From 9f472253d49ac6992d0eecd528230e2e1543514d Mon Sep 17 00:00:00 2001 From: Peter Steinberger Date: Tue, 25 Aug 2026 03:58:42 -0700 Subject: [PATCH] fix(voice): fence realtime bridge lifecycle (#129249) --- extensions/google/index.test.ts | 70 +++++++++ extensions/google/index.ts | 54 ++++--- extensions/xai/lazy-capability-providers.ts | 155 ++++++++------------ 3 files changed, 159 insertions(+), 120 deletions(-) diff --git a/extensions/google/index.test.ts b/extensions/google/index.test.ts index a4f9d71972ce..bf19dc0d21c6 100644 --- a/extensions/google/index.test.ts +++ b/extensions/google/index.test.ts @@ -656,6 +656,76 @@ describe("google provider plugin hooks", () => { bridge.close(); }); + it("reopens the provider bridge after an explicit close", async () => { + let firstConnected = false; + let replacementConnected = false; + const first = createMockRealtimeBridge(async () => { + firstConnected = true; + }); + first.bridge.isConnected = vi.fn(() => firstConnected); + const replacement = createMockRealtimeBridge(async () => { + replacementConnected = true; + }); + replacement.close.mockImplementation(() => { + replacementConnected = false; + }); + replacement.bridge.isConnected = vi.fn(() => replacementConnected); + createRealtimeBridgeMock + .mockReturnValueOnce(first.bridge) + .mockReturnValueOnce(replacement.bridge); + const onReady = vi.fn(); + const onClose = vi.fn(); + const { bridge } = createLazyRealtimeBridge(vi.fn(), onReady, onClose); + + await bridge.connect(); + const firstRequest = createRealtimeBridgeMock.mock.calls[0]?.[0]; + expect(bridge.isConnected()).toBe(true); + + bridge.close(); + bridge.sendAudio(Buffer.from([0x01])); + expect(bridge.isConnected()).toBe(false); + + const reconnectPromise = bridge.connect(); + bridge.sendAudio(Buffer.from([0x02])); + await reconnectPromise; + signalRealtimeBridgeReady(); + + expect(bridge.isConnected()).toBe(true); + expect(first.close).toHaveBeenCalledOnce(); + expect(first.sendAudio).not.toHaveBeenCalled(); + expect(replacement.connect).toHaveBeenCalledOnce(); + expect(replacement.close).not.toHaveBeenCalled(); + expect(replacement.sendAudio).toHaveBeenCalledExactlyOnceWith(Buffer.from([0x02])); + firstRequest?.onReady?.(); + firstRequest?.onClose?.("error"); + expect(onReady).toHaveBeenCalledOnce(); + expect(onClose).toHaveBeenCalledExactlyOnceWith("completed"); + }); + + it("fences a provider generation closed during lazy load before reconnecting", async () => { + const first = createMockRealtimeBridge(); + const replacement = createMockRealtimeBridge(); + replacement.bridge.isConnected = vi.fn(() => replacement.connect.mock.calls.length > 0); + createRealtimeBridgeMock + .mockReturnValueOnce(first.bridge) + .mockReturnValueOnce(replacement.bridge); + const { bridge } = createLazyRealtimeBridge(); + + const staleConnect = bridge.connect(); + bridge.close(); + const replacementConnect = bridge.connect(); + bridge.sendAudio(Buffer.from([0x02])); + await Promise.all([staleConnect, replacementConnect]); + signalRealtimeBridgeReady(); + + expect(first.connect).not.toHaveBeenCalled(); + expect(first.close).toHaveBeenCalledOnce(); + expect(replacement.connect).toHaveBeenCalledOnce(); + expect(replacement.close).not.toHaveBeenCalled(); + expect(replacement.sendAudio).toHaveBeenCalledExactlyOnceWith(Buffer.from([0x02])); + expect(bridge.isConnected()).toBe(true); + }); + it("reports and cleans up a lazy realtime connect failure before reconnecting", async () => { const failure = new Error("Google realtime connect rejected"); const errorCallbackFailure = new Error("Google realtime error callback rejected"); diff --git a/extensions/google/index.ts b/extensions/google/index.ts index 45f7e91eee0a..2ca9f5e91e2b 100644 --- a/extensions/google/index.ts +++ b/extensions/google/index.ts @@ -202,11 +202,11 @@ function createLazyGoogleRealtimeVoiceBridge( ): RealtimeVoiceBridge { let bridge: RealtimeVoiceBridge | undefined; let bridgePromise: Promise | undefined; + let bridgePromiseGeneration = 0; let bridgeReady = false; - let closed = false; // Provider close is terminal for input admission. Only an explicit connect() // call may reopen it; late callbacks and microphone frames stay ignored. - let providerTerminated = false; + let terminated = false; let generation = 0; let latestMediaTimestamp: number | undefined; let pendingGreeting: string | undefined; @@ -223,7 +223,7 @@ function createLazyGoogleRealtimeVoiceBridge( latestMediaTimestamp = undefined; }; const isCurrentNonterminalGeneration = (candidate: number) => - candidate === generation && !providerTerminated; + candidate === generation && !terminated; // Loading and connecting finish on separate async boundaries. Keep close ownership // here so either late completion closes the provider bridge exactly once. const closeBridge = (loadedBridge = bridge) => { @@ -238,7 +238,7 @@ function createLazyGoogleRealtimeVoiceBridge( return; } bridgeReady = false; - providerTerminated = true; + terminated = true; clearPendingInput(); req.onClose?.(reason); }; @@ -271,15 +271,16 @@ function createLazyGoogleRealtimeVoiceBridge( const loadBridge = async () => { if (!bridgePromise) { const loadGeneration = generation; + bridgePromiseGeneration = loadGeneration; bridgePromise = loadGoogleRealtimeVoiceProvider().then((provider) => provider.createBridge({ ...req, onReady: () => { - if (loadGeneration !== generation || closed || providerTerminated) { + if (loadGeneration !== generation || terminated) { return; } req.onReady?.(); - if (loadGeneration !== generation || closed || providerTerminated || !bridge) { + if (loadGeneration !== generation || terminated || !bridge) { return; } bridgeReady = true; @@ -293,11 +294,17 @@ function createLazyGoogleRealtimeVoiceBridge( }), ); } - bridge = await bridgePromise; - if (closed) { - closeBridge(bridge); + const loading = bridgePromise; + const loadGeneration = bridgePromiseGeneration; + const loadedBridge = await loading; + // Explicit reconnect can replace the lazy load before it settles. Only the + // matching generation may publish a bridge; stale instances must close. + if (loading !== bridgePromise || loadGeneration !== generation || terminated) { + closeBridge(loadedBridge); + return loadedBridge; } - return bridge; + bridge = loadedBridge; + return loadedBridge; }; const requireBridge = () => { if (!bridge) { @@ -306,7 +313,7 @@ function createLazyGoogleRealtimeVoiceBridge( return bridge; }; const flushPending = (loadedBridge: RealtimeVoiceBridge) => { - if (closed || providerTerminated) { + if (terminated) { return; } if (typeof latestMediaTimestamp === "number") { @@ -332,16 +339,16 @@ function createLazyGoogleRealtimeVoiceBridge( }, supportsToolResultSuppression: false, connect: async () => { - if (providerTerminated) { + if (terminated) { generation += 1; bridge = undefined; bridgePromise = undefined; bridgeReady = false; - providerTerminated = false; + terminated = false; } const connectGeneration = generation; const loadedBridge = await loadBridge(); - if (connectGeneration !== generation || closed) { + if (connectGeneration !== generation || terminated) { closeBridge(loadedBridge); return; } @@ -350,12 +357,12 @@ function createLazyGoogleRealtimeVoiceBridge( } catch (error) { throwTerminalBridgeError(connectGeneration, loadedBridge, error); } - if (connectGeneration !== generation || closed) { + if (connectGeneration !== generation || terminated) { closeBridge(loadedBridge); } }, sendAudio: (audio) => { - if (closed || providerTerminated) { + if (terminated) { return; } if (bridgeReady && bridge) { @@ -365,14 +372,14 @@ function createLazyGoogleRealtimeVoiceBridge( pendingAudio.enqueue(audio); }, setMediaTimestamp: (ts) => { - if (closed || providerTerminated) { + if (terminated) { return; } latestMediaTimestamp = ts; bridge?.setMediaTimestamp(ts); }, sendUserMessage: (text) => { - if (closed || providerTerminated) { + if (terminated) { return; } if (bridgeReady && bridge) { @@ -393,7 +400,7 @@ function createLazyGoogleRealtimeVoiceBridge( pendingUserMessageBytes += messageBytes; }, triggerGreeting: (instructions) => { - if (closed || providerTerminated) { + if (terminated) { return; } if (bridgeReady && bridge) { @@ -407,19 +414,18 @@ function createLazyGoogleRealtimeVoiceBridge( requireBridge().submitToolResult(callId, result, options), acknowledgeMark: () => requireBridge().acknowledgeMark(), close: () => { - if (closed) { + if (terminated) { return; } - const closeGeneration = generation; - closed = true; + terminated = true; bridgeReady = false; clearPendingInput(); closeBridge(); // A bridge closed before its first connect has no provider-owned // connection to report the terminal outcome. - emitTerminal(closeGeneration, "completed"); + req.onClose?.("completed"); }, - isConnected: () => bridge?.isConnected() ?? false, + isConnected: () => !terminated && (bridge?.isConnected() ?? false), }; } diff --git a/extensions/xai/lazy-capability-providers.ts b/extensions/xai/lazy-capability-providers.ts index f6dd833b58cd..ede489df1cf7 100644 --- a/extensions/xai/lazy-capability-providers.ts +++ b/extensions/xai/lazy-capability-providers.ts @@ -6,11 +6,13 @@ import type { RealtimeTranscriptionSession, RealtimeTranscriptionSessionCreateRequest, } from "openclaw/plugin-sdk/realtime-transcription"; -import type { - RealtimeVoiceBridge, - RealtimeVoiceBridgeCreateRequest, - RealtimeVoiceProviderPlugin, - RealtimeVoiceToolResultOptions, +import { + RealtimeVoiceSessionLifecycle, + type RealtimeVoiceBridge, + type RealtimeVoiceBridgeCreateRequest, + type RealtimeVoiceProviderPlugin, + type RealtimeVoiceSessionConnection, + type RealtimeVoiceToolResultOptions, } from "openclaw/plugin-sdk/realtime-voice"; import { createRealtimeVoiceAudioQueue } from "openclaw/plugin-sdk/realtime-voice-audio-queue"; import type { @@ -220,20 +222,12 @@ function createLazyXaiRealtimeVoiceBridge( let bridge: RealtimeVoiceBridge | undefined; let bridgeState: | { - generation: number; + connection: RealtimeVoiceSessionConnection; promise: Promise; } | undefined; - let activeConnect: - | { - generation: number; - promise: Promise; - } - | undefined; - let generation = 0; - let terminalGeneration: number | undefined; - let closed = false; let acceptsInput = false; + const lifecycle = new RealtimeVoiceSessionLifecycle("xAI lazy"); let pendingMediaTimestamp: PendingMediaTimestamp | undefined; let pendingGreeting: PendingVoiceGreeting | undefined; let pendingUserMessageCount = 0; @@ -254,19 +248,17 @@ function createLazyXaiRealtimeVoiceBridge( pendingToolResultCount = 0; pendingToolResultBytes = 0; }; - const isCurrentNonterminalGeneration = (candidate: number) => - candidate === generation && terminalGeneration !== candidate; const emitTerminal = ( - terminalForGeneration: number, + connection: RealtimeVoiceSessionConnection, outcome: Parameters>[0], ) => { - if (!isCurrentNonterminalGeneration(terminalForGeneration)) { + const terminalOutcome = lifecycle.close(connection, outcome); + if (!terminalOutcome) { return; } - terminalGeneration = terminalForGeneration; acceptsInput = false; clearPendingInput(); - req.onClose?.(outcome); + req.onClose?.(terminalOutcome); }; const closeBridge = (loadedBridge: RealtimeVoiceBridge | undefined = bridge) => { if (!loadedBridge || closedBridges.has(loadedBridge)) { @@ -276,11 +268,11 @@ function createLazyXaiRealtimeVoiceBridge( loadedBridge.close(); }; const throwTerminalBridgeError = ( - terminalForGeneration: number, + connection: RealtimeVoiceSessionConnection, loadedBridge: RealtimeVoiceBridge, primaryError: unknown, ): never => { - if (isCurrentNonterminalGeneration(terminalForGeneration)) { + if (lifecycle.failure(connection)) { try { req.onError?.( primaryError instanceof Error ? primaryError : new Error(String(primaryError)), @@ -289,7 +281,7 @@ function createLazyXaiRealtimeVoiceBridge( // Consumer callback failures cannot prevent terminal cleanup or replace the provider failure. } try { - emitTerminal(terminalForGeneration, "error"); + emitTerminal(connection, "error"); } catch { // Consumer callback failures cannot prevent cleanup or replace the provider failure. } @@ -301,54 +293,46 @@ function createLazyXaiRealtimeVoiceBridge( } throw primaryError; }; - const acceptsProviderCallback = (callbackGeneration: number) => - !closed && isCurrentNonterminalGeneration(callbackGeneration); + const acceptsProviderCallback = (connection: RealtimeVoiceSessionConnection) => + lifecycle.acceptsEvents(connection); const guardProviderCallback = ( - callbackGeneration: number, + connection: RealtimeVoiceSessionConnection, callback: (...args: TArgs) => void, ) => { return (...args: TArgs) => { - if (acceptsProviderCallback(callbackGeneration)) { + if (acceptsProviderCallback(connection)) { callback(...args); } }; }; - const loadBridge = async (loadGeneration: number) => { + const loadBridge = async (connection: RealtimeVoiceSessionConnection) => { const existingState = bridgeState; const state = - existingState?.generation === loadGeneration + existingState?.connection.id === connection.id ? existingState : { - generation: loadGeneration, + connection, promise: loadXaiRealtimeVoiceProvider().then((provider) => provider.createBridge({ ...req, // An explicit wrapper reconnect owns a new provider bridge. Guard every // nonterminal callback so late events cannot reach its replacement. - onAudio: guardProviderCallback(loadGeneration, req.onAudio), - onClearAudio: guardProviderCallback(loadGeneration, req.onClearAudio), - ...(req.onMark - ? { onMark: guardProviderCallback(loadGeneration, req.onMark) } - : {}), + onAudio: guardProviderCallback(connection, req.onAudio), + onClearAudio: guardProviderCallback(connection, req.onClearAudio), + ...(req.onMark ? { onMark: guardProviderCallback(connection, req.onMark) } : {}), ...(req.onTranscript - ? { onTranscript: guardProviderCallback(loadGeneration, req.onTranscript) } - : {}), - ...(req.onEvent - ? { onEvent: guardProviderCallback(loadGeneration, req.onEvent) } + ? { onTranscript: guardProviderCallback(connection, req.onTranscript) } : {}), + ...(req.onEvent ? { onEvent: guardProviderCallback(connection, req.onEvent) } : {}), ...(req.onResponseDone - ? { onResponseDone: guardProviderCallback(loadGeneration, req.onResponseDone) } + ? { onResponseDone: guardProviderCallback(connection, req.onResponseDone) } : {}), ...(req.onToolCall - ? { onToolCall: guardProviderCallback(loadGeneration, req.onToolCall) } + ? { onToolCall: guardProviderCallback(connection, req.onToolCall) } : {}), - ...(req.onReady - ? { onReady: guardProviderCallback(loadGeneration, req.onReady) } - : {}), - ...(req.onError - ? { onError: guardProviderCallback(loadGeneration, req.onError) } - : {}), - onClose: (outcome) => emitTerminal(loadGeneration, outcome), + ...(req.onReady ? { onReady: guardProviderCallback(connection, req.onReady) } : {}), + ...(req.onError ? { onError: guardProviderCallback(connection, req.onError) } : {}), + onClose: (outcome) => emitTerminal(connection, outcome), }), ), }; @@ -356,7 +340,7 @@ function createLazyXaiRealtimeVoiceBridge( bridgeState = state; } const loadedBridge = await state.promise; - if (bridgeState === state && loadGeneration === generation) { + if (bridgeState === state && lifecycle.isCurrent(connection)) { bridge = loadedBridge; } return loadedBridge; @@ -374,32 +358,23 @@ function createLazyXaiRealtimeVoiceBridge( pendingOperations.push(next); return next; }; - const beginConnectGeneration = () => { - if (closed || terminalGeneration === generation) { - generation += 1; - closed = false; - acceptsInput = false; - bridge = undefined; - } - return generation; - }; - const acceptsCurrentInput = () => !closed && terminalGeneration !== generation; + const acceptsCurrentInput = () => lifecycle.phase() !== "terminal"; const flushPendingInput = async ( loadedBridge: RealtimeVoiceBridge, - connectGeneration: number, + connection: RealtimeVoiceSessionConnection, ) => { - if (connectGeneration !== generation || !acceptsCurrentInput()) { + if (!lifecycle.acceptsEvents(connection)) { return; } while (true) { - if (connectGeneration !== generation || !acceptsCurrentInput()) { + if (!lifecycle.acceptsEvents(connection)) { return; } const operation = pendingOperations.shift(); if (!operation) { // Queue exhaustion and direct admission must change in the same turn. // An await between them can strand input admitted by the next microtask. - acceptsInput = true; + acceptsInput = lifecycle.ready(connection); return; } switch (operation.type) { @@ -434,7 +409,7 @@ function createLazyXaiRealtimeVoiceBridge( loadedBridge.triggerGreeting?.(operation.instructions); break; } - if (connectGeneration !== generation || !acceptsCurrentInput()) { + if (!lifecycle.acceptsEvents(connection)) { return; } if (operation.type === "user-message") { @@ -451,46 +426,33 @@ function createLazyXaiRealtimeVoiceBridge( get supportsToolResultContinuation() { return bridge?.supportsToolResultContinuation ?? false; }, - connect: async () => { - const connectGeneration = beginConnectGeneration(); - if (activeConnect?.generation === connectGeneration) { - await activeConnect.promise; - return; - } - const promise = (async () => { - const loadedBridge = await loadBridge(connectGeneration); - if (connectGeneration !== generation || !acceptsCurrentInput()) { + connect: () => + lifecycle.connect(async (connection) => { + acceptsInput = false; + bridge = undefined; + const loadedBridge = await loadBridge(connection); + if (!lifecycle.acceptsEvents(connection)) { closeBridge(loadedBridge); return; } try { await loadedBridge.connect(); } catch (error) { - throwTerminalBridgeError(connectGeneration, loadedBridge, error); + throwTerminalBridgeError(connection, loadedBridge, error); } - if (connectGeneration !== generation || !acceptsCurrentInput()) { + if (!lifecycle.acceptsEvents(connection)) { closeBridge(loadedBridge); return; } try { - await flushPendingInput(loadedBridge, connectGeneration); + await flushPendingInput(loadedBridge, connection); } catch (error) { - throwTerminalBridgeError(connectGeneration, loadedBridge, error); + throwTerminalBridgeError(connection, loadedBridge, error); } - if (connectGeneration !== generation || !acceptsCurrentInput()) { + if (!lifecycle.acceptsEvents(connection)) { closeBridge(loadedBridge); } - })(); - const connectTask = { generation: connectGeneration, promise }; - activeConnect = connectTask; - try { - await promise; - } finally { - if (activeConnect === connectTask) { - activeConnect = undefined; - } - } - }, + }), sendAudio: (audio) => { if (!acceptsCurrentInput()) { return; @@ -604,17 +566,18 @@ function createLazyXaiRealtimeVoiceBridge( } }, close: () => { - if (closed) { + const connection = lifecycle.currentConnection(); + if (!lifecycle.cancel()) { return; } - const closeGeneration = generation; - closed = true; acceptsInput = false; clearPendingInput(); closeBridge(); - // A bridge closed before its first connect has no provider-owned - // connection to report the terminal outcome. - emitTerminal(closeGeneration, "completed"); + if (connection) { + emitTerminal(connection, "completed"); + } else { + req.onClose?.("completed"); + } }, isConnected: () => acceptsCurrentInput() && (bridge?.isConnected() ?? false), };