fix(openai): copy queued realtime audio buffers (#117028)

* commit '7facf157e67d7902dd645f33026a253e006ed0e0':
  test(openai): parse realtime websocket frames safely
  fix(openai): own queued realtime audio buffer snapshots

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

@@ -0,0 +1,132 @@
+import { once } from "node:events";
+import type { RealtimeVoiceBridge } from "openclaw/plugin-sdk/realtime-voice";
+import { describe, expect, it, vi } from "vitest";
+import WebSocket, { type RawData, WebSocketServer } from "ws";
+import { OpenAIQuicksilverVoiceBridge } from "./realtime-quicksilver-bridge.js";
+import { buildOpenAIRealtimeVoiceProvider } from "./realtime-voice-provider.js";
+
+type RealtimeProviderKind = "native" | "gpt-live";
+
+function parseWebSocketMessage(data: RawData): Record<string, unknown> {
+  const bytes = Buffer.isBuffer(data)
+    ? data
+    : Array.isArray(data)
+      ? Buffer.concat(data)
+      : Buffer.from(data);
+  return JSON.parse(bytes.toString("utf8")) as Record<string, unknown>;
+}
+
+async function withRealtimeProvider(
+  kind: RealtimeProviderKind,
+  prepareAudio: (bridge: RealtimeVoiceBridge) => void,
+): Promise<Array<Record<string, unknown>>> {
+  const audioEventType = kind === "native" ? "input_audio_buffer.append" : "input_audio.append";
+  const server = new WebSocketServer({ host: "127.0.0.1", port: 0 });
+  await once(server, "listening");
+  const address = server.address();
+  if (!address || typeof address === "string") {
+    throw new Error("expected an available local realtime WebSocket address");
+  }
+  const received: Array<Record<string, unknown>> = [];
+  server.once("connection", (socket) => {
+    socket.on("message", (payload) => {
+      const event = parseWebSocketMessage(payload);
+      received.push(event);
+      if (event.type === "session.update") {
+        socket.send(
+          JSON.stringify(
+            kind === "native"
+              ? { type: "session.updated" }
+              : {
+                  type: "session.started",
+                  session: { id: "fixture-live", expires_at: Math.floor(Date.now() / 1000) + 60 },
+                },
+          ),
+        );
+      }
+    });
+  });
+
+  const endpoint = `http://127.0.0.1:${address.port}`;
+  const bridge =
+    kind === "native"
+      ? buildOpenAIRealtimeVoiceProvider().createBridge({
+          providerConfig: {
+            apiKey: "fixture-local", // pragma: allowlist secret
+            azureEndpoint: endpoint,
+            azureDeployment: "fixture-realtime",
+          },
+          onAudio: vi.fn(),
+          onClearAudio: vi.fn(),
+        })
+      : new OpenAIQuicksilverVoiceBridge({
+          providerConfig: {},
+          model: "gpt-live-1-codex",
+          audioFormat: { encoding: "pcm16", sampleRateHz: 24000, channels: 1 },
+          resolveAuth: async () => ({ type: "api-key", token: "fixture-local" }),
+          webSocketFactory: (_url, options) => new WebSocket(endpoint, options),
+          onAudio: vi.fn(),
+          onClearAudio: vi.fn(),
+        });
+
+  try {
+    prepareAudio(bridge);
+    await bridge.connect();
+    await vi.waitFor(() => {
+      expect(received.some((event) => event.type === audioEventType)).toBe(true);
+    });
+    return received;
+  } finally {
+    bridge.close();
+    for (const client of server.clients) {
+      client.terminate();
+    }
+    await new Promise<void>((resolve, reject) => {
+      server.close((error) => (error ? reject(error) : resolve()));
+    });
+  }
+}
+
+describe("OpenAI realtime queued audio buffer ownership", () => {
+  it.each<RealtimeProviderKind>(["native", "gpt-live"])(
+    "%s preserves each reusable producer frame until the real WebSocket is ready",
+    async (kind) => {
+      const audioEventType = kind === "native" ? "input_audio_buffer.append" : "input_audio.append";
+      const received = await withRealtimeProvider(kind, (bridge) => {
+        const producerAllocation = Buffer.alloc(2 * 1024 * 1024, 0x7f);
+        const producerView = producerAllocation.subarray(0, 1);
+        bridge.sendAudio(producerView);
+        producerAllocation[0] = 0x41;
+        bridge.sendAudio(producerView);
+        producerAllocation[0] = 0;
+      });
+
+      expect(received.filter((event) => event.type === audioEventType)).toEqual([
+        { type: audioEventType, audio: "fw==" },
+        { type: audioEventType, audio: "QQ==" },
+      ]);
+    },
+  );
+
+  it.each<RealtimeProviderKind>(["native", "gpt-live"])(
+    "%s rejects oversized producer frames before allocating a queued copy",
+    async (kind) => {
+      const audioEventType = kind === "native" ? "input_audio_buffer.append" : "input_audio.append";
+      const received = await withRealtimeProvider(kind, (bridge) => {
+        const oversized = Buffer.alloc(1024 * 1024 + 1);
+        const copyBuffer = vi.spyOn(Buffer, "from");
+        try {
+          bridge.sendAudio(oversized);
+          expect(copyBuffer).not.toHaveBeenCalled();
+        } finally {
+          copyBuffer.mockRestore();
+        }
+        bridge.sendAudio(Buffer.from([0x7f]));
+      });
+
+      expect(received.filter((event) => event.type === audioEventType)).toEqual([
+        { type: audioEventType, audio: "fw==" },
+      ]);
+    },
+  );
+});

extensions/openai/realtime-quicksilver-bridge.ts

@@ -568,8 +568,10 @@ export class OpenAIQuicksilverVoiceBridge implements RealtimeVoiceBridge {
     ) {
       return;
     }
-    this.pendingAudio.push(audio);
-    this.pendingAudioBytes += audio.byteLength;
+    // Capture transports can recycle caller-owned views before the provider becomes ready.
+    const queuedAudio = Buffer.from(audio);
+    this.pendingAudio.push(queuedAudio);
+    this.pendingAudioBytes += queuedAudio.byteLength;
   }
 
   private resetTerminalState(): void {

extensions/openai/realtime-voice-provider.ts

@@ -1684,8 +1684,10 @@ class OpenAIRealtimeVoiceBridge implements RealtimeVoiceBridge {
     ) {
       return;
     }
-    this.pendingAudio.push(audio);
-    this.pendingAudioBytes += audio.byteLength;
+    // Capture transports can recycle caller-owned views before the provider becomes ready.
+    const queuedAudio = Buffer.from(audio);
+    this.pendingAudio.push(queuedAudio);
+    this.pendingAudioBytes += queuedAudio.byteLength;
   }
 
   private clearPendingAudio(): void {