fix(storage): publish cleanup manifests atomically

docs-site/src/content/docs/reference/management-api.md

@@ -239,6 +239,10 @@ Storage cleanup endpoints can move or permanently remove archived session data.
 first and submit the returned digest. Prefer quarantine when recovery may be needed.
 :::
 
+Cleanup recovery manifests are published atomically, preserving the previous complete record
+if a replacement fails before publication. This does not reverse a permanent purge: restore
+can still fail when a recorded session has no surviving rollout file.
+
 ### Models and catalog
 
 | Method and path | Purpose | Notable errors |

src/storage/cleanup.ts

@@ -30,11 +30,11 @@ import {
   writeSync,
   chmodSync,
 } from "node:fs";
-import { basename, isAbsolute, join, relative, resolve, sep } from "node:path";
+import { basename, dirname, isAbsolute, join, relative, resolve, sep } from "node:path";
 import { Database } from "bun:sqlite";
 import { resolveCodexHomeDir } from "../codex/home";
 import { readThreadFieldsFromRollout } from "../codex/history-provider";
-import { renameAtomicFile } from "../config";
+import { renameAtomicFile } from "../lib/windows-atomic-replace";
 
 export const ARCHIVED_SESSIONS_DIR = "archived_sessions";
 export const TRASH_DIR = ".trash";
@@ -115,9 +115,35 @@ function chmodPrivatePath(path: string, mode: number): void {
   try { chmodSync(path, mode); } catch { /* best-effort (e.g. Windows ACLs) */ }
 }
 
-function writePrivateFile(path: string, content: string): void {
-  writeFileSync(path, content, "utf8");
-  chmodPrivatePath(path, 0o600);
+/** Publish complete stage metadata without truncating the last recovery record. */
+function writePrivateFile(
+  path: string,
+  content: string,
+  beforeRename?: (temporaryPath: string, targetPath: string) => void,
+): void {
+  const temporaryPath = `${path}.${process.pid}.${randomUUID()}.tmp`;
+  let descriptor: number | undefined;
+  let created = false;
+  try {
+    descriptor = openSync(temporaryPath, "wx", 0o600);
+    created = true;
+    writeFileSync(descriptor, content, "utf8");
+    fsyncSync(descriptor);
+    closeSync(descriptor);
+    descriptor = undefined;
+    chmodPrivatePath(temporaryPath, 0o600);
+    beforeRename?.(temporaryPath, path);
+    renameAtomicFile(temporaryPath, path, undefined, "storage-cleanup");
+    chmodPrivatePath(path, 0o600);
+    fsyncDirectoryBestEffort(dirname(path));
+  } finally {
+    if (descriptor !== undefined) {
+      try { closeSync(descriptor); } catch { /* preserve publication failure */ }
+    }
+    if (created) {
+      try { unlinkSync(temporaryPath); } catch { /* renamed or cleanup unavailable */ }
+    }
+  }
 }
 
 function chunkIds(ids: string[], chunkSize: number): string[][] {
@@ -812,7 +838,6 @@ interface ReconcileTestHooks {
 const SATELLITE_BACKUP_FILE = "satellite-backup.json";
 /** Marks an incomplete restore so retries can accept dest files and resume metadata. */
 const RESTORE_PENDING_FILE = "restore-pending.json";
-let _satelliteBackupSeq = 0;
 
 type StagedFile = { from: string; to: string; relPath: string };
 
@@ -1070,34 +1095,11 @@ function writeSatelliteBackup(
   if (options?.failWrite) throw new Error("test_fail_satellite_backup_write");
   const dest = join(stageDir, SATELLITE_BACKUP_FILE);
   const replacing = existsSync(dest);
-  const tmp = join(stageDir, `${SATELLITE_BACKUP_FILE}.${process.pid}.${++_satelliteBackupSeq}.tmp`);
-  const payload = Buffer.from(JSON.stringify(backup), "utf8");
-  const fd = openSync(tmp, "w", 0o600);
-  try {
-    let offset = 0;
-    while (offset < payload.length) {
-      offset += writeSync(fd, payload, offset, payload.length - offset, null);
+  writePrivateFile(dest, JSON.stringify(backup), () => {
+    if (options?.failReplaceBeforeRename && replacing) {
+      throw new Error("test_fail_satellite_backup_replace");
     }
-    fsyncSync(fd);
-  } catch (error) {
-    try { closeSync(fd); } catch { /* */ }
-    try { unlinkSync(tmp); } catch { /* */ }
-    throw error;
-  }
-  closeSync(fd);
-  chmodPrivatePath(tmp, 0o600);
-  if (options?.failReplaceBeforeRename && replacing) {
-    try { unlinkSync(tmp); } catch { /* */ }
-    throw new Error("test_fail_satellite_backup_replace");
-  }
-  try {
-    renameAtomicFile(tmp, dest, undefined, "storage-cleanup");
-  } catch (error) {
-    try { unlinkSync(tmp); } catch { /* */ }
-    throw error;
-  }
-  chmodPrivatePath(dest, 0o600);
-  fsyncDirectoryBestEffort(stageDir);
+  });
 }
 
 function clearSatelliteBackup(stageDir: string): void {
@@ -1734,6 +1736,12 @@ export interface ExecuteCleanupOptions {
   /** Test-only failure injection for atomicity regressions. */
   _test?: {
     failManifestWrite?: boolean;
+    /** Observe the complete temp and prior destination before publication. Never serialized. */
+    beforeManifestReplace?: (
+      temporaryPath: string,
+      targetPath: string,
+      phase: "staging" | "pre-commit" | "purge-incomplete",
+    ) => void;
     failPurgeBasenames?: string[];
     failRollbackBasenames?: string[];
     blockStageDestBasenames?: string[];
@@ -1752,14 +1760,14 @@ export interface ExecuteCleanupOptions {
 /** Serializable cleanup test hooks allowed on the management API wire. */
 export type CleanupWireTestHooks = Omit<
   NonNullable<ExecuteCleanupOptions["_test"]>,
-  "afterSatelliteMutations" | "beforeReconcileLock"
+  "afterSatelliteMutations" | "beforeReconcileLock" | "beforeManifestReplace"
 >;
 
 function isStringArray(v: unknown): v is string[] {
   return Array.isArray(v) && v.every(e => typeof e === "string");
 }
 
-/** Pick only allowlisted serializable hooks; drops function hooks (afterSatelliteMutations, beforeReconcileLock) and unknown keys. */
+/** Pick only allowlisted serializable hooks; drops all function hooks and unknown keys. */
 export function pickWireCleanupTestHooks(raw: unknown): CleanupWireTestHooks | undefined {
   if (!raw || typeof raw !== "object") return undefined;
   const o = raw as Record<string, unknown>;
@@ -1911,6 +1919,9 @@ export function executeArchivedCleanup(options: ExecuteCleanupOptions): CleanupR
         entries: manifestEntries,
         ...extra,
       }, null, 2),
+      (temporaryPath, targetPath) => options._test?.beforeManifestReplace?.(
+        temporaryPath, targetPath, extra.staging ? "staging" : "pre-commit",
+      ),
     );
   };
 
@@ -1999,6 +2010,9 @@ export function executeArchivedCleanup(options: ExecuteCleanupOptions): CleanupR
             }))
             .filter(entry => entry.physicalRelPaths.length > 0),
         }, null, 2),
+        (temporaryPath, targetPath) => options._test?.beforeManifestReplace?.(
+          temporaryPath, targetPath, "purge-incomplete",
+        ),
       );
     } catch { /* best-effort: the pre-commit manifest is still on disk */ }
     return {

structure/02_config-and-codex-home.md

@@ -127,6 +127,12 @@ Worker cannot restore unrelated API keys or provider settings from a snapshot re
 If that metadata write is unavailable after cleanup has already completed, the job retains the
 cleanup outcome and exposes a bounded persistence error instead of relabeling the run as a Worker failure.
 
+Cleanup manifests and satellite backups share the stage-local atomic publisher: an exclusive
+private temporary file is fully written and file-synced before the existing Windows-tolerant
+rename replaces the destination. Handled publication failures retain the previous record;
+directory syncing remains best-effort. This does not make a partial permanent purge reversible:
+restore still fails closed when a recorded logical entry has no surviving file.
+
 Windows secret-file hardening resolves the effective token SID through an absolute, trusted
 PowerShell path before granting the owner and removing inherited broad ACL entries. The normal
 path obtains System32 from `GetSystemDirectoryW`. Windows ARM64 Bun builds that cannot execute

tests/storage/storage-cleanup.test.ts

@@ -8,6 +8,7 @@ import {
   readFileSync,
   renameSync,
   rmSync,
+  statSync,
   unlinkSync,
   utimesSync,
   writeFileSync,
@@ -20,6 +21,7 @@ import {
   listArchivedCandidates,
   listTrashEntries,
   normalizeArchivedRolloutPath,
+  pickWireCleanupTestHooks,
   previewArchivedCleanup,
   previewExactArchivedCleanup,
   restoreTrashEntry,
@@ -648,6 +650,127 @@ describe("executeArchivedCleanup", () => {
     expect(ids).toContain("told");
   });
 
+  test("initial manifest publication failure preserves originals and removes its private temp", () => {
+    home = buildHome();
+    const observed: Array<{ priorExists: boolean; next: string; mode: number }> = [];
+    const result = runWithDigest(50, "quarantine", home, {
+      now: 881,
+      _test: {
+        beforeManifestReplace: (temporaryPath, targetPath, phase) => {
+          if (phase !== "staging") return;
+          observed.push({
+            priorExists: existsSync(targetPath),
+            next: readFileSync(temporaryPath, "utf8"),
+            mode: statSync(temporaryPath).mode & 0o777,
+          });
+          throw new Error("injected_manifest_publication_failure");
+        },
+      },
+    });
+    // Assert outside the production catch: an assertion inside the hook could be swallowed.
+    expect(observed).toHaveLength(1);
+    expect(observed[0]!.priorExists).toBe(false);
+    expect(JSON.parse(observed[0]!.next).staging).toBe(true);
+    if (process.platform !== "win32") expect(observed[0]!.mode).toBe(0o600);
+    expect(result.error).toBe("fs_failed");
+    expect(existsSync(join(home, ".trash", "881"))).toBe(false);
+    expect(readFileSync(join(home, "archived_sessions", "rollout-old.jsonl"), "utf8")).toBe("OLD".repeat(10));
+    const db = new Database(join(home, "state_5.sqlite"), { readonly: true });
+    expect(db.query("SELECT id FROM threads WHERE id = 'told'").get()).toBeTruthy();
+    db.close();
+    expect(pickWireCleanupTestHooks({
+      beforeManifestReplace: () => {}, failManifestWrite: true,
+    })).toEqual({ failManifestWrite: true });
+  }, STORE_BUDGET_MS);
+
+  test("failed pre-delete replacement leaves the prior manifest intact during publication and restorable", () => {
+    home = buildHome();
+    let stagingBytes = "";
+    const observed: Array<{ prior: string; next: string }> = [];
+    const result = runWithDigest(50, "quarantine", home, {
+      now: 882,
+      _test: {
+        failRollbackBasenames: ["rollout-old.jsonl"],
+        beforeManifestReplace: (temporaryPath, targetPath, phase) => {
+          if (phase === "staging") stagingBytes = readFileSync(temporaryPath, "utf8");
+          if (phase !== "pre-commit") return;
+          observed.push({ prior: readFileSync(targetPath, "utf8"), next: readFileSync(temporaryPath, "utf8") });
+          throw new Error("injected_manifest_publication_failure");
+        },
+      },
+    });
+    expect(observed).toHaveLength(1);
+    expect(observed[0]!.prior).toBe(stagingBytes);
+    expect(JSON.parse(observed[0]!.prior).staging).toBe(true);
+    expect(JSON.parse(observed[0]!.next).staging).toBeUndefined();
+    expect(result.error).toBe("fs_failed");
+    expect(result.trashDir).toBe(".trash/882");
+    const stage = join(home, ".trash", "882");
+    expect(readFileSync(join(stage, "manifest.json"), "utf8")).toBe(stagingBytes);
+    expect(readFileSync(join(stage, "rollout-old.jsonl"), "utf8")).toBe("OLD".repeat(10));
+    expect(readdirSync(stage).filter(name => name.endsWith(".tmp"))).toEqual([]);
+    const db = new Database(join(home, "state_5.sqlite"), { readonly: true });
+    expect(db.query("SELECT id FROM threads WHERE id = 'told'").get()).toBeTruthy();
+    db.close();
+    const restored = restoreTrashEntry(".trash/882", { codexHome: home });
+    expect(restored.ok).toBe(true);
+    expect(restored.count).toBe(1);
+    expect(readFileSync(join(home, "archived_sessions", "rollout-old.jsonl"), "utf8")).toBe("OLD".repeat(10));
+  }, STORE_BUDGET_MS);
+
+  test.each([false, true])("failed post-purge manifest replacement preserves prior bytes (partial=%s)", partial => {
+    home = buildHome();
+    let preCommitBytes = "";
+    const observed: Array<{ prior: string; next: string }> = [];
+    const result = runWithDigest(100, "permanent", home, {
+      now: 883,
+      _test: {
+        failPurgeBasenames: partial
+          ? ["rollout-mid.jsonl"]
+          : ["rollout-old.jsonl", "rollout-mid.jsonl", "rollout-new.jsonl"],
+        beforeManifestReplace: (temporaryPath, targetPath, phase) => {
+          if (phase === "pre-commit") preCommitBytes = readFileSync(temporaryPath, "utf8");
+          if (phase !== "purge-incomplete") return;
+          observed.push({ prior: readFileSync(targetPath, "utf8"), next: readFileSync(temporaryPath, "utf8") });
+          throw new Error("injected_manifest_publication_failure");
+        },
+      },
+    });
+    expect(observed).toHaveLength(1);
+    expect(observed[0]!.prior).toBe(preCommitBytes);
+    expect(JSON.parse(observed[0]!.next).purgeIncomplete).toBe(true);
+    expect(JSON.parse(observed[0]!.next).entries).toHaveLength(partial ? 1 : 3);
+    expect(result.error).toBe("fs_failed");
+    const stage = join(home, ".trash", "883");
+    expect(readFileSync(join(stage, "manifest.json"), "utf8")).toBe(preCommitBytes);
+    expect(readdirSync(stage).filter(name => name.endsWith(".tmp"))).toEqual([]);
+    const dbBefore = new Database(join(home, "state_5.sqlite"), { readonly: true });
+    const rowsBefore = dbBefore.query("SELECT id FROM threads ORDER BY id").all();
+    dbBefore.close();
+    expect(rowsBefore).toEqual([{ id: "active" }]);
+    const stageBefore = readdirSync(stage).sort();
+    const restored = restoreTrashEntry(".trash/883", { codexHome: home });
+    if (partial) {
+      // A wholly purged old entry still fails closed; valid JSON is not full recovery.
+      expect(restored.error).toBe("fs_failed");
+      expect(restored.restoredPaths).toEqual([]);
+      expect(readdirSync(stage).sort()).toEqual(stageBefore);
+      expect(readFileSync(join(stage, "rollout-mid.jsonl"), "utf8")).toBe("MID".repeat(20));
+      expect(existsSync(join(home, "archived_sessions", "rollout-old.jsonl"))).toBe(false);
+      const dbAfter = new Database(join(home, "state_5.sqlite"), { readonly: true });
+      expect(dbAfter.query("SELECT id FROM threads ORDER BY id").all()).toEqual(rowsBefore);
+      dbAfter.close();
+    } else {
+      expect(restored.ok).toBe(true);
+      expect(restored.count).toBe(3);
+      const dbAfter = new Database(join(home, "state_5.sqlite"), { readonly: true });
+      expect(dbAfter.query("SELECT id FROM threads ORDER BY id").all()).toEqual([
+        { id: "active" }, { id: "tmid" }, { id: "tnew" }, { id: "told" },
+      ]);
+      dbAfter.close();
+    }
+  }, { timeout: STORE_BUDGET_MS });
+
   test("rename-back failure keeps staged file and reports relative trashDir", () => {
     home = buildHome();
     const db = new Database(join(home, "state_5.sqlite"));