diff --git a/src/infra/heartbeat-runner-delivery.ts b/src/infra/heartbeat-runner-delivery.ts index df175a7c560d..282e18881e2f 100644 --- a/src/infra/heartbeat-runner-delivery.ts +++ b/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 }, diff --git a/src/infra/heartbeat-runner.returns-default-unset.test.ts b/src/infra/heartbeat-runner.returns-default-unset.test.ts index a3c76f960370..7434857afd7c 100644 --- a/src/infra/heartbeat-runner.returns-default-unset.test.ts +++ b/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; diff --git a/src/infra/heartbeat-runner.structured-delivery.test.ts b/src/infra/heartbeat-runner.structured-delivery.test.ts new file mode 100644 index 000000000000..a20fb29bed83 --- /dev/null +++ b/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[2]> = {}, + ) { + return seedMainSessionStore(storePath, cfg, { + lastChannel: "telegram", + lastProvider: "telegram", + lastTo: TELEGRAM_GROUP, + ...entry, + }); + } + + function runHeartbeat( + cfg: OpenClawConfig, + replySpy: HeartbeatDeps["getReplyFromConfig"], + sendTelegram: ReturnType, + ) { + 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, + }); + }); + }); +});