From 92d19c241f98f3f89c74ded54d36273cd5ed9603 Mon Sep 17 00:00:00 2001 From: Toubat Date: Mon, 27 Jul 2026 09:18:44 -0700 Subject: [PATCH] fix(voice): signal the interruption before un-gating the audio sink MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit cancelSpeechPause() reopens the sink's pause gate to admit the next speech, but the frames parked at that gate belong to the speech it just interrupted, and they are released first — so audio the user has already barged in over reaches the wire. Python gets the ordering for free by blocking until the interrupted generation finishes; here that await is raced against the interrupt (#1124) and returns before the reply task has run its own clearBuffer(). Co-authored-by: Cursor --- .changeset/clear-sink-before-resume.md | 12 + agents/src/voice/agent_activity.test.ts | 73 +++++ agents/src/voice/agent_activity.ts | 11 + ...rmed_interruption_pause_and_commit.test.ts | 254 ++++++++++++++++++ 4 files changed, 350 insertions(+) create mode 100644 .changeset/clear-sink-before-resume.md create mode 100644 agents/src/voice/confirmed_interruption_pause_and_commit.test.ts diff --git a/.changeset/clear-sink-before-resume.md b/.changeset/clear-sink-before-resume.md new file mode 100644 index 000000000..d348c8e73 --- /dev/null +++ b/.changeset/clear-sink-before-resume.md @@ -0,0 +1,12 @@ +--- +'@livekit/agents': patch +--- + +Signal the interruption before un-gating the audio sink + +`cancelSpeechPause()` reopened the sink's pause gate to admit the next speech, but the frames +parked at that gate belong to the speech it had just interrupted, and they are released first — so +audio the user has already barged in over reached the wire. Python gets the ordering for free by +blocking until the interrupted generation finishes; here that await is raced against the interrupt +(#1124) and returns before the reply task has run its own `clearBuffer()`. Clear the sink first +instead. diff --git a/agents/src/voice/agent_activity.test.ts b/agents/src/voice/agent_activity.test.ts index 5f0f5f86e..8ef725e28 100644 --- a/agents/src/voice/agent_activity.test.ts +++ b/agents/src/voice/agent_activity.test.ts @@ -409,6 +409,79 @@ describe('AgentActivity - mainTask', () => { expect(handle.interrupted).toBe(true); expect(fakeActivity.pausedSpeech).toBeUndefined(); }); + + /** + * Unit-level guard for the ordering measured end to end in + * `confirmed_interruption_pause_and_commit.test.ts`. It pins the cheap invariant that test cannot: + * the sink is only cleared when this call is what interrupted the paused speech. + */ + it('clears the sink before un-gating it, and only when it interrupted the paused speech', async () => { + const makeFixture = (handle: SpeechHandle) => { + const calls: string[] = []; + const audioOutput = { + canPause: true, + pause: vi.fn(() => { + calls.push('pause'); + }), + resume: vi.fn(() => { + calls.push('resume'); + }), + clearBuffer: vi.fn(() => { + calls.push('clearBuffer'); + }), + }; + const fakeActivity = { + cancelSpeechPauseTask: undefined as Promise | undefined, + falseInterruptionTimer: undefined as NodeJS.Timeout | undefined, + pausedSpeech: { handle, agentState: 'speaking', timeout: 2000 } as + | { handle: SpeechHandle; agentState: string; timeout: number } + | undefined, + _currentSpeech: handle, + logger: { debug: vi.fn(), info: vi.fn() }, + agentSession: { + sessionOptions: { + turnHandling: { + interruption: { resumeFalseInterruption: true, falseInterruptionTimeout: 2000 }, + }, + }, + output: { audio: audioOutput }, + }, + }; + const proto = AgentActivity.prototype as unknown as Record< + string, + (...args: never[]) => unknown + >; + return { + calls, + fakeActivity, + cancelSpeechPause: proto['cancelSpeechPause']!.bind(fakeActivity as never) as (options?: { + interrupt?: boolean; + }) => Promise, + }; + }; + + const interrupting = makeFixture( + (() => { + const h = SpeechHandle.create({ allowInterruptions: true }); + h._authorizeGeneration(); + return h; + })(), + ); + await raceTimeout(interrupting.cancelSpeechPause(), 2000); + expect(interrupting.calls).toEqual(['clearBuffer', 'resume']); + + // `interrupt: false` ends the pause without interrupting — the speech is meant to keep + // playing, so clearing its buffer would throw away audio that is still wanted. + const resuming = makeFixture( + (() => { + const h = SpeechHandle.create({ allowInterruptions: true }); + h._authorizeGeneration(); + return h; + })(), + ); + await raceTimeout(resuming.cancelSpeechPause({ interrupt: false }), 2000); + expect(resuming.calls).toEqual(['resume']); + }); }); describe('AgentActivity - speech completion', () => { diff --git a/agents/src/voice/agent_activity.ts b/agents/src/voice/agent_activity.ts index 12c244665..638c2c833 100644 --- a/agents/src/voice/agent_activity.ts +++ b/agents/src/voice/agent_activity.ts @@ -4614,12 +4614,14 @@ export class AgentActivity implements RecognitionHooks { return; } + let interruptedPausedSpeech = false; if ( interrupt && !this.pausedSpeech.handle.interrupted && this.pausedSpeech.handle.allowInterruptions ) { this.pausedSpeech.handle.interrupt(); + interruptedPausedSpeech = true; // ensure the generation is done — but only if a generation // was actually started. Must be raced against interrupt: an interrupted // paused speech may never mark its generation done, and an un-raced @@ -4634,6 +4636,15 @@ export class AgentActivity implements RecognitionHooks { const interruptionOptions = this.agentSession.sessionOptions.turnHandling.interruption; if (interruptionOptions.resumeFalseInterruption && this.agentSession.output.audio) { + // Frames of the speech just interrupted are parked at the sink's pause gate. Opening the + // gate is only meant to admit the *next* speech, but the parked frames are released first + // and audio the user has already barged in over reaches the wire. Python blocks here until + // the generation finishes, which gets it the same ordering; the await above is raced + // against the interrupt (#1124) and so returns immediately, before the interrupted reply + // task has run its own clearBuffer(). Signal the interruption first instead. + if (interruptedPausedSpeech) { + this.agentSession.output.audio.clearBuffer(); + } this.agentSession.output.audio.resume(); } } diff --git a/agents/src/voice/confirmed_interruption_pause_and_commit.test.ts b/agents/src/voice/confirmed_interruption_pause_and_commit.test.ts new file mode 100644 index 000000000..3a1b916a1 --- /dev/null +++ b/agents/src/voice/confirmed_interruption_pause_and_commit.test.ts @@ -0,0 +1,254 @@ +// 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 type { OverlappingSpeechEvent } from '../inference/interruption/types.js'; +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 = 40; +const FALSE_INTERRUPTION_TIMEOUT = 400; + +const sleep = (ms: number) => new Promise((resolve) => setTimeout(resolve, ms)); + +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 verdict shape `AudioRecognition` forwards on `bargein_detected`. */ +function bargeInVerdict(overlapStartedAt: number): OverlappingSpeechEvent { + return { + type: 'overlapping_speech', + detectedAt: Date.now(), + isInterruption: true, + overlapStartedAt, + totalDurationInS: 0.1, + predictionDurationInS: 0.05, + detectionDelayInS: 0.2, + probability: 0.99, + numRequests: 1, + }; +} + +/** 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 spaced out in time so a barge-in can land mid-reply. */ +class FrameAgent extends Agent { + produced = 0; + onFrame?: (produced: number) => void; + + constructor() { + super({ instructions: 'test' }); + } + + async ttsNode(): Promise> { + let emitted = 0; + return new ReadableStream({ + pull: async (controller) => { + if (emitted >= FRAMES_PER_REPLY) { + controller.close(); + return; + } + if (emitted > 0) await sleep(15); + emitted++; + controller.enqueue(frame()); + this.produced++; + this.onFrame?.(this.produced); + }, + }); + } +} + +async function makeHarness() { + const session = new AgentSession({ + llm: new FakeLLM([{ input: 'one', content: 'first reply' }]), + aecWarmupDuration: null, + turnHandling: { interruption: { falseInterruptionTimeout: FALSE_INTERRUPTION_TIMEOUT } }, + }); + + const out = new TestParticipantAudioOutput(); + session.output.audio = out; + + // Frame accounting at the boundary that matters: 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, + agent, + falseInterruptions, + delivered: () => delivered, + paused: () => !gate.playbackEnabledFuture.done, + waitForProduced, + activity: () => session._activity!, + async close() { + await session.close(); + await out.close(); + }, + }; +} + +type Harness = Awaited>; + +/** + * Drives a reply up to the moment the adaptive interruption model rules the overlap a genuine + * barge-in, then lets the user fall silent so the false-interruption timer is armed. + * + * Returns the frame count at the AudioSource taken just before the verdict, so callers can + * measure what reaches the wire after it. + */ +async function bargeInMidReply(h: Harness) { + const handle = h.session.generateReply({ userInput: 'one' }); + await h.waitForProduced(3); + + // The user starts talking over the agent: VAD parks the reply at the pause gate. + const overlapStartedAt = Date.now(); + h.activity().onStartOfSpeech(vadEvent(VADEventType.START_OF_SPEECH)); + h.activity().onVADInferenceDone(vadEvent(VADEventType.INFERENCE_DONE, 600)); + expect(h.paused()).toBe(true); + + // Let the TTS keep producing so frames are genuinely parked at the gate when the verdict + // lands — those parked frames are the ones that can escape. + await sleep(60); + + const deliveredBeforeVerdict = h.delivered(); + h.activity().onInterruption(bargeInVerdict(overlapStartedAt)); + // The user stops talking, which is what arms the false-interruption timer. + h.activity().onEndOfSpeech(vadEvent(VADEventType.END_OF_SPEECH, 600, 200)); + + return { handle, deliveredBeforeVerdict }; +} + +describe('confirmed barge-in: pause, then commit on the transcript', () => { + initializeLogger({ pretty: false, level: 'silent' }); + + /** + * Parity with Python's `AgentActivity.on_interruption` (`agent_activity.py`), which does + * restore → interrupt-by-audio-activity → `_on_end_of_agent_speech` and stops. The model's + * verdict alone only *pauses* the reply; committing the user's turn needs an STT final + * transcript. When none arrives within `falseInterruptionTimeout`, the false-interruption + * timer puts the speech back rather than leaving the user in dead air, and the session + * reports that with `agent_false_interruption(resumed: true)`. + */ + it('resumes the paused reply when no transcript follows the verdict', async () => { + const h = await makeHarness(); + try { + const { handle, deliveredBeforeVerdict } = await bargeInMidReply(h); + + await sleep(FALSE_INTERRUPTION_TIMEOUT + 600); + await handle.waitForPlayout(); + + // The verdict did not commit the turn, so the reply resumes where it stopped. + expect(handle.interrupted).toBe(false); + expect(h.falseInterruptions).toEqual([true]); + expect(h.delivered() - deliveredBeforeVerdict).toBeGreaterThan(0); + expect(h.delivered()).toBe(FRAMES_PER_REPLY); + } finally { + await h.close(); + } + }, 30000); + + /** + * The other half of the same flow, and the one the sink ordering in `cancelSpeechPause()` + * exists for. The final transcript arrives, so `onFinalTranscript` commits the interruption + * through `cancelSpeechPause()`, which un-gates the output so the *next* speech can be + * admitted. Frames of the reply just interrupted are still parked at that gate; unless the + * interruption is signalled to the sink first, they are released before the interrupted + * reply task reaches its own `clearBuffer()` and audio the user has already barged in over + * reaches the wire. + * + * Measured at the AudioSource boundary, not by asserting a call order. + */ + it('delivers no further audio once the final transcript commits the barge-in', async () => { + const h = await makeHarness(); + try { + const { handle, deliveredBeforeVerdict } = await bargeInMidReply(h); + + // STT finalizes what the user said, well inside the false-interruption timeout. + h.activity().onFinalTranscript(sttFinal('stop please'), false); + + await handle.waitForPlayout(); + // Well past the timeout: nothing may resume the speech either. + await sleep(FALSE_INTERRUPTION_TIMEOUT + 300); + + expect(h.delivered() - deliveredBeforeVerdict).toBe(0); + expect(handle.interrupted).toBe(true); + expect(h.falseInterruptions).toEqual([]); + // The reply was cut short: this is a barge-in, not a completed turn. + expect(h.delivered()).toBeLessThan(FRAMES_PER_REPLY); + } finally { + await h.close(); + } + }, 30000); +});