diff --git a/.changeset/avatar-deferred-playback-started-commit.md b/.changeset/avatar-deferred-playback-started-commit.md new file mode 100644 index 000000000..5b4525ed5 --- /dev/null +++ b/.changeset/avatar-deferred-playback-started-commit.md @@ -0,0 +1,5 @@ +--- +'@livekit/agents': patch +--- + +fix(voice): commit interrupted replies when playback start is reported after audio forwarding completes (avatar outputs). `forwardAudio` no longer rejects `firstFrameFut` when forwarding finishes before the remote avatar's deferred `lk.playback_started` notification arrives — previously the interrupted reply was classified "skipped" and silently dropped from the chat context, and the agent never entered the `speaking` state. A reported non-zero playback position on interruption is now also honored as evidence of partial playback. diff --git a/agents/src/voice/agent_activity.ts b/agents/src/voice/agent_activity.ts index 656646681..8b657db65 100644 --- a/agents/src/voice/agent_activity.ts +++ b/agents/src/voice/agent_activity.ts @@ -2895,7 +2895,18 @@ export class AgentActivity implements RecognitionHooks { if (audioOutput) { audioOutput.clearBuffer(); const interruptedPlaybackEv = await audioOutput.waitForPlayout(); - if (output.audioOut?.firstFrameFut.done && !output.audioOut.firstFrameFut.rejected) { + // A reported playback position is proof of partial playback even when + // the playback-started notification hasn't arrived yet (remote avatar + // outputs deliver it via RPC, which can race with the interruption). + // It only counts as evidence when THIS segment forwarded audio: + // waitForPlayout returns the last playback event, which is stale + // (from a previous segment) when no frames were captured for the + // current one. + const forwardedAudio = output.audioOut?.startedForwardingAt !== undefined; + if ( + (output.audioOut?.firstFrameFut.done && !output.audioOut.firstFrameFut.rejected) || + (forwardedAudio && interruptedPlaybackEv.playbackPosition > 0) + ) { output.played = 'partial'; output.playbackPositionInS = interruptedPlaybackEv.playbackPosition; output.synchronizedTranscript = interruptedPlaybackEv.synchronizedTranscript; @@ -2917,6 +2928,13 @@ export class AgentActivity implements RecognitionHooks { } finally { replyAbortController.signal.removeEventListener('abort', abortSegment); await cancelAndWait(forwardTasks, AgentActivity.REPLY_TASK_CANCEL_TIMEOUT); + // The segment's playout window is over; settle a still-pending + // firstFrameFut so forwardAudio's playback-started listener is detached. + if (output.audioOut && !output.audioOut.firstFrameFut.done) { + output.audioOut.firstFrameFut.reject( + new Error('playout finished before playback started'), + ); + } } }; @@ -3449,7 +3467,18 @@ export class AgentActivity implements RecognitionHooks { if (audioOutput) { audioOutput.clearBuffer(); const playbackEv = await audioOutput.waitForPlayout(); - if (output.audioOut?.firstFrameFut.done && !output.audioOut.firstFrameFut.rejected) { + // A reported playback position is proof of partial playback even when + // the playback-started notification hasn't arrived yet (remote avatar + // outputs deliver it via RPC, which can race with the interruption). + // It only counts as evidence when THIS message forwarded audio: + // waitForPlayout returns the last playback event, which is stale + // (from a previous message) when no frames were captured for the + // current one. + const forwardedAudio = output.audioOut?.startedForwardingAt !== undefined; + if ( + (output.audioOut?.firstFrameFut.done && !output.audioOut.firstFrameFut.rejected) || + (forwardedAudio && playbackEv.playbackPosition > 0) + ) { output.played = 'partial'; output.playbackPositionInS = playbackEv.playbackPosition; output.synchronizedTranscript = playbackEv.synchronizedTranscript; @@ -3471,6 +3500,13 @@ export class AgentActivity implements RecognitionHooks { } finally { abortController.signal.removeEventListener('abort', abortMessage); await cancelAndWait(forwardTasks, AgentActivity.REPLY_TASK_CANCEL_TIMEOUT); + // The segment's playout window is over; settle a still-pending + // firstFrameFut so forwardAudio's playback-started listener is detached. + if (output.audioOut && !output.audioOut.firstFrameFut.done) { + output.audioOut.firstFrameFut.reject( + new Error('playout finished before playback started'), + ); + } } }; diff --git a/agents/src/voice/agent_activity_interrupted_commit.test.ts b/agents/src/voice/agent_activity_interrupted_commit.test.ts index 3eeb53e92..90b7b2ddd 100644 --- a/agents/src/voice/agent_activity_interrupted_commit.test.ts +++ b/agents/src/voice/agent_activity_interrupted_commit.test.ts @@ -42,6 +42,37 @@ class InterruptibleOutput extends AudioOutput { } } +// Audio sink that mimics a DataStream avatar output (waitPlaybackStart: true): +// frames are accepted faster than real time and playback-started is only +// reported LATER, via an out-of-band notification (the `lk.playback_started` +// RPC from the remote avatar worker) — typically after forwarding has already +// completed. `notifyPlaybackStarted` simulates that RPC arriving. +class DeferredStartOutput extends AudioOutput { + onFirstFrameCaptured?: () => void; + playbackPosition = 0; + private captured = false; + constructor() { + super(24000); + } + async captureFrame(f: AudioFrame): Promise { + await super.captureFrame(f); + if (!this.captured) { + this.captured = true; + this.onFirstFrameCaptured?.(); + } + } + // The remote avatar reports playback start well after frames were forwarded. + notifyPlaybackStarted(): void { + this.onPlaybackStarted(Date.now()); + } + flush(): void { + super.flush(); + } + clearBuffer(): void { + this.onPlaybackFinished({ playbackPosition: this.playbackPosition, interrupted: true }); + } +} + // Agent that synthesizes a few real frames for whatever text it receives, so the // audio path produces a first frame (no TTS provider needed). class FrameAgent extends Agent { @@ -58,6 +89,24 @@ class FrameAgent extends Agent { } } +// Agent whose TTS produces no frames and stays open long enough that an +// interruption (anchored to ttsNode being invoked, fired well before the +// close) always lands mid-forwarding with zero audio captured for the segment. +class NoFrameAgent extends Agent { + onTtsStarted?: () => void; + constructor() { + super({ instructions: 'test' }); + } + async ttsNode(): Promise | null> { + this.onTtsStarted?.(); + return new ReadableStream({ + start(controller) { + setTimeout(() => controller.close(), 500); + }, + }); + } +} + describe('AgentActivity interrupted-speech commit', () => { initializeLogger({ pretty: false, level: 'silent' }); @@ -97,4 +146,137 @@ describe('AgentActivity interrupted-speech commit', () => { await session.close(); } }); + + it('commits an interrupted reply when playback start is reported after forwarding completes (avatar output)', async () => { + const session = new AgentSession({ + llm: new FakeLLM([{ input: 'hello', content: 'A fairly long spoken reply.' }]), + }); + const audioOut = new DeferredStartOutput(); + session.output.audio = audioOut; + + // Simulate the avatar pipeline: playback starts (RPC arrives) shortly after + // the frames were forwarded, then the user interrupts mid-playback. + audioOut.onFirstFrameCaptured = () => { + setTimeout(() => { + audioOut.playbackPosition = 0.12; + audioOut.notifyPlaybackStarted(); + setTimeout(() => session.interrupt({ force: true }), 50); + }, 100); + }; + + const assistantMessages: { content: string; interrupted: boolean }[] = []; + session.on(AgentSessionEventTypes.ConversationItemAdded, (ev: ConversationItemAddedEvent) => { + if (ev.item.type === 'message' && ev.item.role === 'assistant') { + assistantMessages.push({ + content: ev.item.textContent ?? '', + interrupted: !!ev.item.interrupted, + }); + } + }); + + await session.start({ agent: new FrameAgent() }); + try { + const handle = session.generateReply({ userInput: 'hello' }); + await handle.waitForPlayout(); + await vi.waitFor(() => expect(assistantMessages.length).toBeGreaterThan(0)); + + // Pre-fix, forwardAudio rejected firstFrameFut the moment forwarding + // finished (before the deferred playback-started notification), so the + // audibly-played reply was classified "skipped" and silently dropped from + // the chat context (phantom utterance). + const msg = assistantMessages[0]!; + expect(msg.interrupted).toBe(true); + expect(msg.content.length).toBeGreaterThan(0); + expect('A fairly long spoken reply.'.startsWith(msg.content)).toBe(true); + } finally { + await session.close(); + } + }); + + it('commits an interrupted reply when only a playback position is reported (no playback-started signal)', async () => { + const session = new AgentSession({ + llm: new FakeLLM([{ input: 'hello', content: 'A fairly long spoken reply.' }]), + }); + const audioOut = new DeferredStartOutput(); + session.output.audio = audioOut; + + // The remote avatar never reports playback-started (or the RPC races with + // the interruption), but its playback-finished response carries a non-zero + // position — proof the user heard part of the reply. + audioOut.onFirstFrameCaptured = () => { + setTimeout(() => { + audioOut.playbackPosition = 0.25; + session.interrupt({ force: true }); + }, 100); + }; + + const assistantMessages: { content: string; interrupted: boolean }[] = []; + session.on(AgentSessionEventTypes.ConversationItemAdded, (ev: ConversationItemAddedEvent) => { + if (ev.item.type === 'message' && ev.item.role === 'assistant') { + assistantMessages.push({ + content: ev.item.textContent ?? '', + interrupted: !!ev.item.interrupted, + }); + } + }); + + await session.start({ agent: new FrameAgent() }); + try { + const handle = session.generateReply({ userInput: 'hello' }); + await handle.waitForPlayout(); + await vi.waitFor(() => expect(assistantMessages.length).toBeGreaterThan(0)); + + const msg = assistantMessages[0]!; + expect(msg.interrupted).toBe(true); + expect(msg.content.length).toBeGreaterThan(0); + expect('A fairly long spoken reply.'.startsWith(msg.content)).toBe(true); + } finally { + await session.close(); + } + }); + + it('does not commit an interrupted reply that forwarded no audio when a stale playback position exists', async () => { + const session = new AgentSession({ + llm: new FakeLLM([{ input: 'hello', content: 'A reply that was never spoken.' }]), + }); + const audioOut = new DeferredStartOutput(); + session.output.audio = audioOut; + + // A previously played segment leaves a non-zero lastPlaybackEvent behind. + // waitForPlayout returns it immediately for a segment that captured no + // frames — the playback position must not be attributed to that segment. + await audioOut.captureFrame(frame()); + audioOut.flush(); + audioOut.onPlaybackFinished({ playbackPosition: 0.5, interrupted: false }); + + const assistantMessages: { content: string; interrupted: boolean }[] = []; + session.on(AgentSessionEventTypes.ConversationItemAdded, (ev: ConversationItemAddedEvent) => { + if (ev.item.type === 'message' && ev.item.role === 'assistant') { + assistantMessages.push({ + content: ev.item.textContent ?? '', + interrupted: !!ev.item.interrupted, + }); + } + }); + + const agent = new NoFrameAgent(); + // Anchor the interruption to TTS starting, so it always lands while the + // segment is mid-forwarding (zero frames captured, stream still open). + agent.onTtsStarted = () => setTimeout(() => session.interrupt({ force: true }), 150); + + await session.start({ agent }); + try { + const handle = session.generateReply({ userInput: 'hello' }); + await handle.waitForPlayout(); + + // Close drains the reply task fully — the interrupted-commit block (if + // it were to run) has fired by the time close resolves. Nothing was + // heard: the reply must not be committed on the strength of the stale + // playback position. + await session.close(); + expect(assistantMessages).toHaveLength(0); + } finally { + await session.close(); + } + }); }); diff --git a/agents/src/voice/generation.ts b/agents/src/voice/generation.ts index 3d6a4c831..9a6529b30 100644 --- a/agents/src/voice/generation.ts +++ b/agents/src/voice/generation.ts @@ -936,11 +936,25 @@ async function forwardAudio( throw e; } } finally { - audioOutput.off(AudioOutput.EVENT_PLAYBACK_STARTED, onPlaybackStarted); - - if (!out.firstFrameFut.done) { + // Forwarding completes as soon as all TTS frames are *captured*, which can be + // long before remote playback begins: DataStream avatar outputs (e.g. + // LemonSlice, with `waitPlaybackStart: true`) accept frames faster than + // real time and only send the `lk.playback_started` RPC once the avatar + // worker actually starts playing (typically ~1s later). Rejecting + // `firstFrameFut` here would drop that deferred event on the floor: the + // agent never enters the "speaking" state and an interrupted reply is + // treated as "skipped" and silently removed from the chat context (phantom + // utterance). Mirror the python implementation instead: only reject when + // forwarding was aborted before playback started, and keep the listener + // attached until the future settles. `forwardSegment` settles any future + // still pending once the segment's playout window ends, bounding the + // listener's lifetime. + if (signal?.aborted && !out.firstFrameFut.done) { out.firstFrameFut.reject(new Error('audio forwarding cancelled before playback started')); } + out.firstFrameFut.await + .catch(() => {}) + .finally(() => audioOutput.off(AudioOutput.EVENT_PLAYBACK_STARTED, onPlaybackStarted)); reader?.releaseLock(); audioOutput.flush(); diff --git a/agents/src/voice/generation_tts_timeout.test.ts b/agents/src/voice/generation_tts_timeout.test.ts index 5cb3803f9..14e31b791 100644 --- a/agents/src/voice/generation_tts_timeout.test.ts +++ b/agents/src/voice/generation_tts_timeout.test.ts @@ -115,7 +115,6 @@ describe('TTS stream idle timeout', () => { const controller = new AbortController(); const [task, audioOut] = performAudioForwarding(stalledStream, audioOutput, controller, 500); - // Reject path is expected (no first frame ever captured). audioOut.firstFrameFut.await.catch(() => {}); vi.useFakeTimers(); @@ -131,7 +130,12 @@ describe('TTS stream idle timeout', () => { vi.useRealTimers(); expect(audioOutput.capturedFrames.length).toBe(0); - expect(audioOut.firstFrameFut.rejected).toBe(true); + // Forwarding ended without capturing a frame. The future stays pending — + // playback-started may legitimately arrive after forwarding completes + // (deferred avatar notification), so the caller settles it when the + // segment's playout window ends. Stray events must still be ignored. + audioOutput.onPlaybackStarted(Date.now()); + expect(audioOut.firstFrameFut.done).toBe(false); }); it('resamples a rate-mismatched frame even after a stray PLAYBACK_STARTED', async () => {