fix(voice-call): make transcript adoption transactional

extensions/voice-call/src/webhook/realtime-handler.test.ts

@@ -1993,6 +1993,178 @@ describe("RealtimeCallHandler path routing", () => {
     }
   });
 
+  it("restores the prior transcript owner when replacement bridge creation fails", async () => {
+    const callbacks: RealtimeBridgeRequest[] = [];
+    const createBridge = vi.fn((request: RealtimeBridgeRequest) => {
+      callbacks.push(request);
+      if (callbacks.length === 1) {
+        return makeBridge();
+      }
+      request.onTranscript?.("user", "Failed ", false);
+      throw new Error("replacement bridge failed");
+    });
+    const processEvent = vi.fn();
+    const sharedCallSid = "CA-transcript-rollback";
+    const call = makeCallRecord(sharedCallSid);
+    const handler = makeHandler(undefined, {
+      manager: {
+        getCallByProviderCallId: vi.fn(() => call),
+        processEvent,
+      },
+      realtimeProvider: makeRealtimeProvider(createBridge),
+    });
+    const oldServer = await startRealtimeServer(handler);
+    let replacementServer: Awaited<ReturnType<typeof startRealtimeServer>> | undefined;
+    let oldWs: WebSocket | undefined;
+
+    try {
+      oldWs = await connectWs(oldServer.url);
+      oldWs.send(
+        JSON.stringify({
+          event: "start",
+          start: { streamSid: "MZ-transcript-rollback-old", callSid: sharedCallSid },
+        }),
+      );
+      await waitForRealtimeTest(() => {
+        expect(callbacks).toHaveLength(1);
+      });
+      callbacks[0]?.onTranscript?.("user", "Old ", false);
+
+      replacementServer = await startRealtimeServer(handler);
+      const replacementWs = await connectWs(replacementServer.url);
+      try {
+        replacementWs.send(
+          JSON.stringify({
+            event: "start",
+            start: { streamSid: "MZ-transcript-rollback-new", callSid: sharedCallSid },
+          }),
+        );
+        await waitForRealtimeTest(() => {
+          expect(createBridge).toHaveBeenCalledTimes(2);
+        });
+
+        callbacks[0]?.onTranscript?.("user", "caller", true);
+        await waitForRealtimeTest(() => {
+          expect(
+            processEvent.mock.calls
+              .map(([event]) => event as NormalizedEvent)
+              .filter((event) => event.type === "call.speech")
+              .map((event) => (event.type === "call.speech" ? event.transcript : undefined)),
+          ).toEqual(["Old caller"]);
+        });
+      } finally {
+        if (
+          replacementWs.readyState !== WebSocket.CLOSED &&
+          replacementWs.readyState !== WebSocket.CLOSING
+        ) {
+          replacementWs.close();
+        }
+      }
+    } finally {
+      if (
+        oldWs &&
+        oldWs.readyState !== WebSocket.CLOSED &&
+        oldWs.readyState !== WebSocket.CLOSING
+      ) {
+        oldWs.close();
+      }
+      await replacementServer?.close();
+      await oldServer.close();
+    }
+  });
+
+  it("cleans provisional transcript state when initial bridge creation fails", async () => {
+    const createBridge = vi.fn((request: RealtimeBridgeRequest) => {
+      request.onTranscript?.("user", "orphaned", false);
+      throw new Error("initial bridge failed");
+    });
+    const call = makeCallRecord("CA-transcript-initial-failure");
+    const handler = makeHandler(undefined, {
+      manager: {
+        getCallByProviderCallId: vi.fn(() => call),
+      },
+      realtimeProvider: makeRealtimeProvider(createBridge),
+    });
+    const server = await startRealtimeServer(handler);
+    const ws = await connectWs(server.url);
+
+    try {
+      ws.send(
+        JSON.stringify({
+          event: "start",
+          start: {
+            streamSid: "MZ-transcript-initial-failure",
+            callSid: "CA-transcript-initial-failure",
+          },
+        }),
+      );
+      await waitForRealtimeTest(() => {
+        expect(createBridge).toHaveBeenCalledOnce();
+      });
+      expect(
+        (
+          handler as unknown as {
+            userTranscriptStatesByCallId: Map<string, unknown>;
+          }
+        ).userTranscriptStatesByCallId.size,
+      ).toBe(0);
+    } finally {
+      if (ws.readyState !== WebSocket.CLOSED && ws.readyState !== WebSocket.CLOSING) {
+        ws.close();
+      }
+      await server.close();
+    }
+  });
+
+  it("keeps provisional transcript ownership across synchronous provider close", async () => {
+    let callbacks: RealtimeBridgeRequest | undefined;
+    const createBridge = vi.fn((request: RealtimeBridgeRequest) => {
+      callbacks = request;
+      request.onClose?.("completed");
+      return makeBridge();
+    });
+    const processEvent = vi.fn();
+    const call = makeCallRecord("CA-transcript-synchronous-close");
+    const handler = makeHandler(undefined, {
+      manager: {
+        getCallByProviderCallId: vi.fn(() => call),
+        processEvent,
+      },
+      realtimeProvider: makeRealtimeProvider(createBridge),
+    });
+    const server = await startRealtimeServer(handler);
+    const ws = await connectWs(server.url);
+
+    try {
+      ws.send(
+        JSON.stringify({
+          event: "start",
+          start: {
+            streamSid: "MZ-transcript-synchronous-close",
+            callSid: "CA-transcript-synchronous-close",
+          },
+        }),
+      );
+      await waitForRealtimeTest(() => {
+        expect(createBridge).toHaveBeenCalledOnce();
+      });
+      callbacks?.onTranscript?.("user", "Still listening", true);
+      await waitForRealtimeTest(() => {
+        expect(processEvent).toHaveBeenCalledWith(
+          expect.objectContaining({
+            transcript: "Still listening",
+            type: "call.speech",
+          }),
+        );
+      });
+    } finally {
+      if (ws.readyState !== WebSocket.CLOSED && ws.readyState !== WebSocket.CLOSING) {
+        ws.close();
+      }
+      await server.close();
+    }
+  });
+
   it("does not share a native consult with a replacement realtime session", async () => {
     const callbacks: RealtimeBridgeRequest[] = [];
     const oldSubmitToolResult = vi.fn();

extensions/voice-call/src/webhook/realtime-handler.ts

@@ -300,6 +300,11 @@ type UserTranscriptState = {
   recentFinalTimer?: ReturnType<typeof setTimeout>;
 };
 
+type UserTranscriptOwnerAdoption = {
+  owner: UserTranscriptState;
+  previous?: UserTranscriptState;
+};
+
 type TelephonyCloseReason = "completed" | "error";
 
 async function waitForNativeConsult(state: NativeConsultState): Promise<NativeConsultOutcome> {
@@ -768,11 +773,11 @@ export class RealtimeCallHandler {
         : undefined;
     // Providers may close synchronously before createBridge returns; no consult can exist yet.
     const nativeConsultOwner: { current?: ActiveRealtimeVoiceBridge } = {};
-    // Adopt transcript ownership before bridge creation so synchronous callbacks
-    // belong to this session while late callbacks from replaced sessions are ignored.
-    const userTranscriptOwner: UserTranscriptState = {};
-    this.adoptUserTranscriptOwner(callId, userTranscriptOwner);
-    const session = harness.createBridge({
+    // Provisional ownership accepts callbacks fired during createBridge. Commit
+    // retires the predecessor only after creation succeeds; failure restores it.
+    const userTranscriptAdoption = this.beginUserTranscriptOwnerAdoption(callId);
+    const userTranscriptOwner = userTranscriptAdoption.owner;
+    const bridgeParams: Parameters<typeof harness.createBridge>[0] = {
       provider: this.realtimeProvider,
       cfg: this.coreConfig,
       providerConfig: this.providerConfig,
@@ -994,7 +999,9 @@ export class RealtimeCallHandler {
           this.clearActiveBridgeMappings(callId, callSid, owner);
           this.cancelConsultSession(callId, owner);
         }
-        this.clearUserTranscriptState(callId, userTranscriptOwner);
+        if (ownsCallState) {
+          this.clearUserTranscriptState(callId, userTranscriptOwner);
+        }
         harness.finishOutputAudio(reason);
         harness.emit({
           type: "session.closed",
@@ -1021,7 +1028,15 @@ export class RealtimeCallHandler {
             );
           });
       },
-    });
+    };
+    let session: ActiveRealtimeVoiceBridge;
+    try {
+      session = harness.createBridge(bridgeParams);
+    } catch (error) {
+      this.rollbackUserTranscriptOwnerAdoption(callId, userTranscriptAdoption);
+      throw error;
+    }
+    this.commitUserTranscriptOwnerAdoption(callId, userTranscriptAdoption);
     nativeConsultOwner.current = session;
     providerHandlesInputAudioBargeIn =
       session.bridge.handlesInputAudioBargeIn ?? providerHandlesInputAudioBargeIn;
@@ -1093,12 +1108,44 @@ export class RealtimeCallHandler {
     return session;
   }
 
-  private adoptUserTranscriptOwner(callId: string, owner: UserTranscriptState): void {
-    const previous = this.userTranscriptStatesByCallId.get(callId);
-    if (previous?.recentFinalTimer) {
-      clearTimeout(previous.recentFinalTimer);
+  private beginUserTranscriptOwnerAdoption(callId: string): UserTranscriptOwnerAdoption {
+    const adoption = {
+      owner: {},
+      previous: this.userTranscriptStatesByCallId.get(callId),
+    } satisfies UserTranscriptOwnerAdoption;
+    this.userTranscriptStatesByCallId.set(callId, adoption.owner);
+    return adoption;
+  }
+
+  private commitUserTranscriptOwnerAdoption(
+    callId: string,
+    adoption: UserTranscriptOwnerAdoption,
+  ): void {
+    if (this.userTranscriptStatesByCallId.get(callId) !== adoption.owner) {
+      return;
     }
-    this.userTranscriptStatesByCallId.set(callId, owner);
+    if (adoption.previous?.recentFinalTimer) {
+      clearTimeout(adoption.previous.recentFinalTimer);
+      adoption.previous.recentFinalTimer = undefined;
+    }
+  }
+
+  private rollbackUserTranscriptOwnerAdoption(
+    callId: string,
+    adoption: UserTranscriptOwnerAdoption,
+  ): void {
+    if (this.userTranscriptStatesByCallId.get(callId) !== adoption.owner) {
+      return;
+    }
+    if (adoption.owner.recentFinalTimer) {
+      clearTimeout(adoption.owner.recentFinalTimer);
+      adoption.owner.recentFinalTimer = undefined;
+    }
+    if (adoption.previous) {
+      this.userTranscriptStatesByCallId.set(callId, adoption.previous);
+      return;
+    }
+    this.userTranscriptStatesByCallId.delete(callId);
   }
 
   private getUserTranscriptState(