From 5d55258694462c0bcfa0b5a67ef1bf5c38e82d9a Mon Sep 17 00:00:00 2001 From: Peter Steinberger Date: Wed, 29 Jul 2026 01:29:27 -0400 Subject: [PATCH] fix(heartbeat): deliver task-preempted same-session commitments (#115581) --- src/infra/heartbeat-runner-scheduler.ts | 11 +- .../heartbeat-runner.commitments.test.ts | 118 ++++++++++++++++++ 2 files changed, 120 insertions(+), 9 deletions(-) diff --git a/src/infra/heartbeat-runner-scheduler.ts b/src/infra/heartbeat-runner-scheduler.ts index 4819a7e34d6c..496c1ab1ca6e 100644 --- a/src/infra/heartbeat-runner-scheduler.ts +++ b/src/infra/heartbeat-runner-scheduler.ts @@ -19,7 +19,6 @@ import { type HeartbeatConfig, } from "./heartbeat-runner-config.js"; import { runHeartbeatOnce } from "./heartbeat-runner-run.js"; -import { resolveHeartbeatSession } from "./heartbeat-runner-session.js"; import { computeNextHeartbeatPhaseDueMs, resolveHeartbeatPhaseMs, @@ -382,11 +381,8 @@ export function startHeartbeatRunner(opts: { advanceAgentSchedule(agent, now, reason); let agentRan = res.status === "ran"; - const defaultSessionKey = resolveHeartbeatSession( - wakeConfig, - agent.agentId, - agent.heartbeat, - ).sessionKey; + // Re-read pending commitments after the global turn so a task-preempted + // default session gets an isolated follow-up without duplicating sends. const dueSessionKeys = canHeartbeatDeliverCommitments(agent.heartbeat) ? await listDueCommitmentSessionKeys({ cfg: wakeConfig, @@ -396,9 +392,6 @@ export function startHeartbeatRunner(opts: { }) : []; for (const dueSessionKey of dueSessionKeys) { - if (dueSessionKey === defaultSessionKey) { - continue; - } let commitmentRes: HeartbeatRunResult; try { commitmentRes = await runOnce({ diff --git a/src/infra/heartbeat-runner.commitments.test.ts b/src/infra/heartbeat-runner.commitments.test.ts index 37ce826128d7..87abeb1ef3d2 100644 --- a/src/infra/heartbeat-runner.commitments.test.ts +++ b/src/infra/heartbeat-runner.commitments.test.ts @@ -469,6 +469,124 @@ describe("runHeartbeatOnce commitments", () => { }); }); + it.each([ + { taskPreemptsCommitment: true, expectedRunCount: 2 }, + { taskPreemptsCommitment: false, expectedRunCount: 1 }, + ])( + "delivers a default-session commitment exactly once when task preemption is $taskPreemptsCommitment", + async ({ taskPreemptsCommitment, expectedRunCount }) => { + vi.useFakeTimers(); + vi.setSystemTime(nowMs); + + await withTempHeartbeatSandbox(async ({ tmpDir, storePath, replySpy }) => { + setTestEnvValue("OPENCLAW_STATE_DIR", tmpDir); + const sessionKey = "agent:main:telegram:user-155462274"; + const cfg: OpenClawConfig = { + agents: { + defaults: { + workspace: tmpDir, + heartbeat: { every: "5m", target: "last", session: sessionKey }, + }, + }, + channels: { telegram: { allowFrom: ["*"] } }, + session: { store: storePath }, + }; + await seedSessionStore(storePath, sessionKey, { + lastChannel: "telegram", + lastProvider: "telegram", + lastTo: "stale-target", + }); + await saveCommitmentStore(undefined, { + version: 1, + commitments: [buildCommitment({ id: "cm_interview", sessionKey, to: "155462274" })], + }); + + const sendTelegram = vi.fn().mockResolvedValue({ + messageId: "m1", + chatId: "155462274", + }); + replySpy.mockImplementation(async (ctx, opts) => { + if (taskPreemptsCommitment && replySpy.mock.calls.length === 1) { + expect(ctx.Body).toContain("Run the following periodic tasks"); + expect(ctx.Body).toContain("Check deployment status with the normal tools"); + expect(ctx.Body).not.toContain("Due inferred follow-up commitments"); + expect(opts?.disableTools).toBeUndefined(); + return { text: HEARTBEAT_TOKEN }; + } + + expect(ctx.Body).toContain("Due inferred follow-up commitments"); + expect(ctx.Body).toContain("How did the interview go?"); + expect(ctx.Body).not.toContain("Check deployment status with the normal tools"); + expect(ctx.OriginatingTo).toBe("155462274"); + expect(opts?.disableTools).toBe(true); + expect(opts?.skillFilter).toStrictEqual([]); + return { text: "How did the interview go?" }; + }); + + const runOnce = vi.fn(async (opts) => + runHeartbeatOnce({ + ...opts, + ...(taskPreemptsCommitment && opts.runScope === "global" + ? { + tasks: [ + { + jobId: "job-deployment-status", + name: "deployment-status", + prompt: "Check deployment status with the normal tools", + }, + ], + } + : {}), + deps: { + ...opts.deps, + getReplyFromConfig: replySpy, + telegram: sendTelegram, + getQueueSize: () => 0, + nowMs: () => nowMs, + }, + }), + ); + const runner = startHeartbeatRunner({ + cfg, + runOnce, + stableSchedulerSeed: "same-session-commitment-monitor-tick", + }); + + requestHeartbeat({ + source: "interval", + intent: "scheduled", + reason: "interval", + agentId: "main", + scheduledEveryMs: 5 * 60_000, + coalesceMs: 0, + }); + await vi.advanceTimersByTimeAsync(1); + await vi.waitFor(() => expect(runOnce).toHaveBeenCalledTimes(expectedRunCount)); + runner.stop(); + + expect(replySpy).toHaveBeenCalledTimes(expectedRunCount); + if (taskPreemptsCommitment) { + expect(runOnce.mock.calls[1]?.[0]).toMatchObject({ + agentId: "main", + runScope: "commitment-only", + sessionKey, + }); + } + expect(sendTelegram).toHaveBeenCalledTimes(1); + expect(sendTelegram).toHaveBeenCalledWith( + "155462274", + "How did the interview go?", + expect.any(Object), + ); + expectCommitmentFields((await loadCommitmentStore()).commitments[0], { + id: "cm_interview", + status: "sent", + attempts: 1, + }); + }); + }, + ); + it("delivers due commitments on a targeted cron-monitor interval tick", async () => { vi.useFakeTimers(); vi.setSystemTime(nowMs);