ewilliams28/openclaw
1
0
Fork 0
mirror of https://github.com/openclaw/openclaw.git synced 2026-08-26 20:35:39 -06:00
Code Issues Packages Projects Releases Wiki Activity

fix(voice): fence realtime bridge lifecycle (#129249)

Browse Source
This commit is contained in:
Peter Steinberger
2026-08-25 03:58:42 -07:00
committed by GitHub
parent 1ba243c88e
commit 9f472253d4
3 changed files with 159 additions and 120 deletions
Download Patch File Download Diff File Expand all files Collapse all files
extensions/google/index.test.ts
+70
View File
@@ -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");
extensions/google/index.ts
+30 -24
View File
@@ -202,11 +202,11 @@ function createLazyGoogleRealtimeVoiceBridge(
): RealtimeVoiceBridge {
let bridge: RealtimeVoiceBridge | undefined;
let bridgePromise: Promise<RealtimeVoiceBridge> | 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),
};
}
extensions/xai/lazy-capability-providers.ts
+59 -96
View File
@@ -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<RealtimeVoiceBridge>;
}
| undefined;
let activeConnect:
| {
generation: number;
promise: Promise<void>;
}
| 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<NonNullable<RealtimeVoiceBridgeCreateRequest["onClose"]>>[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 = <TArgs extends unknown[]>(
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),
};