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 5fee2ff7a1.
- Required merge gates passed before the squash merge.

Prepared head SHA: 5fee2ff7a1
Review: https://github.com/openclaw/openclaw/pull/90812#issuecomment-4637047870

Co-authored-by: Sahibzada Allahyar <sahibzada@fastino.ai>
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 <781889+takhoffman@users.noreply.github.com>

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

@@ -305,6 +305,42 @@ describe("CallManager verification on restore", () => {
     expect(hangupCall.reason).toBe("timeout");
   });
 
+  it.each(["speaking", "listening"] as const)(
+    "uses call start as max-duration anchor for restored live %s calls without answeredAt",
+    async (state) => {
+      vi.useFakeTimers();
+      const now = new Date("2026-03-17T03:07:00Z").getTime();
+      vi.setSystemTime(now);
+      const startedAt = now - 290_000;
+      const { manager, provider, storePath } = await initializeManager({
+        callOverrides: {
+          callId: `call-${state}`,
+          providerCallId: `provider-${state}`,
+          state,
+          startedAt,
+          answeredAt: undefined,
+        },
+        configOverrides: { maxDurationSeconds: 300 },
+      });
+
+      const activeCall = requireSingleActiveCall(manager);
+      expect(activeCall.state).toBe(state);
+      expect(activeCall.answeredAt).toBe(startedAt);
+      expect(
+        loadActiveCallsFromStore(storePath).activeCalls.get(activeCall.callId)?.answeredAt,
+      ).toBe(startedAt);
+
+      await vi.advanceTimersByTimeAsync(9_000);
+      expect(manager.getActiveCalls()).toHaveLength(1);
+      expect(provider.hangupCalls).toHaveLength(0);
+
+      await vi.advanceTimersByTimeAsync(1_100);
+      expect(manager.getActiveCalls()).toHaveLength(0);
+      const hangupCall = requireSingleHangupCall(provider);
+      expect(hangupCall.reason).toBe("timeout");
+    },
+  );
+
   it("restores dedupe keys from terminal persisted calls so replayed webhooks stay ignored", async () => {
     const storePath = createTestStorePath();
     const persisted = makePersistedCall({

extensions/voice-call/src/manager.ts

@@ -47,6 +47,13 @@ function incrementRestoreStatusCount(
   counts.set(key, (counts.get(key) ?? 0) + 1);
 }
 
+function resolveRestoredMaxDurationAnchor(call: CallRecord): number | undefined {
+  return (
+    call.answeredAt ??
+    (call.state === "speaking" || call.state === "listening" ? call.startedAt : undefined)
+  );
+}
+
 function resolveDefaultStoreBase(config: VoiceCallConfig, storePath?: string): string {
   const rawOverride = storePath?.trim() || config.store?.trim();
   if (rawOverride) {
@@ -126,11 +133,12 @@ export class CallManager {
       }
     }
 
-    // Restart max-duration timers for restored calls that are past the answered state
+    // Restart max-duration timers for restored calls that are past the answered/live state.
     let skippedAlreadyElapsedTimers = 0;
     for (const [callId, call] of verified) {
-      if (call.answeredAt && !TerminalStates.has(call.state)) {
-        const elapsed = Date.now() - call.answeredAt;
+      const maxDurationAnchor = resolveRestoredMaxDurationAnchor(call);
+      if (maxDurationAnchor !== undefined && !TerminalStates.has(call.state)) {
+        const elapsed = Date.now() - maxDurationAnchor;
         const maxDurationMs = resolveVoiceCallSecondsTimerDelayMs(this.config.maxDurationSeconds);
         if (elapsed >= maxDurationMs) {
           // Already expired — remove instead of keeping
@@ -141,6 +149,12 @@ export class CallManager {
           skippedAlreadyElapsedTimers += 1;
           continue;
         }
+        if (call.answeredAt === undefined) {
+          // Twilio streams can restore directly in speaking/listening without an
+          // answered webhook; anchoring at startedAt preserves bounded duration.
+          call.answeredAt = maxDurationAnchor;
+          persistCallRecord(this.storePath, call);
+        }
         startMaxDurationTimer({
           ctx: this.getContext(),
           callId,

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

@@ -7,13 +7,14 @@ import {
   createPluginStateSyncKeyedStoreForTests,
   resetPluginStateStoreForTests,
 } from "openclaw/plugin-sdk/plugin-state-test-runtime";
-import { afterEach, beforeEach, describe, expect, it } from "vitest";
+import { afterEach, beforeEach, describe, expect, it, vi } from "vitest";
 import { VoiceCallConfigSchema } from "../config.js";
 import type { VoiceCallProvider } from "../providers/base.js";
 import { clearVoiceCallStateRuntime, setVoiceCallStateRuntime } from "../runtime-state.js";
 import type { AnswerCallInput, HangupCallInput, NormalizedEvent } from "../types.js";
 import type { CallManagerContext } from "./context.js";
 import { processEvent } from "./events.js";
+import { speakInitialMessage } from "./outbound.js";
 import { flushPendingCallRecordWritesForTest } from "./store.js";
 
 const contexts: CallManagerContext[] = [];
@@ -54,6 +55,8 @@ afterEach(async () => {
   }
   clearVoiceCallStateRuntime();
   resetPluginStateStoreForTests();
+  vi.useRealTimers();
+  vi.restoreAllMocks();
 });
 
 function createContext(overrides: Partial<CallManagerContext> = {}): CallManagerContext {
@@ -363,6 +366,155 @@ describe("processEvent (functional)", () => {
     expect(answeredCallId).toBe("call-2");
   });
 
+  it.each([
+    {
+      name: "speaking",
+      expectedState: "speaking",
+      createEvent: (timestamp: number): NormalizedEvent => ({
+        id: "evt-live-speaking",
+        type: "call.speaking",
+        callId: "call-live",
+        providerCallId: "provider-live",
+        timestamp,
+        text: "hello",
+      }),
+    },
+    {
+      name: "listening",
+      expectedState: "listening",
+      createEvent: (timestamp: number): NormalizedEvent => ({
+        id: "evt-live-listening",
+        type: "call.speech",
+        callId: "call-live",
+        providerCallId: "provider-live",
+        timestamp,
+        transcript: "hello",
+        isFinal: true,
+      }),
+    },
+  ])(
+    "starts max-duration enforcement when $name arrives before answered",
+    async ({ expectedState, createEvent }) => {
+      const now = new Date("2026-03-22T12:00:00.000Z").getTime();
+      vi.useFakeTimers();
+      vi.setSystemTime(now);
+      const hangupCalls: HangupCallInput[] = [];
+      const ctx = createContext({
+        config: VoiceCallConfigSchema.parse({
+          enabled: true,
+          provider: "plivo",
+          fromNumber: "+15550000000",
+          maxDurationSeconds: 1,
+        }),
+        provider: createProvider({
+          hangupCall: async (input: HangupCallInput): Promise<void> => {
+            hangupCalls.push(input);
+          },
+        }),
+      });
+      ctx.activeCalls.set("call-live", {
+        callId: "call-live",
+        providerCallId: "provider-live",
+        provider: "plivo",
+        direction: "inbound",
+        state: "ringing",
+        from: "+15550000002",
+        to: "+15550000000",
+        startedAt: now - 120_000,
+        transcript: [],
+        processedEventIds: [],
+        metadata: {},
+      });
+      ctx.providerCallIdMap.set("provider-live", "call-live");
+      const liveTimestamp = now + 250;
+
+      processEvent(ctx, createEvent(liveTimestamp));
+
+      const call = ctx.activeCalls.get("call-live");
+      if (!call) {
+        throw new Error("expected live call to remain active");
+      }
+      expect(call.state).toBe(expectedState);
+      expect(call.answeredAt).toBe(liveTimestamp);
+      expect(ctx.maxDurationTimers.has("call-live")).toBe(true);
+
+      await vi.advanceTimersByTimeAsync(1_000);
+
+      expect(hangupCalls).toEqual([
+        {
+          callId: "call-live",
+          providerCallId: "provider-live",
+          reason: "timeout",
+        },
+      ]);
+      expect(ctx.activeCalls.has("call-live")).toBe(false);
+      vi.useRealTimers();
+    },
+  );
+
+  it("enforces max duration for Twilio initial-message streams without answeredAt", async () => {
+    const now = new Date("2026-03-22T12:00:00.000Z").getTime();
+    vi.useFakeTimers();
+    vi.setSystemTime(now);
+    const hangupCalls: HangupCallInput[] = [];
+    const provider = createProvider({
+      name: "twilio",
+      hangupCall: async (input: HangupCallInput): Promise<void> => {
+        hangupCalls.push(input);
+      },
+    }) as VoiceCallProvider & { isConversationStreamConnectEnabled?: () => boolean };
+    provider.isConversationStreamConnectEnabled = () => true;
+    const ctx = createContext({
+      config: VoiceCallConfigSchema.parse({
+        enabled: true,
+        provider: "twilio",
+        fromNumber: "+15550000000",
+        maxDurationSeconds: 1,
+        streaming: { enabled: true },
+      }),
+      provider,
+    });
+    ctx.activeCalls.set("call-stream", {
+      callId: "call-stream",
+      providerCallId: "provider-stream",
+      provider: "twilio",
+      direction: "inbound",
+      state: "active",
+      from: "+15550000002",
+      to: "+15550000000",
+      startedAt: now - 120_000,
+      transcript: [],
+      processedEventIds: [],
+      metadata: {
+        initialMessage: "Hello from the bot.",
+        mode: "conversation",
+      },
+    });
+    ctx.providerCallIdMap.set("provider-stream", "call-stream");
+
+    await speakInitialMessage(ctx, "provider-stream");
+
+    const call = ctx.activeCalls.get("call-stream");
+    if (!call) {
+      throw new Error("expected initial-message call to remain active");
+    }
+    expect(call.state).toBe("speaking");
+    expect(call.answeredAt).toBe(now);
+    expect(ctx.maxDurationTimers.has("call-stream")).toBe(true);
+
+    await vi.advanceTimersByTimeAsync(1_000);
+
+    expect(hangupCalls).toEqual([
+      {
+        callId: "call-stream",
+        providerCallId: "provider-stream",
+        reason: "timeout",
+      },
+    ]);
+    expect(ctx.activeCalls.has("call-stream")).toBe(false);
+    vi.useRealTimers();
+  });
+
   it("removes active call even when hangup rejects", () => {
     const provider = createProvider({
       hangupCall: async (): Promise<void> => {

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

@@ -10,7 +10,11 @@ import { findCall } from "./lookup.js";
 import { endCall } from "./outbound.js";
 import { addTranscriptEntry, transitionState } from "./state.js";
 import { persistCallRecord } from "./store.js";
-import { resolveTranscriptWaiter, startMaxDurationTimer } from "./timers.js";
+import {
+  ensureMaxDurationTimerForLiveCall,
+  resolveTranscriptWaiter,
+  startMaxDurationTimer,
+} from "./timers.js";
 
 type EventContext = Pick<
   CallManagerContext,
@@ -277,6 +281,14 @@ export function processEvent(ctx: EventContext, event: NormalizedEvent): void {
       break;
 
     case "call.speaking":
+      ensureMaxDurationTimerForLiveCall({
+        ctx,
+        call,
+        liveAt: event.timestamp,
+        onTimeout: async (callId) => {
+          await endCall(ctx, callId, { reason: "timeout" });
+        },
+      });
       transitionState(call, "speaking");
       break;
 
@@ -297,6 +309,14 @@ export function processEvent(ctx: EventContext, event: NormalizedEvent): void {
         }
         addTranscriptEntry(call, "user", event.transcript);
       }
+      ensureMaxDurationTimerForLiveCall({
+        ctx,
+        call,
+        liveAt: event.timestamp,
+        onTimeout: async (callId) => {
+          await endCall(ctx, callId, { reason: "timeout" });
+        },
+      });
       transitionState(call, "listening");
       break;
 

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

@@ -5,6 +5,7 @@ import { beforeEach, describe, expect, it, vi } from "vitest";
 const {
   addTranscriptEntryMock,
   clearMaxDurationTimerMock,
+  ensureMaxDurationTimerForLiveCallMock,
   generateDtmfRedirectTwimlMock,
   generateNotifyTwimlMock,
   getCallByProviderCallIdMock,
@@ -15,6 +16,11 @@ const {
 } = vi.hoisted(() => ({
   addTranscriptEntryMock: vi.fn(),
   clearMaxDurationTimerMock: vi.fn(),
+  ensureMaxDurationTimerForLiveCallMock: vi.fn(
+    (params: { call: { answeredAt?: number }; liveAt: number }) => {
+      params.call.answeredAt ??= params.liveAt;
+    },
+  ),
   generateDtmfRedirectTwimlMock: vi.fn(),
   generateNotifyTwimlMock: vi.fn(),
   getCallByProviderCallIdMock: vi.fn(),
@@ -36,6 +42,7 @@ vi.mock("./store.js", () => ({
 vi.mock("./timers.js", () => ({
   clearMaxDurationTimer: clearMaxDurationTimerMock,
   clearTranscriptWaiter: vi.fn(),
+  ensureMaxDurationTimerForLiveCall: ensureMaxDurationTimerForLiveCallMock,
   rejectTranscriptWaiter: rejectTranscriptWaiterMock,
   waitForFinalTranscript: vi.fn(),
 }));

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

@@ -21,7 +21,11 @@ import { getCallByProviderCallId } from "./lookup.js";
 import { addTranscriptEntry, transitionState } from "./state.js";
 import { persistCallRecord } from "./store.js";
 import { resolveVoiceCallSecondsTimerDelayMs } from "./timer-delays.js";
-import { clearTranscriptWaiter, waitForFinalTranscript } from "./timers.js";
+import {
+  clearTranscriptWaiter,
+  ensureMaxDurationTimerForLiveCall,
+  waitForFinalTranscript,
+} from "./timers.js";
 import { generateDtmfRedirectTwiml, generateNotifyTwiml } from "./twiml.js";
 
 type InitiateContext = Pick<
@@ -37,7 +41,13 @@ type InitiateContext = Pick<
 
 type SpeakContext = Pick<
   CallManagerContext,
-  "activeCalls" | "providerCallIdMap" | "provider" | "config" | "storePath"
+  | "activeCalls"
+  | "providerCallIdMap"
+  | "provider"
+  | "config"
+  | "storePath"
+  | "transcriptWaiters"
+  | "maxDurationTimers"
 >;
 
 type ConversationContext = Pick<
@@ -267,6 +277,14 @@ export async function speak(
   const { call, providerCallId, provider } = connected;
 
   try {
+    ensureMaxDurationTimerForLiveCall({
+      ctx,
+      call,
+      liveAt: Date.now(),
+      onTimeout: async (id) => {
+        await endCall(ctx, id, { reason: "timeout" });
+      },
+    });
     transitionState(call, "speaking");
     persistCallRecord(ctx.storePath, call);
 

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

@@ -1,5 +1,5 @@
 // Voice Call plugin module implements timers behavior.
-import { TerminalStates, type CallId } from "../types.js";
+import { TerminalStates, type CallId, type CallRecord } from "../types.js";
 import type { CallManagerContext } from "./context.js";
 import { persistCallRecord } from "./store.js";
 import {
@@ -67,6 +67,28 @@ export function startMaxDurationTimer(params: {
   params.ctx.maxDurationTimers.set(params.callId, timer);
 }
 
+/** Backfill max-duration enforcement from the first live conversation signal. */
+export function ensureMaxDurationTimerForLiveCall(params: {
+  ctx: MaxDurationTimerContext;
+  call: CallRecord;
+  liveAt: number;
+  onTimeout: (callId: CallId) => Promise<void>;
+}): void {
+  if (params.call.answeredAt) {
+    return;
+  }
+
+  // Realtime streams can prove the call is live before an answered callback;
+  // use that first live signal so stale cleanup can skip it without losing
+  // maxDurationSeconds enforcement.
+  params.call.answeredAt = params.liveAt;
+  startMaxDurationTimer({
+    ctx: params.ctx,
+    callId: params.call.callId,
+    onTimeout: params.onTimeout,
+  });
+}
+
 /** Clear and forget a pending final-transcript waiter. */
 export function clearTranscriptWaiter(ctx: TranscriptWaiterContext, callId: CallId): void {
   const waiter = ctx.transcriptWaiters.get(callId);

extensions/voice-call/src/webhook/stale-call-reaper.test.ts

@@ -56,6 +56,34 @@ describe("startStaleCallReaper", () => {
     stop?.();
   });
 
+  it.each(["speaking", "listening"] as const)(
+    "does not reap live %s calls without answeredAt",
+    async (state) => {
+      const endCall = vi.fn(async () => {});
+      const manager = {
+        getActiveCalls: vi.fn(() => [
+          {
+            callId: `call-${state}`,
+            startedAt: Date.now() - 120_000,
+            state,
+          },
+        ]),
+        endCall,
+      };
+
+      const stop = startStaleCallReaper({
+        manager: manager as never,
+        staleCallReaperSeconds: 60,
+      });
+
+      await vi.advanceTimersByTimeAsync(30_000);
+
+      expect(endCall).not.toHaveBeenCalled();
+
+      stop?.();
+    },
+  );
+
   it("logs and swallows endCall failures", async () => {
     const warn = vi.spyOn(console, "warn").mockImplementation(() => {});
     const endCallError = new Error("network");

extensions/voice-call/src/webhook/stale-call-reaper.ts

@@ -1,11 +1,18 @@
 // Voice Call plugin module implements stale call reaper behavior.
 import type { CallManager } from "../manager.js";
+import type { CallState } from "../types.js";
 import { TerminalStates } from "../types.js";
 
 // Background cleanup loop for calls that never reached answered/terminal state.
 
 const CHECK_INTERVAL_MS = 30_000;
 
+/** States that indicate a live conversation with speech/transcription.
+ * Inbound Twilio calls may never fire a call.answered event, so answeredAt
+ * can be absent even while the call is actively transcribing. These states
+ * prove the call is live and should not be reaped. */
+const LiveConversationStates: ReadonlySet<CallState> = new Set(["speaking", "listening"]);
+
 /** Start a stale-call reaper and return its cleanup callback. */
 export function startStaleCallReaper(params: {
   manager: CallManager;
@@ -20,7 +27,16 @@ export function startStaleCallReaper(params: {
   const interval = setInterval(() => {
     const now = Date.now();
     for (const call of params.manager.getActiveCalls()) {
-      if (call.answeredAt || TerminalStates.has(call.state)) {
+      // Skip calls that have been answered (answeredAt set) or are in a live
+      // conversation state. Inbound Twilio calls may never fire a call.answered
+      // event so answeredAt may be absent even when the call is actively
+      // transcribing/responding. Without this state guard live calls in
+      // speaking/listening state get reaped as stale.
+      if (
+        call.answeredAt ||
+        TerminalStates.has(call.state) ||
+        LiveConversationStates.has(call.state)
+      ) {
         continue;
       }