From b7b37847c8676f925a9cef28186aac74da780227 Mon Sep 17 00:00:00 2001 From: Vincent Koc Date: Sat, 1 Aug 2026 15:25:50 +0800 Subject: [PATCH] fix(talk): make transport startup transactional --- .../chat/realtime-talk-gateway-relay.test.ts | 255 +++++++++++++++--- .../pages/chat/realtime-talk-gateway-relay.ts | 248 ++++++++++++----- .../chat/realtime-talk-google-live.test.ts | 103 ++++++- .../pages/chat/realtime-talk-google-live.ts | 46 ++-- .../chat/realtime-talk-lifecycle.test.ts | 150 ++++++++++- ui/src/pages/chat/realtime-talk-shared.ts | 5 +- .../chat/realtime-talk-transcript-owner.ts | 20 ++ .../realtime-talk-transcript-queue.test.ts | 2 +- .../pages/chat/realtime-talk-webrtc.test.ts | 77 +++++- ui/src/pages/chat/realtime-talk-webrtc.ts | 73 +++-- ui/src/pages/chat/realtime-talk.test.ts | 6 +- ui/src/pages/chat/realtime-talk.ts | 70 ++++- 12 files changed, 901 insertions(+), 154 deletions(-) diff --git a/ui/src/pages/chat/realtime-talk-gateway-relay.test.ts b/ui/src/pages/chat/realtime-talk-gateway-relay.test.ts index 80f41d8833a1..a70ac4cad21c 100644 --- a/ui/src/pages/chat/realtime-talk-gateway-relay.test.ts +++ b/ui/src/pages/chat/realtime-talk-gateway-relay.test.ts @@ -129,6 +129,11 @@ function createTransport(overrides: Partial = {}) }); } +async function startTransport(transport: GatewayRelayRealtimeTalkTransport): Promise { + await expect(transport.start()).resolves.toBe("ready"); + transport.activate(); +} + function emitGatewayFrame(frame: GatewayFrame): void { for (const listener of listeners) { listener(frame); @@ -198,7 +203,7 @@ describe("GatewayRelayRealtimeTalkTransport", () => { inputDeviceId: "usb-mic", }); - await transport.start(); + await startTransport(transport); expect(getUserMedia).toHaveBeenCalledWith({ audio: { @@ -218,7 +223,7 @@ describe("GatewayRelayRealtimeTalkTransport", () => { sessionKey: "main", }); - await transport.start(); + await startTransport(transport); const processor = processors.at(-1); const sink = inputSinks.at(-1); @@ -250,13 +255,162 @@ describe("GatewayRelayRealtimeTalkTransport", () => { const start = transport.start(); transport.stop(); resolveMedia({ getTracks: () => [{ stop: stopTrack }] } as unknown as MediaStream); - await start; + await expect(start).resolves.toBe("cancelled"); expect(stopTrack).toHaveBeenCalledOnce(); expect(processors).toHaveLength(0); expect(onInputLevel).not.toHaveBeenCalled(); }); + it("defers relay effects until the transport is committed", async () => { + let resolveMedia: (media: MediaStream) => void = () => undefined; + getUserMedia.mockReturnValue( + new Promise((resolve) => { + resolveMedia = resolve; + }), + ); + const client = createClient(); + const onStatus = vi.fn(); + const onTranscript = vi.fn(); + const transport = createTransport({ + client, + callbacks: { onStatus, onTranscript }, + }); + + const start = transport.start(); + emitTalkEvent({ relaySessionId: "relay-1", type: "ready" }); + emitTalkEvent({ + relaySessionId: "relay-1", + type: "transcript", + role: "assistant", + text: "not committed yet", + final: true, + }); + emitTalkEvent({ + relaySessionId: "relay-1", + type: "toolCall", + callId: "call-1", + name: "unknown_tool", + args: {}, + }); + + expect(onStatus).not.toHaveBeenCalled(); + expect(onTranscript).not.toHaveBeenCalled(); + expect(requestCallsFor(client, "talk.session.submitToolResult")).toHaveLength(0); + + resolveMedia({ getTracks: () => [{ stop: vi.fn() }] } as unknown as MediaStream); + await expect(start).resolves.toBe("ready"); + expect(onStatus).not.toHaveBeenCalled(); + expect(onTranscript).not.toHaveBeenCalled(); + + transport.activate(); + + expect(onStatus).toHaveBeenCalledWith("listening"); + expect(onTranscript).toHaveBeenCalledWith({ + role: "assistant", + text: "not committed yet", + final: true, + }); + await waitForFast(() => + expect(requestCallsFor(client, "talk.session.submitToolResult")).toHaveLength(1), + ); + transport.stop(); + }); + + it("fails a provisional relay whose bounded event buffer overflows", async () => { + let resolveMedia: (media: MediaStream) => void = () => undefined; + getUserMedia.mockReturnValue( + new Promise((resolve) => { + resolveMedia = resolve; + }), + ); + const onStatus = vi.fn(); + const client = createClient(); + const transport = createTransport({ callbacks: { onStatus }, client }); + + const start = transport.start(); + for (let index = 0; index < 33; index += 1) { + emitTalkEvent({ relaySessionId: "relay-1", type: "ready" }); + } + expect(requestCallsFor(client, "talk.session.close")).toHaveLength(1); + resolveMedia({ getTracks: () => [{ stop: vi.fn() }] } as unknown as MediaStream); + + await expect(start).rejects.toThrow( + "Realtime relay emitted too much data before browser setup completed", + ); + expect(onStatus).not.toHaveBeenCalled(); + }); + + it("enforces the provisional relay byte bound", async () => { + let resolveMedia: (media: MediaStream) => void = () => undefined; + getUserMedia.mockReturnValue( + new Promise((resolve) => { + resolveMedia = resolve; + }), + ); + const client = createClient(); + const transport = createTransport({ client }); + + const start = transport.start(); + emitTalkEvent({ + relaySessionId: "relay-1", + type: "transcript", + role: "assistant", + text: "x".repeat(131_073), + final: true, + }); + + expect(requestCallsFor(client, "talk.session.close")).toHaveLength(1); + resolveMedia({ getTracks: () => [{ stop: vi.fn() }] } as unknown as MediaStream); + await expect(start).rejects.toThrow( + "Realtime relay emitted too much data before browser setup completed", + ); + }); + + it("rejects a relay that closes before browser setup commits", async () => { + let resolveMedia: (media: MediaStream) => void = () => undefined; + getUserMedia.mockReturnValue( + new Promise((resolve) => { + resolveMedia = resolve; + }), + ); + const client = createClient(); + const onStatus = vi.fn(); + const transport = createTransport({ callbacks: { onStatus }, client }); + + const start = transport.start(); + emitTalkEvent({ + relaySessionId: "relay-1", + type: "error", + message: "provider rejected setup", + }); + emitTalkEvent({ + relaySessionId: "relay-1", + type: "close", + reason: "error", + }); + resolveMedia({ getTracks: () => [{ stop: vi.fn() }] } as unknown as MediaStream); + + await expect(start).rejects.toThrow("provider rejected setup"); + expect(onStatus).not.toHaveBeenCalled(); + expect(requestCallsFor(client, "talk.session.close")).toHaveLength(0); + }); + + it("closes the relay when a provisional callback throws during activation", async () => { + const client = createClient(); + const onStatus = vi.fn(() => { + throw new Error("consumer failed"); + }); + const transport = createTransport({ callbacks: { onStatus }, client }); + + const start = transport.start(); + emitTalkEvent({ relaySessionId: "relay-1", type: "ready" }); + await expect(start).resolves.toBe("ready"); + + expect(() => transport.activate()).toThrow("consumer failed"); + expect(requestCallsFor(client, "talk.session.close")).toHaveLength(1); + }); + it("forwards common Talk events from Gateway relay frames", async () => { const onTalkEvent = vi.fn(); const transport = new GatewayRelayRealtimeTalkTransport(createSession(), { @@ -276,7 +430,7 @@ describe("GatewayRelayRealtimeTalkTransport", () => { payload: {}, } satisfies RealtimeTalkEvent; - await transport.start(); + await startTransport(transport); emitTalkEvent({ relaySessionId: "relay-1", type: "ready", @@ -291,7 +445,7 @@ describe("GatewayRelayRealtimeTalkTransport", () => { const onTalkEvent = vi.fn(); const transport = createTransport({ callbacks: { onTalkEvent } }); - await transport.start(); + await startTransport(transport); emitTalkEvent({ relaySessionId: "relay-other", type: "ready", @@ -316,7 +470,7 @@ describe("GatewayRelayRealtimeTalkTransport", () => { const client = createClient(); const transport = createTransport({ client }); - await transport.start(); + await startTransport(transport); emitTalkEvent({ relaySessionId: "relay-1", type: "audio", @@ -336,7 +490,7 @@ describe("GatewayRelayRealtimeTalkTransport", () => { const client = createClient(); const transport = createTransport({ client }); - await transport.start(); + await startTransport(transport); for (let index = 0; index < 321; index += 1) { emitTalkEvent({ relaySessionId: "relay-1", @@ -382,7 +536,7 @@ describe("GatewayRelayRealtimeTalkTransport", () => { const client = createClient(); const transport = createTransport({ client }); - await transport.start(); + await startTransport(transport); emitTalkEvent({ relaySessionId: "relay-1", type: "audio", @@ -410,7 +564,7 @@ describe("GatewayRelayRealtimeTalkTransport", () => { const client = createClient(); const transport = createTransport({ client }); - await transport.start(); + await startTransport(transport); emitTalkEvent({ relaySessionId: "relay-1", type: "audio", @@ -433,7 +587,7 @@ describe("GatewayRelayRealtimeTalkTransport", () => { const client = createClient(); const transport = createTransport({ client }); - await transport.start(); + await startTransport(transport); emitTalkEvent({ relaySessionId: "relay-1", type: "audio", @@ -451,7 +605,7 @@ describe("GatewayRelayRealtimeTalkTransport", () => { const onInputLevel = vi.fn(); const transport = createTransport({ callbacks: { onInputLevel } }); - await transport.start(); + await startTransport(transport); pumpMicrophone(new Float32Array(4096)); pumpMicrophone(new Float32Array(4096).fill(0.25)); transport.stop(); @@ -490,7 +644,7 @@ describe("GatewayRelayRealtimeTalkTransport", () => { }); const transport = createTransport({ callbacks: { onStatus }, client }); - await transport.start(); + await startTransport(transport); const samples = new Float32Array(4096); for (let index = 0; index < 10_000; index += 1) { pumpMicrophone(samples); @@ -521,7 +675,7 @@ describe("GatewayRelayRealtimeTalkTransport", () => { const client = createClient(); const transport = createTransport({ client }); - await transport.start(); + await startTransport(transport); for (const timestamp of [10, 20, 30, 40]) { audioCurrentTime = timestamp / 1_000; pumpMicrophone(new Float32Array(4096)); @@ -581,7 +735,7 @@ describe("GatewayRelayRealtimeTalkTransport", () => { }); const transport = createTransport({ callbacks: { onStatus }, client }); - await transport.start(); + await startTransport(transport); pumpMicrophone(new Float32Array(4096)); await waitForFast(() => expect(onStatus).toHaveBeenCalledWith("error", "Unknown realtime relay session"), @@ -605,7 +759,7 @@ describe("GatewayRelayRealtimeTalkTransport", () => { const client = createClient(); const transport = createTransport({ callbacks: { onStatus }, client }); - await transport.start(); + await startTransport(transport); pumpMicrophone(new Float32Array(4096)); emitTalkEvent({ relaySessionId: "relay-1", @@ -626,12 +780,53 @@ describe("GatewayRelayRealtimeTalkTransport", () => { expect(closeCalls).toHaveLength(0); }); + it.each(["talk event", "status"] as const)( + "releases local resources when a close %s callback throws", + async (callbackKind) => { + const stopTrack = vi.fn(); + getUserMedia.mockResolvedValue({ + getTracks: () => [{ stop: stopTrack }], + } as unknown as MediaStream); + const throwingCallback = vi.fn(() => { + throw new Error("consumer failed"); + }); + const client = createClient(); + const transport = createTransport({ + client, + callbacks: + callbackKind === "talk event" + ? { onTalkEvent: throwingCallback } + : { onStatus: throwingCallback }, + }); + + await startTransport(transport); + expect(() => + emitTalkEvent({ + relaySessionId: "relay-1", + type: "close", + reason: "error", + talkEvent: + callbackKind === "talk event" + ? ({ type: "session.closed", final: true } satisfies RealtimeTalkEvent) + : undefined, + }), + ).toThrow("consumer failed"); + + expect(stopTrack).toHaveBeenCalledOnce(); + expect(processors.at(-1)?.disconnect).toHaveBeenCalledOnce(); + expect(listeners.size).toBe(0); + emitTalkEvent({ relaySessionId: "relay-1", type: "ready" }); + transport.stop(); + expect(requestCallsFor(client, "talk.session.close")).toHaveLength(0); + }, + ); + it("preserves relay error details across close events", async () => { const onStatus = vi.fn(); const client = createClient(); const transport = createTransport({ callbacks: { onStatus }, client }); - await transport.start(); + await startTransport(transport); emitTalkEvent({ relaySessionId: "relay-1", type: "error", @@ -652,7 +847,7 @@ describe("GatewayRelayRealtimeTalkTransport", () => { const transport = createTransport({ client }); const speech = new Float32Array(4096).fill(0.25); - await transport.start(); + await startTransport(transport); emitTalkEvent({ relaySessionId: "relay-1", type: "audio", @@ -690,7 +885,7 @@ describe("GatewayRelayRealtimeTalkTransport", () => { }); const transport = createTransport({ callbacks: { onStatus }, client }); - await transport.start(); + await startTransport(transport); emitTalkEvent({ relaySessionId: "relay-1", type: "toolCall", @@ -745,7 +940,7 @@ describe("GatewayRelayRealtimeTalkTransport", () => { }); const transport = createTransport({ callbacks: { onStatus }, client }); - await transport.start(); + await startTransport(transport); emitTalkEvent({ relaySessionId: "relay-1", type: "toolCall", @@ -794,7 +989,7 @@ describe("GatewayRelayRealtimeTalkTransport", () => { }); const transport = createTransport({ callbacks: { onStatus }, client }); - await transport.start(); + await startTransport(transport); emitTalkEvent({ relaySessionId: "relay-1", type: "toolCall", @@ -833,7 +1028,7 @@ describe("GatewayRelayRealtimeTalkTransport", () => { }); const transport = createTransport({ client }); - await transport.start(); + await startTransport(transport); emitTalkEvent({ relaySessionId: "relay-1", type: "toolCall", @@ -870,7 +1065,7 @@ describe("GatewayRelayRealtimeTalkTransport", () => { }); const transport = createTransport({ client }); - await transport.start(); + await startTransport(transport); emitTalkEvent({ relaySessionId: "relay-1", type: "audio", @@ -915,7 +1110,7 @@ describe("GatewayRelayRealtimeTalkTransport", () => { }); const transport = createTransport({ client }); - await transport.start(); + await startTransport(transport); emitTalkEvent({ relaySessionId: "relay-1", type: "audio", @@ -967,7 +1162,7 @@ describe("GatewayRelayRealtimeTalkTransport", () => { }); const transport = createTransport({ client }); - await transport.start(); + await startTransport(transport); emitTalkEvent({ relaySessionId: "relay-1", type: "toolCall", @@ -1003,7 +1198,7 @@ describe("GatewayRelayRealtimeTalkTransport", () => { const transport = createTransport({ client }); const speech = new Float32Array(4096).fill(0.25); - await transport.start(); + await startTransport(transport); emitTalkEvent({ relaySessionId: "relay-1", type: "audio", @@ -1096,7 +1291,7 @@ describe("GatewayRelayRealtimeTalkTransport", () => { const transport = createTransport({ callbacks: { onStatus }, client }); const speech = new Float32Array(4096).fill(0.25); - await transport.start(); + await startTransport(transport); emitTalkEvent({ relaySessionId: "relay-1", type: "audio", @@ -1125,7 +1320,7 @@ describe("GatewayRelayRealtimeTalkTransport", () => { expect(requestCallsFor(client, "talk.session.submitToolResult")).toHaveLength(0); expect(requestCallsFor(client, "talk.session.close")).toEqual([ - ["talk.session.close", { sessionId: "relay-1" }], + ["talk.session.close", { sessionId: "relay-1" }, { timeoutMs: 8_000 }], ]); expect(onStatus).toHaveBeenCalledWith("error", "cancel failed"); }); @@ -1140,7 +1335,7 @@ describe("GatewayRelayRealtimeTalkTransport", () => { }); const transport = createTransport({ client }); - await transport.start(); + await startTransport(transport); emitTalkEvent({ relaySessionId: "relay-1", type: "toolCall", @@ -1204,7 +1399,7 @@ describe("GatewayRelayRealtimeTalkTransport", () => { }); const transport = createTransport({ client }); - await transport.start(); + await startTransport(transport); emitTalkEvent({ relaySessionId: "relay-1", type: "toolCall", @@ -1251,7 +1446,7 @@ describe("GatewayRelayRealtimeTalkTransport", () => { }); const transport = createTransport({ client }); - await transport.start(); + await startTransport(transport); emitTalkEvent({ relaySessionId: "relay-1", type: "toolCall", diff --git a/ui/src/pages/chat/realtime-talk-gateway-relay.ts b/ui/src/pages/chat/realtime-talk-gateway-relay.ts index b9b71775bc9c..e25a06b8281f 100644 --- a/ui/src/pages/chat/realtime-talk-gateway-relay.ts +++ b/ui/src/pages/chat/realtime-talk-gateway-relay.ts @@ -17,6 +17,7 @@ import { type RealtimeTalkGatewayRelaySessionResult, type RealtimeTalkTransport, type RealtimeTalkTransportContext, + type RealtimeTalkTransportStartResult, } from "./realtime-talk-shared.ts"; const BARGE_IN_RMS_THRESHOLD = 0.02; @@ -24,6 +25,20 @@ const BARGE_IN_PEAK_THRESHOLD = 0.08; const BARGE_IN_CONSECUTIVE_SPEECH_FRAMES = 2; const MAX_PENDING_AUDIO_APPENDS = 4; const AUDIO_APPEND_TIMEOUT_MS = 8_000; +const RELAY_CLOSE_TIMEOUT_MS = 8_000; +const MAX_PENDING_ACTIVATION_EVENTS = 32; +const MAX_PENDING_ACTIVATION_EVENT_BYTES = 256 * 1024; + +function estimateRelayEventBytes(event: GatewayRelayEvent): number { + try { + const serialized = JSON.stringify(event); + // Browser strings may use two bytes per code unit; use the upper bound without + // allocating a second encoded copy on this realtime event path. + return (serialized?.length ?? 0) * 2; + } catch { + return MAX_PENDING_ACTIVATION_EVENT_BYTES + 1; + } +} export class GatewayRelayRealtimeTalkTransport implements RealtimeTalkTransport { private media: MediaStream | null = null; @@ -33,6 +48,7 @@ export class GatewayRelayRealtimeTalkTransport implements RealtimeTalkTransport private readonly inputPump = new RealtimeTalkPcmInputPump(); private unsubscribe: (() => void) | null = null; private closed = false; + private mediaSetupController: AbortController | null = null; private audioAppendAbortController: AbortController | null = null; private readonly pendingAudioAppends = new Set>(); private readonly outputQueue = new RealtimeTalkPcmOutputQueue(); @@ -46,13 +62,17 @@ export class GatewayRelayRealtimeTalkTransport implements RealtimeTalkTransport private pendingOutputCancellations = 0; private speechFramesDuringPlayback = 0; private lastRelayError: string | undefined; + private activated = false; + private pendingActivationEvents: GatewayRelayEvent[] = []; + private pendingActivationEventBytes = 0; + private startupError: Error | null = null; constructor( private readonly session: RealtimeTalkGatewayRelaySessionResult, private readonly ctx: RealtimeTalkTransportContext, ) {} - async start(): Promise { + async start(): Promise { if (!navigator.mediaDevices?.getUserMedia) { throw new Error("Realtime Talk requires browser microphone access"); } @@ -63,24 +83,46 @@ export class GatewayRelayRealtimeTalkTransport implements RealtimeTalkTransport throw new Error("Gateway-relay realtime Talk currently requires PCM16 audio"); } this.closed = false; + this.activated = false; + this.pendingActivationEvents = []; + this.pendingActivationEventBytes = 0; + this.startupError = null; + this.mediaSetupController?.abort(); + const mediaSetupController = new AbortController(); + this.mediaSetupController = mediaSetupController; this.unsubscribe = this.ctx.client.addEventListener((evt) => { if (evt.event !== "talk.event") { return; } - this.handleRelayEvent(evt.payload as GatewayRelayEvent); + this.handleIncomingRelayEvent(evt.payload as GatewayRelayEvent); }); let media: MediaStream; try { - media = await openRealtimeTalkInput(this.ctx.inputDeviceId); + media = await openRealtimeTalkInput(this.ctx.inputDeviceId, { + signal: mediaSetupController.signal, + }); } catch (error) { + const startupError = this.currentStartupError(); + if (startupError) { + throw startupError; + } if (this.closed) { - return; + return "cancelled"; } throw error; + } finally { + if (this.mediaSetupController === mediaSetupController) { + this.mediaSetupController = null; + } + } + const startupError = this.currentStartupError(); + if (startupError) { + media.getTracks().forEach((track) => track.stop()); + throw startupError; } if (this.closed) { media.getTracks().forEach((track) => track.stop()); - return; + return "cancelled"; } this.media = media; this.inputContext = new AudioContext({ sampleRate: this.session.audio.inputSampleRateHz }); @@ -92,6 +134,28 @@ export class GatewayRelayRealtimeTalkTransport implements RealtimeTalkTransport this.inputMeter.start(this.media, this.inputContext); } this.startMicrophonePump(); + return "ready"; + } + + activate(): void { + if (this.closed || this.activated) { + return; + } + this.activated = true; + const events = this.pendingActivationEvents; + this.pendingActivationEvents = []; + this.pendingActivationEventBytes = 0; + for (const event of events) { + try { + this.handleRelayEvent(event); + } catch (error) { + this.stop(); + throw error; + } + if (this.closed) { + return; + } + } } stop(): void { @@ -99,15 +163,24 @@ export class GatewayRelayRealtimeTalkTransport implements RealtimeTalkTransport this.stopLocal(); if (!wasClosed) { void this.ctx.client - .request("talk.session.close", { - sessionId: this.session.relaySessionId, - }) + .request( + "talk.session.close", + { + sessionId: this.session.relaySessionId, + }, + { timeoutMs: RELAY_CLOSE_TIMEOUT_MS }, + ) .catch(() => undefined); } } private stopLocal(): void { this.closed = true; + this.mediaSetupController?.abort(); + this.mediaSetupController = null; + this.activated = false; + this.pendingActivationEvents = []; + this.pendingActivationEventBytes = 0; this.unsubscribe?.(); this.unsubscribe = null; this.inputPump.stop(); @@ -186,70 +259,119 @@ export class GatewayRelayRealtimeTalkTransport implements RealtimeTalkTransport this.pendingAudioAppends.clear(); } + private currentStartupError(): Error | null { + return this.startupError; + } + + private handleIncomingRelayEvent(event: GatewayRelayEvent): void { + if (event.relaySessionId !== this.session.relaySessionId || this.closed) { + return; + } + if (this.activated) { + this.handleRelayEvent(event); + return; + } + if (event.type === "error") { + this.lastRelayError = event.message ?? "Realtime relay failed"; + } + if (event.type === "close") { + this.startupError = new Error( + event.reason === "error" + ? (this.lastRelayError ?? "Realtime relay closed before browser setup completed") + : "Realtime relay closed before browser setup completed", + ); + // The server already declared this relay terminal; local cleanup must not + // replay its callbacks or send a redundant close request. + this.stopLocal(); + return; + } + const eventBytes = estimateRelayEventBytes(event); + if ( + this.pendingActivationEvents.length >= MAX_PENDING_ACTIVATION_EVENTS || + eventBytes > MAX_PENDING_ACTIVATION_EVENT_BYTES - this.pendingActivationEventBytes + ) { + // The relay starts before browser media permission settles. Keep that provisional + // window bounded and fail the candidate instead of dropping authoritative events. + this.startupError = new Error( + "Realtime relay emitted too much data before browser setup completed", + ); + // Overflow is locally terminal, so release the server relay immediately even + // if the browser's microphone permission prompt never settles. + this.stop(); + return; + } + this.pendingActivationEvents.push(event); + this.pendingActivationEventBytes += eventBytes; + } + private handleRelayEvent(event: GatewayRelayEvent): void { if (event.relaySessionId !== this.session.relaySessionId || this.closed) { return; } - if (event.talkEvent) { - this.ctx.callbacks.onTalkEvent?.(event.talkEvent); - } - switch (event.type) { - case "ready": - this.ctx.callbacks.onStatus?.("listening"); - return; - case "audio": - if (event.audioBase64 && !this.playbackOverflowed) { - this.cancelRequestedForPlayback = false; - this.speechFramesDuringPlayback = 0; - this.playPcm16(event.audioBase64); - } - return; - case "clear": - this.playbackOverflowed = false; - this.stopOutput({ releaseDelayedToolResults: this.pendingOutputCancellations === 0 }); - if (event.talkEvent?.type === "turn.cancelled") { - this.abortConsults(); - } - return; - case "mark": - if (event.markName) { - this.scheduleMarkAck(event.markName); - } - return; - case "transcript": - if (event.role && event.text) { - this.ctx.callbacks.onTranscript?.({ - role: event.role, - text: event.text, - final: event.final ?? false, + const closesRelay = event.type === "close"; + try { + if (event.talkEvent) { + this.ctx.callbacks.onTalkEvent?.(event.talkEvent); + } + switch (event.type) { + case "ready": + this.ctx.callbacks.onStatus?.("listening"); + return; + case "audio": + if (event.audioBase64 && !this.playbackOverflowed) { + this.cancelRequestedForPlayback = false; + this.speechFramesDuringPlayback = 0; + this.playPcm16(event.audioBase64); + } + return; + case "clear": + this.playbackOverflowed = false; + this.stopOutput({ releaseDelayedToolResults: this.pendingOutputCancellations === 0 }); + if (event.talkEvent?.type === "turn.cancelled") { + this.abortConsults(); + } + return; + case "mark": + if (event.markName) { + this.scheduleMarkAck(event.markName); + } + return; + case "transcript": + if (event.role && event.text) { + this.ctx.callbacks.onTranscript?.({ + role: event.role, + text: event.text, + final: event.final ?? false, + }); + } + return; + case "toolCall": + void this.handleToolCall(event).catch((error: unknown) => { + this.reportToolResultSubmissionError(error); }); - } - return; - case "toolCall": - void this.handleToolCall(event).catch((error: unknown) => { - this.reportToolResultSubmissionError(error); - }); - return; - case "toolResult": - if (this.isFinalToolResult(event)) { - this.completeToolCall(event.callId); - } - return; - case "error": - this.lastRelayError = event.message ?? "Realtime relay failed"; - this.ctx.callbacks.onStatus?.("error", this.lastRelayError); - return; - case "close": - this.abortConsults(); - if (!this.closed) { + return; + case "toolResult": + if (this.isFinalToolResult(event)) { + this.completeToolCall(event.callId); + } + return; + case "error": + this.lastRelayError = event.message ?? "Realtime relay failed"; + this.ctx.callbacks.onStatus?.("error", this.lastRelayError); + return; + case "close": + this.abortConsults(); this.ctx.callbacks.onStatus?.( event.reason === "error" ? "error" : "idle", event.reason === "error" ? (this.lastRelayError ?? "Realtime relay closed") : undefined, ); - this.stopLocal(); - } - - default: + default: + } + } finally { + // The provider has ended this relay; consumer exceptions must not retain browser media. + if (closesRelay && !this.closed) { + this.stopLocal(); + } } } diff --git a/ui/src/pages/chat/realtime-talk-google-live.test.ts b/ui/src/pages/chat/realtime-talk-google-live.test.ts index 148697c54c1b..e59a073caf2e 100644 --- a/ui/src/pages/chat/realtime-talk-google-live.test.ts +++ b/ui/src/pages/chat/realtime-talk-google-live.test.ts @@ -21,6 +21,7 @@ type MockWebSocketHandler = (event?: MockWebSocketEvent) => void; type MockWebSocketEventType = "close" | "error" | "message" | "open"; const wsInstances: MockGoogleLiveWebSocket[] = []; +const audioContexts: MockAudioContext[] = []; const createdSources: MockAudioBufferSource[] = []; const inputProcessors: Array<{ connect: ReturnType; @@ -33,6 +34,7 @@ const inputSinks: Array<{ gain: { value: number }; }> = []; let getUserMedia: ReturnType; +let stopInputTrack: ReturnType; async function flushMicrotasks(): Promise { await Promise.resolve(); @@ -79,6 +81,19 @@ class MockGoogleLiveWebSocket { handler({ data }); } } + + emitClose() { + this.readyState = 3; + for (const handler of this.handlers.close) { + handler(); + } + } + + emitError() { + for (const handler of this.handlers.error) { + handler(); + } + } } class MockAudioBufferSource { @@ -97,6 +112,7 @@ class MockAudioContext { constructor(options?: { sampleRate?: number }) { this.sampleRate = options?.sampleRate ?? 24000; + audioContexts.push(this); } createMediaStreamSource() { @@ -230,13 +246,15 @@ function requireFirstTalkEvent(onTalkEvent: ReturnType): Record { beforeEach(() => { wsInstances.length = 0; + audioContexts.length = 0; createdSources.length = 0; inputProcessors.length = 0; inputSinks.length = 0; vi.stubGlobal("WebSocket", MockGoogleLiveWebSocket); vi.stubGlobal("AudioContext", MockAudioContext); + stopInputTrack = vi.fn(); getUserMedia = vi.fn(async () => ({ - getTracks: () => [{ stop: vi.fn() }], + getTracks: () => [{ stop: stopInputTrack }], })); vi.stubGlobal("navigator", { mediaDevices: { @@ -328,7 +346,7 @@ describe("GoogleLiveRealtimeTalkTransport", () => { const start = transport.start(); transport.stop(); resolveMedia({ getTracks: () => [{ stop: stopTrack }] } as unknown as MediaStream); - await start; + await expect(start).resolves.toBe("cancelled"); expect(stopTrack).toHaveBeenCalledOnce(); expect(inputProcessors).toHaveLength(0); @@ -353,6 +371,87 @@ describe("GoogleLiveRealtimeTalkTransport", () => { expect(readyEvent.transport).toBe("provider-websocket"); }); + it("releases owned media when the live socket closes", async () => { + const onStatus = vi.fn(); + const onTranscript = vi.fn(); + const transport = createTransport({ onStatus, onTranscript }); + + await expect(transport.start()).resolves.toBe("ready"); + const ws = latestWebSocket(); + ws.emitOpen(); + pumpMicrophone(new Float32Array(4096)); + ws.emitClose(); + + expect(onStatus).toHaveBeenCalledWith("error", "Realtime connection closed"); + expect(stopInputTrack).toHaveBeenCalledOnce(); + expect(inputProcessors.at(-1)?.disconnect).toHaveBeenCalledOnce(); + expect(audioContexts).toHaveLength(2); + for (const context of audioContexts) { + expect(context.close).toHaveBeenCalledOnce(); + } + + ws.emitMessage( + encodeJsonFrame({ + serverContent: { + inputTranscription: { text: "too late", finished: true }, + }, + }), + ); + await flushMicrotasks(); + expect(onTranscript).not.toHaveBeenCalled(); + }); + + it("preserves socket error precedence and ignores the later close", async () => { + const onStatus = vi.fn(); + const onTranscript = vi.fn(); + const transport = createTransport({ onStatus, onTranscript }); + + await expect(transport.start()).resolves.toBe("ready"); + const ws = latestWebSocket(); + ws.emitOpen(); + ws.emitError(); + ws.emitClose(); + + expect(onStatus).toHaveBeenCalledTimes(1); + expect(onStatus).toHaveBeenCalledWith("error", "Realtime connection failed"); + expect(stopInputTrack).toHaveBeenCalledOnce(); + ws.emitMessage( + encodeJsonFrame({ + serverContent: { + inputTranscription: { text: "too late", finished: true }, + }, + }), + ); + await flushMicrotasks(); + expect(onTranscript).not.toHaveBeenCalled(); + }); + + it.each(["status", "talk event"] as const)( + "releases socket resources when the terminal %s callback throws", + async (callbackKind) => { + const throwingCallback = vi.fn(() => { + throw new Error("consumer failed"); + }); + const transport = createTransport( + callbackKind === "status" + ? { onStatus: throwingCallback } + : { onTalkEvent: throwingCallback }, + ); + + await expect(transport.start()).resolves.toBe("ready"); + const ws = latestWebSocket(); + ws.emitOpen(); + expect(() => ws.emitError()).toThrow("consumer failed"); + + expect(stopInputTrack).toHaveBeenCalledOnce(); + expect(inputProcessors.at(-1)?.disconnect).toHaveBeenCalledOnce(); + for (const context of audioContexts) { + expect(context.close).toHaveBeenCalledOnce(); + } + ws.emitClose(); + }, + ); + it("reports microphone activity and resets it when stopped", async () => { const onInputLevel = vi.fn(); const transport = createTransport({ onInputLevel }); diff --git a/ui/src/pages/chat/realtime-talk-google-live.ts b/ui/src/pages/chat/realtime-talk-google-live.ts index b7b563b1ccfd..3ba479910bb0 100644 --- a/ui/src/pages/chat/realtime-talk-google-live.ts +++ b/ui/src/pages/chat/realtime-talk-google-live.ts @@ -21,6 +21,7 @@ import { submitRealtimeTalkConsult, type RealtimeTalkTransport, type RealtimeTalkTransportContext, + type RealtimeTalkTransportStartResult, } from "./realtime-talk-shared.ts"; import { captureRealtimeTalkVideoFrame, @@ -143,7 +144,7 @@ export class GoogleLiveRealtimeTalkTransport implements RealtimeTalkTransport { }); } - async start(): Promise { + async start(): Promise { if (!navigator.mediaDevices?.getUserMedia || typeof WebSocket === "undefined") { throw new Error("Realtime Talk requires browser WebSocket and microphone access"); } @@ -162,7 +163,7 @@ export class GoogleLiveRealtimeTalkTransport implements RealtimeTalkTransport { }); } catch (error) { if (this.closed) { - return; + return "cancelled"; } throw error; } finally { @@ -172,7 +173,7 @@ export class GoogleLiveRealtimeTalkTransport implements RealtimeTalkTransport { } if (this.closed) { media.getTracks().forEach((track) => track.stop()); - return; + return "cancelled"; } this.media = media; this.inputContext = new AudioContext({ sampleRate: this.session.audio.inputSampleRateHz }); @@ -193,16 +194,9 @@ export class GoogleLiveRealtimeTalkTransport implements RealtimeTalkTransport { this.ws.addEventListener("message", (event) => { void this.handleMessage(event.data); }); - this.ws.addEventListener("close", () => { - if (!this.closed) { - this.ctx.callbacks.onStatus?.("error", "Realtime connection closed"); - } - }); - this.ws.addEventListener("error", () => { - if (!this.closed) { - this.ctx.callbacks.onStatus?.("error", "Realtime connection failed"); - } - }); + this.ws.addEventListener("close", () => this.failConnection("Realtime connection closed")); + this.ws.addEventListener("error", () => this.failConnection("Realtime connection failed")); + return "ready"; } async setVideoEnabled(enabled: boolean): Promise { @@ -214,10 +208,18 @@ export class GoogleLiveRealtimeTalkTransport implements RealtimeTalkTransport { } stop(options?: { emitClosed?: boolean }): void { - if (!this.closed && options?.emitClosed !== false) { - this.emitTalkEvent({ type: "session.closed", final: true }); - } + const emitClosed = !this.closed && options?.emitClosed !== false; this.closed = true; + try { + if (emitClosed) { + this.emitTalkEvent({ type: "session.closed", final: true }); + } + } finally { + this.releaseResources(); + } + } + + private releaseResources(): void { this.mediaSetupController?.abort(); this.mediaSetupController = null; this.setupComplete = false; @@ -241,6 +243,18 @@ export class GoogleLiveRealtimeTalkTransport implements RealtimeTalkTransport { this.ws = null; } + private failConnection(detail: string): void { + if (this.closed) { + return; + } + try { + this.ctx.callbacks.onStatus?.("error", detail); + } finally { + // Socket failure is terminal even when a consumer callback rejects the update. + this.stop(); + } + } + private startMicrophonePump(): void { if (this.closed || !this.media || !this.inputContext) { return; diff --git a/ui/src/pages/chat/realtime-talk-lifecycle.test.ts b/ui/src/pages/chat/realtime-talk-lifecycle.test.ts index 6123a5fef255..21cd39d9196c 100644 --- a/ui/src/pages/chat/realtime-talk-lifecycle.test.ts +++ b/ui/src/pages/chat/realtime-talk-lifecycle.test.ts @@ -4,9 +4,11 @@ import type { RealtimeTalkTransportContext } from "./realtime-talk-shared.ts"; const transportMock = vi.hoisted(() => ({ relayContexts: [] as RealtimeTalkTransportContext[], + relayStops: [] as Array>, webRtcContexts: [] as RealtimeTalkTransportContext[], webRtcStops: [] as Array>, - start: vi.fn(async () => undefined), + relayActivate: vi.fn(), + start: vi.fn(async (): Promise<"ready" | "cancelled"> => "ready"), stop: vi.fn(), })); @@ -16,7 +18,13 @@ vi.mock("./realtime-talk-gateway-relay.ts", () => ({ context: RealtimeTalkTransportContext, ) { transportMock.relayContexts.push(context); - return { start: transportMock.start, stop: transportMock.stop }; + const stop = vi.fn((options?: { emitClosed?: boolean }) => transportMock.stop(options)); + transportMock.relayStops.push(stop); + return { + start: transportMock.start, + activate: transportMock.relayActivate, + stop, + }; }), })); vi.mock("./realtime-talk-google-live.ts", () => ({ @@ -66,8 +74,10 @@ function transcriptContext(contexts: RealtimeTalkTransportContext[], index = 0): describe("RealtimeTalkSession lifecycle", () => { beforeEach(() => { transportMock.relayContexts.length = 0; + transportMock.relayStops.length = 0; transportMock.webRtcContexts.length = 0; transportMock.webRtcStops.length = 0; + transportMock.relayActivate.mockClear(); transportMock.start.mockClear(); transportMock.stop.mockClear(); }); @@ -217,8 +227,109 @@ describe("RealtimeTalkSession lifecycle", () => { session.stop(); }); + it("keeps the active transport when a replacement cancels during startup", async () => { + const transcriptEntryIds: string[] = []; + const request = vi.fn(async (method: string, params?: { entryId?: string }) => { + if (method === "talk.client.create") { + return { + provider: "openai", + transport: "webrtc", + voiceSessionId: "voice-cancelled-replacement", + clientSecret: "secret", + }; + } + if (method === "talk.client.transcript") { + transcriptEntryIds.push(String(params?.entryId)); + } + return { ok: true }; + }); + const session = new RealtimeTalkSession({ request } as never, "agent:main:main"); + await session.start(); + const activeContext = transcriptContext(transportMock.webRtcContexts); + transportMock.start.mockResolvedValueOnce("cancelled"); + + await session.start(); + + expect(transportMock.webRtcStops[0]).not.toHaveBeenCalled(); + expect(transportMock.webRtcStops[1]).toHaveBeenCalledWith({ emitClosed: false }); + expect(request.mock.calls.some(([method]) => method === "talk.client.close")).toBe(false); + activeContext.callbacks.onTranscript?.({ + role: "user", + text: "active transport survived", + final: true, + }); + await activeContext.flushTranscriptWrites?.(); + expect(transcriptEntryIds).toEqual(["1"]); + session.stop(); + }); + + it("restores the active relay when replacement activation throws", async () => { + const request = vi.fn(async (method: string) => { + if (method === "talk.client.create") { + return { + provider: "openai", + transport: "gateway-relay", + relaySessionId: "relay-activation", + audio: { + inputEncoding: "pcm16", + inputSampleRateHz: 24_000, + outputEncoding: "pcm16", + outputSampleRateHz: 24_000, + }, + }; + } + return { ok: true }; + }); + const session = new RealtimeTalkSession({ request } as never, "agent:main:main"); + await session.start(); + transportMock.relayActivate.mockImplementationOnce(() => { + throw new Error("activation failed"); + }); + + await expect(session.start()).rejects.toThrow("activation failed"); + + expect(transportMock.relayStops[0]).not.toHaveBeenCalled(); + expect(transportMock.relayStops[1]).toHaveBeenCalledWith({ emitClosed: false }); + session.stop(); + expect(transportMock.relayStops[0]).toHaveBeenCalledOnce(); + }); + + it("does not restore an active relay after activation stops the session", async () => { + const request = vi.fn(async (method: string) => { + if (method === "talk.client.create") { + return { + provider: "openai", + transport: "gateway-relay", + relaySessionId: "relay-activation-stop", + audio: { + inputEncoding: "pcm16", + inputSampleRateHz: 24_000, + outputEncoding: "pcm16", + outputSampleRateHz: 24_000, + }, + }; + } + return { ok: true }; + }); + const session = new RealtimeTalkSession({ request } as never, "agent:main:main"); + await session.start(); + transportMock.relayActivate.mockImplementationOnce(() => { + session.stop(); + throw new Error("activation stopped"); + }); + + await expect(session.start()).rejects.toThrow("activation stopped"); + + expect(transportMock.relayStops[0]).toHaveBeenCalledOnce(); + expect(transportMock.relayStops[0]).toHaveBeenCalledWith({ emitClosed: false }); + expect(transportMock.relayStops[1]).toHaveBeenCalledOnce(); + session.stop(); + expect(transportMock.relayStops[0]).toHaveBeenCalledOnce(); + expect(transportMock.relayStops[1]).toHaveBeenCalledOnce(); + }); + it("ignores transcripts from a superseded pending replacement", async () => { - const firstReplacementStart = createDeferred(); + const firstReplacementStart = createDeferred<"ready">(); const transcriptEntryIds: string[] = []; const request = vi.fn(async (method: string, params?: { entryId?: string }) => { if (method === "talk.client.create") { @@ -250,7 +361,7 @@ describe("RealtimeTalkSession lifecycle", () => { expect(transcriptEntryIds).toEqual(["1"]); expect(transportMock.webRtcStops[1]).toHaveBeenCalledWith({ emitClosed: false }); - firstReplacementStart.resolve(); + firstReplacementStart.resolve("ready"); await firstReplacement; session.stop(); }); @@ -310,7 +421,7 @@ describe("RealtimeTalkSession lifecycle", () => { it("stops a pending replacement when transcript overflow closes the active call", async () => { const warn = vi.spyOn(console, "warn").mockImplementation(() => undefined); - const replacementStart = createDeferred(); + const replacementStart = createDeferred<"ready">(); const firstTranscript = createDeferred(); const request = vi.fn(async (method: string, params?: { entryId?: string }) => { if (method === "talk.client.create") { @@ -349,7 +460,7 @@ describe("RealtimeTalkSession lifecycle", () => { expect(transportMock.webRtcStops[0]).toHaveBeenCalledWith(); expect(transportMock.webRtcStops[1]).toHaveBeenCalledWith({ emitClosed: false }); - replacementStart.resolve(); + replacementStart.resolve("ready"); await replacement; firstTranscript.resolve(); await vi.waitFor(() => @@ -398,8 +509,32 @@ describe("RealtimeTalkSession lifecycle", () => { recovered.stop(); }); + it("rejects a terminal failure during initial transport setup", async () => { + const request = vi.fn(async (method: string) => { + if (method === "talk.client.create") { + return { + provider: "openai", + transport: "webrtc", + voiceSessionId: "voice-terminal-startup", + clientSecret: "secret", + }; + } + return { ok: true }; + }); + const onStatus = vi.fn(); + const session = new RealtimeTalkSession({ request } as never, "agent:main:main", { + onStatus, + }); + transportMock.start.mockRejectedValueOnce(new Error("Realtime connection closed")); + + await expect(session.start()).rejects.toThrow("Realtime connection closed"); + + expect(onStatus).toHaveBeenCalledWith("connecting"); + expect(transportMock.webRtcStops[0]).toHaveBeenCalledWith({ emitClosed: false }); + }); + it("does not restore a failed replacement after concurrent stop", async () => { - const replacementStart = createDeferred(); + const replacementStart = createDeferred<"ready">(); const transcriptEntryIds: string[] = []; const request = vi.fn(async (method: string, params?: { entryId?: string }) => { if (method === "talk.client.create") { @@ -752,6 +887,7 @@ describe("RealtimeTalkSession lifecycle", () => { const session = new RealtimeTalkSession({ request } as never, "agent:main:main"); await session.start(); const context = transcriptContext(transportMock.relayContexts); + expect(transportMock.relayActivate).toHaveBeenCalledOnce(); context.callbacks.onTranscript?.({ role: "user", text: "server owns this", final: true }); await Promise.resolve(); diff --git a/ui/src/pages/chat/realtime-talk-shared.ts b/ui/src/pages/chat/realtime-talk-shared.ts index 1ba545db9ca9..5788d5a9ade6 100644 --- a/ui/src/pages/chat/realtime-talk-shared.ts +++ b/ui/src/pages/chat/realtime-talk-shared.ts @@ -104,8 +104,11 @@ export type RealtimeTalkSessionResult = | RealtimeTalkGatewayRelaySessionResult | RealtimeTalkManagedRoomSessionResult; +export type RealtimeTalkTransportStartResult = "ready" | "cancelled"; + export type RealtimeTalkTransport = { - start(): Promise; + start(): Promise; + activate?: () => void; stop(options?: { emitClosed?: boolean }): void; setVideoEnabled?: (enabled: boolean) => Promise; switchCamera?: (videoDeviceId: string | undefined) => Promise; diff --git a/ui/src/pages/chat/realtime-talk-transcript-owner.ts b/ui/src/pages/chat/realtime-talk-transcript-owner.ts index ac6dc1d1a14f..2a4df3f7e4ef 100644 --- a/ui/src/pages/chat/realtime-talk-transcript-owner.ts +++ b/ui/src/pages/chat/realtime-talk-transcript-owner.ts @@ -1,6 +1,7 @@ import { DEFAULT_GATEWAY_REQUEST_TIMEOUT_MS } from "@openclaw/gateway-client/browser"; import type { BoundedSerialQueue } from "../../../../src/shared/bounded-serial-queue.js"; import type { GatewayBrowserClient } from "../../api/gateway.ts"; +import type { RealtimeTalkTransport } from "./realtime-talk-shared.ts"; export type ClientVoiceSessionOwner = { signal: AbortSignal; @@ -57,6 +58,25 @@ export function reserveClientVoiceSessionOwner( }; } +export function retireUncommittedRealtimeTalkTransport(params: { + nextTransport: RealtimeTalkTransport | null; + transport: string; + owner: ClientVoiceSessionOwner; + reusesExistingOwner: boolean; + closeVoiceSession: () => void; +}): void { + params.nextTransport?.stop({ emitClosed: false }); + if (params.reusesExistingOwner) { + return; + } + if (params.transport === "gateway-relay" && params.nextTransport) { + // The relay transport owns server close once constructed; release browser ownership. + params.owner.release(); + return; + } + params.closeVoiceSession(); +} + export function transcriptPersistenceAbortError(): Error { const error = new Error("voice transcript persistence aborted"); error.name = "AbortError"; diff --git a/ui/src/pages/chat/realtime-talk-transcript-queue.test.ts b/ui/src/pages/chat/realtime-talk-transcript-queue.test.ts index 4c3fd84dae60..82dd8fe138d6 100644 --- a/ui/src/pages/chat/realtime-talk-transcript-queue.test.ts +++ b/ui/src/pages/chat/realtime-talk-transcript-queue.test.ts @@ -4,7 +4,7 @@ import type { RealtimeTalkTransportContext } from "./realtime-talk-shared.ts"; const transportMock = vi.hoisted(() => ({ context: undefined as RealtimeTalkTransportContext | undefined, - start: vi.fn(async () => undefined), + start: vi.fn(async () => "ready" as const), stop: vi.fn(), })); diff --git a/ui/src/pages/chat/realtime-talk-webrtc.test.ts b/ui/src/pages/chat/realtime-talk-webrtc.test.ts index 81b947269b20..6bef1ac0e370 100644 --- a/ui/src/pages/chat/realtime-talk-webrtc.test.ts +++ b/ui/src/pages/chat/realtime-talk-webrtc.test.ts @@ -5,6 +5,7 @@ import { REALTIME_VOICE_AGENT_CONSULT_TOOL_NAME } from "./realtime-talk-shared.t import { WebRtcSdpRealtimeTalkTransport } from "./realtime-talk-webrtc.ts"; let getUserMedia: ReturnType; +let stopInputTrack: ReturnType; class FakeDataChannel extends EventTarget { readyState: RTCDataChannelState = "open"; @@ -177,7 +178,8 @@ describe("WebRtcSdpRealtimeTalkTransport", () => { beforeEach(() => { FakePeerConnection.instances = []; - const track = { stop: vi.fn() } as unknown as MediaStreamTrack; + stopInputTrack = vi.fn(); + const track = { stop: stopInputTrack } as unknown as MediaStreamTrack; const stream = { getAudioTracks: () => [track], getTracks: () => [track], @@ -249,7 +251,7 @@ describe("WebRtcSdpRealtimeTalkTransport", () => { transport.stop(); resolveMedia(stream); - await expect(startPromise).resolves.toBeUndefined(); + await expect(startPromise).resolves.toBe("cancelled"); expect(peer?.addTrack).not.toHaveBeenCalled(); expect(stopTrack).toHaveBeenCalledTimes(1); expect(fetchMock).not.toHaveBeenCalled(); @@ -276,7 +278,7 @@ describe("WebRtcSdpRealtimeTalkTransport", () => { transport.stop(); rejectOffer(new Error("closed peer rejected offer creation")); - await expect(startPromise).resolves.toBeUndefined(); + await expect(startPromise).resolves.toBe("cancelled"); expect(fetchMock).not.toHaveBeenCalled(); }); @@ -385,10 +387,77 @@ describe("WebRtcSdpRealtimeTalkTransport", () => { transport.stop(); - await expect(startResult).resolves.toBeUndefined(); + await expect(startResult).resolves.toBe("cancelled"); expect(offerSignal?.aborted).toBe(true); }); + it("reports a closed candidate when the peer fails during final setup", async () => { + stubAnswerSdpFetch(); + const onStatus = vi.fn(); + const onTalkEvent = vi.fn(); + const transport = createOpenAiTransport({}, { onStatus, onTalkEvent }); + const start = transport.start(); + const peer = FakePeerConnection.instances[0]; + if (!peer) { + throw new Error("expected WebRTC peer"); + } + let finishRemoteDescription: (() => void) | undefined; + const remoteDescription = vi.spyOn(peer, "setRemoteDescription").mockImplementation( + async () => + await new Promise((resolve) => { + finishRemoteDescription = resolve; + }), + ); + await waitForFast(() => expect(remoteDescription).toHaveBeenCalled()); + + peer.connectionState = "failed"; + peer.dispatchEvent(new Event("connectionstatechange")); + finishRemoteDescription?.(); + + await expect(start).rejects.toThrow("Realtime connection closed"); + expect(onStatus).not.toHaveBeenCalled(); + expect(onTalkEvent).not.toHaveBeenCalled(); + expect(peer.channel.close).toHaveBeenCalledOnce(); + }); + + it("releases an active peer when the terminal status callback throws", async () => { + stubAnswerSdpFetch(); + const onStatus = vi.fn(() => { + throw new Error("consumer failed"); + }); + const transport = createOpenAiTransport({}, { onStatus }); + await expect(transport.start()).resolves.toBe("ready"); + const peer = FakePeerConnection.instances[0]; + if (!peer) { + throw new Error("expected WebRTC peer"); + } + + peer.connectionState = "failed"; + peer.dispatchEvent(new Event("connectionstatechange")); + + expect(onStatus).toHaveBeenCalledWith("error", "Realtime connection closed"); + expect(stopInputTrack).toHaveBeenCalledOnce(); + expect(peer.channel.close).toHaveBeenCalledOnce(); + }); + + it("releases an active peer when its closed-event callback throws", async () => { + stubAnswerSdpFetch(); + const onTalkEvent = vi.fn(() => { + throw new Error("consumer failed"); + }); + const transport = createOpenAiTransport({}, { onTalkEvent }); + await expect(transport.start()).resolves.toBe("ready"); + const peer = FakePeerConnection.instances[0]; + if (!peer) { + throw new Error("expected WebRTC peer"); + } + + expect(() => transport.stop()).toThrow("consumer failed"); + + expect(stopInputTrack).toHaveBeenCalledOnce(); + expect(peer.channel.close).toHaveBeenCalledOnce(); + }); + it("clears the WebRTC offer timeout after setup succeeds", async () => { vi.useFakeTimers(); let offerSignal: AbortSignal | undefined; diff --git a/ui/src/pages/chat/realtime-talk-webrtc.ts b/ui/src/pages/chat/realtime-talk-webrtc.ts index fa7e2d23b48f..e1f6753ff173 100644 --- a/ui/src/pages/chat/realtime-talk-webrtc.ts +++ b/ui/src/pages/chat/realtime-talk-webrtc.ts @@ -14,6 +14,7 @@ import { submitRealtimeTalkConsult, type RealtimeTalkTransport, type RealtimeTalkTransportContext, + type RealtimeTalkTransportStartResult, } from "./realtime-talk-shared.ts"; import { captureRealtimeTalkVideoFrame } from "./realtime-talk-video.ts"; import { @@ -47,6 +48,8 @@ export class WebRtcSdpRealtimeTalkTransport implements RealtimeTalkTransport { private readonly camera: RealtimeTalkCameraController; private readonly consultAbortControllers = new Set(); private readonly emitTalkEvent: ReturnType; + private starting = false; + private startupError: Error | null = null; constructor( private readonly session: RealtimeTalkWebRtcSdpSessionResult, @@ -62,11 +65,13 @@ export class WebRtcSdpRealtimeTalkTransport implements RealtimeTalkTransport { }); } - async start(): Promise { + async start(): Promise { if (!navigator.mediaDevices?.getUserMedia || typeof RTCPeerConnection === "undefined") { throw new Error("Realtime Talk requires browser WebRTC and microphone access"); } this.closed = false; + this.starting = true; + this.startupError = null; this.mediaSetupController?.abort(); const peer = new RTCPeerConnection(); this.peer = peer; @@ -116,11 +121,11 @@ export class WebRtcSdpRealtimeTalkTransport implements RealtimeTalkTransport { } } if (media === cancelledSetup) { - return; + return this.cancelledStart(); } if (!this.isCurrentPeer(peer)) { media.getTracks().forEach((track) => track.stop()); - return; + return this.cancelledStart(); } this.media = media; if (this.ctx.callbacks.onInputLevel) { @@ -135,7 +140,7 @@ export class WebRtcSdpRealtimeTalkTransport implements RealtimeTalkTransport { const channel = peer.createDataChannel("oai-events"); if (!this.isCurrentPeer(peer)) { channel.close(); - return; + return this.cancelledStart(); } this.channel = channel; channel.addEventListener("open", () => { @@ -154,17 +159,17 @@ export class WebRtcSdpRealtimeTalkTransport implements RealtimeTalkTransport { const offer = await this.awaitSetupStep(peer, peer.createOffer()); if (offer === cancelledSetup) { - return; + return this.cancelledStart(); } if (!this.isCurrentPeer(peer)) { - return; + return this.cancelledStart(); } const localDescriptionResult = await this.awaitSetupStep(peer, peer.setLocalDescription(offer)); if (localDescriptionResult === cancelledSetup) { - return; + return this.cancelledStart(); } if (!this.isCurrentPeer(peer)) { - return; + return this.cancelledStart(); } const answerSdp = await this.offerExchange.readAnswer({ session: this.session, @@ -173,18 +178,23 @@ export class WebRtcSdpRealtimeTalkTransport implements RealtimeTalkTransport { isCurrent: () => this.isCurrentPeer(peer), }); if (answerSdp === undefined) { - return; + return this.cancelledStart(); } if (!this.isCurrentPeer(peer)) { - return; + return this.cancelledStart(); } - await this.awaitSetupStep( + const remoteDescriptionResult = await this.awaitSetupStep( peer, peer.setRemoteDescription({ type: "answer", sdp: answerSdp, }), ); + if (remoteDescriptionResult === cancelledSetup || !this.isCurrentPeer(peer)) { + return this.cancelledStart(); + } + this.starting = false; + return "ready"; } async setVideoEnabled(enabled: boolean): Promise { @@ -199,6 +209,18 @@ export class WebRtcSdpRealtimeTalkTransport implements RealtimeTalkTransport { return !this.closed && this.peer === peer; } + private cancelledStart(): RealtimeTalkTransportStartResult { + const startupError = this.currentStartupError(); + if (startupError) { + throw startupError; + } + return "cancelled"; + } + + private currentStartupError(): Error | null { + return this.startupError; + } + private async awaitSetupStep( peer: RTCPeerConnection, promise: Promise, @@ -214,10 +236,19 @@ export class WebRtcSdpRealtimeTalkTransport implements RealtimeTalkTransport { } stop(options?: { emitClosed?: boolean }): void { - if (!this.closed && options?.emitClosed !== false) { - this.emitTalkEvent({ type: "session.closed", final: true }); - } + const emitClosed = !this.closed && options?.emitClosed !== false; this.closed = true; + try { + if (emitClosed) { + this.emitTalkEvent({ type: "session.closed", final: true }); + } + } finally { + this.releaseResources(); + } + } + + private releaseResources(): void { + this.starting = false; this.mediaSetupController?.abort(); this.mediaSetupController = null; this.offerExchange.abort(); @@ -246,9 +277,17 @@ export class WebRtcSdpRealtimeTalkTransport implements RealtimeTalkTransport { if (this.closed) { return; } - this.ctx.callbacks.onStatus?.("error", detail); - // A terminal peer failure still owns live browser media until stop() releases it. - this.stop(); + const wasStarting = this.starting; + try { + if (!wasStarting) { + this.ctx.callbacks.onStatus?.("error", detail); + } else { + this.startupError = new Error(detail); + } + } finally { + // A terminal peer failure still owns browser media if status delivery fails. + this.stop({ emitClosed: !wasStarting }); + } } private send(event: unknown): void { diff --git a/ui/src/pages/chat/realtime-talk.test.ts b/ui/src/pages/chat/realtime-talk.test.ts index 5aa9b68775f6..d8357cc3b99d 100644 --- a/ui/src/pages/chat/realtime-talk.test.ts +++ b/ui/src/pages/chat/realtime-talk.test.ts @@ -20,11 +20,11 @@ const { googleSwitchCamera, webRtcSwitchCamera, } = { - googleStart: vi.fn(async () => undefined), + googleStart: vi.fn(async () => "ready" as const), googleStop: vi.fn(), - relayStart: vi.fn(async () => undefined), + relayStart: vi.fn(async () => "ready" as const), relayStop: vi.fn(), - webRtcStart: vi.fn(async () => undefined), + webRtcStart: vi.fn(async () => "ready" as const), webRtcStop: vi.fn(), googleSetVideoEnabled: vi.fn(async () => undefined), webRtcSetVideoEnabled: vi.fn(async () => undefined), diff --git a/ui/src/pages/chat/realtime-talk.ts b/ui/src/pages/chat/realtime-talk.ts index 479d11c2eae6..c029cb085c24 100644 --- a/ui/src/pages/chat/realtime-talk.ts +++ b/ui/src/pages/chat/realtime-talk.ts @@ -24,6 +24,7 @@ import { type ClientVoiceSessionOwner, type DetachedVoiceSession, reserveClientVoiceSessionOwner, + retireUncommittedRealtimeTalkTransport, transcriptPersistenceAbortError, waitForTranscriptRetry, } from "./realtime-talk-transcript-owner.ts"; @@ -174,6 +175,9 @@ export class RealtimeTalkSession { const existingTransport = this.transport; const existingVoiceSessionId = this.voiceSessionId; const existingOwner = this.clientVoiceSessionOwner; + const existingAcceptingTranscripts = this.acceptingTranscripts; + const existingServerOwnedVoiceSession = this.serverOwnedVoiceSession; + const existingTransportGeneration = this.transportGeneration; const providerVideoCapable = await this.resolveVideoCapability(); if (this.closed || lifecycleGeneration !== this.lifecycleGeneration) { return; @@ -230,6 +234,7 @@ export class RealtimeTalkSession { ); const transcriptQueue = this.transcriptQueue; let nextTransport: RealtimeTalkTransport | null = null; + let startResult: Awaited>; try { nextTransport = createTransport(session, { client: this.client, @@ -246,26 +251,38 @@ export class RealtimeTalkSession { this.callbacks.onVideoCapability?.( providerVideoCapable && typeof nextTransport.setVideoEnabled === "function", ); - await nextTransport.start(); + startResult = await nextTransport.start(); } catch (error) { if (this.pendingTransport === nextTransport) { this.pendingTransport = null; } - nextTransport?.stop({ emitClosed: false }); - if (!(existingOwner && adoptedOwner === existingOwner)) { - this.closeUnadoptedVoiceSession(voiceSessionId, transport, adoptedOwner); - } + retireUncommittedRealtimeTalkTransport({ + nextTransport, + transport, + owner: adoptedOwner, + reusesExistingOwner: Boolean(existingOwner && adoptedOwner === existingOwner), + closeVoiceSession: () => + this.closeUnadoptedVoiceSession(voiceSessionId, transport, adoptedOwner), + }); ownerTransferred = true; throw error; } if (this.pendingTransport === nextTransport) { this.pendingTransport = null; } - if (this.closed || lifecycleGeneration !== this.lifecycleGeneration) { - nextTransport.stop({ emitClosed: false }); - if (!(existingOwner && adoptedOwner === existingOwner)) { - this.closeUnadoptedVoiceSession(voiceSessionId, transport, adoptedOwner); - } + if ( + startResult === "cancelled" || + this.closed || + lifecycleGeneration !== this.lifecycleGeneration + ) { + retireUncommittedRealtimeTalkTransport({ + nextTransport, + transport, + owner: adoptedOwner, + reusesExistingOwner: Boolean(existingOwner && adoptedOwner === existingOwner), + closeVoiceSession: () => + this.closeUnadoptedVoiceSession(voiceSessionId, transport, adoptedOwner), + }); ownerTransferred = true; return; } @@ -280,6 +297,39 @@ export class RealtimeTalkSession { } else { this.clientVoiceSessionOwner = adoptedOwner; } + try { + // Publish the candidate before releasing bounded events buffered during permission. + nextTransport.activate?.(); + } catch (error) { + const canRestoreExistingTransport = + !this.closed && + lifecycleGeneration === this.lifecycleGeneration && + this.transport === nextTransport; + if (canRestoreExistingTransport) { + this.voiceSessionId = existingVoiceSessionId; + this.acceptingTranscripts = existingAcceptingTranscripts; + this.serverOwnedVoiceSession = existingServerOwnedVoiceSession; + this.transportGeneration = existingTransportGeneration; + this.transport = existingTransport; + this.clientVoiceSessionOwner = existingOwner; + retireUncommittedRealtimeTalkTransport({ + nextTransport, + transport, + owner: adoptedOwner, + reusesExistingOwner: Boolean(existingOwner && adoptedOwner === existingOwner), + closeVoiceSession: () => + this.closeUnadoptedVoiceSession(voiceSessionId, transport, adoptedOwner), + }); + } else { + // Stop or supersession wins activation and owns allocation cleanup. + if (this.transport === nextTransport) { + nextTransport.stop({ emitClosed: false }); + } + existingTransport?.stop({ emitClosed: false }); + } + ownerTransferred = true; + throw error; + } ownerTransferred = true; existingTransport?.stop({ emitClosed: false }); } finally {