diff --git a/extensions/google/index.test.ts b/extensions/google/index.test.ts index 118cc4cadc5c..024286e5ad27 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,90 @@ 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("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); + 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..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,7 +202,6 @@ function resolveGoogleRealtimeEnvApiKey(): string | undefined { ); } -const GOOGLE_REALTIME_LAZY_MAX_PENDING_AUDIO_CHUNKS = 320; const GOOGLE_REALTIME_LAZY_MAX_PENDING_USER_MESSAGES = 128; const GOOGLE_REALTIME_LAZY_MAX_PENDING_USER_MESSAGE_BYTES = 256 * 1024; @@ -213,9 +213,13 @@ 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[] = []; + // Lazy startup keeps the newest microphone tail when loading stalls. + const pendingAudio = createGoogleRealtimeAudioQueue("drop-oldest"); const pendingUserMessages: string[] = []; let pendingUserMessageBytes = 0; // Loading and connecting finish on separate async boundaries. Keep close ownership @@ -233,11 +237,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 +249,12 @@ function createLazyGoogleRealtimeVoiceBridge( // Release prompts only after the provider can accept user content. flushPending(bridge); }, + onClose: (reason) => { + bridgeReady = false; + providerTerminated = true; + pendingAudio.clear(); + req.onClose?.(reason); + }, }), ); } @@ -261,13 +271,13 @@ 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)) { + for (const audio of pendingAudio.drain()) { loadedBridge.sendAudio(audio); } const userMessages = pendingUserMessages.splice(0); @@ -292,23 +302,28 @@ function createLazyGoogleRealtimeVoiceBridge( closeBridge(loadedBridge); return; } - await loadedBridge.connect(); + providerTerminated = false; + try { + await loadedBridge.connect(); + } catch (error) { + bridgeReady = false; + providerTerminated = true; + pendingAudio.clear(); + throw error; + } if (closed) { closeBridge(loadedBridge); } }, sendAudio: (audio) => { - if (closed) { + if (closed || providerTerminated) { return; } - if (bridge) { + if (bridgeReady && bridge) { bridge.sendAudio(audio); return; } - if (pendingAudio.length >= GOOGLE_REALTIME_LAZY_MAX_PENDING_AUDIO_CHUNKS) { - pendingAudio.shift(); - } - pendingAudio.push(audio); + pendingAudio.enqueue(audio); }, setMediaTimestamp: (ts) => { if (closed) { @@ -355,7 +370,8 @@ function createLazyGoogleRealtimeVoiceBridge( close: () => { closed = true; bridgeReady = false; - pendingAudio.length = 0; + providerTerminated = true; + 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..ce9f08100aa2 --- /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; + +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; + } + if ( + overflowPolicy === "reject-newest" && + (chunks.length >= GOOGLE_REALTIME_MAX_PENDING_AUDIO_CHUNKS || + bytes + audio.byteLength > GOOGLE_REALTIME_MAX_PENDING_AUDIO_BYTES) + ) { + return false; + } + while ( + chunks.length >= GOOGLE_REALTIME_MAX_PENDING_AUDIO_CHUNKS || + bytes + audio.byteLength > GOOGLE_REALTIME_MAX_PENDING_AUDIO_BYTES + ) { + const dropped = chunks.shift(); + if (!dropped) { + return false; + } + bytes -= dropped.byteLength; + } + const chunk = Buffer.from(audio); + chunks.push(chunk); + bytes += chunk.byteLength; + return true; + }, + }; +} diff --git a/extensions/google/realtime-voice-provider.test.ts b/extensions/google/realtime-voice-provider.test.ts index 7257aa57b7f0..32a7231ed96b 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"; @@ -1128,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; @@ -1142,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); @@ -1213,6 +1232,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..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,7 +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 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; @@ -471,7 +471,8 @@ class GoogleRealtimeVoiceBridge implements RealtimeVoiceBridge { private setupCompleteReceived = false; private sessionConfigured = false; private intentionallyClosed = false; - private pendingAudio: Buffer[] = []; + // Native reconnect keeps the already accepted FIFO prefix stable. + private readonly pendingAudio = createGoogleRealtimeAudioQueue("reject-newest"); private sessionReadyFired = false; private consecutiveSilenceMs = 0; private audioStreamEnded = false; @@ -641,10 +642,11 @@ 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); - } + this.pendingAudio.enqueue(audio); return; } const silent = this.isSilence(audio); @@ -778,7 +780,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 +879,7 @@ class GoogleRealtimeVoiceBridge implements RealtimeVoiceBridge { } this.sessionConfigured = true; this.reconnectAttempts = 0; - for (const chunk of this.pendingAudio.splice(0)) { + for (const chunk of this.pendingAudio.drain()) { this.sendAudio(chunk); } if (!this.sessionReadyFired) { @@ -1015,10 +1017,15 @@ class GoogleRealtimeVoiceBridge implements RealtimeVoiceBridge { if (this.closeNotified) { return; } + this.clearPendingAudio(); this.closeNotified = true; this.config.onClose?.(reason); } + private clearPendingAudio(): void { + this.pendingAudio.clear(); + } + private cancelConnectAttempt(attempt: GoogleLiveConnectionAttempt | undefined): void { if (!attempt) { return;