fix(heartbeat): deliver structured replies and clear completed recovery (#119655)

* fix(heartbeat): preserve structured replies and delivery recovery

* test(heartbeat): isolate structured delivery regressions

src/infra/heartbeat-runner-delivery.ts

@@ -126,9 +126,19 @@ export function classifyHeartbeatAgentOutcome(params: {
     normalized.text = replacement.text;
     normalized.shouldSkip = false;
   }
+  const hasStructuredReplyContent =
+    !heartbeatToolResponse &&
+    replyPayload !== undefined &&
+    hasOutboundReplyContent({
+      ...replyPayload,
+      text: undefined,
+      mediaUrl: undefined,
+      mediaUrls: undefined,
+    });
   const shouldSkipMain =
     normalized.shouldSkip &&
     !normalized.hasMedia &&
+    (!hasStructuredReplyContent || normalized.isInternalPlaceholderOnly) &&
     (!params.hasRelayableExecCompletion || normalized.isInternalPlaceholderOnly);
   if (heartbeatTerminalToolFailure) {
     return {
@@ -147,6 +157,8 @@ export function classifyHeartbeatAgentOutcome(params: {
     kind: "delivery",
     normalized,
     deliveredAgentRunFailure,
+    hasStructuredReplyContent,
+    replyPayload: heartbeatToolResponse ? undefined : replyPayload,
     mediaUrls:
       heartbeatToolResponse || !replyPayload
         ? []
@@ -288,7 +300,13 @@ export async function finalizeHeartbeatOutcome(params: {
     consumeInspectedSystemEvents(params.wake, params.prepared);
     return { status: "ran", durationMs: Date.now() - startedAt };
   }
-  const { deliveredAgentRunFailure, mediaUrls, normalized } = outcome;
+  const {
+    deliveredAgentRunFailure,
+    hasStructuredReplyContent,
+    mediaUrls,
+    normalized,
+    replyPayload,
+  } = outcome;
   // Suppress duplicate heartbeats (same payload) within a short window.
   // This prevents "nagging" when nothing changed but the model repeats the same items.
   const prevHeartbeatText =
@@ -297,9 +315,12 @@ export async function finalizeHeartbeatOutcome(params: {
     typeof entry?.lastHeartbeatSentAt === "number" ? entry.lastHeartbeatSentAt : undefined;
   const isDuplicateMain =
     !mediaUrls.length &&
+    !hasStructuredReplyContent &&
     Boolean(prevHeartbeatText.trim()) &&
     normalized.text.trim() === prevHeartbeatText.trim() &&
     typeof prevHeartbeatAt === "number" &&
+    // A future timestamp after clock rollback cannot prove a recent prior send.
+    prevHeartbeatAt <= startedAt &&
     startedAt - prevHeartbeatAt < 24 * 60 * 60 * 1000;
 
   if (isDuplicateMain) {
@@ -382,7 +403,13 @@ export async function finalizeHeartbeatOutcome(params: {
     session: params.outboundSession,
     identity: params.outboundIdentity,
     threadId: delivery.threadId,
-    payloads: [{ text: normalized.text, mediaUrls }],
+    payloads: [
+      copyReplyPayloadMetadata(replyPayload ?? {}, {
+        ...replyPayload,
+        text: normalized.text,
+        mediaUrls,
+      }),
+    ],
     deps: params.opts.deps,
     silent: normalized.silent,
   });
@@ -394,27 +421,24 @@ export async function finalizeHeartbeatOutcome(params: {
   // commitments and heartbeat dedupe state active so a later heartbeat can retry.
   if (visibleSendSucceeded) {
     await markDueCommitments("sent");
-  }
-
-  // Record last delivered heartbeat payload for dedupe.
-  if (visibleSendSucceeded && normalized.text.trim()) {
+    const hasHeartbeatText = Boolean(normalized.text.trim());
     await patchSessionEntry(
       { storePath, sessionKey },
       (current, context) => {
         if (!context.existingEntry) {
           return null;
         }
-        // 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
-          : {};
+        // Visible structured-only sends satisfy their own pending final too;
+        // preserve old text dedupe markers and another run's recovery state.
+        const ownsPendingFinalDelivery = heartbeatRunOwnsPendingFinalDelivery(current, startedAt);
+        if (!hasHeartbeatText && !ownsPendingFinalDelivery) {
+          return null;
+        }
         return {
-          lastHeartbeatText: normalized.text,
-          lastHeartbeatSentAt: startedAt,
-          ...clearedRecoveryFields,
+          ...(hasHeartbeatText
+            ? { lastHeartbeatText: normalized.text, lastHeartbeatSentAt: startedAt }
+            : {}),
+          ...(ownsPendingFinalDelivery ? CLEARED_PENDING_FINAL_DELIVERY_FIELDS : {}),
         };
       },
       { preserveActivity: true },

src/infra/heartbeat-runner.returns-default-unset.test.ts

@@ -1255,6 +1255,53 @@ describe("runHeartbeatOnce", () => {
     }
   });
 
+  it("delivers a repeated heartbeat when the clock moves behind its previous send", async () => {
+    const tmpDir = await createCaseDir("hb-dup-clock-rollback");
+    const storePath = path.join(tmpDir, "sessions.json");
+    const replySpy = vi.fn();
+    try {
+      const cfg: OpenClawConfig = {
+        agents: {
+          defaults: {
+            workspace: tmpDir,
+            heartbeat: { every: "5m", target: "whatsapp" },
+          },
+        },
+        channels: { whatsapp: { allowFrom: ["*"] } },
+        session: { store: storePath },
+      };
+      const sessionKey = resolveMainSessionKey(cfg);
+      const nowMs = 60_000;
+      await seedWhatsAppSession(storePath, sessionKey, {
+        lastHeartbeatText: "Final alert",
+        lastHeartbeatSentAt: nowMs + 60_000,
+      });
+      replySpy.mockResolvedValue([{ text: "Final alert" }]);
+      const sendWhatsApp = vi
+        .fn<
+          (
+            to: string,
+            text: string,
+            opts?: unknown,
+          ) => Promise<{ messageId: string; toJid: string }>
+        >()
+        .mockResolvedValue({ messageId: "m1", toJid: "jid" });
+
+      await runHeartbeatOnce({
+        cfg,
+        deps: createHeartbeatDeps(sendWhatsApp, { nowMs, getReplyFromConfig: replySpy }),
+      });
+
+      expect(sendWhatsApp).toHaveBeenCalledOnce();
+      expectWhatsAppSendCall(sendWhatsApp, 0, {
+        to: "120363401234567890@g.us",
+        text: "Final alert",
+      });
+    } finally {
+      replySpy.mockReset();
+    }
+  });
+
   it.each(
     typedCases<{
       name: string;

src/infra/heartbeat-runner.structured-delivery.test.ts

@@ -0,0 +1,247 @@
+// Covers structured heartbeat delivery, text-only dedupe, and recovery ownership.
+import { describe, expect, it, vi } from "vitest";
+import { setReplyPayloadMetadata } from "../auto-reply/reply-payload.js";
+import type { OpenClawConfig } from "../config/config.js";
+import { patchSessionEntry } from "../config/sessions/session-accessor.js";
+import type { SessionEntry } from "../config/sessions/types.js";
+import { setActivePluginRegistry } from "../plugins/runtime.js";
+import { createOutboundTestPlugin, createTestRegistry } from "../test-utils/channel-plugins.js";
+import { runHeartbeatOnce, type HeartbeatDeps } from "./heartbeat-runner.js";
+import { installHeartbeatRunnerTestRuntime } from "./heartbeat-runner.test-harness.js";
+import {
+  readSessionStoreForTest,
+  seedMainSessionStore,
+  withTempTelegramHeartbeatSandbox,
+} from "./heartbeat-runner.test-utils.js";
+
+installHeartbeatRunnerTestRuntime();
+
+describe("runHeartbeatOnce structured heartbeat delivery", () => {
+  const TELEGRAM_GROUP = "-1001234567890";
+
+  function createConfig(tmpDir: string, storePath: string): OpenClawConfig {
+    return {
+      agents: {
+        defaults: {
+          workspace: tmpDir,
+          heartbeat: { every: "5m", target: "telegram" },
+        },
+      },
+      messages: { visibleReplies: "automatic" },
+      channels: {
+        telegram: {
+          token: "test-token",
+          allowFrom: ["*"],
+          heartbeat: { showOk: false },
+        },
+      },
+      session: { store: storePath },
+    } as OpenClawConfig;
+  }
+
+  function seedTelegramSession(
+    storePath: string,
+    cfg: OpenClawConfig,
+    entry: Partial<Parameters<typeof seedMainSessionStore>[2]> = {},
+  ) {
+    return seedMainSessionStore(storePath, cfg, {
+      lastChannel: "telegram",
+      lastProvider: "telegram",
+      lastTo: TELEGRAM_GROUP,
+      ...entry,
+    });
+  }
+
+  function runHeartbeat(
+    cfg: OpenClawConfig,
+    replySpy: HeartbeatDeps["getReplyFromConfig"],
+    sendTelegram: ReturnType<typeof vi.fn>,
+  ) {
+    return runHeartbeatOnce({
+      cfg,
+      deps: {
+        telegram: sendTelegram as unknown,
+        getQueueSize: () => 0,
+        nowMs: () => 0,
+        getReplyFromConfig: replySpy,
+      },
+    });
+  }
+
+  it("delivers presentation-only heartbeat replies with their button fallback", async () => {
+    await withTempTelegramHeartbeatSandbox(async ({ tmpDir, storePath, replySpy }) => {
+      const cfg = createConfig(tmpDir, storePath);
+      await seedTelegramSession(storePath, cfg);
+      replySpy.mockResolvedValue({
+        presentation: {
+          blocks: [
+            { type: "text", text: "Deployment approval required." },
+            {
+              type: "buttons",
+              buttons: [{ label: "Approve deployment", value: "approve" }],
+            },
+          ],
+        },
+      });
+      const sendTelegram = vi.fn().mockResolvedValue({ messageId: "presentation-1" });
+
+      const result = await runHeartbeat(cfg, replySpy, sendTelegram);
+
+      expect(result.status).toBe("ran");
+      expect(sendTelegram).toHaveBeenCalledOnce();
+      expect(sendTelegram.mock.calls[0]?.[0]).toBe(TELEGRAM_GROUP);
+      expect(sendTelegram.mock.calls[0]?.[1]).toContain("Deployment approval required.");
+      expect(sendTelegram.mock.calls[0]?.[1]).toContain("Approve deployment");
+    });
+  });
+
+  it("delivers changed heartbeat actions when their visible text matches the previous send", async () => {
+    await withTempTelegramHeartbeatSandbox(async ({ tmpDir, storePath, replySpy }) => {
+      const cfg = createConfig(tmpDir, storePath);
+      await seedTelegramSession(storePath, cfg);
+      const text = "Deployment approval required.";
+      replySpy
+        .mockResolvedValueOnce({
+          text,
+          presentation: {
+            blocks: [
+              {
+                type: "buttons",
+                buttons: [{ label: "Review deployment", value: "review" }],
+              },
+            ],
+          },
+        })
+        .mockResolvedValueOnce({
+          text,
+          presentation: {
+            blocks: [
+              {
+                type: "buttons",
+                buttons: [{ label: "Approve deployment", value: "approve" }],
+              },
+            ],
+          },
+        });
+      const sendTelegram = vi.fn().mockResolvedValue({ messageId: "presentation-1" });
+
+      await runHeartbeat(cfg, replySpy, sendTelegram);
+      await runHeartbeat(cfg, replySpy, sendTelegram);
+
+      expect(sendTelegram).toHaveBeenCalledTimes(2);
+      expect(sendTelegram.mock.calls[0]?.[1]).toContain("Review deployment");
+      expect(sendTelegram.mock.calls[1]?.[1]).toContain("Approve deployment");
+    });
+  });
+
+  it("clears a run-owned transport-only pending final after a presentation-only send", async () => {
+    await withTempTelegramHeartbeatSandbox(async ({ tmpDir, storePath, replySpy }) => {
+      const cfg = createConfig(tmpDir, storePath);
+      const previousText = "Previous successful heartbeat";
+      const previousSentAt = 0;
+      const sessionKey = await seedTelegramSession(storePath, cfg, {
+        lastHeartbeatText: previousText,
+        lastHeartbeatSentAt: previousSentAt,
+      });
+      replySpy.mockImplementation(async () => {
+        await patchSessionEntry(
+          { storePath, sessionKey },
+          () => ({
+            pendingFinalDelivery: {
+              kind: "transport-only",
+              createdAt: 0,
+              intentId: "structured-heartbeat-intent",
+            },
+          }),
+          { preserveActivity: true },
+        );
+        return {
+          presentation: {
+            blocks: [
+              {
+                type: "buttons",
+                buttons: [{ label: "Approve deployment", value: "approve" }],
+              },
+            ],
+          },
+        };
+      });
+      const sendTelegram = vi.fn().mockResolvedValue({ messageId: "presentation-1" });
+
+      const result = await runHeartbeat(cfg, replySpy, sendTelegram);
+
+      expect(result.status).toBe("ran");
+      expect(sendTelegram).toHaveBeenCalledOnce();
+      expect(sendTelegram.mock.calls[0]?.[1]).toContain("Approve deployment");
+      const sessionStore = readSessionStoreForTest<{
+        pendingFinalDelivery?: SessionEntry["pendingFinalDelivery"];
+        lastHeartbeatText?: string;
+        lastHeartbeatSentAt?: number;
+      }>(storePath);
+      expect(sessionStore[sessionKey]).toMatchObject({
+        lastHeartbeatText: previousText,
+        lastHeartbeatSentAt: previousSentAt,
+      });
+      expect(sessionStore[sessionKey]?.pendingFinalDelivery).toBeUndefined();
+    });
+  });
+
+  it("preserves heartbeat reply metadata, channel data, and voice delivery", async () => {
+    await withTempTelegramHeartbeatSandbox(async ({ tmpDir, storePath, replySpy }) => {
+      const cfg = createConfig(tmpDir, storePath);
+      await seedTelegramSession(storePath, cfg);
+      const sendPayload = vi.fn().mockResolvedValue({
+        channel: "telegram",
+        messageId: "metadata-1",
+      });
+      setActivePluginRegistry(
+        createTestRegistry([
+          {
+            pluginId: "telegram",
+            source: "test",
+            plugin: createOutboundTestPlugin({
+              id: "telegram",
+              outbound: {
+                deliveryMode: "direct",
+                sendText: vi.fn().mockResolvedValue({ messageId: "text-1" }),
+                sendPayload,
+              },
+            }),
+          },
+        ]),
+      );
+      const mediaUrl = "https://example.test/heartbeat.ogg";
+      const channelData = {
+        telegram: {
+          buttons: [[{ text: "Open deployment", callback_data: "open" }]],
+        },
+      };
+      replySpy.mockResolvedValue(
+        setReplyPayloadMetadata(
+          {
+            text: "Deployment update",
+            mediaUrl,
+            replyToId: "42",
+            audioAsVoice: true,
+            channelData,
+          },
+          { replyToIdExplicit: true },
+        ),
+      );
+      const sendTelegram = vi.fn().mockResolvedValue({ messageId: "unused-1" });
+
+      const result = await runHeartbeat(cfg, replySpy, sendTelegram);
+
+      expect(result.status).toBe("ran");
+      expect(sendPayload).toHaveBeenCalledOnce();
+      const deliveredPayload = sendPayload.mock.calls[0]?.[0]?.payload;
+      expect(deliveredPayload).toMatchObject({
+        text: "Deployment update",
+        mediaUrl,
+        replyToId: "42",
+        audioAsVoice: true,
+        channelData,
+      });
+    });
+  });
+});