Recovered: fix(bundle): atomic re-commit via per-entry aside (#196 by @SahilRakhaiya05)

src/lib/bundle.commit.test.ts

@@ -0,0 +1,124 @@
+import type * as NodeFsPromises from 'node:fs/promises';
+import {
+  existsSync,
+  mkdirSync,
+  mkdtempSync,
+  readdirSync,
+  readFileSync,
+  writeFileSync,
+} from 'node:fs';
+import { tmpdir } from 'node:os';
+import { join } from 'node:path';
+import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest';
+
+const renameMock = vi.hoisted(() => vi.fn());
+const rmMock = vi.hoisted(() => vi.fn());
+
+vi.mock('node:fs/promises', async importOriginal => {
+  const actual = (await importOriginal()) as typeof NodeFsPromises;
+  return {
+    ...actual,
+    rename: renameMock,
+    rm: rmMock,
+  };
+});
+
+const { commitBundle } = await import('./bundle.js');
+
+describe('commitBundle', () => {
+  let realRename: typeof NodeFsPromises.rename;
+  let realRm: typeof NodeFsPromises.rm;
+
+  beforeEach(async () => {
+    const actual = (await vi.importActual('node:fs/promises')) as typeof NodeFsPromises;
+    realRename = actual.rename;
+    realRm = actual.rm;
+    renameMock.mockImplementation(realRename);
+    rmMock.mockImplementation(realRm);
+  });
+
+  afterEach(() => {
+    renameMock.mockReset();
+    rmMock.mockReset();
+  });
+
+  async function withTempParent(run: (parent: string) => Promise<void>): Promise<void> {
+    const parent = mkdtempSync(join(tmpdir(), 'bundle-commit-parent-'));
+    try {
+      await run(parent);
+    } finally {
+      await realRm(parent, { recursive: true, force: true }).catch(() => undefined);
+    }
+  }
+
+  function seedBundleDirs(parent: string): { dir: string; tmpDir: string; files: string[] } {
+    const dir = join(parent, 'bundle');
+    const tmpDir = join(dir, '.tmp');
+
+    mkdirSync(dir, { recursive: true });
+    writeFileSync(join(dir, 'notes.txt'), 'foreign notes\n', 'utf8');
+    mkdirSync(join(dir, 'steps'), { recursive: true });
+    writeFileSync(join(dir, 'meta.json'), '{"snapshotId":"snap_old"}\n', 'utf8');
+    writeFileSync(join(dir, 'steps', '01-evidence.json'), '{"step":1}\n', 'utf8');
+
+    mkdirSync(join(tmpDir, 'steps'), { recursive: true });
+    writeFileSync(join(tmpDir, 'meta.json'), '{"snapshotId":"snap_new"}\n', 'utf8');
+    writeFileSync(join(tmpDir, 'result.json'), '{}\n', 'utf8');
+    writeFileSync(join(tmpDir, 'steps', '01-evidence.json'), '{"step":9}\n', 'utf8');
+
+    return { dir, tmpDir, files: ['result.json', 'meta.json', 'steps/01-evidence.json'] };
+  }
+
+  it('rolls back to the prior complete bundle when a staged rename fails', async () => {
+    await withTempParent(async parent => {
+      const { dir, tmpDir, files } = seedBundleDirs(parent);
+
+      renameMock.mockImplementation(async (oldPath, newPath) => {
+        const dest = String(newPath);
+        if (dest.endsWith('result.json') && !dest.includes('.aside.')) {
+          throw Object.assign(new Error('simulated install failure'), { code: 'EACCES' });
+        }
+        return realRename(oldPath, newPath);
+      });
+
+      await expect(commitBundle(tmpDir, dir, files)).rejects.toThrow('simulated install failure');
+
+      expect(readFileSync(join(dir, 'meta.json'), 'utf8')).toBe('{"snapshotId":"snap_old"}\n');
+      expect(readFileSync(join(dir, 'steps', '01-evidence.json'), 'utf8')).toBe('{"step":1}\n');
+      expect(readFileSync(join(dir, 'notes.txt'), 'utf8')).toBe('foreign notes\n');
+      const leftovers = readdirSync(parent).filter(name => name.includes('.aside.'));
+      expect(leftovers).toEqual([]);
+    });
+  });
+
+  it('preserves foreign files while installing the new bundle on success', async () => {
+    await withTempParent(async parent => {
+      const { dir, tmpDir, files } = seedBundleDirs(parent);
+
+      await expect(commitBundle(tmpDir, dir, files)).resolves.toBeUndefined();
+
+      expect(readFileSync(join(dir, 'meta.json'), 'utf8')).toBe('{"snapshotId":"snap_new"}\n');
+      expect(readFileSync(join(dir, 'steps', '01-evidence.json'), 'utf8')).toBe('{"step":9}\n');
+      expect(readFileSync(join(dir, 'notes.txt'), 'utf8')).toBe('foreign notes\n');
+      expect(existsSync(join(dir, 'result.json'))).toBe(true);
+    });
+  });
+
+  it('keeps the new bundle when post-commit aside cleanup fails', async () => {
+    await withTempParent(async parent => {
+      const { dir, tmpDir, files } = seedBundleDirs(parent);
+
+      rmMock.mockImplementation(async (path, options) => {
+        if (String(path).includes('.aside.')) {
+          throw Object.assign(new Error('simulated aside cleanup failure'), { code: 'EACCES' });
+        }
+        return realRm(path, options);
+      });
+
+      await expect(commitBundle(tmpDir, dir, files)).resolves.toBeUndefined();
+
+      expect(readFileSync(join(dir, 'meta.json'), 'utf8')).toBe('{"snapshotId":"snap_new"}\n');
+      expect(readFileSync(join(dir, 'notes.txt'), 'utf8')).toBe('foreign notes\n');
+    });
+  });
+});

src/lib/bundle.ts

@@ -33,12 +33,13 @@
  *     when the agent re-runs. M3 may add resume.
  */
 
+import { randomUUID } from 'node:crypto';
 import { mkdir, mkdtemp, readdir, rename, rm, stat, unlink, writeFile } from 'node:fs/promises';
 import type { Writable } from 'node:stream';
 import { pipeline } from 'node:stream/promises';
 import type { ReadableStream as NodeReadableStream } from 'node:stream/web';
 import { createWriteStream } from 'node:fs';
-import { dirname, isAbsolute, join, relative, resolve } from 'node:path';
+import { basename, dirname, isAbsolute, join, relative, resolve } from 'node:path';
 import type { CliFailureContext, CliTestStep } from '../commands/test.js';
 import { ApiError, TransportError, localValidationError } from './errors.js';
 import { requireEnum } from './validate.js';
@@ -519,30 +520,6 @@ async function freshTmpDir(dir: string): Promise<string> {
   return tmpDir;
 }
 
-/**
- * Rename `<tmp>/<file>` → `<dir>/<file>` for every file in `files`.
- *
- * Critical ordering for atomicity (the §3 "agent-safe" contract):
- *
- *   1. **Remove the OLD `meta.json` first.** The bundle's completion
- *      signal is `meta.json`'s presence; an agent reading `<dir>` while
- *      we're mutating it must see "no meta → bundle absent or
- *      mid-write" rather than "meta points at a snapshot that's already
- *      been partially overwritten." Removing the old meta is what
- *      makes the rest of the swap safe to do in place.
- *   2. Wipe stale top-level files (e.g. an old `video.mp4` when the new
- *      bundle has no video). Without this, a fresh bundle could ship
- *      with a stale video lingering at the top level.
- *   3. Replace `<dir>/steps/` wholesale.
- *   4. Rename top-level files into place.
- *   5. **Rename `meta.json` LAST.** Its visible presence is the atomic
- *      completion signal; until step 5 lands, agents see "incomplete."
- *
- * The window between (1) and (5) is bounded by a handful of `rename`
- * syscalls — small enough that a SIGKILL there is rare, and any agent
- * caught reading the dir during it sees no meta and refuses to consume
- * (per §7.3). That's what we want.
- */
 /**
  * Whether a top-level directory entry belongs to the bundle format —
  * i.e. something a prior `writeBundle` could have produced and this
@@ -567,62 +544,101 @@ export function isBundleOwnedEntry(entry: string): boolean {
   return /^code\.[A-Za-z0-9]+$/.test(entry);
 }
 
-async function commitBundle(
+/**
+ * Atomically install the complete bundle staged in `tmpDir` into `dir`.
+ *
+ * Compatible with the #162 data-loss guard: only `isBundleOwnedEntry`
+ * names are moved aside or replaced — foreign files in `--out` survive.
+ *
+ * Re-commit safety: bundle-owned entries are renamed to a sibling
+ * aside directory before the new artifacts land. The prior `meta.json`
+ * is moved aside first (§3/§7.3 — no meta ⇒ refuse to consume) so a
+ * concurrent reader never sees meta pointing at steps already gone.
+ * The new `meta.json` is installed last. On failure, aside entries are
+ * restored so a failed re-commit never leaves the directory unusable.
+ */
+export async function commitBundle(
   tmpDir: string,
   dir: string,
   files: ReadonlyArray<string>,
 ): Promise<void> {
-  // (1) Remove the prior bundle's completion signal FIRST.
-  await unlink(join(dir, 'meta.json')).catch(() => undefined);
-
-  // (2) Sweep stale top-level files that the new bundle won't write.
-  // If the prior run wrote `video.mp4` and the new run has no video,
-  // an in-place rename leaves the old video lingering. Only entries the
-  // bundle format OWNS are candidates: `--out` may point at a directory
-  // that also holds the user's unrelated files, and those must survive
-  // the commit (deleting them would be silent data loss).
-  const topLevel = files.filter(f => !f.startsWith('steps/'));
-  const newTopLevelSet = new Set(topLevel);
-  newTopLevelSet.add('meta.json'); // about to land last, do not delete
-  const existing = await readdir(dir).catch(() => [] as string[]);
-  for (const entry of existing) {
-    // Preserve the writer's own scratch dir + the .partial marker
-    // (we'll re-evaluate .partial at the end of commit). Any other
-    // bundle-owned entry not-listed in the new bundle is stale.
-    if (entry === '.tmp' || entry === '.partial') continue;
-    if (newTopLevelSet.has(entry)) continue;
-    if (entry === 'steps') continue; // handled below
-    if (!isBundleOwnedEntry(entry)) continue; // foreign file — never touch
-    await rm(join(dir, entry), { recursive: true, force: true });
-  }
+  const parent = dirname(dir);
+  const base = basename(dir);
+  const asideDir = join(parent, `.${base}.aside.${randomUUID()}`);
+  const asideLog: Array<{ asidePath: string; restorePath: string }> = [];
+
+  const asideIfPresent = async (entry: string): Promise<void> => {
+    const restorePath = join(dir, entry);
+    if (!(await pathExists(restorePath))) return;
+    const asidePath = join(asideDir, entry);
+    await mkdir(dirname(asidePath), { recursive: true });
+    await rename(restorePath, asidePath);
+    asideLog.push({ asidePath, restorePath });
+  };
 
-  // (3) Replace `<dir>/steps/` with `<tmp>/steps/`.
-  const stepsTmp = join(tmpDir, 'steps');
-  const stepsDir = join(dir, 'steps');
-  await rm(stepsDir, { recursive: true, force: true });
-  if (await dirExists(stepsTmp)) {
-    await rename(stepsTmp, stepsDir);
-  }
+  const rollback = async (): Promise<void> => {
+    for (const { asidePath, restorePath } of [...asideLog].reverse()) {
+      await rm(restorePath, { recursive: true, force: true }).catch(() => undefined);
+      await rename(asidePath, restorePath).catch(() => undefined);
+    }
+    await rm(asideDir, { recursive: true, force: true }).catch(() => undefined);
+  };
 
-  // (4) Top-level files (result/failure/code/video). meta.json renames
-  // LAST; track it separately.
-  const metaIdx = topLevel.indexOf('meta.json');
-  const beforeMeta = metaIdx >= 0 ? topLevel.filter((_, i) => i !== metaIdx) : topLevel;
-  for (const file of beforeMeta) {
-    await rename(join(tmpDir, file), join(dir, file));
-  }
+  try {
+    const topLevel = files.filter(f => !f.startsWith('steps/'));
+    const newTopLevelSet = new Set(topLevel);
+    newTopLevelSet.add('meta.json');
+
+    // meta.json first — its presence is the completion signal; do not
+    // move steps (or anything else) aside while a stale meta is still visible.
+    await asideIfPresent('meta.json');
+
+    const existing = await readdir(dir).catch(() => [] as string[]);
+    for (const entry of existing) {
+      if (entry === '.tmp' || entry === 'meta.json') continue;
+      if (!isBundleOwnedEntry(entry)) continue;
+
+      const isStale = entry !== 'steps' && entry !== '.partial' && !newTopLevelSet.has(entry);
+      const willReplace = entry === 'steps' || newTopLevelSet.has(entry);
+      if (isStale || willReplace) {
+        await asideIfPresent(entry);
+      }
+    }
 
-  // (5) meta.json LAST → atomic completion signal.
-  if (metaIdx >= 0) {
-    await rename(join(tmpDir, 'meta.json'), join(dir, 'meta.json'));
-  }
+    const stepsTmp = join(tmpDir, 'steps');
+    const stepsDir = join(dir, 'steps');
+    if (await dirExists(stepsTmp)) {
+      await rename(stepsTmp, stepsDir);
+    }
 
-  // .partial from a prior aborted run is now stale. Remove it so an
-  // agent inspecting the dir sees only the fresh bundle.
-  await unlink(join(dir, '.partial')).catch(() => undefined);
+    const metaIdx = topLevel.indexOf('meta.json');
+    const beforeMeta = metaIdx >= 0 ? topLevel.filter((_, i) => i !== metaIdx) : topLevel;
+    for (const file of beforeMeta) {
+      await rename(join(tmpDir, file), join(dir, file));
+    }
 
-  // Clean up the now-empty tmp dir.
-  await rm(tmpDir, { recursive: true, force: true });
+    if (metaIdx >= 0) {
+      await rename(join(tmpDir, 'meta.json'), join(dir, 'meta.json'));
+    }
+
+    await unlink(join(dir, '.partial')).catch(() => undefined);
+    await rm(tmpDir, { recursive: true, force: true });
+    // Best-effort aside cleanup — failures must not roll back a committed bundle.
+    await rm(asideDir, { recursive: true, force: true }).catch(() => undefined);
+  } catch (err) {
+    await rollback();
+    await rm(tmpDir, { recursive: true, force: true }).catch(() => undefined);
+    throw err;
+  }
+}
+
+async function pathExists(path: string): Promise<boolean> {
+  try {
+    await stat(path);
+    return true;
+  } catch {
+    return false;
+  }
 }
 
 async function dirExists(path: string): Promise<boolean> {