Skip to content

Commit ac1042b

Browse files
clawsweeper[bot]sahibzada-allahyarTakhoffman
authored
fix(voice-call): preserve live Twilio streams in stale reaper (#90812)
Summary: - The PR updates the voice-call plugin to preserve live `speaking`/`listening` calls without `answeredAt`, backfill max-duration enforcement for live/restored call paths, and add regression tests. - PR surface: Source +90, Tests +223. Total +313 across 9 files. - Reproducibility: yes. source-level: current main and v2026.6.6 still reap aged non-terminal calls solely bec ... king` or `listening` without setting it. I did not run a live Twilio carrier call in this read-only review. Automerge notes: - Ran the ClawSweeper repair loop before final review. - Included post-review commit in the final squash: fix(voice-call): preserve live Twilio streams in stale reaper - Included post-review commit in the final squash: fix(clawsweeper): address review for automerge-openclaw-openclaw-9062… Validation: - ClawSweeper review passed for head 5fee2ff. - Required merge gates passed before the squash merge. Prepared head SHA: 5fee2ff Review: #90812 (comment) Co-authored-by: Sahibzada Allahyar <[email protected]> Co-authored-by: clawsweeper <274271284+clawsweeper[bot]@users.noreply.github.com> Co-authored-by: clawsweeper[bot] <274271284+clawsweeper[bot]@users.noreply.github.com> Approved-by: takhoffman Co-authored-by: takhoffman <[email protected]>
1 parent fd80e0d commit ac1042b

9 files changed

Lines changed: 322 additions & 9 deletions

File tree

‎extensions/voice-call/src/manager.restore.test.ts‎

Lines changed: 36 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -305,6 +305,42 @@ describe("CallManager verification on restore", () => {
305305
expect(hangupCall.reason).toBe("timeout");
306306
});
307307

308+
it.each(["speaking", "listening"] as const)(
309+
"uses call start as max-duration anchor for restored live %s calls without answeredAt",
310+
async (state) => {
311+
vi.useFakeTimers();
312+
const now = new Date("2026-03-17T03:07:00Z").getTime();
313+
vi.setSystemTime(now);
314+
const startedAt = now - 290_000;
315+
const { manager, provider, storePath } = await initializeManager({
316+
callOverrides: {
317+
callId: `call-${state}`,
318+
providerCallId: `provider-${state}`,
319+
state,
320+
startedAt,
321+
answeredAt: undefined,
322+
},
323+
configOverrides: { maxDurationSeconds: 300 },
324+
});
325+
326+
const activeCall = requireSingleActiveCall(manager);
327+
expect(activeCall.state).toBe(state);
328+
expect(activeCall.answeredAt).toBe(startedAt);
329+
expect(
330+
loadActiveCallsFromStore(storePath).activeCalls.get(activeCall.callId)?.answeredAt,
331+
).toBe(startedAt);
332+
333+
await vi.advanceTimersByTimeAsync(9_000);
334+
expect(manager.getActiveCalls()).toHaveLength(1);
335+
expect(provider.hangupCalls).toHaveLength(0);
336+
337+
await vi.advanceTimersByTimeAsync(1_100);
338+
expect(manager.getActiveCalls()).toHaveLength(0);
339+
const hangupCall = requireSingleHangupCall(provider);
340+
expect(hangupCall.reason).toBe("timeout");
341+
},
342+
);
343+
308344
it("restores dedupe keys from terminal persisted calls so replayed webhooks stay ignored", async () => {
309345
const storePath = createTestStorePath();
310346
const persisted = makePersistedCall({

‎extensions/voice-call/src/manager.ts‎

Lines changed: 17 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -47,6 +47,13 @@ function incrementRestoreStatusCount(
4747
counts.set(key, (counts.get(key) ?? 0) + 1);
4848
}
4949

50+
function resolveRestoredMaxDurationAnchor(call: CallRecord): number | undefined {
51+
return (
52+
call.answeredAt ??
53+
(call.state === "speaking" || call.state === "listening" ? call.startedAt : undefined)
54+
);
55+
}
56+
5057
function resolveDefaultStoreBase(config: VoiceCallConfig, storePath?: string): string {
5158
const rawOverride = storePath?.trim() || config.store?.trim();
5259
if (rawOverride) {
@@ -126,11 +133,12 @@ export class CallManager {
126133
}
127134
}
128135

129-
// Restart max-duration timers for restored calls that are past the answered state
136+
// Restart max-duration timers for restored calls that are past the answered/live state.
130137
let skippedAlreadyElapsedTimers = 0;
131138
for (const [callId, call] of verified) {
132-
if (call.answeredAt && !TerminalStates.has(call.state)) {
133-
const elapsed = Date.now() - call.answeredAt;
139+
const maxDurationAnchor = resolveRestoredMaxDurationAnchor(call);
140+
if (maxDurationAnchor !== undefined && !TerminalStates.has(call.state)) {
141+
const elapsed = Date.now() - maxDurationAnchor;
134142
const maxDurationMs = resolveVoiceCallSecondsTimerDelayMs(this.config.maxDurationSeconds);
135143
if (elapsed >= maxDurationMs) {
136144
// Already expired — remove instead of keeping
@@ -141,6 +149,12 @@ export class CallManager {
141149
skippedAlreadyElapsedTimers += 1;
142150
continue;
143151
}
152+
if (call.answeredAt === undefined) {
153+
// Twilio streams can restore directly in speaking/listening without an
154+
// answered webhook; anchoring at startedAt preserves bounded duration.
155+
call.answeredAt = maxDurationAnchor;
156+
persistCallRecord(this.storePath, call);
157+
}
144158
startMaxDurationTimer({
145159
ctx: this.getContext(),
146160
callId,

‎extensions/voice-call/src/manager/events.test.ts‎

Lines changed: 153 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -7,13 +7,14 @@ import {
77
createPluginStateSyncKeyedStoreForTests,
88
resetPluginStateStoreForTests,
99
} from "openclaw/plugin-sdk/plugin-state-test-runtime";
10-
import { afterEach, beforeEach, describe, expect, it } from "vitest";
10+
import { afterEach, beforeEach, describe, expect, it, vi } from "vitest";
1111
import { VoiceCallConfigSchema } from "../config.js";
1212
import type { VoiceCallProvider } from "../providers/base.js";
1313
import { clearVoiceCallStateRuntime, setVoiceCallStateRuntime } from "../runtime-state.js";
1414
import type { AnswerCallInput, HangupCallInput, NormalizedEvent } from "../types.js";
1515
import type { CallManagerContext } from "./context.js";
1616
import { processEvent } from "./events.js";
17+
import { speakInitialMessage } from "./outbound.js";
1718
import { flushPendingCallRecordWritesForTest } from "./store.js";
1819

1920
const contexts: CallManagerContext[] = [];
@@ -54,6 +55,8 @@ afterEach(async () => {
5455
}
5556
clearVoiceCallStateRuntime();
5657
resetPluginStateStoreForTests();
58+
vi.useRealTimers();
59+
vi.restoreAllMocks();
5760
});
5861

5962
function createContext(overrides: Partial<CallManagerContext> = {}): CallManagerContext {
@@ -363,6 +366,155 @@ describe("processEvent (functional)", () => {
363366
expect(answeredCallId).toBe("call-2");
364367
});
365368

369+
it.each([
370+
{
371+
name: "speaking",
372+
expectedState: "speaking",
373+
createEvent: (timestamp: number): NormalizedEvent => ({
374+
id: "evt-live-speaking",
375+
type: "call.speaking",
376+
callId: "call-live",
377+
providerCallId: "provider-live",
378+
timestamp,
379+
text: "hello",
380+
}),
381+
},
382+
{
383+
name: "listening",
384+
expectedState: "listening",
385+
createEvent: (timestamp: number): NormalizedEvent => ({
386+
id: "evt-live-listening",
387+
type: "call.speech",
388+
callId: "call-live",
389+
providerCallId: "provider-live",
390+
timestamp,
391+
transcript: "hello",
392+
isFinal: true,
393+
}),
394+
},
395+
])(
396+
"starts max-duration enforcement when $name arrives before answered",
397+
async ({ expectedState, createEvent }) => {
398+
const now = new Date("2026-03-22T12:00:00.000Z").getTime();
399+
vi.useFakeTimers();
400+
vi.setSystemTime(now);
401+
const hangupCalls: HangupCallInput[] = [];
402+
const ctx = createContext({
403+
config: VoiceCallConfigSchema.parse({
404+
enabled: true,
405+
provider: "plivo",
406+
fromNumber: "+15550000000",
407+
maxDurationSeconds: 1,
408+
}),
409+
provider: createProvider({
410+
hangupCall: async (input: HangupCallInput): Promise<void> => {
411+
hangupCalls.push(input);
412+
},
413+
}),
414+
});
415+
ctx.activeCalls.set("call-live", {
416+
callId: "call-live",
417+
providerCallId: "provider-live",
418+
provider: "plivo",
419+
direction: "inbound",
420+
state: "ringing",
421+
from: "+15550000002",
422+
to: "+15550000000",
423+
startedAt: now - 120_000,
424+
transcript: [],
425+
processedEventIds: [],
426+
metadata: {},
427+
});
428+
ctx.providerCallIdMap.set("provider-live", "call-live");
429+
const liveTimestamp = now + 250;
430+
431+
processEvent(ctx, createEvent(liveTimestamp));
432+
433+
const call = ctx.activeCalls.get("call-live");
434+
if (!call) {
435+
throw new Error("expected live call to remain active");
436+
}
437+
expect(call.state).toBe(expectedState);
438+
expect(call.answeredAt).toBe(liveTimestamp);
439+
expect(ctx.maxDurationTimers.has("call-live")).toBe(true);
440+
441+
await vi.advanceTimersByTimeAsync(1_000);
442+
443+
expect(hangupCalls).toEqual([
444+
{
445+
callId: "call-live",
446+
providerCallId: "provider-live",
447+
reason: "timeout",
448+
},
449+
]);
450+
expect(ctx.activeCalls.has("call-live")).toBe(false);
451+
vi.useRealTimers();
452+
},
453+
);
454+
455+
it("enforces max duration for Twilio initial-message streams without answeredAt", async () => {
456+
const now = new Date("2026-03-22T12:00:00.000Z").getTime();
457+
vi.useFakeTimers();
458+
vi.setSystemTime(now);
459+
const hangupCalls: HangupCallInput[] = [];
460+
const provider = createProvider({
461+
name: "twilio",
462+
hangupCall: async (input: HangupCallInput): Promise<void> => {
463+
hangupCalls.push(input);
464+
},
465+
}) as VoiceCallProvider & { isConversationStreamConnectEnabled?: () => boolean };
466+
provider.isConversationStreamConnectEnabled = () => true;
467+
const ctx = createContext({
468+
config: VoiceCallConfigSchema.parse({
469+
enabled: true,
470+
provider: "twilio",
471+
fromNumber: "+15550000000",
472+
maxDurationSeconds: 1,
473+
streaming: { enabled: true },
474+
}),
475+
provider,
476+
});
477+
ctx.activeCalls.set("call-stream", {
478+
callId: "call-stream",
479+
providerCallId: "provider-stream",
480+
provider: "twilio",
481+
direction: "inbound",
482+
state: "active",
483+
from: "+15550000002",
484+
to: "+15550000000",
485+
startedAt: now - 120_000,
486+
transcript: [],
487+
processedEventIds: [],
488+
metadata: {
489+
initialMessage: "Hello from the bot.",
490+
mode: "conversation",
491+
},
492+
});
493+
ctx.providerCallIdMap.set("provider-stream", "call-stream");
494+
495+
await speakInitialMessage(ctx, "provider-stream");
496+
497+
const call = ctx.activeCalls.get("call-stream");
498+
if (!call) {
499+
throw new Error("expected initial-message call to remain active");
500+
}
501+
expect(call.state).toBe("speaking");
502+
expect(call.answeredAt).toBe(now);
503+
expect(ctx.maxDurationTimers.has("call-stream")).toBe(true);
504+
505+
await vi.advanceTimersByTimeAsync(1_000);
506+
507+
expect(hangupCalls).toEqual([
508+
{
509+
callId: "call-stream",
510+
providerCallId: "provider-stream",
511+
reason: "timeout",
512+
},
513+
]);
514+
expect(ctx.activeCalls.has("call-stream")).toBe(false);
515+
vi.useRealTimers();
516+
});
517+
366518
it("removes active call even when hangup rejects", () => {
367519
const provider = createProvider({
368520
hangupCall: async (): Promise<void> => {

‎extensions/voice-call/src/manager/events.ts‎

Lines changed: 21 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -10,7 +10,11 @@ import { findCall } from "./lookup.js";
1010
import { endCall } from "./outbound.js";
1111
import { addTranscriptEntry, transitionState } from "./state.js";
1212
import { persistCallRecord } from "./store.js";
13-
import { resolveTranscriptWaiter, startMaxDurationTimer } from "./timers.js";
13+
import {
14+
ensureMaxDurationTimerForLiveCall,
15+
resolveTranscriptWaiter,
16+
startMaxDurationTimer,
17+
} from "./timers.js";
1418

1519
type EventContext = Pick<
1620
CallManagerContext,
@@ -277,6 +281,14 @@ export function processEvent(ctx: EventContext, event: NormalizedEvent): void {
277281
break;
278282

279283
case "call.speaking":
284+
ensureMaxDurationTimerForLiveCall({
285+
ctx,
286+
call,
287+
liveAt: event.timestamp,
288+
onTimeout: async (callId) => {
289+
await endCall(ctx, callId, { reason: "timeout" });
290+
},
291+
});
280292
transitionState(call, "speaking");
281293
break;
282294

@@ -297,6 +309,14 @@ export function processEvent(ctx: EventContext, event: NormalizedEvent): void {
297309
}
298310
addTranscriptEntry(call, "user", event.transcript);
299311
}
312+
ensureMaxDurationTimerForLiveCall({
313+
ctx,
314+
call,
315+
liveAt: event.timestamp,
316+
onTimeout: async (callId) => {
317+
await endCall(ctx, callId, { reason: "timeout" });
318+
},
319+
});
300320
transitionState(call, "listening");
301321
break;
302322

‎extensions/voice-call/src/manager/outbound.test.ts‎

Lines changed: 7 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -5,6 +5,7 @@ import { beforeEach, describe, expect, it, vi } from "vitest";
55
const {
66
addTranscriptEntryMock,
77
clearMaxDurationTimerMock,
8+
ensureMaxDurationTimerForLiveCallMock,
89
generateDtmfRedirectTwimlMock,
910
generateNotifyTwimlMock,
1011
getCallByProviderCallIdMock,
@@ -15,6 +16,11 @@ const {
1516
} = vi.hoisted(() => ({
1617
addTranscriptEntryMock: vi.fn(),
1718
clearMaxDurationTimerMock: vi.fn(),
19+
ensureMaxDurationTimerForLiveCallMock: vi.fn(
20+
(params: { call: { answeredAt?: number }; liveAt: number }) => {
21+
params.call.answeredAt ??= params.liveAt;
22+
},
23+
),
1824
generateDtmfRedirectTwimlMock: vi.fn(),
1925
generateNotifyTwimlMock: vi.fn(),
2026
getCallByProviderCallIdMock: vi.fn(),
@@ -36,6 +42,7 @@ vi.mock("./store.js", () => ({
3642
vi.mock("./timers.js", () => ({
3743
clearMaxDurationTimer: clearMaxDurationTimerMock,
3844
clearTranscriptWaiter: vi.fn(),
45+
ensureMaxDurationTimerForLiveCall: ensureMaxDurationTimerForLiveCallMock,
3946
rejectTranscriptWaiter: rejectTranscriptWaiterMock,
4047
waitForFinalTranscript: vi.fn(),
4148
}));

‎extensions/voice-call/src/manager/outbound.ts‎

Lines changed: 20 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -21,7 +21,11 @@ import { getCallByProviderCallId } from "./lookup.js";
2121
import { addTranscriptEntry, transitionState } from "./state.js";
2222
import { persistCallRecord } from "./store.js";
2323
import { resolveVoiceCallSecondsTimerDelayMs } from "./timer-delays.js";
24-
import { clearTranscriptWaiter, waitForFinalTranscript } from "./timers.js";
24+
import {
25+
clearTranscriptWaiter,
26+
ensureMaxDurationTimerForLiveCall,
27+
waitForFinalTranscript,
28+
} from "./timers.js";
2529
import { generateDtmfRedirectTwiml, generateNotifyTwiml } from "./twiml.js";
2630

2731
type InitiateContext = Pick<
@@ -37,7 +41,13 @@ type InitiateContext = Pick<
3741

3842
type SpeakContext = Pick<
3943
CallManagerContext,
40-
"activeCalls" | "providerCallIdMap" | "provider" | "config" | "storePath"
44+
| "activeCalls"
45+
| "providerCallIdMap"
46+
| "provider"
47+
| "config"
48+
| "storePath"
49+
| "transcriptWaiters"
50+
| "maxDurationTimers"
4151
>;
4252

4353
type ConversationContext = Pick<
@@ -267,6 +277,14 @@ export async function speak(
267277
const { call, providerCallId, provider } = connected;
268278

269279
try {
280+
ensureMaxDurationTimerForLiveCall({
281+
ctx,
282+
call,
283+
liveAt: Date.now(),
284+
onTimeout: async (id) => {
285+
await endCall(ctx, id, { reason: "timeout" });
286+
},
287+
});
270288
transitionState(call, "speaking");
271289
persistCallRecord(ctx.storePath, call);
272290

0 commit comments

Comments
 (0)