From 113ff97faddffcda2f5ff7ea5b6ee71a45511f74 Mon Sep 17 00:00:00 2001 From: Toubat Date: Fri, 24 Jul 2026 18:21:26 -0700 Subject: [PATCH 1/5] fix(recorder): register recorder segment before forwarding frames downstream MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit RecorderAudioOutput forwarded each frame to the wrapped output before calling super.captureFrame, so the recorder had not yet counted its own segment when the frame left. A downstream output that parks frames — ParticipantAudioOutput holds them at its pause gate — can emit an interrupted finish while a frame is parked. That finish arrived while the recorder owned zero segments, so the base AudioOutput discarded it as surplus. The segment registered a moment later then had no finish left to settle it and its waitForPlayout never resolved, stalling the speech scheduler for the rest of the session. Invert the ordering so the segment is registered first, and track playout state per segment instead of in global counters and a single shared frame buffer. Each finish is now attributed to the segment it belongs to, a finish that arrives with no segment to match stays queued instead of being dropped, and waitForPlayout resolves with its own segment's event rather than whatever the base class last recorded. Co-authored-by: Cursor --- .../fix-recorder-capture-finish-race.md | 8 + .../src/voice/recorder_io/recorder_io.test.ts | 401 +++++++++++++++++- agents/src/voice/recorder_io/recorder_io.ts | 244 ++++++++--- 3 files changed, 604 insertions(+), 49 deletions(-) create mode 100644 .changeset/fix-recorder-capture-finish-race.md diff --git a/.changeset/fix-recorder-capture-finish-race.md b/.changeset/fix-recorder-capture-finish-race.md new file mode 100644 index 000000000..0d152a79e --- /dev/null +++ b/.changeset/fix-recorder-capture-finish-race.md @@ -0,0 +1,8 @@ +--- +'@livekit/agents': patch +--- + +Fix a deadlock where a recorder-wrapped audio output could leave `waitForPlayout` stranded when +an interrupt arrived before the recorder had registered its segment. `RecorderAudioOutput` now +registers its own segment before forwarding a frame downstream, and attributes each playback +finish to the segment it belongs to instead of relying on a global counter. diff --git a/agents/src/voice/recorder_io/recorder_io.test.ts b/agents/src/voice/recorder_io/recorder_io.test.ts index 27a5f7448..b24f88dd7 100644 --- a/agents/src/voice/recorder_io/recorder_io.test.ts +++ b/agents/src/voice/recorder_io/recorder_io.test.ts @@ -10,7 +10,7 @@ import { initializeLogger } from '../../log.js'; import { type StreamChannel, createStreamChannel } from '../../stream/stream_channel.js'; import { Future, isWritableStreamClosedError } from '../../utils.js'; import type { AgentSession } from '../agent_session.js'; -import { AudioInput, AudioOutput } from '../io.js'; +import { AudioInput, AudioOutput, type PlaybackFinishedEvent } from '../io.js'; import { RecorderIO } from './recorder_io.js'; class FakeAudioInput extends AudioInput { @@ -51,6 +51,176 @@ class WaitAwareAudioOutput extends FakeAudioOutput { } } +class FinishDuringCaptureOutput extends AudioOutput { + private captures = 0; + + constructor() { + super(24000); + } + + async captureFrame(frame: AudioFrame): Promise { + await super.captureFrame(frame); + this.captures++; + if (this.captures === 1) { + this.onPlaybackFinished({ playbackPosition: 0, interrupted: true }); + } + } + + clearBuffer(): void {} +} + +class DroppingAudioOutput extends AudioOutput { + constructor() { + super(24000); + } + + async captureFrame(_frame: AudioFrame): Promise {} + + clearBuffer(): void {} +} + +class PreviousFinishThenDropOutput extends AudioOutput { + private captures = 0; + onPreviousFinishForwarded?: () => void; + + constructor() { + super(24000); + } + + async captureFrame(frame: AudioFrame): Promise { + this.captures++; + if (this.captures === 1) { + await super.captureFrame(frame); + return; + } + + this.onPlaybackFinished({ playbackPosition: 0, interrupted: true }); + this.onPreviousFinishForwarded?.(); + } + + clearBuffer(): void {} +} + +class FinishDuringLaterFrameOutput extends AudioOutput { + private captures = 0; + onFinishForwarded?: () => void; + + constructor() { + super(24000); + } + + async captureFrame(frame: AudioFrame): Promise { + await super.captureFrame(frame); + this.captures++; + if (this.captures === 2) { + this.onPlaybackFinished({ playbackPosition: 0.04, interrupted: true }); + this.onFinishForwarded?.(); + } + } + + clearBuffer(): void {} +} + +class DropThenFinishOutput extends AudioOutput { + private captures = 0; + + constructor() { + super(24000); + } + + async captureFrame(frame: AudioFrame): Promise { + this.captures++; + if (this.captures === 1) { + return; + } + + await super.captureFrame(frame); + this.onPlaybackFinished({ + playbackPosition: 0, + interrupted: true, + synchronizedTranscript: 'accepted-second-segment', + }); + } + + clearBuffer(): void {} +} + +class BlockingSecondCaptureOutput extends AudioOutput { + readonly secondCaptureStarted = new Future(); + private readonly continueSecondCapture = new Future(); + private captures = 0; + + constructor() { + super(24000); + } + + async captureFrame(frame: AudioFrame): Promise { + this.captures++; + if (this.captures === 2) { + this.secondCaptureStarted.resolve(); + await this.continueSecondCapture.await; + } + await super.captureFrame(frame); + } + + releaseSecondCapture(): void { + this.continueSecondCapture.resolve(); + } + + finishSegment(): void { + this.onPlaybackFinished({ playbackPosition: 0, interrupted: true }); + } + + clearBuffer(): void {} +} + +class RejectingAudioOutput extends AudioOutput { + constructor() { + super(24000); + } + + async captureFrame(_frame: AudioFrame): Promise { + throw new Error('capture rejected'); + } + + clearBuffer(): void {} +} + +class CountThenRejectOutput extends AudioOutput { + constructor() { + super(24000); + } + + async captureFrame(frame: AudioFrame): Promise { + await super.captureFrame(frame); + throw new Error('capture rejected after counting'); + } + + clearBuffer(): void {} +} + +class AcceptThenCountRejectOutput extends AudioOutput { + private captures = 0; + + constructor() { + super(24000); + } + + async captureFrame(frame: AudioFrame): Promise { + await super.captureFrame(frame); + this.captures++; + if (this.captures === 2) { + throw new Error('second capture rejected after counting'); + } + } + + finishFirstSegment(): void { + this.onPlaybackFinished({ playbackPosition: 0, interrupted: false }); + } + + clearBuffer(): void {} +} + function makeFrame(durationMs: number, sampleRate = 48000, channels = 1): AudioFrame { const samplesPerChannel = Math.floor((durationMs / 1000) * sampleRate); return new AudioFrame( @@ -139,6 +309,235 @@ describe('RecorderIO close', () => { }); describe('RecorderAudioOutput', () => { + it('does not lose a playback finish emitted during first-frame capture', async () => { + const recorder = new RecorderIO({ agentSession: {} as AgentSession }); + const output = recorder.recordOutput(new FinishDuringCaptureOutput()); + + await output.captureFrame(makeFrame(20, 24000)); + output.flush(); + const result = await Promise.race([ + output.waitForPlayout().then(() => 'resolved' as const), + new Promise<'timeout'>((resolve) => setTimeout(() => resolve('timeout'), 100)), + ]); + + expect(result).toBe('resolved'); + await recorder.close(); + }); + + it('keeps an early-finished segment active until its remaining frames are flushed', async () => { + const recorder = new RecorderIO({ agentSession: {} as AgentSession }); + const output = recorder.recordOutput(new FinishDuringCaptureOutput()); + + await output.captureFrame(makeFrame(20, 24000)); + await output.captureFrame(makeFrame(20, 24000)); + output.flush(); + const result = await Promise.race([ + output.waitForPlayout().then(() => 'resolved' as const), + new Promise<'timeout'>((resolve) => setTimeout(() => resolve('timeout'), 100)), + ]); + + expect(result).toBe('resolved'); + await recorder.close(); + }); + + it('settles a recorder segment when the wrapped output drops its first frame', async () => { + const recorder = new RecorderIO({ agentSession: {} as AgentSession }); + const output = recorder.recordOutput(new DroppingAudioOutput()); + + await output.captureFrame(makeFrame(20, 24000)); + output.flush(); + const result = await Promise.race([ + output.waitForPlayout(), + new Promise<'timeout'>((resolve) => setTimeout(() => resolve('timeout'), 100)), + ]); + + expect(result).toEqual({ playbackPosition: 0, interrupted: true }); + await recorder.close(); + }); + + it('preserves an older finish while reconciling a dropped overlapping segment', async () => { + const recorder = new RecorderIO({ agentSession: {} as AgentSession }); + const downstream = new PreviousFinishThenDropOutput(); + const output = recorder.recordOutput(downstream); + const finishes: PlaybackFinishedEvent[] = []; + output.on(AudioOutput.EVENT_PLAYBACK_FINISHED, (event: PlaybackFinishedEvent) => { + finishes.push(event); + }); + downstream.onPreviousFinishForwarded = () => { + expect(finishes).toHaveLength(1); + }; + + await output.captureFrame(makeFrame(20, 24000)); + output.flush(); + await output.captureFrame(makeFrame(20, 24000)); + output.flush(); + const event = await output.waitForPlayout(); + + expect(finishes).toHaveLength(2); + expect(event).toEqual({ playbackPosition: 0, interrupted: true }); + await recorder.close(); + }); + + it('defers a current-segment finish until a later frame capture is recorded', async () => { + const recorder = new RecorderIO({ agentSession: {} as AgentSession }); + const downstream = new FinishDuringLaterFrameOutput(); + const output = recorder.recordOutput(downstream); + const finishes: PlaybackFinishedEvent[] = []; + output.on(AudioOutput.EVENT_PLAYBACK_FINISHED, (event: PlaybackFinishedEvent) => { + finishes.push(event); + }); + downstream.onFinishForwarded = () => { + expect(finishes).toHaveLength(0); + }; + + await output.captureFrame(makeFrame(20, 24000)); + await output.captureFrame(makeFrame(20, 24000)); + output.flush(); + const event = await output.waitForPlayout(); + + expect(finishes).toHaveLength(1); + expect(event.interrupted).toBe(true); + await recorder.close(); + }); + + it('settles a dropped older segment before applying a real finish to the next segment', async () => { + const recorder = new RecorderIO({ agentSession: {} as AgentSession }); + const output = recorder.recordOutput(new DropThenFinishOutput()); + const finishes: PlaybackFinishedEvent[] = []; + output.on(AudioOutput.EVENT_PLAYBACK_FINISHED, (event: PlaybackFinishedEvent) => { + finishes.push(event); + }); + + await output.captureFrame(makeFrame(20, 24000)); + output.flush(); + await output.captureFrame(makeFrame(20, 24000)); + output.flush(); + const result = await Promise.race([ + output.waitForPlayout(), + new Promise<'timeout'>((resolve) => setTimeout(() => resolve('timeout'), 100)), + ]); + + expect(result).not.toBe('timeout'); + expect(finishes).toEqual([ + { playbackPosition: 0, interrupted: true }, + { + playbackPosition: 0, + interrupted: true, + synchronizedTranscript: 'accepted-second-segment', + }, + ]); + await recorder.close(); + }); + + it('does not reconcile a segment while its downstream capture is still in flight', async () => { + const recorder = new RecorderIO({ agentSession: {} as AgentSession }); + const downstream = new BlockingSecondCaptureOutput(); + const output = recorder.recordOutput(downstream); + const finishes: PlaybackFinishedEvent[] = []; + output.on(AudioOutput.EVENT_PLAYBACK_FINISHED, (event: PlaybackFinishedEvent) => { + finishes.push(event); + }); + + await output.captureFrame(makeFrame(20, 24000)); + output.flush(); + const waitForFirstSegment = output.waitForPlayout(); + const captureSecondSegment = output.captureFrame(makeFrame(20, 24000)); + await downstream.secondCaptureStarted.await; + + downstream.finishSegment(); + await waitForFirstSegment; + expect(finishes).toHaveLength(1); + + downstream.releaseSecondCapture(); + await captureSecondSegment; + output.flush(); + downstream.finishSegment(); + await output.waitForPlayout(); + + expect(finishes).toHaveLength(2); + await recorder.close(); + }); + + it('keeps later segment audio when an older overlapping segment finishes', async () => { + const writes: AudioFrame[][] = []; + const recorder = new RecorderIO({ agentSession: {} as AgentSession }); + const recorderState = recorder as unknown as { + started: boolean; + writeCb: (buf: AudioFrame[]) => void; + }; + recorderState.started = true; + recorderState.writeCb = (buf) => writes.push(buf); + const downstream = new FakeAudioOutput(); + const output = recorder.recordOutput(downstream); + + await output.captureFrame(makeFrame(1)); + output.flush(); + await output.captureFrame(makeFrame(1)); + output.flush(); + await new Promise((resolve) => setTimeout(resolve, 5)); + + downstream.onPlaybackFinished({ playbackPosition: 0.001, interrupted: false }); + downstream.onPlaybackFinished({ playbackPosition: 0.001, interrupted: false }); + + expect(writes).toHaveLength(2); + recorderState.started = false; + await recorder.close(); + }); + + it('settles recorder state when downstream capture rejects', async () => { + const recorder = new RecorderIO({ agentSession: {} as AgentSession }); + const output = recorder.recordOutput(new RejectingAudioOutput()); + + await expect(output.captureFrame(makeFrame(20, 24000))).rejects.toThrow('capture rejected'); + const event = await output.waitForPlayout(); + + expect(event).toEqual({ playbackPosition: 0, interrupted: true }); + await recorder.close(); + }); + + it('settles both outputs when downstream capture rejects after counting', async () => { + const recorder = new RecorderIO({ agentSession: {} as AgentSession }); + const output = recorder.recordOutput(new CountThenRejectOutput()); + + await expect(output.captureFrame(makeFrame(20, 24000))).rejects.toThrow( + 'capture rejected after counting', + ); + const result = await Promise.race([ + output.waitForPlayout(), + new Promise<'timeout'>((resolve) => setTimeout(() => resolve('timeout'), 100)), + ]); + + expect(result).toEqual({ playbackPosition: 0, interrupted: true }); + await recorder.close(); + }); + + it('settles a counted failed segment only after its older segment finishes', async () => { + const recorder = new RecorderIO({ agentSession: {} as AgentSession }); + const downstream = new AcceptThenCountRejectOutput(); + const output = recorder.recordOutput(downstream); + const finishes: PlaybackFinishedEvent[] = []; + output.on(AudioOutput.EVENT_PLAYBACK_FINISHED, (event: PlaybackFinishedEvent) => { + finishes.push(event); + }); + + await output.captureFrame(makeFrame(20, 24000)); + output.flush(); + const waitForFirstSegment = output.waitForPlayout(); + await expect(output.captureFrame(makeFrame(20, 24000))).rejects.toThrow( + 'second capture rejected after counting', + ); + downstream.finishFirstSegment(); + const firstEvent = await waitForFirstSegment; + await output.waitForPlayout(); + + expect(firstEvent).toEqual({ playbackPosition: 0, interrupted: false }); + expect(finishes).toEqual([ + { playbackPosition: 0, interrupted: false }, + { playbackPosition: 0, interrupted: true }, + ]); + await recorder.close(); + }); + it('snapshots its segment before delegating the playout wait', async () => { const recorder = new RecorderIO({ agentSession: {} as AgentSession }); const downstream = new WaitAwareAudioOutput(); diff --git a/agents/src/voice/recorder_io/recorder_io.ts b/agents/src/voice/recorder_io/recorder_io.ts index 63911598c..cc20e2ff1 100644 --- a/agents/src/voice/recorder_io/recorder_io.ts +++ b/agents/src/voice/recorder_io/recorder_io.ts @@ -569,19 +569,29 @@ class RecorderAudioInput extends AudioInput { } } +interface RecorderOutputSegment { + frames: AudioFrame[]; + acceptedDownstream: boolean; + captureFailed: boolean; + capturesInFlight: number; + finishRequested: boolean; + flushed: boolean; + playbackEvent?: PlaybackFinishedEvent; + speechStartTime?: number; + currentPauseStart?: number; + pauseWallTimes: Array<[number, number]>; +} + class RecorderAudioOutput extends AudioOutput { private recorderIO: RecorderIO; private writeFn: (buf: AudioFrame[]) => void; - private accFrames: AudioFrame[] = []; + private segments: RecorderOutputSegment[] = []; + private currentSegment?: RecorderOutputSegment; + private deferredFinishes: PlaybackFinishedEvent[] = []; private _startedWallTime?: number; private _logger = log(); _lastSpeechEndTime?: number; - private _lastSpeechStartTime?: number; - - // Pause tracking - private currentPauseStart?: number; - private pauseWallTimes: Array<[number, number]> = []; // [start, end] pairs constructor( recorderIO: RecorderIO, @@ -598,12 +608,13 @@ class RecorderAudioOutput extends AudioOutput { } get hasPendingData(): boolean { - return this.accFrames.length > 0; + return this.segments.some((segment) => segment.frames.length > 0); } pause(): void { - if (this.currentPauseStart === undefined && this.recorderIO.recording) { - this.currentPauseStart = Date.now(); + const segment = this.segments[0]; + if (segment && segment.currentPauseStart === undefined && this.recorderIO.recording) { + segment.currentPauseStart = Date.now(); } if (this.nextInChain) { @@ -615,9 +626,10 @@ class RecorderAudioOutput extends AudioOutput { * Resume playback and record the pause interval */ resume(): void { - if (this.currentPauseStart !== undefined && this.recorderIO.recording) { - this.pauseWallTimes.push([this.currentPauseStart, Date.now()]); - this.currentPauseStart = undefined; + const segment = this.segments[0]; + if (segment?.currentPauseStart !== undefined && this.recorderIO.recording) { + segment.pauseWallTimes.push([segment.currentPauseStart, Date.now()]); + segment.currentPauseStart = undefined; } if (this.nextInChain) { @@ -625,19 +637,88 @@ class RecorderAudioOutput extends AudioOutput { } } - private resetPauseState(): void { - this.currentPauseStart = undefined; - this.pauseWallTimes = []; + onPlaybackFinished(options: PlaybackFinishedEvent): void { + this.deferredFinishes.push(options); + this.drainFinishes(); } - onPlaybackFinished(options: PlaybackFinishedEvent): void { - const finishTime = this.currentPauseStart ?? Date.now(); + /** + * Settle segments in capture order against the finishes the downstream output has sent. + * + * Segments are settled oldest-first so a finish is always attributed to the segment it + * belongs to. A finish that arrives with nothing to attribute it to yet stays queued in + * `deferredFinishes` rather than being forwarded (and dropped) immediately. + */ + private drainFinishes(): void { + while (this.segments.length > 0) { + const segment = this.segments[0]!; + if (segment.capturesInFlight > 0) { + return; + } + + if (!segment.acceptedDownstream) { + // A segment the downstream output never counted will never receive a real finish, so + // we synthesize one. Waiting for the flush first is a stricter precondition than the + // old code had: it guarantees no further frames can join this segment, so we cannot + // settle it while it is still growing. All in-tree callers satisfy it — `generation.ts` + // flushes in a `finally`, the interrupted path in `agent_activity.ts` awaits + // `cancelAndWait` on the forward tasks before waiting for playout, and + // `RecorderIO.close()` bounds its own wait with `closePlayoutFlushTimeoutMs`. + if (!segment.flushed) { + return; + } + this.finishSegment(segment, { playbackPosition: 0, interrupted: true }); + continue; + } + + if (!segment.flushed) { + return; + } + + const event = this.deferredFinishes.shift(); + if (event) { + this.finishSegment(segment, event); + continue; + } + + if (segment.captureFailed && !segment.finishRequested && this.nextInChain) { + // Reaching down to the wrapped output looks like a layering inversion, and normally it + // would be. This branch only runs when the downstream output already counted the + // segment (`acceptedDownstream`) and our capture then threw, so the sink is holding a + // segment it will never be told about and its own `waitForPlayout` would hang. Nobody + // else can unstick it: the frame never reached the sink's completion path. Guarded by + // `finishRequested` so we ask exactly once. + segment.finishRequested = true; + this.nextInChain.onPlaybackFinished({ playbackPosition: 0, interrupted: true }); + } + return; + } + + // No segments left to attribute these to. Forwarding them to `super.onPlaybackFinished` + // would only trip its "more finishes than segments" warning, so drop them and say so at + // debug level instead of polluting the logs with a warning we caused. + const leftovers = this.deferredFinishes.splice(0); + if (leftovers.length > 0) { + this._logger.debug( + { count: leftovers.length }, + 'discarding playback finishes with no matching recorder segment', + ); + } + } + + private finishSegment(segment: RecorderOutputSegment, options: PlaybackFinishedEvent): void { + this.segments.shift(); + if (this.currentSegment === segment) { + this.currentSegment = undefined; + } + + const finishTime = segment.currentPauseStart ?? Date.now(); const trailingSilenceDuration = Math.max(0, Date.now() - finishTime); // Convert playbackPosition from seconds to ms for internal calculations let playbackPosition = options.playbackPosition * 1000; - if (this._lastSpeechStartTime === undefined) { + if (segment.speechStartTime === undefined) { this._logger.warn( { finishTime, @@ -652,25 +733,24 @@ class RecorderAudioOutput extends AudioOutput { // Clamp playbackPosition to actual elapsed time (all in ms) playbackPosition = Math.max( 0, - Math.min(finishTime - (this._lastSpeechStartTime ?? 0), playbackPosition), + Math.min(finishTime - (segment.speechStartTime ?? 0), playbackPosition), ); // Convert back to seconds for the event - super.onPlaybackFinished({ ...options, playbackPosition: playbackPosition / 1000 }); + segment.playbackEvent = { ...options, playbackPosition: playbackPosition / 1000 }; + super.onPlaybackFinished(segment.playbackEvent); if (!this.recorderIO.recording) { return; } - if (this.currentPauseStart !== undefined) { - this.pauseWallTimes.push([this.currentPauseStart, finishTime]); - this.currentPauseStart = undefined; + if (segment.currentPauseStart !== undefined) { + segment.pauseWallTimes.push([segment.currentPauseStart, finishTime]); + segment.currentPauseStart = undefined; } - if (this.accFrames.length === 0) { - this.resetPauseState(); + if (segment.frames.length === 0) { this._lastSpeechEndTime = Date.now(); - this._lastSpeechStartTime = undefined; return; } @@ -678,15 +758,15 @@ class RecorderAudioOutput extends AudioOutput { const pauseEvents: Array<[number, number]> = []; let playbackStartTime = finishTime - playbackPosition; - if (this.pauseWallTimes.length > 0) { - const totalPauseDuration = this.pauseWallTimes.reduce( + if (segment.pauseWallTimes.length > 0) { + const totalPauseDuration = segment.pauseWallTimes.reduce( (sum, [start, end]) => sum + (end - start), 0, ); playbackStartTime = finishTime - playbackPosition - totalPauseDuration; let accumulatedPause = 0; - for (const [pauseStart, pauseEnd] of this.pauseWallTimes) { + for (const [pauseStart, pauseEnd] of segment.pauseWallTimes) { let position = pauseStart - playbackStartTime - accumulatedPause; const duration = pauseEnd - pauseStart; position = Math.max(0, Math.min(position, playbackPosition)); @@ -697,13 +777,13 @@ class RecorderAudioOutput extends AudioOutput { const buf: AudioFrame[] = []; let accDur = 0; - const sampleRate = this.accFrames[0]!.sampleRate; - const numChannels = this.accFrames[0]!.channels; + const sampleRate = segment.frames[0]!.sampleRate; + const numChannels = segment.frames[0]!.channels; let pauseIdx = 0; let shouldBreak = false; - for (const frame of this.accFrames) { + for (const frame of segment.frames) { let currentFrame = frame; const frameDuration = (frame.samplesPerChannel / frame.sampleRate) * 1000; @@ -764,46 +844,114 @@ class RecorderAudioOutput extends AudioOutput { this.writeFn(filteredBuf); } - this.accFrames = []; - this.resetPauseState(); this._lastSpeechEndTime = Date.now(); - this._lastSpeechStartTime = undefined; } async captureFrame(frame: AudioFrame): Promise { - if (this.nextInChain) { - await this.nextInChain.captureFrame(frame); - } + // Register our own segment BEFORE handing the frame downstream. A downstream output may + // park this frame (ParticipantAudioOutput holds frames at its pause gate) and emit an + // interrupted finish while it is parked. If we had not counted the segment yet, that + // finish would arrive while we own zero segments and `AudioOutput.onPlaybackFinished` + // would discard it as surplus — leaving the segment we register afterwards with no + // finish left to settle it, and `waitForPlayout` stuck forever. + const capturedBefore = this.capturedPlayoutSegments; + const capture = super.captureFrame(frame); + const startedNewSegment = this.capturedPlayoutSegments > capturedBefore; + let segment = this.currentSegment; + if (startedNewSegment) { + segment = { + frames: [], + acceptedDownstream: this.nextInChain === undefined, + captureFailed: false, + capturesInFlight: 0, + finishRequested: false, + flushed: false, + pauseWallTimes: [], + }; + this.segments.push(segment); + this.currentSegment = segment; + } + if (!segment) { + throw new Error('recorder capture has no active segment'); + } + + const downstreamCapturedBefore = this.nextInChain?.capturedPlayoutSegments ?? 0; + segment.capturesInFlight++; + let captureCompleted = false; + try { + await capture; + if (this.nextInChain) { + await this.nextInChain.captureFrame(frame); + if (this.nextInChain.capturedPlayoutSegments > downstreamCapturedBefore) { + segment.acceptedDownstream = true; + } + } - await super.captureFrame(frame); + if (this.recorderIO.recording) { + segment.frames.push(frame); + } - if (this.recorderIO.recording) { - this.accFrames.push(frame); - } + if (this._startedWallTime === undefined) { + this._startedWallTime = Date.now(); + } - if (this._startedWallTime === undefined) { - this._startedWallTime = Date.now(); - } + if (segment.speechStartTime === undefined) { + segment.speechStartTime = Date.now(); + } - if (this._lastSpeechStartTime === undefined) { - this._lastSpeechStartTime = Date.now(); + captureCompleted = true; + } finally { + if (this.nextInChain && this.nextInChain.capturedPlayoutSegments > downstreamCapturedBefore) { + segment.acceptedDownstream = true; + } + if (!captureCompleted) { + segment.captureFailed = true; + segment.flushed = true; + if (this.currentSegment === segment) { + this.currentSegment = undefined; + } + } + segment.capturesInFlight--; + this.drainFinishes(); } } + /** + * Wait for the segment that is open at call time to finish playing. + * + * Unlike the base {@link AudioOutput}, this resolves with *that segment's* own + * `playbackEvent` rather than whatever `lastPlaybackEvent` happens to hold when the wait + * unblocks. With multiple segments in flight the base behavior can hand a caller another + * segment's event — e.g. report `interrupted: true` for a segment that played to completion. + * This is a deliberate divergence from the base class (and from Python, whose + * `voice/io.py` also returns the last event). Note the `playedOwnFrame` bookkeeping in + * `agent_activity.ts` exists precisely to work around stale events from waits like this one, + * so it is now partly redundant here; it is left in place because it still guards the other + * outputs. Giving the base class the same per-segment attribution — which would also fix + * `ParticipantAudioOutput` — is follow-up work. + */ async waitForPlayout(): Promise { + const targetSegment = this.segments[this.segments.length - 1]; const waitForRecorder = super.waitForPlayout(); if (this.nextInChain) { await this.nextInChain.waitForPlayout(); } - return waitForRecorder; + this.drainFinishes(); + const event = await waitForRecorder; + return targetSegment?.playbackEvent ?? event; } flush(): void { super.flush(); + if (this.currentSegment) { + this.currentSegment.flushed = true; + this.currentSegment = undefined; + } if (this.nextInChain) { this.nextInChain.flush(); } + this.drainFinishes(); } clearBuffer(): void { From 7741aa68cb71ae8df7dd6c0b2faa85a6a17e969f Mon Sep 17 00:00:00 2001 From: Toubat Date: Fri, 24 Jul 2026 18:29:29 -0700 Subject: [PATCH 2/5] fix(recorder): seal the open playout segment when the recorder closes MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit drainFinishes() will only settle a segment the downstream sink never accepted once that segment has been flushed, so it cannot settle one that is still growing. RecorderIO.close() was the one caller that could wait on a segment the turn never flushed: an interrupt that tore the turn down after the sink dropped the frame left the segment open, so close() burned the full CLOSE_PLAYOUT_FLUSH_TIMEOUT_MS and then warned about dropping audio that the synthetic zero-position finish would have truncated to nothing anyway. Mark the open segment flushed at the top of close(). Closing already guarantees no further frames can reach the output, which is exactly the condition the flush gate stands in for, so the segment can settle immediately. Segments with a capture still in flight are unaffected — drainFinishes() continues to hold them — and a segment the sink did accept still waits for its real finish. Co-authored-by: Cursor --- .../src/voice/recorder_io/recorder_io.test.ts | 36 ++++++++++++++++++- agents/src/voice/recorder_io/recorder_io.ts | 36 +++++++++++++++---- 2 files changed, 65 insertions(+), 7 deletions(-) diff --git a/agents/src/voice/recorder_io/recorder_io.test.ts b/agents/src/voice/recorder_io/recorder_io.test.ts index b24f88dd7..05da96af2 100644 --- a/agents/src/voice/recorder_io/recorder_io.test.ts +++ b/agents/src/voice/recorder_io/recorder_io.test.ts @@ -5,7 +5,7 @@ import { AudioFrame } from '@livekit/rtc-node'; import fs from 'node:fs'; import os from 'node:os'; import path from 'node:path'; -import { beforeAll, describe, expect, it } from 'vitest'; +import { beforeAll, describe, expect, it, vi } from 'vitest'; import { initializeLogger } from '../../log.js'; import { type StreamChannel, createStreamChannel } from '../../stream/stream_channel.js'; import { Future, isWritableStreamClosedError } from '../../utils.js'; @@ -286,6 +286,40 @@ describe('RecorderIO close', () => { expect(fs.existsSync(outputPath)).toBe(false); }, 15000); + it('settles a dropped, never-flushed segment on close without stalling or warning', async () => { + const dir = fs.mkdtempSync(path.join(os.tmpdir(), 'recorder-io-test-')); + const outputPath = path.join(dir, 'audio.ogg'); + const recorder = new RecorderIO({ agentSession: {} as AgentSession }); + recorder.recordInput(new FakeAudioInput()); + const outWrapped = recorder.recordOutput(new DroppingAudioOutput()); + await recorder.start(outputPath); + + // An interrupt tore the turn down before the segment was flushed, and the sink dropped the + // frame, so no real playbackFinished will ever arrive for it. + await outWrapped.captureFrame(makeFrame(200, 24000)); + expect(outWrapped.hasPendingData).toBe(true); + + const warnSpy = vi.spyOn( + (recorder as unknown as { logger: { warn: (...args: unknown[]) => void } }).logger, + 'warn', + ); + + const start = Date.now(); + await recorder.close(); + const elapsed = Date.now() - start; + + expect(outWrapped.hasPendingData).toBe(false); + expect(elapsed).toBeLessThan(1000); + expect( + warnSpy.mock.calls.some((args) => + args.some( + (arg) => + typeof arg === 'string' && arg.includes('closed before the last playback finished'), + ), + ), + ).toBe(false); + }, 15000); + it('flushes trailing input audio on close', async () => { const { recorder, input, inWrapped, outputPath } = makeRecorder(); await recorder.start(outputPath); diff --git a/agents/src/voice/recorder_io/recorder_io.ts b/agents/src/voice/recorder_io/recorder_io.ts index cc20e2ff1..c0fe43adb 100644 --- a/agents/src/voice/recorder_io/recorder_io.ts +++ b/agents/src/voice/recorder_io/recorder_io.ts @@ -113,6 +113,12 @@ export class RecorderIO { try { if (!this.started) return; + // No further frames can reach the output once we are closing, so seal the open segment. + // A segment the downstream output never accepted can then settle immediately instead of + // stalling teardown for the full flush timeout and warning about audio it was never + // going to keep. + this.outRecord?._sealOpenSegment(); + // On a force-interrupted shutdown, the session marks the speech done // before playout settles, so the playout finished event may still be in flight. // Give it a bounded window to land before fencing writers out. @@ -658,12 +664,13 @@ class RecorderAudioOutput extends AudioOutput { if (!segment.acceptedDownstream) { // A segment the downstream output never counted will never receive a real finish, so - // we synthesize one. Waiting for the flush first is a stricter precondition than the - // old code had: it guarantees no further frames can join this segment, so we cannot - // settle it while it is still growing. All in-tree callers satisfy it — `generation.ts` - // flushes in a `finally`, the interrupted path in `agent_activity.ts` awaits - // `cancelAndWait` on the forward tasks before waiting for playout, and - // `RecorderIO.close()` bounds its own wait with `closePlayoutFlushTimeoutMs`. + // we synthesize one. Requiring the flush first is a stricter precondition than the old + // code had: it guarantees no further frames can join this segment, so we cannot settle + // one that is still growing. Every path that waits on a segment reaches this state — + // `generation.ts` flushes in a `finally`, the interrupted path in `agent_activity.ts` + // awaits `cancelAndWait` on the forward tasks before waiting for playout, and + // `RecorderIO.close()` seals the open segment before it waits, since closing means no + // further frames are possible. if (!segment.flushed) { return; } @@ -941,6 +948,23 @@ class RecorderAudioOutput extends AudioOutput { return targetSegment?.playbackEvent ?? event; } + /** + * Mark the currently open segment as flushed because no more frames can arrive for it. + * + * Called by {@link RecorderIO.close}. Unlike {@link flush} this does not notify the base class + * or the wrapped output — closing is not a segment boundary they need to hear about, it just + * means our own segment can never grow again, which is the guarantee `drainFinishes` needs + * before it may settle a segment the downstream output never accepted. A segment with a + * capture still in flight is unaffected: `drainFinishes` continues to hold it. + */ + _sealOpenSegment(): void { + if (this.currentSegment) { + this.currentSegment.flushed = true; + this.currentSegment = undefined; + } + this.drainFinishes(); + } + flush(): void { super.flush(); if (this.currentSegment) { From e02173fde2328f4e99d586f1ae1eccf27f397348 Mon Sep 17 00:00:00 2001 From: Toubat Date: Fri, 24 Jul 2026 21:44:39 -0700 Subject: [PATCH 3/5] fix(recorder): timestamp segments at open and recover from capture rejections Two review findings on the segment-ordering fix: - `speechStartTime` was assigned only after the downstream capture resolved, so a finish landing while the sink parked the first frame clamped the segment's playback position to ~zero and truncated away every frame the sink had just reported as played (and to a negative window, if the pause was still open when the segment settled). Stamp the segment when it opens instead. The field is now always set, which makes the "playback finished before speech started" warning unreachable, so it goes away with it. - Registering our segment before forwarding means the base class has already latched its capture state by the time a downstream capture throws. We clear the failed segment, so a caller that caught the rejection and retried without an explicit flush hit `recorder capture has no active segment` forever. Release the latch on this output and on the wrapped one when we abandon the segment, so a retry opens a fresh segment on both sides rather than joining a segment we already declared finished or drifting the segment accounting. Co-authored-by: Cursor --- .../fix-recorder-capture-finish-race.md | 7 + agents/src/voice/io.ts | 13 ++ .../src/voice/recorder_io/recorder_io.test.ts | 164 ++++++++++++++++++ agents/src/voice/recorder_io/recorder_io.ts | 42 +++-- 4 files changed, 208 insertions(+), 18 deletions(-) diff --git a/.changeset/fix-recorder-capture-finish-race.md b/.changeset/fix-recorder-capture-finish-race.md index 0d152a79e..b205af63e 100644 --- a/.changeset/fix-recorder-capture-finish-race.md +++ b/.changeset/fix-recorder-capture-finish-race.md @@ -6,3 +6,10 @@ Fix a deadlock where a recorder-wrapped audio output could leave `waitForPlayout an interrupt arrived before the recorder had registered its segment. `RecorderAudioOutput` now registers its own segment before forwarding a frame downstream, and attributes each playback finish to the segment it belongs to instead of relying on a global counter. + +A recorded segment is also timestamped when it opens rather than when the wrapped output accepts +its first frame, so a finish that lands while that frame is parked no longer clamps the segment's +playback position to zero and drop the audio the sink reported as played. And a segment whose +downstream capture throws now releases the capture latch on both the recorder and the wrapped +output, so a caller that retries after a transient rejection is no longer rejected forever with +`recorder capture has no active segment`. diff --git a/agents/src/voice/io.ts b/agents/src/voice/io.ts index 3f9911246..134286d8c 100644 --- a/agents/src/voice/io.ts +++ b/agents/src/voice/io.ts @@ -215,6 +215,19 @@ export abstract class AudioOutput extends EventEmitter { this._capturing = false; } + /** + * Forget the segment currently being captured, without treating it as a flush boundary. + * + * For an output whose open segment was abandoned rather than flushed — e.g. a capture threw and + * the segment has already been reported finished. Without this the output keeps believing a + * segment is open, so the next `captureFrame` silently joins a segment that no longer exists + * instead of counting a new one. + * @internal + */ + abandonOpenSegment(): void { + this._capturing = false; + } + /** * Clear the buffer, stopping playback immediately */ diff --git a/agents/src/voice/recorder_io/recorder_io.test.ts b/agents/src/voice/recorder_io/recorder_io.test.ts index 05da96af2..27849e0cd 100644 --- a/agents/src/voice/recorder_io/recorder_io.test.ts +++ b/agents/src/voice/recorder_io/recorder_io.test.ts @@ -186,6 +186,93 @@ class RejectingAudioOutput extends AudioOutput { clearBuffer(): void {} } +/** + * Holds the first frame at a gate (like `ParticipantAudioOutput` does while paused) after + * counting it, and lets the test report a playback finish while the frame is still parked. + */ +class ParkFirstFrameOutput extends AudioOutput { + readonly frameParked = new Future(); + private readonly gate = new Future(); + private captures = 0; + + constructor() { + super(48000); + } + + async captureFrame(frame: AudioFrame): Promise { + this.captures++; + // Count before parking so the finish reported during the park isn't surplus downstream. + await super.captureFrame(frame); + if (this.captures === 1) { + this.frameParked.resolve(); + await this.gate.await; + } + } + + reportFinished(playbackPosition: number): void { + this.onPlaybackFinished({ playbackPosition, interrupted: true }); + } + + releaseGate(): void { + this.gate.resolve(); + } + + clearBuffer(): void {} +} + +// `interrupted: false` can only come from a real downstream finish: a segment the recorder has to +// settle on its own is always synthesized as interrupted. The transcript marker makes the +// attribution unambiguous. +const RETRIED_SEGMENT_FINISH: PlaybackFinishedEvent = { + playbackPosition: 0, + interrupted: false, + synchronizedTranscript: 'retried-segment', +}; + +class RejectFirstCaptureOutput extends AudioOutput { + private captures = 0; + + constructor() { + super(24000); + } + + async captureFrame(frame: AudioFrame): Promise { + this.captures++; + if (this.captures === 1) { + throw new Error('capture rejected'); + } + await super.captureFrame(frame); + } + + finishSegment(): void { + this.onPlaybackFinished(RETRIED_SEGMENT_FINISH); + } + + clearBuffer(): void {} +} + +class CountThenRejectFirstCaptureOutput extends AudioOutput { + private captures = 0; + + constructor() { + super(24000); + } + + async captureFrame(frame: AudioFrame): Promise { + await super.captureFrame(frame); + this.captures++; + if (this.captures === 1) { + throw new Error('capture rejected after counting'); + } + } + + finishSegment(): void { + this.onPlaybackFinished(RETRIED_SEGMENT_FINISH); + } + + clearBuffer(): void {} +} + class CountThenRejectOutput extends AudioOutput { constructor() { super(24000); @@ -518,6 +605,83 @@ describe('RecorderAudioOutput', () => { await recorder.close(); }); + it('keeps the audio a finish reports as played while the first frame was parked', async () => { + const writes: AudioFrame[][] = []; + const recorder = new RecorderIO({ agentSession: {} as AgentSession }); + const recorderState = recorder as unknown as { + started: boolean; + writeCb: (buf: AudioFrame[]) => void; + }; + recorderState.started = true; + recorderState.writeCb = (buf) => writes.push(buf); + const downstream = new ParkFirstFrameOutput(); + const output = recorder.recordOutput(downstream); + + const capture = output.captureFrame(makeFrame(100)); + await downstream.frameParked.await; + // Wall-clock advances well past the position the sink is about to report. + await new Promise((resolve) => setTimeout(resolve, 150)); + downstream.reportFinished(0.05); + downstream.releaseGate(); + await capture; + output.flush(); + + const capturedSamples = writes + .flat() + .reduce((total, frame) => total + frame.samplesPerChannel, 0); + expect(capturedSamples).toBe(0.05 * 48000); + recorderState.started = false; + await recorder.close(); + }); + + it('accepts a retried capture after a rejection without an explicit flush', async () => { + const recorder = new RecorderIO({ agentSession: {} as AgentSession }); + const downstream = new RejectFirstCaptureOutput(); + const output = recorder.recordOutput(downstream); + + await expect(output.captureFrame(makeFrame(20, 24000))).rejects.toThrow('capture rejected'); + + // A caller that catches the rejection and retries must not be permanently poisoned. + await output.captureFrame(makeFrame(20, 24000)); + output.flush(); + const waitForRetriedSegment = output.waitForPlayout(); + downstream.finishSegment(); + const event = await Promise.race([ + waitForRetriedSegment, + new Promise<'timeout'>((resolve) => setTimeout(() => resolve('timeout'), 100)), + ]); + + expect(event).toEqual(RETRIED_SEGMENT_FINISH); + await recorder.close(); + }); + + it('accepts a retried capture after a rejection the sink had already counted', async () => { + const recorder = new RecorderIO({ agentSession: {} as AgentSession }); + const downstream = new CountThenRejectFirstCaptureOutput(); + const output = recorder.recordOutput(downstream); + const finishes: PlaybackFinishedEvent[] = []; + output.on(AudioOutput.EVENT_PLAYBACK_FINISHED, (event: PlaybackFinishedEvent) => { + finishes.push(event); + }); + + await expect(output.captureFrame(makeFrame(20, 24000))).rejects.toThrow( + 'capture rejected after counting', + ); + + await output.captureFrame(makeFrame(20, 24000)); + output.flush(); + const waitForRetriedSegment = output.waitForPlayout(); + downstream.finishSegment(); + const event = await Promise.race([ + waitForRetriedSegment, + new Promise<'timeout'>((resolve) => setTimeout(() => resolve('timeout'), 100)), + ]); + + expect(event).toEqual(RETRIED_SEGMENT_FINISH); + expect(finishes).toEqual([{ playbackPosition: 0, interrupted: true }, RETRIED_SEGMENT_FINISH]); + await recorder.close(); + }); + it('settles recorder state when downstream capture rejects', async () => { const recorder = new RecorderIO({ agentSession: {} as AgentSession }); const output = recorder.recordOutput(new RejectingAudioOutput()); diff --git a/agents/src/voice/recorder_io/recorder_io.ts b/agents/src/voice/recorder_io/recorder_io.ts index c0fe43adb..821894650 100644 --- a/agents/src/voice/recorder_io/recorder_io.ts +++ b/agents/src/voice/recorder_io/recorder_io.ts @@ -583,7 +583,8 @@ interface RecorderOutputSegment { finishRequested: boolean; flushed: boolean; playbackEvent?: PlaybackFinishedEvent; - speechStartTime?: number; + /** Wall-clock time the segment was opened, i.e. when its first frame entered `captureFrame`. */ + speechStartTime: number; currentPauseStart?: number; pauseWallTimes: Array<[number, number]>; } @@ -697,6 +698,14 @@ class RecorderAudioOutput extends AudioOutput { // `finishRequested` so we ask exactly once. segment.finishRequested = true; this.nextInChain.onPlaybackFinished({ playbackPosition: 0, interrupted: true }); + if (!this.currentSegment) { + // The sink also still has this segment latched open. A retried frame would silently + // join the segment we just declared finished — the sink would never count it, so we + // could not tell the frame had been accepted either, and the retried segment would be + // written off as interrupted at position zero. Only safe while we hold no open segment + // of our own; otherwise the latch belongs to a newer segment that is still growing. + this.nextInChain.abandonOpenSegment(); + } } return; } @@ -725,22 +734,10 @@ class RecorderAudioOutput extends AudioOutput { // Convert playbackPosition from seconds to ms for internal calculations let playbackPosition = options.playbackPosition * 1000; - if (segment.speechStartTime === undefined) { - this._logger.warn( - { - finishTime, - playbackPosition, - interrupted: options.interrupted, - }, - 'playback finished before speech started', - ); - playbackPosition = 0; - } - // Clamp playbackPosition to actual elapsed time (all in ms) playbackPosition = Math.max( 0, - Math.min(finishTime - (segment.speechStartTime ?? 0), playbackPosition), + Math.min(finishTime - segment.speechStartTime, playbackPosition), ); // Convert back to seconds for the event @@ -873,6 +870,14 @@ class RecorderAudioOutput extends AudioOutput { capturesInFlight: 0, finishRequested: false, flushed: false, + // Stamped here, before the frame leaves, rather than once the downstream output accepts + // it. A downstream output may park the frame (the `ParticipantAudioOutput` pause gate) + // and a finish can land while it is parked. `finishSegment` clamps the reported playback + // position against `finishTime - speechStartTime`, so a timestamp taken after the park + // would make the elapsed window ~zero (negative if we are still paused, since + // `finishTime` is then the pause start) and truncate away every frame the sink just + // reported as played. + speechStartTime: Date.now(), pauseWallTimes: [], }; this.segments.push(segment); @@ -902,10 +907,6 @@ class RecorderAudioOutput extends AudioOutput { this._startedWallTime = Date.now(); } - if (segment.speechStartTime === undefined) { - segment.speechStartTime = Date.now(); - } - captureCompleted = true; } finally { if (this.nextInChain && this.nextInChain.capturedPlayoutSegments > downstreamCapturedBefore) { @@ -916,6 +917,11 @@ class RecorderAudioOutput extends AudioOutput { segment.flushed = true; if (this.currentSegment === segment) { this.currentSegment = undefined; + // We just closed this segment and `drainFinishes` reports it finished, so the base class + // must stop counting it as open. Otherwise its capture latch is still set, the next + // `captureFrame` neither counts a new segment nor finds one of ours to attribute the + // frame to, and a caller that retries after a transient rejection is rejected forever. + this.abandonOpenSegment(); } } segment.capturesInFlight--; From 569ecd0c00a2e700b1aae774f3941e14ad00ef47 Mon Sep 17 00:00:00 2001 From: Toubat Date: Sat, 25 Jul 2026 16:47:15 -0700 Subject: [PATCH 4/5] fix(interruption): don't drop committed audio at the transport send boundary The WS transport re-read `overlapSpeechStarted` right before writing to the socket, after `await reconnecting`. `overlap-speech-ended`, `agent-speech-ended` and `bargein_detected` can all clear that flag inside the await window, so a slice the buffering stage had already committed to could be silently discarded and never counted in `numRequests`. Python's `send_task` gates on nothing and relies solely on the receive-side overlap check, which JS already has. Also make the verdict diagnosable: an overlap resolved without a usable inference result now warns with overlap duration, `numRequests`, buffered samples and agent-speech state, and every verdict logs probability / isInterruption / numRequests at debug. Adds the first coverage of the audio-to-verdict path. Co-authored-by: Cursor --- .changeset/adaptive-interruption-send-gate.md | 9 + .../interruption_pipeline.test.ts | 396 ++++++++++++++++++ .../interruption/interruption_stream.ts | 33 +- .../inference/interruption/ws_transport.ts | 7 +- agents/src/voice/events.ts | 6 + 5 files changed, 447 insertions(+), 4 deletions(-) create mode 100644 .changeset/adaptive-interruption-send-gate.md create mode 100644 agents/src/inference/interruption/interruption_pipeline.test.ts diff --git a/.changeset/adaptive-interruption-send-gate.md b/.changeset/adaptive-interruption-send-gate.md new file mode 100644 index 000000000..5f58e7a0b --- /dev/null +++ b/.changeset/adaptive-interruption-send-gate.md @@ -0,0 +1,9 @@ +--- +'@livekit/agents': patch +--- + +Adaptive interruption: stop dropping audio at the send boundary, and make the verdict diagnosable. + +The WebSocket transport re-checked `overlapSpeechStarted` immediately before writing to the socket, after awaiting any in-flight reconnect. Because `overlap-speech-ended`, `agent-speech-ended` and `bargein_detected` can all clear that flag inside the await window, audio the pipeline had already committed to sending could be discarded and never counted in `numRequests`. The buffering stage upstream is the only place that decides whether a slice belongs to an overlap, which matches the Python implementation, whose send task gates on nothing. + +An overlap that ends without any usable inference result now logs at `warn` (previously `debug`) with the overlap duration, `numRequests`, buffered samples and agent-speech state, so a fallback backchannel verdict is no longer indistinguishable from a genuine low-probability one. Every verdict also logs `probability`, `isInterruption` and `numRequests` at `debug`. `OverlappingSpeechEvent` is now exported by name for typing `overlapping_speech` handlers. diff --git a/agents/src/inference/interruption/interruption_pipeline.test.ts b/agents/src/inference/interruption/interruption_pipeline.test.ts new file mode 100644 index 000000000..9b6c0df0a --- /dev/null +++ b/agents/src/inference/interruption/interruption_pipeline.test.ts @@ -0,0 +1,396 @@ +// SPDX-FileCopyrightText: 2026 LiveKit, Inc. +// +// SPDX-License-Identifier: Apache-2.0 +// +// End-to-end (transport-mocked) test of the adaptive-interruption pipeline as it is wired by +// AudioRecognition: room audio -> interruption stream channel -> audio transformer -> WS transport. +// +// Regression coverage for "every overlap is classified as backchannel": the classifier can only +// return `isInterruption: true` from the inference-response path, which requires audio to actually +// reach the transport while the overlap is open. +import { AudioFrame } from '@livekit/rtc-node'; +import { ReadableStream } from 'node:stream/web'; +import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest'; +import { ChatContext } from '../../llm/chat_context.js'; +import { initializeLogger } from '../../log.js'; +import { AudioRecognition, type RecognitionHooks } from '../../voice/audio_recognition.js'; +import { createEndpointing } from '../../voice/turn_config/endpointing.js'; +import { MockWebSocket } from './_mock_ws.js'; +import type { InterruptionCacheEntry } from './interruption_cache_entry.js'; +import { AdaptiveInterruptionDetector } from './interruption_detector.js'; +import { InterruptionStreamBase, InterruptionStreamSentinel } from './interruption_stream.js'; +import { BoundedCache } from './utils.js'; +import { type WsTransportState, createWsTransport } from './ws_transport.js'; + +vi.mock('ws', async () => { + const { MockWebSocket } = await import('./_mock_ws.js'); + return { default: MockWebSocket, WebSocket: MockWebSocket }; +}); + +initializeLogger({ pretty: false, level: 'silent' }); + +const sleep = (ms: number) => new Promise((resolve) => setTimeout(resolve, ms)); + +async function waitFor(predicate: () => boolean, timeoutMs = 3000): Promise { + const start = performance.now(); + while (!predicate()) { + if (performance.now() - start > timeoutMs) { + throw new Error('condition not met within timeout'); + } + await sleep(5); + } +} + +function createHooks(): RecognitionHooks { + return { + onInterruption: vi.fn(), + onBackchannelConfirmed: vi.fn(), + onStartOfSpeech: vi.fn(), + onVADInferenceDone: vi.fn(), + onEndOfSpeech: vi.fn(), + onInterimTranscript: vi.fn(), + onFinalTranscript: vi.fn(), + onPreemptiveGeneration: vi.fn(), + onAgentBackchannelOpportunity: vi.fn(), + retrieveChatCtx: () => ChatContext.empty(), + onEndOfTurn: vi.fn(async () => true), + } as unknown as RecognitionHooks; +} + +/** 10ms of non-silent mono PCM, as a room track would deliver it. */ +function makeFrame(sampleRate: number): AudioFrame { + const samples = Math.floor(sampleRate / 100); + const data = new Int16Array(samples); + for (let i = 0; i < samples; i++) { + data[i] = Math.round(8000 * Math.sin((2 * Math.PI * 220 * i) / sampleRate)); + } + return new AudioFrame(data, sampleRate, 1, samples); +} + +/** Number of binary (audio) frames the transport pushed onto the socket. */ +function audioSendCount(ws: MockWebSocket): number { + return ws.sent.filter((s) => s instanceof Uint8Array).length; +} + +/** `created_at` header the transport stamped on a binary frame. */ +function createdAtOf(buf: Uint8Array): number { + const view = new DataView(buf.buffer, buf.byteOffset, buf.byteLength); + return view.getUint32(4, true) * 0x100000000 + view.getUint32(0, true); +} + +interface Harness { + recognition: AudioRecognition; + detector: AdaptiveInterruptionDetector; + hooks: RecognitionHooks; + ws: MockWebSocket; + events: OverlapEvent[]; + /** Requests the mock gateway has answered so far. */ + requestCount: () => number; + /** Answer the next request with a bargein verdict instead of a plain inference_done. */ + bargeinOnNextRequest: () => void; + close: () => Promise; +} + +interface OverlapEvent { + isInterruption: boolean; + numRequests: number; + probability: number; +} + +async function createHarness({ + sampleRate = 48000, +}: { sampleRate?: number } = {}): Promise { + const detector = new AdaptiveInterruptionDetector({ + baseUrl: 'http://localhost:9999', + apiKey: 'test-key', + apiSecret: 'test-secret', + }); + const hooks = createHooks(); + const recognition = new AudioRecognition({ + recognitionHooks: hooks, + interruptionDetection: detector, + endpointing: createEndpointing({ mode: 'dynamic', minDelay: 500, maxDelay: 3000, alpha: 0.9 }), + }); + // The pipeline wiring keys off `interruptionDetection`; the enabled flag additionally requires a + // VAD, which this test substitutes for by driving the overlap callbacks directly. + (recognition as unknown as { isInterruptionEnabled: boolean }).isInterruptionEnabled = true; + + const events: OverlapEvent[] = []; + detector.on('overlapping_speech', (ev) => events.push(ev)); + + // Continuous room audio, pumped in the background for the whole test, exactly as a subscribed + // track behaves — the interruption stream must pick it up on its own. + let pumping = true; + let pushFrame!: (frame: AudioFrame) => void; + const audioStream = new ReadableStream({ + start(controller) { + pushFrame = (frame) => controller.enqueue(frame); + }, + }); + recognition.setInputAudioStream(audioStream); + const pump = (async () => { + while (pumping) { + pushFrame(makeFrame(sampleRate)); + await sleep(10); + } + })(); + + const ac = new AbortController(); + const task = ( + recognition as unknown as { + createInterruptionTask: ( + d: AdaptiveInterruptionDetector, + signal: AbortSignal, + ) => Promise; + } + ).createInterruptionTask(detector, ac.signal); + + await waitFor(() => MockWebSocket.instances.length > 0); + const ws = MockWebSocket.instances[MockWebSocket.instances.length - 1]!; + ws.simulateOpen(); + await waitFor(() => ws.sent.length > 0); // session.create + ws.simulateMessage({ type: 'session.created', default_threshold: 0.5 }); + await sleep(20); + + // Stand-in for the gateway: answer every audio frame promptly, as the real service does. Without + // this the transport's own 0.7s inference timeout tears the stream down. + let answered = 0; + let bargeinPending = false; + const responder = (async () => { + while (pumping) { + const binary = ws.sent.filter((s): s is Uint8Array => s instanceof Uint8Array); + while (answered < binary.length) { + const createdAt = createdAtOf(binary[answered]!); + answered++; + if (bargeinPending) { + bargeinPending = false; + ws.simulateMessage({ + type: 'bargein_detected', + created_at: createdAt, + probabilities: [0.91, 0.93, 0.95], + prediction_duration: 0.02, + }); + } else { + ws.simulateMessage({ + type: 'inference_done', + created_at: createdAt, + probabilities: [0.01, 0.02], + prediction_duration: 0.02, + is_bargein: false, + }); + } + } + await sleep(5); + } + })(); + + return { + recognition, + detector, + hooks, + ws, + events, + requestCount: () => answered, + bargeinOnNextRequest: () => { + bargeinPending = true; + }, + close: async () => { + pumping = false; + await pump; + await responder; + ac.abort(); + await task.catch(() => {}); + await recognition.close().catch(() => {}); + }, + }; +} + +describe('adaptive interruption pipeline', () => { + beforeEach(() => { + MockWebSocket.instances.length = 0; + }); + + afterEach(() => { + vi.clearAllMocks(); + }); + + it('sends inference requests for user audio that overlaps agent speech', async () => { + const h = await createHarness(); + + await h.recognition.onStartOfAgentSpeech(Date.now()); + await sleep(200); + await h.recognition.onStartOfOverlapSpeech(200, Date.now()); + await sleep(600); + + expect(audioSendCount(h.ws)).toBeGreaterThan(0); + + await h.close(); + }); + + it('classifies a server bargein verdict as an interruption', async () => { + const h = await createHarness(); + + await h.recognition.onStartOfAgentSpeech(Date.now()); + await sleep(200); + h.bargeinOnNextRequest(); + await h.recognition.onStartOfOverlapSpeech(200, Date.now()); + await sleep(600); + + expect(audioSendCount(h.ws)).toBeGreaterThan(0); + await waitFor(() => h.events.length > 0); + + expect(h.events[0]!.isInterruption).toBe(true); + await waitFor(() => vi.mocked(h.hooks.onInterruption).mock.calls.length > 0); + + await h.close(); + }); + + it('keeps making requests across repeated short overlaps in one agent turn', async () => { + const h = await createHarness(); + + await h.recognition.onStartOfAgentSpeech(Date.now()); + await sleep(300); + + const perOverlapRequests: number[] = []; + for (let i = 0; i < 3; i++) { + const before = audioSendCount(h.ws); + await h.recognition.onStartOfOverlapSpeech(200, Date.now()); + await sleep(1200); // the user's overlaps were 1-2s each + await h.recognition.onEndOfOverlapSpeech(Date.now()); + await sleep(50); + perOverlapRequests.push(audioSendCount(h.ws) - before); + } + + expect(h.events).toHaveLength(3); + expect(perOverlapRequests.every((n) => n > 0)).toBe(true); + expect(h.events.map((e) => e.numRequests).every((n) => n > 0)).toBe(true); + + await h.close(); + }); +}); + +// --------------------------------------------------------------------------- +// Send-time overlap gate (regression) +// --------------------------------------------------------------------------- + +/** + * The audio transformer commits a slice while the overlap is open, but the transport's `transform` + * runs later — the pipe between the two stages resolves on its own tick, and `transform` itself + * parks on `await reconnecting`. In that window an `overlap-speech-ended` sentinel (or a + * `bargein_detected` message) can clear `overlapSpeechStarted`. Re-reading that flag at send time + * therefore discards audio the pipeline already decided to send, and the request is never counted. + * + * Python's `send_task` has no such gate: it sends every slice the buffering stage hands it, and + * gates only on the receive side. + */ +describe('interruption transport send gate', () => { + /** One frame worth exactly the 100 ms detection interval, so a single push commits a slice. */ + function detectionIntervalFrame(sampleRate = 16000): AudioFrame { + const samples = Math.floor(sampleRate * 0.1); + return new AudioFrame(new Int16Array(samples), sampleRate, 1, samples); + } + + async function openStream(): Promise<{ + stream: InterruptionStreamBase; + ws: MockWebSocket; + drained: Promise; + }> { + const detector = new AdaptiveInterruptionDetector({ + baseUrl: 'http://localhost:9999', + apiKey: 'test-key', + apiSecret: 'test-secret', + }); + const priorSockets = MockWebSocket.instances.length; + const stream = new InterruptionStreamBase(detector, {}); + + // The event side must be consumed or the pipeline stalls on backpressure. + const reader = stream.stream().getReader(); + const drained = (async () => { + for (;;) { + const { done } = await reader.read(); + if (done) return; + } + })().catch(() => {}); + + await waitFor(() => MockWebSocket.instances.length > priorSockets); + const ws = MockWebSocket.instances[priorSockets]!; + ws.simulateOpen(); + await waitFor(() => ws.sent.length > 0); // session.create + ws.simulateMessage({ type: 'session.created', default_threshold: 0.5 }); + await sleep(10); + + return { stream, ws, drained }; + } + + /** + * Park the transport mid-send by starting an in-place reconnect and leaving the replacement + * socket un-opened: `transform` is then blocked on `await reconnecting` with the slice in hand. + */ + async function stallOnReconnect(stream: InterruptionStreamBase): Promise { + const priorSockets = MockWebSocket.instances.length; + await stream.updateOptions({ threshold: 0.7 }); + await waitFor(() => MockWebSocket.instances.length > priorSockets); + return MockWebSocket.instances[priorSockets]!; + } + + it('sends a slice committed during an overlap that ends while a reconnect is in flight', async () => { + const { stream } = await openStream(); + + await stream.pushFrame(InterruptionStreamSentinel.agentSpeechStarted()); + await stream.pushFrame(InterruptionStreamSentinel.overlapSpeechStarted(200, Date.now())); + + const ws2 = await stallOnReconnect(stream); + + // Committed while the overlap is open; `transform` parks on the pending reconnect. + await stream.pushFrame(detectionIntervalFrame()); + // The overlap ends while the slice is still parked, clearing `overlapSpeechStarted`. + await stream.pushFrame(InterruptionStreamSentinel.overlapSpeechEnded(Date.now())); + await sleep(20); + + ws2.simulateOpen(); // reconnect completes, the parked slice resumes + await waitFor(() => ws2.sent.length > 0); // session.create on the new socket + await sleep(50); + + expect(audioSendCount(ws2)).toBe(1); + + await stream.close(); + }); + + // The buffering stage in interruption_stream.ts is the single place that decides whether a slice + // belongs to an overlap; the transport must not second-guess it. Pinning that directly (rather + // than only through the reconnect race above) keeps the JS send path aligned with Python's + // `send_task`, which gates on nothing. + it('forwards a slice regardless of the overlap flag at send time', async () => { + const state: WsTransportState = { + overlapSpeechStarted: false, + overlapSpeechStartedAt: undefined, + cache: new BoundedCache(10), + }; + const priorSockets = MockWebSocket.instances.length; + const { transport, close } = createWsTransport( + { + baseUrl: 'http://localhost:9999', + apiKey: 'test-key', + apiSecret: 'test-secret', + sampleRate: 16000, + minFrames: 2, + timeout: 0, + connectTimeout: 2000, + }, + () => state, + (partial) => Object.assign(state, partial), + ); + + await waitFor(() => MockWebSocket.instances.length > priorSockets); + const ws = MockWebSocket.instances[priorSockets]!; + ws.simulateOpen(); + await waitFor(() => ws.sent.length > 0); // session.create + + await transport.writable.getWriter().write(new Int16Array(1600)); + await sleep(20); + + expect(audioSendCount(ws)).toBe(1); + + close(); + }); +}); diff --git a/agents/src/inference/interruption/interruption_stream.ts b/agents/src/inference/interruption/interruption_stream.ts index efb02819f..4eb20a0ce 100644 --- a/agents/src/inference/interruption/interruption_stream.ts +++ b/agents/src/inference/interruption/interruption_stream.ts @@ -278,8 +278,24 @@ export class InterruptionStreamBase { let latestEntry = cache.pop( (entry) => entry.totalDurationInS !== undefined && entry.totalDurationInS > 0, ); + const numRequests = getAndResetNumRequests(); if (!latestEntry) { - this.logger.debug('no request made for overlap speech'); + // The verdict below is a fallback, not a model decision. Warn rather than debug: + // without this, an unanswered overlap is indistinguishable from a genuine + // low-probability backchannel in production logs. + this.logger.warn( + { + overlapDuration: + this.overlapSpeechStartedAt !== undefined + ? chunk.endedAt - this.overlapSpeechStartedAt + : undefined, + numRequests, + accumulatedSamples, + agentSpeechStarted, + agentEnded: chunk.agentEnded, + }, + 'no interruption inference result for overlap speech, defaulting to backchannel', + ); latestEntry = InterruptionCacheEntry.default(); } const e = latestEntry ?? InterruptionCacheEntry.default(); @@ -295,7 +311,7 @@ export class InterruptionStreamBase { detectionDelayInS: e.detectionDelayInS, predictionDurationInS: e.predictionDurationInS, probability: e.probability, - numRequests: getAndResetNumRequests(), + numRequests, }; controller.enqueue(event); overlapSpeechStarted = false; @@ -328,6 +344,19 @@ export class InterruptionStreamBase { const eventEmitter = new TransformStream({ transform: (chunk, controller) => { + // Once per overlap. `numRequests: 0` here means the model was never asked, which is what + // distinguishes "scored below the threshold" from "never classified". + this.logger.debug( + { + isInterruption: chunk.isInterruption, + probability: chunk.probability, + numRequests: chunk.numRequests, + agentEnded: chunk.agentEnded, + totalDuration: chunk.totalDurationInS * 1000, + detectionDelay: chunk.detectionDelayInS * 1000, + }, + 'interruption verdict', + ); this.model.emit('overlapping_speech', chunk); const metrics: InterruptionMetrics = { diff --git a/agents/src/inference/interruption/ws_transport.ts b/agents/src/inference/interruption/ws_transport.ts index a1b252660..65a0c4adb 100644 --- a/agents/src/inference/interruption/ws_transport.ts +++ b/agents/src/inference/interruption/ws_transport.ts @@ -504,9 +504,12 @@ export function createWsTransport( // rejects — a failed reconnect has already errored the stream via outputController. if (reconnecting) await reconnecting; - // Only forwards buffered audio while overlap speech is actively on. + // Deliberately no overlap-state gate here: whether a slice belongs to an overlap is decided + // once, upstream, when the slice is cut. Re-reading the flag after the await above would + // drop audio the pipeline already committed to, since `overlap-speech-ended`, + // `agent-speech-ended` and `bargein_detected` can all clear it in that window. Late + // responses are harmless — handleMessage() ignores anything outside an open overlap. const state = getState(); - if (!state.overlapSpeechStartedAt || !state.overlapSpeechStarted) return; if (options.timeout > 0) { const now = performance.now(); diff --git a/agents/src/voice/events.ts b/agents/src/voice/events.ts index 3a3050f4b..45d19489b 100644 --- a/agents/src/voice/events.ts +++ b/agents/src/voice/events.ts @@ -436,6 +436,12 @@ export const createAgentFalseInterruptionEvent = ({ createdAt, }); +/** + * Payload of {@link AgentSessionEventTypes.OverlappingSpeech}. Re-exported here so handlers can be + * typed by name; it carries `probability`, `probabilities` and `numRequests` alongside the verdict. + */ +export type { OverlappingSpeechEvent }; + export type AgentEvent = | UserInputTranscribedEvent | UserStateChangedEvent From 537d3748ed8f6baeb232088fd478d76fa3b99ccc Mon Sep 17 00:00:00 2001 From: Toubat Date: Sat, 25 Jul 2026 17:24:55 -0700 Subject: [PATCH 5/5] fix(interruption): charge a request to the overlap its slice was cut for MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Removing the send-time overlap gate means a slice can reach the socket after its overlap has closed, and possibly after the next one has opened. onRequestSent() then incremented the shared counter for whichever overlap was current, so a later overlap could report a numRequests it never made — the one signal that tells "the model was never asked" apart from "the model scored it low". Slices now carry the overlap generation they were cut for, and the count is only taken when that generation is still the open one. Co-authored-by: Cursor --- .changeset/adaptive-interruption-send-gate.md | 2 + .../interruption_pipeline.test.ts | 48 ++++++++++++++++++- .../interruption/interruption_stream.ts | 19 ++++++-- agents/src/inference/interruption/types.ts | 13 +++++ .../inference/interruption/ws_transport.ts | 22 +++++---- 5 files changed, 91 insertions(+), 13 deletions(-) diff --git a/.changeset/adaptive-interruption-send-gate.md b/.changeset/adaptive-interruption-send-gate.md index 5f58e7a0b..5f88c7929 100644 --- a/.changeset/adaptive-interruption-send-gate.md +++ b/.changeset/adaptive-interruption-send-gate.md @@ -6,4 +6,6 @@ Adaptive interruption: stop dropping audio at the send boundary, and make the ve The WebSocket transport re-checked `overlapSpeechStarted` immediately before writing to the socket, after awaiting any in-flight reconnect. Because `overlap-speech-ended`, `agent-speech-ended` and `bargein_detected` can all clear that flag inside the await window, audio the pipeline had already committed to sending could be discarded and never counted in `numRequests`. The buffering stage upstream is the only place that decides whether a slice belongs to an overlap, which matches the Python implementation, whose send task gates on nothing. +Slices now carry the overlap they were cut for, so a send that lands after that overlap closed is no longer counted against whichever overlap happens to be open when the socket accepts it. Without this, a later overlap could report a `numRequests` it never made. + An overlap that ends without any usable inference result now logs at `warn` (previously `debug`) with the overlap duration, `numRequests`, buffered samples and agent-speech state, so a fallback backchannel verdict is no longer indistinguishable from a genuine low-probability one. Every verdict also logs `probability`, `isInterruption` and `numRequests` at `debug`. `OverlappingSpeechEvent` is now exported by name for typing `overlapping_speech` handlers. diff --git a/agents/src/inference/interruption/interruption_pipeline.test.ts b/agents/src/inference/interruption/interruption_pipeline.test.ts index 9b6c0df0a..5e7b41b89 100644 --- a/agents/src/inference/interruption/interruption_pipeline.test.ts +++ b/agents/src/inference/interruption/interruption_pipeline.test.ts @@ -292,6 +292,7 @@ describe('interruption transport send gate', () => { async function openStream(): Promise<{ stream: InterruptionStreamBase; + detector: AdaptiveInterruptionDetector; ws: MockWebSocket; drained: Promise; }> { @@ -319,7 +320,7 @@ describe('interruption transport send gate', () => { ws.simulateMessage({ type: 'session.created', default_threshold: 0.5 }); await sleep(10); - return { stream, ws, drained }; + return { stream, detector, ws, drained }; } /** @@ -386,11 +387,54 @@ describe('interruption transport send gate', () => { ws.simulateOpen(); await waitFor(() => ws.sent.length > 0); // session.create - await transport.writable.getWriter().write(new Int16Array(1600)); + await transport.writable.getWriter().write({ + type: 'audio-slice', + audio: new Int16Array(1600), + overlapGeneration: 1, + }); await sleep(20); expect(audioSendCount(ws)).toBe(1); close(); }); + + /** + * Sending a slice unconditionally means the send can land after the overlap it was cut for has + * closed — possibly after the *next* overlap has already opened. `numRequests` is per-overlap + * accounting ("was the model ever asked about this overlap?"), so the count has to follow the + * overlap the slice was cut for rather than whichever one happens to be open when the socket + * finally accepts it. Otherwise a later overlap inherits a request it never made. + */ + it('does not charge a parked slice to the overlap that follows it', async () => { + const { stream, detector } = await openStream(); + const events: OverlapEvent[] = []; + detector.on('overlapping_speech', (ev) => events.push(ev)); + + await stream.pushFrame(InterruptionStreamSentinel.agentSpeechStarted()); + await stream.pushFrame(InterruptionStreamSentinel.overlapSpeechStarted(200, Date.now())); + + const ws2 = await stallOnReconnect(stream); + + // Cut during the first overlap, then parked on the pending reconnect. + await stream.pushFrame(detectionIntervalFrame()); + await stream.pushFrame(InterruptionStreamSentinel.overlapSpeechEnded(Date.now())); + await sleep(20); + + // A second overlap opens while the slice is still parked. It cuts no audio of its own. + await stream.pushFrame(InterruptionStreamSentinel.overlapSpeechStarted(200, Date.now())); + await sleep(20); + + ws2.simulateOpen(); // the parked slice resumes, now that a later overlap is open + await waitFor(() => ws2.sent.length > 0); // session.create on the new socket + await sleep(50); + expect(audioSendCount(ws2)).toBe(1); + + await stream.pushFrame(InterruptionStreamSentinel.overlapSpeechEnded(Date.now())); + await waitFor(() => events.length === 2); + + expect(events[1]!.numRequests).toBe(0); + + await stream.close(); + }); }); diff --git a/agents/src/inference/interruption/interruption_stream.ts b/agents/src/inference/interruption/interruption_stream.ts index 4eb20a0ce..ffb4b2ca6 100644 --- a/agents/src/inference/interruption/interruption_stream.ts +++ b/agents/src/inference/interruption/interruption_stream.ts @@ -17,6 +17,7 @@ import { type AgentSpeechStarted, type ApiConnectOptions, type Flush, + type InterruptionAudioSlice, type InterruptionOptions, type InterruptionSentinel, type OverlapSpeechEnded, @@ -167,6 +168,9 @@ export class InterruptionStreamBase { let accumulatedSamples = 0; let overlapSpeechStarted = false; let overlapCount = 0; + // Monotonic across the life of the stream — unlike `overlapCount`, which restarts each agent + // turn — so a slice cut for an earlier overlap can never be mistaken for the current one. + let overlapGeneration = 0; const cache = new BoundedCache(10); const inferenceS16Data = new Int16Array( Math.ceil(this.options.maxAudioDurationInS * this.options.sampleRate), @@ -191,7 +195,11 @@ export class InterruptionStreamBase { } }; - const onRequestSent = () => { + // A slice is sent unconditionally once cut, so the send can land after its overlap closed and + // the next one opened. Charging it to whatever overlap is open at send time would hand a later + // overlap a request it never made, which is exactly what `numRequests: 0` is meant to detect. + const onRequestSent = (sliceOverlapGeneration: number) => { + if (sliceOverlapGeneration !== overlapGeneration) return; this.numRequests++; }; @@ -204,7 +212,7 @@ export class InterruptionStreamBase { // First transform: process input frames/sentinels and output audio slices or events const audioTransformer = new TransformStream< InterruptionSentinel | AudioFrame, - Int16Array | OverlappingSpeechEvent + InterruptionAudioSlice | OverlappingSpeechEvent >( { transform: (chunk, controller) => { @@ -233,7 +241,11 @@ export class InterruptionStreamBase { ) { const audioSlice = inferenceS16Data.slice(0, startIdx); accumulatedSamples = 0; - controller.enqueue(audioSlice); + controller.enqueue({ + type: 'audio-slice', + audio: audioSlice, + overlapGeneration, + }); } } else if (chunk.type === 'agent-speech-started') { this.logger.debug('agent speech started'); @@ -262,6 +274,7 @@ export class InterruptionStreamBase { overlapSpeechStarted = true; accumulatedSamples = 0; overlapCount += 1; + overlapGeneration += 1; if (overlapCount <= 1) { const keepSize = Math.round((chunk.speechDuration / 1000) * this.options.sampleRate) + diff --git a/agents/src/inference/interruption/types.ts b/agents/src/inference/interruption/types.ts index d062eb739..32bfc7e97 100644 --- a/agents/src/inference/interruption/types.ts +++ b/agents/src/inference/interruption/types.ts @@ -23,6 +23,19 @@ export interface OverlappingSpeechEvent { numRequests: number; } +/** + * An audio slice on its way to the transport, tagged with the overlap it was cut for. + * + * The send itself is unconditional, so it can land after that overlap has closed — and after the + * next one has opened. The tag keeps per-overlap request accounting attached to the right overlap + * regardless of when the socket accepts the slice. + */ +export interface InterruptionAudioSlice { + type: 'audio-slice'; + audio: Int16Array; + overlapGeneration: number; +} + /** * Configuration options for interruption detection. */ diff --git a/agents/src/inference/interruption/ws_transport.ts b/agents/src/inference/interruption/ws_transport.ts index 65a0c4adb..b00c6cb24 100644 --- a/agents/src/inference/interruption/ws_transport.ts +++ b/agents/src/inference/interruption/ws_transport.ts @@ -10,7 +10,7 @@ import { log } from '../../log.js'; import { Event } from '../../utils.js'; import { buildMetadataHeaders, createAccessToken } from '../utils.js'; import { InterruptionCacheEntry } from './interruption_cache_entry.js'; -import type { OverlappingSpeechEvent } from './types.js'; +import type { InterruptionAudioSlice, OverlappingSpeechEvent } from './types.js'; import type { BoundedCache } from './utils.js'; // WebSocket message types @@ -147,7 +147,10 @@ async function connectWebSocket( } export interface WsTransportResult { - transport: TransformStream; + transport: TransformStream< + InterruptionAudioSlice | OverlappingSpeechEvent, + OverlappingSpeechEvent + >; reconnect: () => Promise; close: () => void; } @@ -164,7 +167,7 @@ export function createWsTransport( getState: () => WsTransportState, setState: (partial: Partial) => void, updateUserSpeakingSpan?: (entry: InterruptionCacheEntry) => void, - onRequestSent?: () => void, + onRequestSent?: (overlapGeneration: number) => void, getAndResetNumRequests?: () => number, ): WsTransportResult { const logger = log(); @@ -383,13 +386,14 @@ export function createWsTransport( } } - function sendAudioData(audioSlice: Int16Array): void { + function sendAudioData(slice: InterruptionAudioSlice): void { // Backstop for a genuine unexpected drop: throws a retryable error so the stream fails over. An // intentional reconnect is awaited in transform() before we get here, so it won't fire then. if (!activeWs || activeWs.readyState !== WebSocket.OPEN) { throw new APIConnectionError({ message: 'WebSocket not connected' }); } + const audioSlice = slice.audio; const state = getState(); const createdAt = Math.floor(performance.now()); @@ -417,7 +421,7 @@ export function createWsTransport( combined.set(audioBytes, 8); activeWs.send(combined); - onRequestSent?.(); + onRequestSent?.(slice.overlapGeneration); } // Close the current socket without ending the transport (used by both close() and reconnect). @@ -482,7 +486,7 @@ export function createWsTransport( } const transport = new TransformStream< - Int16Array | OverlappingSpeechEvent, + InterruptionAudioSlice | OverlappingSpeechEvent, OverlappingSpeechEvent >( { @@ -495,7 +499,7 @@ export function createWsTransport( }, async transform(chunk, controller) { - if (!(chunk instanceof Int16Array)) { + if (chunk.type !== 'audio-slice') { controller.enqueue(chunk); return; } @@ -508,7 +512,9 @@ export function createWsTransport( // once, upstream, when the slice is cut. Re-reading the flag after the await above would // drop audio the pipeline already committed to, since `overlap-speech-ended`, // `agent-speech-ended` and `bargein_detected` can all clear it in that window. Late - // responses are harmless — handleMessage() ignores anything outside an open overlap. + // responses are harmless — handleMessage() ignores anything outside an open overlap, and + // the slice carries the overlap it was cut for so a late send is never charged to a + // later one. const state = getState(); if (options.timeout > 0) {