fix(heartbeat): only clear pendingFinalDelivery this run owns

agocs · claude · agocs · commit 675dcd8ab6e6 · 2026-06-23T06:20:03.000Z
The send-success and duplicate-skip clears retired any pendingFinalDelivery unconditionally, so a heartbeat could erase an older, unsatisfied user-facing final it never produced (e.g. a message_tool_only run that does not refresh pending). Gate both clears on ownership via a shared helper: clear only when pendingFinalDeliveryCreatedAt is at/after the run start. Add regression tests covering the older-pending preserve case on both clear paths. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01SkPieH67337ieQPWVXQgMD
diff --git a/src/infra/heartbeat-runner.clears-pending-final-delivery.test.ts b/src/infra/heartbeat-runner.clears-pending-final-delivery.test.ts
@@ -37,11 +37,14 @@ describe("runHeartbeatOnce clears stuck pendingFinalDelivery state once delivery
   function heartbeatDeps(
     sendTelegram: ReturnType<typeof vi.fn>,
     replySpy: HeartbeatReplySpy,
+    now?: number,
   ): HeartbeatDeps {
     return {
       telegram: sendTelegram as unknown,
       getQueueSize: () => 0,
-      nowMs: () => Date.now(),
+      // A fixed clock lets a test seed pendingFinalDeliveryCreatedAt relative to
+      // the run's startedAt, which is what the ownership guard compares against.
+      nowMs: () => now ?? Date.now(),
       getReplyFromConfig: replySpy,
     } satisfies HeartbeatDeps;
   }
@@ -78,23 +81,25 @@ describe("runHeartbeatOnce clears stuck pendingFinalDelivery state once delivery
   it("nulls every pendingFinalDelivery* field after delivering substantive heartbeat content", async () => {
     await withTempHeartbeatSandbox(async ({ storePath, replySpy }) => {
       const cfg = createHeartbeatConfig(storePath);
+      const NOW = Date.now();
 
-      // Seed a session that carries a stuck pendingFinalDelivery from a prior run.
-      // pendingFinalDeliveryText is a heartbeat-ack token so the pendingFinalDelivery
-      // gate at heartbeat-runner.ts:~1390 does not bail before the send.
+      // Seed a stuck pendingFinalDelivery this run owns: createdAt at run start
+      // marks it as produced by this heartbeat (the case the original fix
+      // targets). pendingFinalDeliveryText is a heartbeat-ack token so the
+      // pendingFinalDelivery defer gate does not bail before the send.
       const sessionKey = await seedMainSessionStore(storePath, cfg, {
         lastChannel: "telegram",
         lastProvider: "telegram",
         lastTo: TELEGRAM_GROUP,
-        updatedAt: Date.now(),
+        updatedAt: NOW,
         pendingFinalDelivery: true,
         pendingFinalDeliveryText: "HEARTBEAT_OK",
-        pendingFinalDeliveryCreatedAt: 1,
+        pendingFinalDeliveryCreatedAt: NOW,
         pendingFinalDeliveryAttemptCount: 3,
         pendingFinalDeliveryLastError: "prior-error",
       });
       await patchEntry(storePath, sessionKey, {
-        pendingFinalDeliveryLastAttemptAt: 2,
+        pendingFinalDeliveryLastAttemptAt: NOW,
         pendingFinalDeliveryContext: { foo: "bar" },
         pendingFinalDeliveryIntentId: "intent-send-success",
       });
@@ -105,7 +110,10 @@ describe("runHeartbeatOnce clears stuck pendingFinalDelivery state once delivery
       replySpy.mockResolvedValue({ text: replyText });
       const sendTelegram = vi.fn().mockResolvedValue({ messageId: "m1", toJid: "jid" });
 
-      const result = await runHeartbeatOnce({ cfg, deps: heartbeatDeps(sendTelegram, replySpy) });
+      const result = await runHeartbeatOnce({
+        cfg,
+        deps: heartbeatDeps(sendTelegram, replySpy, NOW),
+      });
 
       expect(result.status).toBe("ran");
       expect(sendTelegram).toHaveBeenCalledTimes(1);
@@ -136,10 +144,11 @@ describe("runHeartbeatOnce clears stuck pendingFinalDelivery state once delivery
 
       const body = "Heartbeat update: everything is green.";
       const deliveredText = `🤖 ${body}`;
+      const NOW = Date.now();
       // updatedAt is well past the 30s defer window so the pendingFinalDelivery
       // gate does not bail before the duplicate branch (the pending text is
       // substantive, not a heartbeat ack).
-      const staleAt = Date.now() - 60_000;
+      const staleAt = NOW - 60_000;
 
       const sessionKey = await seedMainSessionStore(storePath, cfg, {
         lastChannel: "telegram",
@@ -148,7 +157,9 @@ describe("runHeartbeatOnce clears stuck pendingFinalDelivery state once delivery
         updatedAt: staleAt,
         pendingFinalDelivery: true,
         pendingFinalDeliveryText: body, // prefix-less; diverges from deliveredText
-        pendingFinalDeliveryCreatedAt: 1,
+        // createdAt at run start: this run produced the pending (real runs stamp
+        // a fresh createdAt during the agent turn the defer gate ran ahead of).
+        pendingFinalDeliveryCreatedAt: NOW,
         pendingFinalDeliveryAttemptCount: 3,
         pendingFinalDeliveryLastError: "prior-error",
       });
@@ -157,7 +168,7 @@ describe("runHeartbeatOnce clears stuck pendingFinalDelivery state once delivery
         // which is what makes this run a duplicate and the pending clear safe.
         lastHeartbeatText: deliveredText,
         lastHeartbeatSentAt: staleAt,
-        pendingFinalDeliveryLastAttemptAt: 2,
+        pendingFinalDeliveryLastAttemptAt: NOW,
         pendingFinalDeliveryContext: { foo: "bar" },
         pendingFinalDeliveryIntentId: "intent-duplicate-skip",
       });
@@ -167,7 +178,10 @@ describe("runHeartbeatOnce clears stuck pendingFinalDelivery state once delivery
       replySpy.mockResolvedValue({ text: body });
       const sendTelegram = vi.fn().mockResolvedValue({ messageId: "m1", toJid: "jid" });
 
-      const result = await runHeartbeatOnce({ cfg, deps: heartbeatDeps(sendTelegram, replySpy) });
+      const result = await runHeartbeatOnce({
+        cfg,
+        deps: heartbeatDeps(sendTelegram, replySpy, NOW),
+      });
 
       // status "ran" (not "skipped") proves the run reached the duplicate branch
       // rather than bailing at the pendingFinalDelivery gate.
@@ -181,4 +195,104 @@ describe("runHeartbeatOnce clears stuck pendingFinalDelivery state once delivery
       expectPendingFinalDeliveryCleared(entry);
     });
   });
+
+  it("preserves an older unsatisfied pendingFinalDelivery the heartbeat send did not create", async () => {
+    await withTempHeartbeatSandbox(async ({ storePath, replySpy }) => {
+      const cfg = createHeartbeatConfig(storePath);
+      const NOW = Date.now();
+      // An older user-facing final that failed delivery and still owns its own
+      // get-reply redelivery recovery path. It predates this run (createdAt <
+      // startedAt) and a message_tool_only/response-tool heartbeat would not
+      // refresh it, so the send-success clear must NOT retire it.
+      const olderCreatedAt = NOW - 60_000;
+      const olderText = "Older final the user never received";
+      const sessionKey = await seedMainSessionStore(storePath, cfg, {
+        lastChannel: "telegram",
+        lastProvider: "telegram",
+        lastTo: TELEGRAM_GROUP,
+        // Stale so the substantive-pending defer gate does not bail this run.
+        updatedAt: NOW - 60_000,
+        pendingFinalDelivery: true,
+        pendingFinalDeliveryText: olderText,
+        pendingFinalDeliveryCreatedAt: olderCreatedAt,
+        pendingFinalDeliveryAttemptCount: 2,
+        pendingFinalDeliveryLastError: "prior-delivery-failure",
+      });
+      await patchEntry(storePath, sessionKey, {
+        pendingFinalDeliveryLastAttemptAt: NOW - 50_000,
+        pendingFinalDeliveryContext: { channel: "telegram", to: "older-chat" },
+        pendingFinalDeliveryIntentId: "intent-older-unsatisfied",
+      });
+
+      // A fresh, different heartbeat payload that gets delivered this run.
+      const replyText = "Fresh heartbeat content unrelated to the older final.";
+      replySpy.mockResolvedValue({ text: replyText });
+      const sendTelegram = vi.fn().mockResolvedValue({ messageId: "m1", toJid: "jid" });
+
+      const result = await runHeartbeatOnce({
+        cfg,
+        deps: heartbeatDeps(sendTelegram, replySpy, NOW),
+      });
+
+      expect(result.status).toBe("ran");
+      expect(sendTelegram).toHaveBeenCalledTimes(1);
+
+      const entry = await readEntry(storePath, sessionKey);
+      // Send-success records the dedupe markers for the delivered payload...
+      expect(entry?.lastHeartbeatText).toBe(replyText);
+      // ...but the older, unowned pending-final survives for its own recovery.
+      expect(entry?.pendingFinalDelivery).toBe(true);
+      expect(entry?.pendingFinalDeliveryText).toBe(olderText);
+      expect(entry?.pendingFinalDeliveryCreatedAt).toBe(olderCreatedAt);
+      expect(entry?.pendingFinalDeliveryIntentId).toBe("intent-older-unsatisfied");
+    });
+  });
+
+  it("preserves an older unowned pendingFinalDelivery on a duplicate skip", async () => {
+    await withTempHeartbeatSandbox(async ({ storePath, replySpy }) => {
+      const cfg = createHeartbeatConfig(storePath);
+      const NOW = Date.now();
+      const body = "Recurring heartbeat status line.";
+      const olderCreatedAt = NOW - 60_000;
+      const olderText = "A different older final still awaiting delivery";
+      const sessionKey = await seedMainSessionStore(storePath, cfg, {
+        lastChannel: "telegram",
+        lastProvider: "telegram",
+        lastTo: TELEGRAM_GROUP,
+        updatedAt: NOW - 60_000,
+        pendingFinalDelivery: true,
+        pendingFinalDeliveryText: olderText,
+        pendingFinalDeliveryCreatedAt: olderCreatedAt,
+        pendingFinalDeliveryAttemptCount: 2,
+        pendingFinalDeliveryLastError: "prior-delivery-failure",
+      });
+      await patchEntry(storePath, sessionKey, {
+        // Same payload already delivered within 24h -> this run is a duplicate skip.
+        lastHeartbeatText: body,
+        lastHeartbeatSentAt: NOW - 60_000,
+        pendingFinalDeliveryLastAttemptAt: NOW - 50_000,
+        pendingFinalDeliveryContext: { channel: "telegram", to: "older-chat" },
+        pendingFinalDeliveryIntentId: "intent-older-dupe",
+      });
+
+      replySpy.mockResolvedValue({ text: body });
+      const sendTelegram = vi.fn().mockResolvedValue({ messageId: "m1", toJid: "jid" });
+
+      const result = await runHeartbeatOnce({
+        cfg,
+        deps: heartbeatDeps(sendTelegram, replySpy, NOW),
+      });
+
+      expect(result.status).toBe("ran");
+      // Duplicate payload: nothing is sent this run.
+      expect(sendTelegram).not.toHaveBeenCalled();
+
+      const entry = await readEntry(storePath, sessionKey);
+      // The duplicate-skip clear must not retire the older, unowned pending-final.
+      expect(entry?.pendingFinalDelivery).toBe(true);
+      expect(entry?.pendingFinalDeliveryText).toBe(olderText);
+      expect(entry?.pendingFinalDeliveryCreatedAt).toBe(olderCreatedAt);
+      expect(entry?.pendingFinalDeliveryIntentId).toBe("intent-older-dupe");
+    });
+  });
 });
diff --git a/src/infra/heartbeat-runner.ts b/src/infra/heartbeat-runner.ts
@@ -1322,6 +1322,17 @@ const CLEARED_PENDING_FINAL_DELIVERY_FIELDS = {
   pendingFinalDeliveryIntentId: undefined,
 } as const;
 
+// Clear pending-final only when this run produced it: the agent run stamps
+// createdAt during the run, so createdAt >= run start means we own it. An older
+// final (e.g. one a message_tool_only run never refreshed) must keep its recovery path.
+function heartbeatRunOwnsPendingFinalDelivery(
+  entry: SessionEntry | undefined,
+  runStartedAt: number,
+): boolean {
+  const createdAt = entry?.pendingFinalDeliveryCreatedAt;
+  return typeof createdAt === "number" && createdAt >= runStartedAt;
+}
+
 export async function runHeartbeatOnce(opts: {
   cfg?: OpenClawConfig;
   agentId?: string;
@@ -1705,13 +1716,13 @@ export async function runHeartbeatOnce(opts: {
   };
 
   // The duplicate-suppression branch returns before any send, so it never hits
-  // the send-success clear. A duplicate means this exact payload was already
-  // delivered within the dedupe window, and the agent run refreshed
-  // pendingFinalDelivery to this run's output, so the recovery state is
-  // satisfied. Clear it the same way the send-success path does. We must not
+  // the send-success clear. A duplicate means this run's own output was already
+  // delivered within the dedupe window, so this run's pending-final is satisfied
+  // and gets cleared the same way the send-success path does. We must not
   // text-match the pending against the delivered text: agent-runner stores it
   // pre-normalization (no responsePrefix), so a byte compare would leave
-  // prefixed agents permanently stuck.
+  // prefixed agents permanently stuck. Ownership is gated on createdAt instead,
+  // so an older final this run did not produce is preserved, not erased.
   const clearSatisfiedPendingFinalDelivery = async () => {
     await updateSessionStore(
       storePath,
@@ -1720,6 +1731,9 @@ export async function runHeartbeatOnce(opts: {
         if (current?.pendingFinalDelivery !== true && !current?.pendingFinalDeliveryText) {
           return false;
         }
+        if (!heartbeatRunOwnsPendingFinalDelivery(current, startedAt)) {
+          return false;
+        }
         store[sessionKey] = { ...current, ...CLEARED_PENDING_FINAL_DELIVERY_FIELDS };
         return true;
       },
@@ -2161,13 +2175,18 @@ export async function runHeartbeatOnce(opts: {
         if (!current) {
           return;
         }
+        // A heartbeat-driven agent run can leave its own pendingFinalDelivery
+        // set; a successful send completes it, so clear the recovery fields.
+        // Only clear the pending-final this run owns — an older final the run
+        // did not produce keeps its own recovery path.
+        const clearedRecoveryFields = heartbeatRunOwnsPendingFinalDelivery(current, startedAt)
+          ? CLEARED_PENDING_FINAL_DELIVERY_FIELDS
+          : {};
         store[sessionKey] = {
           ...current,
           lastHeartbeatText: normalized.text,
           lastHeartbeatSentAt: startedAt,
-          // Heartbeat-driven agent runs can leave pendingFinalDelivery set; a
-          // successful send completes it, so clear the recovery fields.
-          ...CLEARED_PENDING_FINAL_DELIVERY_FIELDS,
+          ...clearedRecoveryFields,
         };
       });
     }
