fix(cron): roll back live scheduler state when persisting add/update/remove fails (#99960)

* fix(cron): roll back live store when persist fails during add/update/remove

* fix(cron): restore catch-up deferral markers on failed persists

* fix(cron): defer rollback-sensitive notifications

---------

Co-authored-by: Vincent Koc <vincentkoc@ieee.org>

src/cron/service/jobs.ts

@@ -492,6 +492,7 @@ export function recordScheduleComputeError(params: {
   state: CronServiceState;
   job: CronJob;
   err: unknown;
+  deferredAutoDisableNotifications?: Array<() => void>;
 }): boolean {
   const { state, job, err } = params;
   const errorCount = (job.state.scheduleErrorCount ?? 0) + 1;
@@ -508,20 +509,27 @@ export function recordScheduleComputeError(params: {
       "cron: auto-disabled job after repeated schedule errors",
     );
 
-    // Notify the user so the auto-disable is not silent (#28861).
     const notifyText = `⚠️ Cron job "${job.name}" has been auto-disabled after ${errorCount} consecutive schedule errors. Last error: ${errText}`;
-    state.deps.enqueueSystemEvent(notifyText, {
-      agentId: job.agentId,
-      sessionKey: job.sessionKey,
-      contextKey: `cron:${job.id}:auto-disabled`,
-    });
-    state.deps.requestHeartbeat({
-      source: "cron",
-      intent: "event",
-      reason: `cron:${job.id}:auto-disabled`,
-      agentId: job.agentId,
-      sessionKey: job.sessionKey,
-    });
+    const notify = () => {
+      state.deps.enqueueSystemEvent(notifyText, {
+        agentId: job.agentId,
+        sessionKey: job.sessionKey,
+        contextKey: `cron:${job.id}:auto-disabled`,
+      });
+      state.deps.requestHeartbeat({
+        source: "cron",
+        intent: "event",
+        reason: `cron:${job.id}:auto-disabled`,
+        agentId: job.agentId,
+        sessionKey: job.sessionKey,
+      });
+    };
+    if (params.deferredAutoDisableNotifications) {
+      params.deferredAutoDisableNotifications.push(notify);
+    } else {
+      // Notify the user so the auto-disable is not silent (#28861).
+      notify();
+    }
   } else {
     state.deps.log.warn(
       { jobId: job.id, name: job.name, errorCount, err: errText },
@@ -617,7 +625,12 @@ function walkSchedulableJobs(
   return changed;
 }
 
-function recomputeJobNextRunAtMs(params: { state: CronServiceState; job: CronJob; nowMs: number }) {
+function recomputeJobNextRunAtMs(params: {
+  state: CronServiceState;
+  job: CronJob;
+  nowMs: number;
+  deferredAutoDisableNotifications?: Array<() => void>;
+}) {
   let changed = false;
   try {
     let newNext = computeJobNextRunAtMs(params.job, params.nowMs);
@@ -644,7 +657,14 @@ function recomputeJobNextRunAtMs(params: { state: CronServiceState; job: CronJob
       changed = true;
     }
   } catch (err) {
-    if (recordScheduleComputeError({ state: params.state, job: params.job, err })) {
+    if (
+      recordScheduleComputeError({
+        state: params.state,
+        job: params.job,
+        err,
+        deferredAutoDisableNotifications: params.deferredAutoDisableNotifications,
+      })
+    ) {
       changed = true;
     }
   }
@@ -682,10 +702,18 @@ export function recomputeNextRunsForMaintenance(
     recomputeExpired?: boolean;
     nowMs?: number;
     repairFutureCronNextRunAtMs?: boolean;
+    deferredAutoDisableNotifications?: Array<() => void>;
   },
 ): boolean {
   const recomputeExpired = opts?.recomputeExpired ?? false;
   const repairFutureCronNextRunAtMs = opts?.repairFutureCronNextRunAtMs ?? true;
+  const recomputeJob = (job: CronJob, nowMs: number) =>
+    recomputeJobNextRunAtMs({
+      state,
+      job,
+      nowMs,
+      deferredAutoDisableNotifications: opts?.deferredAutoDisableNotifications,
+    });
   const deferralIds = state.pendingCatchupDeferralJobIds;
   // Drop deferral markers for jobs that no longer exist in the store or
   // are disabled. They will not fire, so no deferral is needed.
@@ -715,7 +743,7 @@ export function recomputeNextRunsForMaintenance(
       }
 
       if (!hasScheduledNextRunAtMs(job.state.nextRunAtMs)) {
-        if (recomputeJobNextRunAtMs({ state, job, nowMs: now })) {
+        if (recomputeJob(job, now)) {
           changed = true;
         }
       } else if (
@@ -723,7 +751,7 @@ export function recomputeNextRunsForMaintenance(
         !deferralIds.has(job.id) &&
         shouldRepairFutureCronNextRunAtMs({ state, job, nowMs: now })
       ) {
-        if (recomputeJobNextRunAtMs({ state, job, nowMs: now })) {
+        if (recomputeJob(job, now)) {
           changed = true;
         }
       } else if (
@@ -745,7 +773,7 @@ export function recomputeNextRunsForMaintenance(
           now < backoffUntilMs &&
           job.state.nextRunAtMs < backoffUntilMs;
         if (alreadyExecutedSlot || isStaleBackoffSlot) {
-          if (recomputeJobNextRunAtMs({ state, job, nowMs: now })) {
+          if (recomputeJob(job, now)) {
             changed = true;
           }
         }

src/cron/service/ops.test.ts

@@ -1,16 +1,18 @@
 // Cron service ops tests cover high-level service operations and state transitions.
 import fs from "node:fs/promises";
 import path from "node:path";
-import { describe, expect, it, vi } from "vitest";
+import { afterEach, describe, expect, it, vi } from "vitest";
 import { runOpenClawStateWriteTransaction } from "../../state/openclaw-state-db.js";
 import * as detachedTaskRuntime from "../../tasks/detached-task-runtime.js";
 import { findTaskByRunId, resetTaskRegistryForTests } from "../../tasks/task-registry.js";
 import { formatTaskStatusDetail } from "../../tasks/task-status.js";
 import { withEnvAsync } from "../../test-utils/env.js";
+import * as cronSchedule from "../schedule.js";
 import { setupCronServiceSuite, writeCronStoreSnapshot } from "../service.test-harness.js";
+import * as cronStoreModule from "../store.js";
 import { loadCronJobsStoreWithConfigJobs, loadCronStore } from "../store.js";
 import type { CronJob } from "../types.js";
-import { add, run, start, stop, update } from "./ops.js";
+import { add, list, remove, run, start, stop, update } from "./ops.js";
 import { createCronServiceState } from "./state.js";
 import { runMissedJobs } from "./timer.js";
 
@@ -827,3 +829,164 @@ describe("cron service ops seam coverage", () => {
     expect(job.state.nextRunAtMs).toBeGreaterThan(finalBaseRunAtMs);
   });
 });
+
+describe("cron service ops persist rollback", () => {
+  afterEach(() => {
+    vi.restoreAllMocks();
+  });
+
+  function makeCreateInput(name: string) {
+    return {
+      name,
+      enabled: true,
+      schedule: { kind: "cron", expr: "0 0 * * *" },
+      sessionTarget: "isolated",
+      wakeMode: "next-heartbeat",
+      payload: { kind: "agentTurn", message: "do work" },
+    } as const;
+  }
+
+  it("rolls back an added job from the live store when persist fails", async () => {
+    const { storePath } = await makeStorePath();
+    const now = Date.parse("2026-06-09T00:00:00.000Z");
+    const state = createOkIsolatedCronState({ storePath, now });
+
+    vi.spyOn(cronStoreModule, "saveCronJobsStore").mockRejectedValueOnce(new Error("disk full"));
+
+    await expect(add(state, makeCreateInput("daily cleanup"))).rejects.toThrow("disk full");
+
+    expect(state.timer).toBeNull();
+    expect(state.store?.jobs ?? []).toEqual([]);
+    const listed = await list(state, { includeDisabled: true });
+    if (state.timer) {
+      clearTimeout(state.timer);
+    }
+    expect(listed).toEqual([]);
+    const loaded = await loadCronStore(storePath);
+    expect(loaded.jobs).toEqual([]);
+  });
+
+  it("keeps the pre-update job in the live store when persist fails", 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("daily cleanup"));
+    if (state.timer) {
+      clearTimeout(state.timer);
+    }
+
+    vi.spyOn(cronStoreModule, "saveCronJobsStore").mockRejectedValueOnce(new Error("disk full"));
+
+    await expect(update(state, job.id, { name: "renamed cleanup" })).rejects.toThrow("disk full");
+
+    const inMemory = state.store?.jobs.find((entry) => entry.id === job.id);
+    expect(inMemory?.name).toBe("daily cleanup");
+    const loaded = await loadCronStore(storePath);
+    const stored = loaded.jobs.find((entry) => entry.id === job.id);
+    expect(stored?.name).toBe("daily cleanup");
+  });
+
+  it("keeps a removed job in the live store when persist fails", 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("daily cleanup"));
+    if (state.timer) {
+      clearTimeout(state.timer);
+    }
+
+    vi.spyOn(cronStoreModule, "saveCronJobsStore").mockRejectedValueOnce(new Error("disk full"));
+
+    await expect(remove(state, job.id)).rejects.toThrow("disk full");
+
+    expect(state.store?.jobs.map((entry) => entry.id)).toEqual([job.id]);
+    const loaded = await loadCronStore(storePath);
+    expect(loaded.jobs.map((entry) => entry.id)).toEqual([job.id]);
+  });
+
+  it("keeps a job's catch-up deferral marker when a remove persist fails", 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("daily cleanup"));
+    if (state.timer) {
+      clearTimeout(state.timer);
+    }
+    state.pendingCatchupDeferralJobIds.add(job.id);
+
+    vi.spyOn(cronStoreModule, "saveCronJobsStore").mockRejectedValueOnce(new Error("disk full"));
+
+    await expect(remove(state, job.id)).rejects.toThrow("disk full");
+
+    expect(state.pendingCatchupDeferralJobIds.has(job.id)).toBe(true);
+    expect(state.store?.jobs.map((entry) => entry.id)).toEqual([job.id]);
+  });
+
+  it("recovers after a failed persist so the next mutation succeeds", async () => {
+    const { storePath } = await makeStorePath();
+    const now = Date.parse("2026-06-09T00:00:00.000Z");
+    const state = createOkIsolatedCronState({ storePath, now });
+
+    vi.spyOn(cronStoreModule, "saveCronJobsStore").mockRejectedValueOnce(new Error("disk full"));
+    await expect(add(state, makeCreateInput("daily cleanup"))).rejects.toThrow("disk full");
+
+    const job = await add(state, makeCreateInput("daily cleanup"));
+    if (state.timer) {
+      clearTimeout(state.timer);
+    }
+
+    const listed = await list(state, { includeDisabled: true });
+    if (state.timer) {
+      clearTimeout(state.timer);
+    }
+    expect(listed.map((entry) => entry.id)).toEqual([job.id]);
+    const loaded = await loadCronStore(storePath);
+    expect(loaded.jobs.map((entry) => entry.id)).toEqual([job.id]);
+  });
+
+  it("notifies about schedule auto-disable only after the mutation persists", async () => {
+    const { storePath } = await makeStorePath();
+    const now = Date.parse("2026-06-09T00:00:00.000Z");
+    const state = createOkIsolatedCronState({ storePath, now });
+
+    const malformed = await add(state, {
+      ...makeCreateInput("malformed sibling"),
+      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);
+    });
+
+    vi.spyOn(cronStoreModule, "saveCronJobsStore").mockRejectedValueOnce(new Error("disk full"));
+    await expect(add(state, makeCreateInput("failed mutation"))).rejects.toThrow("disk full");
+
+    expect(state.store?.jobs.find((job) => job.id === malformed.id)?.enabled).toBe(true);
+    expect(enqueueSystemEvent).not.toHaveBeenCalled();
+    expect(requestHeartbeat).not.toHaveBeenCalled();
+
+    await add(state, makeCreateInput("successful mutation"));
+    if (state.timer) {
+      clearTimeout(state.timer);
+    }
+
+    expect(state.store?.jobs.find((job) => job.id === malformed.id)?.enabled).toBe(false);
+    expect(enqueueSystemEvent).toHaveBeenCalledTimes(1);
+    expect(requestHeartbeat).toHaveBeenCalledTimes(1);
+  });
+});

src/cron/service/ops.ts

@@ -19,7 +19,7 @@ import { resolveCronDeliveryPlan, resolveFailureDestination } from "../delivery-
 import { createCronRunDiagnosticsFromError } from "../run-diagnostics.js";
 import { createCronExecutionId } from "../run-id.js";
 import { cronSchedulingInputsEqual } from "../schedule-identity.js";
-import type { CronJob, CronJobCreate, CronJobPatch, CronPayload } from "../types.js";
+import type { CronJob, CronJobCreate, CronJobPatch, CronPayload, CronStoreFile } from "../types.js";
 import { normalizeCronRunErrorText } from "./execution-errors.js";
 import { failureNotificationDeliveryFromJobState } from "./failure-alerts.js";
 import {
@@ -473,17 +473,58 @@ export async function listPage(state: CronServiceState, opts?: CronListPageOptio
   });
 }
 
+type CronRollbackSnapshot = {
+  store: CronStoreFile | null;
+  pendingCatchupDeferralJobIds: Set<string>;
+};
+
+// Rolls the live scheduler state back to its pre-mutation snapshot when the
+// durable write fails. recomputeNextRunsForMaintenance mutates schedule state
+// across all jobs and drops catch-up deferral markers for removed/disabled
+// jobs, so restoring only the touched job would leave siblings and deferrals
+// ahead of disk; without the rollback a failed add/update/remove keeps
+// running, reverting, or resurrecting jobs the caller was told did not apply.
+async function persistOrRestore(
+  state: CronServiceState,
+  snapshot: CronRollbackSnapshot,
+  postPersistAutoDisableNotifications: Array<() => void> = [],
+) {
+  try {
+    await persist(state);
+  } catch (err) {
+    state.store = snapshot.store;
+    state.pendingCatchupDeferralJobIds = snapshot.pendingCatchupDeferralJobIds;
+    throw err;
+  }
+  for (const notify of postPersistAutoDisableNotifications) {
+    notify();
+  }
+}
+
+function snapshotStoreForRollback(state: CronServiceState): CronRollbackSnapshot {
+  return {
+    store: state.store ? structuredClone(state.store) : null,
+    pendingCatchupDeferralJobIds: new Set(state.pendingCatchupDeferralJobIds),
+  };
+}
+
 /** Adds a cron job, recomputes scheduler state, persists, and re-arms the timer. */
 export async function add(state: CronServiceState, input: CronJobCreate) {
   return await locked(state, async () => {
     warnIfDisabled(state, "add");
     await ensureLoaded(state, { skipRecompute: true });
+    const snapshot = snapshotStoreForRollback(state);
     const job = createJob(state, input);
     state.store?.jobs.push(job);
 
-    recomputeNextRunsForMaintenance(state);
+    // Auto-disable notifications describe durable state, so publish them only
+    // after the write succeeds instead of leaking a rolled-back transition.
+    const postPersistAutoDisableNotifications: Array<() => void> = [];
+    recomputeNextRunsForMaintenance(state, {
+      deferredAutoDisableNotifications: postPersistAutoDisableNotifications,
+    });
 
-    await persist(state);
+    await persistOrRestore(state, snapshot, postPersistAutoDisableNotifications);
     armTimer(state);
 
     state.deps.log.info(
@@ -513,6 +554,7 @@ export async function update(state: CronServiceState, id: string, patch: CronJob
   return await locked(state, async () => {
     warnIfDisabled(state, "update");
     await ensureLoaded(state, { skipRecompute: true });
+    const snapshot = snapshotStoreForRollback(state);
     const job = findJobOrThrow(state, id);
     const now = state.deps.nowMs();
     const nextJob = structuredClone(job);
@@ -581,7 +623,7 @@ export async function update(state: CronServiceState, id: string, patch: CronJob
       }
     }
 
-    await persist(state);
+    await persistOrRestore(state, snapshot);
     armTimer(state);
     emit(state, {
       jobId: id,
@@ -602,13 +644,17 @@ export async function remove(state: CronServiceState, id: string) {
     if (!state.store) {
       return { ok: false, removed: false } as const;
     }
+    const snapshot = snapshotStoreForRollback(state);
     const removedJob = state.store.jobs.find((j) => j.id === id);
     state.store.jobs = state.store.jobs.filter((j) => j.id !== id);
     const removed = (state.store.jobs.length ?? 0) !== before;
 
-    recomputeNextRunsForMaintenance(state);
+    const postPersistAutoDisableNotifications: Array<() => void> = [];
+    recomputeNextRunsForMaintenance(state, {
+      deferredAutoDisableNotifications: postPersistAutoDisableNotifications,
+    });
 
-    await persist(state);
+    await persistOrRestore(state, snapshot, postPersistAutoDisableNotifications);
     armTimer(state);
     if (removed) {
       emit(state, { jobId: id, action: "removed", job: removedJob });