fix(voice): stop an interrupted reply from muting the session forever

.changeset/fix-interrupted-speech-wedges-session.md

@@ -0,0 +1,22 @@
+---
+'@livekit/agents': patch
+---
+
+fix(voice): stop an interrupted reply from muting the session forever
+
+A reply interrupted before its audio started playing could leave its pipeline reply task
+parked in the post-interrupt `waitForPlayout()`, which races only the reply's own abort
+signal — a signal nothing on the ordinary interrupt path ever fires. The speech scheduling
+loop waits on that reply's generation, so `_currentSpeech` stayed pinned on the interrupted
+handle and every later turn was queued but never authorized: the agent went silent for the
+rest of the session. On the evidence so far this needs an audio sink whose playback-finished
+event the pipeline does not produce itself — remote avatar outputs (`DataStreamAudioOutput`
+and the avatar plugins built on it) and user-supplied `AudioOutput`s; a plain room output
+settled both of the affected waits on its own across six live runs.
+
+`SpeechHandle` now arms a 5s watchdog when a speech is interrupted (a port of python's
+`INTERRUPTION_TIMEOUT`): if the speech has not finished by then, its tasks are cancelled —
+firing exactly the abort signal those waits are already watching — and the handle is marked
+done, releasing the scheduler.
+
+Fixes #2065.

agents/src/voice/agent_activity.test.ts

@@ -25,6 +25,7 @@ import {
 import { LLM, type LLMStream } from '../llm/llm.js';
 import { type GenerationCreatedEvent, RealtimeError } from '../llm/realtime.js';
 import { type Tool, ToolContext, ToolFlag, Toolset, tool } from '../llm/tool_context.js';
+import { log } from '../log.js';
 import { Future, Task } from '../utils.js';
 import { AgentTask, _getActivityTaskInfo } from './agent.js';
 import { AgentActivity, onEnterStorage } from './agent_activity.js';
@@ -1236,6 +1237,7 @@ describe('AgentActivity - interruption while waiting for tools', () => {
     Object.assign(activity, {
       _backgroundSpeeches: new Set<SpeechHandle>(),
       _commitInterruptedToolOutputs: commitInterruptedToolOutputs,
+      logger: log(),
     });
     const waitForToolExecution = (
       activity as unknown as { _waitForToolExecution: WaitForToolExecution }
@@ -1287,4 +1289,29 @@ describe('AgentActivity - interruption while waiting for tools', () => {
     expect(commitInterruptedToolOutputs).toHaveBeenCalledWith(toolOutput, speechHandle, 456);
     expect(activity['_backgroundSpeeches']).not.toContain(speechHandle);
   });
+
+  it('still commits outputs when cancelling the tool task overruns its budget', async () => {
+    // `Task.cancelAndWait` throws once the cooperative-cancel budget expires. A tool that
+    // ignores its abort signal must not take the commit down with it — the LLM has already
+    // seen these outputs, so dropping them leaves the function calls dangling.
+    const { commitInterruptedToolOutputs, waitForToolExecution } = buildActivity();
+    const speechHandle = SpeechHandle.create();
+    speechHandle.interrupt();
+    const toolOutput = buildToolOutput();
+
+    const shouldContinue = await waitForToolExecution({
+      executeToolsTask: {
+        result: new Promise<void>(() => {}),
+        cancelAndWait: vi.fn(async () => {
+          throw new Error('Task cancellation timed out');
+        }),
+      },
+      toolOutput,
+      speechHandle,
+      createdAt: 789,
+    });
+
+    expect(shouldContinue).toBe(false);
+    expect(commitInterruptedToolOutputs).toHaveBeenCalledWith(toolOutput, speechHandle, 789);
+  });
 });

agents/src/voice/agent_activity.ts

@@ -136,7 +136,7 @@ import {
   updateInstructions,
 } from './generation.js';
 import type { PlaybackFinishedEvent, TimedString } from './io.js';
-import { type InputDetails, SpeechHandle } from './speech_handle.js';
+import { type InputDetails, REPLY_TASK_CANCEL_TIMEOUT, SpeechHandle } from './speech_handle.js';
 import {
   ToolExecutor,
   cancelTaskTool,
@@ -257,8 +257,6 @@ export class AgentActivity implements RecognitionHooks {
   agent: Agent;
   agentSession: AgentSession;
 
-  private static readonly REPLY_TASK_CANCEL_TIMEOUT = 5000;
-
   private started = false;
   private audioRecognition?: AudioRecognition;
   private realtimeSession?: RealtimeSession;
@@ -2653,7 +2651,7 @@ export class AgentActivity implements RecognitionHooks {
 
       if (speechHandle.interrupted) {
         replyAbortController.abort();
-        await cancelAndWait(tasks, AgentActivity.REPLY_TASK_CANCEL_TIMEOUT);
+        await cancelAndWait(tasks, REPLY_TASK_CANCEL_TIMEOUT);
         if (audioOutput) {
           audioOutput.clearBuffer();
           await audioOutput.waitForPlayout();
@@ -2903,7 +2901,7 @@ export class AgentActivity implements RecognitionHooks {
 
     if (speechHandle.interrupted) {
       replyAbortController.abort();
-      await cancelAndWait(tasks, AgentActivity.REPLY_TASK_CANCEL_TIMEOUT);
+      await cancelAndWait(tasks, REPLY_TASK_CANCEL_TIMEOUT);
       return;
     }
 
@@ -3016,7 +3014,7 @@ export class AgentActivity implements RecognitionHooks {
         }
 
         if (speechHandle.interrupted) {
-          await cancelAndWait(forwardTasks, AgentActivity.REPLY_TASK_CANCEL_TIMEOUT);
+          await cancelAndWait(forwardTasks, REPLY_TASK_CANCEL_TIMEOUT);
           if (audioOutput) {
             audioOutput.clearBuffer();
             // During shutdown (room disconnected / activity closing) the
@@ -3071,7 +3069,7 @@ export class AgentActivity implements RecognitionHooks {
         return output;
       } finally {
         replyAbortController.signal.removeEventListener('abort', abortSegment);
-        await cancelAndWait(forwardTasks, AgentActivity.REPLY_TASK_CANCEL_TIMEOUT);
+        await cancelAndWait(forwardTasks, REPLY_TASK_CANCEL_TIMEOUT);
         // The segment's playout window is over; settle a still-pending
         // firstFrameFut so the playback-started listener is detached.
         this.settleFirstFrameFut(output.audioOut);
@@ -3176,7 +3174,7 @@ export class AgentActivity implements RecognitionHooks {
       );
 
       replyAbortController.abort();
-      await cancelAndWait(tasks, AgentActivity.REPLY_TASK_CANCEL_TIMEOUT);
+      await cancelAndWait(tasks, REPLY_TASK_CANCEL_TIMEOUT);
 
       const forwardedText = segmentOutputs.map(forwardedTextFor).join('');
 
@@ -3217,7 +3215,7 @@ export class AgentActivity implements RecognitionHooks {
       if (speechHandle._hasGenerations) {
         speechHandle._markGenerationDone();
       }
-      await executeToolsTask.cancelAndWait(AgentActivity.REPLY_TASK_CANCEL_TIMEOUT);
+      await this.cancelToolExecutions(executeToolsTask, speechHandle, toolOutput);
       this._commitInterruptedToolOutputs(toolOutput, speechHandle, replyStartedAt);
       return;
     }
@@ -3601,7 +3599,7 @@ export class AgentActivity implements RecognitionHooks {
         }
 
         if (speechHandle.interrupted) {
-          await cancelAndWait(forwardTasks, AgentActivity.REPLY_TASK_CANCEL_TIMEOUT);
+          await cancelAndWait(forwardTasks, REPLY_TASK_CANCEL_TIMEOUT);
           if (audioOutput) {
             audioOutput.clearBuffer();
             const playbackEv = await audioOutput.waitForPlayout();
@@ -3639,7 +3637,7 @@ export class AgentActivity implements RecognitionHooks {
         return output;
       } finally {
         abortController.signal.removeEventListener('abort', abortMessage);
-        await cancelAndWait(forwardTasks, AgentActivity.REPLY_TASK_CANCEL_TIMEOUT);
+        await cancelAndWait(forwardTasks, REPLY_TASK_CANCEL_TIMEOUT);
         // The message's playout window is over; settle a still-pending
         // firstFrameFut so the playback-started listener is detached.
         this.settleFirstFrameFut(output.audioOut);
@@ -3786,7 +3784,7 @@ export class AgentActivity implements RecognitionHooks {
         'Aborting all realtime generation tasks due to interruption',
       );
       replyAbortController.abort();
-      await cancelAndWait(tasks, AgentActivity.REPLY_TASK_CANCEL_TIMEOUT);
+      await cancelAndWait(tasks, REPLY_TASK_CANCEL_TIMEOUT);
       addRealtimeMessageOutputs(messageOutputs);
 
       const anySkipped = messageOutputs.some((output) => output.played === 'skipped');
@@ -3808,7 +3806,7 @@ export class AgentActivity implements RecognitionHooks {
         }
       }
       speechHandle._markGenerationDone();
-      await executeToolsTask.cancelAndWait(AgentActivity.REPLY_TASK_CANCEL_TIMEOUT);
+      await this.cancelToolExecutions(executeToolsTask, speechHandle, toolOutput);
 
       // TODO(brian): close tees
       return;
@@ -3977,6 +3975,36 @@ export class AgentActivity implements RecognitionHooks {
     this.scheduleSpeech(replySpeechHandle, SpeechHandle.SPEECH_PRIORITY_NORMAL, true);
   }
 
+  /**
+   * Cancel the tool-execution task, tolerating a cancellation that overruns its budget.
+   *
+   * `Task.cancelAndWait` throws once the budget expires, and every caller still has work to do
+   * afterwards — committing the interrupted tool outputs above all. A tool that ignores its abort
+   * signal must degrade to a warning, not propagate and drop outputs the LLM has already seen.
+   */
+  private async cancelToolExecutions(
+    executeToolsTask: Pick<Task<void>, 'cancelAndWait'>,
+    speechHandle: SpeechHandle,
+    toolOutput: ToolOutput,
+  ): Promise<void> {
+    try {
+      await executeToolsTask.cancelAndWait(REPLY_TASK_CANCEL_TIMEOUT);
+    } catch (error) {
+      this.logger.warn(
+        {
+          error,
+          speech_id: speechHandle.id,
+          timeout: REPLY_TASK_CANCEL_TIMEOUT,
+          tool_calls: toolOutput.output.map((output) => ({
+            function: output.toolCall.name,
+            call_id: output.toolCall.callId,
+          })),
+        },
+        'tool execution task did not settle within the cancellation budget, continuing teardown',
+      );
+    }
+  }
+
   /** @internal */
   async _waitForToolExecution({
     executeToolsTask,
@@ -3990,7 +4018,7 @@ export class AgentActivity implements RecognitionHooks {
     createdAt: number;
   }): Promise<boolean> {
     if (speechHandle.interrupted) {
-      await executeToolsTask.cancelAndWait(AgentActivity.REPLY_TASK_CANCEL_TIMEOUT);
+      await this.cancelToolExecutions(executeToolsTask, speechHandle, toolOutput);
       this._commitInterruptedToolOutputs(toolOutput, speechHandle, createdAt);
       return false;
     }
@@ -4373,7 +4401,7 @@ export class AgentActivity implements RecognitionHooks {
         this._currentSpeech._cancel();
       }
 
-      await cancelAndWait(Array.from(this.speechTasks), AgentActivity.REPLY_TASK_CANCEL_TIMEOUT);
+      await cancelAndWait(Array.from(this.speechTasks), REPLY_TASK_CANCEL_TIMEOUT);
       await this._toolExecutor.drain();
 
       if (this._currentSpeech && !this._currentSpeech.done()) {