From aab146eeec49c54bb2543dbc80dfe9e6d481d8e9 Mon Sep 17 00:00:00 2001 From: Vincent Koc Date: Sun, 2 Aug 2026 03:59:37 +0800 Subject: [PATCH] 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;