diff --git a/.changeset/fix-interrupted-speech-wedges-session.md b/.changeset/fix-interrupted-speech-wedges-session.md new file mode 100644 index 000000000..7ba536d02 --- /dev/null +++ b/.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. 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 23d8b4ea9..12c244665 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 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, '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, @@ -3990,7 +4018,7 @@ export class AgentActivity implements RecognitionHooks { createdAt: number; }): Promise { 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()) { 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..8398c4801 100644 --- a/agents/src/voice/speech_handle.test.ts +++ b/agents/src/voice/speech_handle.test.ts @@ -9,8 +9,9 @@ // 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'; +import { REPLY_TASK_CANCEL_TIMEOUT, SpeechHandle } from './speech_handle.js'; async function raceTimeout(promise: Promise, ms: number): Promise<'resolved' | 'timeout'> { let timer: ReturnType; @@ -185,6 +186,60 @@ 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); + + // 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); + // 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..4e45ca493 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,46 @@ import { functionCallStorage } from './agent.js'; /** Symbol used to identify SpeechHandle instances */ const SPEECH_HANDLE_SYMBOL = Symbol.for('livekit.agents.SpeechHandle'); +/** + * 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 tasks are aborted cooperatively and waited on with this bound instead. + * + * 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 = 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`, + * 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 = 5000; + /** * Type guard to check if a value is a SpeechHandle. */ @@ -96,6 +137,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 +355,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 +466,8 @@ export class SpeechHandle { if (this.generations.length > 0) { this._markGenerationDone(); } + + this.clearInterruptTimeout(); } /** @internal */