fix(meeting-bot): track realtime output ownership

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

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