diff --git a/extensions/openai/realtime-quicksilver-gateway-bridge.test.ts b/extensions/openai/realtime-quicksilver-gateway-bridge.test.ts index 839363f3a345..36ec9f4f7d56 100644 --- a/extensions/openai/realtime-quicksilver-gateway-bridge.test.ts +++ b/extensions/openai/realtime-quicksilver-gateway-bridge.test.ts @@ -4,6 +4,7 @@ import { OPENAI_QUICKSILVER_RELAY_FRAME_BYTES } from "./realtime-quicksilver-aud import { OpenAIQuicksilverGatewayBridge } from "./realtime-quicksilver-gateway-bridge.js"; import { OpenAIQuicksilverAudioPeer, + type OpenAIQuicksilverAudioPeerCallbacks, type OpenAIQuicksilverAudioPeerContract, } from "./realtime-quicksilver-peer.runtime.js"; import { @@ -546,9 +547,13 @@ describe("GPT-Live werift audio peer", () => { }); describe("GPT-Live gateway relay bridge", () => { - function createPendingPeerBridge() { + function createPendingPeerBridge(params?: { + onClose?: (reason: "completed" | "error") => void; + onError?: (error: Error) => void; + }) { let resolvePeer: ((peer: OpenAIQuicksilverAudioPeerContract) => void) | undefined; let rejectPeer: ((error: Error) => void) | undefined; + let peerCallbacks: OpenAIQuicksilverAudioPeerCallbacks | undefined; const peerPromise = new Promise((resolve, reject) => { resolvePeer = resolve; rejectPeer = reject; @@ -567,7 +572,8 @@ describe("GPT-Live gateway relay bridge", () => { audioFormat: { encoding: "pcm16", sampleRateHz: 24_000, channels: 1 }, onAudio: vi.fn(), onClearAudio: vi.fn(), - onClose, + onClose: params?.onClose ?? onClose, + onError: params?.onError, runAgentConsult: vi.fn(async () => ({ text: "done" })), logger: { debug: vi.fn(), warn: vi.fn() }, resolveAuth: vi.fn(async () => ({ @@ -575,7 +581,10 @@ describe("GPT-Live gateway relay bridge", () => { token: "oauth-token", accountId: "account-1", })), - createPeer: vi.fn(() => peerPromise), + createPeer: vi.fn((callbacks) => { + peerCallbacks = callbacks; + return peerPromise; + }), fetchImpl: vi.fn(async () => createCallResponse("v=answer\r\n", "rtc_pending_audio")), webSocketFactory: () => new FakeSocket(), }); @@ -587,6 +596,7 @@ describe("GPT-Live gateway relay bridge", () => { peer, rejectPeer: (error: Error) => rejectPeer?.(error), resolvePeer: () => resolvePeer?.(peer), + triggerPeerError: (error: Error) => peerCallbacks?.onError(error), }; } @@ -644,6 +654,53 @@ describe("GPT-Live gateway relay bridge", () => { expect(peer.sendAudio).not.toHaveBeenCalled(); }); + it("keeps error precedence when onError reentrantly closes the bridge", async () => { + const onClose = vi.fn(); + const bridgeRef: { current?: OpenAIQuicksilverGatewayBridge } = {}; + const harness = createPendingPeerBridge({ + onClose, + onError: () => bridgeRef.current?.close(), + }); + bridgeRef.current = harness.bridge; + const connectionRejected = expect(harness.connection).rejects.toThrow( + "GPT-Live gateway relay bridge closed", + ); + + harness.triggerPeerError(new Error("media peer failed")); + + expect(onClose).toHaveBeenCalledOnce(); + expect(onClose).toHaveBeenCalledWith("error"); + await connectionRejected; + }); + + it("releases queued audio and rejects a late peer when onError throws", async () => { + const callbackError = new Error("error callback failed"); + const onClose = vi.fn(); + const harness = createPendingPeerBridge({ + onClose, + onError: () => { + throw callbackError; + }, + }); + const testBridge = harness.bridge as unknown as TestableGatewayBridge; + harness.bridge.sendAudio(Buffer.from([0x41, 0x42])); + const connectionRejected = expect(harness.connection).rejects.toThrow( + "GPT-Live gateway relay bridge closed", + ); + + expect(() => harness.triggerPeerError(new Error("media peer failed"))).toThrow(callbackError); + const retainedAudioBytes = testBridge.pendingAudio.byteLength; + const closeReason = onClose.mock.calls[0]?.[0]; + harness.bridge.close(); + harness.resolvePeer(); + + await connectionRejected; + await vi.waitFor(() => expect(harness.peer.close).toHaveBeenCalledOnce()); + expect(retainedAudioBytes).toBe(0); + expect(closeReason).toBe("error"); + expect(harness.peer.sendAudio).not.toHaveBeenCalled(); + }); + it("closes a sideband that opens in the abort handoff", async () => { const controller = new AbortController(); const socket = new FakeSocket("manual"); diff --git a/extensions/openai/realtime-quicksilver-gateway-bridge.ts b/extensions/openai/realtime-quicksilver-gateway-bridge.ts index 83422c071124..f4dfc4f120d3 100644 --- a/extensions/openai/realtime-quicksilver-gateway-bridge.ts +++ b/extensions/openai/realtime-quicksilver-gateway-bridge.ts @@ -524,22 +524,24 @@ export class OpenAIQuicksilverGatewayBridge implements RealtimeVoiceBridge { } private fail(error: Error): void { - if (this.closed) { - return; - } - this.config.onError?.(error); - this.teardown("error"); + this.teardown("error", () => this.config.onError?.(error)); } - private teardown(reason: "completed" | "error"): void { + private teardown(reason: "completed" | "error", beforeClose?: () => void): void { if (this.closed) { return; } + // Claim terminal ownership and release resources before callbacks so reentrant close + // cannot replace the outcome, while finally preserves error-before-close ordering. this.closed = true; this.releaseResources(); - if (!this.closeNotified) { - this.closeNotified = true; - this.config.onClose?.(reason); + try { + beforeClose?.(); + } finally { + if (!this.closeNotified) { + this.closeNotified = true; + this.config.onClose?.(reason); + } } }