From 53d227970e1bdeb6fb2081bfcba7d0472f700e54 Mon Sep 17 00:00:00 2001 From: Vincent Koc Date: Sun, 2 Aug 2026 03:58:44 +0800 Subject: [PATCH 1/7] fix(google): bound native realtime audio queue --- .../google/realtime-voice-provider.test.ts | 105 +++++++++++++++++- extensions/google/realtime-voice-provider.ts | 28 ++++- 2 files changed, 128 insertions(+), 5 deletions(-) diff --git a/extensions/google/realtime-voice-provider.test.ts b/extensions/google/realtime-voice-provider.test.ts index 7257aa57b7f0..47efc00225ae 100644 --- a/extensions/google/realtime-voice-provider.test.ts +++ b/extensions/google/realtime-voice-provider.test.ts @@ -1,5 +1,8 @@ // Google tests cover realtime voice provider plugin behavior. -import { REALTIME_VOICE_AUDIO_FORMAT_PCM16_24KHZ } from "openclaw/plugin-sdk/realtime-voice"; +import { + REALTIME_VOICE_AUDIO_FORMAT_PCM16_24KHZ, + resamplePcm, +} from "openclaw/plugin-sdk/realtime-voice"; import type { RealtimeVoiceTool } from "openclaw/plugin-sdk/realtime-voice"; import { afterAll, afterEach, beforeEach, describe, expect, it, vi } from "vitest"; import { buildGoogleRealtimeVoiceProvider } from "./realtime-voice-provider.js"; @@ -1213,6 +1216,106 @@ describe("buildGoogleRealtimeVoiceProvider", () => { expect(connectedSession.sendRealtimeInput).toHaveBeenCalledTimes(1); }); + it("copies and bounds pending audio by aggregate bytes before activation", async () => { + const connectedSession = createMockGoogleLiveSession(); + connectMock.mockResolvedValueOnce(connectedSession); + const provider = buildGoogleRealtimeVoiceProvider(); + const bridge = provider.createBridge({ + providerConfig: { apiKey: "gemini-key" }, + audioFormat: REALTIME_VOICE_AUDIO_FORMAT_PCM16_24KHZ, + onAudio: vi.fn(), + onClearAudio: vi.fn(), + }); + const backing = Buffer.alloc(2 * 1024 * 1024); + const firstChunk = backing.subarray(0, 512 * 1024); + firstChunk.writeInt16LE(513); + const expectedFirstSample = resamplePcm(Buffer.from(firstChunk), 24_000, 16_000).readInt16LE(0); + + bridge.sendAudio(firstChunk); + bridge.sendAudio(Buffer.alloc(512 * 1024, 0x7f)); + bridge.sendAudio(Buffer.from([0x01])); + firstChunk.fill(0); + + await bridge.connect(); + lastConnectParams().callbacks.onopen(); + lastConnectParams().callbacks.onmessage({ setupComplete: { sessionId: "session-1" } }); + + expect(connectedSession.sendRealtimeInput).toHaveBeenCalledTimes(2); + const firstAudio = connectedSession.sendRealtimeInput.mock.calls[0]?.[0]?.audio as + | { data?: unknown } + | undefined; + expect(Buffer.from(String(firstAudio?.data), "base64").readInt16LE(0)).toBe( + expectedFirstSample, + ); + }); + + it("bounds pending audio by chunk count before activation", async () => { + const connectedSession = createMockGoogleLiveSession(); + connectMock.mockResolvedValueOnce(connectedSession); + const provider = buildGoogleRealtimeVoiceProvider(); + const bridge = provider.createBridge({ + providerConfig: { apiKey: "gemini-key" }, + audioFormat: REALTIME_VOICE_AUDIO_FORMAT_PCM16_24KHZ, + onAudio: vi.fn(), + onClearAudio: vi.fn(), + }); + + for (let index = 0; index < 321; index += 1) { + bridge.sendAudio(Buffer.alloc(2, index & 0xff)); + } + + await bridge.connect(); + lastConnectParams().callbacks.onopen(); + lastConnectParams().callbacks.onmessage({ setupComplete: { sessionId: "session-1" } }); + + expect(connectedSession.sendRealtimeInput).toHaveBeenCalledTimes(320); + }); + + it("drops reconnect audio on terminal exhaustion until an explicit reconnect owns admission", async () => { + vi.useFakeTimers(); + const reconnectedSession = createMockGoogleLiveSession(); + const provider = buildGoogleRealtimeVoiceProvider(); + const onClose = vi.fn(); + const bridge = provider.createBridge({ + providerConfig: { apiKey: "gemini-key" }, + audioFormat: REALTIME_VOICE_AUDIO_FORMAT_PCM16_24KHZ, + onAudio: vi.fn(), + onClearAudio: vi.fn(), + onClose, + }); + + await bridge.connect(); + const firstSession = lastConnectParams().callbacks; + firstSession.onopen(); + firstSession.onmessage({ setupComplete: { sessionId: "session-1" } }); + connectMock + .mockRejectedValueOnce(new Error("connect failed 1")) + .mockRejectedValueOnce(new Error("connect failed 2")) + .mockRejectedValueOnce(new Error("connect failed 3")) + .mockResolvedValueOnce(reconnectedSession); + firstSession.onclose({ code: 1011, reason: "temporary" }); + bridge.sendAudio(Buffer.from([0x01, 0x00])); + + await vi.advanceTimersByTimeAsync(1_750); + bridge.sendAudio(Buffer.from([0x02, 0x00])); + + expect(onClose).toHaveBeenCalledOnce(); + expect(onClose).toHaveBeenCalledWith("error"); + + await bridge.connect(); + const reconnected = lastConnectParams().callbacks; + reconnected.onopen(); + reconnected.onmessage({ setupComplete: { sessionId: "session-2" } }); + bridge.sendAudio(Buffer.alloc(480, 0x03)); + + expect(reconnectedSession.sendRealtimeInput).toHaveBeenCalledOnce(); + const sent = reconnectedSession.sendRealtimeInput.mock.calls[0]?.[0]?.audio as + | { data?: unknown } + | undefined; + expect(sent?.data).toBeTypeOf("string"); + bridge.close(); + }); + it("does not activate a late session after close during setup", async () => { const pendingSession = createDeferred(); const lateSession = createMockGoogleLiveSession(); diff --git a/extensions/google/realtime-voice-provider.ts b/extensions/google/realtime-voice-provider.ts index 09b63d02b6be..3ddbd60403a1 100644 --- a/extensions/google/realtime-voice-provider.ts +++ b/extensions/google/realtime-voice-provider.ts @@ -62,6 +62,7 @@ const GOOGLE_REALTIME_BROWSER_API_VERSION = "v1alpha"; const GOOGLE_REALTIME_BROWSER_WEBSOCKET_URL = "wss://generativelanguage.googleapis.com/ws/google.ai.generativelanguage.v1alpha.GenerativeService.BidiGenerateContentConstrained"; const MAX_PENDING_AUDIO_CHUNKS = 320; +const MAX_PENDING_AUDIO_BYTES = 1024 * 1024; const DEFAULT_AUDIO_STREAM_END_SILENCE_MS = 500; const GOOGLE_REALTIME_BROWSER_SESSION_TTL_MS = 30 * 60 * 1000; const GOOGLE_REALTIME_BROWSER_NEW_SESSION_TTL_MS = 60 * 1000; @@ -472,6 +473,7 @@ class GoogleRealtimeVoiceBridge implements RealtimeVoiceBridge { private sessionConfigured = false; private intentionallyClosed = false; private pendingAudio: Buffer[] = []; + private pendingAudioBytes = 0; private sessionReadyFired = false; private consecutiveSilenceMs = 0; private audioStreamEnded = false; @@ -641,10 +643,19 @@ class GoogleRealtimeVoiceBridge implements RealtimeVoiceBridge { } sendAudio(audio: Buffer): void { + if (this.terminalError || this.intentionallyClosed || this.closeNotified) { + return; + } if (!this.session || !this.connected || !this.sessionConfigured) { - if (this.pendingAudio.length < MAX_PENDING_AUDIO_CHUNKS) { - this.pendingAudio.push(audio); + if ( + this.pendingAudio.length >= MAX_PENDING_AUDIO_CHUNKS || + this.pendingAudioBytes + audio.byteLength > MAX_PENDING_AUDIO_BYTES + ) { + return; } + const queuedAudio = Buffer.from(audio); + this.pendingAudio.push(queuedAudio); + this.pendingAudioBytes += queuedAudio.byteLength; return; } const silent = this.isSilence(audio); @@ -778,7 +789,7 @@ class GoogleRealtimeVoiceBridge implements RealtimeVoiceBridge { clearTimeout(this.reconnectTimer); this.reconnectTimer = undefined; } - this.pendingAudio = []; + this.clearPendingAudio(); this.consecutiveSilenceMs = 0; this.audioStreamEnded = false; this.pendingFunctionNames.clear(); @@ -877,7 +888,10 @@ class GoogleRealtimeVoiceBridge implements RealtimeVoiceBridge { } this.sessionConfigured = true; this.reconnectAttempts = 0; - for (const chunk of this.pendingAudio.splice(0)) { + const pendingAudio = this.pendingAudio; + this.pendingAudio = []; + this.pendingAudioBytes = 0; + for (const chunk of pendingAudio) { this.sendAudio(chunk); } if (!this.sessionReadyFired) { @@ -1015,10 +1029,16 @@ class GoogleRealtimeVoiceBridge implements RealtimeVoiceBridge { if (this.closeNotified) { return; } + this.clearPendingAudio(); this.closeNotified = true; this.config.onClose?.(reason); } + private clearPendingAudio(): void { + this.pendingAudio = []; + this.pendingAudioBytes = 0; + } + private cancelConnectAttempt(attempt: GoogleLiveConnectionAttempt | undefined): void { if (!attempt) { return; From aab146eeec49c54bb2543dbc80dfe9e6d481d8e9 Mon Sep 17 00:00:00 2001 From: Vincent Koc Date: Sun, 2 Aug 2026 03:59:37 +0800 Subject: [PATCH 2/7] fix(google): bound lazy realtime audio queue --- extensions/google/index.test.ts | 81 +++++++++++++++++++++++++++++++-- extensions/google/index.ts | 58 +++++++++++++++++++---- 2 files changed, 126 insertions(+), 13 deletions(-) diff --git a/extensions/google/index.test.ts b/extensions/google/index.test.ts index 118cc4cadc5c..c7391a1f6acf 100644 --- a/extensions/google/index.test.ts +++ b/extensions/google/index.test.ts @@ -55,6 +55,7 @@ function createDeferred() { function createMockRealtimeBridge(connectImpl: () => Promise = 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 = async () => supportsToolResultContinuation: false, supportsToolResultSuppression: false, connect, - sendAudio: vi.fn(), + sendAudio, setMediaTimestamp: vi.fn(), sendUserMessage, triggerGreeting, @@ -72,10 +73,14 @@ function createMockRealtimeBridge(connectImpl: () => Promise = 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(); const loaded = createMockRealtimeBridge(() => connected.promise); diff --git a/extensions/google/index.ts b/extensions/google/index.ts index 619371579db0..6e9d4c791676 100644 --- a/extensions/google/index.ts +++ b/extensions/google/index.ts @@ -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; From 24e6354ec2c61eadd2c1e50a6530f1be06da22e9 Mon Sep 17 00:00:00 2001 From: Vincent Koc Date: Sun, 2 Aug 2026 04:00:22 +0800 Subject: [PATCH 3/7] fix(google): preserve realtime startup audio order --- extensions/google/index.test.ts | 23 +++++++++++++++++++++++ extensions/google/index.ts | 2 +- 2 files changed, 24 insertions(+), 1 deletion(-) diff --git a/extensions/google/index.test.ts b/extensions/google/index.test.ts index c7391a1f6acf..024286e5ad27 100644 --- a/extensions/google/index.test.ts +++ b/extensions/google/index.test.ts @@ -517,6 +517,29 @@ describe("google provider plugin hooks", () => { expect(loaded.sendAudio.mock.calls.at(-1)?.[0]).toEqual(Buffer.from([65])); }); + it("preserves lazy audio order across bridge loading and provider readiness", async () => { + const connected = createDeferred(); + const loaded = createMockRealtimeBridge(() => connected.promise); + createRealtimeBridgeMock.mockReturnValue(loaded.bridge); + const { bridge } = createLazyRealtimeBridge(); + + bridge.sendAudio(Buffer.from([0x01])); + const connectPromise = bridge.connect(); + await vi.waitFor(() => expect(loaded.connect).toHaveBeenCalledOnce()); + bridge.sendAudio(Buffer.from([0x02])); + + expect(loaded.sendAudio).not.toHaveBeenCalled(); + connected.resolve(); + await connectPromise; + expect(loaded.sendAudio).not.toHaveBeenCalled(); + + signalRealtimeBridgeReady(); + expect(loaded.sendAudio.mock.calls.map(([audio]) => audio)).toEqual([ + Buffer.from([0x01]), + Buffer.from([0x02]), + ]); + }); + it("copies lazy audio and evicts oldest chunks to enforce the byte limit", async () => { const loaded = createMockRealtimeBridge(); createRealtimeBridgeMock.mockReturnValue(loaded.bridge); diff --git a/extensions/google/index.ts b/extensions/google/index.ts index 6e9d4c791676..77d7a8caabc0 100644 --- a/extensions/google/index.ts +++ b/extensions/google/index.ts @@ -326,7 +326,7 @@ function createLazyGoogleRealtimeVoiceBridge( if (closed || providerTerminated) { return; } - if (bridge) { + if (bridgeReady && bridge) { bridge.sendAudio(audio); return; } From 213c05182f29c794bfa6f727bbb2847434255365 Mon Sep 17 00:00:00 2001 From: Vincent Koc Date: Sun, 2 Aug 2026 05:07:49 +0800 Subject: [PATCH 4/7] refactor(google): centralize realtime audio queue bounds --- extensions/google/index.ts | 38 +++----------- .../google/realtime-audio-queue.test.ts | 34 +++++++++++++ extensions/google/realtime-audio-queue.ts | 49 +++++++++++++++++++ extensions/google/realtime-voice-provider.ts | 25 +++------- 4 files changed, 97 insertions(+), 49 deletions(-) create mode 100644 extensions/google/realtime-audio-queue.test.ts create mode 100644 extensions/google/realtime-audio-queue.ts diff --git a/extensions/google/index.ts b/extensions/google/index.ts index 77d7a8caabc0..8a8a95a357a8 100644 --- a/extensions/google/index.ts +++ b/extensions/google/index.ts @@ -20,6 +20,7 @@ import { } from "./generation-provider-metadata.js"; import { geminiMemoryEmbeddingProviderAdapter } from "./memory-embedding-adapter.js"; import { registerGoogleProvider } from "./provider-registration.js"; +import { createGoogleRealtimeAudioQueue } from "./realtime-audio-queue.js"; import { buildGoogleSpeechProvider } from "./speech-provider.js"; import { createGeminiWebSearchProvider } from "./src/gemini-web-search-provider.js"; @@ -201,8 +202,6 @@ 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; @@ -219,14 +218,10 @@ function createLazyGoogleRealtimeVoiceBridge( let providerTerminated = false; let latestMediaTimestamp: number | undefined; let pendingGreeting: string | undefined; - const pendingAudio: Buffer[] = []; - let pendingAudioBytes = 0; + // Lazy startup keeps the newest microphone tail when loading stalls. + const pendingAudio = createGoogleRealtimeAudioQueue("drop-oldest"); 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) => { @@ -257,7 +252,7 @@ function createLazyGoogleRealtimeVoiceBridge( onClose: (reason) => { bridgeReady = false; providerTerminated = true; - clearPendingAudio(); + pendingAudio.clear(); req.onClose?.(reason); }, }), @@ -282,9 +277,7 @@ function createLazyGoogleRealtimeVoiceBridge( if (typeof latestMediaTimestamp === "number") { loadedBridge.setMediaTimestamp(latestMediaTimestamp); } - const audioChunks = pendingAudio.splice(0); - pendingAudioBytes = 0; - for (const audio of audioChunks) { + for (const audio of pendingAudio.drain()) { loadedBridge.sendAudio(audio); } const userMessages = pendingUserMessages.splice(0); @@ -315,7 +308,7 @@ function createLazyGoogleRealtimeVoiceBridge( } catch (error) { bridgeReady = false; providerTerminated = true; - clearPendingAudio(); + pendingAudio.clear(); throw error; } if (closed) { @@ -330,22 +323,7 @@ function createLazyGoogleRealtimeVoiceBridge( bridge.sendAudio(audio); return; } - if (audio.byteLength > GOOGLE_REALTIME_LAZY_MAX_PENDING_AUDIO_BYTES) { - return; - } - 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; + pendingAudio.enqueue(audio); }, setMediaTimestamp: (ts) => { if (closed) { @@ -393,7 +371,7 @@ function createLazyGoogleRealtimeVoiceBridge( closed = true; bridgeReady = false; providerTerminated = true; - clearPendingAudio(); + pendingAudio.clear(); pendingUserMessages.length = 0; pendingUserMessageBytes = 0; pendingGreeting = undefined; diff --git a/extensions/google/realtime-audio-queue.test.ts b/extensions/google/realtime-audio-queue.test.ts new file mode 100644 index 000000000000..7aee17a24840 --- /dev/null +++ b/extensions/google/realtime-audio-queue.test.ts @@ -0,0 +1,34 @@ +import { describe, expect, it } from "vitest"; +import { createGoogleRealtimeAudioQueue } from "./realtime-audio-queue.js"; + +describe("Google realtime audio queue", () => { + it("rejects newest audio without retaining caller-owned buffers", () => { + const queue = createGoogleRealtimeAudioQueue("reject-newest"); + const backing = Buffer.alloc(2 * 1024 * 1024, 0x01); + const retainedView = backing.subarray(0, 512 * 1024); + + expect(queue.enqueue(retainedView)).toBe(true); + retainedView.fill(0); + expect(queue.enqueue(Buffer.alloc(512 * 1024, 0x02))).toBe(true); + expect(queue.enqueue(Buffer.from([0x03]))).toBe(false); + + expect(queue.drain()).toEqual([Buffer.alloc(512 * 1024, 0x01), Buffer.alloc(512 * 1024, 0x02)]); + }); + + it("drops oldest audio and resets accounting on clear", () => { + const queue = createGoogleRealtimeAudioQueue("drop-oldest"); + for (let index = 0; index < 322; index += 1) { + expect(queue.enqueue(Buffer.from([index & 0xff]))).toBe(true); + } + + const drained = queue.drain(); + expect(drained).toHaveLength(320); + expect(drained[0]).toEqual(Buffer.from([2])); + expect(drained.at(-1)).toEqual(Buffer.from([65])); + + expect(queue.enqueue(Buffer.alloc(1024 * 1024, 0x04))).toBe(true); + queue.clear(); + expect(queue.enqueue(Buffer.from([0x05]))).toBe(true); + expect(queue.drain()).toEqual([Buffer.from([0x05])]); + }); +}); diff --git a/extensions/google/realtime-audio-queue.ts b/extensions/google/realtime-audio-queue.ts new file mode 100644 index 000000000000..bcb32c7bdbf7 --- /dev/null +++ b/extensions/google/realtime-audio-queue.ts @@ -0,0 +1,49 @@ +const GOOGLE_REALTIME_MAX_PENDING_AUDIO_CHUNKS = 320; +const GOOGLE_REALTIME_MAX_PENDING_AUDIO_BYTES = 1024 * 1024; + +export type GoogleRealtimeAudioOverflowPolicy = "drop-oldest" | "reject-newest"; + +export function createGoogleRealtimeAudioQueue(overflowPolicy: GoogleRealtimeAudioOverflowPolicy) { + let chunks: Buffer[] = []; + let bytes = 0; + + const clear = () => { + chunks = []; + bytes = 0; + }; + + return { + clear, + drain: (): Buffer[] => { + const drained = chunks; + clear(); + return drained; + }, + enqueue: (audio: Buffer): boolean => { + if (audio.byteLength > GOOGLE_REALTIME_MAX_PENDING_AUDIO_BYTES) { + return false; + } + const chunk = Buffer.from(audio); + if ( + overflowPolicy === "reject-newest" && + (chunks.length >= GOOGLE_REALTIME_MAX_PENDING_AUDIO_CHUNKS || + bytes + chunk.byteLength > GOOGLE_REALTIME_MAX_PENDING_AUDIO_BYTES) + ) { + return false; + } + while ( + chunks.length >= GOOGLE_REALTIME_MAX_PENDING_AUDIO_CHUNKS || + bytes + chunk.byteLength > GOOGLE_REALTIME_MAX_PENDING_AUDIO_BYTES + ) { + const dropped = chunks.shift(); + if (!dropped) { + return false; + } + bytes -= dropped.byteLength; + } + chunks.push(chunk); + bytes += chunk.byteLength; + return true; + }, + }; +} diff --git a/extensions/google/realtime-voice-provider.ts b/extensions/google/realtime-voice-provider.ts index 3ddbd60403a1..7ae20b887343 100644 --- a/extensions/google/realtime-voice-provider.ts +++ b/extensions/google/realtime-voice-provider.ts @@ -52,6 +52,7 @@ import { normalizeOptionalString, } from "openclaw/plugin-sdk/string-coerce-runtime"; import { createGoogleGenAI } from "./google-genai-runtime.js"; +import { createGoogleRealtimeAudioQueue } from "./realtime-audio-queue.js"; import { resolveGoogleGemini3ThinkingLevel } from "./thinking.js"; const GOOGLE_REALTIME_DEFAULT_MODEL = "gemini-3.1-flash-live-preview"; @@ -61,8 +62,6 @@ const GOOGLE_REALTIME_INPUT_SAMPLE_RATE = 16_000; const GOOGLE_REALTIME_BROWSER_API_VERSION = "v1alpha"; const GOOGLE_REALTIME_BROWSER_WEBSOCKET_URL = "wss://generativelanguage.googleapis.com/ws/google.ai.generativelanguage.v1alpha.GenerativeService.BidiGenerateContentConstrained"; -const MAX_PENDING_AUDIO_CHUNKS = 320; -const MAX_PENDING_AUDIO_BYTES = 1024 * 1024; const DEFAULT_AUDIO_STREAM_END_SILENCE_MS = 500; const GOOGLE_REALTIME_BROWSER_SESSION_TTL_MS = 30 * 60 * 1000; const GOOGLE_REALTIME_BROWSER_NEW_SESSION_TTL_MS = 60 * 1000; @@ -472,8 +471,8 @@ class GoogleRealtimeVoiceBridge implements RealtimeVoiceBridge { private setupCompleteReceived = false; private sessionConfigured = false; private intentionallyClosed = false; - private pendingAudio: Buffer[] = []; - private pendingAudioBytes = 0; + // Native reconnect keeps the already accepted FIFO prefix stable. + private readonly pendingAudio = createGoogleRealtimeAudioQueue("reject-newest"); private sessionReadyFired = false; private consecutiveSilenceMs = 0; private audioStreamEnded = false; @@ -647,15 +646,7 @@ class GoogleRealtimeVoiceBridge implements RealtimeVoiceBridge { return; } if (!this.session || !this.connected || !this.sessionConfigured) { - if ( - this.pendingAudio.length >= MAX_PENDING_AUDIO_CHUNKS || - this.pendingAudioBytes + audio.byteLength > MAX_PENDING_AUDIO_BYTES - ) { - return; - } - const queuedAudio = Buffer.from(audio); - this.pendingAudio.push(queuedAudio); - this.pendingAudioBytes += queuedAudio.byteLength; + this.pendingAudio.enqueue(audio); return; } const silent = this.isSilence(audio); @@ -888,10 +879,7 @@ class GoogleRealtimeVoiceBridge implements RealtimeVoiceBridge { } this.sessionConfigured = true; this.reconnectAttempts = 0; - const pendingAudio = this.pendingAudio; - this.pendingAudio = []; - this.pendingAudioBytes = 0; - for (const chunk of pendingAudio) { + for (const chunk of this.pendingAudio.drain()) { this.sendAudio(chunk); } if (!this.sessionReadyFired) { @@ -1035,8 +1023,7 @@ class GoogleRealtimeVoiceBridge implements RealtimeVoiceBridge { } private clearPendingAudio(): void { - this.pendingAudio = []; - this.pendingAudioBytes = 0; + this.pendingAudio.clear(); } private cancelConnectAttempt(attempt: GoogleLiveConnectionAttempt | undefined): void { From 1ae9b1cfc54ba91836b9012676a2da6ac924c93d Mon Sep 17 00:00:00 2001 From: Vincent Koc Date: Sun, 2 Aug 2026 05:12:21 +0800 Subject: [PATCH 5/7] test(google): preserve queued audio across fresh reconnect --- .../google/realtime-voice-provider.test.ts | 22 ++++++++++++++++--- 1 file changed, 19 insertions(+), 3 deletions(-) diff --git a/extensions/google/realtime-voice-provider.test.ts b/extensions/google/realtime-voice-provider.test.ts index 47efc00225ae..32a7231ed96b 100644 --- a/extensions/google/realtime-voice-provider.test.ts +++ b/extensions/google/realtime-voice-provider.test.ts @@ -1131,6 +1131,8 @@ describe("buildGoogleRealtimeVoiceProvider", () => { firstCallbacks.onopen(); firstCallbacks.onmessage({ setupComplete: {} }); firstCallbacks.onclose({ code: 1011, reason: "temporary" }); + const queuedAudio = Buffer.from([0x7f]); + bridge.sendAudio(queuedAudio); await vi.advanceTimersByTimeAsync(250); const freshCallbacks = lastConnectParams().callbacks; @@ -1145,17 +1147,31 @@ describe("buildGoogleRealtimeVoiceProvider", () => { "session.created", ]); expect(onReady).toHaveBeenCalledTimes(1); + expect(freshSession.sendRealtimeInput).not.toHaveBeenCalled(); pendingSession.resolve(freshSession); await vi.waitFor(() => { expect(onReady).toHaveBeenCalledTimes(2); }); const sessionCreatedOrder = onEvent.mock.invocationCallOrder[1]; + const queuedAudioOrder = freshSession.sendRealtimeInput.mock.invocationCallOrder[0]; const freshReadyOrder = onReady.mock.invocationCallOrder[1]; - if (sessionCreatedOrder === undefined || freshReadyOrder === undefined) { - throw new Error("expected fresh session creation before readiness"); + if ( + sessionCreatedOrder === undefined || + queuedAudioOrder === undefined || + freshReadyOrder === undefined + ) { + throw new Error("expected fresh session creation, queued audio, and readiness"); } - expect(sessionCreatedOrder).toBeLessThan(freshReadyOrder); + expect(sessionCreatedOrder).toBeLessThan(queuedAudioOrder); + expect(queuedAudioOrder).toBeLessThan(freshReadyOrder); + expect(freshSession.sendRealtimeInput).toHaveBeenCalledOnce(); + expect(freshSession.sendRealtimeInput).toHaveBeenCalledWith({ + audio: { + data: expect.any(String), + mimeType: "audio/pcm;rate=16000", + }, + }); freshCallbacks.onclose({ code: 1011, reason: "temporary again" }); await vi.advanceTimersByTimeAsync(250); From 30f8fbc67ae2afa74cc7cbf530942c83998b473e Mon Sep 17 00:00:00 2001 From: Vincent Koc Date: Sun, 2 Aug 2026 05:23:15 +0800 Subject: [PATCH 6/7] perf(google): avoid copying rejected realtime audio --- extensions/google/realtime-audio-queue.ts | 6 +++--- 1 file changed, 3 insertions(+), 3 deletions(-) diff --git a/extensions/google/realtime-audio-queue.ts b/extensions/google/realtime-audio-queue.ts index bcb32c7bdbf7..88462050e5c1 100644 --- a/extensions/google/realtime-audio-queue.ts +++ b/extensions/google/realtime-audio-queue.ts @@ -23,17 +23,16 @@ export function createGoogleRealtimeAudioQueue(overflowPolicy: GoogleRealtimeAud if (audio.byteLength > GOOGLE_REALTIME_MAX_PENDING_AUDIO_BYTES) { return false; } - const chunk = Buffer.from(audio); if ( overflowPolicy === "reject-newest" && (chunks.length >= GOOGLE_REALTIME_MAX_PENDING_AUDIO_CHUNKS || - bytes + chunk.byteLength > GOOGLE_REALTIME_MAX_PENDING_AUDIO_BYTES) + bytes + audio.byteLength > GOOGLE_REALTIME_MAX_PENDING_AUDIO_BYTES) ) { return false; } while ( chunks.length >= GOOGLE_REALTIME_MAX_PENDING_AUDIO_CHUNKS || - bytes + chunk.byteLength > GOOGLE_REALTIME_MAX_PENDING_AUDIO_BYTES + bytes + audio.byteLength > GOOGLE_REALTIME_MAX_PENDING_AUDIO_BYTES ) { const dropped = chunks.shift(); if (!dropped) { @@ -41,6 +40,7 @@ export function createGoogleRealtimeAudioQueue(overflowPolicy: GoogleRealtimeAud } bytes -= dropped.byteLength; } + const chunk = Buffer.from(audio); chunks.push(chunk); bytes += chunk.byteLength; return true; From ee6302161a182dcb45aca328de21018888196d28 Mon Sep 17 00:00:00 2001 From: Vincent Koc Date: Sun, 2 Aug 2026 05:54:34 +0800 Subject: [PATCH 7/7] fix(google): keep realtime queue policy internal --- extensions/google/realtime-audio-queue.ts | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/extensions/google/realtime-audio-queue.ts b/extensions/google/realtime-audio-queue.ts index 88462050e5c1..ce9f08100aa2 100644 --- a/extensions/google/realtime-audio-queue.ts +++ b/extensions/google/realtime-audio-queue.ts @@ -1,7 +1,7 @@ const GOOGLE_REALTIME_MAX_PENDING_AUDIO_CHUNKS = 320; const GOOGLE_REALTIME_MAX_PENDING_AUDIO_BYTES = 1024 * 1024; -export type GoogleRealtimeAudioOverflowPolicy = "drop-oldest" | "reject-newest"; +type GoogleRealtimeAudioOverflowPolicy = "drop-oldest" | "reject-newest"; export function createGoogleRealtimeAudioQueue(overflowPolicy: GoogleRealtimeAudioOverflowPolicy) { let chunks: Buffer[] = [];