fix(openai): unify gateway microphone buffering

extensions/openai/realtime-quicksilver-audio-buffer.test.ts

@@ -0,0 +1,37 @@
+import { describe, expect, it } from "vitest";
+import {
+  appendOpenAIQuicksilverPendingAudio,
+  OPENAI_QUICKSILVER_MAX_PENDING_AUDIO_BYTES,
+} from "./realtime-quicksilver-audio-buffer.js";
+
+describe("GPT-Live pending microphone audio", () => {
+  it("copies caller-owned PCM16 and drops an incomplete sample", () => {
+    const source = Buffer.from([0x01, 0x02, 0x03]);
+    const pending = appendOpenAIQuicksilverPendingAudio(Buffer.alloc(0), source);
+    source.fill(0xff);
+
+    expect(pending).toEqual(Buffer.from([0x01, 0x02]));
+  });
+
+  it("appends audio in capture order while it fits", () => {
+    const first = appendOpenAIQuicksilverPendingAudio(Buffer.alloc(0), Buffer.from([0x01, 0x02]));
+    const second = appendOpenAIQuicksilverPendingAudio(first, Buffer.from([0x03, 0x04]));
+
+    expect(second).toEqual(Buffer.from([0x01, 0x02, 0x03, 0x04]));
+  });
+
+  it("retains the newest bounded tail across existing and oversized input", () => {
+    const existing = Buffer.alloc(OPENAI_QUICKSILVER_MAX_PENDING_AUDIO_BYTES, 0x01);
+    const appended = appendOpenAIQuicksilverPendingAudio(existing, Buffer.from([0x02, 0x02]));
+    const oversized = Buffer.alloc(OPENAI_QUICKSILVER_MAX_PENDING_AUDIO_BYTES + 2, 0x03);
+
+    expect(appended).toHaveLength(OPENAI_QUICKSILVER_MAX_PENDING_AUDIO_BYTES);
+    expect(appended.subarray(0, -2).every((byte) => byte === 0x01)).toBe(true);
+    expect(appended.subarray(-2)).toEqual(Buffer.from([0x02, 0x02]));
+    expect(
+      appendOpenAIQuicksilverPendingAudio(appended, oversized).equals(
+        Buffer.alloc(OPENAI_QUICKSILVER_MAX_PENDING_AUDIO_BYTES, 0x03),
+      ),
+    ).toBe(true);
+  });
+});

extensions/openai/realtime-quicksilver-audio-buffer.ts

@@ -0,0 +1,27 @@
+const RELAY_FRAME_SAMPLES = 480;
+const MAX_PENDING_RELAY_FRAMES = 250;
+
+export const OPENAI_QUICKSILVER_RELAY_FRAME_BYTES = RELAY_FRAME_SAMPLES * 2;
+// One five-second tail spans peer startup and the connected media pump.
+// Keeping the newest PCM bounds latency without changing policy at adoption.
+export const OPENAI_QUICKSILVER_MAX_PENDING_AUDIO_BYTES =
+  OPENAI_QUICKSILVER_RELAY_FRAME_BYTES * MAX_PENDING_RELAY_FRAMES;
+
+export function appendOpenAIQuicksilverPendingAudio(pending: Buffer, incoming: Buffer): Buffer {
+  const evenLength = incoming.length - (incoming.length % 2);
+  if (evenLength === 0) {
+    return pending;
+  }
+  const audio = incoming.subarray(0, evenLength);
+  if (audio.length >= OPENAI_QUICKSILVER_MAX_PENDING_AUDIO_BYTES) {
+    return Buffer.from(audio.subarray(audio.length - OPENAI_QUICKSILVER_MAX_PENDING_AUDIO_BYTES));
+  }
+  const pendingBytes = Math.min(
+    pending.length,
+    OPENAI_QUICKSILVER_MAX_PENDING_AUDIO_BYTES - audio.length,
+  );
+  return Buffer.concat(
+    [pending.subarray(pending.length - pendingBytes), audio],
+    pendingBytes + audio.length,
+  );
+}

extensions/openai/realtime-quicksilver-gateway-bridge.test.ts

@@ -1,5 +1,9 @@
 import { spawnSync } from "node:child_process";
 import { describe, expect, it, vi } from "vitest";
+import {
+  OPENAI_QUICKSILVER_MAX_PENDING_AUDIO_BYTES,
+  OPENAI_QUICKSILVER_RELAY_FRAME_BYTES,
+} from "./realtime-quicksilver-audio-buffer.js";
 import { OpenAIQuicksilverGatewayBridge } from "./realtime-quicksilver-gateway-bridge.js";
 import {
   OpenAIQuicksilverAudioPeer,
@@ -46,7 +50,7 @@ type TestableAudioPeer = {
     };
     peer: {
       connectionStateChange: {
-        execute(state: "closed" | "disconnected"): void;
+        execute(state: "closed" | "connected" | "disconnected"): void;
       };
     };
     transceiver: {
@@ -574,66 +578,117 @@ describe("GPT-Live gateway relay bridge", () => {
       resolvePeer();
       await connection;
 
-      expect(peer.sendAudio.mock.calls.map(([audio]) => audio)).toEqual([
-        Buffer.from([0x7f, 0x41]),
-        Buffer.from([0x22, 0x23]),
-      ]);
+      expect(peer.sendAudio).toHaveBeenCalledWith(Buffer.from([0x7f, 0x41, 0x22, 0x23]));
       bridge.sendAudio(Buffer.from([0x30, 0x31]));
-      expect(peer.sendAudio).toHaveBeenCalledTimes(3);
-    } finally {
-      bridge.close();
-    }
-  });
-
-  it("bounds queued microphone bytes before copying rejected audio", async () => {
-    const { bridge, connection, peer, resolvePeer } = createPendingPeerBridge();
-    try {
-      bridge.sendAudio(Buffer.alloc(512 * 1024, 0x01));
-      bridge.sendAudio(Buffer.alloc(512 * 1024, 0x02));
-      const overflow = Buffer.alloc(1, 0x03);
-      const copy = vi.spyOn(Buffer, "from");
-      try {
-        bridge.sendAudio(overflow);
-        expect(copy).not.toHaveBeenCalled();
-      } finally {
-        copy.mockRestore();
-      }
-
-      resolvePeer();
-      await connection;
-
       expect(peer.sendAudio).toHaveBeenCalledTimes(2);
-      expect(peer.sendAudio.mock.calls.map(([audio]) => audio.byteLength)).toEqual([
-        512 * 1024,
-        512 * 1024,
-      ]);
     } finally {
       bridge.close();
     }
   });
 
-  it("bounds queued microphone frame count before copying rejected audio", async () => {
-    const { bridge, connection, peer, resolvePeer } = createPendingPeerBridge();
+  it("retains the newest five seconds through delayed peer adoption and the RTP pump", async () => {
+    vi.useFakeTimers();
+    let peerCallbacks:
+      | Parameters<typeof OpenAIQuicksilverAudioPeer.create>[0]["callbacks"]
+      | undefined;
+    let resolvePeer: ((peer: OpenAIQuicksilverAudioPeerContract) => void) | undefined;
+    const peerPromise = new Promise<OpenAIQuicksilverAudioPeerContract>((resolve) => {
+      resolvePeer = resolve;
+    });
+    const onError = vi.fn();
+    const bridge = new OpenAIQuicksilverGatewayBridge({
+      providerConfig: {},
+      model: "gpt-live-1-codex",
+      voice: "marin",
+      audioFormat: { encoding: "pcm16", sampleRateHz: 24_000, channels: 1 },
+      onAudio: vi.fn(),
+      onClearAudio: vi.fn(),
+      onError,
+      runAgentConsult: vi.fn(async () => ({ text: "done" })),
+      logger: { debug: vi.fn(), warn: vi.fn() },
+      resolveAuth: vi.fn(async () => ({
+        type: "oauth" as const,
+        token: "oauth-token",
+        accountId: "account-1",
+      })),
+      createPeer: vi.fn((callbacks) => {
+        peerCallbacks = callbacks;
+        return peerPromise;
+      }),
+      fetchImpl: vi.fn(async () => createCallResponse("v=answer\r\n", "rtc_pending_audio")),
+      webSocketFactory: () => new FakeSocket(),
+    });
+    const connection = bridge.connect();
+    const source = Buffer.alloc(256 * OPENAI_QUICKSILVER_RELAY_FRAME_BYTES);
+    for (let frame = 0; frame < 256; frame += 1) {
+      source
+        .subarray(
+          frame * OPENAI_QUICKSILVER_RELAY_FRAME_BYTES,
+          (frame + 1) * OPENAI_QUICKSILVER_RELAY_FRAME_BYTES,
+        )
+        .fill(frame + 1);
+    }
+    for (let offset = 0; offset < source.length; offset += 8_192) {
+      const callerBuffer = Buffer.from(source.subarray(offset, offset + 8_192));
+      bridge.sendAudio(callerBuffer);
+      callerBuffer.fill(0xff);
+      await vi.advanceTimersByTimeAsync(171);
+    }
+    if (!peerCallbacks) {
+      throw new Error("expected the bridge to start peer creation");
+    }
+    const peer = await OpenAIQuicksilverAudioPeer.create({
+      callbacks: peerCallbacks,
+      iceServers: [],
+    });
+    vi.spyOn(peer, "createOffer").mockResolvedValue("v=offer\r\n");
+    vi.spyOn(peer, "applyAnswer").mockResolvedValue();
+    const testPeer = peer as unknown as TestableAudioPeer;
+    const sendRtp = vi
+      .spyOn(testPeer.state.transceiver.sender, "sendRtp")
+      .mockResolvedValue(undefined);
+    const emittedFrames: Buffer[] = [];
+    const takeNextRelayFrame = testPeer.takeNextRelayFrame.bind(testPeer);
+    vi.spyOn(testPeer, "takeNextRelayFrame").mockImplementation(() => {
+      const frame = takeNextRelayFrame();
+      emittedFrames.push(frame);
+      return frame;
+    });
+    const initialSequenceNumber = testPeer.sequenceNumber;
+    const initialTimestamp = testPeer.timestamp;
     try {
-      for (let index = 0; index < 320; index += 1) {
-        bridge.sendAudio(Buffer.alloc(2, index));
-      }
-      const overflow = Buffer.alloc(2, 0xff);
-      const copy = vi.spyOn(Buffer, "from");
-      try {
-        bridge.sendAudio(overflow);
-        expect(copy).not.toHaveBeenCalled();
-      } finally {
-        copy.mockRestore();
-      }
-
-      resolvePeer();
+      resolvePeer?.(peer);
       await connection;
 
-      expect(peer.sendAudio).toHaveBeenCalledTimes(320);
-      expect(peer.sendAudio.mock.calls.at(-1)?.[0]).toEqual(Buffer.alloc(2, 319));
+      expect(testPeer.pendingAudio).toEqual(
+        source.subarray(source.length - OPENAI_QUICKSILVER_MAX_PENDING_AUDIO_BYTES),
+      );
+
+      testPeer.state.peer.connectionStateChange.execute("connected");
+      await vi.advanceTimersByTimeAsync(4_980);
+
+      expect(testPeer.pendingAudio).toHaveLength(0);
+      expect(emittedFrames).toHaveLength(250);
+      expect(emittedFrames.map((frame) => frame[0])).toEqual([
+        ...Array.from({ length: 249 }, (_, index) => index + 7),
+        0,
+      ]);
+      expect(sendRtp).toHaveBeenCalledTimes(250);
+      const packets = sendRtp.mock.calls.map(
+        ([packet]) =>
+          packet as {
+            header: { payloadType: number; sequenceNumber: number; timestamp: number };
+            payload: Buffer;
+          },
+      );
+      expect(packets.every((packet) => packet.header.payloadType === 111)).toBe(true);
+      expect(packets.every((packet) => packet.payload.length > 0)).toBe(true);
+      expect(packets.at(-1)?.header.sequenceNumber).toBe((initialSequenceNumber + 249) & 0xffff);
+      expect(packets.at(-1)?.header.timestamp).toBe((initialTimestamp + 249 * 960) >>> 0);
+      expect(onError).not.toHaveBeenCalled();
     } finally {
       bridge.close();
+      vi.useRealTimers();
     }
   });
 
@@ -653,17 +708,15 @@ describe("GPT-Live gateway relay bridge", () => {
   it("discards queued microphone audio when media peer creation fails", async () => {
     const { bridge, connection, peer, rejectPeer } = createPendingPeerBridge();
     const pendingAudioState = bridge as unknown as {
-      pendingAudio: Buffer[];
-      pendingAudioBytes: number;
+      pendingAudio: Buffer;
     };
     bridge.sendAudio(Buffer.from([0x41, 0x42]));
     rejectPeer(new Error("media peer unavailable"));
 
     await expect(connection).rejects.toThrow("media peer unavailable");
-    expect(pendingAudioState.pendingAudio).toEqual([]);
-    expect(pendingAudioState.pendingAudioBytes).toBe(0);
+    expect(pendingAudioState.pendingAudio).toHaveLength(0);
     bridge.sendAudio(Buffer.from([0x43, 0x44]));
-    expect(pendingAudioState.pendingAudio).toEqual([]);
+    expect(pendingAudioState.pendingAudio).toHaveLength(0);
     expect(peer.sendAudio).not.toHaveBeenCalled();
   });
 

extensions/openai/realtime-quicksilver-gateway-bridge.ts

@@ -7,6 +7,7 @@ import type {
   RealtimeVoiceBridgeCreateRequest,
 } from "openclaw/plugin-sdk/realtime-voice";
 import WebSocket, { type RawData } from "ws";
+import { appendOpenAIQuicksilverPendingAudio } from "./realtime-quicksilver-audio-buffer.js";
 import {
   buildOpenAIQuicksilverDelegationPrompt,
   type OpenAIQuicksilverTranscriptEntry,
@@ -40,7 +41,6 @@ const RELAY_SAMPLE_RATE = 24_000;
 const QUICKSILVER_SESSION_TTL_MS = 30 * 60_000;
 const QUICKSILVER_CONNECT_TIMEOUT_MS = 30_000;
 const WEBSOCKET_OPEN = 1;
-const MAX_PENDING_AUDIO = { chunks: 320, bytes: 1024 * 1024 };
 
 function toError(error: unknown): Error {
   return error instanceof Error ? error : new Error(String(error));
@@ -165,8 +165,7 @@ export class OpenAIQuicksilverGatewayBridge implements RealtimeVoiceBridge {
   private closed = false;
   private closeNotified = false;
   private peer: OpenAIQuicksilverAudioPeerContract | undefined;
-  private pendingAudio: Buffer[] = [];
-  private pendingAudioBytes = 0;
+  private pendingAudio: Buffer = Buffer.alloc(0);
   private ready = false;
   private sideband: ActiveSideband | undefined;
   private timer: ReturnType<typeof setTimeout> | undefined;
@@ -186,15 +185,9 @@ export class OpenAIQuicksilverGatewayBridge implements RealtimeVoiceBridge {
   sendAudio(audio: Buffer): void {
     if (this.peer) {
       this.peer.sendAudio(audio);
-    } else if (
-      !this.closed &&
-      !this.abortController.signal.aborted &&
-      this.pendingAudio.length < MAX_PENDING_AUDIO.chunks &&
-      this.pendingAudioBytes + audio.byteLength <= MAX_PENDING_AUDIO.bytes
-    ) {
+    } else if (!this.closed && !this.abortController.signal.aborted) {
       // Relay capture starts before asynchronous peer creation and may recycle its input buffers.
-      this.pendingAudio.push(Buffer.from(audio));
-      this.pendingAudioBytes += audio.byteLength;
+      this.pendingAudio = appendOpenAIQuicksilverPendingAudio(this.pendingAudio, audio);
     }
   }
 
@@ -269,10 +262,10 @@ export class OpenAIQuicksilverGatewayBridge implements RealtimeVoiceBridge {
         () => undefined,
       );
       this.peer = await waitForConnectStep(peerPromise, connectSignal);
-      for (const audio of this.pendingAudio.splice(0)) {
-        this.peer.sendAudio(audio);
+      if (this.pendingAudio.length > 0) {
+        this.peer.sendAudio(this.pendingAudio);
+        this.pendingAudio = Buffer.alloc(0);
       }
-      this.pendingAudioBytes = 0;
       const offerSdp = await waitForConnectStep(this.peer.createOffer(), connectSignal);
       const auth = await waitForConnectStep(this.config.resolveAuth(), connectSignal);
       const requestIds = {
@@ -553,8 +546,7 @@ export class OpenAIQuicksilverGatewayBridge implements RealtimeVoiceBridge {
   private releaseResources(): void {
     releaseOpenAIQuicksilverSession(this);
     this.connected = false;
-    this.pendingAudio = [];
-    this.pendingAudioBytes = 0;
+    this.pendingAudio = Buffer.alloc(0);
     this.abortController.abort(new Error("GPT-Live gateway relay bridge closed"));
     this.consultController?.abort(new Error("GPT-Live delegation stopped"));
     this.consultController = undefined;

extensions/openai/realtime-quicksilver-peer.runtime.ts

@@ -1,15 +1,16 @@
 // Lazy GPT-Live media runtime: werift peer plus WASM Opus framing and PCM conversion.
 import { randomInt } from "node:crypto";
 import { resamplePcm } from "openclaw/plugin-sdk/realtime-voice";
+import {
+  appendOpenAIQuicksilverPendingAudio,
+  OPENAI_QUICKSILVER_RELAY_FRAME_BYTES,
+} from "./realtime-quicksilver-audio-buffer.js";
 
 const QUICKSILVER_SAMPLE_RATE = 48_000;
 const RELAY_SAMPLE_RATE = 24_000;
 const QUICKSILVER_CHANNELS = 2;
 const OPUS_FRAME_SAMPLES = 960;
 const OPUS_FRAME_DURATION_MS = 20;
-const RELAY_FRAME_SAMPLES = 480;
-const RELAY_FRAME_BYTES = RELAY_FRAME_SAMPLES * 2;
-const MAX_PENDING_RELAY_FRAMES = 250;
 const INBOUND_REORDER_DEPTH = 4;
 // More than two seconds behind cannot be useful 20 ms reordering; fail instead of corrupting Opus state.
 const INBOUND_MAX_LATE_PACKETS = 100;
@@ -167,7 +168,7 @@ export class OpenAIQuicksilverAudioPeer implements OpenAIQuicksilverAudioPeerCon
   private activeInboundSsrc: number | undefined;
   private inboundRtpState: InboundRtpState = { pendingPackets: new Map() };
   private mediaTimer: ReturnType<typeof setInterval> | undefined;
-  private pendingAudio = Buffer.alloc(0);
+  private pendingAudio: Buffer = Buffer.alloc(0);
   private sequenceNumber = randomInt(0x1_0000);
   private subscribedTracks = new Set<string>();
   private timestamp = randomInt(0x1_0000_0000);
@@ -222,16 +223,7 @@ export class OpenAIQuicksilverAudioPeer implements OpenAIQuicksilverAudioPeerCon
     if (this.closed || audio.length < 2) {
       return;
     }
-    const evenAudio = audio.subarray(0, audio.length - (audio.length % 2));
-    this.pendingAudio =
-      this.pendingAudio.length > 0
-        ? Buffer.concat([this.pendingAudio, evenAudio])
-        : Buffer.from(evenAudio);
-    const maxPendingBytes = RELAY_FRAME_BYTES * MAX_PENDING_RELAY_FRAMES;
-    if (this.pendingAudio.length > maxPendingBytes) {
-      // Keep the newest complete frames. Old microphone audio is less useful than bounded latency.
-      this.pendingAudio = this.pendingAudio.subarray(this.pendingAudio.length - maxPendingBytes);
-    }
+    this.pendingAudio = appendOpenAIQuicksilverPendingAudio(this.pendingAudio, audio);
   }
 
   close(): void {
@@ -440,8 +432,8 @@ export class OpenAIQuicksilverAudioPeer implements OpenAIQuicksilverAudioPeerCon
   private takeNextRelayFrame(): Buffer {
     // Relay ticks are framing boundaries: pad partial PCM now, or its tail survives
     // silence and is prepended to a later utterance as stale audio.
-    const frame = Buffer.alloc(RELAY_FRAME_BYTES);
-    const queuedBytes = Math.min(this.pendingAudio.length, RELAY_FRAME_BYTES);
+    const frame = Buffer.alloc(OPENAI_QUICKSILVER_RELAY_FRAME_BYTES);
+    const queuedBytes = Math.min(this.pendingAudio.length, OPENAI_QUICKSILVER_RELAY_FRAME_BYTES);
     if (queuedBytes > 0) {
       this.pendingAudio.copy(frame, 0, 0, queuedBytes);
       this.pendingAudio = this.pendingAudio.subarray(queuedBytes);