diff --git a/extensions/openai/realtime-audio-buffer-ownership.test.ts b/extensions/openai/realtime-audio-buffer-ownership.test.ts new file mode 100644 index 000000000000..dd26c050f524 --- /dev/null +++ b/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 { + const bytes = Buffer.isBuffer(data) + ? data + : Array.isArray(data) + ? Buffer.concat(data) + : Buffer.from(data); + return JSON.parse(bytes.toString("utf8")) as Record; +} + +async function withRealtimeProvider( + kind: RealtimeProviderKind, + prepareAudio: (bridge: RealtimeVoiceBridge) => void, +): Promise>> { + 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> = []; + 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((resolve, reject) => { + server.close((error) => (error ? reject(error) : resolve())); + }); + } +} + +describe("OpenAI realtime queued audio buffer ownership", () => { + it.each(["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(["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==" }, + ]); + }, + ); +}); diff --git a/extensions/openai/realtime-quicksilver-bridge.ts b/extensions/openai/realtime-quicksilver-bridge.ts index de8583cfcca1..6d43551f56df 100644 --- a/extensions/openai/realtime-quicksilver-bridge.ts +++ b/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 { diff --git a/extensions/openai/realtime-voice-provider.ts b/extensions/openai/realtime-voice-provider.ts index 9e8e448c65ab..907783437518 100644 --- a/extensions/openai/realtime-voice-provider.ts +++ b/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 {