diff --git a/src/cron/service/jobs-scheduling.ts b/src/cron/service/jobs-scheduling.ts index 9c277011a638..f5c716de9229 100644 --- a/src/cron/service/jobs-scheduling.ts +++ b/src/cron/service/jobs-scheduling.ts @@ -535,6 +535,7 @@ function recomputeJobNextRunAtMs(params: { job: CronJob; nowMs: number; deferredNotifications?: DeferredCronNotifications; + skipScheduleErrorHandling?: boolean; }) { let changed = false; try { @@ -562,6 +563,9 @@ function recomputeJobNextRunAtMs(params: { changed = true; } } catch (err) { + if (params.skipScheduleErrorHandling) { + return false; + } if ( recordScheduleComputeError({ state: params.state, @@ -611,6 +615,7 @@ export function recomputeNextRunsForMaintenance( repairFutureCronNextRunAtMs?: boolean; preserveExpiredPacedNextRunJobId?: string; deferredNotifications?: DeferredCronNotifications; + skipScheduleErrorHandling?: boolean; }, ): boolean { const recomputeExpired = opts?.recomputeExpired ?? false; @@ -621,6 +626,7 @@ export function recomputeNextRunsForMaintenance( job, nowMs, deferredNotifications: opts?.deferredNotifications, + skipScheduleErrorHandling: opts?.skipScheduleErrorHandling, }); return walkSchedulableJobs( state, diff --git a/src/cron/service/ops-mutations.ts b/src/cron/service/ops-mutations.ts index bfabd0c00bda..93c0d8e62231 100644 --- a/src/cron/service/ops-mutations.ts +++ b/src/cron/service/ops-mutations.ts @@ -504,13 +504,19 @@ export async function removeAgentJobsTransactional( state.store.jobs = state.store.jobs.filter( (job) => resolveEffectiveJobAgentId(job, defaultAgentId) !== id, ); - recomputeNextRunsForMaintenance(state); + const postPersistNotifications: DeferredCronNotifications = []; + recomputeNextRunsForMaintenance(state, { deferredNotifications: postPersistNotifications }); + // Cron is durable first, but notifications stay speculative until the roster commits. await persistOrRestore(state, snapshot); let result: T; try { result = await commit(); } catch (error) { if (error instanceof AgentDeletionCommitUncertainError) { + // Uncertain roster writes intentionally keep the cron deletion durable. + for (const notify of postPersistNotifications) { + notify(); + } armTimer(state); for (const job of removedJobs) { noteActiveCronJobRemoval(job.id); @@ -534,6 +540,9 @@ export async function removeAgentJobsTransactional( } throw error; } + for (const notify of postPersistNotifications) { + notify(); + } for (const job of removedJobs) { noteActiveCronJobRemoval(job.id); try { diff --git a/src/cron/service/ops-run-preparation.ts b/src/cron/service/ops-run-preparation.ts index 630eec278d43..df9473eacd31 100644 --- a/src/cron/service/ops-run-preparation.ts +++ b/src/cron/service/ops-run-preparation.ts @@ -207,6 +207,15 @@ async function skipInvalidPersistedManualRun(params: { armTimer(params.state); } +function recomputeManualRunPreflight(state: CronServiceState, id: string, mode?: "due" | "force") { + // Preflight is advisory and may be called by read-shaped queue checks. Do not + // let a schedule error turn that check into an auto-disable transition. + return recomputeNextRunsForMaintenance(state, { + ...(mode === "force" ? { preserveExpiredPacedNextRunJobId: id } : {}), + skipScheduleErrorHandling: true, + }); +} + async function inspectManualRunPreflight( state: CronServiceState, id: string, @@ -228,10 +237,7 @@ async function inspectManualRunPreflight( // Normalize job tick state (clears stale runningAtMs markers) before // checking if already running, so a stale marker from a crashed Phase-1 // persist does not block manual triggers for up to STUCK_RUN_MS (#17554). - recomputeNextRunsForMaintenance( - state, - mode === "force" ? { preserveExpiredPacedNextRunJobId: id } : undefined, - ); + recomputeManualRunPreflight(state, id, mode); const job = findJobOrThrow(state, id); if (!admitsStreamSourceRun(job, streamScheduleKey, streamSourceIdentity)) { return { ok: true, ran: false, reason: "not-due" } as const; @@ -308,10 +314,7 @@ export async function prepareManualRun( // The initial preflight is advisory. A command-lane wait or another cron // run can change this job before its reservation is persisted. await ensureLoaded(state, { skipRecompute: true }); - recomputeNextRunsForMaintenance( - state, - mode === "force" ? { preserveExpiredPacedNextRunJobId: id } : undefined, - ); + recomputeManualRunPreflight(state, id, mode); const job = findJobOrThrow(state, id); if (!admitsStreamSourceRun(job, opts?.streamScheduleKey, opts?.streamSourceIdentity)) { return { ok: true, ran: false, reason: "not-due" as const }; diff --git a/src/cron/service/ops-shared.ts b/src/cron/service/ops-shared.ts index af39cb929e6f..40641f5523f9 100644 --- a/src/cron/service/ops-shared.ts +++ b/src/cron/service/ops-shared.ts @@ -5,8 +5,8 @@ import { cronStreamScheduleKey } from "../stream-schedule.js"; import type { CronJob } from "../types.js"; import { recomputeNextRunsForMaintenance } from "./jobs.js"; import { normalizeOptionalAgentId } from "./normalize.js"; -import type { CronServiceState } from "./state.js"; -import { ensureLoaded, persist } from "./store.js"; +import type { CronServiceState, DeferredCronNotifications } from "./state.js"; +import { ensureLoaded, persistOrRestore, snapshotStoreForRollback } from "./store.js"; import { type IsolatedAgentSetupTimeoutSignal, maybeNotifyIsolatedAgentSetupTimeout, @@ -74,9 +74,13 @@ export async function ensureLoadedForRead(state: CronServiceState) { } // Use the maintenance-only version so that read-only operations never // advance a past-due nextRunAtMs without executing the job (#16156). - const changed = recomputeNextRunsForMaintenance(state); + const rollbackSnapshot = snapshotStoreForRollback(state); + const postPersistNotifications: DeferredCronNotifications = []; + const changed = recomputeNextRunsForMaintenance(state, { + deferredNotifications: postPersistNotifications, + }); if (changed) { - await persist(state); + await persistOrRestore(state, rollbackSnapshot, { postPersistNotifications }); } } diff --git a/src/cron/service/ops.test.ts b/src/cron/service/ops.test.ts index d0ce61aa3f79..0c37b5915555 100644 --- a/src/cron/service/ops.test.ts +++ b/src/cron/service/ops.test.ts @@ -2,6 +2,7 @@ import fs from "node:fs/promises"; import path from "node:path"; import { afterEach, describe, expect, it, vi } from "vitest"; +import { AgentDeletionCommitUncertainError } from "../../agents/agent-lifecycle-registry.js"; import { runOpenClawStateWriteTransaction } from "../../state/openclaw-state-db.js"; import * as taskExecutor from "../../tasks/task-executor.js"; import { findTaskByRunId, listTaskRecordsUnsorted } from "../../tasks/task-registry.js"; @@ -16,8 +17,15 @@ import { loadCronJobsStoreWithConfigJobs, loadCronStore } from "../store.js"; import { cronStoreKey } from "../store/key.js"; import type { CronJob } from "../types.js"; import { start, stop } from "./ops-lifecycle.js"; -import { add, remove, removeStaleJobFamily, update } from "./ops-mutations.js"; +import { + add, + remove, + removeAgentJobsTransactional, + removeStaleJobFamily, + update, +} from "./ops-mutations.js"; import { list } from "./ops-read.js"; +import { inspectManualRunDisposition } from "./ops-run-preparation.js"; import { run } from "./ops-run.js"; import { createCronServiceState, type CronEvent } from "./state.js"; import { tryCreateCronTaskRun, tryFinishCronTaskRun } from "./task-runs.js"; @@ -1738,5 +1746,112 @@ describe("cron service ops persist rollback", () => { expect(enqueueSystemEvent).toHaveBeenCalledTimes(1); expect(requestHeartbeat).toHaveBeenCalledTimes(1); }); + + it.each(["failed", "committed", "uncertain"] as const)( + "publishes agent-removal auto-disable notifications only after a %s roster outcome", + async (outcome) => { + const { storePath } = await makeStorePath(); + const now = Date.parse("2026-06-09T00:00:00.000Z"); + const state = createOkIsolatedCronState({ storePath, now }); + const removed = await add(state, { + ...makeCreateInput("deleted agent job"), + agentId: "doomed", + }); + const malformed = await add(state, { + ...makeCreateInput("malformed surviving job"), + agentId: "survivor", + schedule: { kind: "cron", expr: "0 1 * * *" }, + }); + if (state.timer) { + clearTimeout(state.timer); + } + malformed.state.nextRunAtMs = undefined; + malformed.state.scheduleErrorCount = 2; + const enqueueSystemEvent = vi.mocked(state.deps.enqueueSystemEvent); + const requestHeartbeat = vi.mocked(state.deps.requestHeartbeat); + enqueueSystemEvent.mockClear(); + requestHeartbeat.mockClear(); + const computeNextRunAtMs = cronSchedule.computeNextRunAtMs; + vi.spyOn(cronSchedule, "computeNextRunAtMs").mockImplementation((schedule, nowMs) => { + if (schedule.kind === "cron" && schedule.expr === "0 1 * * *") { + throw new Error("simulated schedule failure"); + } + return computeNextRunAtMs(schedule, nowMs); + }); + + const commit = vi.fn(async () => { + expect(enqueueSystemEvent).not.toHaveBeenCalled(); + expect(requestHeartbeat).not.toHaveBeenCalled(); + const persisted = await loadCronStore(storePath); + expect(persisted.jobs.find((job) => job.id === removed.id)).toBeUndefined(); + expect(persisted.jobs.find((job) => job.id === malformed.id)?.enabled).toBe(false); + if (outcome === "failed") { + throw new Error("roster commit failed"); + } + if (outcome === "uncertain") { + throw new AgentDeletionCommitUncertainError(new Error("roster commit uncertain")); + } + return "roster committed"; + }); + const transaction = removeAgentJobsTransactional(state, "doomed", commit); + if (outcome === "committed") { + await expect(transaction).resolves.toBe("roster committed"); + } else if (outcome === "uncertain") { + await expect(transaction).rejects.toBeInstanceOf(AgentDeletionCommitUncertainError); + } else { + await expect(transaction).rejects.toThrow("roster commit failed"); + } + if (state.timer) { + clearTimeout(state.timer); + } + + const rolledBack = outcome === "failed"; + const notificationCount = rolledBack ? 0 : 1; + expect(commit).toHaveBeenCalledOnce(); + expect(enqueueSystemEvent).toHaveBeenCalledTimes(notificationCount); + expect(requestHeartbeat).toHaveBeenCalledTimes(notificationCount); + expect(state.store?.jobs.some((job) => job.id === removed.id)).toBe(rolledBack); + expect(state.store?.jobs.find((job) => job.id === malformed.id)?.enabled).toBe(rolledBack); + const persisted = await loadCronStore(storePath); + expect(persisted.jobs.some((job) => job.id === removed.id)).toBe(rolledBack); + expect(persisted.jobs.find((job) => job.id === malformed.id)?.enabled).toBe(rolledBack); + }, + ); + + it("does not auto-disable a job during manual-run preflight", async () => { + const { storePath } = await makeStorePath(); + const now = Date.parse("2026-06-09T00:00:00.000Z"); + const state = createOkIsolatedCronState({ storePath, now }); + const job = await add(state, { + ...makeCreateInput("preflight schedule failure"), + schedule: { kind: "cron", expr: "0 1 * * *" }, + }); + if (state.timer) { + clearTimeout(state.timer); + } + job.state.nextRunAtMs = undefined; + job.state.scheduleErrorCount = 2; + const enqueueSystemEvent = vi.mocked(state.deps.enqueueSystemEvent); + const requestHeartbeat = vi.mocked(state.deps.requestHeartbeat); + enqueueSystemEvent.mockClear(); + requestHeartbeat.mockClear(); + const computeSpy = vi.spyOn(cronSchedule, "computeNextRunAtMs").mockImplementation(() => { + throw new Error("simulated preflight schedule failure"); + }); + + try { + await expect(inspectManualRunDisposition(state, job.id)).resolves.toEqual({ + ok: true, + ran: false, + reason: "not-due", + }); + expect(job.enabled).toBe(true); + expect(job.state.scheduleErrorCount).toBe(2); + expect(enqueueSystemEvent).not.toHaveBeenCalled(); + expect(requestHeartbeat).not.toHaveBeenCalled(); + } finally { + computeSpy.mockRestore(); + } + }); }); /* oxlint-disable max-lines -- TODO: split this grandfathered oversized file. */ diff --git a/src/cron/service/timer-catchup.ts b/src/cron/service/timer-catchup.ts index 5e8770b2a006..8f3fb0cbd0d2 100644 --- a/src/cron/service/timer-catchup.ts +++ b/src/cron/service/timer-catchup.ts @@ -15,7 +15,7 @@ import { runWithCronAdmission, updateQueuedCronRunReservationMarker, } from "./run-admission.js"; -import { type CronServiceState, emit } from "./state.js"; +import { type CronServiceState, type DeferredCronNotifications, emit } from "./state.js"; import { ensureLoaded, persist, persistOrRestore, snapshotStoreForRollback } from "./store.js"; import { tryCreateCronTaskRun } from "./task-runs.js"; import { @@ -88,8 +88,12 @@ async function releaseStartupCatchupReservationsAfterFailure( if (pendingReleases.length === 0) { return; } - recomputeNextRunsForMaintenance(state, { repairFutureCronNextRunAtMs: false }); - await persistOrRestore(state, rollbackSnapshot); + const postPersistNotifications: DeferredCronNotifications = []; + recomputeNextRunsForMaintenance(state, { + repairFutureCronNextRunAtMs: false, + deferredNotifications: postPersistNotifications, + }); + await persistOrRestore(state, rollbackSnapshot, { postPersistNotifications }); for (const pending of pendingReleases) { releaseQueuedCronRun(state, pending.jobId, pending.reservationIdentity); } @@ -309,8 +313,12 @@ async function executeStartupCatchupPlan( ) { const rollbackSnapshot = snapshotStoreForRollback(state); delete job.state.queuedAtMs; - recomputeNextRunsForMaintenance(state, { repairFutureCronNextRunAtMs: false }); - await persistOrRestore(state, rollbackSnapshot); + const postPersistNotifications: DeferredCronNotifications = []; + recomputeNextRunsForMaintenance(state, { + repairFutureCronNextRunAtMs: false, + deferredNotifications: postPersistNotifications, + }); + await persistOrRestore(state, rollbackSnapshot, { postPersistNotifications }); releaseQueuedCronRun(state, candidate.jobId, candidate.reservationIdentity); return undefined; } @@ -437,8 +445,12 @@ async function applyStartupCatchupOutcomes( const rollbackSnapshot = snapshotStoreForRollback(state); const pendingReleases = clearUnstartedStartupCatchupReservationMarkers(state, plan, outcomes); if (pendingReleases.length > 0) { - recomputeNextRunsForMaintenance(state, { repairFutureCronNextRunAtMs: false }); - await persistOrRestore(state, rollbackSnapshot); + const postPersistNotifications: DeferredCronNotifications = []; + recomputeNextRunsForMaintenance(state, { + repairFutureCronNextRunAtMs: false, + deferredNotifications: postPersistNotifications, + }); + await persistOrRestore(state, rollbackSnapshot, { postPersistNotifications }); for (const pending of pendingReleases) { releaseQueuedCronRun(state, pending.jobId, pending.reservationIdentity); } @@ -450,8 +462,12 @@ async function applyStartupCatchupOutcomes( const pendingReleases = clearUnstartedStartupCatchupReservationMarkers(state, plan, outcomes); if (outcomes.length === 0 && plan.deferredJobs.length === 0) { if (pendingReleases.length > 0) { - recomputeNextRunsForMaintenance(state, { repairFutureCronNextRunAtMs: false }); - await persistOrRestore(state, rollbackSnapshot); + const postPersistNotifications: DeferredCronNotifications = []; + recomputeNextRunsForMaintenance(state, { + repairFutureCronNextRunAtMs: false, + deferredNotifications: postPersistNotifications, + }); + await persistOrRestore(state, rollbackSnapshot, { postPersistNotifications }); for (const pending of pendingReleases) { releaseQueuedCronRun(state, pending.jobId, pending.reservationIdentity); } @@ -484,8 +500,12 @@ async function applyStartupCatchupOutcomes( // Startup overflow owns these staggered wake times; repairing future // schedules here would silently move a deferred run to its natural slot. - recomputeNextRunsForMaintenance(state, { repairFutureCronNextRunAtMs: false }); - await persistOrRestore(state, rollbackSnapshot); + const postPersistNotifications: DeferredCronNotifications = []; + recomputeNextRunsForMaintenance(state, { + repairFutureCronNextRunAtMs: false, + deferredNotifications: postPersistNotifications, + }); + await persistOrRestore(state, rollbackSnapshot, { postPersistNotifications }); for (const pending of pendingReleases) { releaseQueuedCronRun(state, pending.jobId, pending.reservationIdentity); } diff --git a/src/cron/service/timer-scheduler.ts b/src/cron/service/timer-scheduler.ts index 140b55e79746..e51e7c4b8f56 100644 --- a/src/cron/service/timer-scheduler.ts +++ b/src/cron/service/timer-scheduler.ts @@ -28,8 +28,8 @@ import { runWithCronAdmission, updateQueuedCronRunReservationMarker, } from "./run-admission.js"; -import { type CronServiceState, emit } from "./state.js"; -import { ensureLoaded, persist, persistOrRestore, snapshotStoreForRollback } from "./store.js"; +import { type CronServiceState, type DeferredCronNotifications, emit } from "./state.js"; +import { ensureLoaded, persistOrRestore, snapshotStoreForRollback } from "./store.js"; import { tryCreateCronTaskRun } from "./task-runs.js"; import { resolveCronJobTimeoutMs } from "./timeout-policy.js"; import { @@ -208,12 +208,15 @@ async function onAdmittedTimer(state: CronServiceState) { // Use maintenance-only recompute to avoid advancing past-due nextRunAtMs // values without execution. This prevents jobs from being silently skipped // when the timer wakes up but findDueJobs returns empty (see #13992). + const rollbackSnapshot = snapshotStoreForRollback(state); + const postPersistNotifications: DeferredCronNotifications = []; const changed = recomputeNextRunsForMaintenance(state, { recomputeExpired: true, nowMs: dueCheckNow, + deferredNotifications: postPersistNotifications, }); if (changed) { - await persist(state); + await persistOrRestore(state, rollbackSnapshot, { postPersistNotifications }); } return []; } @@ -262,8 +265,11 @@ async function onAdmittedTimer(state: CronServiceState) { releaseQueuedCronRun(state, candidate.id, candidate.reservationIdentity); } } - recomputeNextRunsForMaintenance(state); - await persistOrRestore(state, rollbackSnapshot); + const postPersistNotifications: DeferredCronNotifications = []; + recomputeNextRunsForMaintenance(state, { + deferredNotifications: postPersistNotifications, + }); + await persistOrRestore(state, rollbackSnapshot, { postPersistNotifications }); for (const candidate of pendingReleases) { releaseQueuedCronRun(state, candidate.id, candidate.reservationIdentity); } @@ -380,8 +386,9 @@ async function onAdmittedTimer(state: CronServiceState) { releaseQueuedCronRun(state, due.id, due.reservationIdentity); } } - recomputeNextRunsForMaintenance(state); - await persistOrRestore(state, rollbackSnapshot); + const postPersistNotifications: DeferredCronNotifications = []; + recomputeNextRunsForMaintenance(state, { deferredNotifications: postPersistNotifications }); + await persistOrRestore(state, rollbackSnapshot, { postPersistNotifications }); for (const due of pendingReleases) { releaseQueuedCronRun(state, due.id, due.reservationIdentity); }