fix(google): keep realtime startup audio bounded and ordered (#117640)

* commit 'ee6302161a182dcb45aca328de21018888196d28':
  fix(google): keep realtime queue policy internal
  perf(google): avoid copying rejected realtime audio
  test(google): preserve queued audio across fresh reconnect
  refactor(google): centralize realtime audio queue bounds
  fix(google): preserve realtime startup audio order
  fix(google): bound lazy realtime audio queue
  fix(google): bound native realtime audio queue

extensions/google/index.test.ts

@@ -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,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<void>();
+    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<void>();
     const loaded = createMockRealtimeBridge(() => connected.promise);

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;

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])]);
+  });
+});

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;
+    },
+  };
+}

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<MockGoogleLiveSession>();
     const lateSession = createMockGoogleLiveSession();

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;