fix(cron): preserve watched commands across gateway hot reload (#115761)

Co-authored-by: Andy Ye <35905412+TurboTheTurtle@users.noreply.github.com>

src/gateway/cron-exit-watchers.test.ts

@@ -146,6 +146,92 @@ describe("createCronExitWatchers", () => {
     expect(order).toEqual(["persist", "fire"]);
   });
 
+  it("rebinds live watchers but drains callbacks already owned by the previous scheduler", async () => {
+    const { supervisor, runs } = makeFakeSupervisor();
+    let releasePersistence = () => {};
+    const persistenceGate = new Promise<void>((resolve) => {
+      releasePersistence = resolve;
+    });
+    const releaseCompletion = vi.fn();
+    const oldPersistCompletion = vi.fn(async () => {
+      await persistenceGate;
+      return releaseCompletion;
+    });
+    const oldFireOnExit = vi.fn(async () => {});
+    const newPersistCompletion = vi.fn(async () => {});
+    const newFireOnExit = vi.fn(async () => {});
+    const watchers = createCronExitWatchers({
+      getProcessSupervisor: () => supervisor as never,
+      persistCompletion: oldPersistCompletion,
+      fireOnExit: oldFireOnExit,
+      logger: noopLogger,
+    });
+
+    watchers.reconcile([onExitJob("old-owner"), onExitJob("new-owner")]);
+    await flush();
+    expectDefined(runs[0], "runs[0] test invariant").deferred.resolve({
+      exitCode: 0,
+      reason: "exit",
+    });
+    await vi.waitFor(() => expect(oldPersistCompletion).toHaveBeenCalledOnce());
+
+    const handoff = watchers.updateHandlers({
+      getProcessSupervisor: () => supervisor as never,
+      persistCompletion: newPersistCompletion,
+      fireOnExit: newFireOnExit,
+      logger: noopLogger,
+    });
+    const handoffSettled = vi.fn();
+    void handoff?.then(handoffSettled);
+
+    expectDefined(runs[1], "runs[1] test invariant").deferred.resolve({
+      exitCode: 0,
+      reason: "exit",
+    });
+    await vi.waitFor(() => expect(newFireOnExit).toHaveBeenCalledOnce());
+    expect(newPersistCompletion).toHaveBeenCalledOnce();
+    expect(oldFireOnExit).not.toHaveBeenCalled();
+    expect(handoffSettled).not.toHaveBeenCalled();
+
+    releasePersistence();
+    await handoff;
+    await vi.waitFor(() => expect(oldFireOnExit).toHaveBeenCalledOnce());
+    expect(releaseCompletion).toHaveBeenCalledOnce();
+    expect(supervisor.spawn).toHaveBeenCalledTimes(2);
+    expect(supervisor.cancelScope).not.toHaveBeenCalled();
+  });
+
+  it("routes post-handoff retries through the replacement scheduler owner", async () => {
+    const { supervisor, runs } = makeFakeSupervisor();
+    const oldUpdateWatcherState = vi.fn(async () => {});
+    const newUpdateWatcherState = vi.fn(async () => {});
+    const watchers = createCronExitWatchers({
+      getProcessSupervisor: () => supervisor as never,
+      persistCompletion: vi.fn(async () => {}),
+      fireOnExit: vi.fn(async () => {}),
+      updateWatcherState: oldUpdateWatcherState,
+      logger: noopLogger,
+      retryBackoffMs: [0],
+    });
+
+    watchers.reconcile([onExitJob("job-a")]);
+    await flush();
+    await watchers.updateHandlers({
+      getProcessSupervisor: () => supervisor as never,
+      persistCompletion: vi.fn(async () => {}),
+      fireOnExit: vi.fn(async () => {}),
+      updateWatcherState: newUpdateWatcherState,
+      logger: noopLogger,
+    });
+    expectDefined(runs[0], "runs[0] test invariant").deferred.reject(new Error("wait failed"));
+
+    await vi.waitFor(() => expect(newUpdateWatcherState).toHaveBeenCalledOnce());
+    expect(oldUpdateWatcherState).not.toHaveBeenCalled();
+    await delay(5);
+    await flush();
+    expect(supervisor.spawn).toHaveBeenCalledTimes(2);
+  });
+
   it("a fired job stays unarmed across a simulated restart (disabled in store → not re-run)", async () => {
     // persistCompletion disables the job; after a restart the reconcile sees a
     // disabled job and must NOT re-arm (which would re-run the command).

src/gateway/cron-exit-watchers.ts

@@ -29,11 +29,23 @@ export type CronExitResult = {
   noOutputTimedOut: boolean;
 };
 
-type CronExitWatchers = {
+export type CronExitWatcherHandlers = {
+  getProcessSupervisor: () => ProcessSupervisor;
+  persistCompletion: (job: OnExitCronJob) => Promise<(() => void) | void>;
+  fireOnExit: (job: CronJob, exit: CronExitResult) => void | Promise<void>;
+  updateWatcherState?: (
+    job: OnExitCronJob,
+    patch: Pick<CronJob["state"], "lastError" | "consecutiveErrors">,
+  ) => Promise<CronJob | void>;
+  logger: Logger;
+};
+
+export type CronExitWatchers = {
   reconcile: (jobs: CronJob[]) => void;
   cancel: (jobId: string) => void;
   cancelAll: () => Promise<void>;
   activeJobIds: () => string[];
+  updateHandlers: (handlers: CronExitWatcherHandlers) => Promise<void> | void;
 };
 
 const SCOPE_PREFIX = "cron-exit";
@@ -46,18 +58,26 @@ function isWatchableExitJob(job: CronJob): job is OnExitCronJob {
   return job.enabled && job.schedule.kind === "on-exit";
 }
 
-export function createCronExitWatchers(params: {
-  getProcessSupervisor: () => ProcessSupervisor;
-  persistCompletion: (job: OnExitCronJob) => Promise<(() => void) | void>;
-  fireOnExit: (job: CronJob, exit: CronExitResult) => void | Promise<void>;
-  updateWatcherState?: (
-    job: OnExitCronJob,
-    patch: Pick<CronJob["state"], "lastError" | "consecutiveErrors">,
-  ) => Promise<CronJob | void>;
-  logger: Logger;
-  shell?: { command: string; argsFor: (command: string) => string[] };
-  retryBackoffMs?: readonly number[];
-}): CronExitWatchers {
+export function createCronExitWatchers(
+  params: CronExitWatcherHandlers & {
+    shell?: { command: string; argsFor: (command: string) => string[] };
+    retryBackoffMs?: readonly number[];
+  },
+): CronExitWatchers {
+  let handlers: CronExitWatcherHandlers = params;
+  const ownerSettlements = new Set<Promise<void>>();
+  const settleOwnerCallback = async <T>(operation: Promise<T>): Promise<T> => {
+    const settlement = operation.then(
+      () => undefined,
+      () => undefined,
+    );
+    ownerSettlements.add(settlement);
+    try {
+      return await operation;
+    } finally {
+      ownerSettlements.delete(settlement);
+    }
+  };
   const shell = params.shell ?? resolveExitWatchShell();
   const retryBackoffMs =
     params.retryBackoffMs && params.retryBackoffMs.length > 0
@@ -110,9 +130,9 @@ export function createCronExitWatchers(params: {
     // killed by the arm() ownership check once it resolves.
     slot.run?.cancel("manual-cancel");
     try {
-      params.getProcessSupervisor().cancelScope(scopeKey(jobId), "manual-cancel");
+      handlers.getProcessSupervisor().cancelScope(scopeKey(jobId), "manual-cancel");
     } catch (err) {
-      params.logger.warn({ err: String(err), jobId }, "cron-exit: cancel watcher failed");
+      handlers.logger.warn({ err: String(err), jobId }, "cron-exit: cancel watcher failed");
     }
   };
 
@@ -141,16 +161,17 @@ export function createCronExitWatchers(params: {
     const persistWatcherState = async (
       patch: Pick<CronJob["state"], "lastError" | "consecutiveErrors">,
     ) => {
-      if (!params.updateWatcherState) {
+      const owner = handlers;
+      if (!owner.updateWatcherState) {
         return;
       }
       try {
-        const updated = await params.updateWatcherState(slot.job, patch);
+        const updated = await settleOwnerCallback(owner.updateWatcherState(slot.job, patch));
         if (owns() && updated && isWatchableExitJob(updated)) {
           slot.job = updated;
         }
       } catch (err) {
-        params.logger.warn(
+        owner.logger.warn(
           { err: String(err), jobId: slot.job.id },
           "cron-exit: failed to persist watcher state",
         );
@@ -180,7 +201,7 @@ export function createCronExitWatchers(params: {
         arm(slot.job, slot.consecutiveFailures);
       }, delayMs);
       slot.retryTimer.unref?.();
-      params.logger.warn(
+      handlers.logger.warn(
         { err: String(error), jobId: slot.job.id, retryInMs: delayMs },
         `cron-exit: watcher ${phase} failed; retry scheduled`,
       );
@@ -188,7 +209,7 @@ export function createCronExitWatchers(params: {
     void (async () => {
       let run: ManagedRun;
       try {
-        run = await params.getProcessSupervisor().spawn({
+        run = await handlers.getProcessSupervisor().spawn({
           sessionId: `cron-exit:${job.id}`,
           backendId: "cron-exit-watch",
           scopeKey: scopeKey(job.id),
@@ -223,7 +244,10 @@ export function createCronExitWatchers(params: {
         return;
       }
       slot.run = run;
-      params.logger.info({ jobId: job.id, runId: run.runId, command }, "cron-exit: watcher armed");
+      handlers.logger.info(
+        { jobId: job.id, runId: run.runId, command },
+        "cron-exit: watcher armed",
+      );
       let exit: Awaited<ReturnType<ManagedRun["wait"]>>;
       try {
         exit = await run.wait();
@@ -237,7 +261,8 @@ export function createCronExitWatchers(params: {
       if (!owns()) {
         return;
       }
-      params.logger.info(
+      const owner = handlers;
+      owner.logger.info(
         { jobId: job.id, exitCode: exit.exitCode, reason: exit.reason },
         "cron-exit: watched command exited; firing job",
       );
@@ -245,45 +270,52 @@ export function createCronExitWatchers(params: {
       // Persist the terminal one-shot state BEFORE firing. FAIL CLOSED: if the
       // store write fails we do NOT wake — waking without a persisted terminal
       // state would let a gateway restart re-arm and re-run the command.
-      let releaseCompletion: (() => void) | void;
       try {
-        releaseCompletion = await params.persistCompletion(slot.job);
-      } catch (err) {
-        if (owns()) {
-          active.delete(job.id);
-        }
-        params.logger.warn(
-          { err: String(err), jobId: job.id },
-          "cron-exit: persistCompletion failed; NOT firing (fail closed to avoid replay)",
+        await settleOwnerCallback(
+          (async () => {
+            let releaseCompletion: (() => void) | void;
+            try {
+              releaseCompletion = await owner.persistCompletion(slot.job);
+            } catch (err) {
+              if (owns()) {
+                active.delete(job.id);
+              }
+              owner.logger.warn(
+                { err: String(err), jobId: job.id },
+                "cron-exit: persistCompletion failed; NOT firing (fail closed to avoid replay)",
+              );
+              return;
+            }
+            try {
+              if (!owns() || slot.cancelled) {
+                if (active.get(job.id) === slot) {
+                  active.delete(job.id);
+                }
+                return;
+              }
+              slot.fired = true;
+              try {
+                await owner.fireOnExit(slot.job, {
+                  exitCode: exit.exitCode,
+                  reason: exit.reason,
+                  stdout: exit.stdout,
+                  stderr: exit.stderr,
+                  timedOut: exit.timedOut,
+                  noOutputTimedOut: exit.noOutputTimedOut,
+                });
+              } catch (err) {
+                owner.logger.warn(
+                  { err: String(err), jobId: job.id },
+                  "cron-exit: fireOnExit after exit failed",
+                );
+              }
+            } finally {
+              releaseCompletion?.();
+            }
+          })(),
         );
-        return;
-      }
-      slot.terminalPersisting = false;
-      try {
-        if (!owns() || slot.cancelled) {
-          if (active.get(job.id) === slot) {
-            active.delete(job.id);
-          }
-          return;
-        }
-        slot.fired = true;
-        try {
-          await params.fireOnExit(slot.job, {
-            exitCode: exit.exitCode,
-            reason: exit.reason,
-            stdout: exit.stdout,
-            stderr: exit.stderr,
-            timedOut: exit.timedOut,
-            noOutputTimedOut: exit.noOutputTimedOut,
-          });
-        } catch (err) {
-          params.logger.warn(
-            { err: String(err), jobId: job.id },
-            "cron-exit: fireOnExit after exit failed",
-          );
-        }
       } finally {
-        releaseCompletion?.();
+        slot.terminalPersisting = false;
       }
     })().finally(() => {
       slot.lifecycleSettled = true;
@@ -296,10 +328,23 @@ export function createCronExitWatchers(params: {
   };
 
   const reconcile = (jobs: CronJob[]) => {
+    const jobsById = new Map(jobs.map((job) => [job.id, job] as const));
     const want = new Map(jobs.filter(isWatchableExitJob).map((j) => [j.id, j] as const));
     // Cancel watchers whose job is gone or no longer watchable.
-    for (const jobId of Array.from(active.keys())) {
+    for (const [jobId, slot] of Array.from(active.entries())) {
       if (!want.has(jobId)) {
+        const storedJob = jobsById.get(jobId);
+        // A replacement can observe the terminal disable before its previous
+        // owner's completion callback has settled.
+        if (
+          slot.terminalPersisting &&
+          storedJob?.schedule.kind === "on-exit" &&
+          !storedJob.enabled &&
+          slot.command === storedJob.schedule.command &&
+          slot.cwd === storedJob.schedule.cwd
+        ) {
+          continue;
+        }
         cancel(jobId);
       }
     }
@@ -342,5 +387,14 @@ export function createCronExitWatchers(params: {
           ...Array.from(settlingCancelledSlots, (slot) => slot.job.id),
         ]),
       ),
+    updateHandlers: (nextHandlers) => {
+      handlers = nextHandlers;
+      if (ownerSettlements.size > 0) {
+        // Finish callbacks that already captured the old scheduler, but keep
+        // live watched children running under their newly adopted owner.
+        return Promise.all(ownerSettlements).then(() => undefined);
+      }
+      return undefined;
+    },
   };
 }

src/gateway/server-cron-lazy.test.ts

@@ -87,6 +87,49 @@ describe("createLazyGatewayCronState", () => {
     expect(cron["readJob"]).toHaveBeenCalledWith("demo");
   });
 
+  it("does not load cron solely to prepare a watcher handoff", async () => {
+    const lazy = createLazyGatewayCronState(createParams());
+
+    await expect(lazy.prepareExitWatcherHandoff?.()).resolves.toBeUndefined();
+    expect(hoisted.buildGatewayCronService).not.toHaveBeenCalled();
+  });
+
+  it("preserves a watcher owner when hot reload overtakes lazy startup", async () => {
+    const finishStart = deferred();
+    const cron = createCronService();
+    cron.start = vi.fn(async () => await finishStart.promise);
+    const watchers = {
+      reconcile: vi.fn(),
+      cancel: vi.fn(),
+      cancelAll: vi.fn(async () => {}),
+      activeJobIds: vi.fn(() => ["watched-job"]),
+      updateHandlers: vi.fn(),
+    };
+    const stopOwner = vi.fn(async () => {});
+    hoisted.setState({
+      ...createCronState(cron),
+      prepareExitWatcherHandoff: vi.fn(async () => ({
+        current: () => watchers,
+        adopt: vi.fn(),
+        stopOwner,
+      })),
+    });
+    const lazy = createLazyGatewayCronState(createParams());
+
+    const start = lazy.cron.start();
+    await vi.waitFor(() => expect(cron["start"]).toHaveBeenCalledOnce());
+    const handoff = await lazy.prepareExitWatcherHandoff?.();
+    expect(handoff?.current()).toBe(watchers);
+
+    await handoff?.stopOwner();
+    finishStart.resolve();
+    await start;
+
+    expect(stopOwner).toHaveBeenCalledOnce();
+    expect(watchers.cancelAll).not.toHaveBeenCalled();
+    expect(cron["stop"]).not.toHaveBeenCalled();
+  });
+
   it("forwards run payload overrides to the loaded cron service", async () => {
     const cron = createCronService();
     hoisted.setState(createCronState(cron));

src/gateway/server-cron-lazy.ts

@@ -5,7 +5,7 @@ import type { OpenClawConfig } from "../config/types.openclaw.js";
 import { resolveCronJobsStorePathFromConfig } from "../cron/store.js";
 import { createLazyPromiseLoader } from "../shared/lazy-runtime.js";
 import type { GatewayCronServiceContract } from "./server-cron-contract.js";
-import type { GatewayCronState } from "./server-cron.js";
+import type { GatewayCronExitWatcherHandoff, GatewayCronState } from "./server-cron.js";
 import type { GatewayRequestContext } from "./server-methods/types.js";
 
 type LazyGatewayCronParams = {
@@ -38,6 +38,8 @@ export function createLazyGatewayCronState(params: LazyGatewayCronParams): Gatew
   const cronEnabled = env.OPENCLAW_SKIP_CRON !== "1" && params.cfg.cron?.enabled !== false;
   let loaded: LoadedGatewayCronState | null = null;
   let stopped = false;
+  let preserveExitWatchersOnStop = false;
+  let exitWatcherHandoffStop: Promise<void> | undefined;
   let lifecycleGeneration = 0;
   let schedulingPaused = false;
   const schedulingResumeWaiters = new Set<() => void>();
@@ -86,6 +88,48 @@ export function createLazyGatewayCronState(params: LazyGatewayCronParams): Gatew
     return await cronStateLoader.load();
   };
 
+  const stopResolvedCron = async (
+    resolved: LoadedGatewayCronState,
+    preserveExitWatchers: boolean,
+  ): Promise<void> => {
+    resolved.phase = "stopped";
+    resolved.underlyingStarted = false;
+    const handoff = preserveExitWatchers
+      ? await resolved.state.prepareExitWatcherHandoff?.()
+      : undefined;
+    if (handoff) {
+      await handoff.stopOwner();
+    } else if (resolved.state.cron.stopAndDrain) {
+      await resolved.state.cron.stopAndDrain();
+    } else {
+      resolved.state.cron.stop();
+      await resolved.state.stopStreamWatchers();
+    }
+  };
+
+  const stopResolvedCronOnce = (
+    resolved: LoadedGatewayCronState,
+    preserveExitWatchers: boolean,
+  ): Promise<void> => {
+    if (!preserveExitWatchers) {
+      return stopResolvedCron(resolved, false);
+    }
+    exitWatcherHandoffStop ??= stopResolvedCron(resolved, true);
+    return exitWatcherHandoffStop;
+  };
+
+  const stopLoadedCronAndDrain = async (preserveExitWatchers = false): Promise<void> => {
+    stopped = true;
+    preserveExitWatchersOnStop ||= preserveExitWatchers;
+    lifecycleGeneration += 1;
+    releaseSchedulingResumeWaiters();
+    const loading = cronStateLoader.peek();
+    const resolved = loaded ?? (loading ? await loading : null);
+    if (resolved) {
+      await stopResolvedCronOnce(resolved, preserveExitWatchersOnStop);
+    }
+  };
+
   const cron: GatewayCronServiceContract = {
     async start() {
       stopped = false;
@@ -140,10 +184,7 @@ export function createLazyGatewayCronState(params: LazyGatewayCronParams): Gatew
           resolved.underlyingStartInFlight = false;
         }
         if (startCancelled()) {
-          resolved.phase = "stopped";
-          resolved.underlyingStarted = false;
-          resolved.state.cron.stop();
-          await resolved.state.stopStreamWatchers();
+          await stopResolvedCronOnce(resolved, preserveExitWatchersOnStop);
           return;
         }
         if (schedulingPaused) {
@@ -164,10 +205,7 @@ export function createLazyGatewayCronState(params: LazyGatewayCronParams): Gatew
           throw err;
         }
         if (startCancelled()) {
-          resolved.phase = "stopped";
-          resolved.underlyingStarted = false;
-          resolved.state.cron.stop();
-          await resolved.state.stopStreamWatchers();
+          await stopResolvedCronOnce(resolved, preserveExitWatchersOnStop);
           return;
         }
         resolved.phase = "started";
@@ -209,21 +247,7 @@ export function createLazyGatewayCronState(params: LazyGatewayCronParams): Gatew
       }
     },
     async stopAndDrain() {
-      stopped = true;
-      lifecycleGeneration += 1;
-      releaseSchedulingResumeWaiters();
-      const resolved = loaded ?? (cronStateLoader.peek() ? await cronStateLoader.peek() : null);
-      if (!resolved) {
-        return;
-      }
-      resolved.phase = "stopped";
-      resolved.underlyingStarted = false;
-      if (resolved.state.cron.stopAndDrain) {
-        await resolved.state.cron.stopAndDrain();
-      } else {
-        resolved.state.cron.stop();
-        await resolved.state.stopStreamWatchers();
-      }
+      await stopLoadedCronAndDrain();
     },
     pauseScheduling() {
       schedulingPaused = true;
@@ -320,6 +344,20 @@ export function createLazyGatewayCronState(params: LazyGatewayCronParams): Gatew
     cron,
     storePath,
     cronEnabled,
+    prepareExitWatcherHandoff: async (): Promise<GatewayCronExitWatcherHandoff | undefined> => {
+      const loading = cronStateLoader.peek();
+      const resolved = loaded ?? (loading ? await loading : null);
+      const handoff = await resolved?.state.prepareExitWatcherHandoff?.();
+      if (!handoff) {
+        return undefined;
+      }
+      return {
+        ...handoff,
+        stopOwner: async () => {
+          await stopLoadedCronAndDrain(true);
+        },
+      };
+    },
     // Reload rules invoke these hooks on whatever cronState is live; the lazy
     // proxy must forward every GatewayCronState member or hot reloads silently
     // no-op until a gateway restart (heartbeat cadence changes never applied).

src/gateway/server-cron.drain.test.ts

@@ -24,6 +24,7 @@ vi.mock("./cron-exit-watchers.js", async (importOriginal) => ({
     cancel: vi.fn(),
     cancelAll: cancelAllMock,
     activeJobIds: () => [],
+    updateHandlers: vi.fn(),
   }),
 }));
 

src/gateway/server-cron.test.ts

@@ -903,6 +903,79 @@ describe("buildGatewayCronService", () => {
     }
   });
 
+  it.each(["update", "updateWithPrecondition", "add"] as const)(
+    "honors an explicit %s disable while terminal persistence is settling",
+    async (mutation) => {
+      const commandExit = createDeferred<RunExit>();
+      const completionPersistCommitted = createDeferred();
+      const allowCompletionPersist = createDeferred();
+      const cancel = vi.fn();
+      const cancelScope = vi.fn();
+      const spawn = vi.fn(async () => ({
+        runId: "run-on-exit-explicit-disable",
+        startedAtMs: Date.now(),
+        cancel,
+        wait: () => commandExit.promise,
+      }));
+      getProcessSupervisorMock.mockReturnValue({ spawn, cancelScope });
+      const state = loadCronService(
+        createCronConfig(`server-cron-on-exit-explicit-disable-${mutation}`),
+      );
+      const originalUpdateWithPrecondition = state.cron.updateWithPrecondition.bind(state.cron);
+      let gateTerminalCompletion = true;
+      vi.spyOn(state.cron, "updateWithPrecondition").mockImplementation(async (...args) => {
+        if (!gateTerminalCompletion) {
+          return await originalUpdateWithPrecondition(...args);
+        }
+        gateTerminalCompletion = false;
+        const result = await originalUpdateWithPrecondition(...args);
+        completionPersistCommitted.resolve();
+        await allowCompletionPersist.promise;
+        return result;
+      });
+      const run = vi.spyOn(state.cron, "run");
+
+      try {
+        const input = {
+          name: "watch and honor explicit disable",
+          declarationKey: "agent:main:watch-and-honor-explicit-disable",
+          enabled: true,
+          schedule: { kind: "on-exit" as const, command: "true" },
+          payload: { kind: "systemEvent" as const, text: "must not fire" },
+          sessionTarget: "main" as const,
+          wakeMode: "now" as const,
+        };
+        const job = await state.cron.add(input);
+        await state.reconcileExitWatchers();
+        await vi.waitFor(() => expect(spawn).toHaveBeenCalledOnce());
+
+        commandExit.resolve(runExit());
+        await completionPersistCommitted.promise;
+
+        if (mutation === "updateWithPrecondition") {
+          await state.cron.updateWithPrecondition(job.id, { enabled: false }, () => {});
+        } else if (mutation === "add") {
+          await state.cron.add({ ...input, enabled: false }, { enabledExplicit: true });
+        } else {
+          await state.cron.update(job.id, { enabled: false });
+        }
+        expect(cancel).toHaveBeenCalledWith("manual-cancel");
+        expect(cancelScope).toHaveBeenCalledWith(`cron-exit:${job.id}`, "manual-cancel");
+        allowCompletionPersist.resolve();
+
+        await vi.waitFor(async () => {
+          const handoff = await state.prepareExitWatcherHandoff?.();
+          expect(handoff?.current().activeJobIds()).toEqual([]);
+        });
+        expect(run).not.toHaveBeenCalled();
+      } finally {
+        allowCompletionPersist.resolve();
+        commandExit.resolve(runExit());
+        await state.cron.stopAndDrain?.();
+      }
+    },
+  );
+
   it("aborts and drains active cron runs during shutdown", async () => {
     const controller = new AbortController();
     const coreRun = new Promise<void>((resolve) => {

src/gateway/server-cron.ts

@@ -81,7 +81,12 @@ import {
 } from "../routing/session-key.js";
 import { defaultRuntime } from "../runtime.js";
 import { parseAgentSessionKey } from "../sessions/session-key-utils.js";
-import { createCronExitWatchers, type CronExitResult } from "./cron-exit-watchers.js";
+import {
+  createCronExitWatchers,
+  type CronExitResult,
+  type CronExitWatcherHandlers,
+  type CronExitWatchers,
+} from "./cron-exit-watchers.js";
 import {
   createCronStreamWatchers,
   type CronStreamFireDisposition,
@@ -112,6 +117,7 @@ export type GatewayCronState = {
   cron: GatewayCronServiceContract;
   storePath: string;
   cronEnabled: boolean;
+  prepareExitWatcherHandoff?: () => Promise<GatewayCronExitWatcherHandoff | undefined>;
   // Required, not optional: reload rules call these hooks directly on whatever
   // cronState is live (including the lazy proxy). An optional member here let
   // the proxy silently omit reconcileHeartbeatJobs, turning every
@@ -123,6 +129,12 @@ export type GatewayCronState = {
   reconcileHeartbeatJobs: (cfg?: OpenClawConfig) => Promise<void>;
 };
 
+export type GatewayCronExitWatcherHandoff = {
+  current: () => CronExitWatchers;
+  adopt: (watchers: CronExitWatchers) => Promise<void> | void;
+  stopOwner: () => Promise<void>;
+};
+
 function formatOnExitRunSummary(exit: CronExitResult): string {
   const lines = [
     "Watched command finished.",
@@ -562,7 +574,10 @@ export function buildGatewayCronService(params: {
   } = { current: undefined };
   let exitWatcherReconciliations = 0;
   let streamWatcherReconciliations = 0;
-  const terminalExitCompletionTokens = new Map<string, object>();
+  const terminalExitCompletionTokens = new Map<
+    string,
+    Parameters<CronService["updateWithPrecondition"]>[2]
+  >();
   let exitWatcherGeneration = 0;
   let exitWatcherMutationRevision = 0;
   let exitWatchersStopped = false;
@@ -1136,10 +1151,14 @@ export function buildGatewayCronService(params: {
     },
   });
 
-  exitWatchersRef.current = createCronExitWatchers({
+  const exitWatcherHandlers = {
     getProcessSupervisor,
     persistCompletion: async (job) => {
-      const completionToken = {};
+      const completionToken: Parameters<CronService["updateWithPrecondition"]>[2] = (current) => {
+        if (!current.enabled || current.updatedAtMs !== job.updatedAtMs) {
+          throw new Error("cron on-exit job changed before completion");
+        }
+      };
       terminalExitCompletionTokens.set(job.id, completionToken);
       const releaseCompletionToken = () => {
         if (terminalExitCompletionTokens.get(job.id) === completionToken) {
@@ -1148,11 +1167,7 @@ export function buildGatewayCronService(params: {
       };
       try {
         await runWithGatewayIndependentRootWorkAdmission(async () => {
-          await cron.updateWithPrecondition(job.id, { enabled: false }, (current) => {
-            if (!current.enabled || current.updatedAtMs !== job.updatedAtMs) {
-              throw new Error("cron on-exit job changed before completion");
-            }
-          });
+          await cron.updateWithPrecondition(job.id, { enabled: false }, completionToken);
         });
         return () => {
           releaseCompletionToken();
@@ -1191,7 +1206,8 @@ export function buildGatewayCronService(params: {
         }
       }),
     logger: cronLogger,
-  });
+  } satisfies CronExitWatcherHandlers;
+  exitWatchersRef.current = createCronExitWatchers(exitWatcherHandlers);
   const updateCron = cron.update.bind(cron);
   streamWatchersRef.current = createCronStreamWatchers({
     getProcessSupervisor,
@@ -1270,10 +1286,21 @@ export function buildGatewayCronService(params: {
       patch.schedule !== undefined ? "schedule-update" : "disabled",
     );
   };
+  const cancelDisabledExitWatcher = (job: CronJob) => {
+    if (job.enabled || job.schedule.kind !== "on-exit") {
+      return;
+    }
+    // An operator disable wins over a completion retained during owner handoff.
+    exitWatcherMutationRevision += 1;
+    exitWatchersRef.current?.cancel(job.id);
+  };
   const addCron = cron.add.bind(cron);
   cron.add = async (input, options) => {
     const result = await addCron(input, options);
     const addedJob = "job" in result ? result.job : result;
+    if (options?.enabledExplicit && !input.enabled) {
+      cancelDisabledExitWatcher(addedJob);
+    }
     await routeCurrentStreamJob(addedJob.id, addedJob, "added");
     return result;
   };
@@ -1314,6 +1341,9 @@ export function buildGatewayCronService(params: {
     };
     try {
       const result = await updateCronWithPrecondition(jobId, patch, routeAfterValidation, opts);
+      if (patch.enabled === false) {
+        cancelDisabledExitWatcher(result);
+      }
       await settleStopAfterCommittedUpdate(jobId, lifecycleStop);
       await routeLiveStreamJobLogged(jobId);
       return result;
@@ -1333,6 +1363,9 @@ export function buildGatewayCronService(params: {
     };
     try {
       const result = await updateCronWithPrecondition(jobId, patch, routeAfterPrecondition, opts);
+      if (patch.enabled === false && terminalExitCompletionTokens.get(jobId) !== precondition) {
+        cancelDisabledExitWatcher(result);
+      }
       await settleStopAfterCommittedUpdate(jobId, lifecycleStop);
       await routeLiveStreamJobLogged(jobId);
       return result;
@@ -1408,10 +1441,17 @@ export function buildGatewayCronService(params: {
   };
   const automationEpoch = claimSessionAutomationEpoch();
   const stopCron = cron.stop.bind(cron);
-  cron.stop = () => {
+  const stopCronLifecycle = (preserveExitWatchers = false) => {
     try {
       stopCron();
-      stopExitWatchers();
+      if (preserveExitWatchers) {
+        // A committed replacement owns these children; fence this scheduler
+        // without terminating the adopted manager.
+        exitWatchersStopped = true;
+        exitWatcherGeneration += 1;
+      } else {
+        stopExitWatchers();
+      }
       stopHeartbeatReconcileRetry();
       void stopStreamWatchers().catch((err: unknown) => {
         cronLogger.warn(
@@ -1425,8 +1465,11 @@ export function buildGatewayCronService(params: {
       unregisterSessionAutomationSource(automationSource);
     }
   };
-  cron.stopAndDrain = async () => {
-    cron.stop();
+  cron.stop = () => {
+    stopCronLifecycle();
+  };
+  const stopAndDrainCron = async (preserveExitWatchers = false) => {
+    stopCronLifecycle(preserveExitWatchers);
     const exitWatchersStop = exitWatchersStopPromise ?? Promise.resolve();
     const streamWatchersStop = stopStreamWatchers().then(
       () => ({ ok: true as const }),
@@ -1448,6 +1491,9 @@ export function buildGatewayCronService(params: {
       throw streamWatchersResult.error;
     }
   };
+  cron.stopAndDrain = async () => {
+    await stopAndDrainCron();
+  };
   // 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.
@@ -1536,6 +1582,16 @@ export function buildGatewayCronService(params: {
     cron,
     storePath,
     cronEnabled,
+    prepareExitWatcherHandoff: async () => ({
+      current: () => exitWatchersRef.current!,
+      adopt: (watchers) => {
+        exitWatchersRef.current = watchers;
+        return watchers.updateHandlers(exitWatcherHandlers);
+      },
+      stopOwner: async () => {
+        await stopAndDrainCron(true);
+      },
+    }),
     reconcileExitWatchers,
     stopExitWatchers,
     reconcileStreamWatchers,

src/gateway/server-reload-handlers.test.ts

@@ -2,6 +2,7 @@
  * Gateway config reload handler tests.
  */
 import fs from "node:fs";
+import { readFile, writeFile } from "node:fs/promises";
 import path from "node:path";
 import { afterEach, beforeEach, describe, expect, it, vi } from "vitest";
 import { createDeferred } from "../../test/helpers/promise.js";
@@ -58,6 +59,8 @@ import {
   tryBeginGatewaySuspendAdmission,
 } from "../process/gateway-work-admission.js";
 import { CommandLane } from "../process/lanes.js";
+import { getProcessSupervisor } from "../process/supervisor/index.js";
+import { buildWindowsCmdExeCommandLine } from "../process/windows-command.js";
 import { resolveAuthProfileSecretOwnerId } from "../secrets/runtime-auth-profile-owner.js";
 import { listActiveDegradedSecretOwners } from "../secrets/runtime-degraded-state.js";
 import { createEmptyRuntimeWebToolsMetadata } from "../secrets/runtime-fast-path.js";
@@ -1270,6 +1273,110 @@ describe("provider auth hot reload path ownership", () => {
 });
 
 describe("gateway hot reload model state", () => {
+  it("keeps a supervised on-exit child alive exactly once across same-store cron reload", async () => {
+    const fixtureDir = autoCleanupTempDirs.make("openclaw-cron-exit-reload-");
+    const childScriptPath = path.join(fixtureDir, "watcher.cjs");
+    const markerPath = path.join(fixtureDir, "watcher-runs.txt");
+    const releasePath = path.join(fixtureDir, "release-watcher");
+    const config = {
+      session: { mainKey: "main", store: path.join(fixtureDir, "sessions.json") },
+      cron: { enabled: true, store: path.join(fixtureDir, "jobs.json") },
+    } as OpenClawConfig;
+    await writeFile(
+      childScriptPath,
+      "const fs=require('node:fs');" +
+        "fs.appendFileSync(process.argv[2],'run\\n');" +
+        "const timer=setInterval(()=>{if(fs.existsSync(process.argv[3]))clearInterval(timer)},10)",
+      "utf8",
+    );
+    const childArgs = [childScriptPath, markerPath, releasePath];
+    const command =
+      process.platform === "win32"
+        ? buildWindowsCmdExeCommandLine(process.execPath, childArgs)
+        : [process.execPath, ...childArgs].map((argument) => JSON.stringify(argument)).join(" ");
+    const supervisor = getProcessSupervisor();
+    const spawn = vi.spyOn(supervisor, "spawn");
+    let state: ReturnType<ReloadHandlerParams["getState"]> | undefined;
+
+    vi.stubEnv("OPENCLAW_STATE_DIR", fixtureDir);
+    vi.stubEnv("OPENCLAW_SKIP_CRON", "0");
+    hoisted.runtimeConfig.value = config;
+    setRuntimeConfigSnapshot(config, config);
+
+    try {
+      const actualCron =
+        await vi.importActual<typeof import("./server-cron.js")>("./server-cron.js");
+      const initialCronState = actualCron.buildGatewayCronService({
+        cfg: config,
+        deps: {} as never,
+        broadcast: vi.fn(),
+      });
+      state = {
+        ...createDefaultGatewayReloadState(),
+        cronState: initialCronState,
+      };
+      await initialCronState.cron.start();
+      const job = await initialCronState.cron.add({
+        name: "preserve the real watched child",
+        enabled: true,
+        schedule: { kind: "on-exit", command },
+        sessionTarget: "main",
+        wakeMode: "next-heartbeat",
+        payload: { kind: "systemEvent", text: "watched child finished" },
+      });
+      await initialCronState.reconcileExitWatchers();
+      await waitForFast(async () => expect(await readFile(markerPath, "utf8")).toBe("run\n"), {
+        timeout: 10_000,
+      });
+      expect(spawn).toHaveBeenCalledOnce();
+      const watchedRun = await spawn.mock.results[0]?.value;
+      if (!watchedRun) {
+        throw new Error("expected the supervised cron exit watcher to start");
+      }
+
+      hoisted.buildGatewayCronService.mockImplementationOnce(
+        (params) =>
+          actualCron.buildGatewayCronService(
+            params as Parameters<typeof actualCron.buildGatewayCronService>[0],
+          ) as unknown as ReturnType<typeof hoisted.buildGatewayCronService>,
+      );
+      const handlers = createGatewayReloadHandlers({
+        getState: () => {
+          if (!state) {
+            throw new Error("expected gateway state");
+          }
+          return state;
+        },
+        setState: (nextState) => {
+          state = nextState;
+        },
+      });
+
+      await withGatewayRestartSignal(async () => {
+        await handlers.applyHotReload(createCronRestartPlan(), config);
+      });
+
+      expect(supervisor.getRecord(watchedRun.runId)).toMatchObject({
+        pid: watchedRun.pid,
+        state: "running",
+      });
+      expect(await readFile(markerPath, "utf8")).toBe("run\n");
+      expect(spawn).toHaveBeenCalledOnce();
+
+      await writeFile(releasePath, "release");
+      await waitForFast(() => expect(state?.cronState.cron.getJob(job.id)?.enabled).toBe(false), {
+        timeout: 10_000,
+      });
+      expect(await readFile(markerPath, "utf8")).toBe("run\n");
+      expect(spawn).toHaveBeenCalledOnce();
+    } finally {
+      await writeFile(releasePath, "release").catch(() => {});
+      await state?.cronState.cron.stopAndDrain?.();
+      spawn.mockRestore();
+      vi.unstubAllEnvs();
+    }
+  });
+
   it.each([
     "agents.defaults.compaction.model",
     "agents.defaults.compaction.maxActiveTranscriptBytes",

src/gateway/server-reload-hot.ts

@@ -20,7 +20,7 @@ import {
 } from "./config-reload-recovery.js";
 import type { GatewayReloadPlan } from "./config-reload.js";
 import { commitHooksConfigReload, resolveHooksConfig } from "./hooks.js";
-import { buildGatewayCronService } from "./server-cron.js";
+import { buildGatewayCronService, type GatewayCronExitWatcherHandoff } from "./server-cron.js";
 import { applyGatewayLaneConcurrency, resolveGatewayLaneConcurrency } from "./server-lanes.js";
 import { createGatewayActiveWorkTracker } from "./server-reload-active-work.js";
 import {
@@ -114,6 +114,9 @@ export function createGatewayReloadHandlers(params: GatewayReloadHandlerParams)
     }
     nextState.hookClientIpConfig = resolveHookClientIpConfig(nextConfig);
 
+    let cronExitWatcherHandoff:
+      | { previous: GatewayCronExitWatcherHandoff; next: GatewayCronExitWatcherHandoff }
+      | undefined;
     if (plan.restartCron) {
       nextState.cronState = buildGatewayCronService({
         cfg: nextConfig,
@@ -126,6 +129,19 @@ export function createGatewayReloadHandlers(params: GatewayReloadHandlerParams)
           ? { resolveGatewayContext: params.resolveGatewayContext }
           : {}),
       });
+      if (
+        state.cronState.cronEnabled &&
+        nextState.cronState.cronEnabled &&
+        state.cronState.storePath === nextState.cronState.storePath
+      ) {
+        const [previous, next] = await Promise.all([
+          state.cronState.prepareExitWatcherHandoff?.(),
+          nextState.cronState.prepareExitWatcherHandoff?.(),
+        ]);
+        if (previous && next) {
+          cronExitWatcherHandoff = { previous, next };
+        }
+      }
     }
 
     resetDirectoryCache();
@@ -201,7 +217,10 @@ export function createGatewayReloadHandlers(params: GatewayReloadHandlerParams)
         if (plan.restartCron) {
           params.cronReconciliation.invalidate();
           params.onCronRestart?.();
-          if (state.cronState.cron.stopAndDrain) {
+          if (cronExitWatcherHandoff) {
+            await cronExitWatcherHandoff.next.adopt(cronExitWatcherHandoff.previous.current());
+            await cronExitWatcherHandoff.previous.stopOwner();
+          } else if (state.cronState.cron.stopAndDrain) {
             await state.cronState.cron.stopAndDrain();
           } else {
             state.cronState.cron.stop();