diff --git a/extensions/openai/realtime-quicksilver-audio-buffer.test.ts b/extensions/openai/realtime-quicksilver-audio-buffer.test.ts new file mode 100644 index 000000000000..af92af2dc172 --- /dev/null +++ b/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); + }); +}); diff --git a/extensions/openai/realtime-quicksilver-audio-buffer.ts b/extensions/openai/realtime-quicksilver-audio-buffer.ts new file mode 100644 index 000000000000..5616cb7635aa --- /dev/null +++ b/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, + ); +} diff --git a/extensions/openai/realtime-quicksilver-gateway-bridge.test.ts b/extensions/openai/realtime-quicksilver-gateway-bridge.test.ts index 0c5e54d4d670..bcdf9afd2a7a 100644 --- a/extensions/openai/realtime-quicksilver-gateway-bridge.test.ts +++ b/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[0]["callbacks"] + | undefined; + let resolvePeer: ((peer: OpenAIQuicksilverAudioPeerContract) => void) | undefined; + const peerPromise = new Promise((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(); }); diff --git a/extensions/openai/realtime-quicksilver-gateway-bridge.ts b/extensions/openai/realtime-quicksilver-gateway-bridge.ts index 9e34ed4927b2..83422c071124 100644 --- a/extensions/openai/realtime-quicksilver-gateway-bridge.ts +++ b/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 | 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; diff --git a/extensions/openai/realtime-quicksilver-peer.runtime.ts b/extensions/openai/realtime-quicksilver-peer.runtime.ts index cd14cb012e3d..9eee52524c1a 100644 --- a/extensions/openai/realtime-quicksilver-peer.runtime.ts +++ b/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 | undefined; - private pendingAudio = Buffer.alloc(0); + private pendingAudio: Buffer = Buffer.alloc(0); private sequenceNumber = randomInt(0x1_0000); private subscribedTracks = new Set(); 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);