Skip to content

fix(voice): stop dropping turns whose playback starts after audio forwarding completes (combines #1910 + #1960) #1966

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
toubatbrian wants to merge 3 commits into
base: main
Choose a base branch
from
Open

fix(voice): stop dropping turns whose playback starts after audio forwarding completes (combines #1910 + #1960) #1966

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
aede2a9
fix(voice): stop dropping turns whose playback starts after audio for…
toubatbrian Jul 6, 2026
43def48
fix(voice): gate interrupted-commit playback evidence on captured seg…
toubatbrian Jul 7, 2026
4de626d
fix(test): handle race rejection in paused-gate test for throws-check
toubatbrian Jul 7, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
9 changes: 9 additions & 0 deletions .changeset/late-playback-started-dropped-turns.md
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -0,0 +1,9 @@
---
'@livekit/agents': patch
---

fix(voice): stop dropping agent turns whose playback starts after audio forwarding completes (#1909, #1960; port of livekit/agents#5039).

`forwardAudio` used to reject `firstFrameFut` (and detach its `PLAYBACK_STARTED` listener) in its `finally` block whenever no frame had played by the time forwarding finished. Two real scenarios hit this window: a speech paused in the thinking state by a brief user sound, whose buffered first frame only plays after the false interruption clears (#1909), and DataStream avatar outputs with `waitPlaybackStart: true`, which deliver `lk.playback_started` ~1s after frames were captured (#1960). In both cases the late playback-started event found nothing listening, the reply was classified "skipped", and the turn was silently removed from the chat context while the agent never entered the `speaking` state.

The `PLAYBACK_STARTED` listener now lives in `performAudioForwarding` so it outlives the forwarding task, and `forwardAudio` no longer settles the future; the reply tasks (including the `say()` path) settle it once the playout window ends, which also detaches the listener. A reported non-zero playback position on interruption is additionally honored as evidence of partial playback — but only when the segment actually captured a frame into the output (tracked via the output's segment count, which is also what makes the reported position fresh rather than stale) — covering avatars whose playback-started RPC races the interruption itself.
130 changes: 91 additions & 39 deletions agents/src/voice/agent_activity.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -2493,14 +2493,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(
Expand Down Expand Up @@ -2541,53 +2541,73 @@ 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();
}
}
Comment on lines +2551 to 2558

@devin-ai-integration devin-ai-integration Bot Jul 6, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

🔍 The say() path lacks the new partial-playback-position heuristic

The ttsTask method (used by session.say()) at agents/src/voice/agent_activity.ts:2551-2558 handles interruption by aborting and clearing the buffer, but unlike the pipeline (forwardSegment at line 2925-2928) and realtime (processOneMessage at line 3493-3496) paths, it does NOT check startedForwardingAt or playbackPosition > 0 as a fallback for detecting partial playback when firstFrameFut hasn't resolved. This means the say() path still relies solely on firstFrameFut resolving (via the late PLAYBACK_STARTED listener) to enter the 'speaking' state and record metrics. For avatar outputs where the playback-started RPC races with interruption, say() may still miss the speaking-state transition. This is a pre-existing gap that the PR's new heuristic doesn't cover for this path, though the say() path always commits to chat context regardless (line 2560-2583), so the dropped-turn symptom doesn't apply here.

Open in Devin Review

Was this helpful? React with 👍 or 👎 to provide feedback.

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, dropping the PLAYBACK_STARTED
* listener registered in `performAudioForwarding`. Called by the reply tasks once
* the playout window has ended (finished or interrupted). `forwardAudio` no longer
* settles the future itself so that a frame which plays late — after a
* false-interruption resume (#1909) or a remote avatar's deferred
* `lk.playback_started` notification (#1960) — can still resolve it.
*/
private settleFirstFrameFut(audioOut: _AudioOut | null | undefined): void {
if (audioOut && !audioOut.firstFrameFut.done) {
audioOut.firstFrameFut.reject(new Error('playout finished before playback started'));
}
}

Expand Down Expand Up @@ -2895,7 +2915,20 @@ export class AgentActivity implements RecognitionHooks {
if (audioOutput) {
audioOutput.clearBuffer();
const interruptedPlaybackEv = await audioOutput.waitForPlayout();

@chenghao-mou chenghao-mou Jul 6, 2026

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

should we use something like

// generation.ts — performAudioForwarding(): record the baseline
const out: _AudioOut = {
  audio: [], firstFrameFut: new Future(), _hasCapturedOwnFrame: false,
  capturedSegmentsBefore: audioOutput.capturedPlayoutSegments,
};

// agent_activity.ts — the interrupted branch (both the segment and say paths)
- const forwardedAudio = output.audioOut?.startedForwardingAt !== undefined;
+ const playedOwnFrame =
+   output.audioOut !== null &&
+   audioOutput.capturedPlayoutSegments > output.audioOut.capturedSegmentsBefore;

if ((firstFrameFut.done && !firstFrameFut.rejected) ||
    (playedOwnFrame && playbackEv.playbackPosition > 0)) {
  output.played = 'partial';   // now only when a frame really played
  ...
}

basically forwarding check is replaced with captured check

@toubatbrian toubatbrian Jul 7, 2026

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I can do it, but it will no longer have parity match with python code-level, tho I think its fine.

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 bumped the output's
// segment count — the same condition under which waitForPlayout waits
// for this segment's playback event instead of returning a stale one
// from a previous segment (see _AudioOut.capturedSegmentsBefore).
const playedOwnFrame =
output.audioOut != null &&
audioOutput.capturedPlayoutSegments > output.audioOut.capturedSegmentsBefore;
if (
(output.audioOut?.firstFrameFut.done && !output.audioOut.firstFrameFut.rejected) ||
(playedOwnFrame && interruptedPlaybackEv.playbackPosition > 0)
) {
output.played = 'partial';
output.playbackPositionInS = interruptedPlaybackEv.playbackPosition;
output.synchronizedTranscript = interruptedPlaybackEv.synchronizedTranscript;
Expand All @@ -2917,6 +2950,9 @@ 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 the playback-started listener is detached.
this.settleFirstFrameFut(output.audioOut);
}
};

Expand Down Expand Up @@ -3449,7 +3485,20 @@ 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 bumped the output's
// segment count — the same condition under which waitForPlayout waits
// for this message's playback event instead of returning a stale one
// from a previous message (see _AudioOut.capturedSegmentsBefore).
const playedOwnFrame =
output.audioOut != null &&
audioOutput.capturedPlayoutSegments > output.audioOut.capturedSegmentsBefore;
if (
(output.audioOut?.firstFrameFut.done && !output.audioOut.firstFrameFut.rejected) ||
(playedOwnFrame && playbackEv.playbackPosition > 0)
) {
output.played = 'partial';
output.playbackPositionInS = playbackEv.playbackPosition;
output.synchronizedTranscript = playbackEv.synchronizedTranscript;
Expand All @@ -3471,6 +3520,9 @@ export class AgentActivity implements RecognitionHooks {
} finally {
abortController.signal.removeEventListener('abort', abortMessage);
await cancelAndWait(forwardTasks, AgentActivity.REPLY_TASK_CANCEL_TIMEOUT);
// The message's playout window is over; settle a still-pending
// firstFrameFut so the playback-started listener is detached.
this.settleFirstFrameFut(output.audioOut);
}
};

Expand Down
Loading
Loading