fix(cron): defer auto-disable notifications until persistence (#118384)

* fix(cron): defer auto-disable notifications until persistence

* fix(cron): emit disable notifications only after durable roster commits

---------

Co-authored-by: Peter Steinberger <steipete@gmail.com>

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,

src/cron/service/ops-mutations.ts

@@ -504,13 +504,19 @@ export async function removeAgentJobsTransactional<T>(
     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<T>(
       }
       throw error;
     }
+    for (const notify of postPersistNotifications) {
+      notify();
+    }
     for (const job of removedJobs) {
       noteActiveCronJobRemoval(job.id);
       try {

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 };

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 });
   }
 }
 

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. */

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);
     }

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);
         }