diff --git a/packages/gateway-client/src/browser.ts b/packages/gateway-client/src/browser.ts index 791146902982..28cc53c409ed 100644 --- a/packages/gateway-client/src/browser.ts +++ b/packages/gateway-client/src/browser.ts @@ -7,7 +7,11 @@ export * from "./protocol-client.js"; export * from "./reconnect-policy.js"; export * from "./session-projection.js"; export * from "./session-subscriptions.js"; -export { DEFAULT_PREAUTH_HANDSHAKE_TIMEOUT_MS, resolveSafeTimeoutDelayMs } from "./timeouts.js"; +export { + DEFAULT_GATEWAY_REQUEST_TIMEOUT_MS, + DEFAULT_PREAUTH_HANDSHAKE_TIMEOUT_MS, + resolveSafeTimeoutDelayMs, +} from "./timeouts.js"; export * from "@openclaw/gateway-protocol/client-info"; export * from "@openclaw/gateway-protocol/connect-error-details"; export * from "@openclaw/gateway-protocol/gateway-error-details"; diff --git a/ui/src/pages/chat/realtime-talk-lifecycle.test.ts b/ui/src/pages/chat/realtime-talk-lifecycle.test.ts new file mode 100644 index 000000000000..ab231487ed82 --- /dev/null +++ b/ui/src/pages/chat/realtime-talk-lifecycle.test.ts @@ -0,0 +1,478 @@ +// @vitest-environment node +import { beforeEach, describe, expect, it, vi } from "vitest"; +import type { RealtimeTalkTransportContext } from "./realtime-talk-shared.ts"; + +const transportMock = vi.hoisted(() => ({ + relayContexts: [] as RealtimeTalkTransportContext[], + webRtcContexts: [] as RealtimeTalkTransportContext[], + start: vi.fn(async () => undefined), + stop: vi.fn(), +})); + +vi.mock("./realtime-talk-gateway-relay.ts", () => ({ + GatewayRelayRealtimeTalkTransport: vi.fn(function ( + _session: unknown, + context: RealtimeTalkTransportContext, + ) { + transportMock.relayContexts.push(context); + return { start: transportMock.start, stop: transportMock.stop }; + }), +})); +vi.mock("./realtime-talk-google-live.ts", () => ({ + GoogleLiveRealtimeTalkTransport: vi.fn(), +})); +vi.mock("./realtime-talk-webrtc.ts", () => ({ + WebRtcSdpRealtimeTalkTransport: vi.fn(function ( + _session: unknown, + context: RealtimeTalkTransportContext, + ) { + transportMock.webRtcContexts.push(context); + return { start: transportMock.start, stop: transportMock.stop }; + }), +})); + +import { RealtimeTalkSession } from "./realtime-talk.ts"; + +const requestTimeoutOptions = { timeoutMs: 30_000 }; + +type TranscriptContext = RealtimeTalkTransportContext & { + callbacks: { + onTranscript?: (entry: { role: "user" | "assistant"; text: string; final: boolean }) => void; + }; + flushTranscriptWrites?: () => Promise; +}; + +function createDeferred() { + let resolve!: (value: T) => void; + const promise = new Promise((resolvePromise) => { + resolve = resolvePromise; + }); + return { promise, resolve }; +} + +function transcriptContext(contexts: RealtimeTalkTransportContext[], index = 0): TranscriptContext { + const context = contexts[index]; + if (!context) { + throw new Error("expected realtime transport context"); + } + return context as TranscriptContext; +} + +describe("RealtimeTalkSession lifecycle", () => { + beforeEach(() => { + transportMock.relayContexts.length = 0; + transportMock.webRtcContexts.length = 0; + transportMock.start.mockClear(); + transportMock.stop.mockClear(); + }); + + it("retries finalized transcript writes in order", async () => { + vi.useFakeTimers(); + try { + const transcriptEntryIds: string[] = []; + let firstAttempt = true; + const request = vi.fn(async (method: string, params?: { entryId?: string }) => { + if (method === "talk.client.create") { + return { + provider: "openai", + transport: "webrtc", + voiceSessionId: "voice-queue", + clientSecret: "secret", + }; + } + if (method === "talk.client.transcript") { + transcriptEntryIds.push(String(params?.entryId)); + if (params?.entryId === "1" && firstAttempt) { + firstAttempt = false; + throw new Error("temporary failure"); + } + return { ok: true }; + } + return { ok: true }; + }); + const session = new RealtimeTalkSession({ request } as never, "agent:main:main"); + await session.start(); + const context = transcriptContext(transportMock.webRtcContexts); + context.callbacks.onTranscript?.({ role: "user", text: "first", final: true }); + context.callbacks.onTranscript?.({ role: "assistant", text: "second", final: true }); + + await vi.advanceTimersByTimeAsync(500); + await vi.waitFor(() => expect(transcriptEntryIds).toEqual(["1", "1", "2"])); + session.stop(); + await vi.runAllTimersAsync(); + } finally { + vi.useRealTimers(); + } + }); + + it("continues entry ids when the same voice session replaces its transport", 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-resume", + clientSecret: "secret", + }; + } + if (method === "talk.client.transcript") { + transcriptEntryIds.push(String(params?.entryId)); + } + return { ok: true }; + }); + const client = { request } as never; + const session = new RealtimeTalkSession(client, "agent:main:main", {}); + + await session.start(); + const firstContext = transcriptContext(transportMock.webRtcContexts); + firstContext.callbacks.onTranscript?.({ role: "user", text: "first", final: true }); + await firstContext.flushTranscriptWrites?.(); + + await session.start(); + const secondContext = transcriptContext(transportMock.webRtcContexts, 1); + secondContext.callbacks.onTranscript?.({ role: "assistant", text: "second", final: true }); + await secondContext.flushTranscriptWrites?.(); + + expect(transcriptEntryIds).toEqual(["1", "2"]); + expect(request.mock.calls.filter(([method]) => method === "talk.client.create")).toEqual([ + [ + "talk.client.create", + { sessionKey: "agent:main:main", capabilities: ["voice-transcript"] }, + requestTimeoutOptions, + ], + [ + "talk.client.create", + { + sessionKey: "agent:main:main", + voiceSessionId: "voice-resume", + capabilities: ["voice-transcript"], + }, + requestTimeoutOptions, + ], + ]); + session.stop(); + await vi.waitFor(() => + expect(request.mock.calls.filter(([method]) => method === "talk.client.close")).toHaveLength( + 1, + ), + ); + await Promise.resolve(); + + const firstReplacement = new RealtimeTalkSession(client, "agent:main:main"); + const secondReplacement = new RealtimeTalkSession(client, "agent:main:main"); + await firstReplacement.start(); + await secondReplacement.start(); + firstReplacement.stop(); + secondReplacement.stop(); + }); + + it("surfaces transcript failure after three attempts", async () => { + vi.useFakeTimers(); + const warn = vi.spyOn(console, "warn").mockImplementation(() => undefined); + try { + const request = vi.fn(async (method: string) => { + if (method === "talk.client.create") { + return { + provider: "openai", + transport: "webrtc", + voiceSessionId: "voice-failure", + clientSecret: "secret", + }; + } + if (method === "talk.client.transcript") { + throw new Error("still unavailable"); + } + return { ok: true }; + }); + const onStatus = vi.fn(); + const session = new RealtimeTalkSession({ request } as never, "agent:main:main", { + onStatus, + }); + await session.start(); + const context = transcriptContext(transportMock.webRtcContexts); + context.callbacks.onTranscript?.({ role: "user", text: "save me", final: true }); + + await vi.advanceTimersByTimeAsync(2_500); + await vi.waitFor(() => + expect(onStatus).toHaveBeenCalledWith( + "error", + expect.stringContaining("Voice transcript could not be saved"), + ), + ); + expect( + request.mock.calls.filter(([method]) => method === "talk.client.transcript"), + ).toHaveLength(3); + expect(warn).toHaveBeenCalled(); + session.stop(); + await vi.runAllTimersAsync(); + } finally { + warn.mockRestore(); + vi.useRealTimers(); + } + }); + + it("retries logical voice-session close after transient failures", async () => { + vi.useFakeTimers(); + try { + let closeAttempts = 0; + const request = vi.fn(async (method: string) => { + if (method === "talk.client.create") { + return { + provider: "openai", + transport: "webrtc", + voiceSessionId: "voice-close-retry", + clientSecret: "secret", + }; + } + if (method === "talk.client.close" && ++closeAttempts < 3) { + throw new Error("temporary close failure"); + } + return { ok: true }; + }); + const session = new RealtimeTalkSession({ request } as never, "agent:main:main"); + await session.start(); + + session.stop(); + await vi.runAllTimersAsync(); + + expect(closeAttempts).toBe(3); + } finally { + vi.useRealTimers(); + } + }); + + it("starts a new call without resuming the voice session being closed", async () => { + let createCount = 0; + let finishClose: (() => void) | undefined; + const closing = new Promise((resolve) => { + finishClose = resolve; + }); + const request = vi.fn(async (method: string) => { + if (method === "talk.client.create") { + createCount += 1; + return { + provider: "openai", + transport: "webrtc", + voiceSessionId: `voice-${createCount}`, + clientSecret: "secret", + }; + } + if (method === "talk.client.close") { + await closing; + } + return { ok: true }; + }); + const session = new RealtimeTalkSession({ request } as never, "agent:main:main"); + await session.start(); + + session.stop(); + await session.start(); + + const creates = request.mock.calls.filter(([method]) => method === "talk.client.create"); + expect(creates).toEqual([ + [ + "talk.client.create", + { sessionKey: "agent:main:main", capabilities: ["voice-transcript"] }, + requestTimeoutOptions, + ], + [ + "talk.client.create", + { sessionKey: "agent:main:main", capabilities: ["voice-transcript"] }, + requestTimeoutOptions, + ], + ]); + finishClose?.(); + await Promise.resolve(); + }); + + it("bounds active and draining client voice owners across session objects", async () => { + let createCount = 0; + const closes: Array>> = []; + const request = vi.fn(async (method: string) => { + if (method === "talk.client.create") { + createCount += 1; + return { + provider: "openai", + transport: "webrtc", + voiceSessionId: `voice-${createCount}`, + clientSecret: "secret", + }; + } + if (method === "talk.client.close") { + const close = createDeferred(); + closes.push(close); + await close.promise; + } + return { ok: true }; + }); + const client = { request } as never; + const first = new RealtimeTalkSession(client, "agent:main:main"); + const second = new RealtimeTalkSession(client, "agent:main:main"); + const third = new RealtimeTalkSession(client, "agent:main:main"); + + await first.start(); + first.stop(); + await vi.waitFor(() => expect(closes).toHaveLength(1)); + await second.start(); + second.stop(); + await vi.waitFor(() => expect(closes).toHaveLength(2)); + + await expect(third.start()).rejects.toThrow( + "Too many active or closing realtime Talk voice sessions", + ); + expect(createCount).toBe(2); + + closes[0]?.resolve(); + await vi.waitFor(async () => { + await third.start(); + expect(createCount).toBe(3); + }); + + third.stop(); + await vi.waitFor(() => expect(closes).toHaveLength(3)); + closes[1]?.resolve(); + closes[2]?.resolve(); + }); + + it("releases bounded startup owners after request deadlines", async () => { + vi.useFakeTimers(); + let failCreate = true; + try { + const request = vi.fn( + async (method: string, _params?: unknown, options?: { timeoutMs?: number }) => { + if (method !== "talk.client.create") { + return { ok: true }; + } + if (!failCreate) { + return { + provider: "openai", + transport: "webrtc", + voiceSessionId: "voice-recovered", + clientSecret: "secret", + }; + } + await new Promise((_resolve, reject) => { + setTimeout(() => reject(new Error("request timeout")), options?.timeoutMs); + }); + return { ok: true }; + }, + ); + const client = { request } as never; + const first = new RealtimeTalkSession(client, "agent:main:main", {}, { transport: "webrtc" }); + const second = new RealtimeTalkSession( + client, + "agent:main:main", + {}, + { transport: "webrtc" }, + ); + const third = new RealtimeTalkSession(client, "agent:main:main", {}, { transport: "webrtc" }); + + const startsSettled = Promise.allSettled([first.start(), second.start()]); + await Promise.resolve(); + await Promise.resolve(); + await expect(third.start()).rejects.toThrow( + "Too many active or closing realtime Talk voice sessions", + ); + expect(request.mock.calls.filter(([method]) => method === "talk.client.create")).toHaveLength( + 2, + ); + + await vi.advanceTimersByTimeAsync(30_000); + const settled = await startsSettled; + expect(settled.map((result) => result.status)).toEqual(["rejected", "rejected"]); + + failCreate = false; + await third.start(); + third.stop(); + } finally { + vi.useRealTimers(); + } + }); + + it("ignores final transcript callbacks emitted after shutdown begins", async () => { + const request = vi.fn(async (method: string) => { + if (method === "talk.client.create") { + return { + provider: "openai", + transport: "webrtc", + voiceSessionId: "voice-shutdown", + clientSecret: "secret", + }; + } + return { ok: true }; + }); + const session = new RealtimeTalkSession({ request } as never, "agent:main:main"); + await session.start(); + const context = transcriptContext(transportMock.webRtcContexts); + + session.stop(); + context.callbacks.onTranscript?.({ role: "user", text: "too late", final: true }); + await Promise.resolve(); + + expect(request.mock.calls.some(([method]) => method === "talk.client.transcript")).toBe(false); + }); + + it("drops a previous transport's delayed transcript after stop and restart", async () => { + let createCount = 0; + const request = vi.fn(async (method: string) => { + if (method === "talk.client.create") { + createCount += 1; + return { + provider: "openai", + transport: "webrtc", + voiceSessionId: `voice-${createCount}`, + clientSecret: "secret", + }; + } + return { ok: true }; + }); + const onTranscript = vi.fn(); + const session = new RealtimeTalkSession({ request } as never, "agent:main:main", { + onTranscript, + }); + await session.start(); + const previousContext = transcriptContext(transportMock.webRtcContexts); + + session.stop(); + await session.start(); + previousContext.callbacks.onTranscript?.({ + role: "user", + text: "stale transcript", + final: true, + }); + await Promise.resolve(); + + expect(onTranscript).not.toHaveBeenCalled(); + expect(request.mock.calls.some(([method]) => method === "talk.client.transcript")).toBe(false); + }); + + it("does not report Gateway relay transcripts through the client RPC", async () => { + const request = vi.fn(async (method: string) => { + if (method === "talk.client.create") { + return { + provider: "openai", + transport: "gateway-relay", + relaySessionId: "relay-voice", + 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(); + const context = transcriptContext(transportMock.relayContexts); + context.callbacks.onTranscript?.({ role: "user", text: "server owns this", final: true }); + await Promise.resolve(); + + expect(request.mock.calls.some(([method]) => method === "talk.client.transcript")).toBe(false); + session.stop(); + await Promise.resolve(); + expect(request.mock.calls.some(([method]) => method === "talk.client.close")).toBe(false); + }); +}); diff --git a/ui/src/pages/chat/realtime-talk-transcript-queue.test.ts b/ui/src/pages/chat/realtime-talk-transcript-queue.test.ts new file mode 100644 index 000000000000..4c3fd84dae60 --- /dev/null +++ b/ui/src/pages/chat/realtime-talk-transcript-queue.test.ts @@ -0,0 +1,251 @@ +// @vitest-environment node +import { beforeEach, describe, expect, it, vi } from "vitest"; +import type { RealtimeTalkTransportContext } from "./realtime-talk-shared.ts"; + +const transportMock = vi.hoisted(() => ({ + context: undefined as RealtimeTalkTransportContext | undefined, + start: vi.fn(async () => undefined), + stop: vi.fn(), +})); + +vi.mock("./realtime-talk-gateway-relay.ts", () => ({ + GatewayRelayRealtimeTalkTransport: vi.fn(), +})); +vi.mock("./realtime-talk-google-live.ts", () => ({ + GoogleLiveRealtimeTalkTransport: vi.fn(), +})); +vi.mock("./realtime-talk-webrtc.ts", () => ({ + WebRtcSdpRealtimeTalkTransport: vi.fn(function ( + _session: unknown, + context: RealtimeTalkTransportContext, + ) { + transportMock.context = context; + return { start: transportMock.start, stop: transportMock.stop }; + }), +})); + +import { RealtimeTalkSession } from "./realtime-talk.ts"; + +describe("RealtimeTalkSession transcript queue", () => { + beforeEach(() => { + transportMock.context = undefined; + transportMock.start.mockClear(); + transportMock.stop.mockClear(); + }); + + it("does not admit normalized-empty finals into a stalled transcript queue", async () => { + let releaseFirstTranscript!: () => void; + const firstTranscriptPending = new Promise((resolve) => { + releaseFirstTranscript = resolve; + }); + 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-empty-finals", + clientSecret: "secret", + }; + } + if (method === "talk.client.transcript") { + transcriptEntryIds.push(String(params?.entryId)); + if (params?.entryId === "1") { + await firstTranscriptPending; + } + } + return { ok: true }; + }); + const onStatus = vi.fn(); + const onTranscript = vi.fn(); + const session = new RealtimeTalkSession({ request } as never, "agent:main:main", { + onStatus, + onTranscript, + }); + await session.start(); + const context = transportMock.context; + const handleTranscript = context?.callbacks.onTranscript; + if (!context || !handleTranscript) { + throw new Error("expected realtime transcript callback"); + } + + handleTranscript({ role: "user", text: "first", final: true }); + for (let index = 0; index < 10_000; index += 1) { + handleTranscript({ role: "assistant", text: " ", final: true }); + } + for (let index = 0; index < 40; index += 1) { + handleTranscript({ role: "assistant", text: `queued-${index}`, final: true }); + } + + expect(transcriptEntryIds).toEqual(["1"]); + expect(transportMock.stop).not.toHaveBeenCalled(); + expect(onStatus.mock.calls.filter(([status]) => status === "error")).toHaveLength(0); + expect(onTranscript).toHaveBeenCalledTimes(10_041); + + releaseFirstTranscript(); + await context.flushTranscriptWrites?.(); + + expect(transcriptEntryIds).toEqual(Array.from({ length: 41 }, (_, index) => String(index + 1))); + session.stop(); + }); + + it("stops once when stalled transcript persistence exceeds its bounded queue", async () => { + vi.useFakeTimers(); + const warn = vi.spyOn(console, "warn").mockImplementation(() => undefined); + let releaseFirstTranscript!: () => void; + const firstTranscriptPending = new Promise((resolve) => { + releaseFirstTranscript = resolve; + }); + const transcriptEntryIds: string[] = []; + let firstAttempt = true; + try { + const request = vi.fn( + async (method: string, params?: { entryId?: string; text?: string }) => { + if (method === "talk.client.create") { + return { + provider: "openai", + transport: "webrtc", + voiceSessionId: "voice-overflow", + clientSecret: "secret", + }; + } + if (method === "talk.client.transcript") { + transcriptEntryIds.push(String(params?.entryId)); + if (params?.entryId === "1") { + if (firstAttempt) { + firstAttempt = false; + await firstTranscriptPending; + } + throw new Error("still unavailable"); + } + } + return { ok: true }; + }, + ); + const onStatus = vi.fn(); + const session = new RealtimeTalkSession({ request } as never, "agent:main:main", { + onStatus, + }); + await session.start(); + const onTranscript = transportMock.context?.callbacks.onTranscript; + if (!onTranscript) { + throw new Error("expected realtime transcript callback"); + } + + for (let index = 0; index < 10_000; index += 1) { + onTranscript({ + role: index % 2 === 0 ? "user" : "assistant", + text: ` ${"x".repeat(9_000)} `, + final: true, + }); + } + + expect(transcriptEntryIds).toEqual(["1"]); + expect(transportMock.stop).toHaveBeenCalledOnce(); + expect(onStatus.mock.calls.filter(([status]) => status === "error")).toHaveLength(1); + expect(warn).toHaveBeenCalledOnce(); + expect(request.mock.calls.filter(([method]) => method === "talk.client.close")).toHaveLength( + 0, + ); + + releaseFirstTranscript(); + await vi.advanceTimersByTimeAsync(2_500); + await vi.runAllTimersAsync(); + + expect(transcriptEntryIds).toEqual([ + "1", + "1", + "1", + ...Array.from({ length: 40 }, (_, index) => String(index + 2)), + ]); + expect( + request.mock.calls + .filter(([method]) => method === "talk.client.transcript") + .every(([, params]) => String(params?.text).length === 8_000), + ).toBe(true); + expect(request.mock.calls.filter(([method]) => method === "talk.client.close")).toHaveLength( + 1, + ); + expect(onStatus.mock.calls.filter(([status]) => status === "error")).toHaveLength(1); + expect(warn).toHaveBeenCalledTimes(2); + + onTranscript({ role: "user", text: "too late", final: true }); + session.stop(); + await Promise.resolve(); + expect(transcriptEntryIds).toHaveLength(43); + expect(transportMock.stop).toHaveBeenCalledOnce(); + } finally { + vi.useRealTimers(); + } + }); + + it("aborts a detached transcript drain before releasing its client owner", async () => { + vi.useFakeTimers(); + const transcriptSignals: AbortSignal[] = []; + try { + const request = vi.fn( + async ( + method: string, + _params?: unknown, + options?: { signal?: AbortSignal; timeoutMs?: number }, + ) => { + if (method === "talk.client.create") { + return { + provider: "openai", + transport: "webrtc", + voiceSessionId: "voice-drain-timeout", + clientSecret: "secret", + }; + } + if (method === "talk.client.transcript") { + const signal = options?.signal; + if (!signal) { + throw new Error("expected transcript abort signal"); + } + transcriptSignals.push(signal); + await new Promise((_resolve, reject) => { + signal.addEventListener("abort", () => reject(new Error("aborted")), { once: true }); + }); + } + return { ok: true }; + }, + ); + const onStatus = vi.fn(); + const session = new RealtimeTalkSession({ request } as never, "agent:main:main", { + onStatus, + }); + await session.start(); + const handleTranscript = transportMock.context?.callbacks.onTranscript; + if (!handleTranscript) { + throw new Error("expected realtime transcript callback"); + } + + handleTranscript({ role: "user", text: "stalled", final: true }); + for (let index = 0; index < 40; index += 1) { + handleTranscript({ role: "assistant", text: `queued-${index}`, final: true }); + } + session.stop(); + + expect(transcriptSignals).toHaveLength(1); + expect(transcriptSignals[0]?.aborted).toBe(false); + await vi.advanceTimersByTimeAsync(30_000); + await vi.runAllTimersAsync(); + + expect(transcriptSignals[0]?.aborted).toBe(true); + expect( + request.mock.calls.filter(([method]) => method === "talk.client.transcript"), + ).toHaveLength(1); + expect(request).toHaveBeenCalledWith( + "talk.client.close", + { + sessionKey: "agent:main:main", + voiceSessionId: "voice-drain-timeout", + }, + { timeoutMs: 30_000 }, + ); + expect(onStatus.mock.calls.filter(([status]) => status === "error")).toHaveLength(0); + } finally { + vi.useRealTimers(); + } + }); +}); diff --git a/ui/src/pages/chat/realtime-talk.test.ts b/ui/src/pages/chat/realtime-talk.test.ts index 34133d9f5ef5..5aa9b68775f6 100644 --- a/ui/src/pages/chat/realtime-talk.test.ts +++ b/ui/src/pages/chat/realtime-talk.test.ts @@ -39,6 +39,7 @@ type MockTransport = RealtimeTalkTransport & { ctx: RealtimeTalkTransportContext const googleInstances: MockTransport[] = []; const relayInstances: MockTransport[] = []; const webRtcInstances: MockTransport[] = []; +const requestTimeoutOptions = { timeoutMs: 30_000 }; function transportContext(transport: object | undefined): RealtimeTalkTransportContext { if (!transport) { @@ -129,10 +130,14 @@ describe("RealtimeTalkSession", () => { await session.start(); - expect(request).toHaveBeenCalledWith("talk.client.create", { - sessionKey: "main", - capabilities: ["voice-transcript"], - }); + expect(request).toHaveBeenCalledWith( + "talk.client.create", + { + sessionKey: "main", + capabilities: ["voice-transcript"], + }, + requestTimeoutOptions, + ); expect(googleInstances).toHaveLength(1); expect(googleStart).toHaveBeenCalledTimes(1); expect(webRtcInstances).toHaveLength(0); @@ -162,7 +167,8 @@ describe("RealtimeTalkSession", () => { transport: "webrtc-sdp", clientSecret: "secret", })); - const session = new RealtimeTalkSession({ request } as never, "main"); + const client = { request } as never; + const session = new RealtimeTalkSession(client, "main"); await session.start(); @@ -242,7 +248,11 @@ describe("RealtimeTalkSession", () => { const starting = session.start(); await vi.waitFor(() => - expect(request).toHaveBeenCalledWith("talk.client.create", expect.anything()), + expect(request).toHaveBeenCalledWith( + "talk.client.create", + expect.anything(), + requestTimeoutOptions, + ), ); session.stop(); create.resolve({ @@ -258,7 +268,11 @@ describe("RealtimeTalkSession", () => { }); await starting; - expect(request).toHaveBeenCalledWith("talk.session.close", { sessionId: "relay-stale" }); + expect(request).toHaveBeenCalledWith( + "talk.session.close", + { sessionId: "relay-stale" }, + { timeoutMs: 30_000 }, + ); expect(relayInstances).toHaveLength(0); }); @@ -275,7 +289,8 @@ describe("RealtimeTalkSession", () => { } throw new Error(`Unexpected request: ${method}`); }); - const session = new RealtimeTalkSession({ request } as never, "main"); + const client = { request } as never; + const session = new RealtimeTalkSession(client, "main"); const firstStart = session.start(); await vi.waitFor(() => expect(creates).toHaveLength(1)); @@ -289,6 +304,11 @@ describe("RealtimeTalkSession", () => { clientSecret: "secret", }); await secondStart; + const blocked = new RealtimeTalkSession(client, "main"); + await expect(blocked.start()).rejects.toThrow( + "Too many active or closing realtime Talk voice sessions", + ); + expect(creates).toHaveLength(2); creates[0]!.resolve({ provider: "openai", transport: "webrtc", @@ -298,10 +318,16 @@ describe("RealtimeTalkSession", () => { await firstStart; await vi.waitFor(() => - expect(request).toHaveBeenCalledWith("talk.client.close", { - sessionKey: "main", - voiceSessionId: "voice-stale", - }), + expect(request).toHaveBeenCalledWith( + "talk.client.close", + { + sessionKey: "main", + voiceSessionId: "voice-stale", + }, + { + timeoutMs: 30_000, + }, + ), ); expect(webRtcInstances).toHaveLength(1); expect(webRtcStart).toHaveBeenCalledTimes(1); @@ -334,19 +360,29 @@ describe("RealtimeTalkSession", () => { await session.start(); - expect(request).toHaveBeenNthCalledWith(1, "talk.client.create", { - sessionKey: "main", - provider: "xai", - transport: "gateway-relay", - capabilities: ["voice-transcript"], - }); - expect(request).toHaveBeenNthCalledWith(2, "talk.session.create", { - sessionKey: "main", - provider: "xai", - transport: "gateway-relay", - mode: "realtime", - brain: "agent-consult", - }); + expect(request).toHaveBeenNthCalledWith( + 1, + "talk.client.create", + { + sessionKey: "main", + provider: "xai", + transport: "gateway-relay", + capabilities: ["voice-transcript"], + }, + requestTimeoutOptions, + ); + expect(request).toHaveBeenNthCalledWith( + 2, + "talk.session.create", + { + sessionKey: "main", + provider: "xai", + transport: "gateway-relay", + mode: "realtime", + brain: "agent-consult", + }, + requestTimeoutOptions, + ); expect(relayInstances).toHaveLength(1); expect(relayStart).toHaveBeenCalledTimes(1); }); @@ -389,19 +425,29 @@ describe("RealtimeTalkSession", () => { await session.start(); - expect(request).toHaveBeenNthCalledWith(2, "talk.client.create", { - sessionKey: "main", - provider: "openai", - transport: "gateway-relay", - capabilities: ["voice-transcript", "camera-frame"], - }); - expect(request).toHaveBeenNthCalledWith(3, "talk.session.create", { - sessionKey: "main", - provider: "openai", - transport: "gateway-relay", - mode: "realtime", - brain: "agent-consult", - }); + expect(request).toHaveBeenNthCalledWith( + 2, + "talk.client.create", + { + sessionKey: "main", + provider: "openai", + transport: "gateway-relay", + capabilities: ["voice-transcript", "camera-frame"], + }, + requestTimeoutOptions, + ); + expect(request).toHaveBeenNthCalledWith( + 3, + "talk.session.create", + { + sessionKey: "main", + provider: "openai", + transport: "gateway-relay", + mode: "realtime", + brain: "agent-consult", + }, + requestTimeoutOptions, + ); expect(onVideoCapability).toHaveBeenCalledOnce(); expect(onVideoCapability).toHaveBeenCalledWith(false); expect(relayInstances).toHaveLength(1); @@ -452,18 +498,22 @@ describe("RealtimeTalkSession", () => { await session.start(); - expect(request).toHaveBeenCalledWith("talk.client.create", { - sessionKey: "main", - provider: "openai", - model: "gpt-realtime-2", - voice: "marin", - transport: "webrtc", - vadThreshold: 0.45, - silenceDurationMs: 650, - prefixPaddingMs: 250, - reasoningEffort: "low", - capabilities: ["voice-transcript"], - }); + expect(request).toHaveBeenCalledWith( + "talk.client.create", + { + sessionKey: "main", + provider: "openai", + model: "gpt-realtime-2", + voice: "marin", + transport: "webrtc", + vadThreshold: 0.45, + silenceDurationMs: 650, + prefixPaddingMs: 250, + reasoningEffort: "low", + capabilities: ["voice-transcript"], + }, + requestTimeoutOptions, + ); expect(transportContext(webRtcInstances[0])).toEqual( expect.objectContaining({ inputDeviceId: "usb-mic", videoDeviceId: "desk-camera" }), ); @@ -493,11 +543,16 @@ describe("RealtimeTalkSession", () => { await session.start(); - expect(request).toHaveBeenNthCalledWith(1, "talk.catalog", {}); - expect(request).toHaveBeenNthCalledWith(2, "talk.client.create", { - sessionKey: "main", - capabilities: ["voice-transcript", "camera-frame"], - }); + expect(request).toHaveBeenNthCalledWith(1, "talk.catalog", {}, requestTimeoutOptions); + expect(request).toHaveBeenNthCalledWith( + 2, + "talk.client.create", + { + sessionKey: "main", + capabilities: ["voice-transcript", "camera-frame"], + }, + requestTimeoutOptions, + ); expect(onVideoCapability).toHaveBeenCalledWith(true); expect(transportContext(webRtcInstances[0])).toEqual( expect.not.objectContaining({ videoEnabled: expect.anything() }), @@ -604,10 +659,15 @@ describe("RealtimeTalkSession", () => { await session.start(); - expect(request).toHaveBeenNthCalledWith(2, "talk.client.create", { - sessionKey: "main", - capabilities: ["voice-transcript"], - }); + expect(request).toHaveBeenNthCalledWith( + 2, + "talk.client.create", + { + sessionKey: "main", + capabilities: ["voice-transcript"], + }, + requestTimeoutOptions, + ); expect(onVideoCapability).toHaveBeenCalledWith(false); }); @@ -633,8 +693,12 @@ describe("RealtimeTalkSession", () => { await expect(session.start()).rejects.toBe(clientError); expect(request.mock.calls).toEqual([ - ["talk.client.create", { sessionKey: "main", capabilities: ["voice-transcript"] }], - ["talk.config", {}], + [ + "talk.client.create", + { sessionKey: "main", capabilities: ["voice-transcript"] }, + requestTimeoutOptions, + ], + ["talk.config", {}, requestTimeoutOptions], ]); expect(relayInstances).toHaveLength(0); }); @@ -672,12 +736,17 @@ describe("RealtimeTalkSession", () => { await session.start(); - expect(request).toHaveBeenNthCalledWith(3, "talk.session.create", { - sessionKey: "main", - mode: "realtime", - transport: "gateway-relay", - brain: "agent-consult", - }); + expect(request).toHaveBeenNthCalledWith( + 3, + "talk.session.create", + { + sessionKey: "main", + mode: "realtime", + transport: "gateway-relay", + brain: "agent-consult", + }, + requestTimeoutOptions, + ); expect(relayInstances).toHaveLength(1); expect(relayStart).toHaveBeenCalledTimes(1); }); @@ -709,12 +778,17 @@ describe("RealtimeTalkSession", () => { await session.start(); - expect(request).toHaveBeenNthCalledWith(3, "talk.session.create", { - sessionKey: "main", - mode: "realtime", - transport: "gateway-relay", - brain: "agent-consult", - }); + expect(request).toHaveBeenNthCalledWith( + 3, + "talk.session.create", + { + sessionKey: "main", + mode: "realtime", + transport: "gateway-relay", + brain: "agent-consult", + }, + requestTimeoutOptions, + ); expect(relayInstances).toHaveLength(1); }); @@ -734,8 +808,12 @@ describe("RealtimeTalkSession", () => { await expect(session.start()).rejects.toBe(clientError); expect(request.mock.calls).toEqual([ - ["talk.client.create", { sessionKey: "main", capabilities: ["voice-transcript"] }], - ["talk.config", {}], + [ + "talk.client.create", + { sessionKey: "main", capabilities: ["voice-transcript"] }, + requestTimeoutOptions, + ], + ["talk.config", {}, requestTimeoutOptions], ]); expect(relayInstances).toHaveLength(0); }); @@ -756,338 +834,13 @@ describe("RealtimeTalkSession", () => { await expect(session.start()).rejects.toBe(clientError); expect(request.mock.calls).toEqual([ - ["talk.client.create", { sessionKey: "main", capabilities: ["voice-transcript"] }], - ["talk.config", {}], + [ + "talk.client.create", + { sessionKey: "main", capabilities: ["voice-transcript"] }, + requestTimeoutOptions, + ], + ["talk.config", {}, requestTimeoutOptions], ]); expect(relayInstances).toHaveLength(0); }); - - it("retries finalized transcript writes in order", async () => { - vi.useFakeTimers(); - try { - const transcriptEntryIds: string[] = []; - let firstAttempt = true; - const request = vi.fn(async (method: string, params?: { entryId?: string }) => { - if (method === "talk.client.create") { - return { - provider: "openai", - transport: "webrtc", - voiceSessionId: "voice-queue", - clientSecret: "secret", - }; - } - if (method === "talk.client.transcript") { - transcriptEntryIds.push(String(params?.entryId)); - if (params?.entryId === "1" && firstAttempt) { - firstAttempt = false; - throw new Error("temporary failure"); - } - return { ok: true }; - } - return { ok: true }; - }); - const session = new RealtimeTalkSession({ request } as never, "agent:main:main"); - await session.start(); - const context = transportContext(webRtcInstances[0]) as { - callbacks: { - onTranscript?: (entry: { - role: "user" | "assistant"; - text: string; - final: boolean; - }) => void; - }; - }; - context.callbacks.onTranscript?.({ role: "user", text: "first", final: true }); - context.callbacks.onTranscript?.({ role: "assistant", text: "second", final: true }); - - await vi.advanceTimersByTimeAsync(500); - await vi.waitFor(() => expect(transcriptEntryIds).toEqual(["1", "1", "2"])); - session.stop(); - await vi.runAllTimersAsync(); - } finally { - vi.useRealTimers(); - } - }); - - it("continues entry ids when the same voice session replaces its transport", 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-resume", - 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 firstContext = transportContext(webRtcInstances[0]) as { - callbacks: { - onTranscript?: (entry: { - role: "user" | "assistant"; - text: string; - final: boolean; - }) => void; - }; - flushTranscriptWrites?: () => Promise; - }; - firstContext.callbacks.onTranscript?.({ role: "user", text: "first", final: true }); - await firstContext.flushTranscriptWrites?.(); - - await session.start(); - const secondContext = transportContext(webRtcInstances[1]) as typeof firstContext; - secondContext.callbacks.onTranscript?.({ role: "assistant", text: "second", final: true }); - await secondContext.flushTranscriptWrites?.(); - - expect(transcriptEntryIds).toEqual(["1", "2"]); - expect(request.mock.calls.filter(([method]) => method === "talk.client.create")).toEqual([ - ["talk.client.create", { sessionKey: "agent:main:main", capabilities: ["voice-transcript"] }], - [ - "talk.client.create", - { - sessionKey: "agent:main:main", - voiceSessionId: "voice-resume", - capabilities: ["voice-transcript"], - }, - ], - ]); - session.stop(); - await Promise.resolve(); - }); - - it("surfaces transcript failure after three attempts", async () => { - vi.useFakeTimers(); - const warn = vi.spyOn(console, "warn").mockImplementation(() => undefined); - try { - const request = vi.fn(async (method: string) => { - if (method === "talk.client.create") { - return { - provider: "openai", - transport: "webrtc", - voiceSessionId: "voice-failure", - clientSecret: "secret", - }; - } - if (method === "talk.client.transcript") { - throw new Error("still unavailable"); - } - return { ok: true }; - }); - const onStatus = vi.fn(); - const session = new RealtimeTalkSession({ request } as never, "agent:main:main", { - onStatus, - }); - await session.start(); - const context = transportContext(webRtcInstances[0]) as { - callbacks: { - onTranscript?: (entry: { - role: "user" | "assistant"; - text: string; - final: boolean; - }) => void; - }; - }; - context.callbacks.onTranscript?.({ role: "user", text: "save me", final: true }); - - await vi.advanceTimersByTimeAsync(2_500); - await vi.waitFor(() => - expect(onStatus).toHaveBeenCalledWith( - "error", - expect.stringContaining("Voice transcript could not be saved"), - ), - ); - expect( - request.mock.calls.filter(([method]) => method === "talk.client.transcript"), - ).toHaveLength(3); - expect(warn).toHaveBeenCalled(); - session.stop(); - await vi.runAllTimersAsync(); - } finally { - warn.mockRestore(); - vi.useRealTimers(); - } - }); - - it("retries logical voice-session close after transient failures", async () => { - vi.useFakeTimers(); - try { - let closeAttempts = 0; - const request = vi.fn(async (method: string) => { - if (method === "talk.client.create") { - return { - provider: "openai", - transport: "webrtc", - voiceSessionId: "voice-close-retry", - clientSecret: "secret", - }; - } - if (method === "talk.client.close" && ++closeAttempts < 3) { - throw new Error("temporary close failure"); - } - return { ok: true }; - }); - const session = new RealtimeTalkSession({ request } as never, "agent:main:main"); - await session.start(); - - session.stop(); - await vi.runAllTimersAsync(); - - expect(closeAttempts).toBe(3); - } finally { - vi.useRealTimers(); - } - }); - - it("starts a new call without resuming the voice session being closed", async () => { - let createCount = 0; - let finishClose: (() => void) | undefined; - const closing = new Promise((resolve) => { - finishClose = resolve; - }); - const request = vi.fn(async (method: string) => { - if (method === "talk.client.create") { - createCount += 1; - return { - provider: "openai", - transport: "webrtc", - voiceSessionId: `voice-${createCount}`, - clientSecret: "secret", - }; - } - if (method === "talk.client.close") { - await closing; - } - return { ok: true }; - }); - const session = new RealtimeTalkSession({ request } as never, "agent:main:main"); - await session.start(); - - session.stop(); - await session.start(); - - const creates = request.mock.calls.filter(([method]) => method === "talk.client.create"); - expect(creates).toEqual([ - ["talk.client.create", { sessionKey: "agent:main:main", capabilities: ["voice-transcript"] }], - ["talk.client.create", { sessionKey: "agent:main:main", capabilities: ["voice-transcript"] }], - ]); - finishClose?.(); - await Promise.resolve(); - }); - - it("ignores final transcript callbacks emitted after shutdown begins", async () => { - const request = vi.fn(async (method: string) => { - if (method === "talk.client.create") { - return { - provider: "openai", - transport: "webrtc", - voiceSessionId: "voice-shutdown", - clientSecret: "secret", - }; - } - return { ok: true }; - }); - const session = new RealtimeTalkSession({ request } as never, "agent:main:main"); - await session.start(); - const context = transportContext(webRtcInstances[0]) as { - callbacks: { - onTranscript?: (entry: { - role: "user" | "assistant"; - text: string; - final: boolean; - }) => void; - }; - }; - - session.stop(); - context.callbacks.onTranscript?.({ role: "user", text: "too late", final: true }); - await Promise.resolve(); - - expect(request.mock.calls.some(([method]) => method === "talk.client.transcript")).toBe(false); - }); - - it("drops a previous transport's delayed transcript after stop and restart", async () => { - let createCount = 0; - const request = vi.fn(async (method: string) => { - if (method === "talk.client.create") { - createCount += 1; - return { - provider: "openai", - transport: "webrtc", - voiceSessionId: `voice-${createCount}`, - clientSecret: "secret", - }; - } - return { ok: true }; - }); - const onTranscript = vi.fn(); - const session = new RealtimeTalkSession({ request } as never, "agent:main:main", { - onTranscript, - }); - await session.start(); - const previousContext = transportContext(webRtcInstances[0]) as { - callbacks: { - onTranscript?: (entry: { - role: "user" | "assistant"; - text: string; - final: boolean; - }) => void; - }; - }; - - session.stop(); - await session.start(); - previousContext.callbacks.onTranscript?.({ - role: "user", - text: "stale transcript", - final: true, - }); - await Promise.resolve(); - - expect(onTranscript).not.toHaveBeenCalled(); - expect(request.mock.calls.some(([method]) => method === "talk.client.transcript")).toBe(false); - }); - - it("does not report Gateway relay transcripts through the client RPC", async () => { - const request = vi.fn(async (method: string) => { - if (method === "talk.client.create") { - return { - provider: "openai", - transport: "gateway-relay", - relaySessionId: "relay-voice", - 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(); - const context = transportContext(relayInstances[0]) as { - callbacks: { - onTranscript?: (entry: { - role: "user" | "assistant"; - text: string; - final: boolean; - }) => void; - }; - }; - context.callbacks.onTranscript?.({ role: "user", text: "server owns this", final: true }); - await Promise.resolve(); - - expect(request.mock.calls.some(([method]) => method === "talk.client.transcript")).toBe(false); - session.stop(); - await Promise.resolve(); - expect(request.mock.calls.some(([method]) => method === "talk.client.close")).toBe(false); - }); }); diff --git a/ui/src/pages/chat/realtime-talk.ts b/ui/src/pages/chat/realtime-talk.ts index 20633051545c..9bff063598f4 100644 --- a/ui/src/pages/chat/realtime-talk.ts +++ b/ui/src/pages/chat/realtime-talk.ts @@ -1,6 +1,12 @@ // Control UI chat module implements realtime talk behavior. +import { DEFAULT_GATEWAY_REQUEST_TIMEOUT_MS } from "@openclaw/gateway-client/browser"; import type { TalkCatalogResult } from "@openclaw/gateway-protocol"; +import type { BoundedSerialQueue } from "../../../../src/shared/bounded-serial-queue.js"; import { normalizeTalkTransport } from "../../../../src/talk/talk-session-controller.js"; +import { + normalizeVoiceTranscriptText, + VOICE_TRANSCRIPT_QUEUE_POLICY, +} from "../../../../src/talk/voice-transcript.js"; import type { GatewayBrowserClient } from "../../api/gateway.ts"; import { GatewayRelayRealtimeTalkTransport } from "./realtime-talk-gateway-relay.ts"; import { GoogleLiveRealtimeTalkTransport } from "./realtime-talk-google-live.ts"; @@ -62,7 +68,8 @@ type DetachedVoiceSession = { voiceSessionId: string; serverOwned: boolean; generation?: number; - transcriptWrites: Promise; + transcriptQueue: BoundedSerialQueue; + owner?: ClientVoiceSessionOwner; }; type RealtimeTalkConfigResult = { @@ -75,6 +82,79 @@ type RealtimeTalkConfigResult = { }; }; +type ClientVoiceSessionOwner = { + signal: AbortSignal; + abort: () => void; + release: () => void; +}; + +const MAX_CLIENT_VOICE_SESSION_OWNERS = 2; +const CLIENT_VOICE_TRANSCRIPT_DRAIN_TIMEOUT_MS = DEFAULT_GATEWAY_REQUEST_TIMEOUT_MS; +// Count the active call and one retiring replacement across session objects. +// The token stays owned through durable close so rapid restarts cannot orphan drains. +const clientVoiceSessionOwnerCounts = new WeakMap>(); + +function reserveClientVoiceSessionOwner( + client: GatewayBrowserClient, + sessionKey: string, +): ClientVoiceSessionOwner { + let counts = clientVoiceSessionOwnerCounts.get(client); + if (!counts) { + counts = new Map(); + clientVoiceSessionOwnerCounts.set(client, counts); + } + const count = counts.get(sessionKey) ?? 0; + if (count >= MAX_CLIENT_VOICE_SESSION_OWNERS) { + throw new Error("Too many active or closing realtime Talk voice sessions"); + } + counts.set(sessionKey, count + 1); + const ownerCounts = counts; + const controller = new AbortController(); + let released = false; + return { + signal: controller.signal, + abort: () => controller.abort(), + release: () => { + if (released) { + return; + } + released = true; + const nextCount = (ownerCounts.get(sessionKey) ?? 1) - 1; + if (nextCount > 0) { + ownerCounts.set(sessionKey, nextCount); + } else { + ownerCounts.delete(sessionKey); + } + }, + }; +} + +function transcriptPersistenceAbortError(): Error { + const error = new Error("voice transcript persistence aborted"); + error.name = "AbortError"; + return error; +} + +async function waitForTranscriptRetry(delayMs: number, signal: AbortSignal): Promise { + if (signal.aborted) { + throw transcriptPersistenceAbortError(); + } + if (delayMs <= 0) { + return; + } + await new Promise((resolve, reject) => { + const timer = setTimeout(() => { + signal.removeEventListener("abort", onAbort); + resolve(); + }, delayMs); + const onAbort = () => { + clearTimeout(timer); + reject(transcriptPersistenceAbortError()); + }; + signal.addEventListener("abort", onAbort, { once: true }); + }); +} + function normalizeLaunchTransport(value: unknown): RealtimeTalkLaunchTransport | undefined { if (typeof value !== "string") { return undefined; @@ -145,7 +225,8 @@ export class RealtimeTalkSession { private readonly transcriptSeqByVoiceSessionId = new Map(); private acceptingTranscripts = false; private serverOwnedVoiceSession = false; - private transcriptWrites: Promise = Promise.resolve(); + private transcriptQueue = VOICE_TRANSCRIPT_QUEUE_POLICY.createQueue(); + private clientVoiceSessionOwner: ClientVoiceSessionOwner | undefined; constructor( private readonly client: GatewayBrowserClient, @@ -159,58 +240,93 @@ export class RealtimeTalkSession { const lifecycleGeneration = ++this.lifecycleGeneration; this.closed = false; this.callbacks.onStatus?.("connecting"); - const providerVideoCapable = await this.resolveVideoCapability(); - if (this.closed || lifecycleGeneration !== this.lifecycleGeneration) { - return; + const existingVoiceSessionId = this.voiceSessionId; + const existingOwner = this.clientVoiceSessionOwner; + const owner = reserveClientVoiceSessionOwner(this.client, this.sessionKey); + let ownerTransferred = false; + try { + const providerVideoCapable = await this.resolveVideoCapability(); + if (this.closed || lifecycleGeneration !== this.lifecycleGeneration) { + return; + } + // Declaring voice-transcript arms the server-side spoken-confirmation gate; + // this client reports every finalized utterance, so the gate is completable. + const capabilities: Array<"camera-frame" | "voice-transcript"> = ["voice-transcript"]; + if (providerVideoCapable) { + capabilities.push("camera-frame"); + } + const session = await this.createSession({ ...this.options, capabilities }); + const transport = resolveTransport(session); + // Managed-room stays unsupported here and carries no voice bookkeeping; + // reject it before the voice-session requirement produces a misleading error. + if (transport === "managed-room") { + throw new Error("Managed-room realtime Talk sessions are not available in this UI yet"); + } + const voiceSessionId = + session.voiceSessionId ?? + (transport === "gateway-relay" + ? (session as RealtimeTalkGatewayRelaySessionResult).relaySessionId + : undefined); + if (!voiceSessionId) { + throw new Error("Realtime Talk session did not return a voice session id"); + } + if (this.closed || lifecycleGeneration !== this.lifecycleGeneration) { + this.closeUnadoptedVoiceSession(voiceSessionId, transport, owner); + ownerTransferred = true; + return; + } + if ( + existingOwner && + (transport === "gateway-relay" || voiceSessionId !== existingVoiceSessionId) + ) { + this.closeUnadoptedVoiceSession(voiceSessionId, transport, owner); + ownerTransferred = true; + throw new Error("Realtime Talk replacement changed the active voice session"); + } + const adoptedOwner = + existingOwner && voiceSessionId === existingVoiceSessionId ? existingOwner : owner; + if (adoptedOwner !== owner) { + owner.release(); + } + this.voiceSessionId = voiceSessionId; + this.acceptingTranscripts = true; + this.serverOwnedVoiceSession = transport === "gateway-relay"; + this.transportGeneration += 1; + if (this.serverOwnedVoiceSession) { + owner.release(); + } else { + this.clientVoiceSessionOwner = adoptedOwner; + } + ownerTransferred = true; + const callbacks = + transport === "gateway-relay" + ? this.callbacks + : this.clientOwnedTranscriptCallbacks( + voiceSessionId, + this.transportGeneration, + adoptedOwner.signal, + ); + const transcriptQueue = this.transcriptQueue; + this.transport = createTransport(session, { + client: this.client, + sessionKey: this.sessionKey, + voiceSessionId, + flushTranscriptWrites: async () => await transcriptQueue.flush(), + callbacks, + inputDeviceId: this.localOptions.inputDeviceId, + videoDeviceId: this.localOptions.videoDeviceId, + consultThinkingLevel: session.consultThinkingLevel, + consultFastMode: session.consultFastMode, + }); + this.callbacks.onVideoCapability?.( + providerVideoCapable && typeof this.transport.setVideoEnabled === "function", + ); + await this.transport.start(); + } finally { + if (!ownerTransferred) { + owner.release(); + } } - // Declaring voice-transcript arms the server-side spoken-confirmation gate; - // this client reports every finalized utterance, so the gate is completable. - const capabilities: Array<"camera-frame" | "voice-transcript"> = ["voice-transcript"]; - if (providerVideoCapable) { - capabilities.push("camera-frame"); - } - const session = await this.createSession({ ...this.options, capabilities }); - const transport = resolveTransport(session); - // Managed-room stays unsupported here and carries no voice bookkeeping; - // reject it before the voice-session requirement produces a misleading error. - if (transport === "managed-room") { - throw new Error("Managed-room realtime Talk sessions are not available in this UI yet"); - } - const voiceSessionId = - session.voiceSessionId ?? - (transport === "gateway-relay" - ? (session as RealtimeTalkGatewayRelaySessionResult).relaySessionId - : undefined); - if (!voiceSessionId) { - throw new Error("Realtime Talk session did not return a voice session id"); - } - if (this.closed || lifecycleGeneration !== this.lifecycleGeneration) { - this.closeUnadoptedVoiceSession(voiceSessionId, transport); - return; - } - this.voiceSessionId = voiceSessionId; - this.acceptingTranscripts = true; - this.serverOwnedVoiceSession = transport === "gateway-relay"; - this.transportGeneration += 1; - const callbacks = - transport === "gateway-relay" - ? this.callbacks - : this.clientOwnedTranscriptCallbacks(voiceSessionId, this.transportGeneration); - this.transport = createTransport(session, { - client: this.client, - sessionKey: this.sessionKey, - voiceSessionId, - flushTranscriptWrites: async () => await this.transcriptWrites, - callbacks, - inputDeviceId: this.localOptions.inputDeviceId, - videoDeviceId: this.localOptions.videoDeviceId, - consultThinkingLevel: session.consultThinkingLevel, - consultFastMode: session.consultFastMode, - }); - this.callbacks.onVideoCapability?.( - providerVideoCapable && typeof this.transport.setVideoEnabled === "function", - ); - await this.transport.start(); } private async resolveVideoCapability(): Promise { @@ -218,7 +334,11 @@ export class RealtimeTalkSession { return false; } try { - const catalog = await this.client.request("talk.catalog", {}); + const catalog = await this.client.request( + "talk.catalog", + {}, + { timeoutMs: DEFAULT_GATEWAY_REQUEST_TIMEOUT_MS }, + ); const selectedProvider = this.options.provider ?? catalog.realtime.activeProvider; if (!selectedProvider) { return false; @@ -248,13 +368,18 @@ export class RealtimeTalkSession { voiceSessionId: this.voiceSessionId, ...launchOptions, }), + { timeoutMs: DEFAULT_GATEWAY_REQUEST_TIMEOUT_MS }, ); } catch (error) { let transport = launchOptions.transport; if (!transport) { let result: RealtimeTalkConfigResult; try { - result = await this.client.request("talk.config", {}); + result = await this.client.request( + "talk.config", + {}, + { timeoutMs: DEFAULT_GATEWAY_REQUEST_TIMEOUT_MS }, + ); } catch { throw error; } @@ -284,6 +409,7 @@ export class RealtimeTalkSession { transport: transport ?? "gateway-relay", brain: "agent-consult", }), + { timeoutMs: DEFAULT_GATEWAY_REQUEST_TIMEOUT_MS }, ); return resolveTransport(relaySession) === "gateway-relay" ? { @@ -313,25 +439,38 @@ export class RealtimeTalkSession { } } - private closeUnadoptedVoiceSession(voiceSessionId: string, transport: string): void { + private closeUnadoptedVoiceSession( + voiceSessionId: string, + transport: string, + owner: ClientVoiceSessionOwner, + ): void { // A stopped or superseded create still owns the allocation returned to it. // Close at the provider boundary without installing a stale transport. if (transport === "gateway-relay") { void this.client - .request("talk.session.close", { sessionId: voiceSessionId }) - .catch(() => undefined); + .request( + "talk.session.close", + { sessionId: voiceSessionId }, + { timeoutMs: DEFAULT_GATEWAY_REQUEST_TIMEOUT_MS }, + ) + .catch(() => undefined) + .finally(owner.release); return; } + const transcriptQueue = VOICE_TRANSCRIPT_QUEUE_POLICY.createQueue(); + transcriptQueue.seal(); this.closeLogicalVoiceSession({ voiceSessionId, serverOwned: false, - transcriptWrites: Promise.resolve(), + transcriptQueue, + owner, }); } private clientOwnedTranscriptCallbacks( owningVoiceSessionId: string, owningGeneration: number, + transcriptSignal: AbortSignal, ): RealtimeTalkCallbacks { return { ...this.callbacks, @@ -350,61 +489,112 @@ export class RealtimeTalkSession { if (entry.final) { const transcriptSeq = (this.transcriptSeqByVoiceSessionId.get(owningVoiceSessionId) ?? 0) + 1; - this.transcriptSeqByVoiceSessionId.set(owningVoiceSessionId, transcriptSeq); const entryId = String(transcriptSeq); - // One promise tail preserves transcript order and makes consult flushes - // observe the same dedupe sequence used by close. - const write = this.transcriptWrites.then(async () => { - await this.writeTranscriptWithRetry({ - voiceSessionId: owningVoiceSessionId, - entryId, - role: entry.role, - text: entry.text, - }); - }); - this.transcriptWrites = write.catch((error: unknown) => { - // The utterance exists only in client memory; after retries and surfacing the error, - // keeping the record open cannot recover it, while server entryId dedupe preserves order. - // Deferring close would only shift the identical loss to the 6h stale sweep. - const detail = `Voice transcript could not be saved: ${error instanceof Error ? error.message : String(error)}`; - console.warn(detail, error); - // Only surface to the user if this transport is still the active one; a - // retired call's late failure must not error a healthy replacement call. - if (this.transportGeneration === owningGeneration) { - this.callbacks.onStatus?.("error", detail); + const role = entry.role; + const text = normalizeVoiceTranscriptText(entry.text); + if (text) { + const admission = this.transcriptQueue.enqueue( + async () => + await this.writeTranscriptWithRetry({ + voiceSessionId: owningVoiceSessionId, + entryId, + role, + text, + signal: transcriptSignal, + }), + { weight: text.length }, + ); + if (!admission.accepted) { + if (admission.reason === "overflow") { + this.failTranscriptPersistence(owningGeneration); + } + return; } - }); + this.transcriptSeqByVoiceSessionId.set(owningVoiceSessionId, transcriptSeq); + void admission.completion.catch((error: unknown) => { + if (transcriptSignal.aborted) { + return; + } + // The utterance exists only in client memory; after retries and surfacing the error, + // keeping the record open cannot recover it, while server entryId dedupe preserves order. + // Deferring close would only shift the identical loss to the 6h stale sweep. + const detail = `Voice transcript could not be saved: ${error instanceof Error ? error.message : String(error)}`; + console.warn(detail, error); + // Only surface to the user if this transport is still the active one; a + // retired call's late failure must not error a healthy replacement call. + if (this.transportGeneration === owningGeneration) { + this.callbacks.onStatus?.("error", detail); + } + }); + } } this.callbacks.onTranscript?.(entry); }, }; } + private failTranscriptPersistence(owningGeneration: number): void { + if ( + this.transportGeneration !== owningGeneration || + !this.acceptingTranscripts || + !this.voiceSessionId + ) { + return; + } + this.lifecycleGeneration += 1; + this.closed = true; + this.videoOperation += 1; + this.videoEnabled = false; + activeRealtimeTalkSessions.delete(this); + const detached = this.detachVoiceSession(); + // Retire the overflowing transport before accepted-write and close failures + // settle so the first terminal persistence error keeps precedence. + this.transportGeneration += 1; + this.transport?.stop(); + this.transport = null; + console.warn(VOICE_TRANSCRIPT_QUEUE_POLICY.overflowMessage); + this.callbacks.onStatus?.("error", VOICE_TRANSCRIPT_QUEUE_POLICY.overflowMessage); + if (detached) { + this.closeLogicalVoiceSession(detached); + } + } + private async writeTranscriptWithRetry(params: { voiceSessionId: string; entryId: string; role: "user" | "assistant"; text: string; + signal: AbortSignal; }): Promise { const retryDelaysMs = [0, 500, 2_000]; let lastError: unknown; for (const delayMs of retryDelaysMs) { if (delayMs > 0) { - await new Promise((resolve) => { - setTimeout(resolve, delayMs); - }); + await waitForTranscriptRetry(delayMs, params.signal); + } else if (params.signal.aborted) { + throw transcriptPersistenceAbortError(); } try { - await this.client.request("talk.client.transcript", { - sessionKey: this.sessionKey, - voiceSessionId: params.voiceSessionId, - entryId: params.entryId, - role: params.role, - text: params.text, - timestamp: Date.now(), - }); + await this.client.request( + "talk.client.transcript", + { + sessionKey: this.sessionKey, + voiceSessionId: params.voiceSessionId, + entryId: params.entryId, + role: params.role, + text: params.text, + timestamp: Date.now(), + }, + { + signal: params.signal, + timeoutMs: DEFAULT_GATEWAY_REQUEST_TIMEOUT_MS, + }, + ); return; } catch (error) { + if (params.signal.aborted) { + throw transcriptPersistenceAbortError(); + } lastError = error; } } @@ -420,20 +610,30 @@ export class RealtimeTalkSession { voiceSessionId, serverOwned: this.serverOwnedVoiceSession, generation: this.transportGeneration, - transcriptWrites: this.transcriptWrites, + transcriptQueue: this.transcriptQueue, + owner: this.clientVoiceSessionOwner, } satisfies DetachedVoiceSession; + detached.transcriptQueue.seal(); + this.transcriptSeqByVoiceSessionId.delete(voiceSessionId); this.voiceSessionId = undefined; this.acceptingTranscripts = false; this.serverOwnedVoiceSession = false; - this.transcriptWrites = Promise.resolve(); + this.transcriptQueue = VOICE_TRANSCRIPT_QUEUE_POLICY.createQueue(); + this.clientVoiceSessionOwner = undefined; return detached; } private closeLogicalVoiceSession(detached: DetachedVoiceSession): void { if (detached.serverOwned) { + detached.owner?.release(); return; } - void detached.transcriptWrites + const drainTimer = setTimeout( + () => detached.owner?.abort(), + CLIENT_VOICE_TRANSCRIPT_DRAIN_TIMEOUT_MS, + ); + void detached.transcriptQueue + .flush() .then(async () => { let lastError: unknown; for (const delayMs of [0, 500, 2_000]) { @@ -443,10 +643,14 @@ export class RealtimeTalkSession { }); } try { - await this.client.request("talk.client.close", { - sessionKey: this.sessionKey, - voiceSessionId: detached.voiceSessionId, - }); + await this.client.request( + "talk.client.close", + { + sessionKey: this.sessionKey, + voiceSessionId: detached.voiceSessionId, + }, + { timeoutMs: DEFAULT_GATEWAY_REQUEST_TIMEOUT_MS }, + ); return; } catch (error) { lastError = error; @@ -461,6 +665,10 @@ export class RealtimeTalkSession { if (this.transportGeneration === detached.generation) { this.callbacks.onStatus?.("error", "Realtime Talk voice session close failed"); } + }) + .finally(() => { + clearTimeout(drainTimer); + detached.owner?.release(); }); }