fix(talk): make transport startup transactional

2026-08-02 05:51:39 +00:00 · 2026-08-01 15:25:50 +08:00
parent 5394bb1069
commit b7b37847c8
12 changed files with 901 additions and 154 deletions
--- a/ui/src/pages/chat/realtime-talk-gateway-relay.test.ts
+++ b/ui/src/pages/chat/realtime-talk-gateway-relay.test.ts
@@ -129,6 +129,11 @@ function createTransport(overrides: Partial<RealtimeTalkTransportContext> = {})
  });
 }

+async function startTransport(transport: GatewayRelayRealtimeTalkTransport): Promise<void> {
+  await expect(transport.start()).resolves.toBe("ready");
+  transport.activate();
+}
+
 function emitGatewayFrame(frame: GatewayFrame): void {
  for (const listener of listeners) {
    listener(frame);
@@ -198,7 +203,7 @@ describe("GatewayRelayRealtimeTalkTransport", () => {
      inputDeviceId: "usb-mic",
    });

-    await transport.start();
+    await startTransport(transport);

    expect(getUserMedia).toHaveBeenCalledWith({
      audio: {
@@ -218,7 +223,7 @@ describe("GatewayRelayRealtimeTalkTransport", () => {
      sessionKey: "main",
    });

-    await transport.start();
+    await startTransport(transport);

    const processor = processors.at(-1);
    const sink = inputSinks.at(-1);
@@ -250,13 +255,162 @@ describe("GatewayRelayRealtimeTalkTransport", () => {
    const start = transport.start();
    transport.stop();
    resolveMedia({ getTracks: () => [{ stop: stopTrack }] } as unknown as MediaStream);
-    await start;
+    await expect(start).resolves.toBe("cancelled");

    expect(stopTrack).toHaveBeenCalledOnce();
    expect(processors).toHaveLength(0);
    expect(onInputLevel).not.toHaveBeenCalled();
  });

+  it("defers relay effects until the transport is committed", async () => {
+    let resolveMedia: (media: MediaStream) => void = () => undefined;
+    getUserMedia.mockReturnValue(
+      new Promise<MediaStream>((resolve) => {
+        resolveMedia = resolve;
+      }),
+    );
+    const client = createClient();
+    const onStatus = vi.fn();
+    const onTranscript = vi.fn();
+    const transport = createTransport({
+      client,
+      callbacks: { onStatus, onTranscript },
+    });
+
+    const start = transport.start();
+    emitTalkEvent({ relaySessionId: "relay-1", type: "ready" });
+    emitTalkEvent({
+      relaySessionId: "relay-1",
+      type: "transcript",
+      role: "assistant",
+      text: "not committed yet",
+      final: true,
+    });
+    emitTalkEvent({
+      relaySessionId: "relay-1",
+      type: "toolCall",
+      callId: "call-1",
+      name: "unknown_tool",
+      args: {},
+    });
+
+    expect(onStatus).not.toHaveBeenCalled();
+    expect(onTranscript).not.toHaveBeenCalled();
+    expect(requestCallsFor(client, "talk.session.submitToolResult")).toHaveLength(0);
+
+    resolveMedia({ getTracks: () => [{ stop: vi.fn() }] } as unknown as MediaStream);
+    await expect(start).resolves.toBe("ready");
+    expect(onStatus).not.toHaveBeenCalled();
+    expect(onTranscript).not.toHaveBeenCalled();
+
+    transport.activate();
+
+    expect(onStatus).toHaveBeenCalledWith("listening");
+    expect(onTranscript).toHaveBeenCalledWith({
+      role: "assistant",
+      text: "not committed yet",
+      final: true,
+    });
+    await waitForFast(() =>
+      expect(requestCallsFor(client, "talk.session.submitToolResult")).toHaveLength(1),
+    );
+    transport.stop();
+  });
+
+  it("fails a provisional relay whose bounded event buffer overflows", async () => {
+    let resolveMedia: (media: MediaStream) => void = () => undefined;
+    getUserMedia.mockReturnValue(
+      new Promise<MediaStream>((resolve) => {
+        resolveMedia = resolve;
+      }),
+    );
+    const onStatus = vi.fn();
+    const client = createClient();
+    const transport = createTransport({ callbacks: { onStatus }, client });
+
+    const start = transport.start();
+    for (let index = 0; index < 33; index += 1) {
+      emitTalkEvent({ relaySessionId: "relay-1", type: "ready" });
+    }
+    expect(requestCallsFor(client, "talk.session.close")).toHaveLength(1);
+    resolveMedia({ getTracks: () => [{ stop: vi.fn() }] } as unknown as MediaStream);
+
+    await expect(start).rejects.toThrow(
+      "Realtime relay emitted too much data before browser setup completed",
+    );
+    expect(onStatus).not.toHaveBeenCalled();
+  });
+
+  it("enforces the provisional relay byte bound", async () => {
+    let resolveMedia: (media: MediaStream) => void = () => undefined;
+    getUserMedia.mockReturnValue(
+      new Promise<MediaStream>((resolve) => {
+        resolveMedia = resolve;
+      }),
+    );
+    const client = createClient();
+    const transport = createTransport({ client });
+
+    const start = transport.start();
+    emitTalkEvent({
+      relaySessionId: "relay-1",
+      type: "transcript",
+      role: "assistant",
+      text: "x".repeat(131_073),
+      final: true,
+    });
+
+    expect(requestCallsFor(client, "talk.session.close")).toHaveLength(1);
+    resolveMedia({ getTracks: () => [{ stop: vi.fn() }] } as unknown as MediaStream);
+    await expect(start).rejects.toThrow(
+      "Realtime relay emitted too much data before browser setup completed",
+    );
+  });
+
+  it("rejects a relay that closes before browser setup commits", async () => {
+    let resolveMedia: (media: MediaStream) => void = () => undefined;
+    getUserMedia.mockReturnValue(
+      new Promise<MediaStream>((resolve) => {
+        resolveMedia = resolve;
+      }),
+    );
+    const client = createClient();
+    const onStatus = vi.fn();
+    const transport = createTransport({ callbacks: { onStatus }, client });
+
+    const start = transport.start();
+    emitTalkEvent({
+      relaySessionId: "relay-1",
+      type: "error",
+      message: "provider rejected setup",
+    });
+    emitTalkEvent({
+      relaySessionId: "relay-1",
+      type: "close",
+      reason: "error",
+    });
+    resolveMedia({ getTracks: () => [{ stop: vi.fn() }] } as unknown as MediaStream);
+
+    await expect(start).rejects.toThrow("provider rejected setup");
+    expect(onStatus).not.toHaveBeenCalled();
+    expect(requestCallsFor(client, "talk.session.close")).toHaveLength(0);
+  });
+
+  it("closes the relay when a provisional callback throws during activation", async () => {
+    const client = createClient();
+    const onStatus = vi.fn(() => {
+      throw new Error("consumer failed");
+    });
+    const transport = createTransport({ callbacks: { onStatus }, client });
+
+    const start = transport.start();
+    emitTalkEvent({ relaySessionId: "relay-1", type: "ready" });
+    await expect(start).resolves.toBe("ready");
+
+    expect(() => transport.activate()).toThrow("consumer failed");
+    expect(requestCallsFor(client, "talk.session.close")).toHaveLength(1);
+  });
+
  it("forwards common Talk events from Gateway relay frames", async () => {
    const onTalkEvent = vi.fn();
    const transport = new GatewayRelayRealtimeTalkTransport(createSession(), {
@@ -276,7 +430,7 @@ describe("GatewayRelayRealtimeTalkTransport", () => {
      payload: {},
    } satisfies RealtimeTalkEvent;

-    await transport.start();
+    await startTransport(transport);
    emitTalkEvent({
      relaySessionId: "relay-1",
      type: "ready",
@@ -291,7 +445,7 @@ describe("GatewayRelayRealtimeTalkTransport", () => {
    const onTalkEvent = vi.fn();
    const transport = createTransport({ callbacks: { onTalkEvent } });

-    await transport.start();
+    await startTransport(transport);
    emitTalkEvent({
      relaySessionId: "relay-other",
      type: "ready",
@@ -316,7 +470,7 @@ describe("GatewayRelayRealtimeTalkTransport", () => {
    const client = createClient();
    const transport = createTransport({ client });

-    await transport.start();
+    await startTransport(transport);
    emitTalkEvent({
      relaySessionId: "relay-1",
      type: "audio",
@@ -336,7 +490,7 @@ describe("GatewayRelayRealtimeTalkTransport", () => {
    const client = createClient();
    const transport = createTransport({ client });

-    await transport.start();
+    await startTransport(transport);
    for (let index = 0; index < 321; index += 1) {
      emitTalkEvent({
        relaySessionId: "relay-1",
@@ -382,7 +536,7 @@ describe("GatewayRelayRealtimeTalkTransport", () => {
    const client = createClient();
    const transport = createTransport({ client });

-    await transport.start();
+    await startTransport(transport);
    emitTalkEvent({
      relaySessionId: "relay-1",
      type: "audio",
@@ -410,7 +564,7 @@ describe("GatewayRelayRealtimeTalkTransport", () => {
    const client = createClient();
    const transport = createTransport({ client });

-    await transport.start();
+    await startTransport(transport);
    emitTalkEvent({
      relaySessionId: "relay-1",
      type: "audio",
@@ -433,7 +587,7 @@ describe("GatewayRelayRealtimeTalkTransport", () => {
    const client = createClient();
    const transport = createTransport({ client });

-    await transport.start();
+    await startTransport(transport);
    emitTalkEvent({
      relaySessionId: "relay-1",
      type: "audio",
@@ -451,7 +605,7 @@ describe("GatewayRelayRealtimeTalkTransport", () => {
    const onInputLevel = vi.fn();
    const transport = createTransport({ callbacks: { onInputLevel } });

-    await transport.start();
+    await startTransport(transport);
    pumpMicrophone(new Float32Array(4096));
    pumpMicrophone(new Float32Array(4096).fill(0.25));
    transport.stop();
@@ -490,7 +644,7 @@ describe("GatewayRelayRealtimeTalkTransport", () => {
    });
    const transport = createTransport({ callbacks: { onStatus }, client });

-    await transport.start();
+    await startTransport(transport);
    const samples = new Float32Array(4096);
    for (let index = 0; index < 10_000; index += 1) {
      pumpMicrophone(samples);
@@ -521,7 +675,7 @@ describe("GatewayRelayRealtimeTalkTransport", () => {
    const client = createClient();
    const transport = createTransport({ client });

-    await transport.start();
+    await startTransport(transport);
    for (const timestamp of [10, 20, 30, 40]) {
      audioCurrentTime = timestamp / 1_000;
      pumpMicrophone(new Float32Array(4096));
@@ -581,7 +735,7 @@ describe("GatewayRelayRealtimeTalkTransport", () => {
    });
    const transport = createTransport({ callbacks: { onStatus }, client });

-    await transport.start();
+    await startTransport(transport);
    pumpMicrophone(new Float32Array(4096));
    await waitForFast(() =>
      expect(onStatus).toHaveBeenCalledWith("error", "Unknown realtime relay session"),
@@ -605,7 +759,7 @@ describe("GatewayRelayRealtimeTalkTransport", () => {
    const client = createClient();
    const transport = createTransport({ callbacks: { onStatus }, client });

-    await transport.start();
+    await startTransport(transport);
    pumpMicrophone(new Float32Array(4096));
    emitTalkEvent({
      relaySessionId: "relay-1",
@@ -626,12 +780,53 @@ describe("GatewayRelayRealtimeTalkTransport", () => {
    expect(closeCalls).toHaveLength(0);
  });

+  it.each(["talk event", "status"] as const)(
+    "releases local resources when a close %s callback throws",
+    async (callbackKind) => {
+      const stopTrack = vi.fn();
+      getUserMedia.mockResolvedValue({
+        getTracks: () => [{ stop: stopTrack }],
+      } as unknown as MediaStream);
+      const throwingCallback = vi.fn(() => {
+        throw new Error("consumer failed");
+      });
+      const client = createClient();
+      const transport = createTransport({
+        client,
+        callbacks:
+          callbackKind === "talk event"
+            ? { onTalkEvent: throwingCallback }
+            : { onStatus: throwingCallback },
+      });
+
+      await startTransport(transport);
+      expect(() =>
+        emitTalkEvent({
+          relaySessionId: "relay-1",
+          type: "close",
+          reason: "error",
+          talkEvent:
+            callbackKind === "talk event"
+              ? ({ type: "session.closed", final: true } satisfies RealtimeTalkEvent)
+              : undefined,
+        }),
+      ).toThrow("consumer failed");
+
+      expect(stopTrack).toHaveBeenCalledOnce();
+      expect(processors.at(-1)?.disconnect).toHaveBeenCalledOnce();
+      expect(listeners.size).toBe(0);
+      emitTalkEvent({ relaySessionId: "relay-1", type: "ready" });
+      transport.stop();
+      expect(requestCallsFor(client, "talk.session.close")).toHaveLength(0);
+    },
+  );
+
  it("preserves relay error details across close events", async () => {
    const onStatus = vi.fn();
    const client = createClient();
    const transport = createTransport({ callbacks: { onStatus }, client });

-    await transport.start();
+    await startTransport(transport);
    emitTalkEvent({
      relaySessionId: "relay-1",
      type: "error",
@@ -652,7 +847,7 @@ describe("GatewayRelayRealtimeTalkTransport", () => {
    const transport = createTransport({ client });
    const speech = new Float32Array(4096).fill(0.25);

-    await transport.start();
+    await startTransport(transport);
    emitTalkEvent({
      relaySessionId: "relay-1",
      type: "audio",
@@ -690,7 +885,7 @@ describe("GatewayRelayRealtimeTalkTransport", () => {
    });
    const transport = createTransport({ callbacks: { onStatus }, client });

-    await transport.start();
+    await startTransport(transport);
    emitTalkEvent({
      relaySessionId: "relay-1",
      type: "toolCall",
@@ -745,7 +940,7 @@ describe("GatewayRelayRealtimeTalkTransport", () => {
    });
    const transport = createTransport({ callbacks: { onStatus }, client });

-    await transport.start();
+    await startTransport(transport);
    emitTalkEvent({
      relaySessionId: "relay-1",
      type: "toolCall",
@@ -794,7 +989,7 @@ describe("GatewayRelayRealtimeTalkTransport", () => {
    });
    const transport = createTransport({ callbacks: { onStatus }, client });

-    await transport.start();
+    await startTransport(transport);
    emitTalkEvent({
      relaySessionId: "relay-1",
      type: "toolCall",
@@ -833,7 +1028,7 @@ describe("GatewayRelayRealtimeTalkTransport", () => {
    });
    const transport = createTransport({ client });

-    await transport.start();
+    await startTransport(transport);
    emitTalkEvent({
      relaySessionId: "relay-1",
      type: "toolCall",
@@ -870,7 +1065,7 @@ describe("GatewayRelayRealtimeTalkTransport", () => {
    });
    const transport = createTransport({ client });

-    await transport.start();
+    await startTransport(transport);
    emitTalkEvent({
      relaySessionId: "relay-1",
      type: "audio",
@@ -915,7 +1110,7 @@ describe("GatewayRelayRealtimeTalkTransport", () => {
    });
    const transport = createTransport({ client });

-    await transport.start();
+    await startTransport(transport);
    emitTalkEvent({
      relaySessionId: "relay-1",
      type: "audio",
@@ -967,7 +1162,7 @@ describe("GatewayRelayRealtimeTalkTransport", () => {
    });
    const transport = createTransport({ client });

-    await transport.start();
+    await startTransport(transport);
    emitTalkEvent({
      relaySessionId: "relay-1",
      type: "toolCall",
@@ -1003,7 +1198,7 @@ describe("GatewayRelayRealtimeTalkTransport", () => {
    const transport = createTransport({ client });
    const speech = new Float32Array(4096).fill(0.25);

-    await transport.start();
+    await startTransport(transport);
    emitTalkEvent({
      relaySessionId: "relay-1",
      type: "audio",
@@ -1096,7 +1291,7 @@ describe("GatewayRelayRealtimeTalkTransport", () => {
    const transport = createTransport({ callbacks: { onStatus }, client });
    const speech = new Float32Array(4096).fill(0.25);

-    await transport.start();
+    await startTransport(transport);
    emitTalkEvent({
      relaySessionId: "relay-1",
      type: "audio",
@@ -1125,7 +1320,7 @@ describe("GatewayRelayRealtimeTalkTransport", () => {

    expect(requestCallsFor(client, "talk.session.submitToolResult")).toHaveLength(0);
    expect(requestCallsFor(client, "talk.session.close")).toEqual([
-      ["talk.session.close", { sessionId: "relay-1" }],
+      ["talk.session.close", { sessionId: "relay-1" }, { timeoutMs: 8_000 }],
    ]);
    expect(onStatus).toHaveBeenCalledWith("error", "cancel failed");
  });
@@ -1140,7 +1335,7 @@ describe("GatewayRelayRealtimeTalkTransport", () => {
    });
    const transport = createTransport({ client });

-    await transport.start();
+    await startTransport(transport);
    emitTalkEvent({
      relaySessionId: "relay-1",
      type: "toolCall",
@@ -1204,7 +1399,7 @@ describe("GatewayRelayRealtimeTalkTransport", () => {
    });
    const transport = createTransport({ client });

-    await transport.start();
+    await startTransport(transport);
    emitTalkEvent({
      relaySessionId: "relay-1",
      type: "toolCall",
@@ -1251,7 +1446,7 @@ describe("GatewayRelayRealtimeTalkTransport", () => {
    });
    const transport = createTransport({ client });

-    await transport.start();
+    await startTransport(transport);
    emitTalkEvent({
      relaySessionId: "relay-1",
      type: "toolCall",
--- a/ui/src/pages/chat/realtime-talk-gateway-relay.ts
+++ b/ui/src/pages/chat/realtime-talk-gateway-relay.ts
@@ -17,6 +17,7 @@ import {
  type RealtimeTalkGatewayRelaySessionResult,
  type RealtimeTalkTransport,
  type RealtimeTalkTransportContext,
+  type RealtimeTalkTransportStartResult,
 } from "./realtime-talk-shared.ts";

 const BARGE_IN_RMS_THRESHOLD = 0.02;
@@ -24,6 +25,20 @@ const BARGE_IN_PEAK_THRESHOLD = 0.08;
 const BARGE_IN_CONSECUTIVE_SPEECH_FRAMES = 2;
 const MAX_PENDING_AUDIO_APPENDS = 4;
 const AUDIO_APPEND_TIMEOUT_MS = 8_000;
+const RELAY_CLOSE_TIMEOUT_MS = 8_000;
+const MAX_PENDING_ACTIVATION_EVENTS = 32;
+const MAX_PENDING_ACTIVATION_EVENT_BYTES = 256 * 1024;
+
+function estimateRelayEventBytes(event: GatewayRelayEvent): number {
+  try {
+    const serialized = JSON.stringify(event);
+    // Browser strings may use two bytes per code unit; use the upper bound without
+    // allocating a second encoded copy on this realtime event path.
+    return (serialized?.length ?? 0) * 2;
+  } catch {
+    return MAX_PENDING_ACTIVATION_EVENT_BYTES + 1;
+  }
+}

 export class GatewayRelayRealtimeTalkTransport implements RealtimeTalkTransport {
  private media: MediaStream | null = null;
@@ -33,6 +48,7 @@ export class GatewayRelayRealtimeTalkTransport implements RealtimeTalkTransport
  private readonly inputPump = new RealtimeTalkPcmInputPump();
  private unsubscribe: (() => void) | null = null;
  private closed = false;
+  private mediaSetupController: AbortController | null = null;
  private audioAppendAbortController: AbortController | null = null;
  private readonly pendingAudioAppends = new Set<Promise<unknown>>();
  private readonly outputQueue = new RealtimeTalkPcmOutputQueue();
@@ -46,13 +62,17 @@ export class GatewayRelayRealtimeTalkTransport implements RealtimeTalkTransport
  private pendingOutputCancellations = 0;
  private speechFramesDuringPlayback = 0;
  private lastRelayError: string | undefined;
+  private activated = false;
+  private pendingActivationEvents: GatewayRelayEvent[] = [];
+  private pendingActivationEventBytes = 0;
+  private startupError: Error | null = null;

  constructor(
    private readonly session: RealtimeTalkGatewayRelaySessionResult,
    private readonly ctx: RealtimeTalkTransportContext,
  ) {}

-  async start(): Promise<void> {
+  async start(): Promise<RealtimeTalkTransportStartResult> {
    if (!navigator.mediaDevices?.getUserMedia) {
      throw new Error("Realtime Talk requires browser microphone access");
    }
@@ -63,24 +83,46 @@ export class GatewayRelayRealtimeTalkTransport implements RealtimeTalkTransport
      throw new Error("Gateway-relay realtime Talk currently requires PCM16 audio");
    }
    this.closed = false;
+    this.activated = false;
+    this.pendingActivationEvents = [];
+    this.pendingActivationEventBytes = 0;
+    this.startupError = null;
+    this.mediaSetupController?.abort();
+    const mediaSetupController = new AbortController();
+    this.mediaSetupController = mediaSetupController;
    this.unsubscribe = this.ctx.client.addEventListener((evt) => {
      if (evt.event !== "talk.event") {
        return;
      }
-      this.handleRelayEvent(evt.payload as GatewayRelayEvent);
+      this.handleIncomingRelayEvent(evt.payload as GatewayRelayEvent);
    });
    let media: MediaStream;
    try {
-      media = await openRealtimeTalkInput(this.ctx.inputDeviceId);
+      media = await openRealtimeTalkInput(this.ctx.inputDeviceId, {
+        signal: mediaSetupController.signal,
+      });
    } catch (error) {
+      const startupError = this.currentStartupError();
+      if (startupError) {
+        throw startupError;
+      }
      if (this.closed) {
-        return;
+        return "cancelled";
      }
      throw error;
+    } finally {
+      if (this.mediaSetupController === mediaSetupController) {
+        this.mediaSetupController = null;
+      }
+    }
+    const startupError = this.currentStartupError();
+    if (startupError) {
+      media.getTracks().forEach((track) => track.stop());
+      throw startupError;
    }
    if (this.closed) {
      media.getTracks().forEach((track) => track.stop());
-      return;
+      return "cancelled";
    }
    this.media = media;
    this.inputContext = new AudioContext({ sampleRate: this.session.audio.inputSampleRateHz });
@@ -92,6 +134,28 @@ export class GatewayRelayRealtimeTalkTransport implements RealtimeTalkTransport
      this.inputMeter.start(this.media, this.inputContext);
    }
    this.startMicrophonePump();
+    return "ready";
+  }
+
+  activate(): void {
+    if (this.closed || this.activated) {
+      return;
+    }
+    this.activated = true;
+    const events = this.pendingActivationEvents;
+    this.pendingActivationEvents = [];
+    this.pendingActivationEventBytes = 0;
+    for (const event of events) {
+      try {
+        this.handleRelayEvent(event);
+      } catch (error) {
+        this.stop();
+        throw error;
+      }
+      if (this.closed) {
+        return;
+      }
+    }
  }

  stop(): void {
@@ -99,15 +163,24 @@ export class GatewayRelayRealtimeTalkTransport implements RealtimeTalkTransport
    this.stopLocal();
    if (!wasClosed) {
      void this.ctx.client
-        .request("talk.session.close", {
-          sessionId: this.session.relaySessionId,
-        })
+        .request(
+          "talk.session.close",
+          {
+            sessionId: this.session.relaySessionId,
+          },
+          { timeoutMs: RELAY_CLOSE_TIMEOUT_MS },
+        )
        .catch(() => undefined);
    }
  }

  private stopLocal(): void {
    this.closed = true;
+    this.mediaSetupController?.abort();
+    this.mediaSetupController = null;
+    this.activated = false;
+    this.pendingActivationEvents = [];
+    this.pendingActivationEventBytes = 0;
    this.unsubscribe?.();
    this.unsubscribe = null;
    this.inputPump.stop();
@@ -186,70 +259,119 @@ export class GatewayRelayRealtimeTalkTransport implements RealtimeTalkTransport
    this.pendingAudioAppends.clear();
  }

+  private currentStartupError(): Error | null {
+    return this.startupError;
+  }
+
+  private handleIncomingRelayEvent(event: GatewayRelayEvent): void {
+    if (event.relaySessionId !== this.session.relaySessionId || this.closed) {
+      return;
+    }
+    if (this.activated) {
+      this.handleRelayEvent(event);
+      return;
+    }
+    if (event.type === "error") {
+      this.lastRelayError = event.message ?? "Realtime relay failed";
+    }
+    if (event.type === "close") {
+      this.startupError = new Error(
+        event.reason === "error"
+          ? (this.lastRelayError ?? "Realtime relay closed before browser setup completed")
+          : "Realtime relay closed before browser setup completed",
+      );
+      // The server already declared this relay terminal; local cleanup must not
+      // replay its callbacks or send a redundant close request.
+      this.stopLocal();
+      return;
+    }
+    const eventBytes = estimateRelayEventBytes(event);
+    if (
+      this.pendingActivationEvents.length >= MAX_PENDING_ACTIVATION_EVENTS ||
+      eventBytes > MAX_PENDING_ACTIVATION_EVENT_BYTES - this.pendingActivationEventBytes
+    ) {
+      // The relay starts before browser media permission settles. Keep that provisional
+      // window bounded and fail the candidate instead of dropping authoritative events.
+      this.startupError = new Error(
+        "Realtime relay emitted too much data before browser setup completed",
+      );
+      // Overflow is locally terminal, so release the server relay immediately even
+      // if the browser's microphone permission prompt never settles.
+      this.stop();
+      return;
+    }
+    this.pendingActivationEvents.push(event);
+    this.pendingActivationEventBytes += eventBytes;
+  }
+
  private handleRelayEvent(event: GatewayRelayEvent): void {
    if (event.relaySessionId !== this.session.relaySessionId || this.closed) {
      return;
    }
-    if (event.talkEvent) {
-      this.ctx.callbacks.onTalkEvent?.(event.talkEvent);
-    }
-    switch (event.type) {
-      case "ready":
-        this.ctx.callbacks.onStatus?.("listening");
-        return;
-      case "audio":
-        if (event.audioBase64 && !this.playbackOverflowed) {
-          this.cancelRequestedForPlayback = false;
-          this.speechFramesDuringPlayback = 0;
-          this.playPcm16(event.audioBase64);
-        }
-        return;
-      case "clear":
-        this.playbackOverflowed = false;
-        this.stopOutput({ releaseDelayedToolResults: this.pendingOutputCancellations === 0 });
-        if (event.talkEvent?.type === "turn.cancelled") {
-          this.abortConsults();
-        }
-        return;
-      case "mark":
-        if (event.markName) {
-          this.scheduleMarkAck(event.markName);
-        }
-        return;
-      case "transcript":
-        if (event.role && event.text) {
-          this.ctx.callbacks.onTranscript?.({
-            role: event.role,
-            text: event.text,
-            final: event.final ?? false,
+    const closesRelay = event.type === "close";
+    try {
+      if (event.talkEvent) {
+        this.ctx.callbacks.onTalkEvent?.(event.talkEvent);
+      }
+      switch (event.type) {
+        case "ready":
+          this.ctx.callbacks.onStatus?.("listening");
+          return;
+        case "audio":
+          if (event.audioBase64 && !this.playbackOverflowed) {
+            this.cancelRequestedForPlayback = false;
+            this.speechFramesDuringPlayback = 0;
+            this.playPcm16(event.audioBase64);
+          }
+          return;
+        case "clear":
+          this.playbackOverflowed = false;
+          this.stopOutput({ releaseDelayedToolResults: this.pendingOutputCancellations === 0 });
+          if (event.talkEvent?.type === "turn.cancelled") {
+            this.abortConsults();
+          }
+          return;
+        case "mark":
+          if (event.markName) {
+            this.scheduleMarkAck(event.markName);
+          }
+          return;
+        case "transcript":
+          if (event.role && event.text) {
+            this.ctx.callbacks.onTranscript?.({
+              role: event.role,
+              text: event.text,
+              final: event.final ?? false,
+            });
+          }
+          return;
+        case "toolCall":
+          void this.handleToolCall(event).catch((error: unknown) => {
+            this.reportToolResultSubmissionError(error);
          });
-        }
-        return;
-      case "toolCall":
-        void this.handleToolCall(event).catch((error: unknown) => {
-          this.reportToolResultSubmissionError(error);
-        });
-        return;
-      case "toolResult":
-        if (this.isFinalToolResult(event)) {
-          this.completeToolCall(event.callId);
-        }
-        return;
-      case "error":
-        this.lastRelayError = event.message ?? "Realtime relay failed";
-        this.ctx.callbacks.onStatus?.("error", this.lastRelayError);
-        return;
-      case "close":
-        this.abortConsults();
-        if (!this.closed) {
+          return;
+        case "toolResult":
+          if (this.isFinalToolResult(event)) {
+            this.completeToolCall(event.callId);
+          }
+          return;
+        case "error":
+          this.lastRelayError = event.message ?? "Realtime relay failed";
+          this.ctx.callbacks.onStatus?.("error", this.lastRelayError);
+          return;
+        case "close":
+          this.abortConsults();
          this.ctx.callbacks.onStatus?.(
            event.reason === "error" ? "error" : "idle",
            event.reason === "error" ? (this.lastRelayError ?? "Realtime relay closed") : undefined,
          );
-          this.stopLocal();
-        }
-
-      default:
+        default:
+      }
+    } finally {
+      // The provider has ended this relay; consumer exceptions must not retain browser media.
+      if (closesRelay && !this.closed) {
+        this.stopLocal();
+      }
    }
  }

--- a/ui/src/pages/chat/realtime-talk-google-live.test.ts
+++ b/ui/src/pages/chat/realtime-talk-google-live.test.ts
@@ -21,6 +21,7 @@ type MockWebSocketHandler = (event?: MockWebSocketEvent) => void;
 type MockWebSocketEventType = "close" | "error" | "message" | "open";

 const wsInstances: MockGoogleLiveWebSocket[] = [];
+const audioContexts: MockAudioContext[] = [];
 const createdSources: MockAudioBufferSource[] = [];
 const inputProcessors: Array<{
  connect: ReturnType<typeof vi.fn>;
@@ -33,6 +34,7 @@ const inputSinks: Array<{
  gain: { value: number };
 }> = [];
 let getUserMedia: ReturnType<typeof vi.fn>;
+let stopInputTrack: ReturnType<typeof vi.fn>;

 async function flushMicrotasks(): Promise<void> {
  await Promise.resolve();
@@ -79,6 +81,19 @@ class MockGoogleLiveWebSocket {
      handler({ data });
    }
  }
+
+  emitClose() {
+    this.readyState = 3;
+    for (const handler of this.handlers.close) {
+      handler();
+    }
+  }
+
+  emitError() {
+    for (const handler of this.handlers.error) {
+      handler();
+    }
+  }
 }

 class MockAudioBufferSource {
@@ -97,6 +112,7 @@ class MockAudioContext {

  constructor(options?: { sampleRate?: number }) {
    this.sampleRate = options?.sampleRate ?? 24000;
+    audioContexts.push(this);
  }

  createMediaStreamSource() {
@@ -230,13 +246,15 @@ function requireFirstTalkEvent(onTalkEvent: ReturnType<typeof vi.fn>): Record<st
 describe("GoogleLiveRealtimeTalkTransport", () => {
  beforeEach(() => {
    wsInstances.length = 0;
+    audioContexts.length = 0;
    createdSources.length = 0;
    inputProcessors.length = 0;
    inputSinks.length = 0;
    vi.stubGlobal("WebSocket", MockGoogleLiveWebSocket);
    vi.stubGlobal("AudioContext", MockAudioContext);
+    stopInputTrack = vi.fn();
    getUserMedia = vi.fn(async () => ({
-      getTracks: () => [{ stop: vi.fn() }],
+      getTracks: () => [{ stop: stopInputTrack }],
    }));
    vi.stubGlobal("navigator", {
      mediaDevices: {
@@ -328,7 +346,7 @@ describe("GoogleLiveRealtimeTalkTransport", () => {
    const start = transport.start();
    transport.stop();
    resolveMedia({ getTracks: () => [{ stop: stopTrack }] } as unknown as MediaStream);
-    await start;
+    await expect(start).resolves.toBe("cancelled");

    expect(stopTrack).toHaveBeenCalledOnce();
    expect(inputProcessors).toHaveLength(0);
@@ -353,6 +371,87 @@ describe("GoogleLiveRealtimeTalkTransport", () => {
    expect(readyEvent.transport).toBe("provider-websocket");
  });

+  it("releases owned media when the live socket closes", async () => {
+    const onStatus = vi.fn();
+    const onTranscript = vi.fn();
+    const transport = createTransport({ onStatus, onTranscript });
+
+    await expect(transport.start()).resolves.toBe("ready");
+    const ws = latestWebSocket();
+    ws.emitOpen();
+    pumpMicrophone(new Float32Array(4096));
+    ws.emitClose();
+
+    expect(onStatus).toHaveBeenCalledWith("error", "Realtime connection closed");
+    expect(stopInputTrack).toHaveBeenCalledOnce();
+    expect(inputProcessors.at(-1)?.disconnect).toHaveBeenCalledOnce();
+    expect(audioContexts).toHaveLength(2);
+    for (const context of audioContexts) {
+      expect(context.close).toHaveBeenCalledOnce();
+    }
+
+    ws.emitMessage(
+      encodeJsonFrame({
+        serverContent: {
+          inputTranscription: { text: "too late", finished: true },
+        },
+      }),
+    );
+    await flushMicrotasks();
+    expect(onTranscript).not.toHaveBeenCalled();
+  });
+
+  it("preserves socket error precedence and ignores the later close", async () => {
+    const onStatus = vi.fn();
+    const onTranscript = vi.fn();
+    const transport = createTransport({ onStatus, onTranscript });
+
+    await expect(transport.start()).resolves.toBe("ready");
+    const ws = latestWebSocket();
+    ws.emitOpen();
+    ws.emitError();
+    ws.emitClose();
+
+    expect(onStatus).toHaveBeenCalledTimes(1);
+    expect(onStatus).toHaveBeenCalledWith("error", "Realtime connection failed");
+    expect(stopInputTrack).toHaveBeenCalledOnce();
+    ws.emitMessage(
+      encodeJsonFrame({
+        serverContent: {
+          inputTranscription: { text: "too late", finished: true },
+        },
+      }),
+    );
+    await flushMicrotasks();
+    expect(onTranscript).not.toHaveBeenCalled();
+  });
+
+  it.each(["status", "talk event"] as const)(
+    "releases socket resources when the terminal %s callback throws",
+    async (callbackKind) => {
+      const throwingCallback = vi.fn(() => {
+        throw new Error("consumer failed");
+      });
+      const transport = createTransport(
+        callbackKind === "status"
+          ? { onStatus: throwingCallback }
+          : { onTalkEvent: throwingCallback },
+      );
+
+      await expect(transport.start()).resolves.toBe("ready");
+      const ws = latestWebSocket();
+      ws.emitOpen();
+      expect(() => ws.emitError()).toThrow("consumer failed");
+
+      expect(stopInputTrack).toHaveBeenCalledOnce();
+      expect(inputProcessors.at(-1)?.disconnect).toHaveBeenCalledOnce();
+      for (const context of audioContexts) {
+        expect(context.close).toHaveBeenCalledOnce();
+      }
+      ws.emitClose();
+    },
+  );
+
  it("reports microphone activity and resets it when stopped", async () => {
    const onInputLevel = vi.fn();
    const transport = createTransport({ onInputLevel });
--- a/ui/src/pages/chat/realtime-talk-google-live.ts
+++ b/ui/src/pages/chat/realtime-talk-google-live.ts
@@ -21,6 +21,7 @@ import {
  submitRealtimeTalkConsult,
  type RealtimeTalkTransport,
  type RealtimeTalkTransportContext,
+  type RealtimeTalkTransportStartResult,
 } from "./realtime-talk-shared.ts";
 import {
  captureRealtimeTalkVideoFrame,
@@ -143,7 +144,7 @@ export class GoogleLiveRealtimeTalkTransport implements RealtimeTalkTransport {
    });
  }

-  async start(): Promise<void> {
+  async start(): Promise<RealtimeTalkTransportStartResult> {
    if (!navigator.mediaDevices?.getUserMedia || typeof WebSocket === "undefined") {
      throw new Error("Realtime Talk requires browser WebSocket and microphone access");
    }
@@ -162,7 +163,7 @@ export class GoogleLiveRealtimeTalkTransport implements RealtimeTalkTransport {
      });
    } catch (error) {
      if (this.closed) {
-        return;
+        return "cancelled";
      }
      throw error;
    } finally {
@@ -172,7 +173,7 @@ export class GoogleLiveRealtimeTalkTransport implements RealtimeTalkTransport {
    }
    if (this.closed) {
      media.getTracks().forEach((track) => track.stop());
-      return;
+      return "cancelled";
    }
    this.media = media;
    this.inputContext = new AudioContext({ sampleRate: this.session.audio.inputSampleRateHz });
@@ -193,16 +194,9 @@ export class GoogleLiveRealtimeTalkTransport implements RealtimeTalkTransport {
    this.ws.addEventListener("message", (event) => {
      void this.handleMessage(event.data);
    });
-    this.ws.addEventListener("close", () => {
-      if (!this.closed) {
-        this.ctx.callbacks.onStatus?.("error", "Realtime connection closed");
-      }
-    });
-    this.ws.addEventListener("error", () => {
-      if (!this.closed) {
-        this.ctx.callbacks.onStatus?.("error", "Realtime connection failed");
-      }
-    });
+    this.ws.addEventListener("close", () => this.failConnection("Realtime connection closed"));
+    this.ws.addEventListener("error", () => this.failConnection("Realtime connection failed"));
+    return "ready";
  }

  async setVideoEnabled(enabled: boolean): Promise<void> {
@@ -214,10 +208,18 @@ export class GoogleLiveRealtimeTalkTransport implements RealtimeTalkTransport {
  }

  stop(options?: { emitClosed?: boolean }): void {
-    if (!this.closed && options?.emitClosed !== false) {
-      this.emitTalkEvent({ type: "session.closed", final: true });
-    }
+    const emitClosed = !this.closed && options?.emitClosed !== false;
    this.closed = true;
+    try {
+      if (emitClosed) {
+        this.emitTalkEvent({ type: "session.closed", final: true });
+      }
+    } finally {
+      this.releaseResources();
+    }
+  }
+
+  private releaseResources(): void {
    this.mediaSetupController?.abort();
    this.mediaSetupController = null;
    this.setupComplete = false;
@@ -241,6 +243,18 @@ export class GoogleLiveRealtimeTalkTransport implements RealtimeTalkTransport {
    this.ws = null;
  }

+  private failConnection(detail: string): void {
+    if (this.closed) {
+      return;
+    }
+    try {
+      this.ctx.callbacks.onStatus?.("error", detail);
+    } finally {
+      // Socket failure is terminal even when a consumer callback rejects the update.
+      this.stop();
+    }
+  }
+
  private startMicrophonePump(): void {
    if (this.closed || !this.media || !this.inputContext) {
      return;
--- a/ui/src/pages/chat/realtime-talk-lifecycle.test.ts
+++ b/ui/src/pages/chat/realtime-talk-lifecycle.test.ts
@@ -4,9 +4,11 @@ import type { RealtimeTalkTransportContext } from "./realtime-talk-shared.ts";

 const transportMock = vi.hoisted(() => ({
  relayContexts: [] as RealtimeTalkTransportContext[],
+  relayStops: [] as Array<ReturnType<typeof vi.fn>>,
  webRtcContexts: [] as RealtimeTalkTransportContext[],
  webRtcStops: [] as Array<ReturnType<typeof vi.fn>>,
-  start: vi.fn(async () => undefined),
+  relayActivate: vi.fn(),
+  start: vi.fn(async (): Promise<"ready" | "cancelled"> => "ready"),
  stop: vi.fn(),
 }));

@@ -16,7 +18,13 @@ vi.mock("./realtime-talk-gateway-relay.ts", () => ({
    context: RealtimeTalkTransportContext,
  ) {
    transportMock.relayContexts.push(context);
-    return { start: transportMock.start, stop: transportMock.stop };
+    const stop = vi.fn((options?: { emitClosed?: boolean }) => transportMock.stop(options));
+    transportMock.relayStops.push(stop);
+    return {
+      start: transportMock.start,
+      activate: transportMock.relayActivate,
+      stop,
+    };
  }),
 }));
 vi.mock("./realtime-talk-google-live.ts", () => ({
@@ -66,8 +74,10 @@ function transcriptContext(contexts: RealtimeTalkTransportContext[], index = 0):
 describe("RealtimeTalkSession lifecycle", () => {
  beforeEach(() => {
    transportMock.relayContexts.length = 0;
+    transportMock.relayStops.length = 0;
    transportMock.webRtcContexts.length = 0;
    transportMock.webRtcStops.length = 0;
+    transportMock.relayActivate.mockClear();
    transportMock.start.mockClear();
    transportMock.stop.mockClear();
  });
@@ -217,8 +227,109 @@ describe("RealtimeTalkSession lifecycle", () => {
    session.stop();
  });

+  it("keeps the active transport when a replacement cancels during startup", async () => {
+    const transcriptEntryIds: string[] = [];
+    const request = vi.fn(async (method: string, params?: { entryId?: string }) => {
+      if (method === "talk.client.create") {
+        return {
+          provider: "openai",
+          transport: "webrtc",
+          voiceSessionId: "voice-cancelled-replacement",
+          clientSecret: "secret",
+        };
+      }
+      if (method === "talk.client.transcript") {
+        transcriptEntryIds.push(String(params?.entryId));
+      }
+      return { ok: true };
+    });
+    const session = new RealtimeTalkSession({ request } as never, "agent:main:main");
+    await session.start();
+    const activeContext = transcriptContext(transportMock.webRtcContexts);
+    transportMock.start.mockResolvedValueOnce("cancelled");
+
+    await session.start();
+
+    expect(transportMock.webRtcStops[0]).not.toHaveBeenCalled();
+    expect(transportMock.webRtcStops[1]).toHaveBeenCalledWith({ emitClosed: false });
+    expect(request.mock.calls.some(([method]) => method === "talk.client.close")).toBe(false);
+    activeContext.callbacks.onTranscript?.({
+      role: "user",
+      text: "active transport survived",
+      final: true,
+    });
+    await activeContext.flushTranscriptWrites?.();
+    expect(transcriptEntryIds).toEqual(["1"]);
+    session.stop();
+  });
+
+  it("restores the active relay when replacement activation throws", async () => {
+    const request = vi.fn(async (method: string) => {
+      if (method === "talk.client.create") {
+        return {
+          provider: "openai",
+          transport: "gateway-relay",
+          relaySessionId: "relay-activation",
+          audio: {
+            inputEncoding: "pcm16",
+            inputSampleRateHz: 24_000,
+            outputEncoding: "pcm16",
+            outputSampleRateHz: 24_000,
+          },
+        };
+      }
+      return { ok: true };
+    });
+    const session = new RealtimeTalkSession({ request } as never, "agent:main:main");
+    await session.start();
+    transportMock.relayActivate.mockImplementationOnce(() => {
+      throw new Error("activation failed");
+    });
+
+    await expect(session.start()).rejects.toThrow("activation failed");
+
+    expect(transportMock.relayStops[0]).not.toHaveBeenCalled();
+    expect(transportMock.relayStops[1]).toHaveBeenCalledWith({ emitClosed: false });
+    session.stop();
+    expect(transportMock.relayStops[0]).toHaveBeenCalledOnce();
+  });
+
+  it("does not restore an active relay after activation stops the session", async () => {
+    const request = vi.fn(async (method: string) => {
+      if (method === "talk.client.create") {
+        return {
+          provider: "openai",
+          transport: "gateway-relay",
+          relaySessionId: "relay-activation-stop",
+          audio: {
+            inputEncoding: "pcm16",
+            inputSampleRateHz: 24_000,
+            outputEncoding: "pcm16",
+            outputSampleRateHz: 24_000,
+          },
+        };
+      }
+      return { ok: true };
+    });
+    const session = new RealtimeTalkSession({ request } as never, "agent:main:main");
+    await session.start();
+    transportMock.relayActivate.mockImplementationOnce(() => {
+      session.stop();
+      throw new Error("activation stopped");
+    });
+
+    await expect(session.start()).rejects.toThrow("activation stopped");
+
+    expect(transportMock.relayStops[0]).toHaveBeenCalledOnce();
+    expect(transportMock.relayStops[0]).toHaveBeenCalledWith({ emitClosed: false });
+    expect(transportMock.relayStops[1]).toHaveBeenCalledOnce();
+    session.stop();
+    expect(transportMock.relayStops[0]).toHaveBeenCalledOnce();
+    expect(transportMock.relayStops[1]).toHaveBeenCalledOnce();
+  });
+
  it("ignores transcripts from a superseded pending replacement", async () => {
-    const firstReplacementStart = createDeferred<void>();
+    const firstReplacementStart = createDeferred<"ready">();
    const transcriptEntryIds: string[] = [];
    const request = vi.fn(async (method: string, params?: { entryId?: string }) => {
      if (method === "talk.client.create") {
@@ -250,7 +361,7 @@ describe("RealtimeTalkSession lifecycle", () => {
    expect(transcriptEntryIds).toEqual(["1"]);
    expect(transportMock.webRtcStops[1]).toHaveBeenCalledWith({ emitClosed: false });

-    firstReplacementStart.resolve();
+    firstReplacementStart.resolve("ready");
    await firstReplacement;
    session.stop();
  });
@@ -310,7 +421,7 @@ describe("RealtimeTalkSession lifecycle", () => {

  it("stops a pending replacement when transcript overflow closes the active call", async () => {
    const warn = vi.spyOn(console, "warn").mockImplementation(() => undefined);
-    const replacementStart = createDeferred<void>();
+    const replacementStart = createDeferred<"ready">();
    const firstTranscript = createDeferred<void>();
    const request = vi.fn(async (method: string, params?: { entryId?: string }) => {
      if (method === "talk.client.create") {
@@ -349,7 +460,7 @@ describe("RealtimeTalkSession lifecycle", () => {
    expect(transportMock.webRtcStops[0]).toHaveBeenCalledWith();
    expect(transportMock.webRtcStops[1]).toHaveBeenCalledWith({ emitClosed: false });

-    replacementStart.resolve();
+    replacementStart.resolve("ready");
    await replacement;
    firstTranscript.resolve();
    await vi.waitFor(() =>
@@ -398,8 +509,32 @@ describe("RealtimeTalkSession lifecycle", () => {
    recovered.stop();
  });

+  it("rejects a terminal failure during initial transport setup", async () => {
+    const request = vi.fn(async (method: string) => {
+      if (method === "talk.client.create") {
+        return {
+          provider: "openai",
+          transport: "webrtc",
+          voiceSessionId: "voice-terminal-startup",
+          clientSecret: "secret",
+        };
+      }
+      return { ok: true };
+    });
+    const onStatus = vi.fn();
+    const session = new RealtimeTalkSession({ request } as never, "agent:main:main", {
+      onStatus,
+    });
+    transportMock.start.mockRejectedValueOnce(new Error("Realtime connection closed"));
+
+    await expect(session.start()).rejects.toThrow("Realtime connection closed");
+
+    expect(onStatus).toHaveBeenCalledWith("connecting");
+    expect(transportMock.webRtcStops[0]).toHaveBeenCalledWith({ emitClosed: false });
+  });
+
  it("does not restore a failed replacement after concurrent stop", async () => {
-    const replacementStart = createDeferred<void>();
+    const replacementStart = createDeferred<"ready">();
    const transcriptEntryIds: string[] = [];
    const request = vi.fn(async (method: string, params?: { entryId?: string }) => {
      if (method === "talk.client.create") {
@@ -752,6 +887,7 @@ describe("RealtimeTalkSession lifecycle", () => {
    const session = new RealtimeTalkSession({ request } as never, "agent:main:main");
    await session.start();
    const context = transcriptContext(transportMock.relayContexts);
+    expect(transportMock.relayActivate).toHaveBeenCalledOnce();
    context.callbacks.onTranscript?.({ role: "user", text: "server owns this", final: true });
    await Promise.resolve();

--- a/ui/src/pages/chat/realtime-talk-shared.ts
+++ b/ui/src/pages/chat/realtime-talk-shared.ts
@@ -104,8 +104,11 @@ export type RealtimeTalkSessionResult =
  | RealtimeTalkGatewayRelaySessionResult
  | RealtimeTalkManagedRoomSessionResult;

+export type RealtimeTalkTransportStartResult = "ready" | "cancelled";
+
 export type RealtimeTalkTransport = {
-  start(): Promise<void>;
+  start(): Promise<RealtimeTalkTransportStartResult>;
+  activate?: () => void;
  stop(options?: { emitClosed?: boolean }): void;
  setVideoEnabled?: (enabled: boolean) => Promise<void>;
  switchCamera?: (videoDeviceId: string | undefined) => Promise<void>;
--- a/ui/src/pages/chat/realtime-talk-transcript-owner.ts
+++ b/ui/src/pages/chat/realtime-talk-transcript-owner.ts
@@ -1,6 +1,7 @@
 import { DEFAULT_GATEWAY_REQUEST_TIMEOUT_MS } from "@openclaw/gateway-client/browser";
 import type { BoundedSerialQueue } from "../../../../src/shared/bounded-serial-queue.js";
 import type { GatewayBrowserClient } from "../../api/gateway.ts";
+import type { RealtimeTalkTransport } from "./realtime-talk-shared.ts";

 export type ClientVoiceSessionOwner = {
  signal: AbortSignal;
@@ -57,6 +58,25 @@ export function reserveClientVoiceSessionOwner(
  };
 }

+export function retireUncommittedRealtimeTalkTransport(params: {
+  nextTransport: RealtimeTalkTransport | null;
+  transport: string;
+  owner: ClientVoiceSessionOwner;
+  reusesExistingOwner: boolean;
+  closeVoiceSession: () => void;
+}): void {
+  params.nextTransport?.stop({ emitClosed: false });
+  if (params.reusesExistingOwner) {
+    return;
+  }
+  if (params.transport === "gateway-relay" && params.nextTransport) {
+    // The relay transport owns server close once constructed; release browser ownership.
+    params.owner.release();
+    return;
+  }
+  params.closeVoiceSession();
+}
+
 export function transcriptPersistenceAbortError(): Error {
  const error = new Error("voice transcript persistence aborted");
  error.name = "AbortError";
--- a/ui/src/pages/chat/realtime-talk-transcript-queue.test.ts
+++ b/ui/src/pages/chat/realtime-talk-transcript-queue.test.ts
@@ -4,7 +4,7 @@ import type { RealtimeTalkTransportContext } from "./realtime-talk-shared.ts";

 const transportMock = vi.hoisted(() => ({
  context: undefined as RealtimeTalkTransportContext | undefined,
-  start: vi.fn(async () => undefined),
+  start: vi.fn(async () => "ready" as const),
  stop: vi.fn(),
 }));

--- a/ui/src/pages/chat/realtime-talk-webrtc.test.ts
+++ b/ui/src/pages/chat/realtime-talk-webrtc.test.ts
@@ -5,6 +5,7 @@ import { REALTIME_VOICE_AGENT_CONSULT_TOOL_NAME } from "./realtime-talk-shared.t
 import { WebRtcSdpRealtimeTalkTransport } from "./realtime-talk-webrtc.ts";

 let getUserMedia: ReturnType<typeof vi.fn>;
+let stopInputTrack: ReturnType<typeof vi.fn>;

 class FakeDataChannel extends EventTarget {
  readyState: RTCDataChannelState = "open";
@@ -177,7 +178,8 @@ describe("WebRtcSdpRealtimeTalkTransport", () => {

  beforeEach(() => {
    FakePeerConnection.instances = [];
-    const track = { stop: vi.fn() } as unknown as MediaStreamTrack;
+    stopInputTrack = vi.fn();
+    const track = { stop: stopInputTrack } as unknown as MediaStreamTrack;
    const stream = {
      getAudioTracks: () => [track],
      getTracks: () => [track],
@@ -249,7 +251,7 @@ describe("WebRtcSdpRealtimeTalkTransport", () => {
    transport.stop();
    resolveMedia(stream);

-    await expect(startPromise).resolves.toBeUndefined();
+    await expect(startPromise).resolves.toBe("cancelled");
    expect(peer?.addTrack).not.toHaveBeenCalled();
    expect(stopTrack).toHaveBeenCalledTimes(1);
    expect(fetchMock).not.toHaveBeenCalled();
@@ -276,7 +278,7 @@ describe("WebRtcSdpRealtimeTalkTransport", () => {
    transport.stop();
    rejectOffer(new Error("closed peer rejected offer creation"));

-    await expect(startPromise).resolves.toBeUndefined();
+    await expect(startPromise).resolves.toBe("cancelled");
    expect(fetchMock).not.toHaveBeenCalled();
  });

@@ -385,10 +387,77 @@ describe("WebRtcSdpRealtimeTalkTransport", () => {

    transport.stop();

-    await expect(startResult).resolves.toBeUndefined();
+    await expect(startResult).resolves.toBe("cancelled");
    expect(offerSignal?.aborted).toBe(true);
  });

+  it("reports a closed candidate when the peer fails during final setup", async () => {
+    stubAnswerSdpFetch();
+    const onStatus = vi.fn();
+    const onTalkEvent = vi.fn();
+    const transport = createOpenAiTransport({}, { onStatus, onTalkEvent });
+    const start = transport.start();
+    const peer = FakePeerConnection.instances[0];
+    if (!peer) {
+      throw new Error("expected WebRTC peer");
+    }
+    let finishRemoteDescription: (() => void) | undefined;
+    const remoteDescription = vi.spyOn(peer, "setRemoteDescription").mockImplementation(
+      async () =>
+        await new Promise<void>((resolve) => {
+          finishRemoteDescription = resolve;
+        }),
+    );
+    await waitForFast(() => expect(remoteDescription).toHaveBeenCalled());
+
+    peer.connectionState = "failed";
+    peer.dispatchEvent(new Event("connectionstatechange"));
+    finishRemoteDescription?.();
+
+    await expect(start).rejects.toThrow("Realtime connection closed");
+    expect(onStatus).not.toHaveBeenCalled();
+    expect(onTalkEvent).not.toHaveBeenCalled();
+    expect(peer.channel.close).toHaveBeenCalledOnce();
+  });
+
+  it("releases an active peer when the terminal status callback throws", async () => {
+    stubAnswerSdpFetch();
+    const onStatus = vi.fn(() => {
+      throw new Error("consumer failed");
+    });
+    const transport = createOpenAiTransport({}, { onStatus });
+    await expect(transport.start()).resolves.toBe("ready");
+    const peer = FakePeerConnection.instances[0];
+    if (!peer) {
+      throw new Error("expected WebRTC peer");
+    }
+
+    peer.connectionState = "failed";
+    peer.dispatchEvent(new Event("connectionstatechange"));
+
+    expect(onStatus).toHaveBeenCalledWith("error", "Realtime connection closed");
+    expect(stopInputTrack).toHaveBeenCalledOnce();
+    expect(peer.channel.close).toHaveBeenCalledOnce();
+  });
+
+  it("releases an active peer when its closed-event callback throws", async () => {
+    stubAnswerSdpFetch();
+    const onTalkEvent = vi.fn(() => {
+      throw new Error("consumer failed");
+    });
+    const transport = createOpenAiTransport({}, { onTalkEvent });
+    await expect(transport.start()).resolves.toBe("ready");
+    const peer = FakePeerConnection.instances[0];
+    if (!peer) {
+      throw new Error("expected WebRTC peer");
+    }
+
+    expect(() => transport.stop()).toThrow("consumer failed");
+
+    expect(stopInputTrack).toHaveBeenCalledOnce();
+    expect(peer.channel.close).toHaveBeenCalledOnce();
+  });
+
  it("clears the WebRTC offer timeout after setup succeeds", async () => {
    vi.useFakeTimers();
    let offerSignal: AbortSignal | undefined;
--- a/ui/src/pages/chat/realtime-talk-webrtc.ts
+++ b/ui/src/pages/chat/realtime-talk-webrtc.ts
@@ -14,6 +14,7 @@ import {
  submitRealtimeTalkConsult,
  type RealtimeTalkTransport,
  type RealtimeTalkTransportContext,
+  type RealtimeTalkTransportStartResult,
 } from "./realtime-talk-shared.ts";
 import { captureRealtimeTalkVideoFrame } from "./realtime-talk-video.ts";
 import {
@@ -47,6 +48,8 @@ export class WebRtcSdpRealtimeTalkTransport implements RealtimeTalkTransport {
  private readonly camera: RealtimeTalkCameraController;
  private readonly consultAbortControllers = new Set<AbortController>();
  private readonly emitTalkEvent: ReturnType<typeof createRealtimeTalkEventEmitter>;
+  private starting = false;
+  private startupError: Error | null = null;

  constructor(
    private readonly session: RealtimeTalkWebRtcSdpSessionResult,
@@ -62,11 +65,13 @@ export class WebRtcSdpRealtimeTalkTransport implements RealtimeTalkTransport {
    });
  }

-  async start(): Promise<void> {
+  async start(): Promise<RealtimeTalkTransportStartResult> {
    if (!navigator.mediaDevices?.getUserMedia || typeof RTCPeerConnection === "undefined") {
      throw new Error("Realtime Talk requires browser WebRTC and microphone access");
    }
    this.closed = false;
+    this.starting = true;
+    this.startupError = null;
    this.mediaSetupController?.abort();
    const peer = new RTCPeerConnection();
    this.peer = peer;
@@ -116,11 +121,11 @@ export class WebRtcSdpRealtimeTalkTransport implements RealtimeTalkTransport {
      }
    }
    if (media === cancelledSetup) {
-      return;
+      return this.cancelledStart();
    }
    if (!this.isCurrentPeer(peer)) {
      media.getTracks().forEach((track) => track.stop());
-      return;
+      return this.cancelledStart();
    }
    this.media = media;
    if (this.ctx.callbacks.onInputLevel) {
@@ -135,7 +140,7 @@ export class WebRtcSdpRealtimeTalkTransport implements RealtimeTalkTransport {
    const channel = peer.createDataChannel("oai-events");
    if (!this.isCurrentPeer(peer)) {
      channel.close();
-      return;
+      return this.cancelledStart();
    }
    this.channel = channel;
    channel.addEventListener("open", () => {
@@ -154,17 +159,17 @@ export class WebRtcSdpRealtimeTalkTransport implements RealtimeTalkTransport {

    const offer = await this.awaitSetupStep(peer, peer.createOffer());
    if (offer === cancelledSetup) {
-      return;
+      return this.cancelledStart();
    }
    if (!this.isCurrentPeer(peer)) {
-      return;
+      return this.cancelledStart();
    }
    const localDescriptionResult = await this.awaitSetupStep(peer, peer.setLocalDescription(offer));
    if (localDescriptionResult === cancelledSetup) {
-      return;
+      return this.cancelledStart();
    }
    if (!this.isCurrentPeer(peer)) {
-      return;
+      return this.cancelledStart();
    }
    const answerSdp = await this.offerExchange.readAnswer({
      session: this.session,
@@ -173,18 +178,23 @@ export class WebRtcSdpRealtimeTalkTransport implements RealtimeTalkTransport {
      isCurrent: () => this.isCurrentPeer(peer),
    });
    if (answerSdp === undefined) {
-      return;
+      return this.cancelledStart();
    }
    if (!this.isCurrentPeer(peer)) {
-      return;
+      return this.cancelledStart();
    }
-    await this.awaitSetupStep(
+    const remoteDescriptionResult = await this.awaitSetupStep(
      peer,
      peer.setRemoteDescription({
        type: "answer",
        sdp: answerSdp,
      }),
    );
+    if (remoteDescriptionResult === cancelledSetup || !this.isCurrentPeer(peer)) {
+      return this.cancelledStart();
+    }
+    this.starting = false;
+    return "ready";
  }

  async setVideoEnabled(enabled: boolean): Promise<void> {
@@ -199,6 +209,18 @@ export class WebRtcSdpRealtimeTalkTransport implements RealtimeTalkTransport {
    return !this.closed && this.peer === peer;
  }

+  private cancelledStart(): RealtimeTalkTransportStartResult {
+    const startupError = this.currentStartupError();
+    if (startupError) {
+      throw startupError;
+    }
+    return "cancelled";
+  }
+
+  private currentStartupError(): Error | null {
+    return this.startupError;
+  }
+
  private async awaitSetupStep<T>(
    peer: RTCPeerConnection,
    promise: Promise<T>,
@@ -214,10 +236,19 @@ export class WebRtcSdpRealtimeTalkTransport implements RealtimeTalkTransport {
  }

  stop(options?: { emitClosed?: boolean }): void {
-    if (!this.closed && options?.emitClosed !== false) {
-      this.emitTalkEvent({ type: "session.closed", final: true });
-    }
+    const emitClosed = !this.closed && options?.emitClosed !== false;
    this.closed = true;
+    try {
+      if (emitClosed) {
+        this.emitTalkEvent({ type: "session.closed", final: true });
+      }
+    } finally {
+      this.releaseResources();
+    }
+  }
+
+  private releaseResources(): void {
+    this.starting = false;
    this.mediaSetupController?.abort();
    this.mediaSetupController = null;
    this.offerExchange.abort();
@@ -246,9 +277,17 @@ export class WebRtcSdpRealtimeTalkTransport implements RealtimeTalkTransport {
    if (this.closed) {
      return;
    }
-    this.ctx.callbacks.onStatus?.("error", detail);
-    // A terminal peer failure still owns live browser media until stop() releases it.
-    this.stop();
+    const wasStarting = this.starting;
+    try {
+      if (!wasStarting) {
+        this.ctx.callbacks.onStatus?.("error", detail);
+      } else {
+        this.startupError = new Error(detail);
+      }
+    } finally {
+      // A terminal peer failure still owns browser media if status delivery fails.
+      this.stop({ emitClosed: !wasStarting });
+    }
  }

  private send(event: unknown): void {
--- a/ui/src/pages/chat/realtime-talk.test.ts
+++ b/ui/src/pages/chat/realtime-talk.test.ts
@@ -20,11 +20,11 @@ const {
  googleSwitchCamera,
  webRtcSwitchCamera,
 } = {
-  googleStart: vi.fn(async () => undefined),
+  googleStart: vi.fn(async () => "ready" as const),
  googleStop: vi.fn(),
-  relayStart: vi.fn(async () => undefined),
+  relayStart: vi.fn(async () => "ready" as const),
  relayStop: vi.fn(),
-  webRtcStart: vi.fn(async () => undefined),
+  webRtcStart: vi.fn(async () => "ready" as const),
  webRtcStop: vi.fn(),
  googleSetVideoEnabled: vi.fn(async () => undefined),
  webRtcSetVideoEnabled: vi.fn(async () => undefined),
--- a/ui/src/pages/chat/realtime-talk.ts
+++ b/ui/src/pages/chat/realtime-talk.ts
@@ -24,6 +24,7 @@ import {
  type ClientVoiceSessionOwner,
  type DetachedVoiceSession,
  reserveClientVoiceSessionOwner,
+  retireUncommittedRealtimeTalkTransport,
  transcriptPersistenceAbortError,
  waitForTranscriptRetry,
 } from "./realtime-talk-transcript-owner.ts";
@@ -174,6 +175,9 @@ export class RealtimeTalkSession {
      const existingTransport = this.transport;
      const existingVoiceSessionId = this.voiceSessionId;
      const existingOwner = this.clientVoiceSessionOwner;
+      const existingAcceptingTranscripts = this.acceptingTranscripts;
+      const existingServerOwnedVoiceSession = this.serverOwnedVoiceSession;
+      const existingTransportGeneration = this.transportGeneration;
      const providerVideoCapable = await this.resolveVideoCapability();
      if (this.closed || lifecycleGeneration !== this.lifecycleGeneration) {
        return;
@@ -230,6 +234,7 @@ export class RealtimeTalkSession {
            );
      const transcriptQueue = this.transcriptQueue;
      let nextTransport: RealtimeTalkTransport | null = null;
+      let startResult: Awaited<ReturnType<RealtimeTalkTransport["start"]>>;
      try {
        nextTransport = createTransport(session, {
          client: this.client,
@@ -246,26 +251,38 @@ export class RealtimeTalkSession {
        this.callbacks.onVideoCapability?.(
          providerVideoCapable && typeof nextTransport.setVideoEnabled === "function",
        );
-        await nextTransport.start();
+        startResult = await nextTransport.start();
      } catch (error) {
        if (this.pendingTransport === nextTransport) {
          this.pendingTransport = null;
        }
-        nextTransport?.stop({ emitClosed: false });
-        if (!(existingOwner && adoptedOwner === existingOwner)) {
-          this.closeUnadoptedVoiceSession(voiceSessionId, transport, adoptedOwner);
-        }
+        retireUncommittedRealtimeTalkTransport({
+          nextTransport,
+          transport,
+          owner: adoptedOwner,
+          reusesExistingOwner: Boolean(existingOwner && adoptedOwner === existingOwner),
+          closeVoiceSession: () =>
+            this.closeUnadoptedVoiceSession(voiceSessionId, transport, adoptedOwner),
+        });
        ownerTransferred = true;
        throw error;
      }
      if (this.pendingTransport === nextTransport) {
        this.pendingTransport = null;
      }
-      if (this.closed || lifecycleGeneration !== this.lifecycleGeneration) {
-        nextTransport.stop({ emitClosed: false });
-        if (!(existingOwner && adoptedOwner === existingOwner)) {
-          this.closeUnadoptedVoiceSession(voiceSessionId, transport, adoptedOwner);
-        }
+      if (
+        startResult === "cancelled" ||
+        this.closed ||
+        lifecycleGeneration !== this.lifecycleGeneration
+      ) {
+        retireUncommittedRealtimeTalkTransport({
+          nextTransport,
+          transport,
+          owner: adoptedOwner,
+          reusesExistingOwner: Boolean(existingOwner && adoptedOwner === existingOwner),
+          closeVoiceSession: () =>
+            this.closeUnadoptedVoiceSession(voiceSessionId, transport, adoptedOwner),
+        });
        ownerTransferred = true;
        return;
      }
@@ -280,6 +297,39 @@ export class RealtimeTalkSession {
      } else {
        this.clientVoiceSessionOwner = adoptedOwner;
      }
+      try {
+        // Publish the candidate before releasing bounded events buffered during permission.
+        nextTransport.activate?.();
+      } catch (error) {
+        const canRestoreExistingTransport =
+          !this.closed &&
+          lifecycleGeneration === this.lifecycleGeneration &&
+          this.transport === nextTransport;
+        if (canRestoreExistingTransport) {
+          this.voiceSessionId = existingVoiceSessionId;
+          this.acceptingTranscripts = existingAcceptingTranscripts;
+          this.serverOwnedVoiceSession = existingServerOwnedVoiceSession;
+          this.transportGeneration = existingTransportGeneration;
+          this.transport = existingTransport;
+          this.clientVoiceSessionOwner = existingOwner;
+          retireUncommittedRealtimeTalkTransport({
+            nextTransport,
+            transport,
+            owner: adoptedOwner,
+            reusesExistingOwner: Boolean(existingOwner && adoptedOwner === existingOwner),
+            closeVoiceSession: () =>
+              this.closeUnadoptedVoiceSession(voiceSessionId, transport, adoptedOwner),
+          });
+        } else {
+          // Stop or supersession wins activation and owns allocation cleanup.
+          if (this.transport === nextTransport) {
+            nextTransport.stop({ emitClosed: false });
+          }
+          existingTransport?.stop({ emitClosed: false });
+        }
+        ownerTransferred = true;
+        throw error;
+      }
      ownerTransferred = true;
      existingTransport?.stop({ emitClosed: false });
    } finally {