diff --git a/.changeset/stale-interruption-pause-gate.md b/.changeset/stale-interruption-pause-gate.md new file mode 100644 index 000000000..93ea1b07b --- /dev/null +++ b/.changeset/stale-interruption-pause-gate.md @@ -0,0 +1,27 @@ +--- +'@livekit/agents': patch +--- + +Scope the audio pause gate to the segment being captured + +`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. + +The same snapshot is now consulted on every frame rather than only on frames that parked at a +closed 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 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 found the gate already open and +took the unchecked path to the wire. 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/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.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..0c4d02c6d 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,17 +431,43 @@ 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) { - return; + // 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); + } } } + // 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 // bumps playbackSegmentsCount; if a frame interrupted-while-paused bailed at the gate after // that bump, the count would strand ahead of playbackFinishedCount and the next @@ -495,6 +529,7 @@ export class ParticipantAudioOutput extends AudioOutput { */ flush(): void { super.flush(); + this.segmentOpen = false; if (!this.pushedDuration) { return; @@ -522,10 +557,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) { 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(); + }); +});