fix(cron): fence stream reconcile list snapshots against direct mutation routes

A cron.list snapshot captured across the reconcile await could be applied
after a direct add/update route already started the owner, stopping it as
removed and retiring its live identity. A mutation revision bumped at every
direct route start invalidates the stale snapshot; reconcile re-lists
(bounded) instead of applying it.

src/gateway/server-cron.test.ts

@@ -504,6 +504,97 @@ describe("buildGatewayCronService", () => {
     }
   });
 
+  it("discards a stale reconcile list snapshot that raced a direct mutation route", async () => {
+    let resolveWait!: (result: {
+      reason: "manual-cancel";
+      exitCode: null;
+      exitSignal: null;
+      durationMs: number;
+      stdout: string;
+      stderr: string;
+      timedOut: false;
+      noOutputTimedOut: false;
+    }) => void;
+    const wait = new Promise<Parameters<typeof resolveWait>[0]>((resolve) => {
+      resolveWait = resolve;
+    });
+    const cancel = vi.fn(() =>
+      resolveWait({
+        reason: "manual-cancel",
+        exitCode: null,
+        exitSignal: null,
+        durationMs: 1,
+        stdout: "",
+        stderr: "",
+        timedOut: false,
+        noOutputTimedOut: false,
+      }),
+    );
+    const detachOutput = vi.fn();
+    const spawn = vi.fn(async () => ({
+      runId: "run-stale-snapshot",
+      startedAtMs: Date.now(),
+      cancel,
+      detachOutput,
+      wait: () => wait,
+    }));
+    getProcessSupervisorMock.mockReturnValue({ spawn, cancelScope: vi.fn() });
+    const cfg = createCronConfig("server-cron-stream-stale-snapshot");
+    cfg.cron = { ...cfg.cron, triggers: { enabled: true } };
+    loadConfigMock.mockReturnValue(cfg);
+    const state = buildGatewayCronService({
+      cfg,
+      deps: {} as CliDeps,
+      broadcast: () => {},
+    });
+
+    try {
+      // Gate reconcile's first list call: capture the pre-add (empty) snapshot,
+      // hold it while the add's direct route starts the owner, then release the
+      // stale snapshot. The revision fence must re-list instead of stopping the
+      // freshly started owner as "removed".
+      const originalList = state.cron.list.bind(state.cron);
+      let releaseStaleList!: () => void;
+      const staleListGate = new Promise<void>((resolve) => {
+        releaseStaleList = resolve;
+      });
+      let armed = true;
+      state.cron.list = async (opts?: Parameters<typeof originalList>[0]) => {
+        if (!armed) {
+          return await originalList(opts);
+        }
+        armed = false;
+        const snapshot = await originalList(opts);
+        await staleListGate;
+        return snapshot;
+      };
+
+      const reconciling = state.reconcileStreamWatchers?.();
+      const added = await state.cron.add({
+        name: "stale snapshot stream source",
+        enabled: true,
+        schedule: { kind: "stream", command: ["source"] },
+        payload: { kind: "systemEvent", text: "event" },
+        sessionTarget: "main",
+        wakeMode: "next-heartbeat",
+      });
+      const streamJob = "job" in added ? added.job : added;
+      const sourceIdentity = streamJob.state.streamSourceIdentity;
+      expect(spawn).toHaveBeenCalledOnce();
+      releaseStaleList();
+      await reconciling;
+
+      expect(cancel).not.toHaveBeenCalled();
+      expect(detachOutput).not.toHaveBeenCalled();
+      expect(spawn).toHaveBeenCalledOnce();
+      expect(state.cron.getJob(streamJob.id)?.state.streamSourceIdentity).toBe(sourceIdentity);
+    } finally {
+      await state.stopStreamWatchers?.();
+      state.cron.stop();
+      vi.unstubAllEnvs();
+    }
+  });
+
   it("drains stream teardown once when stop and stopAndDrain overlap", async () => {
     const cancel = vi.fn();
     let resolveWait!: () => void;

src/gateway/server-cron.ts

@@ -430,6 +430,9 @@ export function buildGatewayCronService(params: {
   let streamWatcherReconciliations = 0;
   let exitWatcherGeneration = 0;
   let streamWatcherGeneration = 0;
+  // Bumped when a direct watcher route begins; fences reconcile's async list
+  // snapshot against mutations that commit inside the list await.
+  let streamWatcherMutationRevision = 0;
   let streamWatchersStopped = false;
   const reconcileExitWatchers = async () => {
     const generation = exitWatcherGeneration;
@@ -462,16 +465,32 @@ export function buildGatewayCronService(params: {
       if (!watchers || streamWatchersStopped) {
         return;
       }
-      const result = await cron.list({ includeDisabled: true });
-      if (generation !== streamWatcherGeneration || streamWatchersStopped) {
+      // The list snapshot is captured across an await; a direct mutation route
+      // that commits inside that window makes it stale, and reconciling a
+      // stale snapshot could stop a just-added owner as "removed" and retire
+      // its durable identity. Re-list until no route interleaved. Bounded:
+      // under pathological mutation churn we skip this sweep (every mutation
+      // was already routed directly) rather than loop forever.
+      for (let attempt = 0; attempt < 5; attempt += 1) {
+        const revision = streamWatcherMutationRevision;
+        const result = await cron.list({ includeDisabled: true });
+        if (generation !== streamWatcherGeneration || streamWatchersStopped) {
+          return;
+        }
+        if (revision !== streamWatcherMutationRevision) {
+          continue;
+        }
+        const jobs: CronJob[] = Array.isArray(result)
+          ? result
+          : (result as { jobs: CronJob[] }).jobs;
+        await watchers.reconcile(
+          jobs,
+          cronEnabled && params.cfg.cron?.triggers?.enabled === true,
+          params.cfg.cron?.triggers?.enabled === true,
+        );
         return;
       }
-      const jobs: CronJob[] = Array.isArray(result) ? result : (result as { jobs: CronJob[] }).jobs;
-      await watchers.reconcile(
-        jobs,
-        cronEnabled && params.cfg.cron?.triggers?.enabled === true,
-        params.cfg.cron?.triggers?.enabled === true,
-      );
+      cronLogger.warn({}, "cron-stream: reconcile skipped after repeated concurrent mutations");
     } catch (err) {
       cronLogger.warn({ err: String(err) }, "cron-stream: reconcile failed");
     } finally {
@@ -488,6 +507,7 @@ export function buildGatewayCronService(params: {
     if (!watchers || streamWatchersStopped) {
       return;
     }
+    streamWatcherMutationRevision += 1;
     streamWatcherReconciliations += 1;
     try {
       if (action === "removed") {
@@ -1232,7 +1252,10 @@ export function buildGatewayCronService(params: {
     stopCron();
     stopExitWatchers();
     void stopStreamWatchers().catch((err: unknown) => {
-      cronLogger.warn({ err: formatErrorMessage(err) }, "cron-stream: asynchronous teardown failed");
+      cronLogger.warn(
+        { err: formatErrorMessage(err) },
+        "cron-stream: asynchronous teardown failed",
+      );
     });
     // Session rows must stop reporting automation from a stopped scheduler,
     // but a reload's replacement service may already own the registration.