From e2aea761ee6fe04e149dcdde6ee46e0fba351255 Mon Sep 17 00:00:00 2001 From: Toubat Date: Sun, 26 Jul 2026 22:09:08 -0700 Subject: [PATCH 1/6] fix(voice): stop an interrupted reply from muting the session forever MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit A reply interrupted before its audio started playing can leave its pipeline reply task parked in the post-interrupt waitForPlayout(), which races only the reply's own abort signal — and nothing on the ordinary interrupt path ever fires it. The speech scheduling loop waits on that reply's generation, so _currentSpeech stays pinned on the interrupted handle and every later turn is queued but never authorized. Port python's INTERRUPTION_TIMEOUT watchdog: when a speech is interrupted, arm a 5s timer that cancels the handle's tasks — firing exactly the abort signal those waits are already watching — and marks the handle done, releasing the scheduler. Fixes #2065. Co-authored-by: Cursor --- .../fix-interrupted-speech-wedges-session.md | 19 ++ .../interrupt_before_playout_deadlock.test.ts | 192 ++++++++++++++++++ agents/src/voice/speech_handle.test.ts | 49 +++++ agents/src/voice/speech_handle.ts | 52 +++++ 4 files changed, 312 insertions(+) create mode 100644 .changeset/fix-interrupted-speech-wedges-session.md create mode 100644 agents/src/voice/interrupt_before_playout_deadlock.test.ts diff --git a/.changeset/fix-interrupted-speech-wedges-session.md b/.changeset/fix-interrupted-speech-wedges-session.md new file mode 100644 index 000000000..6fade2d2f --- /dev/null +++ b/.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. diff --git a/agents/src/voice/interrupt_before_playout_deadlock.test.ts b/agents/src/voice/interrupt_before_playout_deadlock.test.ts new file mode 100644 index 000000000..47a5e5b8c --- /dev/null +++ b/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> { + let push!: (frame: AudioFrame) => void; + let close!: () => void; + const stream = new ReadableStream({ + 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 { + 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(promise: Promise, 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 { + 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); +}); diff --git a/agents/src/voice/speech_handle.test.ts b/agents/src/voice/speech_handle.test.ts index e75d0bc17..5c05d77d2 100644 --- a/agents/src/voice/speech_handle.test.ts +++ b/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((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(); diff --git a/agents/src/voice/speech_handle.ts b/agents/src/voice/speech_handle.ts index 5fc4c74c0..f9995cf80 100644 --- a/agents/src/voice/speech_handle.ts +++ b/agents/src/voice/speech_handle.ts @@ -4,6 +4,7 @@ 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'; @@ -11,6 +12,14 @@ 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; + 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 */ From e3a49338c35d4ed36b817b6afc67a9f9db565034 Mon Sep 17 00:00:00 2001 From: Toubat Date: Sun, 26 Jul 2026 22:48:17 -0700 Subject: [PATCH 2/6] fix(voice): order the interruption watchdog after the reply cancel budget MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Both timeouts were 5s. Python's cancellation is immediate, so its INTERRUPTION_TIMEOUT only ever elapses when a reply is genuinely stuck; here a reply that unwinds slowly can legitimately spend the whole cooperative cancel budget first, and the watchdog would then mark the handle done at the very moment that reply resumes to commit its turn — trading a muted session for a lost assistant message. Hoist REPLY_TASK_CANCEL_TIMEOUT next to INTERRUPTION_TIMEOUT so the ordering invariant between them is stated once, and derive the watchdog from it. Co-authored-by: Cursor --- agents/src/voice/agent_activity.ts | 28 ++++++++++++-------------- agents/src/voice/speech_handle.test.ts | 10 +++++++-- agents/src/voice/speech_handle.ts | 17 ++++++++++++++-- 3 files changed, 36 insertions(+), 19 deletions(-) diff --git a/agents/src/voice/agent_activity.ts b/agents/src/voice/agent_activity.ts index 23d8b4ea9..89e9e6c92 100644 --- a/agents/src/voice/agent_activity.ts +++ b/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 executeToolsTask.cancelAndWait(REPLY_TASK_CANCEL_TIMEOUT); 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 executeToolsTask.cancelAndWait(REPLY_TASK_CANCEL_TIMEOUT); // TODO(brian): close tees return; @@ -3990,7 +3988,7 @@ export class AgentActivity implements RecognitionHooks { createdAt: number; }): Promise { if (speechHandle.interrupted) { - await executeToolsTask.cancelAndWait(AgentActivity.REPLY_TASK_CANCEL_TIMEOUT); + await executeToolsTask.cancelAndWait(REPLY_TASK_CANCEL_TIMEOUT); this._commitInterruptedToolOutputs(toolOutput, speechHandle, createdAt); return false; } @@ -4373,7 +4371,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()) { diff --git a/agents/src/voice/speech_handle.test.ts b/agents/src/voice/speech_handle.test.ts index 5c05d77d2..8398c4801 100644 --- a/agents/src/voice/speech_handle.test.ts +++ b/agents/src/voice/speech_handle.test.ts @@ -11,7 +11,7 @@ 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'; +import { REPLY_TASK_CANCEL_TIMEOUT, SpeechHandle } from './speech_handle.js'; async function raceTimeout(promise: Promise, ms: number): Promise<'resolved' | 'timeout'> { let timer: ReturnType; @@ -204,7 +204,13 @@ describe('SpeechHandle interruption watchdog (#2065)', () => { handle.interrupt(); expect(handle.done()).toBe(false); - await vi.advanceTimersByTimeAsync(5000); + // A reply that unwinds slowly is allowed the whole cooperative-cancel budget before the + // watchdog is entitled to act. Firing inside that window would cut off a teardown that is + // still on its way to committing the turn. + await vi.advanceTimersByTimeAsync(REPLY_TASK_CANCEL_TIMEOUT); + expect(handle.done()).toBe(false); + + await vi.advanceTimersByTimeAsync(10_000); expect(task.done).toBe(true); expect(handle.done()).toBe(true); diff --git a/agents/src/voice/speech_handle.ts b/agents/src/voice/speech_handle.ts index f9995cf80..f8bf8107a 100644 --- a/agents/src/voice/speech_handle.ts +++ b/agents/src/voice/speech_handle.ts @@ -12,13 +12,26 @@ import { functionCallStorage } from './agent.js'; /** Symbol used to identify SpeechHandle instances */ const SPEECH_HANDLE_SYMBOL = Symbol.for('livekit.agents.SpeechHandle'); +/** + * How long a reply task may take to unwind after its abort signal fires. + * + * Python cancels a task outright and the `CancelledError` lands at its next await point, so + * `utils.aio.cancel_and_wait` needs no ceiling. Nothing here can cancel a pending promise, so the + * reply tasks are aborted cooperatively and waited on with this bound instead. + */ +export const REPLY_TASK_CANCEL_TIMEOUT = 5000; + /** * 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`). + * (`livekit-agents/livekit/agents/voice/speech_handle.py`), but must stay strictly greater than + * {@link REPLY_TASK_CANCEL_TIMEOUT}. Python's 5s can only elapse when a reply is genuinely stuck; + * here a reply that unwinds slowly may legitimately spend the whole cancel budget first. Were the + * two equal, this watchdog would mark the handle done at the very moment such a reply resumes to + * commit its turn, turning a slow teardown into a lost assistant message. */ -const INTERRUPTION_TIMEOUT = 5000; +const INTERRUPTION_TIMEOUT = REPLY_TASK_CANCEL_TIMEOUT + 3000; /** * Type guard to check if a value is a SpeechHandle. From a85d2cdaa227d598694ea25558bc3eccd3e223ca Mon Sep 17 00:00:00 2001 From: Toubat Date: Sun, 26 Jul 2026 22:59:40 -0700 Subject: [PATCH 3/6] fix(voice): keep the interruption watchdog at python's 5s MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Padding the watchdog above the reply cancel budget bought the ordering at the cost of the number that matters: INTERRUPTION_TIMEOUT is dead air the user hears before the session recovers, and python sets it to 5s. Shrink the cooperative cancel budget instead — it has no python counterpart, since a cancelled task there takes CancelledError at its next await point — so the ordering holds with margin and recovery stays at 5s. Co-authored-by: Cursor --- agents/src/voice/speech_handle.ts | 21 +++++++++++---------- 1 file changed, 11 insertions(+), 10 deletions(-) diff --git a/agents/src/voice/speech_handle.ts b/agents/src/voice/speech_handle.ts index f8bf8107a..d215a5a78 100644 --- a/agents/src/voice/speech_handle.ts +++ b/agents/src/voice/speech_handle.ts @@ -15,23 +15,24 @@ const SPEECH_HANDLE_SYMBOL = Symbol.for('livekit.agents.SpeechHandle'); /** * How long a reply task may take to unwind after its abort signal fires. * - * Python cancels a task outright and the `CancelledError` lands at its next await point, so - * `utils.aio.cancel_and_wait` needs no ceiling. Nothing here can cancel a pending promise, so the - * reply tasks are aborted cooperatively and waited on with this bound instead. + * Has no python counterpart: there a cancelled task takes `CancelledError` at its next await point, + * so `utils.aio.cancel_and_wait` needs no ceiling at all. Nothing here can cancel a pending + * promise, so reply tasks are aborted cooperatively and waited on with this bound instead. It has + * to stay comfortably below {@link INTERRUPTION_TIMEOUT}: a reply that spends its whole budget and + * a watchdog that fires at the same instant would mark the handle done just as that reply resumes + * to commit its turn, trading a muted session for a lost assistant message. */ -export const REPLY_TASK_CANCEL_TIMEOUT = 5000; +export const REPLY_TASK_CANCEL_TIMEOUT = 2000; /** * 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`), but must stay strictly greater than - * {@link REPLY_TASK_CANCEL_TIMEOUT}. Python's 5s can only elapse when a reply is genuinely stuck; - * here a reply that unwinds slowly may legitimately spend the whole cancel budget first. Were the - * two equal, this watchdog would mark the handle done at the very moment such a reply resumes to - * commit its turn, turning a slow teardown into a lost assistant message. + * (`livekit-agents/livekit/agents/voice/speech_handle.py`). This is dead air the user hears before + * the session recovers, so it tracks python's value rather than being padded; the ordering against + * {@link REPLY_TASK_CANCEL_TIMEOUT} is bought by keeping that budget small instead. */ -const INTERRUPTION_TIMEOUT = REPLY_TASK_CANCEL_TIMEOUT + 3000; +const INTERRUPTION_TIMEOUT = 5000; /** * Type guard to check if a value is a SpeechHandle. From c117245a3a971d839b98a49510e32e48076cf8f0 Mon Sep 17 00:00:00 2001 From: Toubat Date: Mon, 27 Jul 2026 12:50:45 -0700 Subject: [PATCH 4/6] docs(changeset): scope the interrupted-reply fix to the sinks it reaches The changeset described the defect generally, so a user on a plain room output would read the release note as applying to them. A live reproduction against a real room audio output did not wedge in six runs: `ParticipantAudioOutput` settles both of the affected waits on its own, so reaching the deadlock appears to require a sink whose playback-finished event the pipeline does not produce itself. The regression test's stand-in sink already encodes that scope; the prose now says it. Co-authored-by: Cursor --- .changeset/fix-interrupted-speech-wedges-session.md | 5 ++++- 1 file changed, 4 insertions(+), 1 deletion(-) diff --git a/.changeset/fix-interrupted-speech-wedges-session.md b/.changeset/fix-interrupted-speech-wedges-session.md index 6fade2d2f..7ba536d02 100644 --- a/.changeset/fix-interrupted-speech-wedges-session.md +++ b/.changeset/fix-interrupted-speech-wedges-session.md @@ -9,7 +9,10 @@ parked in the post-interrupt `waitForPlayout()`, which races only the reply's ow 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. +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 — From 2a5a56f60914fa2ab78b366bc7aa910e0932a9cb Mon Sep 17 00:00:00 2001 From: Toubat Date: Mon, 27 Jul 2026 13:37:12 -0700 Subject: [PATCH 5/6] fix(voice): keep the cooperative-cancel budget at its existing value Expressing the watchdog against REPLY_TASK_CANCEL_TIMEOUT was right; buying the ordering by halving that constant was not. It also governs tool-execution cancellation (three sites, each followed immediately by _commitInterruptedToolOutputs) and the shutdown drain, so 5s -> 2s abandoned teardown work that legitimately takes seconds on paths unrelated to an interrupted reply. Restore the value and put the margin on the watchdog instead. Co-authored-by: Cursor --- agents/src/voice/speech_handle.ts | 28 +++++++++++++++++----------- 1 file changed, 17 insertions(+), 11 deletions(-) diff --git a/agents/src/voice/speech_handle.ts b/agents/src/voice/speech_handle.ts index d215a5a78..c09c86f6b 100644 --- a/agents/src/voice/speech_handle.ts +++ b/agents/src/voice/speech_handle.ts @@ -13,26 +13,32 @@ import { functionCallStorage } from './agent.js'; const SPEECH_HANDLE_SYMBOL = Symbol.for('livekit.agents.SpeechHandle'); /** - * How long a reply task may take to unwind after its abort signal fires. + * How long a cooperatively cancelled task may take to unwind after its abort signal fires. * * Has no python counterpart: there a cancelled task takes `CancelledError` at its next await point, * so `utils.aio.cancel_and_wait` needs no ceiling at all. Nothing here can cancel a pending - * promise, so reply tasks are aborted cooperatively and waited on with this bound instead. It has - * to stay comfortably below {@link INTERRUPTION_TIMEOUT}: a reply that spends its whole budget and - * a watchdog that fires at the same instant would mark the handle done just as that reply resumes - * to commit its turn, trading a muted session for a lost assistant message. + * promise, so tasks are aborted cooperatively and waited on with this bound instead. + * + * Moved here from `AgentActivity` only so {@link INTERRUPTION_TIMEOUT} can be expressed against it. + * The value is unchanged, deliberately: this budget also governs tool-execution cancellation and + * the shutdown drain, so shrinking it to buy the ordering below would abandon teardown work that + * legitimately takes seconds, on paths that have nothing to do with an interrupted reply. */ -export const REPLY_TASK_CANCEL_TIMEOUT = 2000; +export const REPLY_TASK_CANCEL_TIMEOUT = 5000; /** * 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`). This is dead air the user hears before - * the session recovers, so it tracks python's value rather than being padded; the ordering against - * {@link REPLY_TASK_CANCEL_TIMEOUT} is bought by keeping that budget small instead. + * Ports `INTERRUPTION_TIMEOUT` from `livekit-agents/livekit/agents/voice/speech_handle.py`, but + * cannot reuse python's 5s: python has no cooperative-cancel budget to collide with, because + * cancellation there lands at the next await point. Here a teardown may legitimately spend the + * whole of {@link REPLY_TASK_CANCEL_TIMEOUT}, and its clock starts *after* the interrupt that arms + * this watchdog — so at equal values the watchdog always preempts a slow-but-healthy teardown and + * marks the handle done just as it resumes to commit its turn, trading a muted session for a lost + * assistant message. The margin is what this watchdog costs: it is dead air the user hears before + * the session recovers, against a session that on current `main` never recovers at all. */ -const INTERRUPTION_TIMEOUT = 5000; +const INTERRUPTION_TIMEOUT = REPLY_TASK_CANCEL_TIMEOUT + 3000; /** * Type guard to check if a value is a SpeechHandle. From 491a4d1be64b82306f6a2da2fc4689d999c821ef Mon Sep 17 00:00:00 2001 From: Toubat Date: Mon, 27 Jul 2026 13:50:45 -0700 Subject: [PATCH 6/6] fix(voice): shrink the cooperative-cancel budget to 2s for python parity The previous commit kept REPLY_TASK_CANCEL_TIMEOUT at 5s on the belief that it also bounded tool-execution cleanup and the shutdown drain. It does not. The tool executor abandons a cancelled tool outright and only logs if execute() is still running after DRAIN_TOOL_TIMEOUT_MS; non-cancellable tools are awaited by drain() with no ceiling at all; and shutdown's drain() is a separate unbounded await that runs after this budget. All this constant bounds is how long an aborted task body takes to reach its own return. Measuring performToolExecutions, that is 0.04-0.33 ms across a tool that observes its abort signal, a 4s tool that ignores it, a non-cancellable 4s tool, three concurrent tools, and a tool body that never settles. The one shape that exceeds 2s never settles at all -- a task parked on a tool-call stream that is never closed -- and no finite ceiling rescues it. So take 2s, which restores INTERRUPTION_TIMEOUT to python's 5s: 5s of dead air before an ignored interrupt recovers instead of 8s. It also keeps the two budgets an interrupted reply spends in sequence (its task group, then its tool-execution task) inside the watchdog, which 5s did not. Separately, harden the three direct executeToolsTask.cancelAndWait sites. Task.cancelAndWait throws on timeout, and two of them commit the interrupted tool outputs on the very next statement -- so an unresponsive tool dropped outputs the LLM had already seen, leaving its function calls dangling. Log and continue instead. Co-authored-by: Cursor --- agents/src/voice/agent_activity.test.ts | 27 +++++++++++++++++ agents/src/voice/agent_activity.ts | 36 ++++++++++++++++++++-- agents/src/voice/speech_handle.ts | 40 ++++++++++++++++--------- 3 files changed, 86 insertions(+), 17 deletions(-) diff --git a/agents/src/voice/agent_activity.test.ts b/agents/src/voice/agent_activity.test.ts index 3b3ed17e2..5f0f5f86e 100644 --- a/agents/src/voice/agent_activity.test.ts +++ b/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(), _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(() => {}), + 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); + }); }); diff --git a/agents/src/voice/agent_activity.ts b/agents/src/voice/agent_activity.ts index 89e9e6c92..12c244665 100644 --- a/agents/src/voice/agent_activity.ts +++ b/agents/src/voice/agent_activity.ts @@ -3215,7 +3215,7 @@ export class AgentActivity implements RecognitionHooks { if (speechHandle._hasGenerations) { speechHandle._markGenerationDone(); } - await executeToolsTask.cancelAndWait(REPLY_TASK_CANCEL_TIMEOUT); + await this.cancelToolExecutions(executeToolsTask, speechHandle, toolOutput); this._commitInterruptedToolOutputs(toolOutput, speechHandle, replyStartedAt); return; } @@ -3806,7 +3806,7 @@ export class AgentActivity implements RecognitionHooks { } } speechHandle._markGenerationDone(); - await executeToolsTask.cancelAndWait(REPLY_TASK_CANCEL_TIMEOUT); + await this.cancelToolExecutions(executeToolsTask, speechHandle, toolOutput); // TODO(brian): close tees return; @@ -3975,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, 'cancelAndWait'>, + speechHandle: SpeechHandle, + toolOutput: ToolOutput, + ): Promise { + 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, @@ -3988,7 +4018,7 @@ export class AgentActivity implements RecognitionHooks { createdAt: number; }): Promise { if (speechHandle.interrupted) { - await executeToolsTask.cancelAndWait(REPLY_TASK_CANCEL_TIMEOUT); + await this.cancelToolExecutions(executeToolsTask, speechHandle, toolOutput); this._commitInterruptedToolOutputs(toolOutput, speechHandle, createdAt); return false; } diff --git a/agents/src/voice/speech_handle.ts b/agents/src/voice/speech_handle.ts index c09c86f6b..4e45ca493 100644 --- a/agents/src/voice/speech_handle.ts +++ b/agents/src/voice/speech_handle.ts @@ -19,26 +19,38 @@ const SPEECH_HANDLE_SYMBOL = Symbol.for('livekit.agents.SpeechHandle'); * so `utils.aio.cancel_and_wait` needs no ceiling at all. Nothing here can cancel a pending * promise, so tasks are aborted cooperatively and waited on with this bound instead. * - * Moved here from `AgentActivity` only so {@link INTERRUPTION_TIMEOUT} can be expressed against it. - * The value is unchanged, deliberately: this budget also governs tool-execution cancellation and - * the shutdown drain, so shrinking it to buy the ordering below would abandon teardown work that - * legitimately takes seconds, on paths that have nothing to do with an interrupted reply. + * All it bounds is how long an *aborted task body* takes to reach its own return — not how long + * the work it started takes. The pipeline's tasks are stream pumps that re-check `signal.aborted` + * each iteration, and `performToolExecutions` never awaits a user tool promise on this path: it + * races the abort signal (`_waitForToolExecutionResult`), while `ToolExecutor` abandons a cancelled + * tool outright and only *logs* if `execute()` is still running after `DRAIN_TOOL_TIMEOUT_MS`. + * Non-cancellable tools are awaited to completion by `ToolExecutor.drain()` with no ceiling at all, + * and the shutdown drain is a separate unbounded await that runs *after* this budget — so tool + * cleanup is governed by the executor's own design, never by this value. + * + * Measured `performToolExecutions` settle latency after `abort()`: 0.04–0.33 ms across a tool that + * observes its abort signal, a 4s tool that ignores it, a non-cancellable 4s tool, three concurrent + * tools, and a tool body that never settles at all. The one shape that exceeds 2s is a task parked + * on a tool-call stream that is never closed — it never settles, so no finite ceiling rescues it. + * + * 2s is therefore ample, and small enough that the two budgets an interrupted reply spends in + * sequence (its task group, then its tool-execution task) still fit inside + * {@link INTERRUPTION_TIMEOUT}; at 5s they would not. */ -export const REPLY_TASK_CANCEL_TIMEOUT = 5000; +export const REPLY_TASK_CANCEL_TIMEOUT = 2000; /** * How long an interrupted speech may keep running before its tasks are cancelled outright. * - * Ports `INTERRUPTION_TIMEOUT` from `livekit-agents/livekit/agents/voice/speech_handle.py`, but - * cannot reuse python's 5s: python has no cooperative-cancel budget to collide with, because - * cancellation there lands at the next await point. Here a teardown may legitimately spend the - * whole of {@link REPLY_TASK_CANCEL_TIMEOUT}, and its clock starts *after* the interrupt that arms - * this watchdog — so at equal values the watchdog always preempts a slow-but-healthy teardown and - * marks the handle done just as it resumes to commit its turn, trading a muted session for a lost - * assistant message. The margin is what this watchdog costs: it is dead air the user hears before - * the session recovers, against a session that on current `main` never recovers at all. + * Ports `INTERRUPTION_TIMEOUT` from `livekit-agents/livekit/agents/voice/speech_handle.py`, + * including its value. It must stay strictly greater than {@link REPLY_TASK_CANCEL_TIMEOUT}: a + * teardown may legitimately spend that whole budget, and its clock starts *after* the interrupt + * that arms this watchdog, so at equal values the watchdog always preempts a slow-but-healthy + * teardown and marks the handle done just as it resumes to commit its turn — trading a muted + * session for a lost assistant message. The margin is what this watchdog costs: dead air the user + * hears before the session recovers, against a session that without it never recovers at all. */ -const INTERRUPTION_TIMEOUT = REPLY_TASK_CANCEL_TIMEOUT + 3000; +const INTERRUPTION_TIMEOUT = 5000; /** * Type guard to check if a value is a SpeechHandle.