fix(voice): resume speech interrupted before its first frame (#1909)

.changeset/interrupt-before-first-frame.md

@@ -0,0 +1,9 @@
+---
+'@livekit/agents': patch
+---
+
+Fix agent speech being silently dropped when interrupted before its first audio frame plays (#1909, port of livekit/agents#5039).
+
+When the agent is in the "thinking" state and the user makes a brief sound before the first TTS frame is forwarded, `onStartOfSpeech` pauses the not-yet-playing speech (this thinking-state pause is intentional and preserved). The frames were still captured into the paused output buffer, but `forwardAudio`'s `finally` block rejected `firstFrameFut` (and removed its `PLAYBACK_STARTED` listener) whenever no frame had played yet. So when a false interruption cleared and the output resumed, the buffered first frame played but nothing was listening — the future stayed rejected, and because the reply tasks gate transcript preservation on `firstFrameFut.done && !firstFrameFut.rejected`, the resumed turn was dropped from history even though audio reached the user.
+
+The `PLAYBACK_STARTED` listener now lives in `performAudioForwarding` so it outlives the forwarding task, and `forwardAudio` no longer rejects the future; a late first frame (e.g. after a `resumeFalseInterruption` resume) can still resolve it. The reply tasks settle the future after playout finishes or is interrupted to remove the listener. A genuine interruption after a resume now keeps its partial synchronized transcript instead of losing the turn.

agents/src/voice/agent_activity.ts

@@ -2394,14 +2394,14 @@ export class AgentActivity implements RecognitionHooks {
       }
     };
 
+    let audioOut: _AudioOut | null = null;
     if (!audioOutput) {
       if (textOut) {
         textOut.firstTextFut.await
           .then(() => onFirstFrame(null))
           .catch(() => this.logger.debug('firstTextFut cancelled before first frame'));
       }
     } else {
-      let audioOut: _AudioOut | null = null;
       if (!audio) {
         // generate audio using TTS
         const [ttsTask, ttsGenData] = performTTSInference(
@@ -2442,53 +2442,72 @@ export class AgentActivity implements RecognitionHooks {
         .catch(() => this.logger.debug('firstFrameFut cancelled before first frame'));
     }
 
-    await speechHandle.waitIfNotInterrupted(tasks.map((task) => task.result));
-
-    if (audioOutput) {
-      await speechHandle.waitIfNotInterrupted([audioOutput.waitForPlayout()]);
-    }
+    try {
+      await speechHandle.waitIfNotInterrupted(tasks.map((task) => task.result));
 
-    if (speechHandle.interrupted) {
-      replyAbortController.abort();
-      await cancelAndWait(tasks, AgentActivity.REPLY_TASK_CANCEL_TIMEOUT);
       if (audioOutput) {
-        audioOutput.clearBuffer();
-        await audioOutput.waitForPlayout();
+        await speechHandle.waitIfNotInterrupted([audioOutput.waitForPlayout()]);
       }
-    }
 
-    if (addToChatCtx) {
-      const replyStoppedSpeakingAt = Date.now();
-      const replyAssistantMetrics: MetricsReport = {};
-      if (replyTtsGenData?.ttfb !== undefined) {
-        replyAssistantMetrics.ttsNodeTtfb = replyTtsGenData.ttfb;
+      if (speechHandle.interrupted) {
+        replyAbortController.abort();
+        await cancelAndWait(tasks, AgentActivity.REPLY_TASK_CANCEL_TIMEOUT);
+        if (audioOutput) {
+          audioOutput.clearBuffer();
+          await audioOutput.waitForPlayout();
+        }
       }
-      if (replyStartedSpeakingAt !== undefined) {
-        replyAssistantMetrics.startedSpeakingAt = replyStartedSpeakingAt / 1000; // ms -> seconds
-        replyAssistantMetrics.stoppedSpeakingAt = replyStoppedSpeakingAt / 1000; // ms -> seconds
 
-        if (replyStartedForwardingAt !== undefined) {
-          replyAssistantMetrics.playbackLatency =
-            (replyStartedSpeakingAt - replyStartedForwardingAt) / 1000; // ms -> seconds
+      if (addToChatCtx) {
+        const replyStoppedSpeakingAt = Date.now();
+        const replyAssistantMetrics: MetricsReport = {};
+        if (replyTtsGenData?.ttfb !== undefined) {
+          replyAssistantMetrics.ttsNodeTtfb = replyTtsGenData.ttfb;
         }
+        if (replyStartedSpeakingAt !== undefined) {
+          replyAssistantMetrics.startedSpeakingAt = replyStartedSpeakingAt / 1000; // ms -> seconds
+          replyAssistantMetrics.stoppedSpeakingAt = replyStoppedSpeakingAt / 1000; // ms -> seconds
+
+          if (replyStartedForwardingAt !== undefined) {
+            replyAssistantMetrics.playbackLatency =
+              (replyStartedSpeakingAt - replyStartedForwardingAt) / 1000; // ms -> seconds
+          }
+        }
+
+        const message = ChatMessage.create({
+          role: 'assistant',
+          content: textOut?.text || '',
+          interrupted: speechHandle.interrupted,
+          metrics: replyAssistantMetrics,
+        });
+        this.agent._chatCtx.insert(message);
+        this.agentSession._conversationItemAdded(message);
       }
 
-      const message = ChatMessage.create({
-        role: 'assistant',
-        content: textOut?.text || '',
-        interrupted: speechHandle.interrupted,
-        metrics: replyAssistantMetrics,
-      });
-      this.agent._chatCtx.insert(message);
-      this.agentSession._conversationItemAdded(message);
+      if (this.agentSession.agentState === 'speaking') {
+        this.agentSession._updateAgentState('listening');
+        if (this.audioRecognition) {
+          this.audioRecognition.onEndOfAgentSpeech(Date.now());
+        }
+        this.restoreInterruptionByAudioActivity();
+      }
+    } finally {
+      // In a finally so the listener is dropped even if an await above throws —
+      // otherwise it would leak on the shared audioOutput.
+      this.settleFirstFrameFut(audioOut);
     }
+  }
 
-    if (this.agentSession.agentState === 'speaking') {
-      this.agentSession._updateAgentState('listening');
-      if (this.audioRecognition) {
-        this.audioRecognition.onEndOfAgentSpeech(Date.now());
-      }
-      this.restoreInterruptionByAudioActivity();
+  /**
+   * Reject `firstFrameFut` if it is still pending, to drop the PLAYBACK_STARTED listener
+   * registered in `performAudioForwarding`. Called by reply tasks once playout has
+   * finished or been interrupted. `forwardAudio` no longer rejects the future itself so
+   * that a frame which plays late (e.g. after a false-interruption resume) can still
+   * resolve it (#1909).
+   */
+  private settleFirstFrameFut(audioOut: _AudioOut | null | undefined): void {
+    if (audioOut && !audioOut.firstFrameFut.done) {
+      audioOut.firstFrameFut.reject(new Error('audio forwarding finished before first frame'));
     }
   }
 
@@ -2832,6 +2851,10 @@ export class AgentActivity implements RecognitionHooks {
       } finally {
         replyAbortController.signal.removeEventListener('abort', abortSegment);
         await cancelAndWait(forwardTasks, AgentActivity.REPLY_TASK_CANCEL_TIMEOUT);
+        // A frame that played after a false-interruption resume keeps the synchronized
+        // transcript (gated on `firstFrameFut.done && !rejected` above) instead of being
+        // dropped.
+        this.settleFirstFrameFut(output.audioOut);
       }
     };
 
@@ -3389,6 +3412,7 @@ export class AgentActivity implements RecognitionHooks {
       } finally {
         abortController.signal.removeEventListener('abort', abortMessage);
         await cancelAndWait(forwardTasks, AgentActivity.REPLY_TASK_CANCEL_TIMEOUT);
+        this.settleFirstFrameFut(output.audioOut);
       }
     };
 

agents/src/voice/generation.ts

@@ -829,6 +829,13 @@ export interface _AudioOut {
    * metric.
    */
   startedForwardingAt?: number;
+  /**
+   * Whether this segment has captured at least one of its own frames. Read by the
+   * PLAYBACK_STARTED listener registered in `performAudioForwarding` to ignore a
+   * stray event emitted by another overlapping segment before this one has played.
+   * @internal
+   */
+  _hasCapturedOwnFrame?: boolean;
 }
 
 async function forwardAudio(
@@ -842,20 +849,7 @@ async function forwardAudio(
   const reader = ttsStream.getReader();
   let resampler: AudioResampler | null = null;
 
-  // The audio output is shared across overlapping segments, so ignore a
-  // PLAYBACK_STARTED from another segment until we capture our own first frame.
-  // Resolving `firstFrameFut` early skips resampler creation and pushes an
-  // unresampled frame (`RtcError: sample_rate and num_channels don't match`).
-  let hasCapturedOwnFrame = false;
-
-  const onPlaybackStarted = (ev: { createdAt: number }) => {
-    if (hasCapturedOwnFrame && !out.firstFrameFut.done) {
-      out.firstFrameFut.resolve(ev.createdAt);
-    }
-  };
-
   try {
-    audioOutput.on(AudioOutput.EVENT_PLAYBACK_STARTED, onPlaybackStarted);
     audioOutput.resume();
 
     while (true) {
@@ -876,8 +870,9 @@ async function forwardAudio(
       }
 
       // Mark before capturing so the PLAYBACK_STARTED emitted synchronously inside
-      // the first captureFrame is attributed to this segment.
-      hasCapturedOwnFrame = true;
+      // the first captureFrame is attributed to this segment (see the listener in
+      // `performAudioForwarding`).
+      out._hasCapturedOwnFrame = true;
 
       if (resampler) {
         for (const f of resampler.push(frame)) {
@@ -900,12 +895,15 @@ async function forwardAudio(
       throw e;
     }
   } finally {
-    audioOutput.off(AudioOutput.EVENT_PLAYBACK_STARTED, onPlaybackStarted);
-
-    if (!out.firstFrameFut.done) {
-      out.firstFrameFut.reject(new Error('audio forwarding cancelled before playback started'));
-    }
-
+    // NOTE: `firstFrameFut` is intentionally NOT rejected here. The frames may have
+    // been captured into a paused output (agent interrupted in the thinking state
+    // before its first frame played); playback then starts only once the output is
+    // resumed — possibly after this forwarding task has already finished. The
+    // PLAYBACK_STARTED listener registered in `performAudioForwarding` outlives this
+    // task and resolves the future when that late first frame finally plays, so a
+    // false interruption resumes and plays instead of dropping the turn. The caller
+    // settles the future (and removes the listener) after playout finishes or is
+    // interrupted. See #1909.
     reader?.releaseLock();
     audioOutput.flush();
     if (signal?.aborted) {
@@ -926,6 +924,27 @@ export function performAudioForwarding(
     firstFrameFut: new Future<number>(),
   };
 
+  // Resolve `firstFrameFut` from the output's PLAYBACK_STARTED event. Registered here
+  // (rather than inside `forwardAudio`) so the listener outlives the forwarding task:
+  // when a not-yet-playing speech is paused in the thinking state its frames are
+  // buffered and the forwarding task may finish before playback ever starts. The late
+  // first frame — e.g. once a false interruption clears and the output resumes — must
+  // still resolve the future instead of being dropped (#1909).
+  const onPlaybackStarted = (ev: { createdAt: number }) => {
+    // Ignore a PLAYBACK_STARTED from another overlapping segment until this segment
+    // has captured its own first frame; resolving early would skip resampler creation.
+    if (out._hasCapturedOwnFrame && !out.firstFrameFut.done) {
+      out.firstFrameFut.resolve(ev.createdAt);
+    }
+  };
+  audioOutput.on(AudioOutput.EVENT_PLAYBACK_STARTED, onPlaybackStarted);
+  // Remove the listener once the future settles — resolved by playback above, or
+  // rejected by the caller (via `firstFrameFut.reject`) after playout finishes or is
+  // interrupted without a frame ever playing.
+  const removeListener = () =>
+    audioOutput.off(AudioOutput.EVENT_PLAYBACK_STARTED, onPlaybackStarted);
+  out.firstFrameFut.await.then(removeListener).catch(removeListener);
+
   return [
     Task.from(
       (controller) => forwardAudio(ttsStream, audioOutput, out, idleTimeout, controller.signal),