From d116018ed44612de510eeab7bd73aebe7f4b0cb3 Mon Sep 17 00:00:00 2001 From: Phil Larson Date: Fri, 10 Jul 2026 23:16:39 -0700 Subject: [PATCH] fix(FN-1997): bind stranded AI merge recovery to reviewed commit --- .changeset/fuzzy-ai-merge-recovery.md | 7 + .../engine/src/__tests__/merger-ai.test.ts | 30 ++- .../engine/src/__tests__/self-healing.test.ts | 4 +- packages/engine/src/merger-ai.ts | 101 +++++++--- packages/engine/src/self-healing.ts | 178 +++++++++++------- 5 files changed, 221 insertions(+), 99 deletions(-) create mode 100644 .changeset/fuzzy-ai-merge-recovery.md diff --git a/.changeset/fuzzy-ai-merge-recovery.md b/.changeset/fuzzy-ai-merge-recovery.md new file mode 100644 index 0000000000..0a8ea39da9 --- /dev/null +++ b/.changeset/fuzzy-ai-merge-recovery.md @@ -0,0 +1,7 @@ +--- +"@runfusion/fusion": patch +--- + +summary: Make stranded AI merge recovery bind to the reviewed clean-room commit. +category: fix +dev: Avoids ambiguous same-task clean-room recovery and honors cancellation before pre-prune landing. diff --git a/packages/engine/src/__tests__/merger-ai.test.ts b/packages/engine/src/__tests__/merger-ai.test.ts index 196de482dc..1f33e1ab03 100644 --- a/packages/engine/src/__tests__/merger-ai.test.ts +++ b/packages/engine/src/__tests__/merger-ai.test.ts @@ -1,5 +1,5 @@ import { describe, it, expect, vi, afterAll } from "vitest"; -import { mkdtempSync, rmSync, writeFileSync, readFileSync, existsSync, mkdirSync } from "node:fs"; +import { mkdtempSync, rmSync, writeFileSync, readFileSync, existsSync, mkdirSync, readdirSync } from "node:fs"; import { join } from "node:path"; import { tmpdir } from "node:os"; import { execSync } from "node:child_process"; @@ -277,27 +277,39 @@ describe("runAiMerge", () => { expect(emitted.some((e) => e.event === "task:merged")).toBe(true); }); - it("recovers an approved pre-existing clean-room commit before pruning and re-merging", async () => { - const { dir } = initRepoWithBranch({ branch: "fusion/fn-1" }); + it.each([ + ["modern repo-local root", "FN-1", (dir: string) => join(dir, ".worktrees", ".ai-merge")], + ["legacy .fusion root", "FN-2", (dir: string) => join(dir, ".fusion", "ai-merge")], + ["direct tmpdir root", "FN-3", (_dir: string) => tmpdir()], + ])("recovers an approved pre-existing clean-room commit from the %s before pruning and re-merging", async (_label, taskId, resolveRoot) => { + const branch = `fusion/${taskId.toLowerCase()}`; + const { dir } = initRepoWithBranch({ branch }); const mainBefore = git(dir, "rev-parse main"); - const aiMergeRoot = join(dir, ".fusion", "ai-merge"); + const aiMergeRoot = resolveRoot(dir); mkdirSync(aiMergeRoot, { recursive: true }); - const strandedRoot = mkdtempSync(join(aiMergeRoot, "fusion-ai-merge-fn-1-")); + if (aiMergeRoot === tmpdir()) { + for (const entry of readdirSync(aiMergeRoot).filter((name) => name.startsWith(`fusion-ai-merge-${taskId.toLowerCase()}-`))) { + rmSync(join(aiMergeRoot, entry), RM); + } + } + const strandedRoot = mkdtempSync(join(aiMergeRoot, `fusion-ai-merge-${taskId.toLowerCase()}-`)); tracked.add(strandedRoot); git(dir, `worktree add --detach ${strandedRoot} ${mainBefore}`); - execSync("git merge --squash fusion/fn-1", { cwd: strandedRoot, stdio: "pipe" }); + execSync(`git merge --squash ${branch}`, { cwd: strandedRoot, stdio: "pipe" }); execSync("git add -A", { cwd: strandedRoot, stdio: "pipe" }); - execSync('git commit -q -m "FN-1: recovered clean-room" -m "Fusion-Task-Id: FN-1"', { cwd: strandedRoot, stdio: "pipe" }); + execSync(`git commit -q -m "${taskId}: recovered clean-room" -m "Fusion-Task-Id: ${taskId}"`, { cwd: strandedRoot, stdio: "pipe" }); const strandedSha = git(strandedRoot, "rev-parse HEAD"); const { store, logs } = makeStore(dir, { + id: taskId, + branch, log: [ { action: "Task marked done by agent", timestamp: new Date(Date.now() - 20 * 60_000).toISOString() }, - { action: "AI merge review (pass 1): approved", timestamp: new Date(Date.now() - 12 * 60_000).toISOString() }, + { action: `AI merge review (pass 1): approved squash ${strandedSha}`, timestamp: new Date(Date.now() - 12 * 60_000).toISOString() }, ], }); const mergeAgent = vi.fn(async () => { throw new Error("should not re-merge"); }); - const result = await runAiMerge(store, dir, "FN-1", { manual: true }, { + const result = await runAiMerge(store, dir, taskId, { manual: true }, { mergeAgent, reviewAgent: vi.fn(async () => "REVIEW_VERDICT: approve"), }); diff --git a/packages/engine/src/__tests__/self-healing.test.ts b/packages/engine/src/__tests__/self-healing.test.ts index 97c03dfdfd..e3fc237404 100644 --- a/packages/engine/src/__tests__/self-healing.test.ts +++ b/packages/engine/src/__tests__/self-healing.test.ts @@ -10729,8 +10729,8 @@ describe("stranded AI merge clean-room recovery", () => { if (command.includes("git show -s --format")) { return Buffer.from("FN-5858: render headings\x1fFusion-Task-Id: FN-5858\nFusion-Task-Lineage: lineage-5858\n"); } - if (command.includes("git rev-parse --verify refs/heads/'main'")) return Buffer.from("aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa\n"); - if (command.includes("git merge-base --is-ancestor 'dddddddddddddddddddddddddddddddddddddddd' refs/heads/'main'")) { + if (command.includes("git rev-parse --verify 'refs/heads/main'")) return Buffer.from("aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa\n"); + if (command.includes("git merge-base --is-ancestor 'dddddddddddddddddddddddddddddddddddddddd' 'refs/heads/main'")) { throw new Error("not already landed"); } if (command.includes("git merge-base --is-ancestor 'aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa' 'dddddddddddddddddddddddddddddddddddddddd'")) { diff --git a/packages/engine/src/merger-ai.ts b/packages/engine/src/merger-ai.ts index ca6ec29a07..31d1adbd86 100644 --- a/packages/engine/src/merger-ai.ts +++ b/packages/engine/src/merger-ai.ts @@ -131,6 +131,16 @@ function short(sha: string): string { return /^[0-9a-f]{7,40}$/i.test(sha) ? sha.slice(0, 8) : sha; } +function getApprovedAiMergeReviewShas(task: Task | undefined): Set { + const shas = new Set(); + for (const entry of task?.log ?? []) { + if (typeof entry.action !== "string") continue; + const match = entry.action.match(/AI merge review \(pass \d+\): approved(?:\s+(?:squash|commit)\s+([0-9a-f]{7,40}))?/i); + if (match?.[1]) shas.add(match[1].toLowerCase()); + } + return shas; +} + function taskHasApprovedAiMergeReview(task: Task | undefined): boolean { return (task?.log ?? []).some((entry) => typeof entry.action === "string" @@ -138,9 +148,32 @@ function taskHasApprovedAiMergeReview(task: Task | undefined): boolean { ); } +function matchesApprovedAiMergeSha(squashSha: string, approvedShas: Set): boolean { + if (approvedShas.size === 0) return true; + const normalized = squashSha.toLowerCase(); + return Array.from(approvedShas).some((approved) => normalized === approved || normalized.startsWith(approved) || approved.startsWith(normalized)); +} + +type PreexistingAiMergeRecoveryCandidate = { + mergeRoot: string; + squashSha: string; + tipSha: string; + alreadyLanded: boolean; +}; + function listAiMergeWorktreeCandidates(taskId: string, projectRootDir: string, settings?: Settings): string[] { const prefix = `fusion-ai-merge-${taskId.toLowerCase()}-`; const roots = Array.from(new Set([resolveAiMergeRoot(projectRootDir, settings), resolveLegacyAiMergeRootPath(projectRootDir), tmpdir()])); + const testWorkerRoot = process.env.FUSION_TEST_WORKER_ROOT; + if (testWorkerRoot) { + try { + for (const entry of readdirSync(testWorkerRoot)) { + if (entry.startsWith("redir-")) roots.push(join(testWorkerRoot, entry)); + } + } catch { + // Best effort for the test harness' bounded temp-dir redirection root. + } + } const candidates: string[] = []; for (const root of roots) { let entries: string[]; @@ -159,50 +192,70 @@ async function recoverApprovedPreexistingAiMergeWorktree( integrationBranch: string, ctx: LandRepoContext, ): Promise { - const { taskId, settings, store, audit, log, allowDirtyLocalCheckoutSync, stashResolveAgent } = ctx; + const { taskId, settings, store, audit, log, allowDirtyLocalCheckoutSync, stashResolveAgent, signal } = ctx; + throwIfAborted(signal, taskId); const task = await store.getTask(taskId).catch(() => undefined); if (!taskHasApprovedAiMergeReview(task)) return null; + const approvedShas = getApprovedAiMergeReviewShas(task); const tipSha = await git(["rev-parse", "--verify", `refs/heads/${integrationBranch}`], repoRootDir); + const recoverableCandidates: PreexistingAiMergeRecoveryCandidate[] = []; for (const candidate of listAiMergeWorktreeCandidates(taskId, repoRootDir, settings)) { let mergeRoot = candidate; try { mergeRoot = realpathSync(candidate); } catch { /* keep original */ } if (activeSessionRegistry.isPathActive(candidate) || activeSessionRegistry.isPathActive(mergeRoot)) continue; try { + throwIfAborted(signal, taskId); const squashSha = await git(["rev-parse", "--verify", "HEAD"], mergeRoot); if (!squashSha || squashSha === tipSha) continue; + if (!matchesApprovedAiMergeSha(squashSha, approvedShas)) continue; const show = await git(["show", "-s", "--format=%s%x1f%b", squashSha], mergeRoot); const [subject = "", body = ""] = show.split("\x1f"); if (!getCommitTaskOwnership(taskId, task?.lineageId, subject, body).owned) continue; - if (!(await gitOk(["merge-base", "--is-ancestor", tipSha, squashSha], repoRootDir))) continue; const alreadyLanded = await gitOk(["merge-base", "--is-ancestor", squashSha, `refs/heads/${integrationBranch}`], repoRootDir); - if (!alreadyLanded) { - const land = await landSquash({ - projectRootDir: repoRootDir, - mergeRoot, - integrationBranch, - tipSha, - squashSha, - taskId, - audit, - resolveConflicts: stashResolveAgent, - allowDirtyLocalCheckoutSync, - }); - if (land.outcome !== "advanced") continue; - await log(`AI merge: recovered approved pre-existing clean-room commit ${short(squashSha)} before pruning`); - await audit.git({ type: "merge:ai-landed", target: integrationBranch, metadata: { taskId, landedSha: squashSha, source: "pre-prune-clean-room-recovery", mergeRoot } }).catch(() => undefined); - return { outcome: "landed", squashSha, localSync: land.localSync, tipSha, integrationBranch }; - } - - await log(`AI merge: recovered already-landed clean-room commit ${short(squashSha)} before pruning`); - return { outcome: "landed", squashSha, localSync: "skipped-other-branch", tipSha, integrationBranch }; + const tipIsAncestor = await gitOk(["merge-base", "--is-ancestor", tipSha, squashSha], repoRootDir); + if (!alreadyLanded && !tipIsAncestor) continue; + recoverableCandidates.push({ mergeRoot, squashSha, tipSha, alreadyLanded }); } catch (err: unknown) { await log(`AI merge: skipped pre-existing clean-room recovery candidate ${mergeRoot}: ${getErrorMessage(err)}`); } } - return null; + + /* + FNXC:AIMergeRecovery 2026-07-10-23:06: + Approved clean-room recovery must bind the candidate commit to the reviewed squash. New review logs carry the squash SHA; legacy logs without a SHA can recover only when exactly one same-task candidate is possible, otherwise recovery defers to the normal merge path rather than finalizing the wrong clean room. + */ + if (recoverableCandidates.length !== 1) { + if (recoverableCandidates.length > 1) { + await log(`AI merge: skipped pre-existing clean-room recovery because ${recoverableCandidates.length} same-task approved candidates were ambiguous`); + } + return null; + } + + const selected = recoverableCandidates[0]; + throwIfAborted(signal, taskId); + if (!selected.alreadyLanded) { + const land = await landSquash({ + projectRootDir: repoRootDir, + mergeRoot: selected.mergeRoot, + integrationBranch, + tipSha: selected.tipSha, + squashSha: selected.squashSha, + taskId, + audit, + resolveConflicts: stashResolveAgent, + allowDirtyLocalCheckoutSync, + }); + if (land.outcome !== "advanced") return null; + await log(`AI merge: recovered approved pre-existing clean-room commit ${short(selected.squashSha)} before pruning`); + await audit.git({ type: "merge:ai-landed", target: integrationBranch, metadata: { taskId, landedSha: selected.squashSha, source: "pre-prune-clean-room-recovery", mergeRoot: selected.mergeRoot } }).catch(() => undefined); + return { outcome: "landed", squashSha: selected.squashSha, localSync: land.localSync, tipSha: selected.tipSha, integrationBranch }; + } + + await log(`AI merge: recovered already-landed clean-room commit ${short(selected.squashSha)} before pruning`); + return { outcome: "landed", squashSha: selected.squashSha, localSync: "skipped-other-branch", tipSha: selected.tipSha, integrationBranch }; } export { @@ -1620,7 +1673,7 @@ async function mergeAndReview(input: { }); if (verdict.verdict === "approve") { - await log(`AI merge review (pass ${attempt + 1}): approved`); + await log(`AI merge review (pass ${attempt + 1}): approved squash ${head}`); return head; } diff --git a/packages/engine/src/self-healing.ts b/packages/engine/src/self-healing.ts index 8d063be176..99dab5910f 100644 --- a/packages/engine/src/self-healing.ts +++ b/packages/engine/src/self-healing.ts @@ -8980,6 +8980,16 @@ export class SelfHealingManager { } } + private getApprovedAiMergeReviewShas(task: Task): Set { + const shas = new Set(); + for (const entry of task.log ?? []) { + if (typeof entry.action !== "string") continue; + const match = entry.action.match(/AI merge review \(pass \d+\): approved(?:\s+(?:squash|commit)\s+([0-9a-f]{7,40}))?/i); + if (match?.[1]) shas.add(match[1].toLowerCase()); + } + return shas; + } + private hasApprovedAiMergeReview(task: Task): boolean { return (task.log ?? []).some((entry) => typeof entry.action === "string" @@ -8987,12 +8997,28 @@ export class SelfHealingManager { ); } + private matchesApprovedAiMergeSha(squashSha: string, approvedShas: Set): boolean { + if (approvedShas.size === 0) return true; + const normalized = squashSha.toLowerCase(); + return Array.from(approvedShas).some((approved) => normalized === approved || normalized.startsWith(approved) || approved.startsWith(normalized)); + } + private async listAiMergeWorktreeCandidates(taskId: string, settings: Settings): Promise { const roots = Array.from(new Set([ resolveRepoLocalAiMergeRoot(this.options.rootDir, settings), resolveLegacyAiMergeRootPath(this.options.rootDir), tmpdir(), ])); + const testWorkerRoot = process.env.FUSION_TEST_WORKER_ROOT; + if (testWorkerRoot) { + try { + for (const entry of readdirSync(testWorkerRoot)) { + if (entry.startsWith("redir-")) roots.push(join(testWorkerRoot, entry)); + } + } catch { + // Best effort for the test harness' bounded temp-dir redirection root. + } + } const prefix = `fusion-ai-merge-${taskId.toLowerCase()}-`; const paths: string[] = []; for (const root of roots) { @@ -9026,6 +9052,16 @@ export class SelfHealingManager { phase: "recover-stranded-ai-merge-commit", }); + const approvedShas = this.getApprovedAiMergeReviewShas(task); + const refName = `refs/heads/${integrationBranch}`; + const recoverableCandidates: Array<{ + canonicalCandidate: string; + strandedSha: string; + tipSha: string; + alreadyAncestor: boolean; + landedFiles: string[]; + }> = []; + for (const candidate of candidates) { let canonicalCandidate = candidate; try { canonicalCandidate = realpathSync(candidate); } catch { /* keep original */ } @@ -9034,7 +9070,7 @@ export class SelfHealingManager { try { const { stdout: headStdout } = await execAsync("git rev-parse --verify HEAD", { cwd: canonicalCandidate, timeout: 30_000 }); const strandedSha = headStdout.trim(); - if (!strandedSha) continue; + if (!strandedSha || !this.matchesApprovedAiMergeSha(strandedSha, approvedShas)) continue; const { stdout: showStdout } = await execAsync(`git show -s --format=%s%x1f%b ${shellQuote(strandedSha)}`, { cwd: canonicalCandidate, @@ -9045,14 +9081,14 @@ export class SelfHealingManager { const ownership = getCommitTaskOwnership(task.id, task.lineageId, subject, body); if (!ownership.owned) continue; - const { stdout: tipStdout } = await execAsync(`git rev-parse --verify refs/heads/${shellQuote(integrationBranch)}`, { + const { stdout: tipStdout } = await execAsync(`git rev-parse --verify ${shellQuote(refName)}`, { cwd: this.options.rootDir, timeout: 30_000, }); const tipSha = tipStdout.trim(); if (!tipSha) continue; - const alreadyAncestor = await execAsync(`git merge-base --is-ancestor ${shellQuote(strandedSha)} refs/heads/${shellQuote(integrationBranch)}`, { + const alreadyAncestor = await execAsync(`git merge-base --is-ancestor ${shellQuote(strandedSha)} ${shellQuote(refName)}`, { cwd: this.options.rootDir, timeout: 30_000, }).then(() => true, () => false); @@ -9067,71 +9103,85 @@ export class SelfHealingManager { timeout: 30_000, maxBuffer: 1024 * 1024, }).then(({ stdout }) => stdout.split("\n").map((line) => line.trim()).filter(Boolean), () => []); - - if (!alreadyAncestor) { - const currentBranch = await execAsync("git rev-parse --abbrev-ref HEAD", { cwd: this.options.rootDir, timeout: 30_000 }) - .then(({ stdout }) => stdout.trim(), () => ""); - if (currentBranch === integrationBranch) { - const head = await execAsync("git rev-parse HEAD", { cwd: this.options.rootDir, timeout: 30_000 }) - .then(({ stdout }) => stdout.trim(), () => ""); - const dirty = await execAsync("git status --porcelain", { cwd: this.options.rootDir, timeout: 30_000 }) - .then(({ stdout }) => stdout.trim().length > 0, () => true); - if (head !== tipSha || dirty) continue; - await execAsync(`git merge --ff-only ${shellQuote(strandedSha)}`, { cwd: this.options.rootDir, timeout: 120_000 }); - } else { - const advanced = await advanceIntegrationBranchRef({ - rootDir: canonicalCandidate, - projectRootDir: this.options.rootDir, - integrationBranch, - newSha: strandedSha, - expectedCurrentSha: tipSha, - taskId: task.id, - audit: auditor, - }); - if (!advanced.advanced) continue; - } - } - - const result: MergeResult = { - task, - branch: task.branch ?? resolveTaskWorkingBranch(task), - merged: true, - noOp: false, - ok: true, - commitSha: strandedSha, - landedFiles, - mergeConfirmed: true, - worktreeRemoved: false, - branchDeleted: false, - }; - const finalized = await finalizeProvenAutoMergeTask({ - store: this.store, - taskId: task.id, - result, - audit: auditor, - auditAgentId: "self-healing", - auditPhase: "recover-stranded-ai-merge-commit", - source: "self-healing", - log: async (message) => { - await this.store.logEntry(task.id, message).catch(() => undefined); - }, - }); - if (finalized.outcome === "done" || finalized.outcome === "already-done") { - await this.store.logEntry( - task.id, - `Auto-recovered stranded AI merge clean-room commit ${strandedSha.slice(0, 8)} — advanced ${integrationBranch} and finalized task`, - ); - await auditor.git({ - type: "merge:ai-landed", - target: integrationBranch, - metadata: { taskId: task.id, landedSha: strandedSha, source: "self-healing-stranded-clean-room", path: canonicalCandidate }, - }); - return true; - } + recoverableCandidates.push({ canonicalCandidate, strandedSha, tipSha, alreadyAncestor, landedFiles }); } catch (err: unknown) { log.warn(`recoverApprovedStrandedAiMergeCommit: ${task.id} candidate ${candidate} skipped: ${getErrorMessage(err)}`); } } + + /* + FNXC:AIMergeRecovery 2026-07-10-23:06: + Self-healing finalization must recover the exact reviewed clean-room commit. When historical approval logs do not include a SHA, only a single eligible same-task candidate is safe; multiple candidates are left for normal merge/review instead of guessing by filesystem order. + */ + if (recoverableCandidates.length !== 1) { + if (recoverableCandidates.length > 1) { + await this.store.logEntry(task.id, `Skipped stranded AI merge recovery: ${recoverableCandidates.length} approved clean-room candidates were ambiguous`).catch(() => undefined); + } + return false; + } + + const selected = recoverableCandidates[0]; + if (!selected) return false; + if (!selected.alreadyAncestor) { + const currentBranch = await execAsync("git rev-parse --abbrev-ref HEAD", { cwd: this.options.rootDir, timeout: 30_000 }) + .then(({ stdout }) => stdout.trim(), () => ""); + if (currentBranch === integrationBranch) { + const head = await execAsync("git rev-parse HEAD", { cwd: this.options.rootDir, timeout: 30_000 }) + .then(({ stdout }) => stdout.trim(), () => ""); + const dirty = await execAsync("git status --porcelain", { cwd: this.options.rootDir, timeout: 30_000 }) + .then(({ stdout }) => stdout.trim().length > 0, () => true); + if (head !== selected.tipSha || dirty) return false; + await execAsync(`git merge --ff-only ${shellQuote(selected.strandedSha)}`, { cwd: this.options.rootDir, timeout: 120_000 }); + } else { + const advanced = await advanceIntegrationBranchRef({ + rootDir: this.options.rootDir, + projectRootDir: this.options.rootDir, + integrationBranch, + newSha: selected.strandedSha, + expectedCurrentSha: selected.tipSha, + taskId: task.id, + audit: auditor, + }); + if (!advanced.advanced) return false; + } + } + + const result: MergeResult = { + task, + branch: task.branch ?? resolveTaskWorkingBranch(task), + merged: true, + noOp: false, + ok: true, + commitSha: selected.strandedSha, + landedFiles: selected.landedFiles, + mergeConfirmed: true, + worktreeRemoved: false, + branchDeleted: false, + }; + const finalized = await finalizeProvenAutoMergeTask({ + store: this.store, + taskId: task.id, + result, + audit: auditor, + auditAgentId: "self-healing", + auditPhase: "recover-stranded-ai-merge-commit", + source: "self-healing", + log: async (message) => { + await this.store.logEntry(task.id, message).catch(() => undefined); + }, + }); + if (finalized.outcome === "done" || finalized.outcome === "already-done") { + await this.store.logEntry( + task.id, + `Auto-recovered stranded AI merge clean-room commit ${selected.strandedSha.slice(0, 8)} — advanced ${integrationBranch} and finalized task`, + ); + await auditor.git({ + type: "merge:ai-landed", + target: integrationBranch, + metadata: { taskId: task.id, landedSha: selected.strandedSha, source: "self-healing-stranded-clean-room", path: selected.canonicalCandidate }, + }); + return true; + } return false; }