diff --git a/.changeset/pre-playout-interrupt-wedge.md b/.changeset/pre-playout-interrupt-wedge.md new file mode 100644 index 000000000..c8ec545ed --- /dev/null +++ b/.changeset/pre-playout-interrupt-wedge.md @@ -0,0 +1,5 @@ +--- +'@livekit/agents': patch +--- + +Bound the speech scheduler's wait on an interrupted generation and force-abort the reply tasks on timeout, so a reply interrupted before its playout starts no longer wedges `mainTask` and mutes the agent for the rest of the session. diff --git a/agents/src/voice/agent_activity.ts b/agents/src/voice/agent_activity.ts index 13cfd5bb3..803901c9c 100644 --- a/agents/src/voice/agent_activity.ts +++ b/agents/src/voice/agent_activity.ts @@ -254,6 +254,7 @@ export class AgentActivity implements RecognitionHooks { agentSession: AgentSession; private static readonly REPLY_TASK_CANCEL_TIMEOUT = 5000; + private static readonly INTERRUPTED_GENERATION_TIMEOUT = 3000; private started = false; private audioRecognition?: AudioRecognition; @@ -2018,7 +2019,44 @@ export class AgentActivity implements RecognitionHooks { // Keep the shared outputs serialized through interrupted-generation cleanup. // Bare handles have no owner task that can settle the generation future. if (speechHandle.interrupted && speechHandle._tasks.length > 0) { - await ThrowsPromise.race([generation, abortFuture.await]); + // A reply interrupted before its playout starts can park the reply task on + // audioOutput.waitForPlayout() with nothing left to fire the playback event: + // the replyAbortController only aborts after the segment loop returns, and the + // segment loop is blocked on that same await. Generation then never settles and + // the scheduler stays wedged on this handle forever, leaving the agent mute for + // the rest of the session (#2065). Bound the wait; on timeout, cancel the + // handle's tasks — their abort path runs the normal interrupted-reply cleanup, + // which settles the generation future. + let generationSettled = false; + const watchedGeneration = generation.finally(() => { + generationSettled = true; + }); + const forceAbortFut = new Future(); + const forceAbortTimer = setTimeout( + () => forceAbortFut.resolve(), + AgentActivity.INTERRUPTED_GENERATION_TIMEOUT, + ); + try { + await ThrowsPromise.race([watchedGeneration, abortFuture.await, forceAbortFut.await]); + } finally { + clearTimeout(forceAbortTimer); + } + if (!generationSettled && !signal.aborted) { + this.logger.warn( + { + speech_id: speechHandle.id, + timeout_ms: AgentActivity.INTERRUPTED_GENERATION_TIMEOUT, + }, + 'interrupted speech generation is stuck, forcing pipeline abort', + ); + await cancelAndWait( + speechHandle._tasks.filter((task) => !task.done), + AgentActivity.REPLY_TASK_CANCEL_TIMEOUT, + ); + if (!speechHandle.done()) { + speechHandle._markDone(); + } + } } this._currentSpeech = undefined; } diff --git a/agents/src/voice/agent_activity_pre_playout_interrupt.test.ts b/agents/src/voice/agent_activity_pre_playout_interrupt.test.ts new file mode 100644 index 000000000..c702c532b --- /dev/null +++ b/agents/src/voice/agent_activity_pre_playout_interrupt.test.ts @@ -0,0 +1,123 @@ +// SPDX-FileCopyrightText: 2026 LiveKit, Inc. +// +// SPDX-License-Identifier: Apache-2.0 + +/** + * Regression test for livekit/agents-js#2065. + * + * When a pipeline reply is interrupted before its playout has started, the + * reply task parks on `audioOutput.waitForPlayout()` inside its interrupted + * branch: the playback-finished event never fires (playback never began) and + * the reply abort controller is only aborted by code that runs after the + * segment loop returns — which is blocked on that same await. The handle's + * generation future then never settles, `mainTask` never advances past the + * wedged handle, and the agent stays mute for the rest of the session. + * + * The fix bounds mainTask's interrupted-generation wait and, on timeout, + * cancels the handle's tasks; their abort path runs the normal + * interrupted-reply cleanup, which settles the generation future and lets the + * scheduler advance to the next speech. + */ +import { AudioFrame } from '@livekit/rtc-node'; +import { ReadableStream } from 'node:stream/web'; +import { describe, expect, it, vi } 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); +} + +// Audio sink where playback never starts: frames are accepted but no +// playback-started/-finished event is ever reported — the interruption lands +// before the output began playing. `waitForPlayout()` on a captured segment +// therefore never resolves, which is the wedge condition of #2065. Real +// outputs behave this way when the interrupt wins the race against the first +// played frame (e.g. a reply generated in an agent-handoff onEnter while the +// user is still speaking). +class NeverStartsOutput extends AudioOutput { + onFrameCaptured?: () => void; + framesCaptured = 0; + + constructor() { + super(24000); + } + + async captureFrame(f: AudioFrame): Promise { + await super.captureFrame(f); + this.framesCaptured++; + this.onFrameCaptured?.(); + } + + flush(): void { + super.flush(); + } + + clearBuffer(): void { + // Playback never started, so there is no playback event to report. + } +} + +// Agent that synthesizes a few real frames for whatever text it receives, so +// the audio path produces frames without a TTS provider. +class FrameAgent extends Agent { + constructor() { + super({ instructions: 'test' }); + } + + async ttsNode(): Promise | null> { + return new ReadableStream({ + start(controller) { + for (let i = 0; i < 10; i++) controller.enqueue(frame()); + controller.close(); + }, + }); + } +} + +describe('pre-playout interruption (#2065)', () => { + initializeLogger({ pretty: false, level: 'silent' }); + + it('unwedges the speech scheduler when a reply is interrupted before playout starts', async () => { + const session = new AgentSession({ + llm: new FakeLLM([ + { input: 'hello', content: 'A reply that never reaches the speaker.' }, + { input: 'again', content: 'A follow-up reply.' }, + ]), + }); + const audioOut = new NeverStartsOutput(); + session.output.audio = audioOut; + + // Interrupt (non-force, like a new user turn) as soon as the first frame + // is captured — before the output ever reports playback start. + let interruptedOnce = false; + audioOut.onFrameCaptured = () => { + if (!interruptedOnce) { + interruptedOnce = true; + session.interrupt(); + } + }; + + await session.start({ agent: new FrameAgent() }); + try { + const wedged = session.generateReply({ userInput: 'hello' }); + // Pre-fix, the interrupted handle's generation never settles and this + // await hangs until the test times out. + await wedged.waitForPlayout(); + + // The scheduler must have moved past the wedged handle: a follow-up + // reply gets authorized and its audio reaches the output. + const framesBefore = audioOut.framesCaptured; + session.generateReply({ userInput: 'again' }); + await vi.waitFor(() => expect(audioOut.framesCaptured).toBeGreaterThan(framesBefore), { + timeout: 5000, + }); + } finally { + await session.close(); + } + }, 15000); +});