feat(FN-3865): complete Steps 3-6 — recover already-merged review tasks

This commit is contained in:
Fusion
2026-05-09 12:19:28 -07:00
committed by gsxdsm
parent a1f89a1d31
commit 6a92d62923
4 changed files with 426 additions and 0 deletions

View File

@@ -0,0 +1,5 @@
---
"@runfusion/fusion": patch
---
Add `recoverAlreadyMergedReviewTasks()` self-healing sweep to recover phantom-merge-guard false positives. Detects tasks whose content already landed on the integration branch (via Fusion-Task-Id trailer, branch ancestry, or git patch-id walk) and reconciles them to `done` with proper merge metadata.

View File

@@ -0,0 +1,214 @@
import { afterEach, describe, expect, it, vi } from "vitest";
import { execSync, spawnSync } from "node:child_process";
import { existsSync, mkdtempSync, mkdirSync, rmSync, writeFileSync } from "node:fs";
import os from "node:os";
import path from "node:path";
import { EventEmitter } from "node:events";
import type { Settings, Task, TaskStore } from "@fusion/core";
import { SelfHealingManager } from "../self-healing.js";
const hasGit = spawnSync("git", ["--version"], { stdio: "pipe" }).status === 0;
const describeIfGit = hasGit ? describe : describe.skip;
function git(repo: string, command: string): string {
return execSync(command, { cwd: repo, encoding: "utf-8", stdio: ["pipe", "pipe", "pipe"] }).trim();
}
type TaskMap = Map<string, Task & { comments?: string[] }>;
function createStore(tasks: TaskMap, settings: Partial<Settings> = {}): TaskStore & EventEmitter {
const emitter = new EventEmitter();
const mergedSettings: Settings = {
globalPause: false,
enginePaused: false,
maintenanceIntervalMs: 0,
taskStuckTimeoutMs: 60_000,
autoMerge: false,
...settings,
} as Settings;
const store = Object.assign(emitter, {
getSettings: vi.fn(async () => mergedSettings),
listTasks: vi.fn(async ({ column, includeArchived }: { column?: string; includeArchived?: boolean } = {}) => {
const values = [...tasks.values()];
return values.filter((task) => {
if (!includeArchived && task.column === "archived") return false;
if (column && task.column !== column) return false;
return true;
});
}),
updateTask: vi.fn(async (id: string, updates: Partial<Task>) => {
const current = tasks.get(id)!;
tasks.set(id, { ...current, ...updates, updatedAt: new Date().toISOString() });
return tasks.get(id);
}),
moveTask: vi.fn(async (id: string, column: Task["column"]) => {
const current = tasks.get(id)!;
tasks.set(id, { ...current, column, columnMovedAt: new Date().toISOString(), updatedAt: new Date().toISOString() });
}),
logEntry: vi.fn(async (id: string, message: string) => {
const current = tasks.get(id)!;
const log = current.log ?? [];
tasks.set(id, { ...current, log: [...log, { timestamp: new Date().toISOString(), action: message }] as any });
}),
walCheckpoint: vi.fn(() => ({ busy: 0, log: 0, checkpointed: 0 })),
archiveTaskAndCleanup: vi.fn(async () => ({})),
clearStaleExecutionStartBranchReferences: vi.fn(() => []),
getTask: vi.fn(async (id: string) => tasks.get(id)),
updateSettings: vi.fn(async () => mergedSettings),
mergeTask: vi.fn(async () => undefined),
getRootDir: vi.fn(() => ""),
}) as unknown as TaskStore & EventEmitter;
return store;
}
describeIfGit("SelfHealingManager recoverAlreadyMergedReviewTasks (real git)", () => {
const repos: string[] = [];
afterEach(() => {
for (const repo of repos.splice(0)) {
rmSync(repo, { recursive: true, force: true });
}
});
function setupRepo(): string {
const repo = mkdtempSync(path.join(os.tmpdir(), "fn-3865-"));
repos.push(repo);
git(repo, "git init -b main");
git(repo, 'git config user.email "test@example.com"');
git(repo, 'git config user.name "Test"');
git(repo, "git commit --allow-empty -m 'init'");
return repo;
}
it("recovers via trailer match and removes worktree", async () => {
const repo = setupRepo();
mkdirSync(path.join(repo, "src"), { recursive: true });
writeFileSync(path.join(repo, "src", "file.txt"), "trailer\n", "utf-8");
git(repo, "git add src/file.txt && git commit -m 'feat: landed' -m 'Fusion-Task-Id: FN-TEST-1'");
const landedSha = git(repo, "git rev-parse HEAD");
const worktreePath = path.join(repo, ".worktrees", "fn-test-1");
mkdirSync(path.dirname(worktreePath), { recursive: true });
git(repo, `git worktree add ${JSON.stringify(worktreePath)} -b fusion/fn-test-1`);
const tasks: TaskMap = new Map([
["FN-TEST-1", { id: "FN-TEST-1", column: "in-review", status: "failed", mergeRetries: 3, paused: false, baseBranch: "main", branch: "fusion/fn-test-1", worktree: worktreePath, steps: [], log: [], updatedAt: new Date().toISOString() } as Task],
]);
const store = createStore(tasks);
const manager = new SelfHealingManager(store, { rootDir: repo, getExecutingTaskIds: () => new Set() });
await (manager as any).runMaintenance();
const task = tasks.get("FN-TEST-1")!;
expect(task.column).toBe("done");
expect(task.status).toBeNull();
expect(task.mergeRetries).toBe(0);
expect(task.mergeDetails?.commitSha).toBe(landedSha);
expect(task.mergeDetails?.strategy).toBe("squash");
expect(task.mergeDetails?.branch).toBe("main");
expect(existsSync(worktreePath)).toBe(false);
expect(git(repo, "git worktree list")).not.toContain(worktreePath);
});
it("recovers via patch-id fallback", async () => {
const repo = setupRepo();
git(repo, "git checkout -b fusion/fn-test-2");
mkdirSync(path.join(repo, "src"), { recursive: true });
writeFileSync(path.join(repo, "src", "patch.txt"), "patch-a\n", "utf-8");
git(repo, "git add src/patch.txt && git commit -m 'task branch commit'");
const branchTip = git(repo, "git rev-parse HEAD");
git(repo, "git checkout main");
mkdirSync(path.join(repo, "src"), { recursive: true });
writeFileSync(path.join(repo, "src", "patch.txt"), "patch-a\n", "utf-8");
git(repo, "git add src/patch.txt && git commit -m 'land equivalent change'");
const landedSha = git(repo, "git rev-parse HEAD");
const worktreePath = path.join(repo, ".worktrees", "fn-test-2");
mkdirSync(path.dirname(worktreePath), { recursive: true });
git(repo, `git worktree add ${JSON.stringify(worktreePath)} fusion/fn-test-2`);
const tasks: TaskMap = new Map([
["FN-TEST-2", { id: "FN-TEST-2", column: "in-review", status: "failed", mergeRetries: 3, paused: false, baseBranch: "main", branch: "fusion/fn-test-2", baseCommitSha: git(repo, "git merge-base main fusion/fn-test-2"), worktree: worktreePath, steps: [], log: [], updatedAt: new Date().toISOString() } as Task],
]);
const store = createStore(tasks);
const manager = new SelfHealingManager(store, { rootDir: repo, getExecutingTaskIds: () => new Set() });
expect(branchTip).toBeTruthy();
await (manager as any).runMaintenance();
const task = tasks.get("FN-TEST-2")!;
expect(task.column).toBe("done");
expect(task.mergeDetails?.commitSha).toBe(landedSha);
});
it("does nothing when no match exists", async () => {
const repo = setupRepo();
git(repo, "git checkout -b fusion/fn-test-3");
mkdirSync(path.join(repo, "src"), { recursive: true });
writeFileSync(path.join(repo, "src", "no-match.txt"), "branch-only\n", "utf-8");
git(repo, "git add src/no-match.txt && git commit -m 'branch only'");
git(repo, "git checkout main");
const worktreePath = path.join(repo, ".worktrees", "fn-test-3");
mkdirSync(path.dirname(worktreePath), { recursive: true });
git(repo, `git worktree add ${JSON.stringify(worktreePath)} fusion/fn-test-3`);
const tasks: TaskMap = new Map([
["FN-TEST-3", { id: "FN-TEST-3", column: "in-review", status: "failed", mergeRetries: 3, paused: false, baseBranch: "main", branch: "fusion/fn-test-3", worktree: worktreePath, steps: [], log: [], updatedAt: new Date().toISOString() } as Task],
]);
const store = createStore(tasks);
const manager = new SelfHealingManager(store, { rootDir: repo, getExecutingTaskIds: () => new Set() });
await (manager as any).runMaintenance();
const task = tasks.get("FN-TEST-3")!;
expect(task.column).toBe("in-review");
expect(task.status).toBe("failed");
expect(task.mergeRetries).toBe(3);
expect(existsSync(worktreePath)).toBe(true);
});
it("is idempotent across two maintenance passes", async () => {
const repo = setupRepo();
mkdirSync(path.join(repo, "src"), { recursive: true });
writeFileSync(path.join(repo, "src", "idempotent.txt"), "same\n", "utf-8");
git(repo, "git add src/idempotent.txt && git commit -m 'feat: done' -m 'Fusion-Task-Id: FN-TEST-4'");
const worktreePath = path.join(repo, ".worktrees", "fn-test-4");
mkdirSync(path.dirname(worktreePath), { recursive: true });
git(repo, `git worktree add ${JSON.stringify(worktreePath)} -b fusion/fn-test-4`);
const tasks: TaskMap = new Map([
["FN-TEST-4", { id: "FN-TEST-4", column: "in-review", status: "failed", mergeRetries: 3, paused: false, baseBranch: "main", branch: "fusion/fn-test-4", worktree: worktreePath, steps: [], log: [], updatedAt: new Date().toISOString() } as Task],
]);
const store = createStore(tasks);
const manager = new SelfHealingManager(store, { rootDir: repo, getExecutingTaskIds: () => new Set() });
await (manager as any).runMaintenance();
const firstRecoveryLogs = (store.logEntry as any).mock.calls.filter((call: unknown[]) => String(call[1]).includes("phantom-merge-guard false positive")).length;
await (manager as any).runMaintenance();
const secondRecoveryLogs = (store.logEntry as any).mock.calls.filter((call: unknown[]) => String(call[1]).includes("phantom-merge-guard false positive")).length;
expect(firstRecoveryLogs).toBe(1);
expect(secondRecoveryLogs).toBe(1);
});
it("short-circuits when paused", async () => {
const repo = setupRepo();
const tasks: TaskMap = new Map([
["FN-TEST-5", { id: "FN-TEST-5", column: "in-review", status: "failed", mergeRetries: 3, paused: false, baseBranch: "main", branch: "fusion/fn-test-5", steps: [], log: [], updatedAt: new Date().toISOString() } as Task],
]);
const globalPausedStore = createStore(tasks, { globalPause: true, enginePaused: false });
const globalPausedManager = new SelfHealingManager(globalPausedStore, { rootDir: repo, getExecutingTaskIds: () => new Set() });
await globalPausedManager.recoverAlreadyMergedReviewTasks();
expect(globalPausedStore.listTasks).not.toHaveBeenCalled();
const enginePausedStore = createStore(tasks, { globalPause: false, enginePaused: true });
const enginePausedManager = new SelfHealingManager(enginePausedStore, { rootDir: repo, getExecutingTaskIds: () => new Set() });
await enginePausedManager.recoverAlreadyMergedReviewTasks();
expect(enginePausedStore.listTasks).not.toHaveBeenCalled();
});
});

View File

@@ -2365,6 +2365,119 @@ describe("SelfHealingManager", () => {
}); });
}); });
describe("recoverAlreadyMergedReviewTasks", () => {
it("short-circuits when globalPause or enginePaused is active", async () => {
const managerWithRecovery = new SelfHealingManager(store, { rootDir: "/tmp/test-project" });
(store.getSettings as ReturnType<typeof vi.fn>).mockResolvedValue({ globalPause: true, enginePaused: false });
const pausedResult = await managerWithRecovery.recoverAlreadyMergedReviewTasks();
expect(pausedResult).toBe(0);
expect(store.listTasks).not.toHaveBeenCalled();
(store.getSettings as ReturnType<typeof vi.fn>).mockResolvedValue({ globalPause: false, enginePaused: true });
const enginePausedResult = await managerWithRecovery.recoverAlreadyMergedReviewTasks();
expect(enginePausedResult).toBe(0);
expect(store.listTasks).not.toHaveBeenCalled();
expect(store.updateTask).not.toHaveBeenCalled();
managerWithRecovery.stop();
});
it("filters out non-candidates", async () => {
const managerWithRecovery = new SelfHealingManager(store, {
rootDir: "/tmp/test-project",
getExecutingTaskIds: () => new Set(["FN-executing"]),
});
(store.getSettings as ReturnType<typeof vi.fn>).mockResolvedValue({ globalPause: false, enginePaused: false });
(store.listTasks as ReturnType<typeof vi.fn>).mockResolvedValue([
{ id: "FN-ok-status", column: "in-review", paused: false, status: null, mergeRetries: 3, mergeDetails: undefined, log: [] },
{ id: "FN-low-retries", column: "in-review", paused: false, status: "failed", mergeRetries: 2, mergeDetails: undefined, log: [] },
{ id: "FN-paused", column: "in-review", paused: true, status: "failed", mergeRetries: 3, mergeDetails: undefined, log: [] },
{ id: "FN-executing", column: "in-review", paused: false, status: "failed", mergeRetries: 3, mergeDetails: undefined, log: [] },
{ id: "FN-confirmed", column: "in-review", paused: false, status: "failed", mergeRetries: 3, mergeDetails: { mergeConfirmed: true }, log: [] },
]);
const result = await managerWithRecovery.recoverAlreadyMergedReviewTasks();
expect(result).toBe(0);
expect(store.updateTask).not.toHaveBeenCalled();
expect(store.moveTask).not.toHaveBeenCalled();
managerWithRecovery.stop();
});
it("leaves tasks untouched when no landed commit is detected", async () => {
const managerWithRecovery = new SelfHealingManager(store, { rootDir: "/tmp/test-project" });
(store.getSettings as ReturnType<typeof vi.fn>).mockResolvedValue({ globalPause: false, enginePaused: false });
(store.listTasks as ReturnType<typeof vi.fn>).mockResolvedValue([
{ id: "FN-1", column: "in-review", paused: false, status: "failed", mergeRetries: 3, mergeDetails: undefined, branch: "fusion/fn-1", log: [] },
]);
mockedExecSync.mockImplementation(() => {
throw new Error("missing branch");
});
const result = await managerWithRecovery.recoverAlreadyMergedReviewTasks();
expect(result).toBe(0);
expect(store.updateTask).not.toHaveBeenCalled();
expect(store.moveTask).not.toHaveBeenCalled();
expect(store.logEntry).not.toHaveBeenCalled();
managerWithRecovery.stop();
});
it("isolates per-task failures and still recovers later candidates", async () => {
const managerWithRecovery = new SelfHealingManager(store, { rootDir: "/tmp/test-project" });
(store.getSettings as ReturnType<typeof vi.fn>).mockResolvedValue({ globalPause: false, enginePaused: false });
(store.listTasks as ReturnType<typeof vi.fn>).mockResolvedValue([
{ id: "FN-throw", column: "in-review", paused: false, status: "failed", mergeRetries: 3, mergeDetails: undefined, baseBranch: "main", branch: "fusion/fn-throw", worktree: "/tmp/wt1", log: [] },
{ id: "FN-hit", column: "in-review", paused: false, status: "failed", mergeRetries: 3, mergeDetails: undefined, baseBranch: "main", branch: "fusion/fn-hit", worktree: "/tmp/wt2", log: [] },
]);
mockedExistsSync.mockReturnValue(false);
mockedExecSync.mockImplementation((command: string | Buffer) => {
const cmd = String(command);
if (cmd.includes("Fusion-Task-Id: FN-throw")) throw new Error("trailer fail");
if (cmd.includes("rev-parse --verify") && cmd.includes("fusion/fn-throw")) throw new Error("rev fail");
if (cmd.includes("Fusion-Task-Id: FN-hit")) return "abc123\n" as any;
if (cmd.includes("rev-parse --verify") && cmd.includes("fusion/fn-hit")) return "tip-hit\n" as any;
return "" as any;
});
const result = await managerWithRecovery.recoverAlreadyMergedReviewTasks();
expect(result).toBe(1);
expect(store.updateTask).toHaveBeenCalledWith("FN-hit", expect.objectContaining({ column: "done" }));
expect(store.updateTask).not.toHaveBeenCalledWith("FN-throw", expect.anything());
managerWithRecovery.stop();
});
it("is idempotent across repeated sweeps", async () => {
const managerWithRecovery = new SelfHealingManager(store, { rootDir: "/tmp/test-project" });
(store.getSettings as ReturnType<typeof vi.fn>).mockResolvedValue({ globalPause: false, enginePaused: false });
(store.listTasks as ReturnType<typeof vi.fn>)
.mockResolvedValueOnce([
{ id: "FN-1", column: "in-review", paused: false, status: "failed", mergeRetries: 3, mergeDetails: undefined, baseBranch: "main", branch: "fusion/fn-1", worktree: "/tmp/wt", log: [] },
])
.mockResolvedValueOnce([
{ id: "FN-1", column: "done", paused: false, status: null, mergeRetries: 0, mergeDetails: { mergeConfirmed: true }, baseBranch: "main", log: [] },
]);
mockedExecSync.mockImplementation((command: string | Buffer) => {
if (String(command).includes("Fusion-Task-Id: FN-1")) return "abc123\n" as any;
return "tip\n" as any;
});
mockedExistsSync.mockReturnValue(false);
const first = await managerWithRecovery.recoverAlreadyMergedReviewTasks();
const second = await managerWithRecovery.recoverAlreadyMergedReviewTasks();
expect(first).toBe(1);
expect(second).toBe(0);
managerWithRecovery.stop();
});
});
describe("recoverReviewTasksWithFailedPreMergeSteps", () => { describe("recoverReviewTasksWithFailedPreMergeSteps", () => {
const baseTask = { const baseTask = {
id: "FN-1572", id: "FN-1572",

View File

@@ -832,6 +832,7 @@ export class SelfHealingManager {
{ name: "recover-stale-merging-status", fn: () => this.recoverStaleMergingStatus() }, { name: "recover-stale-merging-status", fn: () => this.recoverStaleMergingStatus() },
{ name: "recover-mergeable-review", fn: () => this.recoverMergeableReviewTasks() }, { name: "recover-mergeable-review", fn: () => this.recoverMergeableReviewTasks() },
{ name: "recover-merged-review", fn: () => this.recoverMergedReviewTasks() }, { name: "recover-merged-review", fn: () => this.recoverMergedReviewTasks() },
{ name: "recover-already-merged-review", fn: () => this.recoverAlreadyMergedReviewTasks() },
{ name: "recover-misclassified-failures", fn: () => this.recoverMisclassifiedFailures() }, { name: "recover-misclassified-failures", fn: () => this.recoverMisclassifiedFailures() },
{ name: "recover-no-progress-no-task-done", fn: () => this.recoverNoProgressNoTaskDoneFailures() }, { name: "recover-no-progress-no-task-done", fn: () => this.recoverNoProgressNoTaskDoneFailures() },
{ name: "recover-partial-progress-no-task-done", fn: () => this.recoverPartialProgressNoTaskDoneFailures() }, { name: "recover-partial-progress-no-task-done", fn: () => this.recoverPartialProgressNoTaskDoneFailures() },
@@ -1603,6 +1604,99 @@ export class SelfHealingManager {
} }
} }
/**
* Recover retry-exhausted failed review tasks whose content already landed on
* the integration branch via a non-canonical merge lineage.
*
* Candidate filter:
* - `column === "in-review"`
* - not paused
* - `status === "failed"`
* - `(mergeRetries ?? 0) >= MAX_AUTO_MERGE_RETRIES`
* - `mergeDetails.mergeConfirmed !== true`
* - not actively executing
*
* Detection order (first match wins):
* 1. Fusion-Task-Id trailer lookup on the base branch
* 2. Task branch ancestry + task-id grep on first-parent base lineage
* 3. Patch-id match between task branch diff and recent base-branch commits
*
* Idempotency: recovered tasks are moved to `done`, status/error are cleared,
* and mergeRetries reset to 0, so subsequent sweeps will not match them.
*/
async recoverAlreadyMergedReviewTasks(): Promise<number> {
try {
const settings = await this.store.getSettings();
if (settings.globalPause || settings.enginePaused) return 0;
const executingIds = this.options.getExecutingTaskIds?.() ?? new Set<string>();
const tasks = await this.store.listTasks({ column: "in-review", slim: true });
const candidates = tasks.filter((task) =>
task.column === "in-review" &&
!task.paused &&
task.status === "failed" &&
(task.mergeRetries ?? 0) >= MAX_AUTO_MERGE_RETRIES &&
task.mergeDetails?.mergeConfirmed !== true &&
!executingIds.has(task.id),
);
if (candidates.length === 0) return 0;
let recovered = 0;
for (const task of candidates) {
try {
const baseBranch = task.baseBranch || task.executionStartBranch || "main";
if (!baseBranch) continue;
const landed = await this.findAlreadyMergedTaskCommit({
taskId: task.id,
repoDir: this.options.rootDir,
baseBranch,
taskBranch: task.branch,
baseCommitSha: task.baseCommitSha,
});
if (!landed) continue;
const mergeDetails: MergeDetails = {
commitSha: landed.sha,
strategy: "squash",
branch: baseBranch,
mergedAt: new Date().toISOString(),
mergeConfirmed: true,
prNumber: task.prInfo?.number,
};
await this.store.updateTask(task.id, {
column: "done",
status: null,
error: null,
mergeRetries: 0,
mergeDetails,
});
await this.store.moveTask(task.id, "done");
await this.store.logEntry(
task.id,
`Auto-recovered: phantom-merge-guard false positive — content found on ${baseBranch} at ${landed.sha.slice(0, 8)} via ${landed.strategy}`,
);
await this.cleanupWorktreeOnly(task);
recovered++;
} catch (err: unknown) {
const errorMessage = err instanceof Error ? err.message : String(err);
log.warn(`recoverAlreadyMergedReviewTasks: failed for ${task.id}: ${errorMessage}`);
}
}
if (recovered > 0) {
log.log(`Recovered ${recovered} already-merged retry-exhausted review task(s) → done`);
}
return recovered;
} catch (err: unknown) {
const errorMessage = err instanceof Error ? err.message : String(err);
log.error(`Already-merged review recovery failed: ${errorMessage}`);
return 0;
}
}
/** /**
* Recover tasks in `in-review` marked as `failed` where all steps are * Recover tasks in `in-review` marked as `failed` where all steps are
* actually done. This catches the case where an agent completed all work * actually done. This catches the case where an agent completed all work