vultr/openclaw
1
0
Fork 0
mirror of https://github.com/openclaw/openclaw.git synced 2026-08-02 03:41:50 +00:00
Code Issues Packages Projects Releases Wiki Activity

fix(google): bound lazy realtime audio queue

Browse Source
This commit is contained in:
Vincent Koc
2026-08-02 03:59:37 +08:00
parent 53d227970e
commit aab146eeec
2 changed files with 126 additions and 13 deletions
Download Patch File Download Diff File Expand all files Collapse all files

81
extensions/google/index.test.ts
View File

@@ -55,6 +55,7 @@ function createDeferred<T>() {
function createMockRealtimeBridge(connectImpl: () => Promise<void> = async () => {}) {
const connect = vi.fn(connectImpl);
const sendAudio = vi.fn();
const sendUserMessage = vi.fn();
const triggerGreeting = vi.fn();
const close = vi.fn();
@@ -62,7 +63,7 @@ function createMockRealtimeBridge(connectImpl: () => Promise<void> = async () =>
supportsToolResultContinuation: false,
supportsToolResultSuppression: false,
connect,
sendAudio: vi.fn(),
sendAudio,
setMediaTimestamp: vi.fn(),
sendUserMessage,
triggerGreeting,
@@ -72,10 +73,14 @@ function createMockRealtimeBridge(connectImpl: () => Promise<void> = async () =>
close,
isConnected: vi.fn(() => false),
};
return { bridge, close, connect, sendUserMessage, triggerGreeting };
return { bridge, close, connect, sendAudio, sendUserMessage, triggerGreeting };
}
function createLazyRealtimeBridge(onError = vi.fn(), onReady?: () => void) {
function createLazyRealtimeBridge(
onError = vi.fn(),
onReady?: () => void,
onClose?: (reason: "completed" | "error") => void,
) {
let realtimeProvider: RealtimeVoiceProviderPlugin | undefined;
googlePlugin.register(
createTestPluginApi({
@@ -90,6 +95,7 @@ function createLazyRealtimeBridge(onError = vi.fn(), onReady?: () => void) {
onClearAudio() {},
onError,
onReady,
onClose,
});
if (!bridge) {
throw new Error("expected Google realtime bridge");
@@ -105,6 +111,14 @@ function signalRealtimeBridgeReady() {
request.onReady?.();
}
function signalRealtimeBridgeClose(reason: "completed" | "error") {
const request = createRealtimeBridgeMock.mock.calls.at(-1)?.[0];
if (!request) {
throw new Error("expected Google realtime bridge request");
}
request.onClose?.(reason);
}
describe("google provider plugin hooks", () => {
beforeEach(() => {
createRealtimeBridgeMock.mockReset();
@@ -487,6 +501,67 @@ describe("google provider plugin hooks", () => {
expect(bridge.sendUserMessage?.("hello")).toBeUndefined();
});
it("evicts the oldest lazy audio when the startup chunk limit is reached", async () => {
const loaded = createMockRealtimeBridge();
createRealtimeBridgeMock.mockReturnValue(loaded.bridge);
const { bridge } = createLazyRealtimeBridge();
for (let index = 0; index < 322; index += 1) {
bridge.sendAudio(Buffer.from([index & 0xff]));
}
await bridge.connect();
signalRealtimeBridgeReady();
expect(loaded.sendAudio).toHaveBeenCalledTimes(320);
expect(loaded.sendAudio.mock.calls[0]?.[0]).toEqual(Buffer.from([2]));
expect(loaded.sendAudio.mock.calls.at(-1)?.[0]).toEqual(Buffer.from([65]));
});
it("copies lazy audio and evicts oldest chunks to enforce the byte limit", async () => {
const loaded = createMockRealtimeBridge();
createRealtimeBridgeMock.mockReturnValue(loaded.bridge);
const { bridge } = createLazyRealtimeBridge();
const backing = Buffer.alloc(2 * 1024 * 1024, 0x02);
const retainedView = backing.subarray(0, 512 * 1024);
bridge.sendAudio(Buffer.alloc(512 * 1024, 0x01));
bridge.sendAudio(retainedView);
retainedView.fill(0);
bridge.sendAudio(Buffer.from([0x03]));
bridge.sendAudio(Buffer.alloc(1024 * 1024 + 1, 0x04));
await bridge.connect();
signalRealtimeBridgeReady();
expect(loaded.sendAudio).toHaveBeenCalledTimes(2);
expect(loaded.sendAudio.mock.calls[0]?.[0]).toEqual(Buffer.alloc(512 * 1024, 0x02));
expect(loaded.sendAudio.mock.calls[1]?.[0]).toEqual(Buffer.from([0x03]));
});
it("clears lazy audio on terminal close and reopens only for an explicit connect", async () => {
const loaded = createMockRealtimeBridge();
createRealtimeBridgeMock.mockReturnValue(loaded.bridge);
const onClose = vi.fn();
const { bridge } = createLazyRealtimeBridge(vi.fn(), undefined, onClose);
bridge.sendAudio(Buffer.from([0x01]));
await bridge.connect();
signalRealtimeBridgeClose("error");
bridge.sendAudio(Buffer.from([0x02]));
expect(onClose).toHaveBeenCalledOnce();
expect(onClose).toHaveBeenCalledWith("error");
expect(loaded.sendAudio).not.toHaveBeenCalled();
await bridge.connect();
signalRealtimeBridgeReady();
expect(loaded.sendAudio).not.toHaveBeenCalled();
bridge.sendAudio(Buffer.from([0x03]));
expect(loaded.sendAudio).toHaveBeenCalledOnce();
expect(loaded.sendAudio).toHaveBeenCalledWith(Buffer.from([0x03]));
bridge.close();
});
it("preserves queued user messages until the loaded bridge reports ready", async () => {
const connected = createDeferred<void>();
const loaded = createMockRealtimeBridge(() => connected.promise);

58
extensions/google/index.ts
View File

@@ -202,6 +202,7 @@ function resolveGoogleRealtimeEnvApiKey(): string | undefined {
}
const GOOGLE_REALTIME_LAZY_MAX_PENDING_AUDIO_CHUNKS = 320;
const GOOGLE_REALTIME_LAZY_MAX_PENDING_AUDIO_BYTES = 1024 * 1024;
const GOOGLE_REALTIME_LAZY_MAX_PENDING_USER_MESSAGES = 128;
const GOOGLE_REALTIME_LAZY_MAX_PENDING_USER_MESSAGE_BYTES = 256 * 1024;
@@ -213,11 +214,19 @@ function createLazyGoogleRealtimeVoiceBridge(
let bridgeReady = false;
let bridgeClosed = 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 latestMediaTimestamp: number | undefined;
let pendingGreeting: string | undefined;
const pendingAudio: Buffer[] = [];
let pendingAudioBytes = 0;
const pendingUserMessages: string[] = [];
let pendingUserMessageBytes = 0;
const clearPendingAudio = () => {
pendingAudio.length = 0;
pendingAudioBytes = 0;
};
// 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) => {
@@ -233,11 +242,11 @@ function createLazyGoogleRealtimeVoiceBridge(
provider.createBridge({
...req,
onReady: () => {
if (closed) {
if (closed || providerTerminated) {
return;
}
req.onReady?.();
if (closed || !bridge) {
if (closed || providerTerminated || !bridge) {
return;
}
bridgeReady = true;
@@ -245,6 +254,12 @@ function createLazyGoogleRealtimeVoiceBridge(
// Release prompts only after the provider can accept user content.
flushPending(bridge);
},
onClose: (reason) => {
bridgeReady = false;
providerTerminated = true;
clearPendingAudio();
req.onClose?.(reason);
},
}),
);
}
@@ -261,13 +276,15 @@ function createLazyGoogleRealtimeVoiceBridge(
return bridge;
};
const flushPending = (loadedBridge: RealtimeVoiceBridge) => {
if (closed) {
if (closed || providerTerminated) {
return;
}
if (typeof latestMediaTimestamp === "number") {
loadedBridge.setMediaTimestamp(latestMediaTimestamp);
}
for (const audio of pendingAudio.splice(0)) {
const audioChunks = pendingAudio.splice(0);
pendingAudioBytes = 0;
for (const audio of audioChunks) {
loadedBridge.sendAudio(audio);
}
const userMessages = pendingUserMessages.splice(0);
@@ -292,23 +309,43 @@ function createLazyGoogleRealtimeVoiceBridge(
closeBridge(loadedBridge);
return;
}
await loadedBridge.connect();
providerTerminated = false;
try {
await loadedBridge.connect();
} catch (error) {
bridgeReady = false;
providerTerminated = true;
clearPendingAudio();
throw error;
}
if (closed) {
closeBridge(loadedBridge);
}
},
sendAudio: (audio) => {
if (closed) {
if (closed || providerTerminated) {
return;
}
if (bridge) {
bridge.sendAudio(audio);
return;
}
if (pendingAudio.length >= GOOGLE_REALTIME_LAZY_MAX_PENDING_AUDIO_CHUNKS) {
pendingAudio.shift();
if (audio.byteLength > GOOGLE_REALTIME_LAZY_MAX_PENDING_AUDIO_BYTES) {
return;
}
pendingAudio.push(audio);
const queuedAudio = Buffer.from(audio);
while (
pendingAudio.length >= GOOGLE_REALTIME_LAZY_MAX_PENDING_AUDIO_CHUNKS ||
pendingAudioBytes + queuedAudio.byteLength > GOOGLE_REALTIME_LAZY_MAX_PENDING_AUDIO_BYTES
) {
const droppedAudio = pendingAudio.shift();
if (!droppedAudio) {
return;
}
pendingAudioBytes -= droppedAudio.byteLength;
}
pendingAudio.push(queuedAudio);
pendingAudioBytes += queuedAudio.byteLength;
},
setMediaTimestamp: (ts) => {
if (closed) {
@@ -355,7 +392,8 @@ function createLazyGoogleRealtimeVoiceBridge(
close: () => {
closed = true;
bridgeReady = false;
pendingAudio.length = 0;
providerTerminated = true;
clearPendingAudio();
pendingUserMessages.length = 0;
pendingUserMessageBytes = 0;
pendingGreeting = undefined;