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

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

@@ -0,0 +1,19 @@
+---
+'@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.
+
+`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/interrupt_before_playout_deadlock.test.ts

@@ -0,0 +1,192 @@
+// SPDX-FileCopyrightText: 2026 LiveKit, Inc.
+//
+// SPDX-License-Identifier: Apache-2.0
+
+/**
+ * Regression test for livekit/agents-js#2065.
+ *
+ * A reply that is interrupted before its audio ever starts playing can leave the pipeline
+ * reply task parked in `forwardSegment`'s post-interrupt `waitForPlayout()`. That wait races
+ * the reply's own abort signal, but nothing on the ordinary interrupt path fires it — the
+ * `replyAbortController.abort()` call lives past the segment loop the wait is blocking. The
+ * speech scheduling loop then waits on that reply's generation future, so `_currentSpeech`
+ * stays pinned on the interrupted handle and every later turn is queued but never authorized:
+ * the agent goes silent for the rest of the session.
+ *
+ * The output below stands in for the class of sinks whose playback-finished event is not the
+ * pipeline's to produce — remote avatar outputs (`DataStreamAudioOutput` and every avatar
+ * plugin built on it) and any user-supplied `AudioOutput`. It honors the `AudioOutput`
+ * contract: every segment it counts is eventually finished exactly once. It just cannot report
+ * a segment the remote never began playing until the next turn's audio reaches it, which is
+ * what closes the loop — the next turn cannot start until the wedged one lets go.
+ *
+ * The fix is `SpeechHandle`'s interruption watchdog (a port of python's `INTERRUPTION_TIMEOUT`
+ * in `voice/speech_handle.py`), which cancels an interrupted speech's tasks — firing exactly
+ * the abort signal those waits are already watching — and marks the handle done.
+ */
+import { AudioFrame } from '@livekit/rtc-node';
+import { ReadableStream } from 'node:stream/web';
+import { describe, expect, it } from 'vitest';
+import { initializeLogger } from '../log.js';
+import { Agent } from './agent.js';
+import { AgentSession } from './agent_session.js';
+import { AudioOutput } from './io.js';
+import { FakeLLM } from './testing/fake_llm.js';
+
+function frame(durationMs = 20, sampleRate = 24000): AudioFrame {
+  const samples = Math.floor((sampleRate * durationMs) / 1000);
+  return new AudioFrame(new Int16Array(samples), sampleRate, 1, samples);
+}
+
+type TtsControls = { push: (frame: AudioFrame) => void; close: () => void };
+
+/** Hands each reply's TTS frame stream back to the test so interrupts can be timed exactly. */
+class ScriptedTtsAgent extends Agent {
+  readonly replies: TtsControls[] = [];
+
+  constructor() {
+    super({ instructions: 'test' });
+  }
+
+  async ttsNode(): Promise<ReadableStream<AudioFrame>> {
+    let push!: (frame: AudioFrame) => void;
+    let close!: () => void;
+    const stream = new ReadableStream<AudioFrame>({
+      start(controller) {
+        push = (f) => {
+          try {
+            controller.enqueue(f);
+          } catch {
+            // stream already closed by a previous interrupt
+          }
+        };
+        close = () => {
+          try {
+            controller.close();
+          } catch {
+            // already closed
+          }
+        };
+      },
+    });
+    this.replies.push({ push, close });
+    return stream;
+  }
+}
+
+/**
+ * A remote sink: frames are handed to a worker that plays them and reports back.
+ *
+ * `startPlayout` decides whether the worker gets far enough to start playing this turn's
+ * audio. A segment it never started is not reported when the turn ends — the worker only
+ * learns the turn is over when the next turn's audio arrives, and settles the old segment
+ * then. Every counted segment is still finished exactly once.
+ */
+class RemoteWorkerAudioOutput extends AudioOutput {
+  startPlayout = true;
+  private unreportedSegments = 0;
+  private playingSegment = false;
+
+  constructor() {
+    super(24000, undefined, { pause: false });
+  }
+
+  async captureFrame(f: AudioFrame): Promise<void> {
+    while (this.unreportedSegments > 0) {
+      this.unreportedSegments--;
+      this.onPlaybackFinished({ playbackPosition: 0, interrupted: true });
+    }
+    await super.captureFrame(f);
+    if (this.startPlayout && !this.playingSegment) {
+      this.playingSegment = true;
+      this.onPlaybackStarted(Date.now());
+    }
+  }
+
+  flush(): void {
+    const played = this.playingSegment;
+    this.playingSegment = false;
+    super.flush();
+    if (played) {
+      this.onPlaybackFinished({ playbackPosition: 0.02, interrupted: false });
+    } else if (this.pendingPlayoutSegments > this.unreportedSegments) {
+      this.unreportedSegments++;
+    }
+  }
+
+  clearBuffer(): void {}
+}
+
+/** Surfaces a stall as a value, so a hang fails the test instead of hanging the suite. */
+async function settleOrStall<T>(promise: Promise<T>, timeoutMs: number) {
+  let timer: NodeJS.Timeout | undefined;
+  const watchdog = new Promise<'did not settle'>((resolve) => {
+    timer = setTimeout(() => resolve('did not settle'), timeoutMs);
+  });
+  try {
+    return await Promise.race([promise, watchdog]);
+  } finally {
+    if (timer) clearTimeout(timer);
+  }
+}
+
+const sleep = (ms: number) => new Promise((resolve) => setTimeout(resolve, ms));
+
+async function waitForReply(agent: ScriptedTtsAgent, index: number): Promise<TtsControls> {
+  for (let i = 0; i < 400 && agent.replies.length <= index; i++) {
+    await sleep(5);
+  }
+  const controls = agent.replies[index];
+  if (!controls) throw new Error(`reply ${index} never reached the TTS node`);
+  return controls;
+}
+
+describe('reply interrupted before playout start (#2065)', () => {
+  initializeLogger({ pretty: false, level: 'silent' });
+
+  it('keeps speaking after a reply is interrupted before its audio starts playing', async () => {
+    const spoken: string[] = [];
+    const agent = new ScriptedTtsAgent();
+    const session = new AgentSession({
+      llm: new FakeLLM([
+        { input: 'one', content: 'first reply' },
+        { input: 'two', content: 'second reply' },
+      ]),
+    });
+    const audioOutput = new RemoteWorkerAudioOutput();
+    session.output.audio = audioOutput;
+    session.on('conversation_item_added', (ev) => {
+      if (ev.item.role === 'assistant') spoken.push(ev.item.textContent);
+    });
+
+    await session.start({ agent });
+    try {
+      // The reply's audio reaches the sink but the remote never starts playing it, and the
+      // user's next turn interrupts in that window.
+      audioOutput.startPlayout = false;
+      const first = session.generateReply({ userInput: 'one' });
+      const firstTts = await waitForReply(agent, 0);
+      await sleep(50);
+      firstTts.push(frame());
+      await sleep(50);
+      session.interrupt();
+      firstTts.close();
+      await settleOrStall(first.waitForPlayout(), 10_000);
+
+      // The turn the customer never hears: a later reply must still be spoken.
+      audioOutput.startPlayout = true;
+      const second = session.generateReply({ userInput: 'two' });
+      const secondTts = await waitForReply(agent, 1);
+      for (let i = 0; i < 5; i++) {
+        secondTts.push(frame());
+        await sleep(5);
+      }
+      secondTts.close();
+
+      expect(await settleOrStall(second.waitForPlayout(), 15_000)).not.toBe('did not settle');
+      expect(spoken).toContain('second reply');
+    } finally {
+      await settleOrStall(session.close(), 5000);
+    }
+  }, 60_000);
+});

agents/src/voice/speech_handle.test.ts

@@ -9,6 +9,7 @@
 // only, and make SpeechHandle itself awaitable.
 import { describe, expect, it, vi } from 'vitest';
 import { FunctionCall } from '../llm/chat_context.js';
+import { Task, waitForAbort } from '../utils.js';
 import { functionCallStorage } from './agent.js';
 import { SpeechHandle } from './speech_handle.js';
 
@@ -185,6 +186,54 @@ describe('SpeechHandle._markDone - generation completion', () => {
   });
 });
 
+describe('SpeechHandle interruption watchdog (#2065)', () => {
+  it('cancels the owned tasks and marks the handle done when an interrupt is ignored', async () => {
+    vi.useFakeTimers();
+    try {
+      const handle = SpeechHandle.create();
+      // A reply task that never observes its interruption — the shape of the #2065 hang,
+      // where the only escape from the post-interrupt playout wait is this abort signal.
+      const task = Task.from(
+        (controller) =>
+          new Promise<void>((resolve) => waitForAbort(controller.signal).then(resolve)),
+      );
+      handle._tasks.push(task);
+      handle._authorizeGeneration();
+      const generationWait = handle._waitForGeneration();
+
+      handle.interrupt();
+      expect(handle.done()).toBe(false);
+
+      await vi.advanceTimersByTimeAsync(5000);
+
+      expect(task.done).toBe(true);
+      expect(handle.done()).toBe(true);
+      // The scheduling loop's wait is released, so the next queued speech can be authorized.
+      await expect(generationWait).resolves.toBeUndefined();
+    } finally {
+      vi.useRealTimers();
+    }
+  });
+
+  it('does not cancel a speech that finishes within the grace period', async () => {
+    vi.useFakeTimers();
+    try {
+      const handle = SpeechHandle.create();
+      const task = Task.from(() => Promise.resolve());
+      const cancel = vi.spyOn(task, 'cancel');
+      handle._tasks.push(task);
+
+      handle.interrupt();
+      handle._markDone();
+      await vi.advanceTimersByTimeAsync(10_000);
+
+      expect(cancel).not.toHaveBeenCalled();
+    } finally {
+      vi.useRealTimers();
+    }
+  });
+});
+
 describe('SpeechHandle.exception', () => {
   it('throws when the handle is not done yet', () => {
     const handle = SpeechHandle.create();

agents/src/voice/speech_handle.ts

@@ -4,13 +4,22 @@
 import { ThrowsPromise } from '@livekit/throws-transformer/throws';
 import type { Context } from '@opentelemetry/api';
 import type { ChatItem } from '../llm/index.js';
+import { log } from '../log.js';
 import type { Task } from '../utils.js';
 import { Event, Future, dedent, shortuuid } from '../utils.js';
 import { functionCallStorage } from './agent.js';
 
 /** Symbol used to identify SpeechHandle instances */
 const SPEECH_HANDLE_SYMBOL = Symbol.for('livekit.agents.SpeechHandle');
 
+/**
+ * How long an interrupted speech may keep running before its tasks are cancelled outright.
+ *
+ * Mirrors `INTERRUPTION_TIMEOUT` in the python implementation
+ * (`livekit-agents/livekit/agents/voice/speech_handle.py`).
+ */
+const INTERRUPTION_TIMEOUT = 5000;
+
 /**
  * Type guard to check if a value is a SpeechHandle.
  */
@@ -96,6 +105,8 @@ export class SpeechHandle {
 
   private itemAddedCallbacks: Set<(item: ChatItem) => void> = new Set();
   private doneCallbacks: Set<(sh: SpeechHandle) => void> = new Set();
+  private interruptTimeout?: ReturnType<typeof setTimeout>;
+  private logger = log();
 
   /** @internal Symbol marker for type identification */
   readonly [SPEECH_HANDLE_SYMBOL] = true;
@@ -312,11 +323,50 @@ export class SpeechHandle {
 
     if (!this.interruptFut.done) {
       this.interruptFut.resolve();
+      this.startInterruptTimeout();
     }
 
     return this;
   }
 
+  /**
+   * Arm the watchdog that force-cancels an interrupted speech that refuses to finish.
+   *
+   * Interrupting only resolves `interruptFut`; it is up to the owning reply task to notice and
+   * unwind. A task parked on something the interruption itself cannot settle — most of the
+   * pipeline reply's post-interrupt waits race the reply's abort signal, and nothing on the
+   * ordinary interrupt path ever fires it — would otherwise never reach
+   * `_markGenerationDone()`. The speech scheduling loop waits on that generation, so a single
+   * stuck reply silently mutes the session for the rest of its life (#2065). Cancelling the
+   * owned tasks aborts exactly the signal those waits are watching; `_markDone` then releases
+   * the scheduler even if a task ignores its signal.
+   *
+   * Ported from python's `SpeechHandle._cancel`.
+   */
+  private startInterruptTimeout(): void {
+    this.interruptTimeout = setTimeout(() => {
+      this.interruptTimeout = undefined;
+      this.logger.error(
+        { speech_id: this._id, timeout: INTERRUPTION_TIMEOUT },
+        'speech not done in time after interruption, cancelling the speech arbitrarily.',
+      );
+      for (const task of this._tasks) {
+        task.cancel();
+      }
+      this._markDone();
+    }, INTERRUPTION_TIMEOUT);
+    // A pending watchdog must not be what keeps a process alive: handles that are interrupted
+    // and then abandoned (never scheduled, so never marked done) would hold the loop open.
+    this.interruptTimeout.unref?.();
+  }
+
+  private clearInterruptTimeout(): void {
+    if (this.interruptTimeout !== undefined) {
+      clearTimeout(this.interruptTimeout);
+      this.interruptTimeout = undefined;
+    }
+  }
+
   /** @internal */
   get _hasGenerations(): boolean {
     return this.generations.length > 0;
@@ -384,6 +434,8 @@ export class SpeechHandle {
     if (this.generations.length > 0) {
       this._markGenerationDone();
     }
+
+    this.clearInterruptTimeout();
   }
 
   /** @internal */