From d7afb133af9e989fe97a87aa472e38cf0b40a8ae Mon Sep 17 00:00:00 2001 From: Vincent Koc Date: Fri, 31 Jul 2026 09:55:23 +0800 Subject: [PATCH] fix(meeting-bot): track realtime output ownership --- src/meeting-bot/realtime-engine.ts | 55 ++++++----- src/meeting-bot/realtime-output-owner.ts | 120 +++++++++++++++++++++++ 2 files changed, 149 insertions(+), 26 deletions(-) create mode 100644 src/meeting-bot/realtime-output-owner.ts diff --git a/src/meeting-bot/realtime-engine.ts b/src/meeting-bot/realtime-engine.ts index 543caa51bc10..eab0d81acc56 100644 --- a/src/meeting-bot/realtime-engine.ts +++ b/src/meeting-bot/realtime-engine.ts @@ -25,6 +25,7 @@ import { meetingOutputBytesPerMs, resolveMeetingRealtimeProvider, } from "./realtime-engine-support.js"; +import { createMeetingRealtimeOutputOwner } from "./realtime-output-owner.js"; export { formatMeetingAgentAudioModelLog, @@ -133,10 +134,10 @@ export async function startMeetingRealtimeEngine(params: { let outputClearAfterActive = false; let outputPendingBytes = 0; let outputPendingFrames = 0; - let outputBlocked: { token: symbol } | undefined; let outputGenerationActive = false; let outputClearTail = Promise.resolve(); const outputQueue: Array<{ audio: Buffer; generation: number }> = []; + const outputOwner = createMeetingRealtimeOutputOwner(); const outputMaxPendingBytes = meetingOutputBytesPerMs(params.config.chrome.audioFormat) * MEETING_REALTIME_OUTPUT_MAX_PENDING_MS; @@ -156,7 +157,7 @@ export async function startMeetingRealtimeEngine(params: { const stop = async () => { if (!stopped) { stopped = true; - outputBlocked = undefined; + outputOwner.reset(); outputClearAfterActive = false; invalidateOutputQueue(); } @@ -308,20 +309,19 @@ export async function startMeetingRealtimeEngine(params: { const clearOutputPlayback = (): void => { void queueOutputClear(); }; + const invalidateOutputPlayback = (): void => { + outputClearAfterActive ||= outputTransportWriteStarted; + invalidateOutputQueue(); + }; const invalidateAndClearOutputPlayback = (): void => { blockOutput(); clearOutputPlayback(); }; const blockOutput = (): { blocked: boolean; token: symbol } => { - if (outputBlocked) { - return { blocked: false, token: outputBlocked.token }; - } - const token = Symbol("meeting-realtime-output-blocked"); - outputBlocked = { token }; - outputClearAfterActive = outputTransportWriteStarted; - invalidateOutputQueue(); - return { blocked: true, token }; + const result = outputOwner.block(); + invalidateOutputPlayback(); + return result; }; const handleOutputBackpressure = () => { @@ -337,15 +337,15 @@ export async function startMeetingRealtimeEngine(params: { harness.flushOutput(clearOutputPlayback); harness.finishOutputAudio("output-backpressure"); queueMicrotask(() => { - if (stopped || outputBlocked?.token !== token) { + if (stopped || !outputOwner.isBlockedBy(token)) { return; } harness.handleBargeIn({ audioPlaybackActive: true, force: true }, () => {}); }); }; - const queueOutputAudio = (audio: Buffer): boolean => { - if (stopped || outputBlocked) { + const queueOutputAudio = (audio: Buffer, responseId: string | undefined): boolean => { + if (stopped || !outputOwner.accept(responseId)) { return false; } if ( @@ -479,7 +479,8 @@ export async function startMeetingRealtimeEngine(params: { audioSink: { isOpen: () => !stopped, sendAudio: (audio) => { - if (!queueOutputAudio(audio)) { + const responseId = outputOwner.takeNextResponseId(); + if (!queueOutputAudio(audio, responseId)) { return; } if (!outputGenerationActive) { @@ -490,7 +491,8 @@ export async function startMeetingRealtimeEngine(params: { harness.recordOutputAudio(audio); }, clearAudio: () => { - if (blockOutput().blocked) { + if (outputOwner.providerClear()) { + invalidateOutputPlayback(); harness.flushOutput(clearOutputPlayback); harness.finishOutputAudio("clear"); } @@ -540,11 +542,14 @@ export async function startMeetingRealtimeEngine(params: { ); return; } + } + if (role === "user" && strategy === "agent") { harness.talkback?.enqueue(text); } } }, onEvent: (event) => { + outputOwner.noteEvent(event); if (event.type === "input_audio_buffer.speech_started") { harness.ensureTurn(); } else if (event.type === "input_audio_buffer.speech_stopped") { @@ -558,21 +563,19 @@ export async function startMeetingRealtimeEngine(params: { payload: { ...outputTalkPayload, source: event.type }, final: true, }); - } else if (event.type === "response.done") { - outputBlocked = undefined; - outputGenerationActive = false; - harness.finishOutputAudio("response.done"); - harness.endTurn("response.done"); - } else if (outputBlocked && event.type === "response.cancelled") { - outputBlocked = undefined; - outputGenerationActive = false; - harness.finishOutputAudio(event.type); + } else if (event.type === "response.done" || event.type === "response.cancelled") { + if (outputOwner.terminal(event.responseId)) { + outputGenerationActive = false; + harness.finishOutputAudio(event.type); + if (event.type === "response.done") { + harness.endTurn(event.type); + } + } } else if ( event.type === "error" && event.detail === MEETING_REALTIME_CANCELLATION_RACE_DETAIL ) { - if (outputBlocked) { - outputBlocked = undefined; + if (outputOwner.clearBlocked()) { outputGenerationActive = false; harness.finishOutputAudio(event.type); } diff --git a/src/meeting-bot/realtime-output-owner.ts b/src/meeting-bot/realtime-output-owner.ts new file mode 100644 index 000000000000..d5dcca297e31 --- /dev/null +++ b/src/meeting-bot/realtime-output-owner.ts @@ -0,0 +1,120 @@ +import type { RealtimeVoiceBridgeEvent } from "../talk/provider-types.js"; + +const STALE_RESPONSE_LIMIT = 16; +const AUDIO_DELTA_EVENTS = new Set([ + "conversation.output_audio.delta", + "response.audio.delta", + "response.output_audio.delta", +]); + +export function createMeetingRealtimeOutputOwner() { + let nextResponseId: string | undefined; + let announcedResponseId: string | undefined; + let currentResponseId: string | undefined; + let blocked: { responseId?: string; token: symbol } | undefined; + const staleResponseIds = new Set(); + + const rememberStale = (responseId: string) => { + staleResponseIds.delete(responseId); + staleResponseIds.add(responseId); + while (staleResponseIds.size > STALE_RESPONSE_LIMIT) { + const oldest = staleResponseIds.values().next().value; + if (!oldest) { + break; + } + staleResponseIds.delete(oldest); + } + }; + + return { + accept(responseId: string | undefined): boolean { + if (responseId && staleResponseIds.has(responseId)) { + return false; + } + if (blocked) { + if (!blocked.responseId || !responseId || responseId === blocked.responseId) { + return false; + } + blocked = undefined; + } + if (responseId) { + currentResponseId = responseId; + } + return true; + }, + block(): { blocked: boolean; token: symbol } { + if (blocked) { + return { blocked: false, token: blocked.token }; + } + const token = Symbol("meeting-realtime-output-blocked"); + const responseId = currentResponseId ?? announcedResponseId; + blocked = { ...(responseId ? { responseId } : {}), token }; + if (responseId) { + rememberStale(responseId); + } + nextResponseId = undefined; + return { blocked: true, token }; + }, + clearBlocked(): boolean { + if (!blocked) { + return false; + } + blocked = undefined; + return true; + }, + isBlockedBy(token: symbol): boolean { + return blocked?.token === token; + }, + noteEvent(event: RealtimeVoiceBridgeEvent): void { + if (event.direction === "server" && event.type === "response.created" && event.responseId) { + announcedResponseId = event.responseId; + nextResponseId = undefined; + return; + } + nextResponseId = + event.direction === "server" && AUDIO_DELTA_EVENTS.has(event.type) + ? (event.responseId ?? announcedResponseId) + : undefined; + }, + providerClear(): boolean { + if (blocked) { + if (!blocked.responseId) { + blocked = undefined; + } + return false; + } + const responseId = currentResponseId ?? announcedResponseId; + if (responseId) { + blocked = { responseId, token: Symbol("meeting-realtime-output-blocked") }; + rememberStale(responseId); + } + nextResponseId = undefined; + return true; + }, + reset(): void { + nextResponseId = undefined; + announcedResponseId = undefined; + currentResponseId = undefined; + blocked = undefined; + staleResponseIds.clear(); + }, + takeNextResponseId(): string | undefined { + const responseId = nextResponseId ?? announcedResponseId; + nextResponseId = undefined; + return responseId; + }, + terminal(responseId: string | undefined): boolean { + if (!responseId || !blocked?.responseId || blocked.responseId === responseId) { + blocked = undefined; + } + if (!responseId || announcedResponseId === responseId) { + announcedResponseId = undefined; + } + if (responseId && currentResponseId && currentResponseId !== responseId) { + return false; + } + currentResponseId = undefined; + return true; + }, + }; +}