root/fusion
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat(FN-3856): add stuck merge deadlock recovery sweep and self-healing aut

Browse Source 
Merges deadlock recovery for stuck merge tasks (FN-3856) — the engine now runs an already-merged sweep before pausing blocked tasks, and recovers metadata alignment with MergeDetails type. Also adds memory backup CLI commands with core support, a resizable split pane to the mailbox view, peer-agent

Fusion-Task-Id: FN-3856
This commit is contained in:
Fusion
2026-05-09 15:16:10 -07:00
committed by gsxdsm
parent 0b69b9917e
commit 3e64668c6a
3 changed files with 309 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files

5
.changeset/FN-3856-deadlock-recovery.md Normal file
View File

@@ -0,0 +1,5 @@
---
"@runfusion/fusion": patch
---
Auto-recover stuck merge deadlocks where task content is already on main.

188
packages/engine/src/__tests__/self-healing.test.ts
View File

@@ -434,6 +434,7 @@ describe("SelfHealingManager", () => {
} as unknown as Settings);
const recoverNoProgressNoTaskDoneFailures = vi.spyOn(manager, "recoverNoProgressNoTaskDoneFailures").mockResolvedValue(1);
const recoverCompletedTasks = vi.spyOn(manager, "recoverCompletedTasks").mockResolvedValue(1);
const recoverStuckMergeDeadlocks = vi.spyOn(manager, "recoverStuckMergeDeadlocks").mockResolvedValue(1);
const recoverMisclassifiedFailures = vi.spyOn(manager, "recoverMisclassifiedFailures").mockResolvedValue(1);
const recoverPartialProgressNoTaskDoneFailures = vi.spyOn(manager, "recoverPartialProgressNoTaskDoneFailures").mockResolvedValue(1);
const recoverOrphanedExecutions = vi.spyOn(manager, "recoverOrphanedExecutions").mockResolvedValue(1);
@@ -444,6 +445,7 @@ describe("SelfHealingManager", () => {
expect(recoverNoProgressNoTaskDoneFailures).toHaveBeenCalledTimes(1);
expect(recoverCompletedTasks).toHaveBeenCalledTimes(1);
expect(recoverStuckMergeDeadlocks).toHaveBeenCalledTimes(1);
expect(recoverMisclassifiedFailures).toHaveBeenCalledTimes(1);
expect(recoverPartialProgressNoTaskDoneFailures).toHaveBeenCalledTimes(1);
expect(recoverOrphanedExecutions).toHaveBeenCalledTimes(1);
@@ -2365,6 +2367,188 @@ describe("SelfHealingManager", () => {
});
});
describe("recoverStuckMergeDeadlocks", () => {
const baseSettings = { globalPause: false, enginePaused: false, defaultBaseBranch: "main" } as unknown as Settings;
it("recovers phantom-merged deadlocks, moves task to done, and clears blocked dependents", async () => {
const managerWithRecovery = new SelfHealingManager(store, { rootDir: "/tmp/test-project" });
(store.getSettings as ReturnType<typeof vi.fn>).mockResolvedValue(baseSettings);
(store.listTasks as ReturnType<typeof vi.fn>)
.mockResolvedValueOnce([
{ id: "FN-stuck", column: "in-review", paused: false, status: "failed", mergeRetries: 3, mergeDetails: undefined, worktree: "/tmp/wt", branch: "fusion/fn-stuck", baseBranch: "main", prInfo: { number: 77 }, log: [] },
])
.mockResolvedValueOnce([])
.mockResolvedValueOnce([{ id: "FN-dep", column: "todo", blockedBy: "FN-stuck", log: [] }])
.mockResolvedValueOnce([]);
mockedExecSync.mockImplementation((command: string | Buffer) => {
const cmd = String(command);
if (cmd.includes("Fusion-Task-Id: FN-stuck")) return "abc12345\x1fRecovered subject\n" as any;
if (cmd.includes("--shortstat")) return " 2 files changed, 3 insertions(+), 1 deletions(-)\n" as any;
return "" as any;
});
const result = await managerWithRecovery.recoverStuckMergeDeadlocks();
expect(result).toBe(1);
expect(store.moveTask).toHaveBeenCalledWith("FN-stuck", "done");
expect(store.updateTask).toHaveBeenCalledWith("FN-stuck", expect.objectContaining({
status: null,
error: null,
mergeRetries: 0,
worktree: null,
branch: null,
mergeDetails: expect.objectContaining({ commitSha: "abc12345", mergeConfirmed: true }),
}));
expect(store.updateTask).toHaveBeenCalledWith("FN-dep", { blockedBy: null });
expect(
mockedExecSync.mock.calls.some((call) => String(call[0]).includes("git worktree remove '/tmp/wt' --force")),
).toBe(true);
expect(getSelfHealingLogger().log).toHaveBeenCalledWith(expect.stringContaining("self-heal:deadlock-recovered"));
managerWithRecovery.stop();
});
it("pauses genuine failures and leaves blockedBy untouched", async () => {
const managerWithRecovery = new SelfHealingManager(store, { rootDir: "/tmp/test-project" });
(store.getSettings as ReturnType<typeof vi.fn>).mockResolvedValue(baseSettings);
(store.listTasks as ReturnType<typeof vi.fn>)
.mockResolvedValueOnce([{ id: "FN-stuck", column: "in-review", paused: false, status: "failed", mergeRetries: 3, mergeDetails: undefined, worktree: "/tmp/wt", log: [] }])
.mockResolvedValueOnce([])
.mockResolvedValueOnce([{ id: "FN-dep", column: "todo", blockedBy: "FN-stuck", log: [] }])
.mockResolvedValueOnce([]);
mockedExecSync.mockReturnValue("" as any);
const result = await managerWithRecovery.recoverStuckMergeDeadlocks();
expect(result).toBe(1);
expect(store.updateTask).toHaveBeenCalledWith("FN-stuck", { paused: true });
expect(store.moveTask).not.toHaveBeenCalled();
expect(store.updateTask).not.toHaveBeenCalledWith("FN-dep", { blockedBy: null });
expect(getSelfHealingLogger().warn).toHaveBeenCalledWith(expect.stringContaining("paused-for-manual"));
managerWithRecovery.stop();
});
it("is idempotent and cooldown-gated", async () => {
const managerWithRecovery = new SelfHealingManager(store, { rootDir: "/tmp/test-project" });
(store.getSettings as ReturnType<typeof vi.fn>).mockResolvedValue(baseSettings);
(store.listTasks as ReturnType<typeof vi.fn>)
.mockResolvedValueOnce([{ id: "FN-stuck", column: "in-review", paused: false, status: "failed", mergeRetries: 3, mergeDetails: undefined, worktree: "/tmp/wt", log: [] }])
.mockResolvedValueOnce([])
.mockResolvedValueOnce([])
.mockResolvedValueOnce([])
.mockResolvedValueOnce([{ id: "FN-stuck", column: "done", paused: false, status: null, mergeRetries: 0, mergeDetails: { mergeConfirmed: true }, log: [] }])
.mockResolvedValueOnce([])
.mockResolvedValueOnce([])
.mockResolvedValueOnce([]);
mockedExecSync.mockImplementation((command: string | Buffer) => String(command).includes("Fusion-Task-Id: FN-stuck") ? ("abc12345\x1fRecovered subject\n" as any) : ("" as any));
const first = await managerWithRecovery.recoverStuckMergeDeadlocks();
const second = await managerWithRecovery.recoverStuckMergeDeadlocks();
expect(first).toBe(1);
expect(second).toBe(0);
managerWithRecovery.stop();
});
it("enforces cooldown for repeated genuine-failure sweeps", async () => {
const managerWithRecovery = new SelfHealingManager(store, { rootDir: "/tmp/test-project" });
(store.getSettings as ReturnType<typeof vi.fn>).mockResolvedValue(baseSettings);
(store.listTasks as ReturnType<typeof vi.fn>)
.mockResolvedValueOnce([{ id: "FN-cool", column: "in-review", paused: false, status: "failed", mergeRetries: 3, mergeDetails: undefined, worktree: "/tmp/wt", log: [] }])
.mockResolvedValueOnce([])
.mockResolvedValueOnce([])
.mockResolvedValueOnce([])
.mockResolvedValueOnce([{ id: "FN-cool", column: "in-review", paused: false, status: "failed", mergeRetries: 3, mergeDetails: undefined, worktree: "/tmp/wt", log: [] }])
.mockResolvedValueOnce([])
.mockResolvedValueOnce([])
.mockResolvedValueOnce([]);
mockedExecSync.mockReturnValue("" as any);
const first = await managerWithRecovery.recoverStuckMergeDeadlocks();
const updateCallsAfterFirst = (store.updateTask as ReturnType<typeof vi.fn>).mock.calls.length;
const second = await managerWithRecovery.recoverStuckMergeDeadlocks();
expect(first).toBe(1);
expect(second).toBe(0);
expect((store.updateTask as ReturnType<typeof vi.fn>).mock.calls.length).toBe(updateCallsAfterFirst);
managerWithRecovery.stop();
});
it("short-circuits when globalPause or enginePaused is active", async () => {
const managerWithRecovery = new SelfHealingManager(store, { rootDir: "/tmp/test-project" });
mockedExecSync.mockClear();
(store.getSettings as ReturnType<typeof vi.fn>).mockResolvedValue({ globalPause: true, enginePaused: false });
expect(await managerWithRecovery.recoverStuckMergeDeadlocks()).toBe(0);
(store.getSettings as ReturnType<typeof vi.fn>).mockResolvedValue({ globalPause: false, enginePaused: true });
expect(await managerWithRecovery.recoverStuckMergeDeadlocks()).toBe(0);
expect(store.listTasks).not.toHaveBeenCalled();
expect(mockedExecSync).not.toHaveBeenCalled();
managerWithRecovery.stop();
});
it("isolates per-task errors and continues with other stuck tasks", async () => {
const managerWithRecovery = new SelfHealingManager(store, { rootDir: "/tmp/test-project" });
(store.getSettings as ReturnType<typeof vi.fn>).mockResolvedValue(baseSettings);
(store.listTasks as ReturnType<typeof vi.fn>)
.mockResolvedValueOnce([
{ id: "FN-err", column: "in-review", paused: false, status: "failed", mergeRetries: 3, mergeDetails: undefined, worktree: "/tmp/wt1", log: [] },
{ id: "FN-ok", column: "in-review", paused: false, status: "failed", mergeRetries: 3, mergeDetails: undefined, worktree: "/tmp/wt2", log: [] },
])
.mockResolvedValueOnce([])
.mockResolvedValueOnce([])
.mockResolvedValueOnce([]);
mockedExecSync.mockImplementation((command: string | Buffer) => {
const cmd = String(command);
if (cmd.includes("Fusion-Task-Id: FN-ok")) return "def67890\x1fok\n" as any;
if (cmd.includes("Fusion-Task-Id: FN-err")) return "abcabc12\x1ferr\n" as any;
return "" as any;
});
(store.updateTask as ReturnType<typeof vi.fn>).mockImplementation(async (id: string) => {
if (id === "FN-err") throw new Error("update failed");
return {} as Task;
});
const result = await managerWithRecovery.recoverStuckMergeDeadlocks();
expect(result).toBe(1);
expect(getSelfHealingLogger().warn).toHaveBeenCalledWith(expect.stringContaining("self-heal:deadlock-recovery-error"));
expect((managerWithRecovery as any).deadlockRecoveryCooldown.get("FN-err")).toBeTypeOf("number");
managerWithRecovery.stop();
});
it("recovers worktree-only orphans and reproduces three-task incident", async () => {
const managerWithRecovery = new SelfHealingManager(store, { rootDir: "/tmp/test-project" });
(store.getSettings as ReturnType<typeof vi.fn>).mockResolvedValue(baseSettings);
(store.listTasks as ReturnType<typeof vi.fn>)
.mockResolvedValueOnce([
{ id: "FN-3794", column: "in-review", paused: false, status: "failed", mergeRetries: 3, mergeDetails: undefined, worktree: "/tmp/wt-a", log: [] },
{ id: "FN-3814", column: "in-review", paused: false, status: "failed", mergeRetries: 3, mergeDetails: undefined, worktree: "/tmp/wt-b", log: [] },
{ id: "FN-3829", column: "in-review", paused: false, status: "failed", mergeRetries: 3, mergeDetails: undefined, worktree: "/tmp/wt-c", log: [] },
])
.mockResolvedValueOnce([])
.mockResolvedValueOnce([{ id: "FN-3842", column: "todo", blockedBy: "FN-3794", log: [] }])
.mockResolvedValueOnce([]);
mockedExecSync.mockImplementation((command: string | Buffer) => {
const cmd = String(command);
if (cmd.includes("Fusion-Task-Id: FN-3794")) return "278a2825\x1fone\n" as any;
if (cmd.includes("Fusion-Task-Id: FN-3814")) return "69c25e2b\x1ftwo\n" as any;
if (cmd.includes("Fusion-Task-Id: FN-3829")) return "0d3f51b6\x1fthree\n" as any;
return "" as any;
});
const result = await managerWithRecovery.recoverStuckMergeDeadlocks();
expect(result).toBe(3);
expect(store.updateTask).toHaveBeenCalledWith("FN-3842", { blockedBy: null });
managerWithRecovery.stop();
});
});
describe("recoverAlreadyMergedReviewTasks", () => {
it("short-circuits when globalPause or enginePaused is active", async () => {
const managerWithRecovery = new SelfHealingManager(store, { rootDir: "/tmp/test-project" });
@@ -3816,6 +4000,7 @@ describe("maintenance cycle concurrency", () => {
(vi.spyOn(manager as any, "recoverStaleMergingStatus").mockResolvedValue(0) as any);
(vi.spyOn(manager as any, "recoverMergeableReviewTasks").mockResolvedValue(0) as any);
(vi.spyOn(manager as any, "recoverMergedReviewTasks").mockResolvedValue(0) as any);
(vi.spyOn(manager as any, "recoverStuckMergeDeadlocks").mockResolvedValue(0) as any);
(vi.spyOn(manager as any, "recoverMisclassifiedFailures").mockResolvedValue(0) as any);
(vi.spyOn(manager as any, "recoverNoProgressNoTaskDoneFailures").mockResolvedValue(0) as any);
(vi.spyOn(manager as any, "recoverPartialProgressNoTaskDoneFailures").mockResolvedValue(0) as any);
@@ -3841,6 +4026,7 @@ describe("maintenance cycle concurrency", () => {
(vi.spyOn(manager as any, "recoverStaleMergingStatus").mockResolvedValue(0) as any);
(vi.spyOn(manager as any, "recoverMergeableReviewTasks").mockResolvedValue(0) as any);
(vi.spyOn(manager as any, "recoverMergedReviewTasks").mockResolvedValue(0) as any);
(vi.spyOn(manager as any, "recoverStuckMergeDeadlocks").mockResolvedValue(0) as any);
(vi.spyOn(manager as any, "recoverMisclassifiedFailures").mockResolvedValue(0) as any);
(vi.spyOn(manager as any, "recoverNoProgressNoTaskDoneFailures").mockResolvedValue(0) as any);
(vi.spyOn(manager as any, "recoverPartialProgressNoTaskDoneFailures").mockResolvedValue(0) as any);
@@ -3910,6 +4096,7 @@ describe("maintenance cycle concurrency", () => {
makeSlow("recoverStaleMergingStatus");
makeSlow("recoverMergeableReviewTasks");
makeSlow("recoverMergedReviewTasks");
makeSlow("recoverStuckMergeDeadlocks");
makeSlow("recoverMisclassifiedFailures");
makeSlow("recoverNoProgressNoTaskDoneFailures");
makeSlow("recoverPartialProgressNoTaskDoneFailures");
@@ -3937,6 +4124,7 @@ describe("maintenance cycle concurrency", () => {
"recoverStaleMergingStatus",
"recoverMergeableReviewTasks",
"recoverMergedReviewTasks",
"recoverStuckMergeDeadlocks",
"recoverMisclassifiedFailures",
"recoverNoProgressNoTaskDoneFailures",
"recoverPartialProgressNoTaskDoneFailures",

116
packages/engine/src/self-healing.ts
View File

@@ -121,6 +121,7 @@ const ORPHANED_WITH_WORKTREE_GRACE_MS = 300_000;
*/
const MAX_TASK_DONE_RETRIES = 3;
const MAX_AUTO_MERGE_RETRIES = 3;
const DEADLOCK_RECOVERY_COOLDOWN_MS = 15 * 60_000;
const DEFAULT_STALE_MERGING_STATUS_MIN_AGE_MS = 5 * 60_000;
interface LandedTaskCommit {
@@ -181,6 +182,9 @@ export class SelfHealingManager {
// ── Event listener cleanup ──────────────────────────────────────────
private settingsListener: ((data: { settings: Settings; previous: Settings }) => void) | null = null;
// ── Per-task deadlock recovery cooldown ─────────────────────────────
private deadlockRecoveryCooldown: Map<string, number> = new Map();
constructor(
private store: TaskStore,
private options: SelfHealingOptions,
@@ -227,6 +231,8 @@ export class SelfHealingManager {
{ name: "failed-pre-merge-steps", fn: () => this.recoverReviewTasksWithFailedPreMergeSteps().then(() => undefined) },
{ name: "interrupted-merging", fn: () => this.recoverInterruptedMergingTasks().then(() => undefined) },
{ name: "done-merge-metadata", fn: () => this.recoverDoneTaskMergeMetadata().then(() => undefined) },
{ name: "recover-already-merged-review", fn: () => this.recoverAlreadyMergedReviewTasks().then(() => undefined) },
{ name: "recover-stuck-merge-deadlocks", fn: () => this.recoverStuckMergeDeadlocks().then(() => undefined) },
{ name: "misclassified-failures", fn: () => this.recoverMisclassifiedFailures().then(() => undefined) },
{ name: "partial-progress-no-task-done", fn: () => this.recoverPartialProgressNoTaskDoneFailures().then(() => undefined) },
{ name: "orphaned-executions", fn: () => this.recoverOrphanedExecutions().then(() => undefined) },
@@ -833,6 +839,7 @@ export class SelfHealingManager {
{ name: "recover-mergeable-review", fn: () => this.recoverMergeableReviewTasks() },
{ name: "recover-merged-review", fn: () => this.recoverMergedReviewTasks() },
{ name: "recover-already-merged-review", fn: () => this.recoverAlreadyMergedReviewTasks() },
{ name: "recover-stuck-merge-deadlocks", fn: () => this.recoverStuckMergeDeadlocks() },
{ name: "recover-misclassified-failures", fn: () => this.recoverMisclassifiedFailures() },
{ name: "recover-no-progress-no-task-done", fn: () => this.recoverNoProgressNoTaskDoneFailures() },
{ name: "recover-partial-progress-no-task-done", fn: () => this.recoverPartialProgressNoTaskDoneFailures() },
@@ -1677,6 +1684,115 @@ export class SelfHealingManager {
}
}
/**
* Recover deadlocked retry-exhausted merge failures that are still blocking
* dispatch via `blockedBy` or retained worktree ownership.
*/
async recoverStuckMergeDeadlocks(): Promise<number> {
try {
const settings = await this.store.getSettings();
if (settings.globalPause || settings.enginePaused) return 0;
const now = Date.now();
const inReview = await this.store.listTasks({ column: "in-review", slim: true });
const triage = await this.store.listTasks({ column: "triage", slim: true });
const todo = await this.store.listTasks({ column: "todo", slim: true });
const inProgress = await this.store.listTasks({ column: "in-progress", slim: true });
const dependentsByBlocker = new Map<string, Task[]>();
for (const task of [...triage, ...todo, ...inProgress]) {
if (!task.blockedBy) continue;
const dependents = dependentsByBlocker.get(task.blockedBy) ?? [];
dependents.push(task);
dependentsByBlocker.set(task.blockedBy, dependents);
}
const candidates = inReview.filter((task) => {
const cooldownStart = this.deadlockRecoveryCooldown.get(task.id) ?? 0;
const cooldownElapsed = now - cooldownStart;
const hasBlockedDependents = (dependentsByBlocker.get(task.id) ?? []).some(
(dep) => dep.column === "triage" || dep.column === "todo",
);
return task.column === "in-review" &&
!task.paused &&
task.status === "failed" &&
(task.mergeRetries ?? 0) >= MAX_AUTO_MERGE_RETRIES &&
task.mergeDetails?.mergeConfirmed !== true &&
(hasBlockedDependents || Boolean(task.worktree)) &&
cooldownElapsed >= DEADLOCK_RECOVERY_COOLDOWN_MS;
});
if (candidates.length === 0) return 0;
let recovered = 0;
for (const task of candidates) {
const blockedDependents = dependentsByBlocker.get(task.id) ?? [];
const blockedTaskIds = blockedDependents.map((dep) => dep.id);
try {
const landedCommit = await this.findLandedTaskCommit(task);
if (landedCommit) {
const mergeDetails: MergeDetails = {
commitSha: landedCommit.sha,
filesChanged: landedCommit.filesChanged,
insertions: landedCommit.insertions,
deletions: landedCommit.deletions,
mergeCommitMessage: landedCommit.subject,
mergedAt: new Date().toISOString(),
mergeConfirmed: true,
prNumber: task.prInfo?.number,
};
await this.store.updateTask(task.id, {
status: null,
error: null,
mergeRetries: 0,
worktree: null,
branch: null,
mergeDetails,
});
await this.store.moveTask(task.id, "done");
await this.cleanupInterruptedMergeArtifacts(task);
const clearedDependents: string[] = [];
for (const dep of blockedDependents) {
try {
await this.store.updateTask(dep.id, { blockedBy: null });
await this.store.logEntry(dep.id, `Auto-recovered: cleared stale blockedBy ${task.id} after deadlock recovery`);
clearedDependents.push(dep.id);
} catch (depErr: unknown) {
const depErrMessage = depErr instanceof Error ? depErr.message : String(depErr);
log.warn(`self-heal:deadlock-recovery-dependent-error ${JSON.stringify({ blockerTaskId: task.id, dependentTaskId: dep.id, error: depErrMessage })}`);
}
}
await this.store.logEntry(
task.id,
`Auto-recovered: merge deadlock resolved via landed commit ${landedCommit.sha.slice(0, 8)}${clearedDependents.length > 0 ? `; cleared blockedBy on ${clearedDependents.join(", ")}` : ""}`,
);
log.log(`self-heal:deadlock-recovered ${JSON.stringify({ stuckTaskId: task.id, blockedTaskIds, attributedSha: landedCommit.sha, action: "reattributed" })}`);
recovered++;
} else {
await this.store.updateTask(task.id, { paused: true });
await this.store.logEntry(task.id, "merge-deadlock-detected: requires manual intervention — verified content not on main");
log.warn(`self-heal:deadlock-recovered ${JSON.stringify({ stuckTaskId: task.id, blockedTaskIds, attributedSha: null, action: "paused-for-manual" })}`);
recovered++;
}
} catch (err: unknown) {
const errorMessage = err instanceof Error ? err.message : String(err);
log.warn(`self-heal:deadlock-recovery-error ${JSON.stringify({ stuckTaskId: task.id, blockedTaskIds, error: errorMessage })}`);
} finally {
this.deadlockRecoveryCooldown.set(task.id, Date.now());
}
}
return recovered;
} catch (err: unknown) {
const errorMessage = err instanceof Error ? err.message : String(err);
log.error(`Stuck merge deadlock recovery failed: ${errorMessage}`);
return 0;
}
}
/**
* Recover retry-exhausted failed review tasks whose content already landed on
* the integration branch via a non-canonical merge lineage.