From df471b0b728221ec733528875fd45470abfb841d Mon Sep 17 00:00:00 2001 From: Toubat Date: Sun, 26 Jul 2026 02:50:28 -0700 Subject: [PATCH 1/2] fix(voice): stop a finished interruption from dropping the next reply MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit ParticipantAudioOutput.clearBuffer() resolves an interruptedFuture that frames parked at the pause gate consult to decide whether to bail. Nothing reset that signal until the *next* segment's flush(), which only runs after all of that segment's frames have been captured — so for the whole of the following reply the signal still described an interruption that was over. Pause the output mid-reply in that window (an ordinary false-interruption pause, on by default) and every remaining frame bails at the gate and never reaches the wire, while the session still reports the reply as fully spoken. Measured with a real AgentSession and a real ParticipantAudioOutput: after a barge-in, the next reply lost 18 of 20 frames (360ms of 400ms) and was committed to chat context as complete. The gate is now scoped to the segment being captured: a frame bails only for an interruption raised at or after its own segment began. Parked frames are woken through a per-frame signal so a concurrent flush() that replaces interruptedFuture can no longer strand one at the gate. Co-authored-by: Cursor --- .changeset/stale-interruption-pause-gate.md | 20 ++ .../false_interruption_audio_loss.test.ts | 258 ++++++++++++++++++ agents/src/voice/room_io/_output.test.ts | 69 ++++- agents/src/voice/room_io/_output.ts | 37 ++- 4 files changed, 380 insertions(+), 4 deletions(-) create mode 100644 .changeset/stale-interruption-pause-gate.md create mode 100644 agents/src/voice/false_interruption_audio_loss.test.ts diff --git a/.changeset/stale-interruption-pause-gate.md b/.changeset/stale-interruption-pause-gate.md new file mode 100644 index 000000000..0979ca221 --- /dev/null +++ b/.changeset/stale-interruption-pause-gate.md @@ -0,0 +1,20 @@ +--- +'@livekit/agents': patch +--- + +Stop a finished interruption from silently discarding the next reply + +`ParticipantAudioOutput.clearBuffer()` resolves an `interruptedFuture` that frames parked at +the pause gate consult to decide whether to bail. Nothing reset that signal until the _next_ +segment's `flush()`, which only runs after that segment's frames have all been captured — so +for the whole of the following reply the signal still described an interruption that was +already over. If the output was paused mid-reply during that window (an ordinary +false-interruption pause, on by default), every remaining frame bailed at the gate and never +reached the wire. The session still reported the reply as fully spoken and committed it to the +chat context, so the loss was silent: in a reproduction with a real `AgentSession` and a real +`ParticipantAudioOutput`, 18 of 20 frames — 360ms of a 400ms reply — were dropped. + +The gate is now scoped to the segment being captured: a frame bails only for an interruption +raised at or after its own segment began. Frames parked at the gate are also woken by a +per-frame signal, so a `flush()` that replaces `interruptedFuture` can no longer strand one +there. diff --git a/agents/src/voice/false_interruption_audio_loss.test.ts b/agents/src/voice/false_interruption_audio_loss.test.ts new file mode 100644 index 000000000..239fbf45f --- /dev/null +++ b/agents/src/voice/false_interruption_audio_loss.test.ts @@ -0,0 +1,258 @@ +// SPDX-FileCopyrightText: 2026 LiveKit, Inc. +// +// SPDX-License-Identifier: Apache-2.0 +import { AudioFrame, type Room, TrackPublishOptions, TrackSource } from '@livekit/rtc-node'; +import { ReadableStream } from 'node:stream/web'; +import { describe, expect, it } from 'vitest'; +import { initializeLogger } from '../log.js'; +import { VADEventType } from '../vad.js'; +import { Agent } from './agent.js'; +import { AgentSession } from './agent_session.js'; +import { AgentSessionEventTypes } from './events.js'; +import { ParticipantAudioOutput } from './room_io/_output.js'; +import { FakeLLM } from './testing/fake_llm.js'; + +const SAMPLE_RATE = 24000; +const FRAME_MS = 20; +const FRAMES_PER_REPLY = 20; +const FALSE_INTERRUPTION_TIMEOUT = 400; + +function frame(): AudioFrame { + const samples = (SAMPLE_RATE * FRAME_MS) / 1000; + return new AudioFrame(new Int16Array(samples), SAMPLE_RATE, 1, samples); +} + +function vadEvent(type: VADEventType, speechDuration = 0, silenceDuration = 0) { + return { + type, + samplesIndex: 0, + timestamp: Date.now(), + speechDuration, + silenceDuration, + frames: [], + probability: 1, + inferenceDuration: 0, + speaking: type === VADEventType.START_OF_SPEECH, + rawAccumulatedSilence: 0, + rawAccumulatedSpeech: 0, + } as never; +} + +function sttFinal(text: string) { + return { + type: 'final_transcript', + alternatives: [{ text, language: 'en', startTime: 0, endTime: 0, confidence: 1 }], + } as never; +} + +/** The real ParticipantAudioOutput; only track publishing is skipped (no LiveKit server here). */ +class TestParticipantAudioOutput extends ParticipantAudioOutput { + constructor() { + super({} as Room, { + sampleRate: SAMPLE_RATE, + numChannels: 1, + trackPublishOptions: new TrackPublishOptions({ source: TrackSource.SOURCE_MICROPHONE }), + queueSizeMs: 100_000, + }); + (this as unknown as { startedFuture: { resolve: () => void } }).startedFuture.resolve(); + } + + override async start(): Promise {} +} + +/** Emits `FRAMES_PER_REPLY` frames spaced out in time so a turn can be interleaved with VAD events. */ +class FrameAgent extends Agent { + produced = 0; + onFrame?: (produced: number) => void; + + constructor() { + super({ instructions: 'test' }); + } + + async ttsNode(): Promise> { + let emitted = 0; + const emit = () => { + this.produced++; + this.onFrame?.(this.produced); + }; + return new ReadableStream({ + pull: async (controller) => { + if (emitted >= FRAMES_PER_REPLY) { + controller.close(); + return; + } + if (emitted > 0) await new Promise((resolve) => setTimeout(resolve, 15)); + emitted++; + controller.enqueue(frame()); + emit(); + }, + }); + } +} + +const sleep = (ms: number) => new Promise((resolve) => setTimeout(resolve, ms)); + +async function makeHarness() { + const session = new AgentSession({ + llm: new FakeLLM([ + { input: 'one', content: 'first reply' }, + { input: 'two', content: 'second reply' }, + ]), + aecWarmupDuration: null, + turnHandling: { interruption: { falseInterruptionTimeout: FALSE_INTERRUPTION_TIMEOUT } }, + }); + + const out = new TestParticipantAudioOutput(); + session.output.audio = out; + + // Frame accounting at the boundary the bug lives on: how many frames actually reached + // the LiveKit AudioSource, i.e. the wire. + const source = ( + out as unknown as { audioSource: { captureFrame: (f: AudioFrame) => Promise } } + ).audioSource; + const captureFrame = source.captureFrame.bind(source); + let delivered = 0; + source.captureFrame = async (f: AudioFrame) => { + delivered++; + return captureFrame(f); + }; + + const falseInterruptions: boolean[] = []; + session.on(AgentSessionEventTypes.AgentFalseInterruption, (ev) => + falseInterruptions.push(ev.resumed), + ); + + const agent = new FrameAgent(); + await session.start({ agent }); + + const gate = out as unknown as { playbackEnabledFuture: { done: boolean } }; + + const waitForProduced = (n: number) => + new Promise((resolve) => { + agent.onFrame = (produced) => { + if (produced >= n) resolve(); + }; + }); + + return { + session, + out, + agent, + falseInterruptions, + delivered: () => delivered, + paused: () => !gate.playbackEnabledFuture.done, + waitForProduced, + activity: () => session._activity!, + async close() { + await session.close(); + await out.close(); + }, + }; +} + +type Harness = Awaited>; + +/** + * One reply that a backchannel pauses mid-flight and the false-interruption timer resumes. + * Every frame the TTS produced must reach the wire: the ones captured while the output was + * paused are held at the gate and pushed on resume. + */ +async function falseInterruptedReply(h: Harness, userInput: string) { + const producedBefore = h.agent.produced; + const deliveredBefore = h.delivered(); + + const handle = h.session.generateReply({ userInput }); + await h.waitForProduced(producedBefore + 3); + + h.activity().onStartOfSpeech(vadEvent(VADEventType.START_OF_SPEECH)); + h.activity().onVADInferenceDone(vadEvent(VADEventType.INFERENCE_DONE, 600)); + const pausedMidReply = h.paused(); + // The user only said "mm-hmm": no final transcript follows, so the false-interruption + // timer is what resumes playback. + h.activity().onEndOfSpeech(vadEvent(VADEventType.END_OF_SPEECH, 0, 100)); + + await handle.waitForPlayout(); + await sleep(FALSE_INTERRUPTION_TIMEOUT + 300); + + return { + pausedMidReply, + produced: h.agent.produced - producedBefore, + delivered: h.delivered() - deliveredBefore, + }; +} + +describe('false interruption after a previous interruption', () => { + initializeLogger({ pretty: false, level: 'silent' }); + + it('delivers the whole reply when nothing was interrupted before', async () => { + const h = await makeHarness(); + try { + const reply = await falseInterruptedReply(h, 'one'); + + expect(reply.pausedMidReply).toBe(true); + expect(reply.produced).toBe(FRAMES_PER_REPLY); + expect(reply.delivered).toBe(reply.produced); + expect(h.falseInterruptions).toEqual([true]); + } finally { + await h.close(); + } + }, 30000); + + it('delivers the whole reply when the previous turn ended in a barge-in', async () => { + const h = await makeHarness(); + try { + // A genuine barge-in: VAD pauses the output, the final transcript confirms the + // interruption, and the reply task calls clearBuffer() on the way out. That leaves + // interruptedFuture resolved, and nothing resets it until the *next* segment's flush — + // which only happens after that segment's frames have all been captured. + const bargedInto = h.session.generateReply({ userInput: 'one' }); + await h.waitForProduced(3); + h.activity().onStartOfSpeech(vadEvent(VADEventType.START_OF_SPEECH)); + h.activity().onVADInferenceDone(vadEvent(VADEventType.INFERENCE_DONE, 600)); + h.activity().onFinalTranscript(sttFinal('stop please'), false); + await bargedInto.waitForPlayout(); + await sleep(150); + + // An ordinary false interruption on the next reply. Every frame captured during the + // pause used to bail at the gate on the stale signal, losing the rest of the reply + // while the session still recorded it as fully spoken. + const reply = await falseInterruptedReply(h, 'two'); + + expect(reply.pausedMidReply).toBe(true); + expect(reply.produced).toBe(FRAMES_PER_REPLY); + expect(reply.delivered).toBe(reply.produced); + expect(h.falseInterruptions).toEqual([true]); + } finally { + await h.close(); + } + }, 30000); + + it('delivers the whole reply when a paused reply completed and was then cancelled', async () => { + const h = await makeHarness(); + try { + // The agent finishes its sentence while the user talks over the tail: all frames are + // captured, then the output is paused and the segment drains to a clean finish. + const completed = h.session.generateReply({ userInput: 'one' }); + await h.waitForProduced(FRAMES_PER_REPLY); + for (let i = 0; i < 200 && h.delivered() < FRAMES_PER_REPLY; i++) await sleep(5); + await sleep(10); + h.activity().onStartOfSpeech(vadEvent(VADEventType.START_OF_SPEECH)); + h.activity().onVADInferenceDone(vadEvent(VADEventType.INFERENCE_DONE, 600)); + await completed.waitForPlayout(); + await sleep(100); + expect(h.delivered()).toBe(FRAMES_PER_REPLY); + + // The user's transcript finalizes and cancelSpeechPause un-gates the output. + h.activity().onFinalTranscript(sttFinal('okay thanks'), false); + await sleep(150); + + const reply = await falseInterruptedReply(h, 'two'); + + expect(reply.pausedMidReply).toBe(true); + expect(reply.produced).toBe(FRAMES_PER_REPLY); + expect(reply.delivered).toBe(reply.produced); + } finally { + await h.close(); + } + }, 30000); +}); diff --git a/agents/src/voice/room_io/_output.test.ts b/agents/src/voice/room_io/_output.test.ts index 49a74c119..95d6e07ce 100644 --- a/agents/src/voice/room_io/_output.test.ts +++ b/agents/src/voice/room_io/_output.test.ts @@ -164,13 +164,20 @@ describe('ParticipantAudioOutput captureFrame segment accounting', () => { firstFrameEmitted: boolean; pushedDuration: number; _capturing: boolean; + interruptCount: number; + segmentInterruptCount: number; + segmentOpen: boolean; + gatedFrames: Set>; playbackSegmentsCount: number; playbackFinishedCount: number; playbackFinishedFuture: Future; onPlaybackStarted: (createdAt: number) => void; + logger: { error: () => void }; audioSource: { clearQueue: () => void; captureFrame: (frame: CaptureFrameArg) => Promise; + waitForPlayout: () => Promise; + queuedDuration: number; }; }; @@ -184,21 +191,39 @@ describe('ParticipantAudioOutput captureFrame segment accounting', () => { output.firstFrameEmitted = false; output.pushedDuration = 0; output._capturing = false; + output.interruptCount = 0; + output.segmentInterruptCount = 0; + output.segmentOpen = false; + output.gatedFrames = new Set(); output.playbackSegmentsCount = 0; output.playbackFinishedCount = 0; output.playbackFinishedFuture = new Future(); output.onPlaybackStarted = vi.fn(); - output.audioSource = { clearQueue: vi.fn(), captureFrame: vi.fn(async () => {}) }; + output.logger = { error: vi.fn() }; + output.audioSource = { + clearQueue: vi.fn(), + captureFrame: vi.fn(async () => {}), + // Playout never drains on its own, so a flush task stays pending like a real + // segment still on the wire. + waitForPlayout: () => new Promise(() => {}), + queuedDuration: 0, + }; return output; }; const frame = () => ({ samplesPerChannel: 480, sampleRate: 24000 }) as unknown as CaptureFrameArg; + const settledWithin = async (promise: Promise, ms: number) => + Promise.race([ + promise.then(() => 'settled' as const), + new Promise<'pending'>((resolve) => setTimeout(() => resolve('pending'), ms)), + ]); + it('does not strand the segment counter when a frame is interrupted while paused', async () => { const output = makeOutput({ paused: true }); const capture = output.captureFrame(frame()); - output.interruptedFuture.resolve(); + output.clearBuffer(); await capture; expect(output.playbackSegmentsCount).toBe(0); @@ -211,6 +236,46 @@ describe('ParticipantAudioOutput captureFrame segment accounting', () => { expect(result).toBe('resolved'); }); + it('does not drop a new segment at the pause gate after the previous one was interrupted', async () => { + const output = makeOutput({ paused: false }); + + // Segment 1 plays, is flushed and then interrupted — the state a barge-in leaves behind. + await output.captureFrame(frame()); + output.flush(); + output.clearBuffer(); + await nextTick(); + + // Segment 2 is a fresh reply that gets paused mid-flight by a new overlap. Its frames + // belong to a speech the earlier interruption knows nothing about, so they must wait + // for the resume rather than be discarded. + output.pause(); + const capture = output.captureFrame(frame()); + expect(await settledWithin(capture, 50)).toBe('pending'); + + output.resume(); + await capture; + + expect(output.audioSource.captureFrame).toHaveBeenCalledTimes(2); + expect(output.playbackSegmentsCount).toBe(2); + }); + + it('releases a parked frame when a concurrent flush has replaced the interruption signal', async () => { + const output = makeOutput({ paused: false }); + + await output.captureFrame(frame()); + output.pause(); + const capture = output.captureFrame(frame()); + await nextTick(); + + // An overlapping segment's flush swaps interruptedFuture out from under the parked + // frame; the interruption that follows must still reach it. + output.flush(); + output.clearBuffer(); + + expect(await settledWithin(capture, 500)).toBe('settled'); + expect(output.audioSource.captureFrame).toHaveBeenCalledTimes(1); + }); + it('registers a segment on the normal non-paused path', async () => { const output = makeOutput({ paused: false }); diff --git a/agents/src/voice/room_io/_output.ts b/agents/src/voice/room_io/_output.ts index a78caa9fe..e5c8e9a0b 100644 --- a/agents/src/voice/room_io/_output.ts +++ b/agents/src/voice/room_io/_output.ts @@ -384,6 +384,14 @@ export class ParticipantAudioOutput extends AudioOutput { private firstFrameEmitted: boolean = false; /** Gate held closed while the output is paused; frame forwarding awaits it. */ private playbackEnabledFuture: Future = new Future(); + /** Monotonic count of interruptions signalled through clearBuffer(). */ + private interruptCount: number = 0; + /** interruptCount as of the start of the segment currently being captured. */ + private segmentInterruptCount: number = 0; + /** Whether a frame has been captured since the last flush boundary. */ + private segmentOpen: boolean = false; + /** Wake-up futures for frames parked at the pause gate, all resolved by clearBuffer(). */ + private gatedFrames: Set> = new Set(); constructor(room: Room, options: AudioOutputOptions) { super(options.sampleRate, undefined, { pause: true }); @@ -423,13 +431,31 @@ export class ParticipantAudioOutput extends AudioOutput { } async captureFrame(frame: AudioFrame): Promise { + if (!this.segmentOpen) { + this.segmentOpen = true; + // An interruption raised before this segment began belongs to a speech that is already + // over. interruptedFuture stays resolved until the next flush, so without this snapshot + // every frame of the new speech bails at the gate below and the reply is lost. + this.segmentInterruptCount = this.interruptCount; + } + await this.startedFuture.await; if (!this.playbackEnabledFuture.done) { this.audioSource.clearQueue(); // Race against interruption so a cancel-while-paused can't deadlock an in-flight frame. - await Promise.race([this.playbackEnabledFuture.await, this.interruptedFuture.await]); - if (this.interruptedFuture.done) { + // The wake-up is per frame rather than the shared interruptedFuture, which + // waitForPlayoutTask may replace while this frame is parked. + if (this.interruptCount === this.segmentInterruptCount) { + const gate = new Future(); + this.gatedFrames.add(gate); + try { + await Promise.race([this.playbackEnabledFuture.await, gate.await]); + } finally { + this.gatedFrames.delete(gate); + } + } + if (this.interruptCount > this.segmentInterruptCount) { return; } } @@ -495,6 +521,7 @@ export class ParticipantAudioOutput extends AudioOutput { */ flush(): void { super.flush(); + this.segmentOpen = false; if (!this.pushedDuration) { return; @@ -522,10 +549,16 @@ export class ParticipantAudioOutput extends AudioOutput { } clearBuffer(): void { + this.interruptCount++; // Signal interruption even if no frame has been pushed yet, so a gated captureFrame can bail. if (!this.interruptedFuture.done) { this.interruptedFuture.resolve(); } + for (const gate of this.gatedFrames) { + if (!gate.done) { + gate.resolve(); + } + } } private async publishTrack(signal: AbortSignal) { From dfa48553e173238fa7e81c97a8aa7d20602c665b Mon Sep 17 00:00:00 2001 From: Toubat Date: Sun, 26 Jul 2026 22:36:12 -0700 Subject: [PATCH 2/2] fix(room_io): drop an interrupted reply's TTS backlog instead of playing it captureFrame only compared the segment's interrupt snapshot while the pause gate was closed, so the check was skipped once the gate reopened. cancelSpeechPause un-gates the sink to admit the next reply as soon as the handle is interrupted, but the interrupted reply's forwardAudio loop keeps running until its abort signal fires a turn later, draining the seconds of audio real TTS providers deliver ahead of realtime. Those frames reached the wire while the next reply's transcript was already streaming. The recorder's "never flushed" case now flushes before the follow-on turn: forwardAudio always flushes in its finally, and that flush is what tells the sink where the interrupted segment ends. Co-authored-by: Cursor --- .../interrupted-reply-backlog-on-the-wire.md | 13 +++ .../src/voice/recorder_io/recorder_io.test.ts | 4 + agents/src/voice/room_io/_output.ts | 14 ++- .../_output_interrupted_segment.test.ts | 104 ++++++++++++++++++ 4 files changed, 132 insertions(+), 3 deletions(-) create mode 100644 .changeset/interrupted-reply-backlog-on-the-wire.md create mode 100644 agents/src/voice/room_io/_output_interrupted_segment.test.ts diff --git a/.changeset/interrupted-reply-backlog-on-the-wire.md b/.changeset/interrupted-reply-backlog-on-the-wire.md new file mode 100644 index 000000000..a4c0e60e5 --- /dev/null +++ b/.changeset/interrupted-reply-backlog-on-the-wire.md @@ -0,0 +1,13 @@ +--- +'@livekit/agents': patch +--- + +Stop an interrupted reply's TTS backlog from reaching the wire + +`captureFrame` only compared a segment's interrupt snapshot while the pause gate was closed, so +the check was skipped entirely once the gate reopened. `cancelSpeechPause` un-gates the sink to +admit the next reply as soon as the handle is interrupted, but the interrupted reply's +`forwardAudio` loop keeps running until its abort signal fires an event loop turn later — and real +TTS providers hand it several seconds of audio ahead of realtime to drain in the meantime. Those +frames took the open-gate path straight to the wire, so the user kept hearing the barged-over +speech resume while the next reply's transcript was already streaming. diff --git a/agents/src/voice/recorder_io/recorder_io.test.ts b/agents/src/voice/recorder_io/recorder_io.test.ts index c6f61aa42..df409ebf3 100644 --- a/agents/src/voice/recorder_io/recorder_io.test.ts +++ b/agents/src/voice/recorder_io/recorder_io.test.ts @@ -1032,6 +1032,10 @@ describe('RecorderAudioOutput in front of a real ParticipantAudioOutput', () => expect(await settleOrStall(output.waitForPlayout(), 2000)).not.toBe('did not settle'); output.resume(); + // The flush `forwardAudio` runs in its `finally` is what ends the interrupted segment. Without + // it the sink cannot tell a follow-on turn apart from the interrupted turn's TTS backlog, which + // keeps arriving after the resume that admits the next reply. + output.flush(); await output.captureFrame(makeFrame(100, 24000)); output.flush(); diff --git a/agents/src/voice/room_io/_output.ts b/agents/src/voice/room_io/_output.ts index e5c8e9a0b..0c4d02c6d 100644 --- a/agents/src/voice/room_io/_output.ts +++ b/agents/src/voice/room_io/_output.ts @@ -455,9 +455,17 @@ export class ParticipantAudioOutput extends AudioOutput { this.gatedFrames.delete(gate); } } - if (this.interruptCount > this.segmentInterruptCount) { - return; - } + } + + // Tested on every frame, not only the ones that parked at the gate. `cancelSpeechPause` + // un-gates the sink to admit the next reply as soon as the handle is interrupted, but the + // interrupted reply's `forwardAudio` loop only stops an event loop turn later, when its + // abort signal fires — and real TTS hands it seconds of audio ahead of realtime to drain in + // the meantime. Those frames find the gate already open, so without this they reach the wire + // while the next reply's transcript is streaming. `forwardAudio` always flushes in its + // `finally`, which is what opens a fresh segment for the next reply. + if (this.interruptCount > this.segmentInterruptCount) { + return; } // Count the playback segment only after the pause/interrupt gate above. super.captureFrame diff --git a/agents/src/voice/room_io/_output_interrupted_segment.test.ts b/agents/src/voice/room_io/_output_interrupted_segment.test.ts new file mode 100644 index 000000000..c18ff40f0 --- /dev/null +++ b/agents/src/voice/room_io/_output_interrupted_segment.test.ts @@ -0,0 +1,104 @@ +// SPDX-FileCopyrightText: 2026 LiveKit, Inc. +// +// SPDX-License-Identifier: Apache-2.0 +import { AudioFrame, type Room, TrackPublishOptions, TrackSource } from '@livekit/rtc-node'; +import { describe, expect, it } from 'vitest'; +import { ParticipantAudioOutput } from './_output.js'; + +const SAMPLE_RATE = 24000; +const FRAME_MS = 20; + +/** `marker` identifies which reply produced the frame once it reaches the wire. */ +function frame(marker: number): AudioFrame { + const samples = (SAMPLE_RATE * FRAME_MS) / 1000; + const data = new Int16Array(samples); + data[0] = marker; + return new AudioFrame(data, SAMPLE_RATE, 1, samples); +} + +/** The real output; only track publishing is skipped (no LiveKit server in a unit test). */ +class TestOutput extends ParticipantAudioOutput { + constructor() { + super({} as Room, { + sampleRate: SAMPLE_RATE, + numChannels: 1, + trackPublishOptions: new TrackPublishOptions({ source: TrackSource.SOURCE_MICROPHONE }), + queueSizeMs: 100_000, + }); + (this as unknown as { startedFuture: { resolve: () => void } }).startedFuture.resolve(); + } + + override async start(): Promise {} +} + +/** Records the marker of every frame that reaches the wire. */ +function wireMarkers(out: ParticipantAudioOutput): number[] { + const source = ( + out as unknown as { audioSource: { captureFrame: (f: AudioFrame) => Promise } } + ).audioSource; + const original = source.captureFrame.bind(source); + const markers: number[] = []; + source.captureFrame = async (f: AudioFrame) => { + markers.push(f.data[0]!); + return original(f); + }; + return markers; +} + +describe('a committed barge-in with TTS audio still in flight', () => { + /** + * Real TTS delivers several seconds of audio ahead of realtime, so at the moment of a barge-in + * the interrupted reply's `forwardAudio` loop is still holding a backlog of already-synthesized + * frames. It stays alive for the whole of the reply task's + * `cancelAndWait(forwardTasks, REPLY_TASK_CANCEL_TIMEOUT)`, and only flushes in its `finally` + * after that — well after `cancelSpeechPause` has un-gated the sink for the next reply. + * + * `captureFrame` only consults `interruptCount` while the pause gate is closed, so every one of + * those backlog frames takes the open-gate path and reaches the wire while the next reply's + * transcript is already streaming. That is the audio/transcript desync. + */ + it('does not put the interrupted reply’s backlog on the wire', async () => { + const out = new TestOutput(); + const markers = wireMarkers(out); + + // Reply A is playing. + await out.captureFrame(frame(1)); + + // The user barges in: VAD pauses the sink, then the commit clears and un-gates it. + out.pause(); + out.clearBuffer(); + out.resume(); + + // Reply A's forwarding loop has not noticed the abort yet and drains its backlog. + await out.captureFrame(frame(1)); + await out.captureFrame(frame(1)); + + expect(markers.filter((m) => m === 1)).toHaveLength(1); + + await out.close(); + }); + + /** Control: once reply A's forwarding loop unwinds and flushes, reply B must be audible. */ + it('plays the next reply after the interrupted one flushes', async () => { + const out = new TestOutput(); + const markers = wireMarkers(out); + + await out.captureFrame(frame(1)); + + out.pause(); + out.clearBuffer(); + out.resume(); + + await out.captureFrame(frame(1)); + + // `forwardAudio` flushes in a `finally`, closing the interrupted segment. + out.flush(); + + await out.captureFrame(frame(2)); + await out.captureFrame(frame(2)); + + expect(markers.filter((m) => m === 2)).toHaveLength(2); + + await out.close(); + }); +});