fix(talk): bound browser transcript ownership

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";

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<void>;
+};
+
+function createDeferred<T>() {
+  let resolve!: (value: T) => void;
+  const promise = new Promise<T>((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<void>((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<ReturnType<typeof createDeferred<void>>> = [];
+    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<void>();
+        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<void>((_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);
+  });
+});

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<void>((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<void>((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<void>((_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();
+    }
+  });
+});

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<void>;
-    };
-    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<void>((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);
-  });
 });

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<void>;
+  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<GatewayBrowserClient, Map<string, number>>();
+
+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<void> {
+  if (signal.aborted) {
+    throw transcriptPersistenceAbortError();
+  }
+  if (delayMs <= 0) {
+    return;
+  }
+  await new Promise<void>((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<string, number>();
   private acceptingTranscripts = false;
   private serverOwnedVoiceSession = false;
-  private transcriptWrites: Promise<void> = 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<boolean> {
@@ -218,7 +334,11 @@ export class RealtimeTalkSession {
       return false;
     }
     try {
-      const catalog = await this.client.request<TalkCatalogResult>("talk.catalog", {});
+      const catalog = await this.client.request<TalkCatalogResult>(
+        "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<RealtimeTalkConfigResult>("talk.config", {});
+          result = await this.client.request<RealtimeTalkConfigResult>(
+            "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<void> {
     const retryDelaysMs = [0, 500, 2_000];
     let lastError: unknown;
     for (const delayMs of retryDelaysMs) {
       if (delayMs > 0) {
-        await new Promise<void>((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();
       });
   }