fix(talk): separate pending and active transports

ui/src/pages/chat/realtime-talk-lifecycle.test.ts

@@ -5,6 +5,7 @@ import type { RealtimeTalkTransportContext } from "./realtime-talk-shared.ts";
 const transportMock = vi.hoisted(() => ({
   relayContexts: [] as RealtimeTalkTransportContext[],
   webRtcContexts: [] as RealtimeTalkTransportContext[],
+  webRtcStops: [] as Array<ReturnType<typeof vi.fn>>,
   start: vi.fn(async () => undefined),
   stop: vi.fn(),
 }));
@@ -27,7 +28,9 @@ vi.mock("./realtime-talk-webrtc.ts", () => ({
     context: RealtimeTalkTransportContext,
   ) {
     transportMock.webRtcContexts.push(context);
-    return { start: transportMock.start, stop: transportMock.stop };
+    const stop = vi.fn((options?: { emitClosed?: boolean }) => transportMock.stop(options));
+    transportMock.webRtcStops.push(stop);
+    return { start: transportMock.start, stop };
   }),
 }));
 
@@ -64,6 +67,7 @@ describe("RealtimeTalkSession lifecycle", () => {
   beforeEach(() => {
     transportMock.relayContexts.length = 0;
     transportMock.webRtcContexts.length = 0;
+    transportMock.webRtcStops.length = 0;
     transportMock.start.mockClear();
     transportMock.stop.mockClear();
   });
@@ -137,8 +141,9 @@ describe("RealtimeTalkSession lifecycle", () => {
     await secondContext.flushTranscriptWrites?.();
 
     expect(transcriptEntryIds).toEqual(["1", "2"]);
-    expect(transportMock.stop).toHaveBeenCalledOnce();
-    expect(transportMock.stop).toHaveBeenCalledWith({ emitClosed: false });
+    expect(transportMock.webRtcStops[0]).toHaveBeenCalledOnce();
+    expect(transportMock.webRtcStops[0]).toHaveBeenCalledWith({ emitClosed: false });
+    expect(transportMock.webRtcStops[1]).not.toHaveBeenCalled();
     expect(request.mock.calls.filter(([method]) => method === "talk.client.create")).toEqual([
       [
         "talk.client.create",
@@ -156,6 +161,8 @@ describe("RealtimeTalkSession lifecycle", () => {
       ],
     ]);
     session.stop();
+    expect(transportMock.webRtcStops[1]).toHaveBeenCalledOnce();
+    expect(transportMock.webRtcStops[1]).toHaveBeenCalledWith();
     await vi.waitFor(() =>
       expect(request.mock.calls.filter(([method]) => method === "talk.client.close")).toHaveLength(
         1,
@@ -196,8 +203,9 @@ describe("RealtimeTalkSession lifecycle", () => {
 
     await expect(session.start()).rejects.toThrow("replacement startup failed");
     expect(request.mock.calls.some(([method]) => method === "talk.client.close")).toBe(false);
-    expect(transportMock.stop).toHaveBeenCalledOnce();
-    expect(transportMock.stop).toHaveBeenCalledWith({ emitClosed: false });
+    expect(transportMock.webRtcStops[0]).not.toHaveBeenCalled();
+    expect(transportMock.webRtcStops[1]).toHaveBeenCalledOnce();
+    expect(transportMock.webRtcStops[1]).toHaveBeenCalledWith({ emitClosed: false });
 
     existingContext.callbacks.onTranscript?.({ role: "user", text: "still active", final: true });
     await existingContext.flushTranscriptWrites?.();
@@ -209,6 +217,44 @@ describe("RealtimeTalkSession lifecycle", () => {
     session.stop();
   });
 
+  it("ignores transcripts from a superseded pending replacement", async () => {
+    const firstReplacementStart = createDeferred<void>();
+    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-overlapping-replacements",
+          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();
+    transportMock.start.mockImplementationOnce(async () => await firstReplacementStart.promise);
+
+    const firstReplacement = session.start();
+    await vi.waitFor(() => expect(transportMock.webRtcContexts).toHaveLength(2));
+    await session.start();
+
+    const supersededContext = transcriptContext(transportMock.webRtcContexts, 1);
+    const activeContext = transcriptContext(transportMock.webRtcContexts, 2);
+    supersededContext.callbacks.onTranscript?.({ role: "user", text: "stale", final: true });
+    activeContext.callbacks.onTranscript?.({ role: "user", text: "active", final: true });
+    await activeContext.flushTranscriptWrites?.();
+    expect(transcriptEntryIds).toEqual(["1"]);
+    expect(transportMock.webRtcStops[1]).toHaveBeenCalledWith({ emitClosed: false });
+
+    firstReplacementStart.resolve();
+    await firstReplacement;
+    session.stop();
+  });
+
   it("releases newly allocated owners after transport startup failures", async () => {
     let createCount = 0;
     const request = vi.fn(async (method: string) => {
@@ -274,13 +320,24 @@ describe("RealtimeTalkSession lifecycle", () => {
 
     const replacing = session.start();
     await vi.waitFor(() => expect(transportMock.webRtcContexts).toHaveLength(2));
+    existingContext.callbacks.onTranscript?.({
+      role: "user",
+      text: "while replacement starts",
+      final: true,
+    });
+    await existingContext.flushTranscriptWrites?.();
+    expect(transcriptEntryIds).toEqual(["1"]);
+
     session.stop();
     replacementStart.reject(new Error("replacement failed after stop"));
     await expect(replacing).rejects.toThrow("replacement failed after stop");
 
     existingContext.callbacks.onTranscript?.({ role: "user", text: "too late", final: true });
     await Promise.resolve();
-    expect(transcriptEntryIds).toEqual([]);
+    expect(transcriptEntryIds).toEqual(["1"]);
+    expect(transportMock.webRtcStops[0]).toHaveBeenCalledOnce();
+    expect(transportMock.webRtcStops[0]).toHaveBeenCalledWith();
+    expect(transportMock.webRtcStops[1]).toHaveBeenCalledWith({ emitClosed: false });
     await vi.waitFor(() =>
       expect(request.mock.calls.filter(([method]) => method === "talk.client.close")).toHaveLength(
         1,

ui/src/pages/chat/realtime-talk.ts

@@ -142,6 +142,7 @@ function compactLaunchParams(
 
 export class RealtimeTalkSession {
   private transport: RealtimeTalkTransport | null = null;
+  private pendingTransport: RealtimeTalkTransport | null = null;
   private closed = false;
   private lifecycleGeneration = 0;
   private videoEnabled = false;
@@ -164,22 +165,14 @@ export class RealtimeTalkSession {
 
   async start(): Promise<void> {
     const lifecycleGeneration = ++this.lifecycleGeneration;
+    const supersededPendingTransport = this.pendingTransport;
+    this.pendingTransport = null;
+    supersededPendingTransport?.stop({ emitClosed: false });
     this.closed = false;
     this.callbacks.onStatus?.("connecting");
     const existingTransport = this.transport;
-    const existingTransportGeneration = this.transportGeneration;
     const existingVoiceSessionId = this.voiceSessionId;
-    const existingAcceptingTranscripts = this.acceptingTranscripts;
-    const existingServerOwnedVoiceSession = this.serverOwnedVoiceSession;
     const existingOwner = this.clientVoiceSessionOwner;
-    const restoreExistingSession = () => {
-      this.transport = existingTransport;
-      this.transportGeneration = existingTransportGeneration;
-      this.voiceSessionId = existingVoiceSessionId;
-      this.acceptingTranscripts = existingAcceptingTranscripts;
-      this.serverOwnedVoiceSession = existingServerOwnedVoiceSession;
-      this.clientVoiceSessionOwner = existingOwner;
-    };
     const owner = reserveClientVoiceSessionOwner(this.client, this.sessionKey);
     let ownerTransferred = false;
     try {
@@ -226,22 +219,15 @@ export class RealtimeTalkSession {
       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;
+      // Candidate generations must be unique without retiring the committed transport.
+      // Overlapping starts can then fence every superseded candidate independently.
+      const nextTransportGeneration = lifecycleGeneration;
       const callbacks =
         transport === "gateway-relay"
           ? this.callbacks
           : this.clientOwnedTranscriptCallbacks(
               voiceSessionId,
-              this.transportGeneration,
+              nextTransportGeneration,
               adoptedOwner.signal,
             );
       const transcriptQueue = this.transcriptQueue;
@@ -258,41 +244,46 @@ export class RealtimeTalkSession {
           consultThinkingLevel: session.consultThinkingLevel,
           consultFastMode: session.consultFastMode,
         });
-        this.transport = nextTransport;
+        this.pendingTransport = nextTransport;
         this.callbacks.onVideoCapability?.(
           providerVideoCapable && typeof nextTransport.setVideoEnabled === "function",
         );
         await nextTransport.start();
-        if (
-          this.closed ||
-          lifecycleGeneration !== this.lifecycleGeneration ||
-          this.transport !== nextTransport
-        ) {
-          nextTransport.stop({ emitClosed: false });
-          return;
-        }
-        existingTransport?.stop({ emitClosed: false });
       } catch (error) {
-        const canRollback =
-          lifecycleGeneration === this.lifecycleGeneration &&
-          (!nextTransport || this.transport === nextTransport);
+        if (this.pendingTransport === nextTransport) {
+          this.pendingTransport = null;
+        }
         nextTransport?.stop({ emitClosed: false });
-        if (!canRollback) {
-          throw error;
-        }
-        if (existingOwner && adoptedOwner === existingOwner) {
-          // A same-session replacement borrows the existing owner and queue.
-          // Restore the live transport instead of closing their shared allocation.
-          restoreExistingSession();
-        } else {
-          const detached = this.detachVoiceSession();
-          restoreExistingSession();
-          if (detached) {
-            this.closeLogicalVoiceSession(detached);
-          }
+        if (!(existingOwner && adoptedOwner === existingOwner)) {
+          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);
+        }
+        ownerTransferred = true;
+        return;
+      }
+      this.voiceSessionId = voiceSessionId;
+      this.acceptingTranscripts = true;
+      this.serverOwnedVoiceSession = transport === "gateway-relay";
+      this.transportGeneration = nextTransportGeneration;
+      this.transport = nextTransport;
+      if (this.serverOwnedVoiceSession) {
+        owner.release();
+        this.clientVoiceSessionOwner = undefined;
+      } else {
+        this.clientVoiceSessionOwner = adoptedOwner;
+      }
+      ownerTransferred = true;
+      existingTransport?.stop({ emitClosed: false });
     } finally {
       if (!ownerTransferred) {
         owner.release();
@@ -402,6 +393,9 @@ export class RealtimeTalkSession {
     this.videoEnabled = false;
     activeRealtimeTalkSessions.delete(this);
     this.callbacks.onStatus?.("idle");
+    const pendingTransport = this.pendingTransport;
+    this.pendingTransport = null;
+    pendingTransport?.stop({ emitClosed: false });
     const detached = this.detachVoiceSession();
     this.transport?.stop();
     this.transport = null;