fix(cron): prune scratch after durable job removal (#121024)

src/cron/scratch-store.ts

@@ -208,8 +208,8 @@ export function writeCronJobScratch(params: {
 
 /**
  * Deletes scratch when its owning job is removed, or — with expectedRevision —
- * atomically reverts a migration write back to the no-row state. Orphans remain
- * harmless on partial failure. Returns false when the guarded revision moved.
+ * atomically reverts a migration write back to the no-row state. Returns false
+ * when the guarded revision moved.
  */
 export function deleteCronJobScratch(
   storePath: string,

src/cron/service.removal-postcommit.test.ts

@@ -1,6 +1,8 @@
 import { afterEach, beforeEach, describe, expect, it, vi } from "vitest";
 import { resetGatewayWorkAdmission } from "../process/gateway-work-admission.js";
+import { readCronJobScratchState, writeCronJobScratch } from "./scratch-store.js";
 import { setupCronServiceSuite } from "./service.test-harness.js";
+import { add } from "./service/ops-mutations.js";
 import { list } from "./service/ops-read.js";
 import { run } from "./service/ops-run.js";
 import { createCronServiceState, type CronEvent, type CronServiceState } from "./service/state.js";
@@ -37,6 +39,18 @@ function createDueOneShot(id: string, nowMs: number): CronJob {
   };
 }
 
+function createReplacementInput(id: string) {
+  return {
+    id,
+    name: `replacement ${id}`,
+    enabled: true,
+    schedule: { kind: "every" as const, everyMs: 60_000 },
+    sessionTarget: "isolated" as const,
+    wakeMode: "next-heartbeat" as const,
+    payload: { kind: "agentTurn" as const, message: "replacement work" },
+  };
+}
+
 function createState(params: {
   storePath: string;
   nowMs: number;
@@ -85,21 +99,37 @@ afterEach(() => {
 });
 
 describe.each(removalPaths)("cron one-shot removal via %s", (path) => {
-  it("emits the full removal snapshot only after the deletion is durable", async () => {
+  it("emits the full removal snapshot only after the job and scratch deletion are durable", async () => {
     const { storePath } = await makeStorePath();
     const nowMs = Date.parse("2026-07-10T12:00:00.000Z");
     const job = createDueOneShot(`postcommit-${path.replaceAll(" ", "-")}`, nowMs);
     await saveCronStore(storePath, { version: 1, jobs: [job] });
+    expect(
+      writeCronJobScratch({
+        storePath,
+        jobId: job.id,
+        content: "original scratch",
+        nowMs: nowMs - 1,
+      }),
+    ).toMatchObject({ ok: true, currentRevision: 1 });
 
     const events: CronEvent[] = [];
-    const durableJobsAtRemoval: Array<Promise<CronJob[]>> = [];
+    const durableStateAtRemoval: Array<
+      Promise<{
+        jobs: CronJob[];
+        scratch: ReturnType<typeof readCronJobScratchState>;
+      }>
+    > = [];
     const state = createState({
       storePath,
       nowMs,
       onEvent: (event) => {
         events.push(structuredClone(event));
         if (event.action === "removed") {
-          durableJobsAtRemoval.push(loadCronStore(storePath).then((store) => store.jobs));
+          const scratch = readCronJobScratchState(storePath, job.id);
+          durableStateAtRemoval.push(
+            loadCronStore(storePath).then((store) => ({ jobs: store.jobs, scratch })),
+          );
         }
       },
     });
@@ -127,9 +157,15 @@ describe.each(removalPaths)("cron one-shot removal via %s", (path) => {
           },
         },
       });
-      expect(durableJobsAtRemoval).toHaveLength(1);
-      await expect(Promise.all(durableJobsAtRemoval)).resolves.toEqual([[]]);
+      expect(durableStateAtRemoval).toHaveLength(1);
+      await expect(Promise.all(durableStateAtRemoval)).resolves.toEqual([
+        { jobs: [], scratch: { currentRevision: 0 } },
+      ]);
       expect(state.store?.jobs).toEqual([]);
+
+      const replacement = await add(state, createReplacementInput(job.id));
+      expect(replacement.id).toBe(job.id);
+      expect(readCronJobScratchState(storePath, job.id)).toEqual({ currentRevision: 0 });
     } finally {
       clearStateTimer(state);
     }
@@ -140,6 +176,14 @@ describe.each(removalPaths)("cron one-shot removal via %s", (path) => {
     const nowMs = Date.parse("2026-07-10T12:00:00.000Z");
     const job = createDueOneShot(`rollback-${path.replaceAll(" ", "-")}`, nowMs);
     await saveCronStore(storePath, { version: 1, jobs: [job] });
+    writeCronJobScratch({
+      storePath,
+      jobId: job.id,
+      content: "scratch must survive rollback",
+      sourceSha256: "original-source",
+      nowMs: nowMs - 1,
+    });
+    const scratchBefore = readCronJobScratchState(storePath, job.id);
 
     const events: CronEvent[] = [];
     let listedAfterFinished: Promise<CronJob[]> | undefined;
@@ -184,6 +228,7 @@ describe.each(removalPaths)("cron one-shot removal via %s", (path) => {
       expect(state.durableNextRunAtMsByJobId).toEqual(
         new Map([[job.id, durableJob?.state.nextRunAtMs]]),
       );
+      expect(readCronJobScratchState(storePath, job.id)).toEqual(scratchBefore);
     } finally {
       clearStateTimer(state);
     }

src/cron/service/ops-lifecycle.ts

@@ -10,7 +10,7 @@ import {
   STARTUP_INTERRUPTED_ERROR,
 } from "./startup-run-repair.js";
 import type { CronServiceState, DeferredCronNotifications } from "./state.js";
-import { ensureLoaded, persist } from "./store.js";
+import { ensureLoaded, persist, pruneCronJobScratchAfterCommit } from "./store.js";
 import { tryFindCronTaskRunIdForRecovery, tryFindFinalizedCronTaskRun } from "./task-runs.js";
 import { armTimer, runMissedJobs, stopTimer } from "./timer.js";
 
@@ -92,10 +92,13 @@ export async function start(state: CronServiceState) {
     if (repairedAnyStartupRun || jobs.length > 0) {
       // Recovery notifications describe repaired durable rows, so never
       // publish them until the startup write has committed successfully.
-      await persist(state, {
+      const persisted = await persist(state, {
         ...(repairedAnyStartupRun ? {} : { stateOnly: true }),
         postPersistNotifications,
       });
+      if (persisted) {
+        pruneCronJobScratchAfterCommit(state, completedJobIdsToDelete);
+      }
     }
   });
 

src/cron/service/ops-mutations.ts

@@ -13,7 +13,6 @@ import {
   onCronJobInactive,
 } from "../active-jobs.js";
 import { cronSchedulingInputsEqual } from "../schedule-identity.js";
-import { deleteCronJobScratch } from "../scratch-store.js";
 import { removeCronJobBaseSession } from "../session-reaper.js";
 import { removeStaleCronJobFamilyRows } from "../store.js";
 import { createCronStreamSourceIdentity, cronStreamScheduleKey } from "../stream-schedule.js";
@@ -51,6 +50,7 @@ import {
   ensureLoaded,
   persist,
   persistOrRestore,
+  pruneCronJobScratchAfterCommit,
   runPostPersistCronNotifications,
   snapshotStoreForRollback,
   type CronRollbackSnapshot,
@@ -533,13 +533,7 @@ export async function remove(
         release,
       };
     }
-    try {
-      deleteCronJobScratch(state.deps.storePath, id);
-    } catch (error) {
-      // The job deletion is already durable. Scratch cleanup is idempotent and
-      // must not turn a committed removal into a retryable API failure.
-      state.deps.log.warn({ jobId: id, err: String(error) }, "cron: scratch cleanup failed");
-    }
+    pruneCronJobScratchAfterCommit(state, [id]);
     armTimer(state);
     emit(state, { jobId: id, action: "removed", job: removedJob });
     return { ok: true, removed: true } as const;
@@ -614,6 +608,12 @@ export async function removeAgentJobsTransactional<T>(
         armTimer(state);
         for (const job of removedJobs) {
           noteActiveCronJobRemoval(job.id);
+        }
+        pruneCronJobScratchAfterCommit(
+          state,
+          removedJobs.map((job) => job.id),
+        );
+        for (const job of removedJobs) {
           emit(state, { jobId: job.id, action: "removed", job });
         }
         throw error;
@@ -637,15 +637,11 @@ export async function removeAgentJobsTransactional<T>(
     runPostPersistCronNotifications(state, postPersistNotifications);
     for (const job of removedJobs) {
       noteActiveCronJobRemoval(job.id);
-      try {
-        deleteCronJobScratch(state.deps.storePath, job.id);
-      } catch (error) {
-        state.deps.log.warn(
-          { jobId: job.id, err: String(error) },
-          "cron: agent scratch cleanup failed",
-        );
-      }
     }
+    pruneCronJobScratchAfterCommit(
+      state,
+      removedJobs.map((job) => job.id),
+    );
     armTimer(state);
     for (const job of removedJobs) {
       emit(state, { jobId: job.id, action: "removed", job });

src/cron/service/ops-run.ts

@@ -22,7 +22,12 @@ import { releaseQueuedCronRun, runWithCronAdmission } from "./run-admission.js";
 import { mergeManualRunSnapshotAfterReload } from "./startup-run-repair.js";
 import type { CronServiceState, CronWakeMode, DeferredCronNotifications } from "./state.js";
 import { emit } from "./state.js";
-import { ensureLoaded, persistOrRestore, snapshotStoreForRollback } from "./store.js";
+import {
+  ensureLoaded,
+  persistOrRestore,
+  pruneCronJobScratchAfterCommit,
+  snapshotStoreForRollback,
+} from "./store.js";
 import { tryFinishCronTaskRunWithoutHistory } from "./task-runs.js";
 import {
   resolveCronRunScheduleOwnership,
@@ -304,6 +309,7 @@ async function finishPreparedManualRun(
         postPersistNotifications,
       });
       if (removedJob) {
+        pruneCronJobScratchAfterCommit(state, [removedJob.id]);
         emit(state, { jobId: removedJob.id, action: "removed", job: removedJob });
       }
       finalized = true;

src/cron/service/ops.test.ts

@@ -11,6 +11,7 @@ import { formatTaskStatusDetail } from "../../tasks/task-status.js";
 import { withEnvAsync } from "../../test-utils/env.js";
 import { createCronExecutionId } from "../run-id.js";
 import * as cronSchedule from "../schedule.js";
+import { readCronJobScratchState, writeCronJobScratch } from "../scratch-store.js";
 import { setupCronServiceSuite, writeCronStoreSnapshot } from "../service.test-harness.js";
 import * as cronStoreModule from "../store.js";
 import { loadCronJobsStoreWithConfigJobs, loadCronStore } from "../store.js";
@@ -836,6 +837,84 @@ describe("cron service ops seam coverage", () => {
     },
   );
 
+  it("prunes scratch when startup deletes a finalized delete-after-run one-shot", async () => {
+    const { storePath } = await makeStorePath();
+    const now = Date.parse("2026-03-23T12:00:00.000Z");
+    const startedAt = now - 30_000;
+    const endedAt = startedAt + 4_000;
+
+    await withStateDirForStorePath(storePath, async () => {
+      const job = createDueIsolatedJob(now);
+      job.id = "startup-finalized-delete-after-run";
+      job.name = "startup finalized delete after run";
+      job.deleteAfterRun = true;
+      job.schedule = { kind: "at", at: new Date(startedAt).toISOString() };
+      job.state = { runningAtMs: startedAt, nextRunAtMs: startedAt };
+      await writeCronStoreSnapshot({ storePath, jobs: [job] });
+      expect(
+        writeCronJobScratch({
+          storePath,
+          jobId: job.id,
+          content: "completed one-shot scratch",
+          nowMs: startedAt,
+        }),
+      ).toMatchObject({ ok: true, currentRevision: 1 });
+
+      const events: CronEvent[] = [];
+      const state = createCronServiceState({
+        storePath,
+        cronEnabled: true,
+        log: logger,
+        nowMs: () => now,
+        enqueueSystemEvent: vi.fn(),
+        requestHeartbeat: vi.fn(),
+        runIsolatedAgentJob: vi.fn(async () => ({ status: "ok" as const })),
+        onEvent: (event) => events.push(structuredClone(event)),
+      });
+      const taskRunId = tryCreateCronTaskRun({ state, job, startedAt });
+      if (!taskRunId) {
+        throw new Error("expected cron task run");
+      }
+      tryFinishCronTaskRun(state, {
+        taskRunId,
+        job,
+        event: {
+          jobId: job.id,
+          action: "finished",
+          job,
+          status: "ok",
+          summary: "completed before restart",
+          runAtMs: startedAt,
+          durationMs: endedAt - startedAt,
+        },
+      });
+
+      try {
+        await start(state);
+
+        expect((await loadCronStore(storePath)).jobs).toEqual([]);
+        expect(readCronJobScratchState(storePath, job.id)).toEqual({ currentRevision: 0 });
+        expect(
+          events.filter((event) => event.action === "finished" || event.action === "removed"),
+        ).toEqual([]);
+
+        const replacement = await add(state, {
+          id: job.id,
+          name: "same-id replacement",
+          enabled: true,
+          schedule: { kind: "every", everyMs: 60_000 },
+          sessionTarget: "isolated",
+          wakeMode: "next-heartbeat",
+          payload: { kind: "agentTurn", message: "replacement work" },
+        });
+        expect(replacement.id).toBe(job.id);
+        expect(readCronJobScratchState(storePath, job.id)).toEqual({ currentRevision: 0 });
+      } finally {
+        stop(state);
+      }
+    });
+  });
+
   it("keeps a finalized one-shot disabled when startup restores its stale marker", async () => {
     const { storePath } = await makeStorePath();
     const now = Date.parse("2026-03-23T12:00:00.000Z");
@@ -2027,6 +2106,16 @@ describe("cron service ops persist rollback", () => {
         ...makeCreateInput("deleted agent job"),
         agentId: "doomed",
       });
+      expect(
+        writeCronJobScratch({
+          storePath,
+          jobId: removed.id,
+          content: "deleted agent scratch",
+          sourceSha256: "deleted-agent-source",
+          nowMs: now - 1,
+        }),
+      ).toMatchObject({ ok: true, currentRevision: 1 });
+      const scratchBefore = readCronJobScratchState(storePath, removed.id);
       const malformed = await add(state, {
         ...makeCreateInput("malformed surviving job"),
         agentId: "survivor",
@@ -2085,6 +2174,21 @@ describe("cron service ops persist rollback", () => {
       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);
+      expect(readCronJobScratchState(storePath, removed.id)).toEqual(
+        rolledBack ? scratchBefore : { currentRevision: 0 },
+      );
+      if (!rolledBack) {
+        const replacement = await add(state, {
+          ...makeCreateInput("same-id replacement"),
+          id: removed.id,
+          agentId: "survivor",
+        });
+        expect(replacement.id).toBe(removed.id);
+        expect(readCronJobScratchState(storePath, removed.id)).toEqual({ currentRevision: 0 });
+        if (state.timer) {
+          clearTimeout(state.timer);
+        }
+      }
     },
   );
 

src/cron/service/store.ts

@@ -3,6 +3,7 @@ import { normalizeCronJobIdentityFields } from "../normalize-job-identity.js";
 import { normalizeCronJobInput } from "../normalize.js";
 import { getInvalidPersistedCronJobReason } from "../persisted-shape.js";
 import { cronSchedulingInputsEqual } from "../schedule-identity.js";
+import { deleteCronJobScratch } from "../scratch-store.js";
 import { isInvalidCronSessionTargetIdError } from "../session-target.js";
 import {
   getCronJobsStoreRevision,
@@ -320,6 +321,23 @@ export function runPostPersistCronNotifications(
   }
 }
 
+/** Best-effort scratch pruning after the owning job deletions are durable. */
+export function pruneCronJobScratchAfterCommit(
+  state: CronServiceState,
+  committedJobIds: Iterable<string>,
+) {
+  for (const jobId of committedJobIds) {
+    try {
+      deleteCronJobScratch(state.deps.storePath, jobId);
+    } catch (error) {
+      state.deps.log.warn(
+        { jobId, err: String(error) },
+        "cron: post-commit scratch cleanup failed",
+      );
+    }
+  }
+}
+
 /** Captures the live cron state that must stay aligned with the durable store. */
 export function snapshotStoreForRollback(state: CronServiceState): CronRollbackSnapshot {
   return {

src/cron/service/timer-outcome-finalization.ts

@@ -6,7 +6,12 @@ import { recomputeNextRunsForMaintenance } from "./jobs.js";
 import { locked } from "./locked.js";
 import { clearQueuedCronRunReservationMarker, releaseQueuedCronRun } from "./run-admission.js";
 import { emit, type CronServiceState, type DeferredCronNotifications } from "./state.js";
-import { ensureLoaded, persistOrRestore, snapshotStoreForRollback } from "./store.js";
+import {
+  ensureLoaded,
+  persistOrRestore,
+  pruneCronJobScratchAfterCommit,
+  snapshotStoreForRollback,
+} from "./store.js";
 import { tryFinishCronTaskRunWithoutHistory } from "./task-runs.js";
 import type { TimedCronRunOutcome } from "./timer-execution-timeout.js";
 import { applyOutcomeToStoredJob } from "./timer-outcomes.js";
@@ -160,6 +165,10 @@ export async function finalizeCompletedCronRunOutcomes(
       await persistOrRestore(state, rollbackSnapshot, {
         postPersistNotifications,
       });
+      pruneCronJobScratchAfterCommit(
+        state,
+        removedJobs.map((job) => job.id),
+      );
       finishPersistedQuietCronTaskRuns(state, finalizedOutcomes);
       for (const removedJob of removedJobs) {
         emit(state, { jobId: removedJob.id, action: "removed", job: removedJob });