From 4e9ae9fbff9177e107136431dceff36b6ffde876 Mon Sep 17 00:00:00 2001 From: Peter Steinberger Date: Wed, 22 Jul 2026 14:03:29 -0700 Subject: [PATCH] feat(cron): system-owned heartbeat monitor jobs replace the dedicated interval scheduler (#112585) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit * feat(cron): system-owned heartbeat monitor jobs replace the interval scheduler - new internal cron payload kind {kind:"heartbeat"}: execution pokes requestHeartbeat({source:"interval"}); reported in the protocol job schema, not accepted from client create/patch - gateway converges one declaration-keyed monitor job per heartbeat-enabled agent (schedule every+deterministic phase anchor) at startup and on config reload; removes monitors for unconfigured agents - heartbeat runner loses its interval setTimeout machinery; nextDueMs stays as the cooldown gate, event wakes unchanged * test(cron): heartbeat monitor regressions; docs for cron-owned cadence - converge/prune/failure-containment tests for heartbeat monitor jobs - heartbeat payload run fires an interval wake, no system event - scheduler tests converted from timer self-fire to wake-queue pokes; timer-mechanics-only tests deleted with the timer - persisted-shape accepts the heartbeat payload kind - docs: heartbeat cadence ownership + system payload kind * fix(cron): heartbeat monitor review round 1 - targeted cron-monitor interval ticks use the full per-agent path so due-commitment sessions still deliver - cron-disabled gateways keep a local fallback interval timer (shipped cron.enabled=false contract; removed when heartbeat config folds into cron in #110950) - heartbeat job reconciliations serialize with latest-wins epochs and a bounded 30s retry after a failed convergence pass * fix(cron): chain clamped fallback heartbeat timers past the setTimeout cap * fix(cron): heartbeat monitor review round 3 - targeted monitor redirect skips wakes carrying heartbeat overrides and surfaces the per-agent terminal skip reason instead of not-due - cron-disabled fallback timer re-arms with a 1s floor after each firing so a dropped wake cannot end the chain - heartbeat payloads are system-owned at the service boundary: add requires the gateway opt-in, patches to the kind are rejected * fix(cron): heartbeat monitor review round 4 — full ownership enforcement - prune only jobs proven to be monitors (prefix AND heartbeat payload) - existing monitors reject every update patch; declarative upserts on the monitor key require the gateway opt-in even with a different payload * fix(cron): complete heartbeat monitor ownership boundary - converge scopes declarative matching to real monitors so a colliding user job with the same key is never adopted or overwritten - monitor removal requires the gateway systemOwned opt-in; ad-hoc API/CLI deletion is rejected, reconciliation cleanup still prunes * docs(cron): record intentional enrollment-snapshot semantics for monitor ticks * fix(cron): repair heartbeat monitor CI gates --- docs/automation/cron-jobs.md | 2 + docs/gateway/heartbeat.md | 2 + packages/gateway-protocol/src/schema/cron.ts | 11 +- src/cron/normalize.ts | 2 +- src/cron/persisted-shape.ts | 3 +- src/cron/service-contract.ts | 2 +- src/cron/service.heartbeat-payload.test.ts | 79 +++ src/cron/service.ts | 4 +- src/cron/service/jobs.ts | 3 +- src/cron/service/ops.ts | 41 +- src/cron/service/payload-merge.ts | 10 + src/cron/service/state.ts | 2 + src/cron/service/timer.ts | 13 + src/cron/store/payload-codec.ts | 17 + src/cron/types.ts | 10 +- .../server-cron-heartbeat-jobs.test.ts | 142 ++++++ src/gateway/server-cron-heartbeat-jobs.ts | 97 ++++ src/gateway/server-cron.test.ts | 4 + src/gateway/server-cron.ts | 49 ++ src/gateway/server-reload-handlers.ts | 5 + ...t-runner.active-hours-schedule.e2e.test.ts | 162 +++--- .../heartbeat-runner.commitments.test.ts | 52 ++ src/infra/heartbeat-runner.scheduler.test.ts | 208 ++++---- .../heartbeat-runner.timeout-warning.test.ts | 67 --- src/infra/heartbeat-runner.ts | 466 +++++++++--------- .../runtime/heartbeat-active-hours-runtime.ts | 55 ++- ui/src/lib/cron/index.ts | 9 +- ui/src/lib/presenter.ts | 3 + 28 files changed, 1006 insertions(+), 514 deletions(-) create mode 100644 src/cron/service.heartbeat-payload.test.ts create mode 100644 src/gateway/server-cron-heartbeat-jobs.test.ts create mode 100644 src/gateway/server-cron-heartbeat-jobs.ts delete mode 100644 src/infra/heartbeat-runner.timeout-warning.test.ts diff --git a/docs/automation/cron-jobs.md b/docs/automation/cron-jobs.md index 7ecc079ea41f..2e84acd74c5a 100644 --- a/docs/automation/cron-jobs.md +++ b/docs/automation/cron-jobs.md @@ -166,6 +166,8 @@ Every job carries exactly one payload kind, chosen by flag: | Command | `--command ` or `--command-argv ` | A shell/process on the Gateway host, no model call | | Script | `--script ` | A headless code-mode script using the owning agent's tools | +One additional payload kind, `heartbeat`, is system-owned: the gateway converges one heartbeat monitor job per heartbeat-enabled agent (see [Heartbeat](/gateway/heartbeat)). It appears in `cron list --all` but cannot be created or edited through the CLI or API — its cadence follows `agents.*.heartbeat` config. + ### Agent-turn options diff --git a/docs/gateway/heartbeat.md b/docs/gateway/heartbeat.md index 13725e75fde0..1f49de07b94d 100644 --- a/docs/gateway/heartbeat.md +++ b/docs/gateway/heartbeat.md @@ -15,6 +15,8 @@ Heartbeat runs **periodic agent turns** in the main session so the model can sur Heartbeat is a scheduled main-session turn - it does **not** create [background task](/automation/tasks) records. Task records are for detached work (ACP runs, subagents, isolated cron jobs). +Under the hood, heartbeat cadence is owned by the cron scheduler: the gateway maintains one system-owned cron job per heartbeat-enabled agent (visible in `openclaw cron list --all` as `Heartbeat (agent-id)`). Each tick requests a heartbeat wake; the heartbeat runner still applies its own cooldown, active-hours, and busy guards, so a tick outside the configured window is skipped, not delivered. These monitor jobs are converged from your heartbeat config at startup and on config reload — edit `agents.*.heartbeat`, not the cron job. + Troubleshooting: [Scheduled Tasks](/automation/cron-jobs#troubleshooting) ## Quick start (beginner) diff --git a/packages/gateway-protocol/src/schema/cron.ts b/packages/gateway-protocol/src/schema/cron.ts index ec6306b6cd0d..ff4e0c6d705f 100644 --- a/packages/gateway-protocol/src/schema/cron.ts +++ b/packages/gateway-protocol/src/schema/cron.ts @@ -286,6 +286,15 @@ const CronPayloadSchema = Type.Union([ }), ]); +/** + * Reported payloads add the system-owned heartbeat monitor kind; it is + * gateway-converged only, so create/patch schemas intentionally omit it. + */ +const CronReportedPayloadSchema = Type.Union([ + ...CronPayloadSchema.anyOf, + closedObject({ kind: Type.Literal("heartbeat") }), +]); + /** Partial cron payload for job updates. */ const CronPayloadPatchSchema = Type.Union([ closedObject({ @@ -520,7 +529,7 @@ export const CronJobSchema = closedObject({ trigger: Type.Optional(CronTriggerSchema), sessionTarget: CronSessionTargetSchema, wakeMode: CronWakeModeSchema, - payload: CronPayloadSchema, + payload: CronReportedPayloadSchema, delivery: Type.Optional(CronDeliverySchema), failureAlert: Type.Optional(Type.Union([Type.Literal(false), CronFailureAlertSchema])), state: CronJobStateSchema, diff --git a/src/cron/normalize.ts b/src/cron/normalize.ts index bd6aa52ed327..43243ad15562 100644 --- a/src/cron/normalize.ts +++ b/src/cron/normalize.ts @@ -488,7 +488,7 @@ export function normalizeCronJobInput( const kind = typeof next.payload.kind === "string" ? next.payload.kind : ""; // Keep create-time defaults explicit: system events join main, while agent // turns isolate by default to avoid unbounded token accumulation. - if (kind === "systemEvent") { + if (kind === "systemEvent" || kind === "heartbeat") { next.sessionTarget = "main"; } else if (kind === "agentTurn" || kind === "command" || kind === "script") { next.sessionTarget = "isolated"; diff --git a/src/cron/persisted-shape.ts b/src/cron/persisted-shape.ts index 5804fe59304d..ba015d6fff92 100644 --- a/src/cron/persisted-shape.ts +++ b/src/cron/persisted-shape.ts @@ -118,7 +118,8 @@ export function getInvalidPersistedCronJobReason( payloadKind !== "systemEvent" && payloadKind !== "agentTurn" && payloadKind !== "command" && - payloadKind !== "script" + payloadKind !== "script" && + payloadKind !== "heartbeat" ) { return "invalid-payload"; } diff --git a/src/cron/service-contract.ts b/src/cron/service-contract.ts index 71248bd26d3e..db2d356c0354 100644 --- a/src/cron/service-contract.ts +++ b/src/cron/service-contract.ts @@ -47,7 +47,7 @@ export interface CronServiceContract { patch: CronUpdateInput, precondition: CronUpdatePrecondition, ): Promise; - remove(id: string): Promise; + remove(id: string, opts?: { systemOwned?: boolean }): Promise; run(id: string, mode?: CronRunMode, opts?: CronServiceRunOptions): Promise; enqueueRun(id: string, mode?: CronRunMode): Promise; getJob(id: string): CronJob | undefined; diff --git a/src/cron/service.heartbeat-payload.test.ts b/src/cron/service.heartbeat-payload.test.ts new file mode 100644 index 000000000000..cb2fd9d9054b --- /dev/null +++ b/src/cron/service.heartbeat-payload.test.ts @@ -0,0 +1,79 @@ +// The system heartbeat monitor payload replaces the dedicated interval +// scheduler: firing it must only poke the heartbeat wake queue. +import { describe, expect, it } from "vitest"; +import { + createCronStoreHarness, + createNoopLogger, + createStartedCronServiceWithFinishedBarrier, + installCronTestHooks, +} from "./service.test-harness.js"; + +const noopLogger = createNoopLogger(); +const { makeStorePath } = createCronStoreHarness(); +installCronTestHooks({ logger: noopLogger }); + +describe("heartbeat payload execution", () => { + it("fires as an interval heartbeat wake without enqueuing a system event", async () => { + const { storePath, cleanup } = await makeStorePath(); + const { cron, enqueueSystemEvent, requestHeartbeat } = + createStartedCronServiceWithFinishedBarrier({ storePath, logger: noopLogger }); + try { + await cron.start(); + const added = await cron.add( + { + declarationKey: "heartbeat:main", + name: "heartbeat-main", + agentId: "main", + enabled: true, + schedule: { kind: "every", everyMs: 60_000 }, + payload: { kind: "heartbeat" }, + sessionTarget: "main", + wakeMode: "next-heartbeat", + }, + { enabledExplicit: true, systemOwned: true }, + ); + const job = "job" in added ? added.job : added; + // System ownership boundary: no caller may create or patch to the + // heartbeat payload without the gateway's opt-in. + await expect( + cron.add({ + name: "rogue", + enabled: true, + schedule: { kind: "every", everyMs: 60_000 }, + payload: { kind: "heartbeat" }, + sessionTarget: "main", + wakeMode: "next-heartbeat", + }), + ).rejects.toThrow(/system-owned/); + await expect(cron.update(job.id, { payload: { kind: "heartbeat" } })).rejects.toThrow( + /system-owned/, + ); + // Existing monitors reject every patch, not just payload-kind edits. + await expect(cron.update(job.id, { enabled: false })).rejects.toThrow(/system-owned/); + // Ad-hoc deletion is rejected too; only reconciliation cleanup removes. + await expect(cron.remove(job.id)).rejects.toThrow(/system-owned/); + // A declarative upsert on the monitor's key cannot repurpose it either. + await expect( + cron.add({ + declarationKey: "heartbeat:main", + name: "rogue-upsert", + enabled: true, + schedule: { kind: "every", everyMs: 60_000 }, + payload: { kind: "systemEvent", text: "hijack" }, + sessionTarget: "main", + wakeMode: "next-heartbeat", + }), + ).rejects.toThrow(/system-owned/); + const result = await cron.run(job.id, "force"); + expect(result.ok).toBe(true); + expect(requestHeartbeat).toHaveBeenCalledWith( + expect.objectContaining({ source: "interval", intent: "scheduled", agentId: "main" }), + ); + // The monitor never fabricates a system event; the wake is the whole run. + expect(enqueueSystemEvent).not.toHaveBeenCalled(); + } finally { + cron.stop(); + await cleanup(); + } + }); +}); diff --git a/src/cron/service.ts b/src/cron/service.ts index f15716da9a67..2e34858564f3 100644 --- a/src/cron/service.ts +++ b/src/cron/service.ts @@ -117,8 +117,8 @@ export class CronService implements CronServiceContract { return await ops.updateWithPrecondition(this.state, id, patch, precondition); } - async remove(id: string) { - return await ops.remove(this.state, id); + async remove(id: string, opts?: { systemOwned?: boolean }) { + return await ops.remove(this.state, id, opts); } async removeAgentJobsTransactional(agentId: string, commit: () => Promise): Promise { diff --git a/src/cron/service/jobs.ts b/src/cron/service/jobs.ts index 5f23eeac46bb..8a3dd5563021 100644 --- a/src/cron/service/jobs.ts +++ b/src/cron/service/jobs.ts @@ -334,7 +334,8 @@ export function assertSupportedJobSpec( if ( job.sessionTarget === "main" && job.payload.kind !== "systemEvent" && - job.payload.kind !== "script" + job.payload.kind !== "script" && + job.payload.kind !== "heartbeat" ) { throw new Error('main cron jobs require payload.kind="systemEvent" or "script"'); } diff --git a/src/cron/service/ops.ts b/src/cron/service/ops.ts index debcff3ac04d..849573c46553 100644 --- a/src/cron/service/ops.ts +++ b/src/cron/service/ops.ts @@ -728,6 +728,12 @@ function declarativeFields(job: CronJob, includeEnabled: boolean) { export async function add(state: CronServiceState, input: CronJobCreate, opts?: CronAddOptions) { return await locked(state, async () => { warnIfDisabled(state, "add"); + // Heartbeat monitors are gateway-converged system jobs; without this + // boundary any internal caller could upsert the declaration key and + // hijack the monitor despite the transport schemas excluding the kind. + if (input.payload?.kind === "heartbeat" && opts?.systemOwned !== true) { + throw new Error("heartbeat payloads are system-owned; jobs cannot be created with them"); + } await ensureLoaded(state, { skipRecompute: true }); const agentId = resolveEffectiveJobAgentId(input, resolveCurrentDefaultAgentId(state)); if (state.deps.isAgentAvailable?.(agentId) === false) { @@ -753,6 +759,13 @@ export async function add(state: CronServiceState, input: CronJobCreate, opts?: const existing = matches[0]; if (existing) { + // A declarative upsert may not repurpose an existing heartbeat monitor + // with a different payload; only the gateway's own convergence touches it. + if (existing.payload.kind === "heartbeat" && opts?.systemOwned !== true) { + throw new Error( + "heartbeat monitor jobs are system-owned; edit agents.*.heartbeat config instead", + ); + } const now = state.deps.nowMs(); const nextJob = structuredClone(existing); applyDeclarativeJobSpec(nextJob, normalizedInput, { @@ -832,9 +845,23 @@ async function updateLoadedJob(params: { }) { const { state, id, patch, precondition } = params; warnIfDisabled(state, "update"); + // Mirrors the add-time boundary: no caller may patch a job into (or edit) + // the system-owned heartbeat payload; the gateway converges via add only. + if (patch.payload?.kind === "heartbeat") { + throw new Error("heartbeat payloads are system-owned; jobs cannot be patched to them"); + } await ensureLoaded(state, { skipRecompute: true }); const snapshot = snapshotStoreForRollback(state); const job = findJobOrThrow(state, id); + // Existing monitors are config-driven: any patch (disable, reschedule, + // repurpose) would silently diverge from agents.*.heartbeat until the next + // reconcile, so updates are rejected outright. Removal stays allowed — a + // removed monitor self-heals at the next convergence. + if (job.payload.kind === "heartbeat") { + throw new Error( + "heartbeat monitor jobs are system-owned; edit agents.*.heartbeat config instead", + ); + } const now = state.deps.nowMs(); await precondition?.(structuredClone(job), now); const nextJob = structuredClone(job); @@ -880,7 +907,11 @@ export async function updateWithPrecondition( } /** Removes a cron job by id and re-arms the timer when the in-memory store changes. */ -export async function remove(state: CronServiceState, id: string) { +export async function remove( + state: CronServiceState, + id: string, + opts?: { systemOwned?: boolean }, +) { return await locked(state, async () => { warnIfDisabled(state, "remove"); await ensureLoaded(state, { skipRecompute: true }); @@ -890,6 +921,14 @@ export async function remove(state: CronServiceState, id: string) { } const snapshot = snapshotStoreForRollback(state); const removedJob = state.store.jobs.find((j) => j.id === id); + // Config is the monitor's source of truth: ad-hoc deletion would disable + // heartbeats until an unrelated reload, so only gateway reconciliation + // (stale-monitor cleanup) may remove one. + if (removedJob?.payload.kind === "heartbeat" && opts?.systemOwned !== true) { + throw new Error( + "heartbeat monitor jobs are system-owned; edit agents.*.heartbeat config instead", + ); + } state.store.jobs = state.store.jobs.filter((j) => j.id !== id); const removed = (state.store.jobs.length ?? 0) !== before; diff --git a/src/cron/service/payload-merge.ts b/src/cron/service/payload-merge.ts index 235a7cb8b534..4db7274789df 100644 --- a/src/cron/service/payload-merge.ts +++ b/src/cron/service/payload-merge.ts @@ -119,6 +119,12 @@ export function mergeCronPayload(existing: CronPayload, patch: CronPayloadPatch) return next; } + if (patch.kind === "heartbeat") { + // Unreachable through the service (system-owned boundary rejects it + // first); keep the merge total for the type union. + return { kind: "heartbeat" }; + } + if (existing.kind !== "agentTurn") { return buildPayloadFromPatch(patch); } @@ -200,6 +206,10 @@ function buildPayloadFromPatch(patch: CronPayloadPatch): CronPayload { return next; } + if (patch.kind === "heartbeat") { + return { kind: "heartbeat" }; + } + if (typeof patch.message !== "string" || patch.message.length === 0) { throw new Error('cron.update payload.kind="agentTurn" requires message'); } diff --git a/src/cron/service/state.ts b/src/cron/service/state.ts index 6142f3a27dd0..c3023958a772 100644 --- a/src/cron/service/state.ts +++ b/src/cron/service/state.ts @@ -358,6 +358,8 @@ export type CronAddInput = CronJobCreate; export type CronAddOptions = { matchesExisting?: (job: CronJob) => boolean; enabledExplicit?: boolean; + /** Gateway-owned system payloads (heartbeat monitors) require this opt-in. */ + systemOwned?: boolean; }; /** Normalized patch input accepted by cron service updates. */ export type CronUpdateInput = CronJobPatch; diff --git a/src/cron/service/timer.ts b/src/cron/service/timer.ts index a601f605a2e3..f1eacc382d51 100644 --- a/src/cron/service/timer.ts +++ b/src/cron/service/timer.ts @@ -2600,6 +2600,19 @@ async function executeJobCore( payload: appendCronPayloadText(effectiveJob.payload, options.streamBatch), }; } + if (effectiveJob.payload.kind === "heartbeat") { + // The monitor only pokes the wake queue: coalescing, busy-retry, and the + // quiet-hours guard all live in the heartbeat runner, exactly as they did + // for the dedicated interval timer this job replaces. + state.deps.requestHeartbeat({ + source: "interval", + intent: "scheduled", + reason: "interval", + agentId: effectiveJob.agentId, + }); + const result = { status: "ok" as const, summary: "heartbeat wake requested" }; + return triggerEval ? { ...result, triggerEval } : result; + } if (effectiveJob.sessionTarget === "main") { const result = await executeMainSessionCronJob( state, diff --git a/src/cron/store/payload-codec.ts b/src/cron/store/payload-codec.ts index 9a8eb8d55989..5c158c5fbd94 100644 --- a/src/cron/store/payload-codec.ts +++ b/src/cron/store/payload-codec.ts @@ -142,6 +142,20 @@ export function bindPayloadColumns( ...bindPayloadToolAllowColumns(payload), }; } + if (payload.kind === "heartbeat") { + return { + payload_kind: "heartbeat", + payload_message: null, + payload_model: null, + payload_fallbacks_json: null, + payload_thinking: null, + payload_timeout_seconds: null, + payload_allow_unsafe_external_content: null, + payload_external_content_source_json: null, + payload_light_context: null, + ...bindPayloadToolAllowColumns(payload), + }; + } if (payload.kind === "command") { const { timeoutSeconds: _timeoutSeconds, @@ -251,6 +265,9 @@ export function payloadFromRow(row: CronJobRow): CronPayload | null { ...payloadToolAllowFromRow(row), }; } + if (row.payload_kind === "heartbeat") { + return { kind: "heartbeat" }; + } if (row.payload_kind === "script") { const script = parseScriptPayloadMessage(row.payload_message); if (!script) { diff --git a/src/cron/types.ts b/src/cron/types.ts index 5f3d1ac02f9c..14368e5e1d18 100644 --- a/src/cron/types.ts +++ b/src/cron/types.ts @@ -269,14 +269,20 @@ export type CronPayload = | ({ kind: "systemEvent"; text: string } & CronPayloadToolAllow) | (CronAgentTurnPayload & CronPayloadToolAllow) | (CronCommandPayload & CronPayloadToolAllow) - | (CronScriptPayload & CronPayloadToolAllow); + | (CronScriptPayload & CronPayloadToolAllow) + // System-owned heartbeat monitor: execution requests an interval heartbeat + // wake. Gateway-converged only; not accepted from client create/patch APIs. + | ({ kind: "heartbeat" } & CronPayloadToolAllow); /** Partial payload update shape used by cron patch/edit flows. */ export type CronPayloadPatch = | ({ kind: "systemEvent"; text?: string } & CronPayloadToolAllowPatch) | (CronAgentTurnPayloadPatch & CronPayloadToolAllowPatch) | (CronCommandPayloadPatch & CronPayloadToolAllowPatch) - | (CronScriptPayloadPatch & CronPayloadToolAllowPatch); + | (CronScriptPayloadPatch & CronPayloadToolAllowPatch) + // Representable so the service can reject it with a typed boundary error; + // transports and tools never accept it. + | ({ kind: "heartbeat" } & CronPayloadToolAllowPatch); type CronPayloadToolAllow = { /** Restricts agentTurn execution, or the trigger runtime for other payload kinds. */ diff --git a/src/gateway/server-cron-heartbeat-jobs.test.ts b/src/gateway/server-cron-heartbeat-jobs.test.ts new file mode 100644 index 000000000000..1d9ffacbdc94 --- /dev/null +++ b/src/gateway/server-cron-heartbeat-jobs.test.ts @@ -0,0 +1,142 @@ +import { describe, expect, it, vi } from "vitest"; +import type { OpenClawConfig } from "../config/types.openclaw.js"; +import type { CronJob } from "../cron/types.js"; +import { reconcileHeartbeatMonitorJobs } from "./server-cron-heartbeat-jobs.js"; + +const logger = { warn: vi.fn() }; + +type AddOptions = { matchesExisting?: (job: CronJob) => boolean }; + +function monitorJob(agentId: string, id = `job-${agentId}`): CronJob { + return { + id, + declarationKey: `heartbeat:${agentId}`, + name: `heartbeat-${agentId}`, + enabled: true, + createdAtMs: 1, + updatedAtMs: 1, + agentId, + schedule: { kind: "every", everyMs: 60_000 }, + sessionTarget: "main", + wakeMode: "next-heartbeat", + payload: { kind: "heartbeat" }, + state: {}, + } as CronJob; +} + +describe("reconcileHeartbeatMonitorJobs", () => { + it("converges one monitor per heartbeat agent and prunes unconfigured ones", async () => { + const add = vi.fn(async (input: { declarationKey?: string }, _options?: AddOptions) => ({ + job: input, + })); + const remove = vi.fn(async () => ({ ok: true })); + const list = vi.fn(async () => [ + monitorJob("stale-agent"), + { + ...monitorJob("main"), + id: "user-job", + declarationKey: undefined, + payload: { kind: "systemEvent", text: "user job" }, + } as CronJob, + // Prefix collision without a heartbeat payload must never be pruned. + { + ...monitorJob("collider"), + id: "prefix-collider", + payload: { kind: "systemEvent", text: "not a monitor" }, + } as CronJob, + ]); + const cfg = { + agents: { + defaults: { heartbeat: { every: "15m" } }, + list: [{ id: "main" }, { id: "ops" }], + }, + } as OpenClawConfig; + + await reconcileHeartbeatMonitorJobs({ + cron: { add, list, remove } as never, + cfg, + logger, + }); + + const declarationKeys = add.mock.calls.map( + ([input]) => (input as { declarationKey?: string }).declarationKey, + ); + expect(declarationKeys.toSorted((a, b) => (a ?? "").localeCompare(b ?? ""))).toEqual([ + "heartbeat:main", + "heartbeat:ops", + ]); + for (const [input] of add.mock.calls) { + const spec = input as { + payload: { kind: string }; + schedule: { kind: string; everyMs: number; anchorMs?: number }; + sessionTarget: string; + enabled: boolean; + }; + expect(spec.payload).toEqual({ kind: "heartbeat" }); + expect(spec.schedule.kind).toBe("every"); + expect(spec.schedule.everyMs).toBe(15 * 60_000); + // Deterministic phase anchor spreads multi-agent beats inside the interval. + expect(spec.schedule.anchorMs).toBeGreaterThanOrEqual(0); + expect(spec.schedule.anchorMs).toBeLessThan(15 * 60_000); + expect(spec.sessionTarget).toBe("main"); + expect(spec.enabled).toBe(true); + } + // Only the stale monitor is pruned; user jobs without the prefix survive. + expect(remove).toHaveBeenCalledTimes(1); + expect(remove).toHaveBeenCalledWith("job-stale-agent", { systemOwned: true }); + // Declarative matching is scoped to real monitors so a colliding user job + // is never adopted by the system upsert. + for (const call of add.mock.calls) { + const opts = call[1]; + if (!opts) { + throw new Error("expected system-owned add options"); + } + expect(opts.matchesExisting?.(monitorJob("x"))).toBe(true); + expect( + opts.matchesExisting?.({ + ...monitorJob("x"), + payload: { kind: "systemEvent", text: "user" }, + } as CronJob), + ).toBe(false); + } + }); + + it("removes all monitors when heartbeats are disabled", async () => { + const add = vi.fn(async () => ({})); + const remove = vi.fn(async () => ({ ok: true })); + const list = vi.fn(async () => [monitorJob("main")]); + const cfg = { + agents: { defaults: { heartbeat: { every: "0m" } } }, + } as OpenClawConfig; + + await reconcileHeartbeatMonitorJobs({ + cron: { add, list, remove } as never, + cfg, + logger, + }); + + expect(add).not.toHaveBeenCalled(); + expect(remove).toHaveBeenCalledWith("job-main", { systemOwned: true }); + }); + + it("keeps converging other agents when one convergence fails", async () => { + const add = vi.fn(async () => ({})).mockRejectedValueOnce(new Error("store busy")); + const remove = vi.fn(async () => ({ ok: true })); + const list = vi.fn(async () => []); + const cfg = { + agents: { + defaults: { heartbeat: { every: "30m" } }, + list: [{ id: "a" }, { id: "b" }], + }, + } as OpenClawConfig; + + await reconcileHeartbeatMonitorJobs({ + cron: { add, list, remove } as never, + cfg, + logger, + }); + + expect(add).toHaveBeenCalledTimes(2); + expect(logger.warn).toHaveBeenCalled(); + }); +}); diff --git a/src/gateway/server-cron-heartbeat-jobs.ts b/src/gateway/server-cron-heartbeat-jobs.ts new file mode 100644 index 000000000000..57d670b69844 --- /dev/null +++ b/src/gateway/server-cron-heartbeat-jobs.ts @@ -0,0 +1,97 @@ +// Converges the system-owned heartbeat monitor jobs that replaced the +// dedicated interval scheduler: one declaration-keyed cron job per +// heartbeat-enabled agent, reconverged at startup and config reload. +import type { OpenClawConfig } from "../config/types.openclaw.js"; +import { + resolveHeartbeatAgents, + resolveHeartbeatSchedulerSeed, +} from "../infra/heartbeat-runner.js"; +import { resolveHeartbeatPhaseMs } from "../infra/heartbeat-schedule.js"; +import { resolveHeartbeatIntervalMs } from "../infra/heartbeat-summary.js"; +import type { GatewayCronServiceContract } from "./server-cron-contract.js"; + +const HEARTBEAT_DECLARATION_PREFIX = "heartbeat:"; + +function heartbeatMonitorDeclarationKey(agentId: string): string { + return `${HEARTBEAT_DECLARATION_PREFIX}${agentId}`; +} + +type HeartbeatJobCron = Pick; + +/** + * Converges one system-owned heartbeat monitor job per heartbeat-enabled + * agent and removes monitors for agents no longer configured. Config is the + * single source of truth: interval changes update the schedule in place and + * the deterministic per-agent phase keeps multi-agent beats spread out. + */ +export async function reconcileHeartbeatMonitorJobs(params: { + cron: HeartbeatJobCron; + cfg: OpenClawConfig; + logger: { warn: (obj: unknown, msg?: string) => void }; +}): Promise<{ ok: boolean }> { + let ok = true; + const schedulerSeed = resolveHeartbeatSchedulerSeed(); + const desired = new Set(); + for (const agent of resolveHeartbeatAgents(params.cfg)) { + const intervalMs = resolveHeartbeatIntervalMs(params.cfg, undefined, agent.heartbeat); + if (!intervalMs) { + continue; + } + desired.add(agent.agentId); + try { + await params.cron.add( + { + declarationKey: heartbeatMonitorDeclarationKey(agent.agentId), + displayName: `Heartbeat (${agent.agentId})`, + name: `heartbeat-${agent.agentId}`, + agentId: agent.agentId, + enabled: true, + schedule: { + kind: "every", + everyMs: intervalMs, + anchorMs: resolveHeartbeatPhaseMs({ + schedulerSeed, + agentId: agent.agentId, + intervalMs, + }), + }, + payload: { kind: "heartbeat" }, + sessionTarget: "main", + wakeMode: "next-heartbeat", + }, + { + enabledExplicit: true, + systemOwned: true, + // Scope declarative matching to real monitors: a pre-existing user + // job that happens to hold this key is left untouched. + matchesExisting: (job) => job.payload.kind === "heartbeat", + }, + ); + } catch (error) { + ok = false; + params.logger.warn( + { agentId: agent.agentId, err: String(error) }, + "cron-heartbeat: monitor convergence failed", + ); + } + } + try { + const jobs = await params.cron.list({ includeDisabled: true }); + for (const job of jobs) { + const key = job.declarationKey; + // Prune only proven monitors: prefix alone must never delete an + // unrelated declaration-keyed job that happens to share the namespace. + if (!key?.startsWith(HEARTBEAT_DECLARATION_PREFIX) || job.payload.kind !== "heartbeat") { + continue; + } + if (desired.has(key.slice(HEARTBEAT_DECLARATION_PREFIX.length))) { + continue; + } + await params.cron.remove(job.id, { systemOwned: true }); + } + } catch (error) { + ok = false; + params.logger.warn({ err: String(error) }, "cron-heartbeat: stale monitor cleanup failed"); + } + return { ok }; +} diff --git a/src/gateway/server-cron.test.ts b/src/gateway/server-cron.test.ts index a386beca0923..b00e5c12b548 100644 --- a/src/gateway/server-cron.test.ts +++ b/src/gateway/server-cron.test.ts @@ -145,6 +145,10 @@ vi.mock("../infra/heartbeat-wake.js", async () => { vi.mock("../infra/heartbeat-runner.js", () => ({ runHeartbeatOnce, + // Heartbeat monitor convergence enumerates agents at cron start; keep it + // inert so these tests exercise cron wiring, not heartbeat enrollment. + resolveHeartbeatAgents: () => [], + resolveHeartbeatSchedulerSeed: () => "test-seed", })); vi.mock("../infra/restart-coordinator.js", async () => { diff --git a/src/gateway/server-cron.ts b/src/gateway/server-cron.ts index 698e6c95c5cb..bfca4eaa5d44 100644 --- a/src/gateway/server-cron.ts +++ b/src/gateway/server-cron.ts @@ -73,6 +73,7 @@ import { resolveStreamStopReason, } from "./cron-stream-watchers.js"; import type { GatewayCronServiceContract } from "./server-cron-contract.js"; +import { reconcileHeartbeatMonitorJobs } from "./server-cron-heartbeat-jobs.js"; import { dispatchGatewayCronFinishedNotifications, sendGatewayCronFailureAlert, @@ -94,6 +95,7 @@ export type GatewayCronState = { stopExitWatchers?: () => void; reconcileStreamWatchers?: () => Promise; stopStreamWatchers?: () => Promise; + reconcileHeartbeatJobs?: (cfg?: OpenClawConfig) => Promise; }; function classifyCronScriptFailure(code: CronTriggerFailureCode): CronRunErrorClassification { @@ -1273,6 +1275,7 @@ export function buildGatewayCronService(params: { cron.stop = () => { stopCron(); stopExitWatchers(); + stopHeartbeatReconcileRetry(); void stopStreamWatchers().catch((err: unknown) => { cronLogger.warn( { err: formatErrorMessage(err) }, @@ -1286,9 +1289,50 @@ export function buildGatewayCronService(params: { cron.stopAndDrain = async () => { stopCron(); stopExitWatchers(); + stopHeartbeatReconcileRetry(); await stopStreamWatchers(); unregisterSessionAutomationSource(automationSource); }; + // Reconciliations serialize on one tail and only the latest requested epoch + // executes, so an older reload's convergence can never clobber a newer one. + // A failed pass schedules one bounded retry; a newer request supersedes it. + let heartbeatReconcileEpoch = 0; + let heartbeatReconcileTail: Promise = Promise.resolve(); + let heartbeatRetryTimer: NodeJS.Timeout | undefined; + const stopHeartbeatReconcileRetry = () => { + // Also invalidate any in-flight pass so a post-stop retry cannot fire. + heartbeatReconcileEpoch += 1; + if (heartbeatRetryTimer) { + clearTimeout(heartbeatRetryTimer); + heartbeatRetryTimer = undefined; + } + }; + const reconcileHeartbeatJobs = (cfgOverride?: OpenClawConfig): Promise => { + const epoch = ++heartbeatReconcileEpoch; + if (heartbeatRetryTimer) { + clearTimeout(heartbeatRetryTimer); + heartbeatRetryTimer = undefined; + } + const pass = async () => { + if (epoch !== heartbeatReconcileEpoch) { + return; + } + const { ok } = await reconcileHeartbeatMonitorJobs({ + cron, + cfg: cfgOverride ?? getRuntimeConfig(), + logger: cronLogger, + }); + if (!ok && epoch === heartbeatReconcileEpoch) { + heartbeatRetryTimer = setTimeout(() => { + heartbeatRetryTimer = undefined; + void reconcileHeartbeatJobs(cfgOverride); + }, 30_000); + heartbeatRetryTimer.unref?.(); + } + }; + heartbeatReconcileTail = heartbeatReconcileTail.then(pass, pass); + return heartbeatReconcileTail; + }; const startCron = cron.start.bind(cron); cron.start = async () => { const generation = streamWatcherGeneration; @@ -1307,6 +1351,10 @@ export function buildGatewayCronService(params: { if (generation !== streamWatcherGeneration) { return; } + await reconcileHeartbeatJobs(); + if (generation !== streamWatcherGeneration) { + return; + } // Register only once started, under the build-time epoch, so a stale lazy // service resolving after a config reload cannot clobber the replacement. registerSessionAutomationSource(automationSource, automationEpoch); @@ -1328,6 +1376,7 @@ export function buildGatewayCronService(params: { stopExitWatchers, reconcileStreamWatchers, stopStreamWatchers, + reconcileHeartbeatJobs, }; } /* oxlint-disable max-lines -- TODO: split this grandfathered oversized file. */ diff --git a/src/gateway/server-reload-handlers.ts b/src/gateway/server-reload-handlers.ts index 559219325e1b..06ca5011102e 100644 --- a/src/gateway/server-reload-handlers.ts +++ b/src/gateway/server-reload-handlers.ts @@ -639,6 +639,11 @@ export function createGatewayReloadHandlers(params: GatewayReloadHandlerParams) const commit = async () => { if (plan.restartHeartbeat) { nextState.heartbeatRunner.updateConfig(nextConfig); + // Heartbeat cadence lives in system-owned cron monitor jobs; + // reconverge them against the new config in the background. + void nextState.cronState.reconcileHeartbeatJobs?.(nextConfig).catch((error: unknown) => { + params.logReload.warn(`heartbeat monitor reconvergence failed: ${String(error)}`); + }); } // Config, plugin hooks, and prepared stores publish as one generation. Synchronously // retire the prior stores at the commit edge so no request can mix generations. diff --git a/src/infra/heartbeat-runner.active-hours-schedule.e2e.test.ts b/src/infra/heartbeat-runner.active-hours-schedule.e2e.test.ts index d3d2dda29215..78640efafc3c 100644 --- a/src/infra/heartbeat-runner.active-hours-schedule.e2e.test.ts +++ b/src/infra/heartbeat-runner.active-hours-schedule.e2e.test.ts @@ -1,12 +1,15 @@ -// Covers heartbeat scheduling within active hours. +// Covers heartbeat active-hours scheduling (#75487). Interval cadence is +// owned by system cron monitor jobs; these tests poke the wake queue with +// `source: "interval"` and assert the runner's `nextDueMs` seek defers +// quiet-hours pokes and admits in-window ones. import { expectDefined } from "@openclaw/normalization-core"; import { afterEach, describe, expect, it, vi } from "vitest"; import type { OpenClawConfig } from "../config/config.js"; import { startHeartbeatRunner } from "./heartbeat-runner.js"; import { computeNextHeartbeatPhaseDueMs, resolveHeartbeatPhaseMs } from "./heartbeat-schedule.js"; +import { requestHeartbeat } from "./heartbeat-wake.js"; -/** Verifies that the scheduler seeks to in-window phase slots (#75487). */ describe("heartbeat scheduler: activeHours-aware scheduling (#75487)", () => { type RunOnce = Parameters[0]["runOnce"]; const TEST_SCHEDULER_SEED = "heartbeat-ah-schedule-test-seed"; @@ -16,6 +19,18 @@ describe("heartbeat scheduler: activeHours-aware scheduling (#75487)", () => { vi.setSystemTime(new Date(startMs)); } + // Stand-in for a system cron monitor tick: the cron job pokes the wake + // queue; the runner decides via `nextDueMs` whether the agent is due. + async function pokeIntervalWake() { + requestHeartbeat({ + source: "interval", + intent: "scheduled", + reason: "interval", + coalesceMs: 0, + }); + await vi.advanceTimersByTimeAsync(1); + } + function heartbeatConfig(overrides?: { every?: string; activeHours?: { start: string; end: string; timezone?: string }; @@ -51,7 +66,7 @@ describe("heartbeat scheduler: activeHours-aware scheduling (#75487)", () => { vi.restoreAllMocks(); }); - it("skips quiet-hours slots and fires at the first in-window phase slot", async () => { + it("defers quiet-hours pokes and admits the first in-window phase slot", async () => { // 09:00–17:00 UTC, 4h interval. Start at 16:30 — raw due is after 17:00. const startMs = Date.parse("2026-06-15T16:30:00.000Z"); useFakeHeartbeatTime(startMs); @@ -74,12 +89,17 @@ describe("heartbeat scheduler: activeHours-aware scheduling (#75487)", () => { const rawDueMs = resolveDueFromNow(startMs, intervalMs, "main"); - // Advance past the raw due — should NOT fire (quiet hours). + // Poke past the raw due slot — still quiet hours, so nextDueMs was + // seeked into tomorrow's window and the poke must defer. await vi.advanceTimersByTimeAsync(rawDueMs - startMs + 1); + await pokeIntervalWake(); + expect(runSpy).not.toHaveBeenCalled(); - // Advance to end of next day's window — should fire within 09:00–17:00. - const safeEndOfWindow = Date.parse("2026-06-16T17:00:00.000Z"); - await vi.advanceTimersByTimeAsync(safeEndOfWindow - Date.now()); + // Poke inside the next day's window, past the seeked slot (4h spacing + // puts the first in-window slot no later than 13:00) — must fire. + const inWindowMs = Date.parse("2026-06-16T16:00:00.000Z"); + await vi.advanceTimersByTimeAsync(inWindowMs - Date.now()); + await pokeIntervalWake(); expect(runSpy).toHaveBeenCalled(); const firstCallHourUTC = new Date( @@ -109,51 +129,12 @@ describe("heartbeat scheduler: activeHours-aware scheduling (#75487)", () => { const rawDueMs = resolveDueFromNow(startMs, intervalMs, "main"); await vi.advanceTimersByTimeAsync(rawDueMs - startMs + 1); + await pokeIntervalWake(); expect(runSpy).toHaveBeenCalledTimes(1); runner.stop(); }); - it("runs a bounded real scheduler with active-hours enabled", async () => { - const startedAt = Date.now(); - let resolveRun: (() => void) | undefined; - const ran = new Promise((resolve) => { - resolveRun = resolve; - }); - const runSpy: RunOnce = vi.fn().mockImplementation(async () => { - resolveRun?.(); - return { status: "ran", durationMs: 1 }; - }); - const runner = startHeartbeatRunner({ - cfg: heartbeatConfig({ - every: "50ms", - activeHours: { start: "00:00", end: "24:00", timezone: "UTC" }, - }), - runOnce: runSpy, - stableSchedulerSeed: TEST_SCHEDULER_SEED, - }); - let timeout: ReturnType | undefined; - - try { - await Promise.race([ - ran, - new Promise((_, reject) => { - timeout = setTimeout( - () => reject(new Error("real heartbeat scheduler did not fire")), - 2_000, - ); - }), - ]); - expect(runSpy).toHaveBeenCalled(); - expect(Date.now() - startedAt).toBeLessThan(2_000); - } finally { - if (timeout) { - clearTimeout(timeout); - } - runner.stop(); - } - }); - it("seeks forward correctly with a non-UTC timezone (e.g. America/New_York)", async () => { // 09:00–17:00 ET (EDT = UTC-4 in June) → 13:00–21:00 UTC. // Start at 21:30 UTC (17:30 ET = outside window). @@ -175,7 +156,16 @@ describe("heartbeat scheduler: activeHours-aware scheduling (#75487)", () => { stableSchedulerSeed: TEST_SCHEDULER_SEED, }); - await vi.advanceTimersByTimeAsync(48 * 60 * 60_000); + // Quiet-hours poke shortly after start must defer. + await vi.advanceTimersByTimeAsync(60 * 60_000); + await pokeIntervalWake(); + expect(runSpy).not.toHaveBeenCalled(); + + // Poke inside the next ET window (first in-window slot is no later than + // 17:00 UTC given 4h spacing from the 13:00 UTC window start). + const inWindowMs = Date.parse("2026-06-16T17:00:00.000Z"); + await vi.advanceTimersByTimeAsync(inWindowMs - Date.now()); + await pokeIntervalWake(); expect(runSpy).toHaveBeenCalled(); const firstCallHourUTC = new Date( @@ -187,43 +177,6 @@ describe("heartbeat scheduler: activeHours-aware scheduling (#75487)", () => { runner.stop(); }); - it("advances to in-window slot after a quiet-hours skip during interval runs", async () => { - // 09:00–17:00 UTC, 4h interval. Verify ALL fires over 48h stay in-window. - const startMs = Date.parse("2026-06-15T09:00:00.000Z"); - useFakeHeartbeatTime(startMs); - - const callTimes: number[] = []; - const runSpy: RunOnce = vi.fn().mockImplementation(async () => { - callTimes.push(Date.now()); - return { status: "ran", durationMs: 1 }; - }); - - const runner = startHeartbeatRunner({ - cfg: heartbeatConfig({ - every: "4h", - activeHours: { start: "09:00", end: "17:00", timezone: "UTC" }, - }), - runOnce: runSpy, - stableSchedulerSeed: TEST_SCHEDULER_SEED, - }); - - await vi.advanceTimersByTimeAsync(48 * 60 * 60_000); - - expect(callTimes.length).toBeGreaterThan(0); - for (const t of callTimes) { - const hour = new Date(t).getUTCHours(); - expect( - hour, - `fire at ${new Date(t).toISOString()} is outside active window`, - ).toBeGreaterThanOrEqual(9); - expect(hour, `fire at ${new Date(t).toISOString()} is outside active window`).toBeLessThan( - 17, - ); - } - - runner.stop(); - }); - it("does not loop indefinitely when activeHours window is zero-width", async () => { // start === end → never-active; seek falls back, runtime guard skips. const startMs = Date.parse("2026-06-15T10:00:00.000Z"); @@ -240,15 +193,24 @@ describe("heartbeat scheduler: activeHours-aware scheduling (#75487)", () => { stableSchedulerSeed: TEST_SCHEDULER_SEED, }); + // Past any 30m slot: the poke reaches runOnce (the runtime guard owns + // the quiet-hours skip when seek cannot find an active slot). await vi.advanceTimersByTimeAsync(2 * 60 * 60_000); + await pokeIntervalWake(); + expect(runSpy).toHaveBeenCalledTimes(1); + + // Non-retryable skip advanced the cadence — an immediate second poke + // must defer instead of hot-looping runOnce. + await vi.advanceTimersByTimeAsync(1_000); + await pokeIntervalWake(); + expect(runSpy).toHaveBeenCalledTimes(1); - expect(runSpy).toHaveBeenCalled(); runner.stop(); }); it("recomputes schedule when activeHours config changes via hot reload", async () => { // Narrow window pushes nextDueMs to tomorrow; widening via updateConfig - // must recompute from `now` so the timer fires today. + // must recompute from `now` so a poke today is admitted. const startMs = Date.parse("2026-06-15T14:00:00.000Z"); useFakeHeartbeatTime(startMs); @@ -268,6 +230,7 @@ describe("heartbeat scheduler: activeHours-aware scheduling (#75487)", () => { }); await vi.advanceTimersByTimeAsync(60 * 60_000); + await pokeIntervalWake(); expect(runSpy).not.toHaveBeenCalled(); // Widen window — scheduler must recompute, not keep stale tomorrow slot. @@ -278,13 +241,11 @@ describe("heartbeat scheduler: activeHours-aware scheduling (#75487)", () => { }), ); + // The recomputed slot lands no later than 19:00 today (4h spacing from + // 15:00), so a poke this evening must fire today — not tomorrow. await vi.advanceTimersByTimeAsync(8 * 60 * 60_000); + await pokeIntervalWake(); expect(runSpy).toHaveBeenCalled(); - const firstCallHour = new Date( - expectDefined(callTimes[0], "callTimes[0] test invariant"), - ).getUTCHours(); - expect(firstCallHour).toBeGreaterThanOrEqual(8); - expect(firstCallHour).toBeLessThan(20); expect(new Date(expectDefined(callTimes[0], "callTimes[0] test invariant")).getUTCDate()).toBe( 15, ); // today, not tomorrow @@ -314,6 +275,7 @@ describe("heartbeat scheduler: activeHours-aware scheduling (#75487)", () => { }); await vi.advanceTimersByTimeAsync(60 * 60_000); + await pokeIntervalWake(); expect(runSpy).not.toHaveBeenCalled(); runner.updateConfig( @@ -324,12 +286,15 @@ describe("heartbeat scheduler: activeHours-aware scheduling (#75487)", () => { }), ); + // With UTC the seed's slot falls inside today's 16:00–17:00 window; a + // poke at window end must be admitted today. Under the stale ET slot + // (tomorrow 20:00 UTC) it would still defer. const endOfUtcWindow = Date.parse("2026-06-15T17:00:00.000Z"); await vi.advanceTimersByTimeAsync(endOfUtcWindow - Date.now()); + await pokeIntervalWake(); expect(runSpy).toHaveBeenCalled(); const firstCall = new Date(expectDefined(callTimes[0], "callTimes[0] test invariant")); - expect(firstCall.getUTCHours()).toBe(16); expect(firstCall.getUTCDate()).toBe(15); runner.stop(); @@ -353,7 +318,16 @@ describe("heartbeat scheduler: activeHours-aware scheduling (#75487)", () => { stableSchedulerSeed: TEST_SCHEDULER_SEED, }); - await vi.advanceTimersByTimeAsync(16 * 60 * 60_000 + 60_000); + // Quiet-hours poke long before tomorrow's window must defer. + await vi.advanceTimersByTimeAsync(3 * 60 * 60_000); + await pokeIntervalWake(); + expect(runSpy).not.toHaveBeenCalled(); + + // 30s spacing puts a slot no later than 09:00:30 inside the one-minute + // window; a poke there must fire within the window. + const inWindowMs = Date.parse("2026-06-16T09:00:30.000Z"); + await vi.advanceTimersByTimeAsync(inWindowMs - Date.now()); + await pokeIntervalWake(); expect(callTimes.length).toBeGreaterThan(0); for (const callTime of callTimes) { diff --git a/src/infra/heartbeat-runner.commitments.test.ts b/src/infra/heartbeat-runner.commitments.test.ts index 3ac9ab132ab4..125d83f8cb0f 100644 --- a/src/infra/heartbeat-runner.commitments.test.ts +++ b/src/infra/heartbeat-runner.commitments.test.ts @@ -476,6 +476,58 @@ describe("runHeartbeatOnce commitments", () => { }); }); + it("delivers due commitments on a targeted cron-monitor interval tick", async () => { + vi.useFakeTimers(); + vi.setSystemTime(nowMs); + + await withTempHeartbeatSandbox(async ({ tmpDir, storePath }) => { + setTestEnvValue("OPENCLAW_STATE_DIR", tmpDir); + const dueSessionKey = "agent:main:telegram:user-155462274"; + const cfg: OpenClawConfig = { + agents: { + defaults: { + workspace: tmpDir, + heartbeat: { every: "5m", target: "last" }, + }, + }, + session: { store: storePath }, + }; + await saveCommitmentStore(undefined, { + version: 1, + commitments: [buildCommitment({ id: "cm_interview", sessionKey: dueSessionKey, to: "1" })], + }); + const runOnce = vi.fn().mockResolvedValue({ status: "ran", durationMs: 1 }); + const runner = startHeartbeatRunner({ + cfg, + runOnce, + stableSchedulerSeed: "commitment-monitor-tick", + }); + + // Reach the agent's due slot first: scheduled-intent wakes defer with + // not-due until the phase boundary passes. + await vi.advanceTimersByTimeAsync(10 * 60_000); + // The cron heartbeat monitor pokes with an agentId; that targeted + // interval tick must keep the commitment fan-out the broadcast timer had. + requestHeartbeat({ + source: "interval", + intent: "scheduled", + reason: "interval", + agentId: "main", + coalesceMs: 0, + }); + await vi.advanceTimersByTimeAsync(1); + await vi.waitFor(() => expect(runOnce).toHaveBeenCalledTimes(2)); + runner.stop(); + + expect(runOnce.mock.calls[0]?.[0]).toMatchObject({ agentId: "main", runScope: "global" }); + expect(runOnce.mock.calls[1]?.[0]).toMatchObject({ + agentId: "main", + runScope: "commitment-only", + sessionKey: dueSessionKey, + }); + }); + }); + it("delivers due commitments to the original scope when heartbeat target is last", async () => { const { result, sendTelegram, store } = await setupCommitmentCase(); diff --git a/src/infra/heartbeat-runner.scheduler.test.ts b/src/infra/heartbeat-runner.scheduler.test.ts index ec0cec36bc08..06741b2b84be 100644 --- a/src/infra/heartbeat-runner.scheduler.test.ts +++ b/src/infra/heartbeat-runner.scheduler.test.ts @@ -1,4 +1,7 @@ -// Tests heartbeat runner scheduling and timer cleanup. +// Tests heartbeat runner wake dispatch, cooldown bookkeeping, and cleanup. +// Interval cadence is owned by system cron monitor jobs; tests drive the +// scheduled path by poking `requestHeartbeat({source:"interval"})` after +// advancing fake time past the due slot. import { afterEach, describe, expect, it, vi } from "vitest"; import { getRuntimeConfig, @@ -9,15 +12,12 @@ import { import { startHeartbeatRunner } from "./heartbeat-runner.js"; import { computeNextHeartbeatPhaseDueMs, resolveHeartbeatPhaseMs } from "./heartbeat-schedule.js"; import { - HEARTBEAT_SKIP_CRON_IN_PROGRESS, HEARTBEAT_SKIP_REQUESTS_IN_FLIGHT, requestHeartbeat, + setHeartbeatWakeHandler, } from "./heartbeat-wake.js"; describe("startHeartbeatRunner", () => { - type RetryableHeartbeatBusySkipReason = - | typeof HEARTBEAT_SKIP_CRON_IN_PROGRESS - | typeof HEARTBEAT_SKIP_REQUESTS_IN_FLIGHT; type RunOnce = Parameters[0]["runOnce"]; type MockRunOnce = RunOnce & { mock: { calls: unknown[][] } }; const TEST_SCHEDULER_SEED = "heartbeat-runner-test-seed"; @@ -46,6 +46,50 @@ describe("startHeartbeatRunner", () => { } as OpenClawConfig; } + it("keeps a local fallback timer when cron is disabled, none when cron owns cadence", async () => { + useFakeHeartbeatTime(); + const cronDisabledRun = vi.fn().mockResolvedValue({ status: "ran", durationMs: 1 }); + const disabledCfg = { + ...heartbeatConfig(), + cron: { enabled: false }, + } as OpenClawConfig; + const fallbackRunner = startHeartbeatRunner({ + cfg: disabledCfg, + runOnce: cronDisabledRun, + stableSchedulerSeed: TEST_SCHEDULER_SEED, + }); + // Shipped contract: cron.enabled=false gateways still get scheduled + // heartbeats — the runner self-fires without any external poke. + await vi.advanceTimersByTimeAsync(31 * 60_000); + expect(cronDisabledRun).toHaveBeenCalled(); + fallbackRunner.stop(); + + const cronOwnedRun = vi.fn().mockResolvedValue({ status: "ran", durationMs: 1 }); + const cronOwnedRunner = startDefaultRunner(cronOwnedRun); + // With cron owning cadence there is no self-firing timer. + await vi.advanceTimersByTimeAsync(31 * 60_000); + expect(cronOwnedRun).not.toHaveBeenCalled(); + cronOwnedRunner.stop(); + }); + + it("chains clamped fallback timers past Node's setTimeout cap to long due times", async () => { + useFakeHeartbeatTime(); + const runOnce = vi.fn().mockResolvedValue({ status: "ran", durationMs: 1 }); + const runner = startHeartbeatRunner({ + cfg: { + agents: { defaults: { heartbeat: { every: "60d" } } }, + cron: { enabled: false }, + } as OpenClawConfig, + runOnce, + stableSchedulerSeed: TEST_SCHEDULER_SEED, + }); + // 60d exceeds the ~24.85d setTimeout cap: each clamped firing defers + // not-due and must re-arm until the real due slot inside 60d passes. + await vi.advanceTimersByTimeAsync(61 * 24 * 60 * 60 * 1000); + expect(runOnce).toHaveBeenCalled(); + runner.stop(); + }); + function resolveDueFromNow(nowMs: number, intervalMs: number, agentId: string) { return computeNextHeartbeatPhaseDueMs({ nowMs, @@ -58,15 +102,16 @@ describe("startHeartbeatRunner", () => { }); } - function createRetryableBusyRunSpy(reason: RetryableHeartbeatBusySkipReason, skipCount: number) { - let callCount = 0; - return vi.fn().mockImplementation(async () => { - callCount++; - if (callCount <= skipCount) { - return { status: "skipped", reason } as const; - } - return { status: "ran", durationMs: 1 } as const; + // Stand-in for a system cron monitor tick: the cron job pokes the wake + // queue; the runner decides via `nextDueMs` whether the agent is due. + async function pokeIntervalWake() { + requestHeartbeat({ + source: "interval", + intent: "scheduled", + reason: "interval", + coalesceMs: 0, }); + await vi.advanceTimersByTimeAsync(1); } function getRunCall(runSpy: MockRunOnce, callIndex: number) { @@ -184,6 +229,7 @@ describe("startHeartbeatRunner", () => { const firstDueMs = resolveDueFromNow(0, 30 * 60_000, "main"); await vi.advanceTimersByTimeAsync(firstDueMs + 1); + await pokeIntervalWake(); expect(runSpy).toHaveBeenCalledTimes(1); expectRunCallFields(runSpy, 0, { agentId: "main", reason: "interval" }); @@ -204,6 +250,7 @@ describe("startHeartbeatRunner", () => { const finalDueMs = Math.max(nextMainDueMs, nextOpsDueMs); await vi.advanceTimersByTimeAsync(finalDueMs - Date.now() + 1); + await pokeIntervalWake(); const reloadedAgentIds = runSpy.mock.calls.slice(1).map((call) => call[0]?.agentId); expect(reloadedAgentIds).toContain("main"); @@ -268,6 +315,7 @@ describe("startHeartbeatRunner", () => { const opsDueMs = resolveDueFromNow(0, 30 * 60_000, "ops"); await vi.advanceTimersByTimeAsync(Math.max(mainDueMs, opsDueMs) + 1); + await pokeIntervalWake(); const agentIds = runSpy.mock.calls.map((call) => call[0]?.agentId); expect(agentIds).toContain("main"); @@ -276,7 +324,7 @@ describe("startHeartbeatRunner", () => { runner.stop(); }); - it("continues scheduling after runOnce throws an unhandled error", async () => { + it("keeps serving interval wakes after runOnce throws an unhandled error", async () => { useFakeHeartbeatTime(); let callCount = 0; @@ -292,12 +340,14 @@ describe("startHeartbeatRunner", () => { const runner = startDefaultRunner(runSpy); const firstDueMs = resolveDueFromNow(0, 30 * 60_000, "main"); - // First heartbeat fires and throws + // First interval poke fires and throws inside runOnce. await vi.advanceTimersByTimeAsync(firstDueMs + 1); + await pokeIntervalWake(); expect(runSpy).toHaveBeenCalledTimes(1); - // Second heartbeat should still fire (scheduler must not be dead) + // A later poke past the next due slot must still run (handler not dead). await vi.advanceTimersByTimeAsync(30 * 60_000); + await pokeIntervalWake(); expect(runSpy).toHaveBeenCalledTimes(2); runner.stop(); @@ -331,8 +381,9 @@ describe("startHeartbeatRunner", () => { // Stop runner A (stale cleanup) — should NOT kill runner B's handler runnerA.stop(); - // Runner B should still fire + // Runner B should still serve interval wakes await vi.advanceTimersByTimeAsync(firstDueMs + 1); + await pokeIntervalWake(); expect(runSpy2).toHaveBeenCalledTimes(1); expect(runSpy1).not.toHaveBeenCalled(); @@ -342,7 +393,7 @@ describe("startHeartbeatRunner", () => { runnerB.stop(); }); - it("run() returns skipped when runner is stopped", async () => { + it("ignores interval wakes after the runner is stopped", async () => { useFakeHeartbeatTime(); const runSpy = vi.fn().mockResolvedValue({ status: "ran", durationMs: 1 }); @@ -351,59 +402,20 @@ describe("startHeartbeatRunner", () => { runner.stop(); - // After stopping, no heartbeats should fire + // After stopping, pokes past the due slot must not reach runOnce. await vi.advanceTimersByTimeAsync(60 * 60_000); + await pokeIntervalWake(); expect(runSpy).not.toHaveBeenCalled(); - }); - it("reschedules timer when runOnce returns requests-in-flight", async () => { - useFakeHeartbeatTime(); - - const runSpy = createRetryableBusyRunSpy(HEARTBEAT_SKIP_REQUESTS_IN_FLIGHT, 1); - - const runner = startHeartbeatRunner({ - cfg: heartbeatConfig(), - runOnce: runSpy, - stableSchedulerSeed: TEST_SCHEDULER_SEED, - }); - const firstDueMs = resolveDueFromNow(0, 30 * 60_000, "main"); - - // First heartbeat returns requests-in-flight - await vi.advanceTimersByTimeAsync(firstDueMs + 1); - expect(runSpy).toHaveBeenCalledTimes(1); - - // The wake layer retries after DEFAULT_RETRY_MS (1 s). No scheduleNext() - // is called inside runOnce, so we must wait for the full cooldown. - await vi.advanceTimersByTimeAsync(1_000); - expect(runSpy).toHaveBeenCalledTimes(2); - - runner.stop(); - }); - - it("reschedules timer when runOnce returns cron-in-progress", async () => { - useFakeHeartbeatTime(); - - const runSpy = createRetryableBusyRunSpy(HEARTBEAT_SKIP_CRON_IN_PROGRESS, 1); - - const runner = startHeartbeatRunner({ - cfg: heartbeatConfig(), - runOnce: runSpy, - stableSchedulerSeed: TEST_SCHEDULER_SEED, - }); - const firstDueMs = resolveDueFromNow(0, 30 * 60_000, "main"); - - await vi.advanceTimersByTimeAsync(firstDueMs + 1); - expect(runSpy).toHaveBeenCalledTimes(1); - - await vi.advanceTimersByTimeAsync(1_000); - expect(runSpy).toHaveBeenCalledTimes(2); - - runner.stop(); + // Drain the wake queued while no handler was registered so it cannot + // leak into the next test's freshly registered handler. + const disposeDrain = setHeartbeatWakeHandler(async () => ({ status: "ran", durationMs: 0 })); + await vi.advanceTimersByTimeAsync(300); + disposeDrain(); }); it("advances cadence after non-retryable disabled skips", async () => { useFakeHeartbeatTime(); - const timeoutSpy = vi.spyOn(globalThis, "setTimeout"); const runSpy = vi.fn().mockResolvedValue({ status: "skipped", reason: "disabled" } as const); const intervalMs = 10 * 60_000; @@ -415,23 +427,20 @@ describe("startHeartbeatRunner", () => { const firstDueMs = resolveDueFromNow(0, intervalMs, "main"); await vi.advanceTimersByTimeAsync(firstDueMs + 1); + await pokeIntervalWake(); expect(runSpy).toHaveBeenCalledTimes(1); - const delays = timeoutSpy.mock.calls - .map((call) => call[1]) - .filter((delay): delay is number => typeof delay === "number"); - expect(delays[delays.length - 1]).toBeGreaterThan(5_000); - + // Non-retryable skip advanced nextDueMs to the next slot, so an interval + // poke shortly after must defer with not-due instead of re-running. await vi.advanceTimersByTimeAsync(2_000); + await pokeIntervalWake(); expect(runSpy).toHaveBeenCalledTimes(1); - timeoutSpy.mockRestore(); runner.stop(); }); it("advances normal cadence after terminal tool failures", async () => { useFakeHeartbeatTime(); - const timeoutSpy = vi.spyOn(globalThis, "setTimeout"); const runSpy = vi .fn() .mockResolvedValue({ status: "failed", reason: "agent-tool-failure" } as const); @@ -445,23 +454,20 @@ describe("startHeartbeatRunner", () => { const firstDueMs = resolveDueFromNow(0, intervalMs, "main"); await vi.advanceTimersByTimeAsync(firstDueMs + 1); + await pokeIntervalWake(); expect(runSpy).toHaveBeenCalledTimes(1); - const delays = timeoutSpy.mock.calls - .map((call) => call[1]) - .filter((delay): delay is number => typeof delay === "number"); - expect(delays[delays.length - 1]).toBeGreaterThan(5_000); - + // Terminal failure still advances the cadence — a poke inside the new + // cooldown window must not re-run the failing heartbeat. await vi.advanceTimersByTimeAsync(2_000); + await pokeIntervalWake(); expect(runSpy).toHaveBeenCalledTimes(1); - timeoutSpy.mockRestore(); runner.stop(); }); - it("advances cadence after flood deferrals without wake-layer retry", async () => { + it("flood guard defers due interval wakes after repeated runs", async () => { useFakeHeartbeatTime(); - const timeoutSpy = vi.spyOn(globalThis, "setTimeout"); const runSpy = vi.fn().mockResolvedValue({ status: "ran", durationMs: 1 } as const); const intervalMs = 1_000; @@ -473,20 +479,19 @@ describe("startHeartbeatRunner", () => { const firstDueMs = resolveDueFromNow(0, intervalMs, "main"); await vi.advanceTimersByTimeAsync(firstDueMs + 1); + await pokeIntervalWake(); for (let i = 0; i < 4; i++) { await vi.advanceTimersByTimeAsync(intervalMs); + await pokeIntervalWake(); } expect(runSpy).toHaveBeenCalledTimes(5); + // Five runs inside the flood window: the next due interval poke defers + // via the flood guard, and the deferral is terminal (no wake-layer retry). await vi.advanceTimersByTimeAsync(intervalMs); + await pokeIntervalWake(); expect(runSpy).toHaveBeenCalledTimes(5); - const delays = timeoutSpy.mock.calls - .map((call) => call[1]) - .filter((delay): delay is number => typeof delay === "number"); - expect(delays[delays.length - 1]).toBeGreaterThan(0); - - timeoutSpy.mockRestore(); runner.stop(); }); @@ -514,23 +519,28 @@ describe("startHeartbeatRunner", () => { const intervalMs = 30 * 60_000; const firstDueMs = resolveDueFromNow(0, intervalMs, "main"); - // Trigger the first heartbeat at the agent's first slot — returns requests-in-flight. + // Poke the first heartbeat at the agent's first slot — returns + // requests-in-flight, so no bookkeeping is recorded. await vi.advanceTimersByTimeAsync(firstDueMs + 1); + await pokeIntervalWake(); expect(runSpy).toHaveBeenCalledTimes(1); - // Simulate 4 more retries at short intervals (wake layer retries). - for (let i = 0; i < 4; i++) { - requestHeartbeat(wake("retry", { coalesceMs: 0 })); + // The wake layer auto-retries the busy interval wake every 1s; the busy + // skips must not advance nextDueMs, so each retry reaches runOnce until + // the 6th attempt succeeds. + for (let i = 0; i < 5; i++) { await vi.advanceTimersByTimeAsync(1_000); } + expect(runSpy).toHaveBeenCalledTimes(6); const scheduledSlotCallsBeforeInterval = callTimes.filter( (time) => time >= firstDueMs + intervalMs, ); expect(scheduledSlotCallsBeforeInterval).toStrictEqual([]); - // The next interval tick at the next scheduled slot should still fire — + // The next interval poke at the next scheduled slot should still fire — // the retries must not push the phase out by multiple intervals. await vi.advanceTimersByTimeAsync(firstDueMs + intervalMs - Date.now() + 1); + await pokeIntervalWake(); const scheduledSlotCallsAfterInterval = callTimes.filter( (time) => time >= firstDueMs + intervalMs, ); @@ -680,24 +690,6 @@ describe("startHeartbeatRunner", () => { runner.stop(); }); - it("clamps oversized scheduler delays so heartbeats do not fire in a tight loop (#71414)", async () => { - useFakeHeartbeatTime(); - const runSpy = vi.fn().mockResolvedValue({ status: "ran", durationMs: 1 }); - // 365d resolves to ~31_536_000_000 ms, well past Node setTimeout's - // 2_147_483_647 ms cap. Without clamping, setTimeout would fire after - // 1ms and re-arm in a tight loop, exhausting the runner. - const runner = startHeartbeatRunner({ - cfg: heartbeatConfig([{ id: "main", heartbeat: { every: "365d" } }]), - runOnce: runSpy, - stableSchedulerSeed: TEST_SCHEDULER_SEED, - }); - // Advance well past the broken 1ms re-arm but well under the clamped cap - // (~24.85d). If the bug is present, runSpy gets called many times. - await vi.advanceTimersByTimeAsync(60_000); - expect(runSpy).not.toHaveBeenCalled(); - runner.stop(); - }); - it("does not fan out to unrelated agents for session-scoped exec wakes", async () => { useFakeHeartbeatTime(); const runSpy = vi.fn().mockResolvedValue({ status: "ran", durationMs: 1 }); diff --git a/src/infra/heartbeat-runner.timeout-warning.test.ts b/src/infra/heartbeat-runner.timeout-warning.test.ts deleted file mode 100644 index 97ec6b86100f..000000000000 --- a/src/infra/heartbeat-runner.timeout-warning.test.ts +++ /dev/null @@ -1,67 +0,0 @@ -// Covers heartbeat timeout warning emission and suppression behavior. -import { afterEach, describe, expect, it, vi } from "vitest"; -import type { OpenClawConfig } from "../config/config.js"; - -function createHeartbeatConfig(every: string): OpenClawConfig { - return { - agents: { - defaults: { heartbeat: { every } }, - list: [{ id: "main", heartbeat: { every } }], - }, - } as OpenClawConfig; -} - -describe("startHeartbeatRunner timeout overflow warnings", () => { - afterEach(() => { - vi.useRealTimers(); - vi.resetModules(); - vi.restoreAllMocks(); - }); - - it("warns once per runner lifetime when clamping an oversized scheduler delay", async () => { - const warn = vi.fn(); - const noop = vi.fn(); - const logger = { - subsystem: "gateway/heartbeat", - isEnabled: vi.fn(() => true), - trace: noop, - debug: noop, - info: noop, - warn, - error: noop, - fatal: noop, - raw: noop, - child: vi.fn(() => logger), - }; - - vi.doMock("../logging/subsystem.js", async () => { - const actual = - await vi.importActual("../logging/subsystem.js"); - return { - ...actual, - createSubsystemLogger: vi.fn(() => logger), - }; - }); - - const { startHeartbeatRunner } = await import("./heartbeat-runner.js"); - - vi.useFakeTimers(); - vi.setSystemTime(new Date(0)); - const cfg = createHeartbeatConfig("365d"); - const runnerA = startHeartbeatRunner({ - cfg, - runOnce: vi.fn().mockResolvedValue({ status: "ran", durationMs: 1 }), - stableSchedulerSeed: "seed-0", - }); - const runnerB = startHeartbeatRunner({ - cfg, - runOnce: vi.fn().mockResolvedValue({ status: "ran", durationMs: 1 }), - stableSchedulerSeed: "seed-0", - }); - - expect(warn).toHaveBeenCalledTimes(2); - - runnerA.stop(); - runnerB.stop(); - }); -}); diff --git a/src/infra/heartbeat-runner.ts b/src/infra/heartbeat-runner.ts index df507ded9022..1e246ea6c00f 100644 --- a/src/infra/heartbeat-runner.ts +++ b/src/infra/heartbeat-runner.ts @@ -111,7 +111,7 @@ import { } from "../routing/session-key.js"; import { defaultRuntime, type RuntimeEnv } from "../runtime.js"; import { createLazyRuntimeModule } from "../shared/lazy-runtime.js"; -import { MAX_SAFE_TIMEOUT_DELAY_MS, resolveSafeTimeoutDelayMs } from "../utils/timer-delay.js"; +import { resolveSafeTimeoutDelayMs } from "../utils/timer-delay.js"; import { loadOrCreateDeviceIdentity } from "./device-identity.js"; import { formatErrorMessage, hasErrnoCode } from "./errors.js"; import { resolveMainScopedEventSessionKey } from "./event-session-routing.js"; @@ -316,7 +316,7 @@ export type HeartbeatRunner = { updateConfig: (cfg: OpenClawConfig) => void; }; -function resolveHeartbeatSchedulerSeed(explicitSeed?: string) { +export function resolveHeartbeatSchedulerSeed(explicitSeed?: string) { const normalized = normalizeOptionalString(explicitSeed); if (normalized) { return normalized; @@ -380,7 +380,7 @@ function resolveHeartbeatForWake(params: { : heartbeat; } -function resolveHeartbeatAgents(cfg: OpenClawConfig): HeartbeatAgent[] { +export function resolveHeartbeatAgents(cfg: OpenClawConfig): HeartbeatAgent[] { const list = cfg.agents?.list ?? []; if (hasExplicitHeartbeatAgents(cfg)) { return list @@ -2321,17 +2321,57 @@ export function startHeartbeatRunner(opts: { }): HeartbeatRunner { const runtime = opts.runtime ?? defaultRuntime; const runOnce = opts.runOnce ?? runHeartbeatOnce; + // Interval cadence is owned by the system cron monitor jobs (one per + // heartbeat agent, converged by the gateway); this runner only executes + // wakes. `nextDueMs` survives as the cooldown gate that decides whether an + // incoming wake — cron tick or event — is due yet. When cron itself is + // disabled (shipped `cron.enabled=false` / OPENCLAW_SKIP_CRON contract), a + // local fallback timer keeps heartbeats alive; removal plan: fold heartbeat + // enablement into cron config in the #110950 config migration. const state = { cfg: opts.cfg ?? getRuntimeConfig(), runtime, schedulerSeed: resolveHeartbeatSchedulerSeed(opts.stableSchedulerSeed), agents: new Map(), timer: null as NodeJS.Timeout | null, + fallbackCadence: false, stopped: false, }; const readCurrentConfig = opts.readCurrentConfig ?? (() => state.cfg); let initialized = false; - let heartbeatTimeoutOverflowWarned = false; + const cronOwnsCadence = (cfg: OpenClawConfig) => + process.env.OPENCLAW_SKIP_CRON !== "1" && cfg.cron?.enabled !== false; + + const scheduleFallbackNext = (minDelayMs = 0) => { + if (state.timer) { + clearTimeout(state.timer); + state.timer = null; + } + if (state.stopped || !state.fallbackCadence || state.agents.size === 0) { + return; + } + const now = Date.now(); + let nextDue = Number.POSITIVE_INFINITY; + for (const agent of state.agents.values()) { + if (agent.nextDueMs < nextDue) { + nextDue = agent.nextDueMs; + } + } + if (!Number.isFinite(nextDue)) { + return; + } + const delay = resolveSafeTimeoutDelayMs(Math.max(minDelayMs, nextDue - now), { minMs: 0 }); + state.timer = setTimeout(() => { + state.timer = null; + requestHeartbeat({ + source: "interval", + intent: "scheduled", + reason: "interval", + coalesceMs: 0, + }); + }, delay); + state.timer.unref?.(); + }; const resolveNextDue = ( now: number, @@ -2374,6 +2414,9 @@ export function startHeartbeatRunner(opts: { // for cooldown purposes, so keep the existing now + interval behavior. now + agent.intervalMs; agent.nextDueMs = seekActiveSlotForAgent(agent, rawDueMs); + // Every due-slot move re-arms the cron-disabled fallback timer; with cron + // owning cadence this is a no-op. + scheduleFallbackNext(); }; const advanceStaleScheduleAfterDeferral = ( @@ -2383,10 +2426,13 @@ export function startHeartbeatRunner(opts: { decision?: DeferDecision, ) => { if (!decision?.defer || decision.reason === "not-due" || agent.nextDueMs > now) { + // A clamped fallback timer (interval beyond Node's setTimeout cap) can + // fire before nextDueMs; re-arm so the chain reaches the real due time. + scheduleFallbackNext(); return; } // Deferrals that do not have wake-layer retry ownership still need to move - // the due slot forward; otherwise scheduleNext() will keep rearming at 0ms. + // the due slot forward; otherwise the fallback timer would rearm at 0ms. advanceAgentSchedule(agent, now, reason); }; @@ -2429,48 +2475,6 @@ export function startHeartbeatRunner(opts: { agent.floodLoggedSinceLastRun = false; }; - const scheduleNext = () => { - if (state.stopped) { - return; - } - if (state.timer) { - clearTimeout(state.timer); - state.timer = null; - } - if (state.agents.size === 0) { - return; - } - const now = Date.now(); - let nextDue = Number.POSITIVE_INFINITY; - for (const agent of state.agents.values()) { - if (agent.nextDueMs < nextDue) { - nextDue = agent.nextDueMs; - } - } - if (!Number.isFinite(nextDue)) { - return; - } - const rawDelay = Math.max(0, nextDue - now); - if (rawDelay > MAX_SAFE_TIMEOUT_DELAY_MS && !heartbeatTimeoutOverflowWarned) { - heartbeatTimeoutOverflowWarned = true; - log.warn("heartbeat: scheduled delay exceeds Node setTimeout cap; clamping to ~24.85d", { - rawDelayMs: rawDelay, - clampedMs: MAX_SAFE_TIMEOUT_DELAY_MS, - }); - } - const delay = resolveSafeTimeoutDelayMs(rawDelay, { minMs: 0 }); - state.timer = setTimeout(() => { - state.timer = null; - requestHeartbeat({ - source: "interval", - intent: "scheduled", - reason: "interval", - coalesceMs: 0, - }); - }, delay); - state.timer.unref?.(); - }; - const updateConfig = (cfg: OpenClawConfig) => { if (state.stopped) { return; @@ -2540,8 +2544,8 @@ export function startHeartbeatRunner(opts: { log.info("heartbeat: started", { intervalMs: Math.min(...intervals) }); } } - - scheduleNext(); + state.fallbackCadence = !cronOwnsCadence(cfg); + scheduleFallbackNext(); }; const run: HeartbeatWakeHandler = async (params) => { @@ -2587,209 +2591,217 @@ export function startHeartbeatRunner(opts: { const startedAt = Date.now(); const now = startedAt; let ran = false; - // Track retryable busy skips so we can skip re-arm in finally — the wake - // layer handles retry for this case (DEFAULT_RETRY_MS = 1 s). - let retryableBusySkip = false; - try { - if (requestedSessionKey || requestedAgentId) { - const targetAgentId = requestedTargetAgentId ?? resolveDefaultAgentId(wakeConfig); - const targetAgent = state.agents.get(targetAgentId); - // A user-present targeted event may wake an unscheduled agent once. It - // must not enroll that agent in the recurring heartbeat scheduler. - if (!targetAgent && !allowsUnscheduledTarget) { - return { status: "skipped", reason: "disabled" }; - } - if (targetAgent) { - const deferral = evaluateWakeDeferral(targetAgent, now, reason, intent); - if (deferral.defer) { - advanceStaleScheduleAfterDeferral(targetAgent, now, reason, deferral); - return { status: "skipped", reason: deferral.reason }; - } - } - try { - const res = await runOnce({ - cfg: wakeConfig, - agentId: targetAgentId, - heartbeat: resolveHeartbeatForWake({ - cfg: wakeConfig, - agentId: targetAgentId, - configuredHeartbeat: targetAgent?.heartbeat, - requestedHeartbeat, - source: params.source, - mergeRequestedHeartbeat: true, - }), - source: params.source, - intent, - reason, - runScope: "global", - sessionKey: requestedSessionKey, - deps: { runtime: state.runtime }, - }); - if (res.status === "skipped" && isRetryableHeartbeatBusySkipReason(res.reason)) { - // Retryable busy — do NOT record run bookkeeping. The wake layer - // retries the same reason shortly; if we recorded `lastRunStartedAtMs` - // here, the retry would falsely defer with `not-due`/`min-spacing` - // because the cooldown would treat this skipped attempt as a real run. - retryableBusySkip = true; - return res; - } - // Non-retryable outcome (ran, disabled, failed-but-not-busy). Record - // bookkeeping and move the due slot so scheduleNext() cannot hot-loop - // on a stale past-due agent. - if (targetAgent) { - recordRunBookkeeping(targetAgent, now); - advanceAgentSchedule(targetAgent, now, reason); - } - return res.status === "ran" ? { status: "ran", durationMs: Date.now() - startedAt } : res; - } catch (err) { - const errMsg = formatErrorMessage(err); - log.error(`heartbeat runner: targeted runOnce threw unexpectedly: ${errMsg}`, { - error: errMsg, - }); - // Throw counts as a non-retryable terminal attempt for cooldown - // purposes — record bookkeeping so the wake layer doesn't tight-loop - // on the same reason. - if (targetAgent) { - recordRunBookkeeping(targetAgent, now); - advanceAgentSchedule(targetAgent, now, reason); - } - return { status: "failed", reason: errMsg }; - } + // Run each agent's wake concurrently. Heartbeat work is per-agent — + // separate session stores, lanes, and delivery targets — so awaiting + // one slow agent (e.g. one whose heartbeat spawns a multi-minute + // subagent) must not starve the others. Bookkeeping mutations only + // touch the owning agent's `HeartbeatAgentState`, so the per-agent + // closures are safe to fan out under `Promise.all`. + type AgentWakeOutcome = { + ran: boolean; + retryableBusySkip?: HeartbeatRunResult; + // Terminal per-agent result so targeted callers can report the real + // skip reason instead of collapsing everything to not-due. + result?: HeartbeatRunResult; + }; + const runOneAgent = async (agent: HeartbeatAgentState): Promise => { + const deferral = evaluateWakeDeferral(agent, now, reason, intent); + if (deferral.defer) { + advanceStaleScheduleAfterDeferral(agent, now, reason, deferral); + return { ran: false, result: { status: "skipped", reason: deferral.reason } }; } - // Run each agent's wake concurrently. Heartbeat work is per-agent — - // separate session stores, lanes, and delivery targets — so awaiting - // one slow agent (e.g. one whose heartbeat spawns a multi-minute - // subagent) must not starve the others. Bookkeeping mutations only - // touch the owning agent's `HeartbeatAgentState`, so the per-agent - // closures are safe to fan out under `Promise.all`. - type AgentWakeOutcome = { - ran: boolean; - retryableBusySkip?: HeartbeatRunResult; - }; - const runOneAgent = async (agent: HeartbeatAgentState): Promise => { - const deferral = evaluateWakeDeferral(agent, now, reason, intent); - if (deferral.defer) { - advanceStaleScheduleAfterDeferral(agent, now, reason, deferral); - return { ran: false }; - } + let res: HeartbeatRunResult; + try { + res = await runOnce({ + cfg: wakeConfig, + agentId: agent.agentId, + heartbeat: agent.heartbeat, + source: params.source, + intent, + reason, + runScope: "global", + deps: { runtime: state.runtime }, + }); + } catch (err) { + const errMsg = formatErrorMessage(err); + log.error(`heartbeat runner: runOnce threw unexpectedly: ${errMsg}`, { + error: errMsg, + agentId: agent.agentId, + }); + // Throw counts as a non-retryable terminal attempt for cooldown + // purposes — record bookkeeping so the wake layer doesn't tight-loop + // on the same reason. + recordRunBookkeeping(agent, now); + advanceAgentSchedule(agent, now, reason); + return { ran: false, result: { status: "failed", reason: formatErrorMessage(err) } }; + } + if (res.status === "skipped" && isRetryableHeartbeatBusySkipReason(res.reason)) { + // Do not advance the schedule or record run bookkeeping for this + // agent — its target runtime is busy and the wake layer retries. + return { ran: false, retryableBusySkip: res }; + } + // Non-retryable outcome — record bookkeeping for cooldown gates. + recordRunBookkeeping(agent, now); + advanceAgentSchedule(agent, now, reason); + let agentRan = res.status === "ran"; - let res: HeartbeatRunResult; + const defaultSessionKey = resolveHeartbeatSession( + wakeConfig, + agent.agentId, + agent.heartbeat, + ).sessionKey; + const dueSessionKeys = canHeartbeatDeliverCommitments(agent.heartbeat) + ? await listDueCommitmentSessionKeys({ + cfg: wakeConfig, + agentId: agent.agentId, + nowMs: now, + limit: 10, + }) + : []; + for (const dueSessionKey of dueSessionKeys) { + if (dueSessionKey === defaultSessionKey) { + continue; + } + let commitmentRes: HeartbeatRunResult; try { - res = await runOnce({ + commitmentRes = await runOnce({ cfg: wakeConfig, agentId: agent.agentId, heartbeat: agent.heartbeat, - source: params.source, - intent, - reason, - runScope: "global", + runScope: "commitment-only", + sessionKey: dueSessionKey, deps: { runtime: state.runtime }, }); } catch (err) { const errMsg = formatErrorMessage(err); - log.error(`heartbeat runner: runOnce threw unexpectedly: ${errMsg}`, { + log.error(`heartbeat runner: commitment runOnce threw unexpectedly: ${errMsg}`, { error: errMsg, agentId: agent.agentId, }); - // Throw counts as a non-retryable terminal attempt for cooldown - // purposes — record bookkeeping so the wake layer doesn't tight-loop - // on the same reason. - recordRunBookkeeping(agent, now); - advanceAgentSchedule(agent, now, reason); - return { ran: false }; + continue; } - if (res.status === "skipped" && isRetryableHeartbeatBusySkipReason(res.reason)) { - // Do not advance the schedule or record run bookkeeping for this - // agent — its target runtime is busy and the wake layer retries. - return { ran: false, retryableBusySkip: res }; + if ( + commitmentRes.status === "skipped" && + isRetryableHeartbeatBusySkipReason(commitmentRes.reason) + ) { + return { ran: agentRan, retryableBusySkip: commitmentRes, result: res }; } - // Non-retryable outcome — record bookkeeping for cooldown gates. - recordRunBookkeeping(agent, now); - advanceAgentSchedule(agent, now, reason); - let agentRan = res.status === "ran"; - - const defaultSessionKey = resolveHeartbeatSession( - wakeConfig, - agent.agentId, - agent.heartbeat, - ).sessionKey; - const dueSessionKeys = canHeartbeatDeliverCommitments(agent.heartbeat) - ? await listDueCommitmentSessionKeys({ - cfg: wakeConfig, - agentId: agent.agentId, - nowMs: now, - limit: 10, - }) - : []; - for (const dueSessionKey of dueSessionKeys) { - if (dueSessionKey === defaultSessionKey) { - continue; - } - let commitmentRes: HeartbeatRunResult; - try { - commitmentRes = await runOnce({ - cfg: wakeConfig, - agentId: agent.agentId, - heartbeat: agent.heartbeat, - runScope: "commitment-only", - sessionKey: dueSessionKey, - deps: { runtime: state.runtime }, - }); - } catch (err) { - const errMsg = formatErrorMessage(err); - log.error(`heartbeat runner: commitment runOnce threw unexpectedly: ${errMsg}`, { - error: errMsg, - agentId: agent.agentId, - }); - continue; - } - if ( - commitmentRes.status === "skipped" && - isRetryableHeartbeatBusySkipReason(commitmentRes.reason) - ) { - return { ran: agentRan, retryableBusySkip: commitmentRes }; - } - if (commitmentRes.status === "ran") { - agentRan = true; - } + if (commitmentRes.status === "ran") { + agentRan = true; } + } - return { ran: agentRan }; - }; + return { ran: agentRan, result: res }; + }; - const agentOutcomes = await Promise.all( - Array.from(state.agents.values()).map((agent) => runOneAgent(agent)), - ); - let firstRetryableBusy: HeartbeatRunResult | undefined; - for (const outcome of agentOutcomes) { + if (requestedSessionKey || requestedAgentId) { + const targetAgentId = requestedTargetAgentId ?? resolveDefaultAgentId(wakeConfig); + const targetAgent = state.agents.get(targetAgentId); + // A user-present targeted event may wake an unscheduled agent once. It + // must not enroll that agent in the recurring heartbeat scheduler. + if (!targetAgent && !allowsUnscheduledTarget) { + return { status: "skipped", reason: "disabled" }; + } + if (isInterval && targetAgent && !requestedSessionKey && !requestedHeartbeat) { + // Cron monitor tick for one enrolled agent: use the full per-agent + // path — including due-commitment sessions — that the broadcast + // interval owned before cadence moved to cron. Wakes carrying + // heartbeat overrides fall through to the targeted merge path. + // Intentional: interval ticks run on the enrollment snapshot + // (agent.heartbeat, refreshed by updateConfig), exactly like the + // replaced broadcast timer — not resolveHeartbeatForWake, which only + // ever served override-carrying targeted event wakes. + const outcome = await runOneAgent(targetAgent); + if (outcome.retryableBusySkip) { + return outcome.retryableBusySkip; + } if (outcome.ran) { - ran = true; + return { status: "ran", durationMs: Date.now() - startedAt }; } - if (outcome.retryableBusySkip && !firstRetryableBusy) { - firstRetryableBusy = outcome.retryableBusySkip; + return outcome.result ?? { status: "skipped", reason: "not-due" }; + } + if (targetAgent) { + const deferral = evaluateWakeDeferral(targetAgent, now, reason, intent); + if (deferral.defer) { + advanceStaleScheduleAfterDeferral(targetAgent, now, reason, deferral); + return { status: "skipped", reason: deferral.reason }; } } - if (firstRetryableBusy) { - // At least one agent's runtime was busy. The wake layer schedules a - // retry; on retry, agents that already advanced their schedule will - // defer via cooldown, so only the still-busy agent actually re-runs. - retryableBusySkip = true; - return firstRetryableBusy; - } - - if (ran) { - return { status: "ran", durationMs: Date.now() - startedAt }; - } - return { status: "skipped", reason: isInterval ? "not-due" : "disabled" }; - } finally { - // Always re-arm the timer — except for retryable busy skips, where the - // wake layer (heartbeat-wake.ts) handles retry via schedule(DEFAULT_RETRY_MS). - if (!retryableBusySkip) { - scheduleNext(); + try { + const res = await runOnce({ + cfg: wakeConfig, + agentId: targetAgentId, + heartbeat: resolveHeartbeatForWake({ + cfg: wakeConfig, + agentId: targetAgentId, + configuredHeartbeat: targetAgent?.heartbeat, + requestedHeartbeat, + source: params.source, + mergeRequestedHeartbeat: true, + }), + source: params.source, + intent, + reason, + runScope: "global", + sessionKey: requestedSessionKey, + deps: { runtime: state.runtime }, + }); + if (res.status === "skipped" && isRetryableHeartbeatBusySkipReason(res.reason)) { + // Retryable busy — do NOT record run bookkeeping. The wake layer + // retries the same reason shortly; if we recorded `lastRunStartedAtMs` + // here, the retry would falsely defer with `not-due`/`min-spacing` + // because the cooldown would treat this skipped attempt as a real run. + return res; + } + // Non-retryable outcome (ran, disabled, failed-but-not-busy). Record + // bookkeeping and move the due slot so scheduleNext() cannot hot-loop + // on a stale past-due agent. + if (targetAgent) { + recordRunBookkeeping(targetAgent, now); + advanceAgentSchedule(targetAgent, now, reason); + } + return res.status === "ran" ? { status: "ran", durationMs: Date.now() - startedAt } : res; + } catch (err) { + const errMsg = formatErrorMessage(err); + log.error(`heartbeat runner: targeted runOnce threw unexpectedly: ${errMsg}`, { + error: errMsg, + }); + // Throw counts as a non-retryable terminal attempt for cooldown + // purposes — record bookkeeping so the wake layer doesn't tight-loop + // on the same reason. + if (targetAgent) { + recordRunBookkeeping(targetAgent, now); + advanceAgentSchedule(targetAgent, now, reason); + } + return { status: "failed", reason: errMsg }; } } + + const agentOutcomes = await Promise.all( + Array.from(state.agents.values()).map((agent) => runOneAgent(agent)), + ); + let firstRetryableBusy: HeartbeatRunResult | undefined; + for (const outcome of agentOutcomes) { + if (outcome.ran) { + ran = true; + } + if (outcome.retryableBusySkip && !firstRetryableBusy) { + firstRetryableBusy = outcome.retryableBusySkip; + } + } + if (firstRetryableBusy) { + // At least one agent's runtime was busy. The wake layer schedules a + // retry; on retry, agents that already advanced their schedule will + // defer via cooldown, so only the still-busy agent actually re-runs. + return firstRetryableBusy; + } + + if (ran) { + return { status: "ran", durationMs: Date.now() - startedAt }; + } + return { status: "skipped", reason: isInterval ? "not-due" : "disabled" }; }; const wakeHandler: HeartbeatWakeHandler = async (params: HeartbeatWakeRequest) => @@ -2812,8 +2824,8 @@ export function startHeartbeatRunner(opts: { disposeWakeHandler(); if (state.timer) { clearTimeout(state.timer); + state.timer = null; } - state.timer = null; }; opts.abortSignal?.addEventListener("abort", cleanup, { once: true }); diff --git a/test/e2e/qa-lab/runtime/heartbeat-active-hours-runtime.ts b/test/e2e/qa-lab/runtime/heartbeat-active-hours-runtime.ts index 1ed2361c1883..0743e02ce45c 100644 --- a/test/e2e/qa-lab/runtime/heartbeat-active-hours-runtime.ts +++ b/test/e2e/qa-lab/runtime/heartbeat-active-hours-runtime.ts @@ -1,4 +1,5 @@ -// Heartbeat active-hours evidence runs the real bounded scheduler and reload path. +// Heartbeat active-hours evidence runs the real wake-lane guards and reload path. +// Interval cadence itself is covered by the system cron monitor integration tests. import fs from "node:fs/promises"; import path from "node:path"; import { pathToFileURL } from "node:url"; @@ -6,10 +7,12 @@ import type { OpenClawConfig } from "../../../../src/config/types.openclaw.js"; import { formatErrorMessage } from "../../../../src/infra/errors.js"; import { isWithinActiveHours } from "../../../../src/infra/heartbeat-active-hours.js"; import { startHeartbeatRunner } from "../../../../src/infra/heartbeat-runner.js"; +import { requestHeartbeat } from "../../../../src/infra/heartbeat-wake.js"; import { createQaScriptEvidenceWriter } from "./script-evidence.js"; const DEFAULT_TIMEOUT_MS = 5_000; -const HEARTBEAT_INTERVAL = "100ms"; +const HEARTBEAT_INTERVAL_MS = 100; +const HEARTBEAT_INTERVAL = `${HEARTBEAT_INTERVAL_MS}ms`; type HeartbeatRuntimeOptions = { artifactBase: string; @@ -77,7 +80,32 @@ async function waitForObservation( setTimeout(resolve, 20); }); } - throw new Error(`heartbeat scheduler did not observe ${outcome} within ${timeoutMs}ms`); + throw new Error(`heartbeat wake lane did not observe ${outcome} within ${timeoutMs}ms`); +} + +async function pokeScheduledHeartbeat(params: { + observations: SchedulerObservation[]; + outcome: SchedulerObservation["outcome"]; + afterCount: number; + timeoutMs: number; +}) { + // The cron monitor fires at or after the configured due slot. Wait past one + // interval so the runner's cooldown gate admits the equivalent scheduled poke. + await new Promise((resolve) => { + setTimeout(resolve, HEARTBEAT_INTERVAL_MS + 50); + }); + requestHeartbeat({ + source: "interval", + intent: "scheduled", + reason: "interval", + coalesceMs: 0, + }); + await waitForObservation( + params.observations, + params.outcome, + params.afterCount, + params.timeoutMs, + ); } function createWriter(options: HeartbeatRuntimeOptions) { @@ -122,15 +150,30 @@ export async function runHeartbeatActiveHoursRuntime(options: HeartbeatRuntimeOp stableSchedulerSeed: "qa-heartbeat-active-hours", }); try { - await waitForObservation(observations, "active-fire", 0, options.timeoutMs); + await pokeScheduledHeartbeat({ + observations, + outcome: "active-fire", + afterCount: 0, + timeoutMs: options.timeoutMs, + }); const beforeQuiet = observations.length; currentConfig = heartbeatConfig(true); runner.updateConfig(currentConfig); - await waitForObservation(observations, "quiet-hours-skip", beforeQuiet, options.timeoutMs); + await pokeScheduledHeartbeat({ + observations, + outcome: "quiet-hours-skip", + afterCount: beforeQuiet, + timeoutMs: options.timeoutMs, + }); const beforeReload = observations.length; currentConfig = heartbeatConfig(false); runner.updateConfig(currentConfig); - await waitForObservation(observations, "active-fire", beforeReload, options.timeoutMs); + await pokeScheduledHeartbeat({ + observations, + outcome: "active-fire", + afterCount: beforeReload, + timeoutMs: options.timeoutMs, + }); const summaryPath = path.join(options.artifactBase, "heartbeat-active-hours-summary.json"); await fs.writeFile(summaryPath, `${JSON.stringify({ observations }, null, 2)}\n`, "utf8"); diff --git a/ui/src/lib/cron/index.ts b/ui/src/lib/cron/index.ts index b92e69128ca2..5d783915847e 100644 --- a/ui/src/lib/cron/index.ts +++ b/ui/src/lib/cron/index.ts @@ -52,7 +52,9 @@ export type CronFormState = { staggerUnit: "seconds" | "minutes"; sessionTarget: "main" | "isolated" | "current" | `session:${string}`; wakeMode: "next-heartbeat" | "now"; - payloadKind: "systemEvent" | "agentTurn" | "command" | "script"; + // "heartbeat" is system-owned and always payloadLocked; the form only + // displays it, never submits it. + payloadKind: "systemEvent" | "agentTurn" | "command" | "script" | "heartbeat"; payloadLocked: boolean; payloadText: string; payloadModel: string; @@ -93,6 +95,9 @@ function isCronPayload(value: unknown): value is CronPayload { if (value.kind === "script") { return typeof value.script === "string"; } + if (value.kind === "heartbeat") { + return true; + } return false; } @@ -749,7 +754,7 @@ function parseStaggerSchedule( } function isReadOnlyCronPayload(payload: CronPayload | null): boolean { - return payload?.kind === "command" || payload?.kind === "script"; + return payload?.kind === "command" || payload?.kind === "script" || payload?.kind === "heartbeat"; } function jobToForm(job: CronJob, prev: CronFormState): CronFormState { diff --git a/ui/src/lib/presenter.ts b/ui/src/lib/presenter.ts index d0937a9931dd..f04b09754678 100644 --- a/ui/src/lib/presenter.ts +++ b/ui/src/lib/presenter.ts @@ -80,6 +80,9 @@ export function formatCronPayload(job: CronJob) { if (p.kind === "script") { return `Script: ${p.script}`; } + if (p.kind === "heartbeat") { + return "Heartbeat monitor"; + } const base = `Agent: ${p.message}`; const delivery = job.delivery; if (delivery && delivery.mode !== "none") {