From 5e465953314fb6de34014fbe923c09ffdf2af2b5 Mon Sep 17 00:00:00 2001 From: Vincent Koc Date: Fri, 31 Jul 2026 14:34:09 +0800 Subject: [PATCH] fix(voice-call): cancel pending replacement consults --- .../src/webhook/realtime-handler.test.ts | 48 +++++++++----- .../src/webhook/realtime-handler.ts | 66 ++++++++++++++----- 2 files changed, 82 insertions(+), 32 deletions(-) diff --git a/extensions/voice-call/src/webhook/realtime-handler.test.ts b/extensions/voice-call/src/webhook/realtime-handler.test.ts index ae8186990460..e456691265cc 100644 --- a/extensions/voice-call/src/webhook/realtime-handler.test.ts +++ b/extensions/voice-call/src/webhook/realtime-handler.test.ts @@ -1561,6 +1561,10 @@ describe("RealtimeCallHandler path routing", () => { }); it("keeps a replacement session's forced consult when the old result resolves late", async () => { + const sessionHarnesses: RealtimeVoiceSessionHarness[] = []; + realtimeVoiceHarnessTestHooks.onCreate = (harness) => { + sessionHarnesses.push(harness); + }; const callbacks: RealtimeBridgeRequest[] = []; const oldSendUserMessage = vi.fn(); const replacementSendUserMessage = vi.fn(); @@ -1626,6 +1630,20 @@ describe("RealtimeCallHandler path routing", () => { await waitForRealtimeTest(() => { expect(consult).toHaveBeenCalledTimes(1); }); + const oldCoordinator = expectDefined( + sessionHarnesses[0], + "old voice-call realtime session harness", + ).forcedConsults; + const oldForcedHandle = expectDefined( + oldCoordinator.handles().find((handle) => handle.question === "Check the old deployment."), + "old forced consult handle", + ); + const stalePendingHandle = expectDefined( + oldCoordinator.prepare("Pending work from the old session."), + "stale pending forced consult handle", + ); + const stalePendingRun = vi.fn(); + oldCoordinator.schedule(stalePendingHandle, 60_000, stalePendingRun); replacementServer = await startRealtimeServer(handler); const replacementWs = await connectWs(replacementServer.url); @@ -1639,6 +1657,8 @@ describe("RealtimeCallHandler path routing", () => { await waitForRealtimeTest(() => { expect(callbacks).toHaveLength(2); }); + expect(oldCoordinator.handles()).not.toContainEqual(stalePendingHandle); + expect(stalePendingRun).not.toHaveBeenCalled(); callbacks[1]?.onTranscript?.("user", "Check the new deployment.", true); await waitForRealtimeTest(() => { expect(consult).toHaveBeenCalledTimes(2); @@ -1651,31 +1671,27 @@ describe("RealtimeCallHandler path routing", () => { name: "openclaw_agent_consult", args: { question: "Check the old deployment." }, }); - await waitForRealtimeTest(() => { - expect(oldSubmitToolResult).toHaveBeenCalledWith( - "stale-native-consult", - { - status: "cancelled", - message: "OpenClaw cancelled this consult before completion. Do not restart it.", - }, - undefined, - ); + await new Promise((resolve) => { + setTimeout(resolve, 0); }); + expect(oldSubmitToolResult).not.toHaveBeenCalled(); expect(consult).toHaveBeenCalledTimes(2); - const oldClosed = waitForClose(oldWs); - oldWs.close(); - await oldClosed; - await waitForRealtimeTest(() => { - expect(oldCloseBridge).toHaveBeenCalledTimes(1); - }); - oldResult.resolve({ text: "The old deployment is healthy." }); await new Promise((resolve) => { setTimeout(resolve, 0); }); expect(clearAudio).toHaveBeenCalledTimes(2); expect(oldSendUserMessage).not.toHaveBeenCalled(); + expect(oldCoordinator.handles()).toContainEqual(oldForcedHandle); + expect(oldCoordinator.isCancelled(oldForcedHandle)).toBe(true); + + const oldClosed = waitForClose(oldWs); + oldWs.close(); + await oldClosed; + await waitForRealtimeTest(() => { + expect(oldCloseBridge).toHaveBeenCalledTimes(1); + }); replacementResult.resolve({ text: "The new deployment is healthy." }); await waitForRealtimeTest(() => { diff --git a/extensions/voice-call/src/webhook/realtime-handler.ts b/extensions/voice-call/src/webhook/realtime-handler.ts index 784634118e15..653400880fbe 100644 --- a/extensions/voice-call/src/webhook/realtime-handler.ts +++ b/extensions/voice-call/src/webhook/realtime-handler.ts @@ -275,6 +275,11 @@ type ForcedConsultState = { completedAt?: number; }; +type ForcedConsultSession = { + owner: ActiveRealtimeVoiceBridge; + coordinator: RealtimeVoiceSessionHarness["forcedConsults"]; +}; + type NativeConsultState = { owner: ActiveRealtimeVoiceBridge; startedAt: number; @@ -345,6 +350,7 @@ export class RealtimeCallHandler { ReturnType >(); private readonly forcedConsultsByCallId = new Map(); + private readonly forcedConsultSessionsByCallId = new Map(); private readonly nativeConsultsInFlightByCallId = new Map(); private closePromise: Promise | null = null; private closing = false; @@ -940,7 +946,7 @@ export class RealtimeCallHandler { if (nativeConsultOwner.current) { this.clearActiveBridgeMappings(callId, callSid, nativeConsultOwner.current); this.cancelNativeConsult(callId, nativeConsultOwner.current); - this.cancelForcedConsult(callId, nativeConsultOwner.current); + this.cancelForcedConsultSession(callId, nativeConsultOwner.current); } this.clearUserTranscriptState(callId); harness.finishOutputAudio(reason); @@ -977,10 +983,14 @@ export class RealtimeCallHandler { emitCallEnd(reason); } }; - const previousSession = this.activeBridgesByCallId.get(callId); - if (previousSession && previousSession !== session) { - this.cancelForcedConsult(callId, previousSession); + const previousForcedConsultSession = this.forcedConsultSessionsByCallId.get(callId); + if (previousForcedConsultSession && previousForcedConsultSession.owner !== session) { + this.cancelForcedConsultSession(callId, previousForcedConsultSession.owner); } + this.forcedConsultSessionsByCallId.set(callId, { + owner: session, + coordinator: harness.forcedConsults, + }); this.activeBridgesByCallId.set(callId, session); this.activeBridgesByCallId.set(callSid, session); this.activeTelephonyClosersByCallId.set(callId, closeTelephony); @@ -1014,7 +1024,7 @@ export class RealtimeCallHandler { } finally { this.clearActiveBridgeMappings(callId, callSid, session); this.cancelNativeConsult(callId, session); - this.cancelForcedConsult(callId, session); + this.cancelForcedConsultSession(callId, session); this.clearUserTranscriptState(callId); harness.close(); audioPacer.close(); @@ -1097,6 +1107,22 @@ export class RealtimeCallHandler { this.forcedConsultsByCallId.delete(callId); } + private cancelForcedConsultSession( + callId: string, + owner: ActiveRealtimeVoiceBridge | undefined, + ): void { + if (!owner) { + return; + } + const session = this.forcedConsultSessionsByCallId.get(callId); + if (!session || session.owner !== owner) { + return; + } + session.coordinator.clearPending(); + this.cancelForcedConsult(callId, owner); + this.forcedConsultSessionsByCallId.delete(callId); + } + private clearActiveBridgeMappings( callId: string, callSid: string, @@ -1188,7 +1214,10 @@ export class RealtimeCallHandler { transcript: string; clearAudio: () => void; }): void { - if (this.config.consultPolicy !== "always") { + if ( + this.config.consultPolicy !== "always" || + this.activeBridgesByCallId.get(params.callId) !== params.session + ) { return; } const question = params.transcript.trim(); @@ -1290,16 +1319,18 @@ export class RealtimeCallHandler { `[voice-call] realtime forced agent consult failed callId=${params.callId} providerCallId=${params.callSid} error=${formatErrorMessage(error)}`, ); } finally { - if (state.cancelled || this.forcedConsultsByCallId.get(params.callId) !== state) { - coordinator.remove(params.handle); - } else { - const cleanupTimer = setTimeout(() => { - if (this.forcedConsultsByCallId.get(params.callId) === state) { - this.forcedConsultsByCallId.delete(params.callId); - coordinator.remove(params.handle); - } - }, FORCED_CONSULT_NATIVE_DEDUPE_MS); - cleanupTimer.unref?.(); + if (!state.cancelled) { + if (this.forcedConsultsByCallId.get(params.callId) !== state) { + coordinator.remove(params.handle); + } else { + const cleanupTimer = setTimeout(() => { + if (this.forcedConsultsByCallId.get(params.callId) === state) { + this.forcedConsultsByCallId.delete(params.callId); + coordinator.remove(params.handle); + } + }, FORCED_CONSULT_NATIVE_DEDUPE_MS); + cleanupTimer.unref?.(); + } } } } @@ -1438,6 +1469,9 @@ export class RealtimeCallHandler { } }; if (name === REALTIME_VOICE_AGENT_CONSULT_TOOL_NAME) { + if (this.activeBridgesByCallId.get(callId) !== bridge) { + return; + } const coordinator = harness.forcedConsults; const forcedMatch = coordinator.recordNativeConsult(args, bridgeCallId); if (forcedMatch.kind === "none") {