Skip to content

Commit 305fa9c

Browse files
agocsclaude
agocs
and
claude
authored
fix(heartbeat): clear pendingFinalDelivery* on send success (#83187)
* fix(infra): clear pendingFinalDelivery* after successful heartbeat send Heartbeat-driven agent runs can leave pendingFinalDelivery set on the session; runHeartbeatOnce wrote lastHeartbeatText/lastHeartbeatSentAt on send success but never cleared the recovery fields, so the next heartbeat hit the get-reply redelivery short-circuit and skipped the agent. Extend the existing post-success updateSessionStore block to null all seven pendingFinalDelivery* fields, mirroring clearPendingFinalDeliveryAfterSuccess in dispatch-from-config.ts. Refs: #83184 * fix(heartbeat): clear pendingFinalDelivery* on duplicate-skip too The duplicate-suppression branch returns before any send, so it never hit the send-success clear, leaving pendingFinalDelivery* recovery state stuck on a payload that was already delivered. Clear it there too, unconditionally (mirroring the send-success path and the canonical clearPendingFinalDeliveryAfterSuccess via a shared field set). A byte-equal text match would no-op for responsePrefix-configured agents, since agent-runner stores the pending text pre-normalization while the delivered text re-adds the prefix; the duplicate check already proves the payload was delivered, so the clear is unconditional. Adds a regression test covering the prefix-divergence case. Co-Authored-By: Claude Opus 4.8 <[email protected]> * fix(telegram): drop unused markdown param in findRichMarkdownTableLineIndexes Unblocks tsgo:prod (TS6133 'markdown' is declared but its value is never read), which main currently fails after b3f3154. The parameter was never read inside the function; the sole caller already derives lines/fenceSpans. Co-Authored-By: Claude Opus 4.8 <[email protected]> * fix(test): type heartbeat reply spy in pendingFinalDelivery clear tests check:test-types failed: the extracted heartbeatDeps helper typed replySpy as the generic ReturnType<typeof vi.fn>, which is not assignable to getReplyFromConfig. Use the exported HeartbeatReplySpy type instead. Co-Authored-By: Claude Opus 4.8 <[email protected]> * fix(heartbeat): clear pendingFinalDeliveryIntentId on delivery success Add the eighth pending-final field to CLEARED_PENDING_FINAL_DELIVERY_FIELDS so the heartbeat send-success and duplicate-skip paths match the canonical clearPendingFinalDeliveryAfterSuccess cleanup; a delivered heartbeat no longer leaves stale intent metadata behind. Seed and assert the field in both regression cases. Co-Authored-By: Claude Opus 4.8 <[email protected]> Claude-Session: https://claude.ai/code/session_01SkPieH67337ieQPWVXQgMD * fix(heartbeat): only clear pendingFinalDelivery this run owns 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 <[email protected]> Claude-Session: https://claude.ai/code/session_01SkPieH67337ieQPWVXQgMD --------- Co-authored-by: Claude Opus 4.8 <[email protected]>
1 parent e8d7b1f commit 305fa9c

2 files changed

Lines changed: 360 additions & 0 deletions

File tree

‎src/infra/heartbeat-runner.clears-pending-final-delivery.test.ts‎

Lines changed: 298 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,298 @@
1+
import fs from "node:fs/promises";
2+
import { describe, expect, it, vi } from "vitest";
3+
import type { OpenClawConfig } from "../config/config.js";
4+
import { runHeartbeatOnce, type HeartbeatDeps } from "./heartbeat-runner.js";
5+
import { installHeartbeatRunnerTestRuntime } from "./heartbeat-runner.test-harness.js";
6+
import {
7+
type HeartbeatReplySpy,
8+
seedMainSessionStore,
9+
withTempHeartbeatSandbox,
10+
} from "./heartbeat-runner.test-utils.js";
11+
12+
installHeartbeatRunnerTestRuntime();
13+
14+
type StoredEntry = Record<string, unknown> | undefined;
15+
16+
describe("runHeartbeatOnce clears stuck pendingFinalDelivery state once delivery is satisfied", () => {
17+
const TELEGRAM_GROUP = "-1001234567890";
18+
19+
function createHeartbeatConfig(storePath: string): OpenClawConfig {
20+
return {
21+
agents: {
22+
defaults: {
23+
heartbeat: { every: "5m", target: "telegram" },
24+
},
25+
},
26+
channels: {
27+
telegram: {
28+
token: "test-token",
29+
allowFrom: ["*"],
30+
heartbeat: { showOk: false },
31+
},
32+
},
33+
session: { store: storePath },
34+
} as unknown as OpenClawConfig;
35+
}
36+
37+
function heartbeatDeps(
38+
sendTelegram: ReturnType<typeof vi.fn>,
39+
replySpy: HeartbeatReplySpy,
40+
now?: number,
41+
): HeartbeatDeps {
42+
return {
43+
telegram: sendTelegram as unknown,
44+
getQueueSize: () => 0,
45+
// A fixed clock lets a test seed pendingFinalDeliveryCreatedAt relative to
46+
// the run's startedAt, which is what the ownership guard compares against.
47+
nowMs: () => now ?? Date.now(),
48+
getReplyFromConfig: replySpy,
49+
} satisfies HeartbeatDeps;
50+
}
51+
52+
// seedMainSessionStore exposes only part of the pendingFinalDelivery* family;
53+
// patch in lastHeartbeat* and the three unexposed pending fields so each test can
54+
// prove all eight recovery fields get cleared.
55+
async function patchEntry(
56+
storePath: string,
57+
sessionKey: string,
58+
patch: Record<string, unknown>,
59+
): Promise<void> {
60+
const store = JSON.parse(await fs.readFile(storePath, "utf-8")) as Record<string, StoredEntry>;
61+
store[sessionKey] = { ...store[sessionKey], ...patch };
62+
await fs.writeFile(storePath, JSON.stringify(store));
63+
}
64+
65+
async function readEntry(storePath: string, sessionKey: string): Promise<StoredEntry> {
66+
const store = JSON.parse(await fs.readFile(storePath, "utf-8")) as Record<string, StoredEntry>;
67+
return store[sessionKey];
68+
}
69+
70+
function expectPendingFinalDeliveryCleared(entry: StoredEntry): void {
71+
expect(entry?.pendingFinalDelivery).toBeUndefined();
72+
expect(entry?.pendingFinalDeliveryText).toBeUndefined();
73+
expect(entry?.pendingFinalDeliveryCreatedAt).toBeUndefined();
74+
expect(entry?.pendingFinalDeliveryLastAttemptAt).toBeUndefined();
75+
expect(entry?.pendingFinalDeliveryAttemptCount).toBeUndefined();
76+
expect(entry?.pendingFinalDeliveryLastError).toBeUndefined();
77+
expect(entry?.pendingFinalDeliveryContext).toBeUndefined();
78+
expect(entry?.pendingFinalDeliveryIntentId).toBeUndefined();
79+
}
80+
81+
it("nulls every pendingFinalDelivery* field after delivering substantive heartbeat content", async () => {
82+
await withTempHeartbeatSandbox(async ({ storePath, replySpy }) => {
83+
const cfg = createHeartbeatConfig(storePath);
84+
const NOW = Date.now();
85+
86+
// Seed a stuck pendingFinalDelivery this run owns: createdAt at run start
87+
// marks it as produced by this heartbeat (the case the original fix
88+
// targets). pendingFinalDeliveryText is a heartbeat-ack token so the
89+
// pendingFinalDelivery defer gate does not bail before the send.
90+
const sessionKey = await seedMainSessionStore(storePath, cfg, {
91+
lastChannel: "telegram",
92+
lastProvider: "telegram",
93+
lastTo: TELEGRAM_GROUP,
94+
updatedAt: NOW,
95+
pendingFinalDelivery: true,
96+
pendingFinalDeliveryText: "HEARTBEAT_OK",
97+
pendingFinalDeliveryCreatedAt: NOW,
98+
pendingFinalDeliveryAttemptCount: 3,
99+
pendingFinalDeliveryLastError: "prior-error",
100+
});
101+
await patchEntry(storePath, sessionKey, {
102+
pendingFinalDeliveryLastAttemptAt: NOW,
103+
pendingFinalDeliveryContext: { foo: "bar" },
104+
pendingFinalDeliveryIntentId: "intent-send-success",
105+
});
106+
107+
// Substantive reply text forces the post-success store write path
108+
// (heartbeat-runner.ts:~2120, `if (visibleSendSucceeded && !shouldSkipMain ...)`).
109+
const replyText = "Heartbeat update: everything is green.";
110+
replySpy.mockResolvedValue({ text: replyText });
111+
const sendTelegram = vi.fn().mockResolvedValue({ messageId: "m1", toJid: "jid" });
112+
113+
const result = await runHeartbeatOnce({
114+
cfg,
115+
deps: heartbeatDeps(sendTelegram, replySpy, NOW),
116+
});
117+
118+
expect(result.status).toBe("ran");
119+
expect(sendTelegram).toHaveBeenCalledTimes(1);
120+
121+
const entry = await readEntry(storePath, sessionKey);
122+
expect(entry?.lastHeartbeatText).toBe(replyText);
123+
expect(typeof entry?.lastHeartbeatSentAt).toBe("number");
124+
expectPendingFinalDeliveryCleared(entry);
125+
});
126+
});
127+
128+
it("clears pendingFinalDelivery* on a duplicate skip even when responsePrefix diverges the stored text", async () => {
129+
await withTempHeartbeatSandbox(async ({ storePath, replySpy }) => {
130+
// The send-success clear never runs here: the run reproduces a payload we
131+
// already delivered, so the duplicate-suppression branch
132+
// (heartbeat-runner.ts:~1944) returns before any send.
133+
//
134+
// agent-runner stores pendingFinalDeliveryText as the token-stripped body
135+
// WITHOUT the responsePrefix (agent-runner.ts:~2490), while
136+
// normalizeHeartbeatReply re-adds the prefix to the delivered text
137+
// (heartbeat-runner.ts:~838). So the stored pending text legitimately
138+
// differs from lastHeartbeatText for the same payload, and the clear must
139+
// not depend on a byte-equal text match or prefixed agents stay stuck.
140+
const cfg = {
141+
...createHeartbeatConfig(storePath),
142+
messages: { responsePrefix: "🤖" },
143+
} as unknown as OpenClawConfig;
144+
145+
const body = "Heartbeat update: everything is green.";
146+
const deliveredText = `🤖 ${body}`;
147+
const NOW = Date.now();
148+
// updatedAt is well past the 30s defer window so the pendingFinalDelivery
149+
// gate does not bail before the duplicate branch (the pending text is
150+
// substantive, not a heartbeat ack).
151+
const staleAt = NOW - 60_000;
152+
153+
const sessionKey = await seedMainSessionStore(storePath, cfg, {
154+
lastChannel: "telegram",
155+
lastProvider: "telegram",
156+
lastTo: TELEGRAM_GROUP,
157+
updatedAt: staleAt,
158+
pendingFinalDelivery: true,
159+
pendingFinalDeliveryText: body, // prefix-less; diverges from deliveredText
160+
// createdAt at run start: this run produced the pending (real runs stamp
161+
// a fresh createdAt during the agent turn the defer gate ran ahead of).
162+
pendingFinalDeliveryCreatedAt: NOW,
163+
pendingFinalDeliveryAttemptCount: 3,
164+
pendingFinalDeliveryLastError: "prior-error",
165+
});
166+
await patchEntry(storePath, sessionKey, {
167+
// lastHeartbeat* proves the same payload already went out within 24h,
168+
// which is what makes this run a duplicate and the pending clear safe.
169+
lastHeartbeatText: deliveredText,
170+
lastHeartbeatSentAt: staleAt,
171+
pendingFinalDeliveryLastAttemptAt: NOW,
172+
pendingFinalDeliveryContext: { foo: "bar" },
173+
pendingFinalDeliveryIntentId: "intent-duplicate-skip",
174+
});
175+
176+
// Reply is the prefix-less body; normalizeHeartbeatReply re-adds "🤖 ", so
177+
// normalized.text === deliveredText === lastHeartbeatText → duplicate skip.
178+
replySpy.mockResolvedValue({ text: body });
179+
const sendTelegram = vi.fn().mockResolvedValue({ messageId: "m1", toJid: "jid" });
180+
181+
const result = await runHeartbeatOnce({
182+
cfg,
183+
deps: heartbeatDeps(sendTelegram, replySpy, NOW),
184+
});
185+
186+
// status "ran" (not "skipped") proves the run reached the duplicate branch
187+
// rather than bailing at the pendingFinalDelivery gate.
188+
expect(result.status).toBe("ran");
189+
// Duplicate payload: nothing is sent this run.
190+
expect(sendTelegram).not.toHaveBeenCalled();
191+
192+
const entry = await readEntry(storePath, sessionKey);
193+
// lastHeartbeat* stays intact; only the satisfied pending state clears.
194+
expect(entry?.lastHeartbeatText).toBe(deliveredText);
195+
expectPendingFinalDeliveryCleared(entry);
196+
});
197+
});
198+
199+
it("preserves an older unsatisfied pendingFinalDelivery the heartbeat send did not create", async () => {
200+
await withTempHeartbeatSandbox(async ({ storePath, replySpy }) => {
201+
const cfg = createHeartbeatConfig(storePath);
202+
const NOW = Date.now();
203+
// An older user-facing final that failed delivery and still owns its own
204+
// get-reply redelivery recovery path. It predates this run (createdAt <
205+
// startedAt) and a message_tool_only/response-tool heartbeat would not
206+
// refresh it, so the send-success clear must NOT retire it.
207+
const olderCreatedAt = NOW - 60_000;
208+
const olderText = "Older final the user never received";
209+
const sessionKey = await seedMainSessionStore(storePath, cfg, {
210+
lastChannel: "telegram",
211+
lastProvider: "telegram",
212+
lastTo: TELEGRAM_GROUP,
213+
// Stale so the substantive-pending defer gate does not bail this run.
214+
updatedAt: NOW - 60_000,
215+
pendingFinalDelivery: true,
216+
pendingFinalDeliveryText: olderText,
217+
pendingFinalDeliveryCreatedAt: olderCreatedAt,
218+
pendingFinalDeliveryAttemptCount: 2,
219+
pendingFinalDeliveryLastError: "prior-delivery-failure",
220+
});
221+
await patchEntry(storePath, sessionKey, {
222+
pendingFinalDeliveryLastAttemptAt: NOW - 50_000,
223+
pendingFinalDeliveryContext: { channel: "telegram", to: "older-chat" },
224+
pendingFinalDeliveryIntentId: "intent-older-unsatisfied",
225+
});
226+
227+
// A fresh, different heartbeat payload that gets delivered this run.
228+
const replyText = "Fresh heartbeat content unrelated to the older final.";
229+
replySpy.mockResolvedValue({ text: replyText });
230+
const sendTelegram = vi.fn().mockResolvedValue({ messageId: "m1", toJid: "jid" });
231+
232+
const result = await runHeartbeatOnce({
233+
cfg,
234+
deps: heartbeatDeps(sendTelegram, replySpy, NOW),
235+
});
236+
237+
expect(result.status).toBe("ran");
238+
expect(sendTelegram).toHaveBeenCalledTimes(1);
239+
240+
const entry = await readEntry(storePath, sessionKey);
241+
// Send-success records the dedupe markers for the delivered payload...
242+
expect(entry?.lastHeartbeatText).toBe(replyText);
243+
// ...but the older, unowned pending-final survives for its own recovery.
244+
expect(entry?.pendingFinalDelivery).toBe(true);
245+
expect(entry?.pendingFinalDeliveryText).toBe(olderText);
246+
expect(entry?.pendingFinalDeliveryCreatedAt).toBe(olderCreatedAt);
247+
expect(entry?.pendingFinalDeliveryIntentId).toBe("intent-older-unsatisfied");
248+
});
249+
});
250+
251+
it("preserves an older unowned pendingFinalDelivery on a duplicate skip", async () => {
252+
await withTempHeartbeatSandbox(async ({ storePath, replySpy }) => {
253+
const cfg = createHeartbeatConfig(storePath);
254+
const NOW = Date.now();
255+
const body = "Recurring heartbeat status line.";
256+
const olderCreatedAt = NOW - 60_000;
257+
const olderText = "A different older final still awaiting delivery";
258+
const sessionKey = await seedMainSessionStore(storePath, cfg, {
259+
lastChannel: "telegram",
260+
lastProvider: "telegram",
261+
lastTo: TELEGRAM_GROUP,
262+
updatedAt: NOW - 60_000,
263+
pendingFinalDelivery: true,
264+
pendingFinalDeliveryText: olderText,
265+
pendingFinalDeliveryCreatedAt: olderCreatedAt,
266+
pendingFinalDeliveryAttemptCount: 2,
267+
pendingFinalDeliveryLastError: "prior-delivery-failure",
268+
});
269+
await patchEntry(storePath, sessionKey, {
270+
// Same payload already delivered within 24h -> this run is a duplicate skip.
271+
lastHeartbeatText: body,
272+
lastHeartbeatSentAt: NOW - 60_000,
273+
pendingFinalDeliveryLastAttemptAt: NOW - 50_000,
274+
pendingFinalDeliveryContext: { channel: "telegram", to: "older-chat" },
275+
pendingFinalDeliveryIntentId: "intent-older-dupe",
276+
});
277+
278+
replySpy.mockResolvedValue({ text: body });
279+
const sendTelegram = vi.fn().mockResolvedValue({ messageId: "m1", toJid: "jid" });
280+
281+
const result = await runHeartbeatOnce({
282+
cfg,
283+
deps: heartbeatDeps(sendTelegram, replySpy, NOW),
284+
});
285+
286+
expect(result.status).toBe("ran");
287+
// Duplicate payload: nothing is sent this run.
288+
expect(sendTelegram).not.toHaveBeenCalled();
289+
290+
const entry = await readEntry(storePath, sessionKey);
291+
// The duplicate-skip clear must not retire the older, unowned pending-final.
292+
expect(entry?.pendingFinalDelivery).toBe(true);
293+
expect(entry?.pendingFinalDeliveryText).toBe(olderText);
294+
expect(entry?.pendingFinalDeliveryCreatedAt).toBe(olderCreatedAt);
295+
expect(entry?.pendingFinalDeliveryIntentId).toBe("intent-older-dupe");
296+
});
297+
});
298+
});

‎src/infra/heartbeat-runner.ts‎

Lines changed: 62 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -1343,6 +1343,32 @@ function selectSystemEventsConsumedByHeartbeat(params: {
13431343
return preflight.pendingEventEntries;
13441344
}
13451345

1346+
// Recovery fields a completed heartbeat delivery must clear. Mirrors the
1347+
// canonical clearPendingFinalDeliveryAfterSuccess in dispatch-from-config.ts so
1348+
// the send-success and duplicate-skip paths drop the exact same set; leaving any
1349+
// behind keeps the session stuck on a delivery that already happened.
1350+
const CLEARED_PENDING_FINAL_DELIVERY_FIELDS = {
1351+
pendingFinalDelivery: undefined,
1352+
pendingFinalDeliveryText: undefined,
1353+
pendingFinalDeliveryCreatedAt: undefined,
1354+
pendingFinalDeliveryLastAttemptAt: undefined,
1355+
pendingFinalDeliveryAttemptCount: undefined,
1356+
pendingFinalDeliveryLastError: undefined,
1357+
pendingFinalDeliveryContext: undefined,
1358+
pendingFinalDeliveryIntentId: undefined,
1359+
} as const;
1360+
1361+
// Clear pending-final only when this run produced it: the agent run stamps
1362+
// createdAt during the run, so createdAt >= run start means we own it. An older
1363+
// final (e.g. one a message_tool_only run never refreshed) must keep its recovery path.
1364+
function heartbeatRunOwnsPendingFinalDelivery(
1365+
entry: SessionEntry | undefined,
1366+
runStartedAt: number,
1367+
): boolean {
1368+
const createdAt = entry?.pendingFinalDeliveryCreatedAt;
1369+
return typeof createdAt === "number" && createdAt >= runStartedAt;
1370+
}
1371+
13461372
export async function runHeartbeatOnce(opts: {
13471373
cfg?: OpenClawConfig;
13481374
agentId?: string;
@@ -1725,6 +1751,33 @@ export async function runHeartbeatOnce(opts: {
17251751
});
17261752
};
17271753

1754+
// The duplicate-suppression branch returns before any send, so it never hits
1755+
// the send-success clear. A duplicate means this run's own output was already
1756+
// delivered within the dedupe window, so this run's pending-final is satisfied
1757+
// and gets cleared the same way the send-success path does. We must not
1758+
// text-match the pending against the delivered text: agent-runner stores it
1759+
// pre-normalization (no responsePrefix), so a byte compare would leave
1760+
// prefixed agents permanently stuck. Ownership is gated on createdAt instead,
1761+
// so an older final this run did not produce is preserved, not erased.
1762+
const clearSatisfiedPendingFinalDelivery = async () => {
1763+
await updateSessionStore(
1764+
storePath,
1765+
(store) => {
1766+
const current = store[sessionKey];
1767+
if (current?.pendingFinalDelivery !== true && !current?.pendingFinalDeliveryText) {
1768+
return false;
1769+
}
1770+
if (!heartbeatRunOwnsPendingFinalDelivery(current, startedAt)) {
1771+
return false;
1772+
}
1773+
store[sessionKey] = { ...current, ...CLEARED_PENDING_FINAL_DELIVERY_FIELDS };
1774+
return true;
1775+
},
1776+
// No pending to clear is the common case; avoid rewriting the store then.
1777+
{ skipSaveWhenResult: (cleared) => !cleared },
1778+
);
1779+
};
1780+
17281781
const consumeInspectedSystemEvents = () => {
17291782
if (!preflight.shouldInspectPendingEvents || inspectedSystemEventsToConsume.length === 0) {
17301783
return;
@@ -2041,6 +2094,7 @@ export async function runHeartbeatOnce(opts: {
20412094
sessionKey,
20422095
updatedAt: previousUpdatedAt,
20432096
});
2097+
await clearSatisfiedPendingFinalDelivery();
20442098

20452099
emitHeartbeatEvent({
20462100
status: "skipped",
@@ -2173,10 +2227,18 @@ export async function runHeartbeatOnce(opts: {
21732227
if (!current) {
21742228
return;
21752229
}
2230+
// A heartbeat-driven agent run can leave its own pendingFinalDelivery
2231+
// set; a successful send completes it, so clear the recovery fields.
2232+
// Only clear the pending-final this run owns — an older final the run
2233+
// did not produce keeps its own recovery path.
2234+
const clearedRecoveryFields = heartbeatRunOwnsPendingFinalDelivery(current, startedAt)
2235+
? CLEARED_PENDING_FINAL_DELIVERY_FIELDS
2236+
: {};
21762237
store[sessionKey] = {
21772238
...current,
21782239
lastHeartbeatText: normalized.text,
21792240
lastHeartbeatSentAt: startedAt,
2241+
...clearedRecoveryFields,
21802242
};
21812243
});
21822244
}

0 commit comments

Comments
 (0)