fix(voice): clear the audio sink before un-gating it for the next speech

.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.

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<void> | 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<void>,
+      };
+    };
+
+    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', () => {

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();
     }
   }

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<void> {}
+}
+
+/** 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<ReadableStream<AudioFrame>> {
+    let emitted = 0;
+    return new ReadableStream<AudioFrame>({
+      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<void> } }
+  ).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<void>((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<ReturnType<typeof makeHarness>>;
+
+/**
+ * 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);
+});