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 1b7785129ca6266579193b46a7f99c3394f0bac3 Mon Sep 17 00:00:00 2001 From: Toubat Date: Sun, 26 Jul 2026 01:56:34 -0700 Subject: [PATCH 4/5] fix(recorder): settle a segment on the sink's finish without waiting for a flush Registering the recorder segment before forwarding a frame made `drainFinishes` gate *every* settle on a flush, including segments the wrapped output had already reported finished. A caller that waited on such a segment without flushing stalled forever. That gate is only sound for a synthesized finish, where the flush is what guarantees the segment can no longer grow. A finish the sink actually sent is authoritative, and the `AudioOutput` contract lets a sink send one as soon as its playout ends. `SyncedAudioOutput.waitForPlayout` does precisely that when it reconciles a segment the output below it dropped, so the stall was reachable on the default chain rather than only with custom sinks. Settling outside a flush boundary leaves the base class's capture latch set, so release it; otherwise the next `captureFrame` neither counts a new base segment nor finds one of ours and throws `recorder capture has no active segment`. Also pins the deliberate behavior change that comes with registering the segment early: `waitForPlayout` now blocks while a frame is in flight inside the wrapped output instead of returning a fabricated `{ playbackPosition: 0, interrupted: false }`. Tests: a sink finish before any flush, a capture following an unflushed settle, the parked-frame wait, the synchronizer drift finish, and an interrupted turn followed by a working turn driven through a real `ParticipantAudioOutput`. Co-authored-by: Cursor --- .../fix-recorder-capture-finish-race.md | 13 ++ .../src/voice/recorder_io/recorder_io.test.ts | 201 +++++++++++++++++- agents/src/voice/recorder_io/recorder_io.ts | 30 ++- 3 files changed, 238 insertions(+), 6 deletions(-) diff --git a/.changeset/fix-recorder-capture-finish-race.md b/.changeset/fix-recorder-capture-finish-race.md index b205af63e..bb00f303b 100644 --- a/.changeset/fix-recorder-capture-finish-race.md +++ b/.changeset/fix-recorder-capture-finish-race.md @@ -13,3 +13,16 @@ playback position to zero and drop the audio the sink reported as played. And a 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`. + +A finish reported by the wrapped output settles its segment whether or not the recorder has been +flushed. The `AudioOutput` contract lets a sink report a finish as soon as its playout ends, and +`TranscriptionSynchronizer` does exactly that when it reconciles a dropped segment from +`waitForPlayout`, so requiring a flush first would strand the caller. Only a _synthesized_ finish — +for a segment the wrapped output never counted — still waits for the flush, which is what +guarantees the segment can no longer grow. + +Behavior change: `waitForPlayout` now blocks while a frame is still in flight inside the wrapped +output. Previously it could return immediately with a fabricated +`{ playbackPosition: 0, interrupted: false }`, reporting a turn as completed while its audio had +not been handed to the sink yet. Callers that relied on the early return will now wait for the +real playback result. diff --git a/agents/src/voice/recorder_io/recorder_io.test.ts b/agents/src/voice/recorder_io/recorder_io.test.ts index 27849e0cd..89edbca1a 100644 --- a/agents/src/voice/recorder_io/recorder_io.test.ts +++ b/agents/src/voice/recorder_io/recorder_io.test.ts @@ -1,7 +1,7 @@ // SPDX-FileCopyrightText: 2025 LiveKit, Inc. // // SPDX-License-Identifier: Apache-2.0 -import { AudioFrame } from '@livekit/rtc-node'; +import { AudioFrame, type Room, TrackPublishOptions } from '@livekit/rtc-node'; import fs from 'node:fs'; import os from 'node:os'; import path from 'node:path'; @@ -10,7 +10,9 @@ 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, type PlaybackFinishedEvent } from '../io.js'; +import { AudioInput, AudioOutput, type PlaybackFinishedEvent, TextOutput } from '../io.js'; +import { ParticipantAudioOutput } from '../room_io/_output.js'; +import { TranscriptionSynchronizer } from '../transcription/synchronizer.js'; import { RecorderIO } from './recorder_io.js'; class FakeAudioInput extends AudioInput { @@ -308,6 +310,77 @@ class AcceptThenCountRejectOutput extends AudioOutput { clearBuffer(): void {} } +/** + * A sink that counts every frame and reports its finish when the test says so — including + * outside a flush boundary, which the {@link AudioOutput} contract permits. + */ +class AcceptThenFinishOutput extends AudioOutput { + constructor() { + super(24000); + } + + reportFinished(event: PlaybackFinishedEvent): void { + this.onPlaybackFinished(event); + } + + clearBuffer(): void {} +} + +/** Parks the first frame *before* counting it, the way `ParticipantAudioOutput`'s pause gate does. */ +class ParkBeforeCountOutput extends AudioOutput { + readonly frameParked = new Future(); + private readonly gate = new Future(); + private captures = 0; + + constructor() { + super(24000); + } + + async captureFrame(frame: AudioFrame): Promise { + this.captures++; + if (this.captures === 1) { + this.frameParked.resolve(); + await this.gate.await; + } + await super.captureFrame(frame); + } + + releaseGate(): void { + this.gate.resolve(); + } + + reportFinished(event: PlaybackFinishedEvent): void { + this.onPlaybackFinished(event); + } + + clearBuffer(): void {} +} + +class FakeTextOutput extends TextOutput { + async captureText(): Promise {} + + flush(): void {} +} + +/** + * Report a stall as a value rather than letting the test time out, so a hang is distinguishable + * from an assertion failure in the output. + */ +async function settleOrStall( + promise: Promise, + timeoutMs = 200, +): Promise { + let timer: NodeJS.Timeout | undefined; + const watchdog = new Promise<'did not settle'>((resolve) => { + timer = setTimeout(() => resolve('did not settle'), timeoutMs); + }); + try { + return await Promise.race([promise, watchdog]); + } finally { + if (timer) clearTimeout(timer); + } +} + function makeFrame(durationMs: number, sampleRate = 48000, channels = 1): AudioFrame { const samplesPerChannel = Math.floor((durationMs / 1000) * sampleRate); return new AudioFrame( @@ -461,6 +534,66 @@ describe('RecorderAudioOutput', () => { await recorder.close(); }); + it('settles a segment the wrapped output finished before we flushed', async () => { + const recorder = new RecorderIO({ agentSession: {} as AgentSession }); + const downstream = new AcceptThenFinishOutput(); + const output = recorder.recordOutput(downstream); + + await output.captureFrame(makeFrame(20, 24000)); + // The `AudioOutput` contract lets a sink report a finish whenever its playout ends; it does + // not have to wait for a flush. Requiring one here stranded the caller forever. + downstream.reportFinished({ playbackPosition: 0, interrupted: true }); + + expect(await settleOrStall(output.waitForPlayout())).toEqual({ + playbackPosition: 0, + interrupted: true, + }); + await recorder.close(); + }); + + it('opens a fresh segment for a capture that follows a settle with no flush', async () => { + const recorder = new RecorderIO({ agentSession: {} as AgentSession }); + const downstream = new AcceptThenFinishOutput(); + const output = recorder.recordOutput(downstream); + + await output.captureFrame(makeFrame(20, 24000)); + downstream.reportFinished({ playbackPosition: 0, interrupted: true }); + await settleOrStall(output.waitForPlayout()); + + // Settling outside a flush boundary leaves the base class's capture latch set. Unless it is + // released, this capture neither counts a new base segment nor finds one of ours, and + // rejects with `recorder capture has no active segment`. + await expect(output.captureFrame(makeFrame(20, 24000))).resolves.toBeUndefined(); + + output.flush(); + expect(await settleOrStall(output.waitForPlayout())).not.toBe('did not settle'); + await recorder.close(); + }); + + it('waits for a frame the wrapped output has parked instead of reporting a fabricated finish', async () => { + const recorder = new RecorderIO({ agentSession: {} as AgentSession }); + const downstream = new ParkBeforeCountOutput(); + const output = recorder.recordOutput(downstream); + + const capture = output.captureFrame(makeFrame(20, 24000)); + await downstream.frameParked.await; + + // Deliberate divergence from the pre-refactor behavior, which returned immediately with a + // fabricated `{ playbackPosition: 0, interrupted: false }` because the segment had not been + // registered yet. The audio has demonstrably not finished playing, so claiming it completed + // is wrong; registering the segment before forwarding is also what makes the finish that + // arrives during the park attributable at all. + const wait = output.waitForPlayout(); + expect(await settleOrStall(wait)).toBe('did not settle'); + + downstream.releaseGate(); + await capture; + downstream.reportFinished({ playbackPosition: 0.02, interrupted: false }); + + expect(await settleOrStall(wait)).toEqual({ playbackPosition: 0.02, interrupted: false }); + 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()); @@ -757,6 +890,70 @@ describe('RecorderAudioOutput', () => { }); }); +describe('RecorderAudioOutput in front of a real ParticipantAudioOutput', () => { + function makeParticipantAudioOutput(): ParticipantAudioOutput { + const out = new ParticipantAudioOutput({} as Room, { + sampleRate: 24000, + numChannels: 1, + trackPublishOptions: new TrackPublishOptions(), + }); + // `publishTrack` normally resolves this; there is no room to publish to here. + (out as unknown as { startedFuture: Future }).startedFuture.resolve(); + return out; + } + + it('runs a follow-on turn after a turn interrupted while the output was paused', async () => { + // The customer's report: an interrupt lands while a frame sits at the pause gate, and the + // session never speaks again because the interrupted turn's `waitForPlayout` never settles. + const recorder = new RecorderIO({ agentSession: {} as AgentSession }); + const output = recorder.recordOutput(makeParticipantAudioOutput()); + + await output.captureFrame(makeFrame(100, 24000)); + output.flush(); + + output.pause(); + const interruptedTurn = output.captureFrame(makeFrame(100, 24000)); + await new Promise((resolve) => setTimeout(resolve, 20)); + + // `clearBuffer` both completes the in-flight playout task with an interrupted finish and + // releases the parked frame without the sink ever counting it. + output.clearBuffer(); + await interruptedTurn; + output.flush(); + + expect(await settleOrStall(output.waitForPlayout(), 2000)).not.toBe('did not settle'); + + output.resume(); + await output.captureFrame(makeFrame(100, 24000)); + output.flush(); + + expect(await settleOrStall(output.waitForPlayout(), 3000)).toEqual({ + playbackPosition: 0.1, + interrupted: false, + }); + await recorder.close(); + }); +}); + +describe('RecorderAudioOutput behind a TranscriptionSynchronizer', () => { + it('settles when the synchronizer emits a drift finish from waitForPlayout', async () => { + // `syncTranscription` is on by default, so this is the default output chain: + // recorder -> synchronizer -> sink. When the sink drops a frame the synchronizer counted, + // `SyncedAudioOutput.waitForPlayout` reconciles the drift by emitting a synthetic finish — + // with no flush anywhere. That makes the "finished before we flushed" case reachable + // without a custom sink. + const sink = new DroppingAudioOutput(); + const synchronizer = new TranscriptionSynchronizer(sink, new FakeTextOutput()); + const recorder = new RecorderIO({ agentSession: {} as AgentSession }); + const output = recorder.recordOutput(synchronizer.audioOutput); + + await output.captureFrame(makeFrame(20, 24000)); + + expect(await settleOrStall(output.waitForPlayout(), 500)).not.toBe('did not settle'); + await recorder.close(); + }); +}); + describe('RecorderIO writable stream error detection', () => { it('detects ERR_INVALID_STATE stream closure errors', () => { const err = new TypeError('Invalid state: WritableStream is closed'); diff --git a/agents/src/voice/recorder_io/recorder_io.ts b/agents/src/voice/recorder_io/recorder_io.ts index 821894650..34bcd74a9 100644 --- a/agents/src/voice/recorder_io/recorder_io.ts +++ b/agents/src/voice/recorder_io/recorder_io.ts @@ -679,16 +679,23 @@ class RecorderAudioOutput extends AudioOutput { continue; } - if (!segment.flushed) { - return; - } - + // A real finish from the downstream output is authoritative: the sink counted this + // segment and is now telling us it is over, so we settle it whether or not we have been + // flushed. The `AudioOutput` contract lets a sink report a finish as soon as its playout + // ends, with no flush involved, and `SyncedAudioOutput.waitForPlayout` does exactly that + // when it reconciles a segment the output below it dropped — which puts this on the + // default chain, not just custom sinks. The flush gate is only needed above, where we + // *synthesize* a finish and therefore have to know the segment can no longer grow. const event = this.deferredFinishes.shift(); if (event) { this.finishSegment(segment, event); continue; } + if (!segment.flushed) { + return; + } + 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 @@ -726,6 +733,13 @@ class RecorderAudioOutput extends AudioOutput { this.segments.shift(); if (this.currentSegment === segment) { this.currentSegment = undefined; + if (!segment.flushed) { + // Settled on the sink's own finish rather than at a flush boundary, so the base class + // still has this segment latched open. Release the latch, otherwise the next + // `captureFrame` neither counts a new base segment nor finds one of ours to attribute + // the frame to and throws `recorder capture has no active segment`. + this.abandonOpenSegment(); + } } const finishTime = segment.currentPauseStart ?? Date.now(); @@ -942,6 +956,14 @@ class RecorderAudioOutput extends AudioOutput { * 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. + * + * This also blocks while a frame is still in flight inside the wrapped output, where the + * pre-refactor code returned immediately with a fabricated + * `{ playbackPosition: 0, interrupted: false }` — the base class default, reachable only + * because the segment had not been registered yet. Registering before forwarding is what + * makes a finish arriving during that window attributable at all, so the wait necessarily + * sees the segment; reporting a turn as completed while its audio has not reached the sink + * would be the wrong answer anyway. */ async waitForPlayout(): Promise { const targetSegment = this.segments[this.segments.length - 1]; From 46a0c5db9540d06aaed87d6729cb0030739f58b7 Mon Sep 17 00:00:00 2001 From: Toubat Date: Sun, 26 Jul 2026 02:23:11 -0700 Subject: [PATCH 5/5] fix(recorder): settle a dropped segment for a caller that never flushed drainFinishes would only synthesize a finish for a segment the wrapped output never accepted once that segment had been flushed. The flush is there to prove the segment can no longer grow, but it is not the only thing that proves it: by the time waitForPlayout has awaited the wrapped output's own playout, that output is holding nothing for us, so a segment it never counted can never receive a finish from anyone. Mark the segment open at call time and let drainFinishes settle it on that instead. No in-tree caller could reach the stall. forwardAudio is the only code that captures frames into the output and it flushes in a finally (generation.ts), which is what made every wait recover; an AgentSession driven through a real ParticipantAudioOutput survives the interrupt-while-paused sequence on the previous head. But that is an accident of statement order inside somebody else's finally rather than a contract anyone is holding to, and it does not extend to the customised recording setups this bug was reported from. A caller that waits without flushing is asking a well-formed question and hanging is the wrong answer to it. The synthesized event is unchanged ({ playbackPosition: 0, interrupted: true }), so callers see the same result they saw once the flush eventually arrived, only without depending on it arriving. Segments with a capture still in flight are still held, and a segment the sink did accept still waits for its real finish. Tests: a dropped segment waited on with no flush, a segment dropped after the wait had already started, the same interrupt-while-paused turn as before minus the flush, and an AgentSession-level run of the customer's scenario through a real ParticipantAudioOutput. Also replaces an exact 0.1 playback-position assertion with a tolerance; the wall-clock clamp in finishSegment lands it a millisecond short often enough to flake. Co-authored-by: Cursor --- .../fix-recorder-capture-finish-race.md | 10 +- .../src/voice/recorder_io/recorder_io.test.ts | 113 ++++++++++++++++- agents/src/voice/recorder_io/recorder_io.ts | 37 ++++-- .../recorder_session_interrupt.test.ts | 114 ++++++++++++++++++ 4 files changed, 259 insertions(+), 15 deletions(-) create mode 100644 agents/src/voice/recorder_io/recorder_session_interrupt.test.ts diff --git a/.changeset/fix-recorder-capture-finish-race.md b/.changeset/fix-recorder-capture-finish-race.md index bb00f303b..0d7c1e5b6 100644 --- a/.changeset/fix-recorder-capture-finish-race.md +++ b/.changeset/fix-recorder-capture-finish-race.md @@ -17,9 +17,13 @@ output, so a caller that retries after a transient rejection is no longer reject A finish reported by the wrapped output settles its segment whether or not the recorder has been flushed. The `AudioOutput` contract lets a sink report a finish as soon as its playout ends, and `TranscriptionSynchronizer` does exactly that when it reconciles a dropped segment from -`waitForPlayout`, so requiring a flush first would strand the caller. Only a _synthesized_ finish — -for a segment the wrapped output never counted — still waits for the flush, which is what -guarantees the segment can no longer grow. +`waitForPlayout`, so requiring a flush first would strand the caller. + +`waitForPlayout` no longer depends on a flush either. A segment the wrapped output never counted +is settled once that output reports its own playout complete, since at that point no finish can +ever arrive for it. Waiting for a flush instead only worked because `performAudioForwarding` — the +one thing that captures frames — happens to flush in a `finally`; a caller that waited without +flushing hung forever. Behavior change: `waitForPlayout` now blocks while a frame is still in flight inside the wrapped output. Previously it could return immediately with a fabricated diff --git a/agents/src/voice/recorder_io/recorder_io.test.ts b/agents/src/voice/recorder_io/recorder_io.test.ts index 89edbca1a..c6f61aa42 100644 --- a/agents/src/voice/recorder_io/recorder_io.test.ts +++ b/agents/src/voice/recorder_io/recorder_io.test.ts @@ -326,6 +326,37 @@ class AcceptThenFinishOutput extends AudioOutput { clearBuffer(): void {} } +/** + * Parks the first frame at a gate and then *drops* it, the way `ParticipantAudioOutput` does + * when `clearBuffer()` releases a frame that was waiting on the pause gate. + */ +class ParkThenDropOutput extends AudioOutput { + readonly frameParked = new Future(); + private readonly gate = new Future(); + private captures = 0; + + constructor() { + super(24000); + } + + async captureFrame(frame: AudioFrame): Promise { + this.captures++; + if (this.captures === 1) { + this.frameParked.resolve(); + await this.gate.await; + return; + } + await super.captureFrame(frame); + } + + /** Releases the parked frame without ever counting it, exactly like an interrupted pause gate. */ + dropParkedFrame(): void { + this.gate.resolve(); + } + + clearBuffer(): void {} +} + /** Parks the first frame *before* counting it, the way `ParticipantAudioOutput`'s pause gate does. */ class ParkBeforeCountOutput extends AudioOutput { readonly frameParked = new Future(); @@ -594,6 +625,43 @@ describe('RecorderAudioOutput', () => { await recorder.close(); }); + it('settles a dropped segment for a caller that waits without flushing', async () => { + const recorder = new RecorderIO({ agentSession: {} as AgentSession }); + const output = recorder.recordOutput(new DroppingAudioOutput()); + + // No flush. Synthesizing a finish normally requires one, because the flush is what proves + // the segment can no longer grow — but once the wrapped output reports its own playout + // complete, a segment it never accepted can never be finished by anyone, so the wait would + // otherwise never end. + await output.captureFrame(makeFrame(20, 24000)); + + expect(await settleOrStall(output.waitForPlayout())).toEqual({ + playbackPosition: 0, + interrupted: true, + }); + await recorder.close(); + }); + + it('settles a segment the wrapped output drops after the wait has already started', async () => { + const recorder = new RecorderIO({ agentSession: {} as AgentSession }); + const downstream = new ParkThenDropOutput(); + const output = recorder.recordOutput(downstream); + + const capture = output.captureFrame(makeFrame(20, 24000)); + await downstream.frameParked.await; + + // The wait starts while the frame is still in flight, so nothing can be settled yet; the + // drop only becomes visible once the capture returns. + const wait = output.waitForPlayout(); + expect(await settleOrStall(wait)).toBe('did not settle'); + + downstream.dropParkedFrame(); + await capture; + + expect(await settleOrStall(wait)).toEqual({ playbackPosition: 0, interrupted: true }); + 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()); @@ -902,6 +970,18 @@ describe('RecorderAudioOutput in front of a real ParticipantAudioOutput', () => return out; } + /** + * The follow-on turn played its 100 ms frame to completion. `finishSegment` clamps the + * reported position against wall-clock elapsed time, which can land a hair under the pushed + * duration, so compare with a tolerance rather than exactly. + */ + function expectFollowOnTurnPlayed(result: PlaybackFinishedEvent | 'did not settle'): void { + expect(result).not.toBe('did not settle'); + const event = result as PlaybackFinishedEvent; + expect(event.interrupted).toBe(false); + expect(event.playbackPosition).toBeCloseTo(0.1, 2); + } + it('runs a follow-on turn after a turn interrupted while the output was paused', async () => { // The customer's report: an interrupt lands while a frame sits at the pause gate, and the // session never speaks again because the interrupted turn's `waitForPlayout` never settles. @@ -927,10 +1007,35 @@ describe('RecorderAudioOutput in front of a real ParticipantAudioOutput', () => await output.captureFrame(makeFrame(100, 24000)); output.flush(); - expect(await settleOrStall(output.waitForPlayout(), 3000)).toEqual({ - playbackPosition: 0.1, - interrupted: false, - }); + expectFollowOnTurnPlayed(await settleOrStall(output.waitForPlayout(), 3000)); + await recorder.close(); + }); + + it('runs a follow-on turn when the interrupted turn never flushed', async () => { + // Same interrupt, but the turn is torn down without a flush ever reaching us. Every + // in-tree caller happens to flush first (`forwardAudio` does it in a `finally`), but the + // wait must not depend on that: the sink dropped the frame, so no finish is coming and + // there is nothing left to wait for. + const recorder = new RecorderIO({ agentSession: {} as AgentSession }); + const output = recorder.recordOutput(makeParticipantAudioOutput()); + + await output.captureFrame(makeFrame(100, 24000)); + output.flush(); + + output.pause(); + const interruptedTurn = output.captureFrame(makeFrame(100, 24000)); + await new Promise((resolve) => setTimeout(resolve, 20)); + + output.clearBuffer(); + await interruptedTurn; + + expect(await settleOrStall(output.waitForPlayout(), 2000)).not.toBe('did not settle'); + + output.resume(); + await output.captureFrame(makeFrame(100, 24000)); + output.flush(); + + expectFollowOnTurnPlayed(await settleOrStall(output.waitForPlayout(), 3000)); await recorder.close(); }); }); diff --git a/agents/src/voice/recorder_io/recorder_io.ts b/agents/src/voice/recorder_io/recorder_io.ts index 34bcd74a9..d60856341 100644 --- a/agents/src/voice/recorder_io/recorder_io.ts +++ b/agents/src/voice/recorder_io/recorder_io.ts @@ -582,6 +582,14 @@ interface RecorderOutputSegment { capturesInFlight: number; finishRequested: boolean; flushed: boolean; + /** + * Set once a caller has waited for this segment's playout *and* the wrapped output has + * reported its own playout complete. At that point the wrapped output is holding nothing + * for us, so a segment it never accepted can no longer be finished by anyone — which is the + * same guarantee a flush gives {@link RecorderAudioOutput.drainFinishes}, arrived at from + * the other side. + */ + playoutAwaited: boolean; playbackEvent?: PlaybackFinishedEvent; /** Wall-clock time the segment was opened, i.e. when its first frame entered `captureFrame`. */ speechStartTime: number; @@ -665,14 +673,19 @@ class RecorderAudioOutput extends AudioOutput { if (!segment.acceptedDownstream) { // A segment the downstream output never counted will never receive a real finish, so - // 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) { + // we synthesize one. Before doing that we need to know the segment can no longer grow, + // or we would settle one that is still being captured into. A flush proves that, and + // so does `playoutAwaited`: a caller is waiting on this segment and the wrapped output + // has already reported its own playout done, so nothing is left that could finish it. + // + // Requiring the flush *alone* is not enough. It happens to hold for every in-tree + // caller today — `forwardAudio` flushes in a `finally` and is the only code that + // captures frames, `agent_activity.ts` awaits `cancelAndWait` on the forward tasks + // before it waits for playout, and `RecorderIO.close()` seals the open segment — but + // that is an accident of ordering inside somebody else's `finally`, not a contract. + // A caller that waits without flushing is asking a well-formed question, and hanging + // is the wrong answer to it. + if (!segment.flushed && !segment.playoutAwaited) { return; } this.finishSegment(segment, { playbackPosition: 0, interrupted: true }); @@ -884,6 +897,7 @@ class RecorderAudioOutput extends AudioOutput { capturesInFlight: 0, finishRequested: false, flushed: false, + playoutAwaited: 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 @@ -971,6 +985,13 @@ class RecorderAudioOutput extends AudioOutput { if (this.nextInChain) { await this.nextInChain.waitForPlayout(); } + if (targetSegment) { + // Marked only after the wrapped output's own wait returns, so this really does mean + // "nothing downstream is still holding this segment" and not merely "someone asked". + // Only the segment open at call time is marked: one opened later is not part of what + // this caller is waiting for, and settling it early would split the recording. + targetSegment.playoutAwaited = true; + } this.drainFinishes(); const event = await waitForRecorder; return targetSegment?.playbackEvent ?? event; diff --git a/agents/src/voice/recorder_io/recorder_session_interrupt.test.ts b/agents/src/voice/recorder_io/recorder_session_interrupt.test.ts new file mode 100644 index 000000000..521a8ba14 --- /dev/null +++ b/agents/src/voice/recorder_io/recorder_session_interrupt.test.ts @@ -0,0 +1,114 @@ +// SPDX-FileCopyrightText: 2026 LiveKit, Inc. +// +// SPDX-License-Identifier: Apache-2.0 +import { AudioFrame, type Room, TrackPublishOptions } from '@livekit/rtc-node'; +import { ReadableStream } from 'node:stream/web'; +import { describe, expect, it } from 'vitest'; +import { initializeLogger } from '../../log.js'; +import type { Future } from '../../utils.js'; +import { Agent } from '../agent.js'; +import { AgentSession } from '../agent_session.js'; +import { ParticipantAudioOutput } from '../room_io/_output.js'; +import { FakeLLM } from '../testing/fake_llm.js'; +import { RecorderIO } from './recorder_io.js'; + +/** + * The customer's scenario end to end, with the real pieces: a `RecorderIO`-wrapped + * `ParticipantAudioOutput` as `session.output.audio`, and the reply/interrupt sequence driven + * by the real `AgentActivity` and `forwardAudio` rather than by hand. + * + * The recorder unit tests pin the recorder's own behavior; this one pins that the pieces + * around it still let a session keep talking after a turn is interrupted while the output is + * paused — the failure mode that started this whole investigation was that they did not. + */ +function frame(durationMs = 20, sampleRate = 24000): AudioFrame { + const samples = Math.floor((sampleRate * durationMs) / 1000); + return new AudioFrame(new Int16Array(samples), sampleRate, 1, samples); +} + +/** Emits frames slowly enough that an interrupt can land mid-forwarding. */ +class FrameAgent extends Agent { + constructor() { + super({ instructions: 'test' }); + } + + async ttsNode(): Promise | null> { + return new ReadableStream({ + async start(controller) { + for (let i = 0; i < 25; i++) { + controller.enqueue(frame()); + await new Promise((resolve) => setTimeout(resolve, 5)); + } + controller.close(); + }, + }); + } +} + +function makeParticipantAudioOutput(): ParticipantAudioOutput { + const out = new ParticipantAudioOutput({} as Room, { + sampleRate: 24000, + numChannels: 1, + trackPublishOptions: new TrackPublishOptions(), + }); + // `publishTrack` normally resolves this; there is no room to publish to here. + (out as unknown as { startedFuture: Future }).startedFuture.resolve(); + return out; +} + +/** Surface a stall as a value, so a hang reads as "did not settle" and not as a bare timeout. */ +async function settleOrStall(promise: Promise, timeoutMs = 10000) { + let timer: NodeJS.Timeout | undefined; + const watchdog = new Promise<'did not settle'>((resolve) => { + timer = setTimeout(() => resolve('did not settle'), timeoutMs); + }); + try { + return await Promise.race([promise, watchdog]); + } finally { + if (timer) clearTimeout(timer); + } +} + +describe('AgentSession recording a real ParticipantAudioOutput', () => { + initializeLogger({ pretty: false, level: 'silent' }); + + it('keeps answering after a turn interrupted while the output was paused', async () => { + const session = new AgentSession({ + llm: new FakeLLM([ + { input: 'one', content: 'first spoken reply from the agent.' }, + { input: 'two', content: 'second spoken reply from the agent.' }, + { input: 'three', content: 'third spoken reply from the agent.' }, + ]), + }); + // Mirrors what `AgentSession.start` does when recording is enabled. + const recorder = new RecorderIO({ agentSession: session }); + session.output.audio = recorder.recordOutput(makeParticipantAudioOutput()); + + await session.start({ agent: new FrameAgent() }); + try { + // A plain interrupted turn. This also latches `ParticipantAudioOutput`'s + // `interruptedFuture`, which is what makes the next turn's frames droppable. + const first = session.generateReply({ userInput: 'one' }); + await new Promise((resolve) => setTimeout(resolve, 120)); + session.interrupt({ force: true }); + expect(await settleOrStall(first.waitForPlayout())).not.toBe('did not settle'); + + // The false-interruption pause (`agent_activity.ts` pauses the output when the user + // starts talking while the agent is thinking), followed by a confirmed interrupt. Frames + // sitting at the pause gate are released without the sink ever counting them. + const second = session.generateReply({ userInput: 'two' }); + await new Promise((resolve) => setTimeout(resolve, 60)); + session.output.audio!.pause(); + await new Promise((resolve) => setTimeout(resolve, 60)); + session.interrupt({ force: true }); + expect(await settleOrStall(second.waitForPlayout())).not.toBe('did not settle'); + + // The part the customer never got: the session still speaks. + session.output.audio!.resume(); + const third = session.generateReply({ userInput: 'three' }); + expect(await settleOrStall(third.waitForPlayout())).not.toBe('did not settle'); + } finally { + await settleOrStall(session.close(), 5000); + } + }, 60000); +});