{"version":3,"sources":["../../src/voice/interrupt_before_playout_deadlock.test.ts"],"sourcesContent":["// SPDX-FileCopyrightText: 2026 LiveKit, Inc.\n//\n// SPDX-License-Identifier: Apache-2.0\n\n/**\n * Regression test for livekit/agents-js#2065.\n *\n * A reply that is interrupted before its audio ever starts playing can leave the pipeline\n * reply task parked in `forwardSegment`'s post-interrupt `waitForPlayout()`. That wait races\n * the reply's own abort signal, but nothing on the ordinary interrupt path fires it — the\n * `replyAbortController.abort()` call lives past the segment loop the wait is blocking. The\n * speech scheduling loop then waits on that reply's generation future, so `_currentSpeech`\n * stays pinned on the interrupted handle and every later turn is queued but never authorized:\n * the agent goes silent for the rest of the session.\n *\n * The output below stands in for the class of sinks whose playback-finished event is not the\n * pipeline's to produce — remote avatar outputs (`DataStreamAudioOutput` and every avatar\n * plugin built on it) and any user-supplied `AudioOutput`. It honors the `AudioOutput`\n * contract: every segment it counts is eventually finished exactly once. It just cannot report\n * a segment the remote never began playing until the next turn's audio reaches it, which is\n * what closes the loop — the next turn cannot start until the wedged one lets go.\n *\n * The fix is `SpeechHandle`'s interruption watchdog (a port of python's `INTERRUPTION_TIMEOUT`\n * in `voice/speech_handle.py`), which cancels an interrupted speech's tasks — firing exactly\n * the abort signal those waits are already watching — and marks the handle done.\n */\nimport { AudioFrame } from '@livekit/rtc-node';\nimport { ReadableStream } from 'node:stream/web';\nimport { describe, expect, it } from 'vitest';\nimport { initializeLogger } from '../log.js';\nimport { Agent } from './agent.js';\nimport { AgentSession } from './agent_session.js';\nimport { AudioOutput } from './io.js';\nimport { FakeLLM } from './testing/fake_llm.js';\n\nfunction frame(durationMs = 20, sampleRate = 24000): AudioFrame {\n const samples = Math.floor((sampleRate * durationMs) / 1000);\n return new AudioFrame(new Int16Array(samples), sampleRate, 1, samples);\n}\n\ntype TtsControls = { push: (frame: AudioFrame) => void; close: () => void };\n\n/** Hands each reply's TTS frame stream back to the test so interrupts can be timed exactly. */\nclass ScriptedTtsAgent extends Agent {\n readonly replies: TtsControls[] = [];\n\n constructor() {\n super({ instructions: 'test' });\n }\n\n async ttsNode(): Promise> {\n let push!: (frame: AudioFrame) => void;\n let close!: () => void;\n const stream = new ReadableStream({\n start(controller) {\n push = (f) => {\n try {\n controller.enqueue(f);\n } catch {\n // stream already closed by a previous interrupt\n }\n };\n close = () => {\n try {\n controller.close();\n } catch {\n // already closed\n }\n };\n },\n });\n this.replies.push({ push, close });\n return stream;\n }\n}\n\n/**\n * A remote sink: frames are handed to a worker that plays them and reports back.\n *\n * `startPlayout` decides whether the worker gets far enough to start playing this turn's\n * audio. A segment it never started is not reported when the turn ends — the worker only\n * learns the turn is over when the next turn's audio arrives, and settles the old segment\n * then. Every counted segment is still finished exactly once.\n */\nclass RemoteWorkerAudioOutput extends AudioOutput {\n startPlayout = true;\n private unreportedSegments = 0;\n private playingSegment = false;\n\n constructor() {\n super(24000, undefined, { pause: false });\n }\n\n async captureFrame(f: AudioFrame): Promise {\n while (this.unreportedSegments > 0) {\n this.unreportedSegments--;\n this.onPlaybackFinished({ playbackPosition: 0, interrupted: true });\n }\n await super.captureFrame(f);\n if (this.startPlayout && !this.playingSegment) {\n this.playingSegment = true;\n this.onPlaybackStarted(Date.now());\n }\n }\n\n flush(): void {\n const played = this.playingSegment;\n this.playingSegment = false;\n super.flush();\n if (played) {\n this.onPlaybackFinished({ playbackPosition: 0.02, interrupted: false });\n } else if (this.pendingPlayoutSegments > this.unreportedSegments) {\n this.unreportedSegments++;\n }\n }\n\n clearBuffer(): void {}\n}\n\n/** Surfaces a stall as a value, so a hang fails the test instead of hanging the suite. */\nasync function settleOrStall(promise: Promise, timeoutMs: number) {\n let timer: NodeJS.Timeout | undefined;\n const watchdog = new Promise<'did not settle'>((resolve) => {\n timer = setTimeout(() => resolve('did not settle'), timeoutMs);\n });\n try {\n return await Promise.race([promise, watchdog]);\n } finally {\n if (timer) clearTimeout(timer);\n }\n}\n\nconst sleep = (ms: number) => new Promise((resolve) => setTimeout(resolve, ms));\n\nasync function waitForReply(agent: ScriptedTtsAgent, index: number): Promise {\n for (let i = 0; i < 400 && agent.replies.length <= index; i++) {\n await sleep(5);\n }\n const controls = agent.replies[index];\n if (!controls) throw new Error(`reply ${index} never reached the TTS node`);\n return controls;\n}\n\ndescribe('reply interrupted before playout start (#2065)', () => {\n initializeLogger({ pretty: false, level: 'silent' });\n\n it('keeps speaking after a reply is interrupted before its audio starts playing', async () => {\n const spoken: string[] = [];\n const agent = new ScriptedTtsAgent();\n const session = new AgentSession({\n llm: new FakeLLM([\n { input: 'one', content: 'first reply' },\n { input: 'two', content: 'second reply' },\n ]),\n });\n const audioOutput = new RemoteWorkerAudioOutput();\n session.output.audio = audioOutput;\n session.on('conversation_item_added', (ev) => {\n if (ev.item.role === 'assistant') spoken.push(ev.item.textContent);\n });\n\n await session.start({ agent });\n try {\n // The reply's audio reaches the sink but the remote never starts playing it, and the\n // user's next turn interrupts in that window.\n audioOutput.startPlayout = false;\n const first = session.generateReply({ userInput: 'one' });\n const firstTts = await waitForReply(agent, 0);\n await sleep(50);\n firstTts.push(frame());\n await sleep(50);\n session.interrupt();\n firstTts.close();\n await settleOrStall(first.waitForPlayout(), 10_000);\n\n // The turn the customer never hears: a later reply must still be spoken.\n audioOutput.startPlayout = true;\n const second = session.generateReply({ userInput: 'two' });\n const secondTts = await waitForReply(agent, 1);\n for (let i = 0; i < 5; i++) {\n secondTts.push(frame());\n await sleep(5);\n }\n secondTts.close();\n\n expect(await settleOrStall(second.waitForPlayout(), 15_000)).not.toBe('did not settle');\n expect(spoken).toContain('second reply');\n } finally {\n await settleOrStall(session.close(), 5000);\n }\n }, 60_000);\n});\n"],"mappings":";AA0BA,sBAA2B;AAC3B,iBAA+B;AAC/B,oBAAqC;AACrC,iBAAiC;AACjC,mBAAsB;AACtB,2BAA6B;AAC7B,gBAA4B;AAC5B,sBAAwB;AAExB,SAAS,MAAM,aAAa,IAAI,aAAa,MAAmB;AAC9D,QAAM,UAAU,KAAK,MAAO,aAAa,aAAc,GAAI;AAC3D,SAAO,IAAI,2BAAW,IAAI,WAAW,OAAO,GAAG,YAAY,GAAG,OAAO;AACvE;AAKA,MAAM,yBAAyB,mBAAM;AAAA,EAC1B,UAAyB,CAAC;AAAA,EAEnC,cAAc;AACZ,UAAM,EAAE,cAAc,OAAO,CAAC;AAAA,EAChC;AAAA,EAEA,MAAM,UAA+C;AACnD,QAAI;AACJ,QAAI;AACJ,UAAM,SAAS,IAAI,0BAA2B;AAAA,MAC5C,MAAM,YAAY;AAChB,eAAO,CAAC,MAAM;AACZ,cAAI;AACF,uBAAW,QAAQ,CAAC;AAAA,UACtB,QAAQ;AAAA,UAER;AAAA,QACF;AACA,gBAAQ,MAAM;AACZ,cAAI;AACF,uBAAW,MAAM;AAAA,UACnB,QAAQ;AAAA,UAER;AAAA,QACF;AAAA,MACF;AAAA,IACF,CAAC;AACD,SAAK,QAAQ,KAAK,EAAE,MAAM,MAAM,CAAC;AACjC,WAAO;AAAA,EACT;AACF;AAUA,MAAM,gCAAgC,sBAAY;AAAA,EAChD,eAAe;AAAA,EACP,qBAAqB;AAAA,EACrB,iBAAiB;AAAA,EAEzB,cAAc;AACZ,UAAM,MAAO,QAAW,EAAE,OAAO,MAAM,CAAC;AAAA,EAC1C;AAAA,EAEA,MAAM,aAAa,GAA8B;AAC/C,WAAO,KAAK,qBAAqB,GAAG;AAClC,WAAK;AACL,WAAK,mBAAmB,EAAE,kBAAkB,GAAG,aAAa,KAAK,CAAC;AAAA,IACpE;AACA,UAAM,MAAM,aAAa,CAAC;AAC1B,QAAI,KAAK,gBAAgB,CAAC,KAAK,gBAAgB;AAC7C,WAAK,iBAAiB;AACtB,WAAK,kBAAkB,KAAK,IAAI,CAAC;AAAA,IACnC;AAAA,EACF;AAAA,EAEA,QAAc;AACZ,UAAM,SAAS,KAAK;AACpB,SAAK,iBAAiB;AACtB,UAAM,MAAM;AACZ,QAAI,QAAQ;AACV,WAAK,mBAAmB,EAAE,kBAAkB,MAAM,aAAa,MAAM,CAAC;AAAA,IACxE,WAAW,KAAK,yBAAyB,KAAK,oBAAoB;AAChE,WAAK;AAAA,IACP;AAAA,EACF;AAAA,EAEA,cAAoB;AAAA,EAAC;AACvB;AAGA,eAAe,cAAiB,SAAqB,WAAmB;AACtE,MAAI;AACJ,QAAM,WAAW,IAAI,QAA0B,CAAC,YAAY;AAC1D,YAAQ,WAAW,MAAM,QAAQ,gBAAgB,GAAG,SAAS;AAAA,EAC/D,CAAC;AACD,MAAI;AACF,WAAO,MAAM,QAAQ,KAAK,CAAC,SAAS,QAAQ,CAAC;AAAA,EAC/C,UAAE;AACA,QAAI,MAAO,cAAa,KAAK;AAAA,EAC/B;AACF;AAEA,MAAM,QAAQ,CAAC,OAAe,IAAI,QAAQ,CAAC,YAAY,WAAW,SAAS,EAAE,CAAC;AAE9E,eAAe,aAAa,OAAyB,OAAqC;AACxF,WAAS,IAAI,GAAG,IAAI,OAAO,MAAM,QAAQ,UAAU,OAAO,KAAK;AAC7D,UAAM,MAAM,CAAC;AAAA,EACf;AACA,QAAM,WAAW,MAAM,QAAQ,KAAK;AACpC,MAAI,CAAC,SAAU,OAAM,IAAI,MAAM,SAAS,KAAK,6BAA6B;AAC1E,SAAO;AACT;AAAA,IAEA,wBAAS,kDAAkD,MAAM;AAC/D,mCAAiB,EAAE,QAAQ,OAAO,OAAO,SAAS,CAAC;AAEnD,wBAAG,+EAA+E,YAAY;AAC5F,UAAM,SAAmB,CAAC;AAC1B,UAAM,QAAQ,IAAI,iBAAiB;AACnC,UAAM,UAAU,IAAI,kCAAa;AAAA,MAC/B,KAAK,IAAI,wBAAQ;AAAA,QACf,EAAE,OAAO,OAAO,SAAS,cAAc;AAAA,QACvC,EAAE,OAAO,OAAO,SAAS,eAAe;AAAA,MAC1C,CAAC;AAAA,IACH,CAAC;AACD,UAAM,cAAc,IAAI,wBAAwB;AAChD,YAAQ,OAAO,QAAQ;AACvB,YAAQ,GAAG,2BAA2B,CAAC,OAAO;AAC5C,UAAI,GAAG,KAAK,SAAS,YAAa,QAAO,KAAK,GAAG,KAAK,WAAW;AAAA,IACnE,CAAC;AAED,UAAM,QAAQ,MAAM,EAAE,MAAM,CAAC;AAC7B,QAAI;AAGF,kBAAY,eAAe;AAC3B,YAAM,QAAQ,QAAQ,cAAc,EAAE,WAAW,MAAM,CAAC;AACxD,YAAM,WAAW,MAAM,aAAa,OAAO,CAAC;AAC5C,YAAM,MAAM,EAAE;AACd,eAAS,KAAK,MAAM,CAAC;AACrB,YAAM,MAAM,EAAE;AACd,cAAQ,UAAU;AAClB,eAAS,MAAM;AACf,YAAM,cAAc,MAAM,eAAe,GAAG,GAAM;AAGlD,kBAAY,eAAe;AAC3B,YAAM,SAAS,QAAQ,cAAc,EAAE,WAAW,MAAM,CAAC;AACzD,YAAM,YAAY,MAAM,aAAa,OAAO,CAAC;AAC7C,eAAS,IAAI,GAAG,IAAI,GAAG,KAAK;AAC1B,kBAAU,KAAK,MAAM,CAAC;AACtB,cAAM,MAAM,CAAC;AAAA,MACf;AACA,gBAAU,MAAM;AAEhB,gCAAO,MAAM,cAAc,OAAO,eAAe,GAAG,IAAM,CAAC,EAAE,IAAI,KAAK,gBAAgB;AACtF,gCAAO,MAAM,EAAE,UAAU,cAAc;AAAA,IACzC,UAAE;AACA,YAAM,cAAc,QAAQ,MAAM,GAAG,GAAI;AAAA,IAC3C;AAAA,EACF,GAAG,GAAM;AACX,CAAC;","names":[]}