ewilliams28/openclaw
1
0
Fork 0
mirror of https://github.com/openclaw/openclaw.git synced 2026-08-27 12:56:01 -06:00
Code Issues Packages Projects Releases Wiki Activity

fix(voice-call): cancel pending replacement consults

Browse Source
This commit is contained in:
Vincent Koc
2026-07-31 14:34:09 +08:00
parent ee4a19e8ac
commit 5e46595331
2 changed files with 82 additions and 32 deletions
Download Patch File Download Diff File Expand all files Collapse all files
extensions/voice-call/src/webhook/realtime-handler.test.ts
+32 -16
View File
@@ -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<void>((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<void>((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(() => {
extensions/voice-call/src/webhook/realtime-handler.ts
+50 -16
View File
@@ -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<typeof setTimeout>
>();
private readonly forcedConsultsByCallId = new Map<string, ForcedConsultState>();
private readonly forcedConsultSessionsByCallId = new Map<string, ForcedConsultSession>();
private readonly nativeConsultsInFlightByCallId = new Map<string, NativeConsultState>();
private closePromise: Promise<void> | 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") {