From 67514f57a811761914b5e5e220c2f8acc6592dd8 Mon Sep 17 00:00:00 2001 From: Chenghao Mou Date: Sat, 25 Jul 2026 00:15:20 +0100 Subject: [PATCH] fix(recorder_io): settle playout waits when the downstream output finishes or drops the opening frame RecorderAudioOutput forwards a frame downstream before registering its own playback segment, so a downstream playback_finished landing during the opening capture was discarded by the base class and waitForPlayout() never settled. A frame the downstream output drops entirely left the same dangling segment. Buffer finish events that arrive mid-capture and replay them once the segment is registered, and queue a synthetic interrupted finish for segments the downstream output never counted, settled in waitForPlayout() after the downstream output drains. The synthetic finish is revoked if the downstream output starts accepting frames mid-segment, so the segment isn't settled twice. Fixes AGT-3177 Co-Authored-By: Claude --- .changeset/steady-recorders-finish.md | 5 + .../src/voice/recorder_io/recorder_io.test.ts | 231 +++++++++++++++++- agents/src/voice/recorder_io/recorder_io.ts | 72 +++++- 3 files changed, 296 insertions(+), 12 deletions(-) create mode 100644 .changeset/steady-recorders-finish.md diff --git a/.changeset/steady-recorders-finish.md b/.changeset/steady-recorders-finish.md new file mode 100644 index 000000000..0aef8eaa2 --- /dev/null +++ b/.changeset/steady-recorders-finish.md @@ -0,0 +1,5 @@ +--- +'@livekit/agents': patch +--- + +Prevent recorder output wrappers from deadlocking playout waits when interruption finishes during first-frame capture. diff --git a/agents/src/voice/recorder_io/recorder_io.test.ts b/agents/src/voice/recorder_io/recorder_io.test.ts index 27a5f7448..ec4c47bd0 100644 --- a/agents/src/voice/recorder_io/recorder_io.test.ts +++ b/agents/src/voice/recorder_io/recorder_io.test.ts @@ -5,12 +5,12 @@ 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'; 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,32 @@ class WaitAwareAudioOutput extends FakeAudioOutput { } } +interface CaptureStep { + accept: boolean; + finish?: PlaybackFinishedEvent; +} + +class ScriptedAudioOutput extends AudioOutput { + constructor(private readonly captureSteps: CaptureStep[]) { + super(24000); + } + + async captureFrame(frame: AudioFrame) { + const step = this.captureSteps.shift(); + if (!step) { + throw new Error('unexpected audio capture'); + } + if (step.accept) { + await super.captureFrame(frame); + } + if (step.finish) { + this.onPlaybackFinished(step.finish); + } + } + + clearBuffer(): void {} +} + function makeFrame(durationMs: number, sampleRate = 48000, channels = 1): AudioFrame { const samplesPerChannel = Math.floor((durationMs / 1000) * sampleRate); return new AudioFrame( @@ -139,6 +165,207 @@ describe('RecorderIO close', () => { }); describe('RecorderAudioOutput', () => { + it('settles when playback finishes during first-frame capture', async () => { + const recorder = new RecorderIO({ agentSession: {} as AgentSession }); + const output = recorder.recordOutput( + new ScriptedAudioOutput([ + { accept: true, finish: { playbackPosition: 0, interrupted: true } }, + ]), + ); + + await output.captureFrame(makeFrame(20, 24000)); + output.flush(); + + await expect(output.waitForPlayout()).resolves.toEqual({ + playbackPosition: 0, + interrupted: true, + }); + }, 1000); + + it('settles as interrupted when the downstream output drops the first frame', async () => { + const recorder = new RecorderIO({ agentSession: {} as AgentSession }); + const output = recorder.recordOutput(new ScriptedAudioOutput([{ accept: false }])); + + await output.captureFrame(makeFrame(20, 24000)); + output.flush(); + + await expect(output.waitForPlayout()).resolves.toEqual({ + playbackPosition: 0, + interrupted: true, + }); + }, 1000); + + it('settles subsequent segments after an early first-frame finish', async () => { + const recorder = new RecorderIO({ agentSession: {} as AgentSession }); + const finish = { playbackPosition: 0, interrupted: true }; + const output = recorder.recordOutput( + new ScriptedAudioOutput([ + { accept: true, finish }, + { accept: true, finish }, + ]), + ); + + for (let i = 0; i < 2; i++) { + await output.captureFrame(makeFrame(20, 24000)); + output.flush(); + await expect(output.waitForPlayout()).resolves.toEqual({ + playbackPosition: 0, + interrupted: true, + }); + } + }); + + it('settles the next segment after the downstream output drops a frame', async () => { + const recorder = new RecorderIO({ agentSession: {} as AgentSession }); + const downstream = new ScriptedAudioOutput([{ accept: false }, { accept: true }]); + const output = recorder.recordOutput(downstream); + + await output.captureFrame(makeFrame(20, 24000)); + output.flush(); + await expect(output.waitForPlayout()).resolves.toEqual({ + playbackPosition: 0, + interrupted: true, + }); + + await output.captureFrame(makeFrame(20, 24000)); + output.flush(); + const waitForSecondSegment = output.waitForPlayout(); + downstream.onPlaybackFinished({ playbackPosition: 0, interrupted: false }); + + await expect(waitForSecondSegment).resolves.toEqual({ + playbackPosition: 0, + interrupted: false, + }); + }); + + it('preserves a previous completion that arrives during the next capture', async () => { + const recorder = new RecorderIO({ agentSession: {} as AgentSession }); + const downstream = new ScriptedAudioOutput([ + { accept: true }, + { accept: true, finish: { playbackPosition: 0, interrupted: false } }, + ]); + const output = recorder.recordOutput(downstream); + + await output.captureFrame(makeFrame(20, 24000)); + output.flush(); + const waitForFirstSegment = output.waitForPlayout(); + + await output.captureFrame(makeFrame(20, 24000)); + await expect(waitForFirstSegment).resolves.toEqual({ + playbackPosition: 0, + interrupted: false, + }); + + output.flush(); + const waitForSecondSegment = output.waitForPlayout(); + downstream.onPlaybackFinished({ playbackPosition: 0, interrupted: true }); + await expect(waitForSecondSegment).resolves.toEqual({ + playbackPosition: 0, + interrupted: true, + }); + }, 1000); + + it('settles a dropped segment after a previous completion arrives during its capture', async () => { + const recorder = new RecorderIO({ agentSession: {} as AgentSession }); + const downstream = new ScriptedAudioOutput([ + { accept: true }, + { + accept: false, + finish: { playbackPosition: 0, interrupted: false }, + }, + ]); + const output = recorder.recordOutput(downstream); + + await output.captureFrame(makeFrame(20, 24000)); + output.flush(); + const waitForFirstSegment = output.waitForPlayout(); + + await output.captureFrame(makeFrame(20, 24000)); + output.flush(); + + await expect(waitForFirstSegment).resolves.toBeDefined(); + await expect(output.waitForPlayout()).resolves.toEqual({ + playbackPosition: 0, + interrupted: true, + }); + }, 1000); + + it('preserves a delayed previous completion before settling a dropped segment', async () => { + const recorder = new RecorderIO({ agentSession: {} as AgentSession }); + const downstream = new ScriptedAudioOutput([{ accept: true }, { accept: false }]); + const output = recorder.recordOutput(downstream); + + await output.captureFrame(makeFrame(20, 24000)); + output.flush(); + const waitForFirstSegment = output.waitForPlayout(); + + await output.captureFrame(makeFrame(20, 24000)); + output.flush(); + const waitForSecondSegment = output.waitForPlayout(); + + downstream.onPlaybackFinished({ playbackPosition: 0, interrupted: false }); + + await expect(waitForFirstSegment).resolves.toBeDefined(); + await expect(waitForSecondSegment).resolves.toEqual({ + playbackPosition: 0, + interrupted: true, + }); + }, 1000); + + it('keeps an owed completion reported during the capture of a dropped frame', async () => { + const recorder = new RecorderIO({ agentSession: {} as AgentSession }); + // The downstream output segments finer than this wrapper, so it still owes a completion for + // its second segment once the wrapper's first segment has settled. + const downstream = new ScriptedAudioOutput([ + { accept: true }, + { accept: true }, + { accept: false, finish: { playbackPosition: 0.5, interrupted: false } }, + ]); + const output = recorder.recordOutput(downstream); + + await output.captureFrame(makeFrame(20, 24000)); + downstream.flush(); + await output.captureFrame(makeFrame(20, 24000)); + output.flush(); + downstream.onPlaybackFinished({ playbackPosition: 0, interrupted: false }); + + // The owed completion lands while the frame that opens the next segment is dropped: it must + // settle that segment instead of being replaced by a synthetic interrupted finish. + await output.captureFrame(makeFrame(20, 24000)); + output.flush(); + + await expect(output.waitForPlayout()).resolves.toEqual({ + playbackPosition: 0, + interrupted: false, + }); + }, 1000); + + it('settles once when the downstream output accepts a later frame of the segment', async () => { + const recorder = new RecorderIO({ agentSession: {} as AgentSession }); + const downstream = new ScriptedAudioOutput([{ accept: false }, { accept: true }]); + const output = recorder.recordOutput(downstream); + const warn = vi.spyOn( + (output as unknown as { logger: { warn: (...args: unknown[]) => void } }).logger, + 'warn', + ); + + // The opening frame is dropped downstream, the second one is accepted, so the downstream + // output reports a real finish for a segment this wrapper had written off as dropped. + await output.captureFrame(makeFrame(20, 24000)); + await output.captureFrame(makeFrame(20, 24000)); + output.flush(); + downstream.onPlaybackFinished({ playbackPosition: 0, interrupted: false }); + + await expect(output.waitForPlayout()).resolves.toEqual({ + playbackPosition: 0, + interrupted: false, + }); + expect(warn).not.toHaveBeenCalledWith( + expect.stringContaining('playback_finished called more times'), + ); + warn.mockRestore(); + }, 1000); + 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..a54fad104 100644 --- a/agents/src/voice/recorder_io/recorder_io.ts +++ b/agents/src/voice/recorder_io/recorder_io.ts @@ -575,6 +575,10 @@ class RecorderAudioOutput extends AudioOutput { private accFrames: AudioFrame[] = []; private _startedWallTime?: number; private _logger = log(); + // Downstream playback events can fire before this wrapper registers its segment. + private captureDeferredPlaybackFinished?: PlaybackFinishedEvent[]; + // Dropped segments settle only after earlier downstream segments complete. + private droppedPlayoutSegments: number[] = []; _lastSpeechEndTime?: number; private _lastSpeechStartTime?: number; @@ -631,6 +635,11 @@ class RecorderAudioOutput extends AudioOutput { } onPlaybackFinished(options: PlaybackFinishedEvent): void { + if (this.captureDeferredPlaybackFinished && this.pendingPlayoutSegments === 0) { + this.captureDeferredPlaybackFinished.push(options); + return; + } + const finishTime = this.currentPauseStart ?? Date.now(); const trailingSilenceDuration = Math.max(0, Date.now() - finishTime); @@ -771,30 +780,73 @@ class RecorderAudioOutput extends AudioOutput { } async captureFrame(frame: AudioFrame): Promise { - if (this.nextInChain) { - await this.nextInChain.captureFrame(frame); - } + const downstreamCapturedBefore = this.nextInChain?.capturedPlayoutSegments; + const recorderCapturedBefore = this.capturedPlayoutSegments; + const deferredPlaybackFinished: PlaybackFinishedEvent[] = []; + + this.captureDeferredPlaybackFinished = deferredPlaybackFinished; + try { + if (this.nextInChain) { + await this.nextInChain.captureFrame(frame); + } + + await super.captureFrame(frame); - await super.captureFrame(frame); + if (this.recorderIO.recording) { + this.accFrames.push(frame); + } + + if (this._startedWallTime === undefined) { + this._startedWallTime = Date.now(); + } - if (this.recorderIO.recording) { - this.accFrames.push(frame); + if (this._lastSpeechStartTime === undefined) { + this._lastSpeechStartTime = Date.now(); + } + } finally { + this.captureDeferredPlaybackFinished = undefined; } - if (this._startedWallTime === undefined) { - this._startedWallTime = Date.now(); + // Replay before the drop check below: a deferred event settles the segment this frame just + // opened, which means the segment must not also be queued for a synthetic finish. + for (const event of deferredPlaybackFinished) { + this.onPlaybackFinished(event); } - if (this._lastSpeechStartTime === undefined) { - this._lastSpeechStartTime = Date.now(); + const segment = this.capturedPlayoutSegments; + const downstreamAcceptedSegment = + this.nextInChain === undefined || + this.nextInChain.capturedPlayoutSegments > (downstreamCapturedBefore ?? 0); + + if (downstreamAcceptedSegment) { + // The downstream output may start accepting frames mid-segment (e.g. a paused output that + // dropped the opening frame and then resumed). Its real finish now covers this segment, so + // drop the queued synthetic one instead of settling the segment twice. + const queued = this.droppedPlayoutSegments.indexOf(segment); + if (queued >= 0) { + this.droppedPlayoutSegments.splice(queued, 1); + } + return; + } + + const openedRecorderSegment = segment > recorderCapturedBefore; + if (openedRecorderSegment && this.pendingPlayoutSegments > 0) { + this.droppedPlayoutSegments.push(segment); } } async waitForPlayout(): Promise { + const target = this.capturedPlayoutSegments; const waitForRecorder = super.waitForPlayout(); if (this.nextInChain) { await this.nextInChain.waitForPlayout(); } + + while (this.droppedPlayoutSegments.length > 0 && this.droppedPlayoutSegments[0]! <= target) { + this.droppedPlayoutSegments.shift(); + this.onPlaybackFinished({ playbackPosition: 0, interrupted: true }); + } + return waitForRecorder; }