From b23c7bd0f841ed4b45a0d8ea327daf1cf653aecd Mon Sep 17 00:00:00 2001 From: Peter Steinberger Date: Thu, 23 Jul 2026 16:13:08 -0700 Subject: [PATCH] feat(cron): convert heartbeat tasks: entries into independent cron jobs (#113165) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Each legacy heartbeat tasks: entry becomes an editable system-created cron job (declaration key heartbeat-task::, per-occurrence identity for duplicate names) that fires a guarded heartbeat wake carrying the task prompt plus current monitor scratch context. Task cadence is now independent of the base heartbeat interval; active-hours, busy-retry, min-spacing, and flood guards are preserved, deferred payload-carrying wakes are retained and retried after the spacing floor, and colliding task/event wakes cannot starve each other. openclaw doctor --fix migrates existing blocks: async parse/plan, then one synchronous SQLite transaction that rereads the pinned scratch revision, upserts job rows, and strips the tasks: block atomically — concurrent doctors serialize on the database and a losing run aborts untouched. Orphan fields, invalid intervals, and incomplete entries block migration of their block with a clear finding instead of silent text loss. The runtime tasks: parser is deleted; leftover text is ordinary scratch prose. Nine Codex gpt-5.6-sol xhigh autoreview cycles; final verdict clean. --- .../.generated/plugin-sdk-api-baseline.sha256 | 2 +- docs/automation/cron-jobs.md | 8 + docs/automation/index.md | 4 +- docs/docs_map.md | 3 +- docs/gateway/heartbeat.md | 48 +- docs/gateway/troubleshooting.md | 5 +- docs/help/faq-first-run.md | 6 +- docs/help/troubleshooting.md | 3 +- docs/reference/templates/HEARTBEAT.md | 6 +- docs/start/openclaw.md | 2 +- docs/tools/thinking.md | 2 +- src/agents/heartbeat-system-prompt.test.ts | 14 +- src/agents/heartbeat-system-prompt.ts | 6 +- src/agents/system-prompt.ts | 2 +- src/agents/tools/heartbeat-response-tool.ts | 9 +- src/auto-reply/heartbeat.test.ts | 32 +- src/auto-reply/heartbeat.ts | 123 +-- .../doctor-heartbeat-task-migration.test.ts | 720 ++++++++++++++++++ .../doctor-heartbeat-task-migration.ts | 611 +++++++++++++++ src/commands/doctor.fast-path-mocks.ts | 5 + src/commands/heartbeat-task-legacy.ts | 195 +++++ src/config/sessions/types.ts | 2 +- src/config/types.agent-defaults.ts | 2 +- src/cron/heartbeat-task.ts | 34 + src/cron/isolated-agent/session.ts | 1 - src/cron/service.heartbeat-payload.test.ts | 40 + src/cron/service/timer-execution.ts | 20 + src/cron/store/row-codec.ts | 21 + src/flows/doctor-health-contributions.test.ts | 43 ++ src/flows/doctor-health-contributions.ts | 25 + src/gateway/server-cron.ts | 1 + src/infra/heartbeat-cooldown.test.ts | 69 +- src/infra/heartbeat-cooldown.ts | 58 +- ...at-runner.ack-token-heartbeat-acks.test.ts | 12 +- .../heartbeat-runner.commitments.test.ts | 43 +- ...beat-runner.isolated-key-stability.test.ts | 47 +- ...eat-runner.isolated-session-mirror.test.ts | 11 +- ...tbeat-runner.returns-default-unset.test.ts | 47 +- src/infra/heartbeat-runner.scheduler.test.ts | 101 ++- .../heartbeat-runner.tool-response.test.ts | 23 +- src/infra/heartbeat-runner.ts | 233 ++---- src/infra/heartbeat-wake.test.ts | 359 +++++++++ src/infra/heartbeat-wake.ts | 290 ++++++- .../codex-dynamic-tools.heartbeat-turn.json | 3 +- .../telegram-heartbeat-codex-tool.md | 21 +- 45 files changed, 2786 insertions(+), 526 deletions(-) create mode 100644 src/commands/doctor-heartbeat-task-migration.test.ts create mode 100644 src/commands/doctor-heartbeat-task-migration.ts create mode 100644 src/commands/heartbeat-task-legacy.ts create mode 100644 src/cron/heartbeat-task.ts diff --git a/docs/.generated/plugin-sdk-api-baseline.sha256 b/docs/.generated/plugin-sdk-api-baseline.sha256 index d620b8ff1c1e..ad58ab246956 100644 --- a/docs/.generated/plugin-sdk-api-baseline.sha256 +++ b/docs/.generated/plugin-sdk-api-baseline.sha256 @@ -103,7 +103,7 @@ aa2a56b4448c8ebdec9d06aac95d809995f533093d42fa32cd75e1d852967245 module/questio bd2355e94248d21c252148085e5afa5db9a2f3f48f665a8b3e0fcf77326d9b6c module/reply-dispatch-runtime ac2b199e95c5c8b1e2a65e62bd41d1b6322e531bca294ef4979a297a12640bce module/reply-history f394fe4d5a7ed9e4d574063ae44e8d6af85c9a0e7d8b329f750ca16b0664325f module/reply-payload -554bbc681b17bf102a5e240e3f563c7d3c3f1dbab31f8fa3c9285f106c3e8111 module/reply-runtime +6b8a50764563f096175162c49d1acf5593eee3363edd45976c1aa641a3d532cf module/reply-runtime d78db621b8f4f0cc679cad2d5d21b6c95b5418c58611dd347cdf09049ed124f6 module/routing ff6cca86f54f94f238205f5b122af36666314e0a380f3ec7f0ccb9ed9208df31 module/run-command 53b0295cec105696a1664c5c7f5576a7b55d197eb95dcd9185486f010bd53750 module/runtime diff --git a/docs/automation/cron-jobs.md b/docs/automation/cron-jobs.md index 9929d66b001e..4e8ffff63ecf 100644 --- a/docs/automation/cron-jobs.md +++ b/docs/automation/cron-jobs.md @@ -77,6 +77,14 @@ Timestamps without a timezone are treated as UTC. Add `--tz America/New_York` to Recurring top-of-hour expressions (minute `0` with a wildcard hour field) are automatically staggered by up to 5 minutes to reduce load spikes. Use `--exact` to force precise timing, or `--stagger 30s` for an explicit window (cron schedules only). +### Heartbeat task migration + +Older heartbeat scratch supported a structured `tasks:` block. Run `openclaw doctor --fix` after upgrading to convert each entry into an ordinary editable main-session cron job. Doctor preserves the interval and previous last-run timing, creates the jobs before removing the block, and safely converges the same declaration keys on rerun. + +These migrated jobs carry public `systemEvent` payloads, so `openclaw cron list`, `get`, `edit`, and `remove` plus the cron tool manage them like other jobs. Their execution uses the guarded heartbeat task wake: active hours, minimum spacing, flood control, and busy retries still apply, while cron owns each task's independent cadence. Jobs due in the same coalescing window can share one heartbeat turn. A scheduled occurrence outside heartbeat active hours is skipped and retried at the job's next occurrence. + +Heartbeat scratch is now monitor prose only. Runtime heartbeats do not parse `tasks:` text as schedules; create new recurring work with cron. + ### Stream sources A stream schedule keeps an operator-authored argv command running under the Gateway and fires the job from its stdout and stderr lines. Stream schedules are event-driven, never time-due, and require `cron.triggers.enabled: true` because the long-lived command has the same unattended trust class as trigger scripts. Disabling or removing the job stops the process; Gateway shutdown waits for process-tree teardown. Fast failures restart with cron's built-in error backoff. Five consecutive runs shorter than 60 seconds leave the job in an error state and use the normal failure-alert path; manually re-enable the job to clear the restart cap. diff --git a/docs/automation/index.md b/docs/automation/index.md index a5367e2725d5..f36a6bf5b734 100644 --- a/docs/automation/index.md +++ b/docs/automation/index.md @@ -95,14 +95,14 @@ See [Hooks](/automation/hooks). ### Heartbeat -Heartbeat is a periodic main-session turn (default every 30 minutes). It batches multiple checks (inbox, calendar, notifications) in one agent turn with full session context. Heartbeat turns do not create task records and do not extend daily/idle session reset freshness. Use `HEARTBEAT.md` for a small checklist, or a `tasks:` block when you want due-only periodic checks inside heartbeat itself. Empty heartbeat files skip as `empty-heartbeat-file`; due-only task mode skips as `no-tasks-due`. Heartbeats defer while cron work is active or queued, and `heartbeat.skipWhenBusy` can also defer an agent while that same agent's session-keyed subagent or nested lanes are busy. +Heartbeat is a periodic main-session turn (default every 30 minutes). It batches checklist-style monitoring (inbox, calendar, notifications) in one agent turn with full session context. Heartbeat turns do not create task records and do not extend daily/idle session reset freshness. Heartbeat scratch is small prompt context; schedule recurring work as cron jobs. Empty heartbeat scratch skips as `empty-heartbeat-file`. Heartbeats defer while cron work is active or queued, and `heartbeat.skipWhenBusy` can also defer an agent while that same agent's session-keyed subagent or nested lanes are busy. See [Heartbeat](/gateway/heartbeat). ## How they work together - **Cron** handles precise schedules (daily reports, weekly reviews) and one-shot reminders. All cron executions create task records. -- **Heartbeat** handles routine monitoring (inbox, calendar, notifications) in one batched turn every 30 minutes. +- **Heartbeat** handles one batched monitoring checklist every 30 minutes; cron owns checks that need independent cadences. - **Hooks** react to specific events (session resets, compaction, message flow) with custom scripts. Plugin hooks cover tool calls. - **Standing orders** give the agent persistent context and authority boundaries. - **Task Flow** coordinates multi-step flows above individual tasks. diff --git a/docs/docs_map.md b/docs/docs_map.md index 59388cc3089e..5f3a6b8b4b5c 100644 --- a/docs/docs_map.md +++ b/docs/docs_map.md @@ -66,6 +66,7 @@ Do not edit it by hand; run `pnpm docs:map:gen`. - H2: Quick start - H2: How cron works - H2: Schedule types + - H3: Heartbeat task migration - H3: Stream sources - H3: Dynamic cadence (pacing) - H3: Day-of-month and day-of-week use OR logic @@ -3602,7 +3603,7 @@ Do not edit it by hand; run `pnpm docs:map:gen`. - H3: Per-channel vs per-account examples - H3: Common patterns - H2: Monitor scratch (optional) - - H3: tasks: blocks + - H3: Schedule recurring checks with cron - H3: Can the agent update its scratch? - H2: Manual wake (on-demand) - H2: Cost awareness diff --git a/docs/gateway/heartbeat.md b/docs/gateway/heartbeat.md index 66a5e8df47c1..c2d0202e28ea 100644 --- a/docs/gateway/heartbeat.md +++ b/docs/gateway/heartbeat.md @@ -28,7 +28,7 @@ Troubleshooting: [Scheduled Tasks](/automation/cron-jobs#troubleshooting) Leave heartbeats enabled (default is `30m`, or `1h` when Anthropic OAuth/token auth is configured, including Claude CLI reuse) or set your own cadence. - Store a tiny checklist or `tasks:` block in the heartbeat monitor's scratch with `openclaw cron scratch --set "..."`. + Store a tiny checklist in the heartbeat monitor's scratch with `openclaw cron scratch --set "..."`. `target: "none"` is the default; set `target: "last"` to route to the last contact. @@ -63,7 +63,7 @@ Example config: ## Defaults - Interval: `30m`. Applying Anthropic provider defaults bumps this to `1h` when the resolved auth mode is OAuth/token (including Claude CLI reuse), but only while `heartbeat.every` is unset. Set `agents.defaults.heartbeat.every` or per-agent `agents.entries.*.heartbeat.every`; use `0m` to disable. -- Prompt body (configurable via `agents.defaults.heartbeat.prompt`): `Follow the heartbeat monitor scratch context when provided. Do not infer or repeat old tasks from prior chats. If nothing needs attention, reply HEARTBEAT_OK.` +- Prompt body (configurable via `agents.defaults.heartbeat.prompt`): `Follow the heartbeat monitor scratch context when provided. Recurring tasks are cron jobs; create or change their schedules with cron tools or the openclaw cron CLI, not heartbeat scratch. Do not infer or repeat old tasks from prior chats. If nothing needs attention, reply HEARTBEAT_OK.` - Timeout: unset heartbeat turns use `agents.defaults.timeoutSeconds` when set. Otherwise, they use the heartbeat cadence capped at 600 seconds. Set `agents.defaults.heartbeat.timeoutSeconds` or per-agent `agents.entries.*.heartbeat.timeoutSeconds` for longer heartbeat work. - The heartbeat prompt is sent **verbatim** as the user message. The system prompt includes a "Heartbeats" section when heartbeats are enabled for the default agent, and the run is flagged internally. - When heartbeats are disabled with `0m`, the monitor cron job stays but is disabled, and its scratch is retained for when you re-enable the cadence. @@ -107,7 +107,7 @@ Outside heartbeats, stray `HEARTBEAT_OK` at the start/end of a message is stripp target: "last", // default: none | options: last | none | (core or plugin, e.g. "imessage") to: "+15551234567", // optional channel-specific override accountId: "ops-bot", // optional multi-account channel id - prompt: "Follow the heartbeat monitor scratch context when provided. Do not infer or repeat old tasks from prior chats. If nothing needs attention, reply HEARTBEAT_OK.", + prompt: "Follow the heartbeat monitor scratch context when provided. Recurring tasks are cron jobs; create or change their schedules with cron tools or the openclaw cron CLI, not heartbeat scratch. Do not infer or repeat old tasks from prior chats. If nothing needs attention, reply HEARTBEAT_OK.", }, }, }, @@ -146,7 +146,7 @@ Example: two agents, only the second agent runs heartbeats. target: "whatsapp", to: "+15551234567", timeoutSeconds: 45, - prompt: "Follow the heartbeat monitor scratch context when provided. Do not infer or repeat old tasks from prior chats. If nothing needs attention, reply HEARTBEAT_OK.", + prompt: "Follow the heartbeat monitor scratch context when provided. Recurring tasks are cron jobs; create or change their schedules with cron tools or the openclaw cron CLI, not heartbeat scratch. Do not infer or repeat old tasks from prior chats. If nothing needs attention, reply HEARTBEAT_OK.", }, }, ], @@ -382,7 +382,7 @@ Writes are compare-and-swap guarded: pass `--expected-revision ` to fail inst The agent can also update its own scratch: during a heartbeat turn, `heartbeat_respond` accepts an optional `scratch` string that fully replaces the monitor's scratch for future heartbeats. -**Migrating from HEARTBEAT.md or config-only cadence?** Run `openclaw doctor --fix`. Doctor first creates or updates the system-owned monitor rows from `agents.*.heartbeat`, then imports each agent's workspace `HEARTBEAT.md` into the monitor's scratch, archives the original under the state directory (`backups/heartbeat-migration/`), and removes the file. For one stable upgrade window, an unmigrated legacy file remains a read-only fallback when no scratch revision exists, with a Gateway warning directing you to Doctor; new workspaces and completed migrations use database scratch only. +**Migrating from HEARTBEAT.md or config-only cadence?** Run `openclaw doctor --fix`. Doctor first creates or updates the system-owned monitor rows from `agents.*.heartbeat`, then imports each agent's workspace `HEARTBEAT.md` into the monitor's scratch, converts any valid legacy `tasks:` entries into cron jobs, archives the original under the state directory (`backups/heartbeat-migration/`), and removes the file. For one stable upgrade window, an unmigrated legacy file remains a read-only fallback when no scratch revision exists, with a Gateway warning directing you to Doctor; new workspaces and completed migrations use database scratch only. If scratch exists but is effectively empty (only blank lines, Markdown/HTML comments, Markdown headings like `# Heading`, fence markers, or empty checklist stubs), OpenClaw skips the heartbeat run to save API calls. That skip is reported as `reason=empty-heartbeat-file`. If no scratch exists, the heartbeat still runs and the model decides what to do. @@ -399,45 +399,17 @@ Example scratch: - If a task is blocked, write down _what is missing_ and ask Peter next time. ``` -### `tasks:` blocks +### Schedule recurring checks with cron -Scratch also supports a small structured `tasks:` block for interval-based checks inside heartbeat itself. +Heartbeat scratch is prompt context, not a scheduler. Create each recurring check as a [cron job](/automation/cron-jobs) so it has its own cadence, enable/disable state, and run history. Cron jobs can still target the main session when the check should use the normal conversation context. -Example: +Older scratch may contain a structured `tasks:` block. Run `openclaw doctor --fix` once after upgrading: Doctor converts every valid entry into an independently scheduled cron job, preserves its interval and previous last-run timing, and removes the retired block while keeping surrounding scratch prose. Runtime heartbeat turns do not parse `tasks:` text as schedules. -```md -tasks: - -- name: inbox-triage - interval: 30m - prompt: "Check for urgent unread emails and flag anything time sensitive." -- name: calendar-scan - interval: 2h - prompt: "Check for upcoming meetings that need prep or follow-up." - -# Additional instructions - -- Keep alerts short. -- If nothing needs attention after all due tasks, reply HEARTBEAT_OK. -``` - - - - - OpenClaw parses the `tasks:` block and checks each task against its own `interval`. - - Only **due** tasks are included in the heartbeat prompt for that tick. - - If no tasks are due, the heartbeat is skipped entirely (`reason=no-tasks-due`) to avoid a wasted model call. - - Non-task scratch content is preserved and appended as additional context after the due-task list. - - Task last-run timestamps are stored in session state (`heartbeatTaskState`), so intervals survive normal restarts. - - Task timestamps are only advanced after a heartbeat run completes its normal reply path. Skipped `empty-heartbeat-file` / `no-tasks-due` runs do not mark tasks as completed. - - - - -Task mode is useful when you want one scratch document to hold several periodic checks without paying for all of them every tick. +Doctor-created heartbeat task jobs keep heartbeat active-hours, cooldown, flood, and busy guards. Jobs due together can coalesce into one heartbeat turn. An occurrence outside active hours is skipped and tried again at its next cron occurrence. ### Can the agent update its scratch? -Yes. During a heartbeat turn, the agent can pass a `scratch` value to `heartbeat_respond` to fully replace the monitor scratch for future heartbeats. You can also ask it in a normal chat to run `openclaw cron scratch --set ...`, or edit the scratch yourself with the same command. +Yes. During a heartbeat turn, the agent can pass a `scratch` value to `heartbeat_respond` to fully replace the monitor prose for future heartbeats. You can also ask it in a normal chat to run `openclaw cron scratch --set ...`, or edit the scratch yourself with the same command. Manage recurring schedules with cron instead of writing scheduler syntax into scratch. Don't put secrets (API keys, phone numbers, private tokens) into monitor scratch - it becomes part of the prompt context. diff --git a/docs/gateway/troubleshooting.md b/docs/gateway/troubleshooting.md index 74c5b0aefad8..298a1e668e32 100644 --- a/docs/gateway/troubleshooting.md +++ b/docs/gateway/troubleshooting.md @@ -756,15 +756,14 @@ Look for: - Cron enabled and next wake present. - Job run history status (`ok`, `skipped`, `error`). -- Heartbeat skip reasons (`quiet-hours`, `requests-in-flight`, `cron-in-progress`, `lanes-busy`, `alerts-disabled`, `empty-heartbeat-file`, `no-tasks-due`). +- Heartbeat skip reasons (`quiet-hours`, `requests-in-flight`, `cron-in-progress`, `lanes-busy`, `alerts-disabled`, `empty-heartbeat-file`). - `cron: scheduler disabled; jobs will not run automatically` → cron disabled. - `cron: timer tick failed` → scheduler tick failed; check file/log/runtime errors. - `heartbeat skipped` with `reason=quiet-hours` → outside active hours window. - - `heartbeat skipped` with `reason=empty-heartbeat-file` → `HEARTBEAT.md` exists but only contains blank, comment, header, fence, or empty-checklist scaffolding, so OpenClaw skips the model call. - - `heartbeat skipped` with `reason=no-tasks-due` → `HEARTBEAT.md` contains a `tasks:` block, but none of the tasks are due on this tick. + - `heartbeat skipped` with `reason=empty-heartbeat-file` → heartbeat monitor scratch only contains blank, comment, header, fence, or empty-checklist scaffolding, so OpenClaw skips the model call. - `heartbeat: unknown accountId` → invalid account id for heartbeat delivery target. - `heartbeat skipped` with `reason=dm-blocked` → heartbeat target resolved to a DM-style destination while `agents.defaults.heartbeat.directPolicy` (or per-agent override) is set to `block`. diff --git a/docs/help/faq-first-run.md b/docs/help/faq-first-run.md index 7727ddad4ef0..a898c06df6ac 100644 --- a/docs/help/faq-first-run.md +++ b/docs/help/faq-first-run.md @@ -57,12 +57,10 @@ and troubleshooting see the main [FAQ](/help/faq). | Skip reason | Meaning | | --- | --- | | `quiet-hours` | Outside the configured active-hours window | - | `empty-heartbeat-file` | `HEARTBEAT.md` exists but only has blank, comment, header, fence, or empty-checklist scaffolding | - | `no-tasks-due` | Task mode is active but no task interval is due yet | + | `empty-heartbeat-file` | Heartbeat monitor scratch exists but only has blank, comment, header, fence, or empty-checklist scaffolding | | `alerts-disabled` | All heartbeat visibility is off (`showOk`, `showAlerts`, and `useIndicator` all disabled) | - In task mode, due timestamps advance only after a real heartbeat run completes. - Skipped runs do not mark tasks as completed. + Older heartbeat `tasks:` blocks migrate to independently scheduled cron jobs with `openclaw doctor --fix`. Docs: [Heartbeat](/gateway/heartbeat), [Automation](/automation). diff --git a/docs/help/troubleshooting.md b/docs/help/troubleshooting.md index 8eb36e6b9f89..6e611f7240a4 100644 --- a/docs/help/troubleshooting.md +++ b/docs/help/troubleshooting.md @@ -330,8 +330,7 @@ flowchart TD - `cron: scheduler disabled; jobs will not run automatically` → cron is disabled. - `heartbeat skipped` reason `quiet-hours` → outside configured active hours. - - `heartbeat skipped` reason `empty-heartbeat-file` → `HEARTBEAT.md` exists but contains only blank, comment, header, fence, or empty-checklist scaffolding. - - `heartbeat skipped` reason `no-tasks-due` → task mode is active but no task interval is due yet. + - `heartbeat skipped` reason `empty-heartbeat-file` → heartbeat monitor scratch contains only blank, comment, header, fence, or empty-checklist scaffolding. - `heartbeat skipped` reason `alerts-disabled` → `showOk`, `showAlerts`, and `useIndicator` are all off. - `requests-in-flight` → main lane busy; heartbeat wake deferred. - `unknown accountId` → heartbeat delivery target account does not exist. diff --git a/docs/reference/templates/HEARTBEAT.md b/docs/reference/templates/HEARTBEAT.md index ee28c2fd542a..0a39ecedca0e 100644 --- a/docs/reference/templates/HEARTBEAT.md +++ b/docs/reference/templates/HEARTBEAT.md @@ -16,12 +16,12 @@ Shipped default content: # Keep this file empty (or with only comments) to skip heartbeat API calls. -# Add tasks below when you want the agent to check something periodically. +# Add a short checklist below when the heartbeat should inspect shared context. ``` -Add short tasks below the comment lines only when you want periodic checks. Keep it small: heartbeat runs read this file every tick (default every 30 minutes), so bloated instructions burn tokens on every wake. +Add a short checklist below the comment lines only when one heartbeat turn should inspect the items together. Keep it small: heartbeat runs read this file every tick (default every 30 minutes), so bloated instructions burn tokens on every wake. -For due-only checks instead of a plain checklist, use a structured `tasks:` block with per-task `interval` and `prompt` fields; see [HEARTBEAT.md](/gateway/heartbeat#heartbeatmd-optional) for the format and behavior. +For independently scheduled or due-only checks, create [cron jobs](/automation/cron-jobs). Heartbeat scratch no longer supports scheduler syntax. Run `openclaw doctor --fix` to convert older `tasks:` blocks. ## Related diff --git a/docs/start/openclaw.md b/docs/start/openclaw.md index 35b129ce2f30..4c6f6d58b105 100644 --- a/docs/start/openclaw.md +++ b/docs/start/openclaw.md @@ -168,7 +168,7 @@ Example: ## Heartbeats (proactive mode) By default, OpenClaw runs a heartbeat every 30 minutes with the prompt: -`Follow the heartbeat monitor scratch context when provided. Do not infer or repeat old tasks from prior chats. If nothing needs attention, reply HEARTBEAT_OK.` +`Follow the heartbeat monitor scratch context when provided. Recurring tasks are cron jobs; create or change their schedules with cron tools or the openclaw cron CLI, not heartbeat scratch. Do not infer or repeat old tasks from prior chats. If nothing needs attention, reply HEARTBEAT_OK.` Set `agents.defaults.heartbeat.every: "0m"` to disable. Heartbeat checklists live in the monitor's cron scratch (see [Heartbeat](/gateway/heartbeat)); `openclaw doctor --fix` migrates a legacy workspace `HEARTBEAT.md` into it. - If the monitor scratch exists but is effectively empty (only blank lines, Markdown/HTML comments, Markdown headings like `# Heading`, fence markers, or empty checklist stubs), OpenClaw skips the heartbeat run to save API calls. diff --git a/docs/tools/thinking.md b/docs/tools/thinking.md index f32e02c81bb3..ae7e4017e4ec 100644 --- a/docs/tools/thinking.md +++ b/docs/tools/thinking.md @@ -122,7 +122,7 @@ Malformed local-model reasoning tags are handled conservatively. Closed ` ## Heartbeats -- Heartbeat probe body is the configured heartbeat prompt (default: `Follow the heartbeat monitor scratch context when provided. Do not infer or repeat old tasks from prior chats. If nothing needs attention, reply HEARTBEAT_OK.`). Inline directives in a heartbeat message apply as usual (but avoid changing session defaults from heartbeats). +- Heartbeat probe body is the configured heartbeat prompt (default: `Follow the heartbeat monitor scratch context when provided. Recurring tasks are cron jobs; create or change their schedules with cron tools or the openclaw cron CLI, not heartbeat scratch. Do not infer or repeat old tasks from prior chats. If nothing needs attention, reply HEARTBEAT_OK.`). Inline directives in a heartbeat message apply as usual (but avoid changing session defaults from heartbeats). - Heartbeat delivery defaults to the final payload only. To also send the separate `Thinking` message (when available), set `agents.defaults.heartbeat.includeReasoning: true` or per-agent `agents.entries.*.heartbeat.includeReasoning: true`. ## Web chat UI diff --git a/src/agents/heartbeat-system-prompt.test.ts b/src/agents/heartbeat-system-prompt.test.ts index ad6025012e0e..cb935fc0c763 100644 --- a/src/agents/heartbeat-system-prompt.test.ts +++ b/src/agents/heartbeat-system-prompt.test.ts @@ -110,7 +110,19 @@ describe("resolveHeartbeatPromptForSystemPrompt", () => { agentId: "main", defaultAgentId: "main", }), - ).toBe("Ops check"); + ).toContain("Ops check"); + expect( + resolveHeartbeatPromptForSystemPrompt({ + config: { + agents: { + defaults: { heartbeat: { every: "30m" } }, + list: [{ id: "main", heartbeat: { prompt: "Ops check" } }], + }, + }, + agentId: "main", + defaultAgentId: "main", + }), + ).toContain("Recurring tasks are cron jobs"); }); it("does not inject the heartbeat section for non-default agents", () => { diff --git a/src/agents/heartbeat-system-prompt.ts b/src/agents/heartbeat-system-prompt.ts index 01b6b4cd91fc..01b7d8b42312 100644 --- a/src/agents/heartbeat-system-prompt.ts +++ b/src/agents/heartbeat-system-prompt.ts @@ -4,6 +4,7 @@ import { normalizeOptionalString } from "@openclaw/normalization-core/string-coerce"; import { DEFAULT_HEARTBEAT_EVERY, + HEARTBEAT_CRON_TASK_GUIDANCE, resolveHeartbeatPrompt as resolveHeartbeatPromptText, } from "../auto-reply/heartbeat.js"; import { parseDurationMs } from "../cli/parse-duration.js"; @@ -88,5 +89,8 @@ export function resolveHeartbeatPromptForSystemPrompt(params: { if (!shouldIncludeHeartbeatGuidanceForSystemPrompt(params)) { return undefined; } - return resolveHeartbeatPromptText(heartbeat?.prompt); + const prompt = resolveHeartbeatPromptText(heartbeat?.prompt); + return prompt.includes(HEARTBEAT_CRON_TASK_GUIDANCE) + ? prompt + : `${prompt} ${HEARTBEAT_CRON_TASK_GUIDANCE}`; } diff --git a/src/agents/system-prompt.ts b/src/agents/system-prompt.ts index 18d371a97b1d..3adcd168973e 100644 --- a/src/agents/system-prompt.ts +++ b/src/agents/system-prompt.ts @@ -83,7 +83,7 @@ const CONTEXT_FILE_ORDER = new Map([ const DYNAMIC_CONTEXT_FILE_BASENAMES = new Set(); const DEFAULT_HEARTBEAT_PROMPT_CONTEXT_BLOCK = - "Default heartbeat prompt:\n`Follow the heartbeat monitor scratch context when provided. Do not infer or repeat old tasks from prior chats. If nothing needs attention, reply HEARTBEAT_OK.`"; + "Default heartbeat prompt:\n`Follow the heartbeat monitor scratch context when provided. Recurring tasks are cron jobs; create or change their schedules with cron tools or the openclaw cron CLI, not heartbeat scratch. Do not infer or repeat old tasks from prior chats. If nothing needs attention, reply HEARTBEAT_OK.`"; const SYSTEM_PROMPT_STABLE_PREFIX_CACHE_LIMIT = 64; type StablePromptPrefixCacheEntry = { diff --git a/src/agents/tools/heartbeat-response-tool.ts b/src/agents/tools/heartbeat-response-tool.ts index 5c418c553dce..1e36d229dda3 100644 --- a/src/agents/tools/heartbeat-response-tool.ts +++ b/src/agents/tools/heartbeat-response-tool.ts @@ -27,7 +27,12 @@ const HeartbeatResponseToolSchema = Type.Object( reason: Type.Optional(Type.String()), priority: optionalStringEnum(HEARTBEAT_TOOL_PRIORITIES), nextCheck: Type.Optional(Type.String()), - scratch: Type.Optional(Type.String()), + scratch: Type.Optional( + Type.String({ + description: + "Complete replacement for heartbeat monitor prose. Recurring schedules belong in cron jobs, not scratch.", + }), + ), }, { additionalProperties: false }, ); @@ -48,7 +53,7 @@ export function createHeartbeatResponseTool(): AnyAgentTool { name: HEARTBEAT_RESPONSE_TOOL_NAME, displaySummary: "Record heartbeat outcome/notify choice.", description: - "Record heartbeat result. `notify=false` no visible send. `notify=true` needs concise notificationText.", + "Record heartbeat result. `notify=false` no visible send. `notify=true` needs concise notificationText. Scratch is monitor prose only; manage recurring tasks with cron.", parameters: HeartbeatResponseToolSchema, execute: async (_toolCallId, args) => { if (!isRecord(args)) { diff --git a/src/auto-reply/heartbeat.test.ts b/src/auto-reply/heartbeat.test.ts index fbb53d954a9f..c4aa82e56de5 100644 --- a/src/auto-reply/heartbeat.test.ts +++ b/src/auto-reply/heartbeat.test.ts @@ -1,10 +1,9 @@ -/** Tests heartbeat prompt, token, task parsing, and due-time helpers. */ +/** Tests heartbeat prompt and token helpers. */ import { describe, expect, it } from "vitest"; import { DEFAULT_HEARTBEAT_ACK_MAX_CHARS, HEARTBEAT_RESPONSE_TOOL_PROMPT, isHeartbeatContentEffectivelyEmpty, - parseHeartbeatTasks, resolveHeartbeatPromptForResponseTool, stripHeartbeatToken, } from "./heartbeat.js"; @@ -320,32 +319,3 @@ describe("resolveHeartbeatPromptForResponseTool", () => { expect(prompt).toContain("notify=false"); }); }); - -describe("parseHeartbeatTasks", () => { - it("does not bleed top-level interval/prompt fields into task parsing", () => { - const content = `tasks: - - name: email-check - interval: 30m - prompt: Check for urgent emails -interval: should-not-bleed -`; - expect(parseHeartbeatTasks(content)).toEqual([ - { - name: "email-check", - interval: "30m", - prompt: "Check for urgent emails", - }, - ]); - }); - - it("ignores task blocks inside HTML comments", () => { - const content = ` -`; - expect(parseHeartbeatTasks(content)).toEqual([]); - }); -}); diff --git a/src/auto-reply/heartbeat.ts b/src/auto-reply/heartbeat.ts index d9549a3a9a6b..a43b031db5af 100644 --- a/src/auto-reply/heartbeat.ts +++ b/src/auto-reply/heartbeat.ts @@ -1,20 +1,13 @@ /** Heartbeat prompt defaults, token stripping, task parsing, and due-time helpers. */ import { normalizeOptionalString } from "@openclaw/normalization-core/string-coerce"; -import { parseDurationMs } from "../cli/parse-duration.js"; import { escapeRegExp } from "../shared/regexp.js"; import { HEARTBEAT_TOKEN } from "./tokens.js"; -/** YAML-like task entry parsed from heartbeat monitor scratch. */ -export type HeartbeatTask = { - name: string; - interval: string; - prompt: string; -}; - // Default heartbeat prompt (used when config.agents.defaults.heartbeat.prompt is unset). // Keep it tight and avoid encouraging the model to invent/rehash "open loops" from prior chat context. -const HEARTBEAT_CONTEXT_PROMPT = - "Follow the heartbeat monitor scratch context when provided. Do not infer or repeat old tasks from prior chats."; +export const HEARTBEAT_CRON_TASK_GUIDANCE = + "Recurring tasks are cron jobs; create or change their schedules with cron tools or the openclaw cron CLI, not heartbeat scratch."; +const HEARTBEAT_CONTEXT_PROMPT = `Follow the heartbeat monitor scratch context when provided. ${HEARTBEAT_CRON_TASK_GUIDANCE} Do not infer or repeat old tasks from prior chats.`; /** Default prompt for heartbeat turns when config does not override it. */ export const HEARTBEAT_PROMPT = `${HEARTBEAT_CONTEXT_PROMPT} If nothing needs attention, reply HEARTBEAT_OK.`; export const HEARTBEAT_RESPONSE_TOOL_INSTRUCTIONS = @@ -250,113 +243,3 @@ export function stripHeartbeatToken( return { shouldSkip: false, text: rest, didStrip: true }; } - -/** - * Parse heartbeat tasks from HEARTBEAT.md content. - * Supports YAML-like task definitions: - * - * tasks: - * - name: email-check - * interval: 30m - * prompt: "Check for urgent unread emails" - */ -export function parseHeartbeatTasks(content: string): HeartbeatTask[] { - const tasks: HeartbeatTask[] = []; - const lines = stripHeartbeatHtmlComments(content); - let inTasksBlock = false; - - for (const [i, line] of lines.entries()) { - const trimmed = line.trim(); - - // Detect tasks block start. - if (trimmed === "tasks:") { - inTasksBlock = true; - continue; - } - - if (!inTasksBlock) { - continue; - } - - // End of tasks block is any new top-level content that is not a task field. - const isTaskField = - trimmed.startsWith("interval:") || - trimmed.startsWith("prompt:") || - trimmed.startsWith("- name:"); - if ( - !isTaskField && - !trimmed.startsWith(" ") && - !trimmed.startsWith("\t") && - trimmed && - !trimmed.startsWith("-") - ) { - inTasksBlock = false; - continue; - } - - // Parse a task entry and scan following indented fields. - if (trimmed.startsWith("- name:")) { - const name = trimmed - .replace("- name:", "") - .trim() - .replace(/^["']|["']$/g, ""); - let interval = ""; - let prompt = ""; - - // Look ahead for interval and prompt - for (const nextLine of lines.slice(i + 1)) { - const nextTrimmed = nextLine.trim(); - - // End of this task - if (nextTrimmed.startsWith("- name:")) { - break; - } - - // Check for task fields BEFORE checking for end of block - if ( - nextTrimmed.startsWith("interval:") && - (nextLine.startsWith(" ") || nextLine.startsWith("\t")) - ) { - interval = nextTrimmed - .replace("interval:", "") - .trim() - .replace(/^["']|["']$/g, ""); - } else if ( - nextTrimmed.startsWith("prompt:") && - (nextLine.startsWith(" ") || nextLine.startsWith("\t")) - ) { - prompt = nextTrimmed - .replace("prompt:", "") - .trim() - .replace(/^["']|["']$/g, ""); - } else if (!nextTrimmed.startsWith(" ") && !nextTrimmed.startsWith("\t") && nextTrimmed) { - // End of tasks block - inTasksBlock = false; - break; - } - } - - if (name && interval && prompt) { - tasks.push({ name, interval, prompt }); - } - } - } - - return tasks; -} - -/** - * Check if a task is due based on its interval and last run time. - */ -export function isTaskDue(lastRunMs: number | undefined, interval: string, nowMs: number): boolean { - if (lastRunMs === undefined) { - return true; // Never run, always due - } - - try { - const intervalMs = parseDurationMs(interval, { defaultUnit: "m" }); - return nowMs - lastRunMs >= intervalMs; - } catch { - return false; - } -} diff --git a/src/commands/doctor-heartbeat-task-migration.test.ts b/src/commands/doctor-heartbeat-task-migration.test.ts new file mode 100644 index 000000000000..f4ca75f99564 --- /dev/null +++ b/src/commands/doctor-heartbeat-task-migration.test.ts @@ -0,0 +1,720 @@ +import fs from "node:fs/promises"; +import os from "node:os"; +import path from "node:path"; +import { afterEach, beforeEach, describe, expect, it, vi } from "vitest"; +import { resolveDefaultAgentId } from "../agents/agent-scope.js"; +import * as sessionAccessor from "../config/sessions/session-accessor.js"; +import { replaceSessionEntry } from "../config/sessions/session-accessor.js"; +import type { OpenClawConfig } from "../config/types.openclaw.js"; +import { resolveHeartbeatMonitorSpecs } from "../cron/heartbeat-monitor.js"; +import { heartbeatTaskDeclarationKey, isHeartbeatTaskCronJob } from "../cron/heartbeat-task.js"; +import { readCronJobScratchState, writeCronJobScratch } from "../cron/scratch-store.js"; +import { CronService } from "../cron/service.js"; +import { loadCronJobsStore, resolveCronJobsStorePathFromConfig } from "../cron/store.js"; +import { resolveHeartbeatSession } from "../infra/heartbeat-runner.js"; +import { closeOpenClawAgentDatabasesForTest } from "../state/openclaw-agent-db.js"; +import { closeOpenClawStateDatabaseForTest } from "../state/openclaw-state-db.js"; +import { + collectHeartbeatTaskMigrationFindings, + maybeMigrateHeartbeatTasksToCron, +} from "./doctor-heartbeat-task-migration.js"; + +const tempDirs: string[] = []; +let originalHome: string | undefined; +let originalStateDir: string | undefined; + +function createTestCronService(storePath: string, cfg: OpenClawConfig, nowMs: number): CronService { + const noop = () => {}; + const log = { debug: noop, info: noop, warn: noop, error: noop }; + return new CronService({ + storePath, + nowMs: () => nowMs, + cronEnabled: false, + cronConfig: cfg.cron, + defaultAgentId: resolveDefaultAgentId(cfg), + log, + enqueueSystemEvent: () => false, + requestHeartbeat: noop, + runIsolatedAgentJob: async () => ({ + status: "skipped", + error: "tests do not execute cron jobs", + }), + }); +} + +beforeEach(() => { + originalHome = process.env.HOME; + originalStateDir = process.env.OPENCLAW_STATE_DIR; +}); + +afterEach(async () => { + vi.restoreAllMocks(); + closeOpenClawAgentDatabasesForTest(); + closeOpenClawStateDatabaseForTest(); + if (originalHome === undefined) { + delete process.env.HOME; + } else { + process.env.HOME = originalHome; + } + if (originalStateDir === undefined) { + delete process.env.OPENCLAW_STATE_DIR; + } else { + process.env.OPENCLAW_STATE_DIR = originalStateDir; + } + await Promise.all(tempDirs.splice(0).map((dir) => fs.rm(dir, { recursive: true, force: true }))); +}); + +async function createFixture( + nowMs: number, + scratchContent = `# Operations + +tasks: + - name: inbox + interval: 1h + prompt: Check urgent inbox items + - name: calendar + interval: 2h + prompt: Check the next meetings + +# Keep alerts concise +`, +) { + const root = await fs.mkdtemp(path.join(os.tmpdir(), "openclaw-heartbeat-task-migration-")); + tempDirs.push(root); + const env = { ...process.env, HOME: path.join(root, "home"), OPENCLAW_STATE_DIR: root }; + process.env.HOME = env.HOME; + process.env.OPENCLAW_STATE_DIR = env.OPENCLAW_STATE_DIR; + const cfg = { + agents: { defaults: { heartbeat: { every: "30m" } }, list: [{ id: "main" }] }, + } as OpenClawConfig; + const storePath = resolveCronJobsStorePathFromConfig(cfg, env); + const cron = createTestCronService(storePath, cfg, nowMs); + const spec = resolveHeartbeatMonitorSpecs(cfg, [])[0]; + if (!spec) { + throw new Error("expected heartbeat monitor spec"); + } + const added = await cron.add(spec.input, { enabledExplicit: true, systemOwned: true }); + const monitor = "job" in added ? added.job : added; + writeCronJobScratch({ + storePath, + jobId: monitor.id, + content: scratchContent, + expectedRevision: 0, + options: { env }, + }); + const session = resolveHeartbeatSession( + cfg, + "main", + cfg.agents?.defaults?.heartbeat, + undefined, + env, + ); + await replaceSessionEntry( + { storePath: session.storePath, sessionKey: session.sessionKey, env }, + { + sessionId: "heartbeat-main", + updatedAt: nowMs, + heartbeatTaskState: { inbox: nowMs - 30 * 60_000 }, + }, + ); + return { cfg, env, monitor, nowMs, session, storePath }; +} + +async function createExistingInboxJob(fixture: Awaited>) { + const cron = createTestCronService(fixture.storePath, fixture.cfg, fixture.nowMs - 60_000); + const result = await cron.add( + { + declarationKey: heartbeatTaskDeclarationKey("main", "inbox"), + displayName: "Previous inbox task", + name: "inbox", + description: "Existing operator-owned state", + agentId: "main", + enabled: true, + schedule: { + kind: "every", + everyMs: 5 * 60 * 60_000, + anchorMs: fixture.nowMs - 60_000, + }, + payload: { kind: "systemEvent", text: "Previous inbox prompt" }, + sessionTarget: "main", + wakeMode: "next-heartbeat", + state: { lastRunAtMs: fixture.nowMs - 10_000 }, + }, + { enabledExplicit: true, matchesExisting: isHeartbeatTaskCronJob }, + ); + return structuredClone("job" in result ? result.job : result); +} + +describe("heartbeat scratch task cron migration", () => { + it("previews, preserves cadence, clears the block, and reruns idempotently", async () => { + const fixture = await createFixture(2_000_000_000_000); + + await expect(collectHeartbeatTaskMigrationFindings(fixture.cfg, fixture.env)).resolves.toEqual([ + expect.objectContaining({ + checkId: "core/doctor/heartbeat-task-cron-migration", + requirement: "heartbeat-tasks-in-scratch", + target: "main", + }), + ]); + const preview = await maybeMigrateHeartbeatTasksToCron({ + cfg: fixture.cfg, + env: fixture.env, + shouldRepair: false, + nowMs: fixture.nowMs, + }); + expect(preview).toEqual({ changes: [], warnings: [] }); + expect( + (await loadCronJobsStore(fixture.storePath)).jobs.filter(isHeartbeatTaskCronJob), + ).toEqual([]); + + const migrated = await maybeMigrateHeartbeatTasksToCron({ + cfg: fixture.cfg, + env: fixture.env, + shouldRepair: true, + nowMs: fixture.nowMs, + }); + expect(migrated.warnings).toEqual([]); + expect(migrated.changes).toHaveLength(1); + + const jobs = (await loadCronJobsStore(fixture.storePath)).jobs + .filter(isHeartbeatTaskCronJob) + .toSorted((a, b) => a.name.localeCompare(b.name)); + expect(jobs).toHaveLength(2); + expect( + jobs.map((job) => ({ name: job.name, schedule: job.schedule, payload: job.payload })), + ).toEqual([ + { + name: "calendar", + schedule: { kind: "every", everyMs: 2 * 60 * 60_000, anchorMs: fixture.nowMs + 1 }, + payload: { kind: "systemEvent", text: "Check the next meetings" }, + }, + { + name: "inbox", + schedule: { + kind: "every", + everyMs: 60 * 60_000, + anchorMs: fixture.nowMs + 30 * 60_000, + }, + payload: { kind: "systemEvent", text: "Check urgent inbox items" }, + }, + ]); + expect(jobs.find((job) => job.name === "calendar")?.state.nextRunAtMs).toBe(fixture.nowMs + 1); + expect(jobs.find((job) => job.name === "inbox")?.state.nextRunAtMs).toBe( + fixture.nowMs + 30 * 60_000, + ); + + const scratch = readCronJobScratchState(fixture.storePath, fixture.monitor.id, { + env: fixture.env, + }).scratch; + expect(scratch?.content).toContain("# Operations"); + expect(scratch?.content).toContain("# Keep alerts concise"); + expect(scratch?.content).not.toContain("tasks:"); + expect( + resolveHeartbeatSession(fixture.cfg, "main", undefined, undefined, fixture.env).entry + ?.heartbeatTaskState, + ).toBeUndefined(); + + const rerun = await maybeMigrateHeartbeatTasksToCron({ + cfg: fixture.cfg, + env: fixture.env, + shouldRepair: true, + nowMs: fixture.nowMs + 10_000, + }); + expect(rerun).toEqual({ changes: [], warnings: [] }); + expect( + (await loadCronJobsStore(fixture.storePath)).jobs.filter(isHeartbeatTaskCronJob), + ).toHaveLength(2); + }); + + it("leaves cron jobs and legacy timestamps untouched when the scratch revision changes", async () => { + const fixture = await createFixture(2_000_000_000_000); + const existingSnapshot = await createExistingInboxJob(fixture); + const concurrentScratch = `# Concurrent replacement +tasks: + - name: follow-up + interval: 3h + prompt: Run the concurrent follow-up +# Keep concurrent prose +`; + const migration = maybeMigrateHeartbeatTasksToCron({ + cfg: fixture.cfg, + env: fixture.env, + shouldRepair: true, + nowMs: fixture.nowMs, + }); + const current = readCronJobScratchState(fixture.storePath, fixture.monitor.id, { + env: fixture.env, + }); + writeCronJobScratch({ + storePath: fixture.storePath, + jobId: fixture.monitor.id, + content: concurrentScratch, + expectedRevision: current.currentRevision, + options: { env: fixture.env }, + }); + const result = await migration; + + expect(result.changes).toEqual([]); + expect(result.warnings.join("\n")).toContain("scratch changed during task migration"); + expect( + readCronJobScratchState(fixture.storePath, fixture.monitor.id, { env: fixture.env }).scratch + ?.content, + ).toBe(concurrentScratch); + const jobs = (await loadCronJobsStore(fixture.storePath)).jobs.filter(isHeartbeatTaskCronJob); + expect(jobs).toEqual([existingSnapshot]); + expect( + resolveHeartbeatSession(fixture.cfg, "main", undefined, undefined, fixture.env).entry + ?.heartbeatTaskState, + ).toEqual({ inbox: fixture.nowMs - 30 * 60_000 }); + }); + + it("serializes two plans pinned to one scratch revision and converges the loser on rerun", async () => { + const fixture = await createFixture(2_000_000_000_000); + const outcomes = await Promise.all([ + maybeMigrateHeartbeatTasksToCron({ + cfg: fixture.cfg, + env: fixture.env, + shouldRepair: true, + nowMs: fixture.nowMs, + }), + maybeMigrateHeartbeatTasksToCron({ + cfg: fixture.cfg, + env: fixture.env, + shouldRepair: true, + nowMs: fixture.nowMs, + }), + ]); + + expect(outcomes.filter((outcome) => outcome.changes.length === 1)).toHaveLength(1); + expect( + outcomes.filter((outcome) => + outcome.warnings.some((warning) => + warning.includes("scratch changed during task migration"), + ), + ), + ).toHaveLength(1); + const committedJobs = (await loadCronJobsStore(fixture.storePath)).jobs.filter( + isHeartbeatTaskCronJob, + ); + expect(committedJobs).toHaveLength(2); + expect(new Set(committedJobs.map((job) => job.declarationKey)).size).toBe(2); + expect( + readCronJobScratchState(fixture.storePath, fixture.monitor.id, { env: fixture.env }).scratch + ?.content, + ).not.toContain("tasks:"); + + const rerun = await maybeMigrateHeartbeatTasksToCron({ + cfg: fixture.cfg, + env: fixture.env, + shouldRepair: true, + nowMs: fixture.nowMs + 10_000, + }); + expect(rerun).toEqual({ changes: [], warnings: [] }); + expect( + (await loadCronJobsStore(fixture.storePath)).jobs.filter(isHeartbeatTaskCronJob), + ).toEqual(committedJobs); + }); + + it("tolerates a crash after the state transaction and before legacy timestamp cleanup", async () => { + const fixture = await createFixture(2_000_000_000_000); + const cleanup = vi + .spyOn(sessionAccessor, "patchSessionEntry") + .mockRejectedValueOnce(new Error("simulated post-commit crash")); + const result = await maybeMigrateHeartbeatTasksToCron({ + cfg: fixture.cfg, + env: fixture.env, + shouldRepair: true, + nowMs: fixture.nowMs, + }); + + expect(result.changes).toHaveLength(1); + expect(result.warnings.join("\n")).toContain("simulated post-commit crash"); + const committedJobs = (await loadCronJobsStore(fixture.storePath)).jobs.filter( + isHeartbeatTaskCronJob, + ); + expect(committedJobs).toHaveLength(2); + expect( + readCronJobScratchState(fixture.storePath, fixture.monitor.id, { env: fixture.env }).scratch + ?.content, + ).not.toContain("tasks:"); + expect( + resolveHeartbeatSession(fixture.cfg, "main", undefined, undefined, fixture.env).entry + ?.heartbeatTaskState, + ).toEqual({ inbox: fixture.nowMs - 30 * 60_000 }); + + cleanup.mockRestore(); + await expect( + maybeMigrateHeartbeatTasksToCron({ + cfg: fixture.cfg, + env: fixture.env, + shouldRepair: true, + nowMs: fixture.nowMs + 10_000, + }), + ).resolves.toEqual({ changes: [], warnings: [] }); + expect( + (await loadCronJobsStore(fixture.storePath)).jobs.filter(isHeartbeatTaskCronJob), + ).toEqual(committedJobs); + }); + + it("migrates duplicate-name tasks with stable identities and shared initial due time", async () => { + const fixture = await createFixture(2_000_000_000_000); + const state = readCronJobScratchState(fixture.storePath, fixture.monitor.id, { + env: fixture.env, + }); + const duplicate = `tasks: + - name: inbox + interval: 1h + prompt: First + - name: inbox + interval: 1h + prompt: Second +`; + writeCronJobScratch({ + storePath: fixture.storePath, + jobId: fixture.monitor.id, + content: duplicate, + expectedRevision: state.currentRevision, + options: { env: fixture.env }, + }); + + const result = await maybeMigrateHeartbeatTasksToCron({ + cfg: fixture.cfg, + env: fixture.env, + shouldRepair: true, + nowMs: fixture.nowMs, + }); + + expect(result.warnings).toEqual([]); + expect(result.changes).toHaveLength(1); + expect( + readCronJobScratchState(fixture.storePath, fixture.monitor.id, { env: fixture.env }).scratch + ?.content, + ).toBe(""); + const jobs = (await loadCronJobsStore(fixture.storePath)).jobs + .filter(isHeartbeatTaskCronJob) + .toSorted((left, right) => left.payload.text.localeCompare(right.payload.text)); + expect(jobs).toHaveLength(2); + expect(jobs.map((job) => job.declarationKey)).toEqual([ + heartbeatTaskDeclarationKey("main", "inbox", 0), + heartbeatTaskDeclarationKey("main", "inbox", 1), + ]); + expect(new Set(jobs.map((job) => job.declarationKey)).size).toBe(2); + expect(jobs.map((job) => job.schedule)).toEqual([ + { + kind: "every", + everyMs: 60 * 60_000, + anchorMs: fixture.nowMs + 30 * 60_000, + }, + { + kind: "every", + everyMs: 60 * 60_000, + anchorMs: fixture.nowMs + 30 * 60_000, + }, + ]); + const initialIdentities = jobs.map((job) => ({ id: job.id, key: job.declarationKey })); + + await expect( + maybeMigrateHeartbeatTasksToCron({ + cfg: fixture.cfg, + env: fixture.env, + shouldRepair: true, + nowMs: fixture.nowMs + 10_000, + }), + ).resolves.toEqual({ changes: [], warnings: [] }); + const rerunJobs = (await loadCronJobsStore(fixture.storePath)).jobs + .filter(isHeartbeatTaskCronJob) + .toSorted((left, right) => left.payload.text.localeCompare(right.payload.text)); + expect(rerunJobs.map((job) => ({ id: job.id, key: job.declarationKey }))).toEqual( + initialIdentities, + ); + }); + + it("migrates a mixed unique and duplicate-name task block completely", async () => { + const content = `tasks: + - name: inbox + interval: 1h + prompt: First inbox pass + - name: calendar + interval: 2h + prompt: Calendar pass + - name: inbox + interval: 1h + prompt: Second inbox pass +`; + const fixture = await createFixture(2_000_000_000_000, content); + + const result = await maybeMigrateHeartbeatTasksToCron({ + cfg: fixture.cfg, + env: fixture.env, + shouldRepair: true, + nowMs: fixture.nowMs, + }); + + expect(result.warnings).toEqual([]); + const jobs = (await loadCronJobsStore(fixture.storePath)).jobs.filter(isHeartbeatTaskCronJob); + expect(jobs).toHaveLength(3); + expect(new Set(jobs.map((job) => job.declarationKey)).size).toBe(3); + expect( + jobs.map((job) => job.payload.text).toSorted((left, right) => left.localeCompare(right)), + ).toEqual(["Calendar pass", "First inbox pass", "Second inbox pass"]); + expect( + readCronJobScratchState(fixture.storePath, fixture.monitor.id, { env: fixture.env }).scratch + ?.content, + ).toBe(""); + }); + + it("refuses orphan task fields beside a valid task without changing scratch", async () => { + const content = `# Operations +tasks: + interval: 15m + prompt: Orphaned work must not disappear + - name: inbox + interval: 1h + prompt: Check urgent inbox items +# Keep alerts concise +`; + const fixture = await createFixture(2_000_000_000_000, content); + + await expect(collectHeartbeatTaskMigrationFindings(fixture.cfg, fixture.env)).resolves.toEqual([ + expect.objectContaining({ + severity: "error", + requirement: "heartbeat-task-migration-blocked", + message: expect.stringContaining("incomplete name/interval/prompt entry"), + }), + ]); + const result = await maybeMigrateHeartbeatTasksToCron({ + cfg: fixture.cfg, + env: fixture.env, + shouldRepair: true, + nowMs: fixture.nowMs, + }); + + expect(result.changes).toEqual([]); + expect(result.warnings.join("\n")).toContain("incomplete name/interval/prompt entry"); + expect( + readCronJobScratchState(fixture.storePath, fixture.monitor.id, { env: fixture.env }).scratch + ?.content, + ).toBe(content); + expect( + (await loadCronJobsStore(fixture.storePath)).jobs.filter(isHeartbeatTaskCronJob), + ).toEqual([]); + }); + + it("does not migrate a task block hidden by a mid-line HTML comment opener", async () => { + const content = `Notes +# Keep this scratch +`; + const fixture = await createFixture(2_000_000_000_000, content); + + await expect(collectHeartbeatTaskMigrationFindings(fixture.cfg, fixture.env)).resolves.toEqual( + [], + ); + await expect( + maybeMigrateHeartbeatTasksToCron({ + cfg: fixture.cfg, + env: fixture.env, + shouldRepair: true, + nowMs: fixture.nowMs, + }), + ).resolves.toEqual({ changes: [], warnings: [] }); + + expect( + (await loadCronJobsStore(fixture.storePath)).jobs.filter(isHeartbeatTaskCronJob), + ).toEqual([]); + expect( + readCronJobScratchState(fixture.storePath, fixture.monitor.id, { env: fixture.env }).scratch + ?.content, + ).toBe(content); + }); + + it("keeps a multiline comment closed when its opener shares a migrated task line", async () => { + const content = `tasks: + - name: active + interval: 1h + prompt: Run the active check +# Keep this scratch +`; + const fixture = await createFixture(2_000_000_000_000, content); + + const migrated = await maybeMigrateHeartbeatTasksToCron({ + cfg: fixture.cfg, + env: fixture.env, + shouldRepair: true, + nowMs: fixture.nowMs, + }); + + expect(migrated.warnings).toEqual([]); + const jobs = (await loadCronJobsStore(fixture.storePath)).jobs.filter(isHeartbeatTaskCronJob); + expect(jobs.map((job) => job.name)).toEqual(["active"]); + const scratch = readCronJobScratchState(fixture.storePath, fixture.monitor.id, { + env: fixture.env, + }).scratch?.content; + expect(scratch).toContain(``); + await expect( + maybeMigrateHeartbeatTasksToCron({ + cfg: fixture.cfg, + env: fixture.env, + shouldRepair: true, + nowMs: fixture.nowMs + 10_000, + }), + ).resolves.toEqual({ changes: [], warnings: [] }); + }); + + it("preserves indented non-task prose and its line endings byte-for-byte", async () => { + const content = + "# Operations\r\n" + + "tasks:\r\n" + + " - name: inbox\r\n" + + " interval: 1h\r\n" + + " prompt: Check urgent inbox items\r\n" + + " Keep this indented note exactly. \r\n" + + "# Keep alerts concise\r\n"; + const fixture = await createFixture(2_000_000_000_000, content); + + const result = await maybeMigrateHeartbeatTasksToCron({ + cfg: fixture.cfg, + env: fixture.env, + shouldRepair: true, + nowMs: fixture.nowMs, + }); + + expect(result.warnings).toEqual([]); + expect( + readCronJobScratchState(fixture.storePath, fixture.monitor.id, { env: fixture.env }).scratch + ?.content, + ).toBe("# Operations\r\n Keep this indented note exactly. \r\n# Keep alerts concise\r\n"); + }); + + it("uses the supplied environment for legacy session timing and cleanup", async () => { + const fixture = await createFixture(2_000_000_000_000); + const suppliedHome = await fs.mkdtemp( + path.join(os.tmpdir(), "openclaw-heartbeat-task-migration-supplied-home-"), + ); + tempDirs.push(suppliedHome); + fixture.cfg.session = { + store: "~/.openclaw/agents/{agentId}/sessions/sessions.json", + }; + const suppliedEnv = { ...fixture.env, HOME: suppliedHome }; + const suppliedSession = resolveHeartbeatSession( + fixture.cfg, + "main", + fixture.cfg.agents?.defaults?.heartbeat, + undefined, + suppliedEnv, + ); + await replaceSessionEntry( + { + storePath: suppliedSession.storePath, + sessionKey: suppliedSession.sessionKey, + env: suppliedEnv, + }, + { + sessionId: "supplied-heartbeat-main", + updatedAt: fixture.nowMs, + heartbeatTaskState: { inbox: fixture.nowMs - 30 * 60_000 }, + }, + ); + const ambientSession = resolveHeartbeatSession( + fixture.cfg, + "main", + fixture.cfg.agents?.defaults?.heartbeat, + ); + await replaceSessionEntry( + { + storePath: ambientSession.storePath, + sessionKey: ambientSession.sessionKey, + }, + { + sessionId: "ambient-heartbeat-main", + updatedAt: fixture.nowMs, + heartbeatTaskState: { + inbox: fixture.nowMs - 10 * 60_000, + untouched: fixture.nowMs - 5_000, + }, + }, + ); + + const result = await maybeMigrateHeartbeatTasksToCron({ + cfg: fixture.cfg, + env: suppliedEnv, + shouldRepair: true, + nowMs: fixture.nowMs, + }); + + expect(result.warnings).toEqual([]); + const inbox = (await loadCronJobsStore(fixture.storePath)).jobs.find( + (job) => isHeartbeatTaskCronJob(job) && job.name === "inbox", + ); + expect(inbox?.schedule).toEqual({ + kind: "every", + everyMs: 60 * 60_000, + anchorMs: fixture.nowMs + 30 * 60_000, + }); + expect( + resolveHeartbeatSession(fixture.cfg, "main", undefined, undefined, suppliedEnv).entry + ?.heartbeatTaskState, + ).toBeUndefined(); + expect(resolveHeartbeatSession(fixture.cfg, "main").entry?.heartbeatTaskState).toEqual({ + inbox: fixture.nowMs - 10 * 60_000, + untouched: fixture.nowMs - 5_000, + }); + }); + + it("removes consecutive task blocks in one idempotent migration", async () => { + const content = `# Operations +tasks: + - name: inbox + interval: 1h + prompt: Check urgent inbox items +tasks: + - name: calendar + interval: 2h + prompt: Check the next meetings +# Keep alerts concise +`; + const fixture = await createFixture(2_000_000_000_000, content); + + const migrated = await maybeMigrateHeartbeatTasksToCron({ + cfg: fixture.cfg, + env: fixture.env, + shouldRepair: true, + nowMs: fixture.nowMs, + }); + + expect(migrated.warnings).toEqual([]); + expect( + readCronJobScratchState(fixture.storePath, fixture.monitor.id, { env: fixture.env }).scratch + ?.content, + ).toBe("# Operations\n# Keep alerts concise\n"); + expect( + (await loadCronJobsStore(fixture.storePath)).jobs.filter(isHeartbeatTaskCronJob), + ).toHaveLength(2); + + await expect( + maybeMigrateHeartbeatTasksToCron({ + cfg: fixture.cfg, + env: fixture.env, + shouldRepair: true, + nowMs: fixture.nowMs + 10_000, + }), + ).resolves.toEqual({ changes: [], warnings: [] }); + expect( + (await loadCronJobsStore(fixture.storePath)).jobs.filter(isHeartbeatTaskCronJob), + ).toHaveLength(2); + }); +}); diff --git a/src/commands/doctor-heartbeat-task-migration.ts b/src/commands/doctor-heartbeat-task-migration.ts new file mode 100644 index 000000000000..5fd64b991122 --- /dev/null +++ b/src/commands/doctor-heartbeat-task-migration.ts @@ -0,0 +1,611 @@ +/** Doctor-owned migration from heartbeat scratch `tasks:` blocks into cron jobs. */ + +import { randomUUID } from "node:crypto"; +import type { DatabaseSync } from "node:sqlite"; +import { isDeepStrictEqual } from "node:util"; +import { truncateUtf16Safe } from "@openclaw/normalization-core/utf16-slice"; +import { note } from "../../packages/terminal-core/src/note.js"; +import { formatCliCommand } from "../cli/command-format.js"; +import { parseDurationMs } from "../cli/parse-duration.js"; +import { patchSessionEntry } from "../config/sessions/session-accessor.js"; +import type { OpenClawConfig } from "../config/types.openclaw.js"; +import { heartbeatTaskDeclarationKey, isHeartbeatTaskCronJob } from "../cron/heartbeat-task.js"; +import { cronSchedulingInputsEqual } from "../cron/schedule-identity.js"; +import { readHeartbeatMonitorScratch } from "../cron/scratch-store.js"; +import { computeJobNextRunAtMs, hasScheduledNextRunAtMs } from "../cron/service/jobs.js"; +import { resolveCronJobsStorePathFromConfig } from "../cron/store.js"; +import { cronStoreKey } from "../cron/store/key.js"; +import { + assertCronStoreCanPersist, + loadedCronStoreFromRows, + loadCronRows, + upsertCronJobRow, +} from "../cron/store/row-codec.js"; +import { getCronStoreKysely } from "../cron/store/schema.js"; +import type { CronJob } from "../cron/types.js"; +import type { HealthFinding } from "../flows/health-checks.js"; +import { resolveHeartbeatAgents, resolveHeartbeatSession } from "../infra/heartbeat-runner.js"; +import { executeSqliteQuerySync } from "../infra/kysely-sync.js"; +import { + openOpenClawStateDatabase, + runOpenClawStateWriteTransaction, +} from "../state/openclaw-state-db.js"; +import { shortenHomePath } from "../utils.js"; +import { analyzeLegacyHeartbeatTasks, type LegacyHeartbeatTask } from "./heartbeat-task-legacy.js"; + +const HEARTBEAT_TASK_MIGRATION_CHECK_ID = "core/doctor/heartbeat-task-cron-migration"; + +type HeartbeatTaskMigrationResult = { changes: string[]; warnings: string[] }; + +function errorMessage(error: unknown): string { + return error instanceof Error ? error.message : String(error); +} + +type ValidatedHeartbeatTask = { + task: LegacyHeartbeatTask; + intervalMs: number; + occurrenceIndex: number; +}; + +function validateTasks( + tasks: readonly LegacyHeartbeatTask[], + declaredEntryCount: number, +): ValidatedHeartbeatTask[] { + if (tasks.length === 0) { + throw new Error("tasks: block has no complete name/interval/prompt entries"); + } + if (tasks.length !== declaredEntryCount) { + throw new Error("tasks: block contains an incomplete name/interval/prompt entry"); + } + const occurrenceCounts = new Map(); + const validated: ValidatedHeartbeatTask[] = []; + for (const task of tasks) { + const intervalMs = parseDurationMs(task.interval, { defaultUnit: "m" }); + if (intervalMs <= 0) { + throw new Error(`task ${JSON.stringify(task.name)} interval must be greater than zero`); + } + const occurrenceIndex = occurrenceCounts.get(task.name) ?? 0; + occurrenceCounts.set(task.name, occurrenceIndex + 1); + validated.push({ task, intervalMs, occurrenceIndex }); + } + return validated; +} + +function migrationFinding(params: { + storePath: string; + agentId: string; + message: string; + severity?: HealthFinding["severity"]; + requirement: string; +}): HealthFinding { + return { + checkId: HEARTBEAT_TASK_MIGRATION_CHECK_ID, + severity: params.severity ?? "warning", + message: params.message, + path: params.storePath, + target: params.agentId, + requirement: params.requirement, + fixHint: `Run ${formatCliCommand("openclaw doctor --fix")} to convert heartbeat tasks into cron jobs.`, + }; +} + +/** Reports task blocks still owned by heartbeat scratch without changing them. */ +export async function collectHeartbeatTaskMigrationFindings( + cfg: OpenClawConfig, + env: NodeJS.ProcessEnv = process.env, +): Promise { + const storePath = resolveCronJobsStorePathFromConfig(cfg, env); + const findings: HealthFinding[] = []; + for (const agent of resolveHeartbeatAgents(cfg)) { + let monitor: ReturnType; + try { + monitor = readHeartbeatMonitorScratch(storePath, agent.agentId, { env }); + } catch (error) { + findings.push( + migrationFinding({ + storePath, + agentId: agent.agentId, + requirement: "heartbeat-task-migration-blocked", + severity: "error", + message: `Agent "${agent.agentId}" heartbeat scratch cannot be inspected: ${errorMessage(error)}`, + }), + ); + continue; + } + const content = monitor?.state.scratch?.content; + if (!content) { + continue; + } + const document = analyzeLegacyHeartbeatTasks(content); + if (!document.hasTasksBlock) { + continue; + } + try { + validateTasks(document.tasks, document.taskEntryCount); + findings.push( + migrationFinding({ + storePath, + agentId: agent.agentId, + requirement: "heartbeat-tasks-in-scratch", + message: `Agent "${agent.agentId}" has ${document.tasks.length} heartbeat task${document.tasks.length === 1 ? "" : "s"} that must become cron jobs.`, + }), + ); + } catch (error) { + findings.push( + migrationFinding({ + storePath, + agentId: agent.agentId, + requirement: "heartbeat-task-migration-blocked", + severity: "error", + message: `Agent "${agent.agentId}" heartbeat tasks cannot be migrated: ${errorMessage(error)}`, + }), + ); + } + } + return findings; +} + +function taskJobInput(params: { + agentId: string; + task: LegacyHeartbeatTask; + occurrenceIndex: number; + intervalMs: number; + lastRunAtMs?: number; + existing?: CronJob; + nowMs: number; +}) { + const existingAnchor = + params.existing?.schedule.kind === "every" && + params.existing.schedule.everyMs === params.intervalMs + ? params.existing.schedule.anchorMs + : undefined; + const nextDueMs = + params.lastRunAtMs === undefined || params.lastRunAtMs + params.intervalMs <= params.nowMs + ? params.nowMs + 1 + : params.lastRunAtMs + params.intervalMs; + return { + declarationKey: heartbeatTaskDeclarationKey( + params.agentId, + params.task.name, + params.occurrenceIndex, + ), + displayName: truncateUtf16Safe(`Heartbeat task: ${params.task.name}`, 200), + name: params.task.name, + description: "Migrated from heartbeat monitor scratch by openclaw doctor.", + agentId: params.agentId, + enabled: true, + schedule: { + kind: "every" as const, + everyMs: params.intervalMs, + anchorMs: existingAnchor ?? nextDueMs, + }, + payload: { kind: "systemEvent" as const, text: params.task.prompt }, + sessionTarget: "main" as const, + wakeMode: "next-heartbeat" as const, + ...(params.lastRunAtMs === undefined ? {} : { state: { lastRunAtMs: params.lastRunAtMs } }), + }; +} + +type TaskJobPlan = { + declarationKey: string; + previous?: CronJob; + job: CronJob; + sortOrder: number; +}; + +type AgentTaskMigrationPlan = { + monitorJobId: string; + scratchRevision: number; + strippedContent: string; + jobs: TaskJobPlan[]; +}; + +type CronPlanningSnapshot = { + jobs: CronJob[]; + sortOrderByJobId: Map; + nextSortOrder: number; +}; + +type MigrationCommitResult = + | { ok: true; currentRevision: number } + | { ok: false; reason: "job-conflict" | "revision-conflict" }; + +function taskDeclarativeFields(job: CronJob) { + return { + schedule: job.schedule, + pacing: job.pacing, + trigger: job.trigger, + payload: job.payload, + delivery: job.delivery, + displayName: job.displayName, + enabled: job.enabled, + }; +} + +function convergeTaskJob(params: { + agentId: string; + task: LegacyHeartbeatTask; + occurrenceIndex: number; + intervalMs: number; + lastRunAtMs?: number; + existing?: CronJob; + nowMs: number; +}): CronJob { + const input = taskJobInput(params); + if (!params.existing) { + const { state, ...fields } = input; + const job: CronJob = { + id: randomUUID(), + ...fields, + createdAtMs: params.nowMs, + updatedAtMs: params.nowMs, + state: { ...state }, + }; + job.state.nextRunAtMs = computeJobNextRunAtMs(job, params.nowMs); + return job; + } + + const previous = params.existing; + const job = structuredClone(previous); + job.displayName = input.displayName; + job.schedule = structuredClone(input.schedule); + job.payload = structuredClone(input.payload); + job.enabled = true; + delete job.pacing; + delete job.trigger; + delete job.delivery; + if (isDeepStrictEqual(taskDeclarativeFields(previous), taskDeclarativeFields(job))) { + return job; + } + + job.updatedAtMs = params.nowMs; + if (!cronSchedulingInputsEqual(previous, job)) { + job.state.startupCatchupAtMs = undefined; + job.state.pacedNextRunAtMs = undefined; + job.state.forcePreservedNextRunAtMs = undefined; + job.state.nextRunAtMs = computeJobNextRunAtMs(job, params.nowMs); + } else if (!hasScheduledNextRunAtMs(job.state.nextRunAtMs)) { + job.state.nextRunAtMs = computeJobNextRunAtMs(job, params.nowMs); + } + return job; +} + +async function loadCronPlanningSnapshot( + storePath: string, + env: NodeJS.ProcessEnv, +): Promise { + const rows = loadCronRows(openOpenClawStateDatabase({ env }).db, cronStoreKey(storePath)); + const sortOrderByJobId = new Map(rows.map((row) => [row.job_id, row.sort_order] as const)); + return { + jobs: loadedCronStoreFromRows(rows).store.jobs, + sortOrderByJobId, + nextSortOrder: rows.reduce((max, row) => Math.max(max, row.sort_order + 1), 0), + }; +} + +function reserveSortOrder(snapshot: CronPlanningSnapshot, existing?: CronJob): number { + const persisted = existing ? snapshot.sortOrderByJobId.get(existing.id) : undefined; + if (persisted !== undefined) { + return persisted; + } + const sortOrder = snapshot.nextSortOrder; + snapshot.nextSortOrder += 1; + return sortOrder; +} + +function readScratchRevision(db: DatabaseSync, storeKey: string, jobId: string): number { + return ( + executeSqliteQuerySync( + db, + getCronStoreKysely(db) + .selectFrom("cron_job_scratch") + .select("revision") + .where("store_key", "=", storeKey) + .where("job_id", "=", jobId), + ).rows[0]?.revision ?? 0 + ); +} + +function commitAgentTaskMigration(params: { + storePath: string; + env: NodeJS.ProcessEnv; + nowMs: number; + plan: AgentTaskMigrationPlan; +}): MigrationCommitResult { + const storeKey = cronStoreKey(params.storePath); + return runOpenClawStateWriteTransaction( + ({ db }) => { + if ( + readScratchRevision(db, storeKey, params.plan.monitorJobId) !== params.plan.scratchRevision + ) { + return { ok: false, reason: "revision-conflict" } as const; + } + + const rows = loadCronRows(db, storeKey); + const jobsById = new Map( + loadedCronStoreFromRows(rows).store.jobs.map((job) => [job.id, job] as const), + ); + for (const jobPlan of params.plan.jobs) { + const matchingRows = rows.filter((row) => row.declaration_key === jobPlan.declarationKey); + if (jobPlan.previous) { + const current = jobsById.get(jobPlan.previous.id); + if ( + matchingRows.length !== 1 || + !current || + !isDeepStrictEqual(current, jobPlan.previous) + ) { + return { ok: false, reason: "job-conflict" } as const; + } + } else if (matchingRows.length > 0 || rows.some((row) => row.job_id === jobPlan.job.id)) { + return { ok: false, reason: "job-conflict" } as const; + } + } + + for (const jobPlan of params.plan.jobs) { + if (!jobPlan.previous || !isDeepStrictEqual(jobPlan.previous, jobPlan.job)) { + upsertCronJobRow(db, storeKey, jobPlan.job, jobPlan.sortOrder); + } + } + + const updated = executeSqliteQuerySync( + db, + getCronStoreKysely(db) + .updateTable("cron_job_scratch") + .set({ + content: params.plan.strippedContent, + revision: params.plan.scratchRevision + 1, + source_sha256: null, + updated_at_ms: params.nowMs, + }) + .where("store_key", "=", storeKey) + .where("job_id", "=", params.plan.monitorJobId) + .where("revision", "=", params.plan.scratchRevision), + ); + if (updated.numAffectedRows !== 1n) { + throw new Error("scratch revision changed inside task migration transaction"); + } + // Like cadence materialization, doctor only commits durable rows. A live + // gateway reloads the cron store through its normal reload path and arms + // these persisted nextRunAtMs values; doctor never owns its timer. + return { ok: true, currentRevision: params.plan.scratchRevision + 1 } as const; + }, + { env: params.env }, + { operationLabel: "doctor.heartbeat-task-migration" }, + ); +} + +async function clearLegacyTaskTimestamps(params: { + storePath: string; + sessionKey: string; + env: NodeJS.ProcessEnv; + tasks: readonly LegacyHeartbeatTask[]; +}): Promise { + await patchSessionEntry( + { storePath: params.storePath, sessionKey: params.sessionKey, env: params.env }, + (entry) => { + const remaining = { ...entry.heartbeatTaskState }; + let changed = false; + for (const task of params.tasks) { + if (Object.hasOwn(remaining, task.name)) { + delete remaining[task.name]; + changed = true; + } + } + if (!changed) { + return null; + } + return { + heartbeatTaskState: Object.keys(remaining).length > 0 ? remaining : undefined, + }; + }, + { preserveActivity: true }, + ); +} + +/** Converts valid scratch tasks and removes their source block in one SQLite transaction. */ +export async function maybeMigrateHeartbeatTasksToCron(params: { + cfg: OpenClawConfig; + shouldRepair: boolean; + env?: NodeJS.ProcessEnv; + nowMs?: number; +}): Promise { + const env = params.env ?? process.env; + const nowMs = params.nowMs ?? Date.now(); + const storePath = resolveCronJobsStorePathFromConfig(params.cfg, env); + const changes: string[] = []; + const warnings: string[] = []; + const candidates: Array<{ + agent: ReturnType[number]; + document: ReturnType; + monitor: NonNullable>; + scratchRevision: number; + validatedTasks: ValidatedHeartbeatTask[]; + }> = []; + for (const agent of resolveHeartbeatAgents(params.cfg)) { + let monitor: ReturnType; + try { + monitor = readHeartbeatMonitorScratch(storePath, agent.agentId, { env }); + } catch (error) { + warnings.push( + `Agent "${agent.agentId}" heartbeat scratch could not be inspected: ${errorMessage(error)}.`, + ); + continue; + } + const scratch = monitor?.state.scratch; + if (!monitor || !scratch) { + continue; + } + const document = analyzeLegacyHeartbeatTasks(scratch.content); + if (!document.hasTasksBlock) { + continue; + } + const tasks = document.tasks; + let validatedTasks: ValidatedHeartbeatTask[]; + try { + validatedTasks = validateTasks(tasks, document.taskEntryCount); + } catch (error) { + warnings.push( + `Agent "${agent.agentId}" heartbeat tasks were not migrated: ${errorMessage(error)}.`, + ); + continue; + } + if (!params.shouldRepair) { + note( + `${tasks.length} task${tasks.length === 1 ? "" : "s"} in ${shortenHomePath(storePath)} will become independently scheduled cron jobs for agent "${agent.agentId}".`, + "Heartbeat task migration preview", + ); + continue; + } + candidates.push({ + agent, + document, + monitor, + scratchRevision: scratch.revision, + validatedTasks, + }); + } + + if (!params.shouldRepair || candidates.length === 0) { + if (warnings.length > 0) { + note(warnings.join("\n"), "Doctor warnings"); + } + return { changes, warnings }; + } + + let snapshot: CronPlanningSnapshot; + try { + // The scratch revisions above are pinned before this async planning read. + // Concurrent doctors can therefore plan R together and serialize at commit. + snapshot = await loadCronPlanningSnapshot(storePath, env); + } catch (error) { + const warning = `Could not inspect cron jobs for heartbeat task migration: ${errorMessage(error)}`; + note(warning, "Doctor warnings"); + return { changes, warnings: [...warnings, warning] }; + } + + for (const candidate of candidates) { + const { agent, document, monitor, scratchRevision, validatedTasks } = candidate; + const session = resolveHeartbeatSession( + params.cfg, + agent.agentId, + agent.heartbeat, + undefined, + env, + ); + const legacyState = session.entry?.heartbeatTaskState ?? {}; + const jobPlans: TaskJobPlan[] = []; + let blocked = false; + for (const { task, intervalMs, occurrenceIndex } of validatedTasks) { + const declarationKey = heartbeatTaskDeclarationKey(agent.agentId, task.name, occurrenceIndex); + const matches = snapshot.jobs.filter((job) => job.declarationKey === declarationKey); + const existing = matches[0]; + if ( + matches.length > 1 || + (existing && + (!isHeartbeatTaskCronJob(existing) || + existing.agentId !== agent.agentId || + existing.name !== task.name)) + ) { + warnings.push( + `Agent "${agent.agentId}" task ${JSON.stringify(task.name)} collides with an incompatible cron declaration; scratch was left unchanged.`, + ); + blocked = true; + break; + } + const legacyLastRun = legacyState[task.name]; + const lastRunAtMs = + typeof legacyLastRun === "number" && Number.isFinite(legacyLastRun) + ? legacyLastRun + : undefined; + const job = convergeTaskJob({ + agentId: agent.agentId, + task, + occurrenceIndex, + intervalMs, + lastRunAtMs, + existing, + nowMs, + }); + const sortOrder = reserveSortOrder(snapshot, existing); + jobPlans.push({ + declarationKey, + ...(existing ? { previous: structuredClone(existing) } : {}), + job, + sortOrder, + }); + } + if (blocked) { + continue; + } + + try { + assertCronStoreCanPersist({ version: 1, jobs: jobPlans.map((plan) => plan.job) }); + } catch (error) { + warnings.push( + `Agent "${agent.agentId}" task jobs could not be planned: ${errorMessage(error)}. Scratch was left unchanged.`, + ); + continue; + } + + const plan: AgentTaskMigrationPlan = { + monitorJobId: monitor.jobId, + scratchRevision, + strippedContent: document.strippedContent, + jobs: jobPlans, + }; + let committed: MigrationCommitResult; + try { + committed = commitAgentTaskMigration({ storePath, env, nowMs, plan }); + } catch (error) { + warnings.push( + `Agent "${agent.agentId}" task migration could not be committed: ${errorMessage(error)}. Scratch and cron jobs were left unchanged.`, + ); + continue; + } + if (!committed.ok) { + warnings.push( + committed.reason === "revision-conflict" + ? `Agent "${agent.agentId}" scratch changed during task migration; no changes were committed.` + : `Agent "${agent.agentId}" cron jobs changed during task migration; no changes were committed.`, + ); + continue; + } + + for (const jobPlan of jobPlans) { + const index = snapshot.jobs.findIndex((job) => job.id === jobPlan.job.id); + if (index >= 0) { + snapshot.jobs[index] = jobPlan.job; + } else { + snapshot.jobs.push(jobPlan.job); + } + snapshot.sortOrderByJobId.set(jobPlan.job.id, jobPlan.sortOrder); + } + changes.push( + `Converted ${document.tasks.length} heartbeat task${document.tasks.length === 1 ? "" : "s"} into cron jobs for agent "${agent.agentId}".`, + ); + + try { + // Session task timestamps live in the per-agent database, so they cannot + // join the state-DB commit. They are advisory once cron owns scheduling; + // this idempotent cleanup may safely be retried or skipped after a crash. + await clearLegacyTaskTimestamps({ + storePath: session.storePath, + sessionKey: session.sessionKey, + env, + tasks: document.tasks, + }); + } catch (error) { + warnings.push( + `Agent "${agent.agentId}" legacy task timestamps could not be cleared after migration: ${errorMessage(error)}. Cron jobs remain authoritative and a rerun is safe.`, + ); + } + } + + if (changes.length > 0) { + note(changes.join("\n"), "Doctor changes"); + } + if (warnings.length > 0) { + note(warnings.join("\n"), "Doctor warnings"); + } + return { changes, warnings }; +} diff --git a/src/commands/doctor.fast-path-mocks.ts b/src/commands/doctor.fast-path-mocks.ts index 9641703bf19c..f9ccb4ecf71f 100644 --- a/src/commands/doctor.fast-path-mocks.ts +++ b/src/commands/doctor.fast-path-mocks.ts @@ -163,6 +163,11 @@ vi.mock("./doctor-heartbeat-scratch-migration.js", () => ({ maybeMigrateHeartbeatFilesToScratch: vi.fn().mockResolvedValue({ changes: [], warnings: [] }), })); +vi.mock("./doctor-heartbeat-task-migration.js", () => ({ + collectHeartbeatTaskMigrationFindings: vi.fn().mockResolvedValue([]), + maybeMigrateHeartbeatTasksToCron: vi.fn().mockResolvedValue({ changes: [], warnings: [] }), +})); + vi.mock("../plugins/provider-openai-chatgpt-oauth-tls.js", () => ({ noteOpenAIOAuthTlsPrerequisites: vi.fn().mockResolvedValue(undefined), })); diff --git a/src/commands/heartbeat-task-legacy.ts b/src/commands/heartbeat-task-legacy.ts new file mode 100644 index 000000000000..d6cc8f1b6f12 --- /dev/null +++ b/src/commands/heartbeat-task-legacy.ts @@ -0,0 +1,195 @@ +/** Doctor-only reader for the retired YAML-like heartbeat `tasks:` syntax. */ + +export type LegacyHeartbeatTask = { + name: string; + interval: string; + prompt: string; +}; + +type LegacyHeartbeatTaskDocument = { + hasTasksBlock: boolean; + taskEntryCount: number; + tasks: LegacyHeartbeatTask[]; + strippedContent: string; +}; + +type HeartbeatLine = { + htmlCommentSource?: string; + raw: string; + source: string; + visible: string; +}; + +type LegacyHeartbeatTaskBuilder = Partial; + +function splitHeartbeatLines(content: string): Array<{ raw: string; source: string }> { + const lines: Array<{ raw: string; source: string }> = []; + const matcher = /[^\r\n]*(?:\r\n|\n|\r|$)/g; + for (const match of content.matchAll(matcher)) { + const source = match[0]; + if (!source) { + continue; + } + const raw = source.replace(/(?:\r\n|\n|\r)$/, ""); + lines.push({ raw, source }); + } + return lines; +} + +function scanHeartbeatLine(raw: string, state: { inHtmlComment: boolean }) { + let cursor = 0; + let hasHtmlComment = state.inHtmlComment; + let htmlCommentRaw = ""; + let visible = ""; + while (cursor < raw.length) { + if (state.inHtmlComment) { + const commentEnd = raw.indexOf("-->", cursor); + if (commentEnd === -1) { + htmlCommentRaw += raw.slice(cursor); + return { hasHtmlComment, htmlCommentRaw, visible }; + } + htmlCommentRaw += raw.slice(cursor, commentEnd + 3); + state.inHtmlComment = false; + cursor = commentEnd + 3; + continue; + } + + const commentStart = raw.indexOf("