fix(voice): commit interrupted replies when playback start arrives after forwarding completes (avatar outputs)

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

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'),
+          );
+        }
       }
     };
 

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<void> {
+    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<ReadableStream<AudioFrame> | null> {
+    this.onTtsStarted?.();
+    return new ReadableStream<AudioFrame>({
+      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();
+    }
+  });
 });

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

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 () => {