diff --git a/.changeset/ai-merge-cleanup-sweep.md b/.changeset/ai-merge-cleanup-sweep.md new file mode 100644 index 0000000000..85b4c84d14 --- /dev/null +++ b/.changeset/ai-merge-cleanup-sweep.md @@ -0,0 +1,5 @@ +--- +"@runfusion/fusion": patch +--- + +Harden AI merge temporary worktree cleanup with same-task pre-merge pruning and task-aware stale tempdir sweeping for completed or deleted tasks. diff --git a/docs/architecture.md b/docs/architecture.md index 8b2e2e8f17..6319f2e9d8 100644 --- a/docs/architecture.md +++ b/docs/architecture.md @@ -670,7 +670,7 @@ Runtime action-gate flow (v1): - `TransientErrorDetector` (`transient-error-detector.ts`) — retriable error classification - `SelfHealingManager` (`self-healing.ts`) — auto-unpause/maintenance recovery actions - Batch 1 maintenance now includes one `fts-maintenance` step for both search indexes. The live `tasks_fts` branch still runs `merge` every tick, `optimize` every 4th tick, and `rebuild` above `32 MiB` or `1 MiB × live task count`. The archive `archived_tasks_fts` branch is lighter because archive writes are mostly append-only: `merge` every 8th tick, `optimize` every 24th tick, and `rebuild` above `64 MiB` or `512 KiB × archived row count`. Each branch is independently guarded by `fts5Available` and emits `task:fts-maintenance` run-audit telemetry with distinct `target` values (`tasks_fts` vs `archived_tasks_fts`). - - Batch 1 also sweeps stale AI merge clean-room worktrees under `tmpdir()` whose names start with `fusion-ai-merge-`. The sweep only considers directories older than 2 hours, canonicalizes paths before checking `activeSessionRegistry`, attempts `git worktree remove --force ` before filesystem removal, and emits `worktree:tempdir-sweep` run-audit telemetry for removal attempts and failures. It is intentionally native even when `worktrunk.enabled` because these temp-dir worktrees are outside the worktrunk-managed project layout. Fresh directories, active-session paths, and individual removal failures are skipped/logged without aborting the maintenance cycle. + - Batch 1 also sweeps stale AI merge clean-room worktrees under `tmpdir()` whose names start with `fusion-ai-merge-`. The default age gate is 2 hours, but task-aware cleanup uses a 10-minute grace period for `done`/`archived` tasks and treats missing/deleted task rows as immediately stale. The sweep canonicalizes paths before checking `activeSessionRegistry`, attempts `git worktree remove --force ` before filesystem removal, and emits `worktree:tempdir-sweep` run-audit telemetry for removal attempts and failures. It is intentionally native even when `worktrunk.enabled` because these temp-dir worktrees are outside the worktrunk-managed project layout. Fresh directories, active-session paths, and individual removal failures are skipped/logged without aborting the maintenance cycle. - `recoverGhostReviewTasks()` is a fallback only for idle, non-terminal `in-review` states. Terminal/actionable states (notably `status: "failed"`) are preserved and **not** auto-kicked back to `todo`. - Mission validation has a dedicated stale-run reaper: startup recovery and Batch 2 maintenance call `reapStaleMissionValidatorRuns()` when wired by the runtime, using `VALIDATOR_RUN_STALE_MAX_AGE_MS` (currently 6 hours). The sweep terminates ownerless `mission_validator_runs.status='running'` rows as `error`, writes the reap reason into `summary`, leaves `lastValidatorRunId` pointing at the now-terminal run, and emits run-audit telemetry with `mutationType: "mission:validator-run-reaped"` plus `runId`/`featureId`/`missionId`/`triggerType`/`elapsedMs` metadata. Active mission features move to `loopState="needs_fix"` + `lastValidatorStatus="error"` unless their parent mission is already `complete`/`archived`. diff --git a/packages/engine/src/__tests__/merger-ai-cleanup.test.ts b/packages/engine/src/__tests__/merger-ai-cleanup.test.ts index b81f7d74ae..26e9647507 100644 --- 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("node:fs"); + return { + ...actual, + readdirSync: vi.fn((path: Parameters[0], options?: Parameters[1]) => { + if (String(path) === fsState.failReaddirPath) throw new Error("simulated readdir failure"); + return actual.readdirSync(path, options as never); + }), + }; +}); + const tracked = new Set(); 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[ 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"); + }); }); diff --git a/packages/engine/src/__tests__/self-healing-tempdir-sweep.test.ts b/packages/engine/src/__tests__/self-healing-tempdir-sweep.test.ts index 37416f188a..94ae335786 100644 --- 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 = {}) { +function makeStore(settings: Record = {}, getTask: () => Promise = 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 = {}) { - const { store, audits } = makeStore(settings); +function makeManager(settings: Record = {}, getTask?: () => Promise) { + 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 { + return async () => ({ id: "FN-999", column }); +} + +function missingTask(): () => Promise { + return async () => { throw new Error("Task FN-999 not found"); }; +} + async function sweep(manager: SelfHealingManager): Promise { 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); diff --git a/packages/engine/src/merger-ai.ts b/packages/engine/src/merger-ai.ts index f59d6c26a5..e06102db25 100644 --- 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, +): Promise { + 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); diff --git a/packages/engine/src/self-healing.ts b/packages/engine/src/self-healing.ts index 5e8b1be703..ec64edb904 100644 --- 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);