FN-6199: harden AI merge worktree cleanup

Guarantee AI merge clean-room worktrees are pruned on merge completion and swept by task state. - prune pre-existing same-task AI merge temp worktrees before starting a new merge and audit cleanup outcomes - add task-aware tempdir sweep rules with a short grace period for done or archived tasks and immediate cleanup for deleted tasks - extend merger and self-healing coverage and document the updated stale worktree policy Files changed: .changeset/ai-merge-cleanup-sweep.md | 5 + docs/architecture.md | 2 +- packages/engine/src/__tests__/merger-ai-cleanup.test.ts | 109 ++++++++++++++++++-- packages/engine/src/__tests__/self-healing-tempdir-sweep.test.ts | 112 ++++++++++++++++++++- packages/engine/src/merger-ai.ts | 63 ++++++++++++ packages/engine/src/self-healing.ts | 26 ++++- 6 files changed, 300 insertions(+), 17 deletions(-) Fusion-Task-Id: FN-6199 Fusion-Task-Lineage: 659cc52f-0994-4315-a187-6c865bbe3983
2026-06-10 10:57:02 -07:00
parent 747bbf51f1
commit d75f861f24
6 changed files with 300 additions and 17 deletions
--- a/packages/engine/src/tests/merger-ai-cleanup.test.ts
+++ b/packages/engine/src/tests/merger-ai-cleanup.test.ts
@@ -1,17 +1,33 @@
 import { afterEach, describe, expect, it, vi } from "vitest";
-import { existsSync, mkdtempSync, rmSync, writeFileSync } from "node:fs";
+import { existsSync, mkdirSync, mkdtempSync, realpathSync, rmSync, writeFileSync } from "node:fs";
 import { rm } from "node:fs/promises";
 import { tmpdir } from "node:os";
 import { join } from "node:path";
 import { execSync } from "node:child_process";
-import { cleanupAiMergeWorktree, runAiMerge } from "../merger-ai.js";
+import { cleanupAiMergeWorktree, pruneExistingAiMergeWorktrees, runAiMerge } from "../merger-ai.js";
+import { activeSessionRegistry } from "../active-session-registry.js";
 import type { RunAuditor } from "../run-audit.js";

+const fsState = vi.hoisted(() => ({ failReaddirPath: "" }));
+
+vi.mock("node:fs", async () => {
+  const actual = await vi.importActual<typeof import("node:fs")>("node:fs");
+  return {
+    ...actual,
+    readdirSync: vi.fn((path: Parameters<typeof actual.readdirSync>[0], options?: Parameters<typeof actual.readdirSync>[1]) => {
+      if (String(path) === fsState.failReaddirPath) throw new Error("simulated readdir failure");
+      return actual.readdirSync(path, options as never);
+    }),
+  };
+});
+
 const tracked = new Set<string>();
 const RM = { recursive: true, force: true, maxRetries: 5, retryDelay: 50 } as const;

 afterEach(() => {
  vi.restoreAllMocks();
+  fsState.failReaddirPath = "";
+  activeSessionRegistry.clear();
  for (const dir of tracked) {
    try { rmSync(dir, RM); } catch { /* best effort */ }
  }
@@ -51,7 +67,8 @@ async function cleanup(input: Partial<Parameters<typeof cleanupAiMergeWorktree>[
  return { mergeRoot, events, logs };
 }

-function initRepoWithBranch(): { dir: string } {
+function initRepoWithBranch(taskId = "FN-1"): { dir: string } {
+  const branch = `fusion/${taskId.toLowerCase()}`;
  const dir = mkdtempSync(join(tmpdir(), "fusion-ai-merge-cleanup-test-"));
  tracked.add(dir);
  git(dir, "init -q -b main");
@@ -60,7 +77,7 @@ function initRepoWithBranch(): { dir: string } {
  writeFileSync(join(dir, "base.txt"), "base\n");
  git(dir, "add -A");
  git(dir, "commit -q -m base");
-  git(dir, "checkout -q -b fusion/fn-1");
+  git(dir, `checkout -q -b ${branch}`);
  writeFileSync(join(dir, "feature.txt"), "feature work\n");
  git(dir, "add -A");
  git(dir, "commit -q -m 'feat: work'");
@@ -68,12 +85,12 @@ function initRepoWithBranch(): { dir: string } {
  return { dir };
 }

-function makeStore() {
+function makeStore(taskId = "FN-1") {
  const task: any = {
-    id: "FN-1",
+    id: taskId,
    column: "in-review",
    status: null,
-    branch: "fusion/fn-1",
+    branch: `fusion/${taskId.toLowerCase()}`,
    worktree: null,
    title: "do the thing",
    steps: [],
@@ -93,9 +110,16 @@ function makeStore() {
  return { store, audits, logs };
 }

-function realMergeAgent() {
+function tempAiMergeDir(name: string): string {
+  const dir = join(tmpdir(), name);
+  mkdirSync(dir, { recursive: true });
+  tracked.add(dir);
+  return dir;
+}
+
+function realMergeAgent(taskId = "FN-1") {
  return vi.fn(async (cwd: string) => {
-    execSync("git merge --squash fusion/fn-1", { cwd, stdio: "pipe" });
+    execSync(`git merge --squash fusion/${taskId.toLowerCase()}`, { cwd, stdio: "pipe" });
    execSync("git add -A", { cwd, stdio: "pipe" });
    execSync('git commit -q -m "squash: feature"', { cwd, stdio: "pipe" });
  });
@@ -151,6 +175,44 @@ describe("AI merge temp worktree cleanup", () => {
    ]));
  });

+  it("pruneExistingAiMergeWorktrees removes stale same-task directories", async () => {
+    const stale = tempAiMergeDir("fusion-ai-merge-fn-777-stale");
+    const { audit, events } = makeAudit();
+    const logs: string[] = [];
+
+    await expect(pruneExistingAiMergeWorktrees("FN-777", process.cwd(), audit, vi.fn(async (message: string) => { logs.push(message); }))).resolves.toBe(1);
+
+    expect(existsSync(stale)).toBe(false);
+    expect(events).toEqual(expect.arrayContaining([
+      expect.objectContaining({ type: "merge:ai-worktree-cleanup", metadata: expect.objectContaining({ taskId: "FN-777", mergeRoot: realpathSync(tmpdir()) + "/fusion-ai-merge-fn-777-stale", phase: "pre-merge-prune", success: true }) }),
+    ]));
+  });
+
+  it("pruneExistingAiMergeWorktrees skips directories for other tasks", async () => {
+    const other = tempAiMergeDir("fusion-ai-merge-fn-778-stale");
+    const { audit, events } = makeAudit();
+
+    await expect(pruneExistingAiMergeWorktrees("FN-777", process.cwd(), audit, vi.fn(async () => undefined))).resolves.toBe(0);
+
+    expect(existsSync(other)).toBe(true);
+    expect(events).toEqual([]);
+  });
+
+  it("pruneExistingAiMergeWorktrees skips active-session paths", async () => {
+    const stale = tempAiMergeDir("fusion-ai-merge-fn-777-active");
+    const canonical = realpathSync(stale);
+    activeSessionRegistry.registerPath(canonical, { taskId: "FN-777", kind: "executor", ownerKey: "FN-777" });
+    const { audit, events } = makeAudit();
+
+    await expect(pruneExistingAiMergeWorktrees("FN-777", process.cwd(), audit, vi.fn(async () => undefined))).resolves.toBe(0);
+    expect(existsSync(stale)).toBe(true);
+    expect(events).toEqual([]);
+
+    activeSessionRegistry.unregisterPath(canonical);
+    await expect(pruneExistingAiMergeWorktrees("FN-777", process.cwd(), audit, vi.fn(async () => undefined))).resolves.toBe(1);
+    expect(existsSync(stale)).toBe(false);
+  });
+
  it("runAiMerge emits success cleanup audit events", async () => {
    const { dir } = initRepoWithBranch();
    const { store, audits } = makeStore();
@@ -166,4 +228,33 @@ describe("AI merge temp worktree cleanup", () => {
      expect.objectContaining({ metadata: expect.objectContaining({ phase: "fs-rm", success: true }) }),
    ]));
  });
+
+  it("runAiMerge calls pre-merge prune before creating worktree", async () => {
+    const taskId = "FN-777";
+    const { dir } = initRepoWithBranch(taskId);
+    const orphan = tempAiMergeDir("fusion-ai-merge-fn-777-orphan");
+    const { store, audits } = makeStore(taskId);
+
+    await runAiMerge(store, dir, taskId, { manual: true }, {
+      mergeAgent: realMergeAgent(taskId),
+      reviewAgent: vi.fn(async () => "REVIEW_VERDICT: approve"),
+    });
+
+    expect(existsSync(orphan)).toBe(false);
+    expect(audits.filter((event) => event.mutationType === "merge:ai-worktree-cleanup")).toEqual(expect.arrayContaining([
+      expect.objectContaining({ metadata: expect.objectContaining({ phase: "pre-merge-prune", success: true }) }),
+    ]));
+  });
+
+  it("pre-merge prune failure does not abort merge", async () => {
+    const { dir } = initRepoWithBranch();
+    const { store, logs } = makeStore();
+    fsState.failReaddirPath = tmpdir();
+
+    await expect(runAiMerge(store, dir, "FN-1", { manual: true }, {
+      mergeAgent: realMergeAgent(),
+      reviewAgent: vi.fn(async () => "REVIEW_VERDICT: approve"),
+    })).resolves.toMatchObject({ ok: true, merged: true });
+    expect(logs.join("\n")).toContain("pre-merge prune failed");
+  });
 });
--- a/packages/engine/src/tests/self-healing-tempdir-sweep.test.ts
+++ b/packages/engine/src/tests/self-healing-tempdir-sweep.test.ts
@@ -72,17 +72,18 @@ afterEach(() => {
  }
 });

-function makeStore(settings: Record<string, unknown> = {}) {
+function makeStore(settings: Record<string, unknown> = {}, getTask: () => Promise<any> = async () => ({ id: "FN-1", column: "in-progress" })) {
  const audits: any[] = [];
  const store: any = {
    getSettings: vi.fn(async () => ({ ...settings })),
+    getTask: vi.fn(getTask),
    recordRunAuditEvent: vi.fn(async (event: any) => { audits.push(event); }),
  };
  return { store, audits };
 }

-function makeManager(settings: Record<string, unknown> = {}) {
-  const { store, audits } = makeStore(settings);
+function makeManager(settings: Record<string, unknown> = {}, getTask?: () => Promise<any>) {
+  const { store, audits } = makeStore(settings, getTask);
  const manager = new SelfHealingManager(store, { rootDir: projectRoot });
  return { manager, audits };
 }
@@ -93,11 +94,27 @@ function tempMergeDir(name = `fusion-ai-merge-fn-1-${Math.random().toString(36).
  return dir;
 }

-function makeStale(path: string): void {
-  const old = new Date(Date.now() - 3 * 60 * 60 * 1000);
+function makeAge(path: string, ageMs: number): void {
+  const old = new Date(Date.now() - ageMs);
  utimesSync(path, old, old);
 }

+function makeStale(path: string): void {
+  makeAge(path, 3 * 60 * 60 * 1000);
+}
+
+function makeDoneTaskStale(path: string): void {
+  makeAge(path, 11 * 60 * 1000);
+}
+
+function taskWithColumn(column: string): () => Promise<any> {
+  return async () => ({ id: "FN-999", column });
+}
+
+function missingTask(): () => Promise<any> {
+  return async () => { throw new Error("Task FN-999 not found"); };
+}
+
 async function sweep(manager: SelfHealingManager): Promise<number> {
  return await (manager as any).cleanupStaleTempMergeWorktrees();
 }
@@ -188,6 +205,91 @@ describe("SelfHealingManager temp-dir AI merge worktree sweep", () => {
    ]));
  });

+  it("removes worktree for done task after grace period", async () => {
+    const stale = tempMergeDir("fusion-ai-merge-fn-999-donetask");
+    makeDoneTaskStale(stale);
+    const { manager, audits } = makeManager({}, taskWithColumn("done"));
+
+    await expect(sweep(manager)).resolves.toBe(1);
+
+    expect(existsSync(stale)).toBe(false);
+    expect(sweepAudits(audits)).toEqual(expect.arrayContaining([
+      expect.objectContaining({ metadata: expect.objectContaining({ path: realpathSync(sandboxRoot) + "/fusion-ai-merge-fn-999-donetask", success: true, reason: "done-task-stale" }) }),
+    ]));
+  });
+
+  it("removes worktree for archived task after grace period", async () => {
+    const stale = tempMergeDir("fusion-ai-merge-fn-999-archivedtask");
+    makeDoneTaskStale(stale);
+    const { manager, audits } = makeManager({}, taskWithColumn("archived"));
+
+    await expect(sweep(manager)).resolves.toBe(1);
+
+    expect(existsSync(stale)).toBe(false);
+    expect(sweepAudits(audits)).toEqual(expect.arrayContaining([
+      expect.objectContaining({ metadata: expect.objectContaining({ success: true, reason: "done-task-stale" }) }),
+    ]));
+  });
+
+  it("removes worktree for deleted task immediately", async () => {
+    const fresh = tempMergeDir("fusion-ai-merge-fn-999-deletedtask");
+    const { manager, audits } = makeManager({}, missingTask());
+
+    await expect(sweep(manager)).resolves.toBe(1);
+
+    expect(existsSync(fresh)).toBe(false);
+    expect(sweepAudits(audits)).toEqual(expect.arrayContaining([
+      expect.objectContaining({ metadata: expect.objectContaining({ success: true, reason: "deleted-task" }) }),
+    ]));
+  });
+
+  it("keeps worktree for in-progress task within 2h gate", async () => {
+    const fresh = tempMergeDir("fusion-ai-merge-fn-999-inprogressfresh");
+    const { manager } = makeManager({}, taskWithColumn("in-progress"));
+
+    await expect(sweep(manager)).resolves.toBe(0);
+
+    expect(existsSync(fresh)).toBe(true);
+  });
+
+  it("removes worktree for in-progress task after 2h gate", async () => {
+    const stale = tempMergeDir("fusion-ai-merge-fn-999-inprogressstale");
+    makeStale(stale);
+    const { manager, audits } = makeManager({}, taskWithColumn("in-progress"));
+
+    await expect(sweep(manager)).resolves.toBe(1);
+
+    expect(existsSync(stale)).toBe(false);
+    expect(sweepAudits(audits)).toEqual(expect.arrayContaining([
+      expect.objectContaining({ metadata: expect.objectContaining({ success: true, reason: "stale" }) }),
+    ]));
+  });
+
+  it("keeps fresh worktree for done task within grace period", async () => {
+    const fresh = tempMergeDir("fusion-ai-merge-fn-999-donefresh");
+    makeAge(fresh, 5 * 60 * 1000);
+    const { manager } = makeManager({}, taskWithColumn("done"));
+
+    await expect(sweep(manager)).resolves.toBe(0);
+
+    expect(existsSync(fresh)).toBe(true);
+  });
+
+  it("handles non-parseable directory names with age-only fallback", async () => {
+    const fresh = tempMergeDir("fusion-ai-merge-unknown-fresh");
+    const stale = tempMergeDir("fusion-ai-merge-unknown-stale");
+    makeStale(stale);
+    const { manager, audits } = makeManager({}, missingTask());
+
+    await expect(sweep(manager)).resolves.toBe(1);
+
+    expect(existsSync(fresh)).toBe(true);
+    expect(existsSync(stale)).toBe(false);
+    expect(sweepAudits(audits)).toEqual(expect.arrayContaining([
+      expect.objectContaining({ metadata: expect.objectContaining({ path: expect.stringContaining("unknown-stale"), success: true, reason: "stale" }) }),
+    ]));
+  });
+
  it("proceeds when worktrunk is enabled", async () => {
    const stale = tempMergeDir();
    makeStale(stale);
--- a/packages/engine/src/merger-ai.ts
+++ b/packages/engine/src/merger-ai.ts
@@ -32,6 +32,7 @@
 */
 import { execFile } from "node:child_process";
 import { promisify } from "node:util";
+import { readdirSync, realpathSync, rmSync } from "node:fs";
 import { mkdtemp, rm } from "node:fs/promises";
 import { tmpdir } from "node:os";
 import { join } from "node:path";
@@ -62,6 +63,7 @@ import { accumulateSessionTokenUsage } from "./session-token-usage.js";
 import { createRunAuditor, generateSyntheticRunId, type RunAuditor } from "./run-audit.js";
 import { createLogger } from "./logger.js";
 import { captureSingleCommitLandedMetadata, type MergerOptions } from "./merger.js";
+import { activeSessionRegistry } from "./active-session-registry.js";

 const execFileAsync = promisify(execFile);
 const aiMergeLog = createLogger("merger-ai");
@@ -103,6 +105,61 @@ function describeCleanupError(err: unknown): string {
  return stderr ? `${message}: ${stderr.trim()}` : message;
 }

+export async function pruneExistingAiMergeWorktrees(
+  taskId: string,
+  projectRootDir: string,
+  audit: RunAuditor,
+  log: (message: string) => Promise<void>,
+): Promise<number> {
+  const prefix = `fusion-ai-merge-${taskId.toLowerCase()}-`;
+  const tempRoot = tmpdir();
+  let entries: string[];
+  try {
+    entries = readdirSync(tempRoot).filter((entry) => entry.startsWith(prefix));
+  } catch (err: unknown) {
+    await log(`AI merge pre-merge prune: failed to read ${tempRoot}: ${getErrorMessage(err)}`);
+    throw err;
+  }
+
+  let pruned = 0;
+  for (const entry of entries) {
+    const candidatePath = join(tempRoot, entry);
+    let canonicalPath = candidatePath;
+    try {
+      canonicalPath = realpathSync(candidatePath);
+    } catch {
+      canonicalPath = candidatePath;
+    }
+
+    if (activeSessionRegistry.isPathActive(canonicalPath) || activeSessionRegistry.isPathActive(candidatePath)) {
+      await log(`AI merge pre-merge prune: skipping active worktree ${canonicalPath}`);
+      continue;
+    }
+
+    try {
+      await execFileAsync("git", ["worktree", "remove", "--force", canonicalPath], {
+        cwd: projectRootDir,
+        timeout: 30_000,
+      });
+    } catch (err: unknown) {
+      await log(`AI merge pre-merge prune: git worktree remove failed for ${canonicalPath}: ${describeCleanupError(err)} — falling back to filesystem removal`);
+    }
+
+    try {
+      rmSync(canonicalPath, { recursive: true, force: true });
+      await audit.git({ type: "merge:ai-worktree-cleanup", target: canonicalPath, metadata: { taskId, mergeRoot: canonicalPath, phase: "pre-merge-prune", success: true } });
+      pruned++;
+    } catch (err: unknown) {
+      const error = getErrorMessage(err);
+      const code = getErrorStringProperty(err, "code");
+      await log(`AI merge pre-merge prune: filesystem rm failed for ${canonicalPath}${code ? ` (${code})` : ""}: ${error}`);
+      await audit.git({ type: "merge:ai-worktree-cleanup", target: canonicalPath, metadata: { taskId, mergeRoot: canonicalPath, phase: "pre-merge-prune", success: false, error, ...(code ? { code } : {}) } });
+    }
+  }
+
+  return pruned;
+}
+
 export async function cleanupAiMergeWorktree(input: {
  taskId: string;
  mergeRoot: string;
@@ -839,6 +896,12 @@ export async function runAiMerge(
  const taskTitle = task.title?.trim() ? task.title.split("\n")[0] : undefined;

  await setStatus("merging");
+  try {
+    const pruned = await pruneExistingAiMergeWorktrees(taskId, projectRootDir, audit, log);
+    if (pruned > 0) await log(`AI merge: pruned ${pruned} pre-existing worktree(s) for ${taskId}`);
+  } catch (err: unknown) {
+    await log(`AI merge: pre-merge prune failed: ${getErrorMessage(err)}`);
+  }
  let advanceRetries = 0;
  while (true) {
    throwIfAborted(options.signal, taskId);
--- a/packages/engine/src/self-healing.ts
+++ b/packages/engine/src/self-healing.ts
@@ -76,6 +76,7 @@ const DB_CORRUPTION_NOTIFICATION_COOLDOWN_MS = 60 * 60 * 1000;
 const FTS_MAINTENANCE_MERGE_CADENCE_TICKS = 1;
 const FTS_MAINTENANCE_OPTIMIZE_CADENCE_TICKS = 4;
 const STALE_TEMP_MERGE_WORKTREE_MS = 2 * 60 * 60 * 1000;
+const DONE_TASK_TEMP_WORKTREE_GRACE_MS = 10 * 60 * 1000;
 // Live pathology peaked around 775 KB/task (~96 MB for ~120 tasks), while a
 // rebuilt healthy index was ~0.1 MB. Keep the steady-state budget generous but
 // bounded so sustained text churn heals before segment growth becomes material.
@@ -94,6 +95,11 @@ export const MAX_COMPLETION_HANDOFF_LIMBO_RECOVERIES = 3;
 export const MAX_POST_DONE_NONCONTINUABLE_WEDGE_RECOVERIES = 3;
 const MAX_NO_PROGRESS_RESUME_ATTEMPTS = 2;

+function extractTaskIdFromTempMergeDir(dirname: string): string | null {
+  const match = /^fusion-ai-merge-(fn-\d+)-[a-z0-9]+$/i.exec(dirname);
+  return match?.[1]?.toUpperCase() ?? null;
+}
+
 type BranchGroupLandingRecorder = {
  recordBranchGroupMemberLanded?: (groupId: string, payload: {
    taskId: string;
@@ -8887,6 +8893,7 @@ export class SelfHealingManager {
      for (const entry of entries) {
        const path = join(tempRoot, entry);
        let canonicalPath = path;
+        let cleanupReason = "stale";
        try {
          const stat = statSync(path);
          if (!stat.isDirectory()) {
@@ -8894,7 +8901,22 @@ export class SelfHealingManager {
            continue;
          }
          const ageMs = now - stat.mtimeMs;
-          if (ageMs < STALE_TEMP_MERGE_WORKTREE_MS) continue;
+          let ageGateMs = STALE_TEMP_MERGE_WORKTREE_MS;
+          cleanupReason = "stale";
+          const taskId = extractTaskIdFromTempMergeDir(entry);
+          if (taskId) {
+            try {
+              const task = await this.store.getTask(taskId);
+              if (task.column === "done" || task.column === "archived") {
+                ageGateMs = DONE_TASK_TEMP_WORKTREE_GRACE_MS;
+                cleanupReason = "done-task-stale";
+              }
+            } catch {
+              ageGateMs = 0;
+              cleanupReason = "deleted-task";
+            }
+          }
+          if (ageGateMs > 0 && ageMs < ageGateMs) continue;
          try {
            canonicalPath = realpathSync(path);
          } catch {
@@ -8927,7 +8949,7 @@ export class SelfHealingManager {
        try {
          rmSync(canonicalPath, { recursive: true, force: true });
          log.log(`[self-healing] temp-dir sweep: cleaned stale AI merge worktree ${canonicalPath}`);
-          await auditor.git({ type: "worktree:tempdir-sweep", target: canonicalPath, metadata: { path: canonicalPath, success: true, reason: "stale" } });
+          await auditor.git({ type: "worktree:tempdir-sweep", target: canonicalPath, metadata: { path: canonicalPath, success: true, reason: cleanupReason } });
          cleaned++;
        } catch (err: unknown) {
          const errorMessage = err instanceof Error ? err.message : String(err);