diff --git a/extensions/voice-call/src/webhook/realtime-handler.test.ts b/extensions/voice-call/src/webhook/realtime-handler.test.ts index 9f74fb14053e..d200a994555c 100644 --- a/extensions/voice-call/src/webhook/realtime-handler.test.ts +++ b/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> | 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; + } + ).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(); diff --git a/extensions/voice-call/src/webhook/realtime-handler.ts b/extensions/voice-call/src/webhook/realtime-handler.ts index ef70f23c33b1..e8db0cb850c6 100644 --- a/extensions/voice-call/src/webhook/realtime-handler.ts +++ b/extensions/voice-call/src/webhook/realtime-handler.ts @@ -300,6 +300,11 @@ type UserTranscriptState = { recentFinalTimer?: ReturnType; }; +type UserTranscriptOwnerAdoption = { + owner: UserTranscriptState; + previous?: UserTranscriptState; +}; + type TelephonyCloseReason = "completed" | "error"; async function waitForNativeConsult(state: NativeConsultState): Promise { @@ -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[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(