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

fix(FN-1673): recover interrupted merge finalization

Browse Source 
Add self-healing for stale in-review tasks left in merging status after the merge commit lands. Recovery uses taskStuckTimeoutMs, records merge details, moves confirmed merges to done, and clears stale merge status for retry when no landed commit is found.
This commit is contained in:
gsxdsm
2026-04-15 15:22:15 -07:00
parent 979ba2c04a
commit 561ca32040
2 changed files with 354 additions and 1 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -852,6 +852,168 @@ describe("SelfHealingManager", () => {
});
describe("recoverMergedReviewTasks", () => {
it("finalizes stale merging tasks when a task commit already landed", async () => {
const managerWithRecovery = new SelfHealingManager(store, {
rootDir: "/tmp/test-project",
});
(store.getSettings as ReturnType<typeof vi.fn>).mockResolvedValue({
taskStuckTimeoutMs: 60_000,
});
const staleUpdatedAt = new Date(Date.now() - 61_000).toISOString();
(store.listTasks as ReturnType<typeof vi.fn>).mockResolvedValue([
{
id: "FN-1673",
column: "in-review",
status: "merging",
error: null,
paused: false,
worktree: "/tmp/test-project/.worktrees/fn-1673",
branch: "fusion/fn-1673",
baseCommitSha: "base123",
updatedAt: staleUpdatedAt,
steps: [{ name: "Ship it", status: "done" }],
workflowStepResults: [],
mergeDetails: undefined,
log: [],
},
]);
mockedExistsSync.mockReturnValue(true);
mockedExecSync.mockImplementation((command) => {
const cmd = String(command);
if (cmd.includes("git log")) {
return "979ba2c04\u001ffeat(FN-1673): add editable AI suggestion drafts before acceptance\n" as any;
}
if (cmd.includes("git show --shortstat")) {
return " 1 file changed, 2 insertions(+), 2 deletions(-)\n" as any;
}
return "" as any;
});
const result = await managerWithRecovery.recoverInterruptedMergingTasks();
expect(result).toBe(1);
expect(store.updateTask).toHaveBeenCalledWith("FN-1673", {
status: null,
error: null,
mergeRetries: 0,
mergeDetails: expect.objectContaining({
commitSha: "979ba2c04",
mergeCommitMessage: "feat(FN-1673): add editable AI suggestion drafts before acceptance",
mergeConfirmed: true,
filesChanged: 1,
insertions: 2,
deletions: 2,
}),
});
expect(store.moveTask).toHaveBeenCalledWith("FN-1673", "done");
expect(store.logEntry).toHaveBeenCalledWith(
"FN-1673",
expect.stringContaining("stale merge status finalized from landed commit 979ba2c"),
);
managerWithRecovery.stop();
});
it("clears stale merging status for retry when no landed commit is found", async () => {
const managerWithRecovery = new SelfHealingManager(store, {
rootDir: "/tmp/test-project",
});
(store.getSettings as ReturnType<typeof vi.fn>).mockResolvedValue({
taskStuckTimeoutMs: 60_000,
});
const staleUpdatedAt = new Date(Date.now() - 61_000).toISOString();
(store.listTasks as ReturnType<typeof vi.fn>).mockResolvedValue([
{
id: "FN-1674",
column: "in-review",
status: "merging",
error: null,
updatedAt: staleUpdatedAt,
steps: [{ name: "Ship it", status: "done" }],
workflowStepResults: [],
log: [],
},
]);
mockedExecSync.mockImplementation((command) => {
if (String(command).includes("git log")) return "" as any;
return "" as any;
});
const result = await managerWithRecovery.recoverInterruptedMergingTasks();
expect(result).toBe(1);
expect(store.updateTask).toHaveBeenCalledWith("FN-1674", {
status: null,
error: null,
});
expect(store.moveTask).not.toHaveBeenCalledWith("FN-1674", "done");
expect(store.logEntry).toHaveBeenCalledWith(
"FN-1674",
expect.stringContaining("stale merge status cleared"),
);
managerWithRecovery.stop();
});
it("does not recover fresh merging tasks before the stuck timeout", async () => {
const managerWithRecovery = new SelfHealingManager(store, {
rootDir: "/tmp/test-project",
});
(store.getSettings as ReturnType<typeof vi.fn>).mockResolvedValue({
taskStuckTimeoutMs: 60_000,
});
(store.listTasks as ReturnType<typeof vi.fn>).mockResolvedValue([
{
id: "FN-1675",
column: "in-review",
status: "merging",
error: null,
updatedAt: new Date().toISOString(),
log: [],
},
]);
const result = await managerWithRecovery.recoverInterruptedMergingTasks();
expect(result).toBe(0);
expect(store.updateTask).not.toHaveBeenCalled();
expect(store.moveTask).not.toHaveBeenCalled();
managerWithRecovery.stop();
});
it("does not recover stale merging tasks when stuck detection is disabled", async () => {
const managerWithRecovery = new SelfHealingManager(store, {
rootDir: "/tmp/test-project",
});
(store.getSettings as ReturnType<typeof vi.fn>).mockResolvedValue({
taskStuckTimeoutMs: 0,
});
(store.listTasks as ReturnType<typeof vi.fn>).mockResolvedValue([
{
id: "FN-1676",
column: "in-review",
status: "merging",
error: null,
updatedAt: new Date(Date.now() - 24 * 60 * 60_000).toISOString(),
log: [],
},
]);
const result = await managerWithRecovery.recoverInterruptedMergingTasks();
expect(result).toBe(0);
expect(store.listTasks).not.toHaveBeenCalled();
expect(store.updateTask).not.toHaveBeenCalled();
expect(store.moveTask).not.toHaveBeenCalled();
managerWithRecovery.stop();
});
it("merges eligible in-review tasks that still have an unmerged worktree", async () => {
const managerWithRecovery = new SelfHealingManager(store, {
rootDir: "/tmp/test-project",

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

@@ -18,7 +18,7 @@ import { exec } from "node:child_process";
import { promisify } from "node:util";
import { existsSync, readdirSync, statSync } from "node:fs";
import { join } from "node:path";
import { getTaskMergeBlocker, type TaskStore, type Settings, type Task } from "@fusion/core";
import { getTaskMergeBlocker, type TaskStore, type Settings, type Task, type MergeDetails } from "@fusion/core";
import { createLogger } from "./logger.js";
import { scanIdleWorktrees, scanOrphanedBranches } from "./worktree-pool.js";
@@ -63,6 +63,7 @@ export interface SelfHealingOptions {
const APPROVED_TRIAGE_RECOVERY_GRACE_MS = 60_000;
const ORPHANED_EXECUTION_RECOVERY_GRACE_MS = 60_000;
const ACTIVE_MERGE_STATUSES = new Set(["merging", "merging-pr"]);
/**
* Longer grace period for tasks that still have a worktree on disk.
* This avoids racing with `executor.resumeOrphaned()` which runs on
@@ -71,6 +72,31 @@ const ORPHANED_EXECUTION_RECOVERY_GRACE_MS = 60_000;
*/
const ORPHANED_WITH_WORKTREE_GRACE_MS = 300_000;
interface LandedTaskCommit {
sha: string;
subject?: string;
filesChanged?: number;
insertions?: number;
deletions?: number;
}
function shellQuote(value: string): string {
return `'${value.replace(/'/g, "'\\''")}'`;
}
function parseShortstat(output: string): Pick<LandedTaskCommit, "filesChanged" | "insertions" | "deletions"> {
const normalized = output.trim().replace(/\n/g, " ");
const filesMatch = normalized.match(/(\d+) files? changed/);
const insertionsMatch = normalized.match(/(\d+) insertions?\(\+\)/);
const deletionsMatch = normalized.match(/(\d+) deletions?\(-\)/);
return {
filesChanged: filesMatch ? Number.parseInt(filesMatch[1], 10) : 0,
insertions: insertionsMatch ? Number.parseInt(insertionsMatch[1], 10) : 0,
deletions: deletionsMatch ? Number.parseInt(deletionsMatch[1], 10) : 0,
};
}
export class SelfHealingManager {
// ── Auto-unpause state ──────────────────────────────────────────────
private unpauseTimer: ReturnType<typeof setTimeout> | null = null;
@@ -114,6 +140,7 @@ export class SelfHealingManager {
async runStartupRecovery(): Promise<void> {
await this.recoverNoProgressNoTaskDoneFailures();
await this.recoverCompletedTasks();
await this.recoverInterruptedMergingTasks();
await this.recoverMisclassifiedFailures();
await this.recoverOrphanedExecutions();
await this.recoverApprovedTriageTasks();
@@ -342,6 +369,82 @@ export class SelfHealingManager {
}, intervalMs);
}
private isPastInterruptedMergeGrace(task: Task, timeoutMs: number): boolean {
const updatedAt = task.updatedAt ? Date.parse(task.updatedAt) : 0;
if (!Number.isFinite(updatedAt) || updatedAt <= 0) return false;
return Date.now() - updatedAt >= timeoutMs;
}
private async findLandedTaskCommit(task: Task): Promise<LandedTaskCommit | null> {
const readLog = async (range: string) => {
const command = [
"git log",
"--format=%H%x1f%s",
"--max-count=20",
"--fixed-strings",
`--grep=${shellQuote(task.id)}`,
shellQuote(range),
].join(" ");
return execAsync(command, {
cwd: this.options.rootDir,
maxBuffer: 1024 * 1024,
});
};
let stdout: string;
try {
const result = await readLog(task.baseCommitSha ? `${task.baseCommitSha}..HEAD` : "HEAD");
stdout = result.stdout;
} catch {
if (!task.baseCommitSha) return null;
const result = await readLog("HEAD");
stdout = result.stdout;
}
const firstLine = stdout.trim().split("\n").find(Boolean);
if (!firstLine) return null;
const [sha, subject] = firstLine.split("\x1f");
if (!sha) return null;
const commit: LandedTaskCommit = { sha, subject };
try {
const stats = await execAsync(`git show --shortstat --format= ${shellQuote(sha)}`, {
cwd: this.options.rootDir,
maxBuffer: 1024 * 1024,
});
Object.assign(commit, parseShortstat(stats.stdout));
} catch {
// Stats are useful for the task detail view but not required for recovery.
}
return commit;
}
private async cleanupInterruptedMergeArtifacts(task: Task): Promise<void> {
if (task.worktree && existsSync(task.worktree)) {
try {
await execAsync(`git worktree remove ${shellQuote(task.worktree)} --force`, {
cwd: this.options.rootDir,
timeout: 120_000,
});
} catch {
// Non-fatal; existing orphan/worktree cleanup can retry later.
}
}
const branch = task.branch || `fusion/${task.id.toLowerCase()}`;
try {
await execAsync(`git branch -D ${shellQuote(branch)}`, {
cwd: this.options.rootDir,
timeout: 120_000,
});
} catch {
// Non-fatal; branch may be gone or still checked out.
}
}
private async runMaintenance(): Promise<void> {
const startMs = Date.now();
log.log("Maintenance cycle starting");
@@ -353,6 +456,7 @@ export class SelfHealingManager {
this.checkpointWal();
await this.enforceWorktreeCap();
await this.recoverCompletedTasks();
await this.recoverInterruptedMergingTasks();
await this.recoverMergeableReviewTasks();
await this.recoverMergedReviewTasks();
await this.recoverMisclassifiedFailures();
@@ -519,6 +623,93 @@ export class SelfHealingManager {
}
}
/**
* Recover stale `in-review` tasks left in a transient merge status.
*
* The direct AI merger can successfully create the final commit and then be
* interrupted before it stores mergeDetails and moves the task to `done`.
* When that happens no future task:moved event fires, so the merge queue has
* nothing to retry. This recovery confirms the task-specific commit exists on
* the current main lineage before finalizing the task.
*
* If no landed commit is found, it only clears the stale transient status so
* the normal mergeable-review recovery can retry the merge.
*
* @returns Number of tasks finalized or unblocked
*/
async recoverInterruptedMergingTasks(): Promise<number> {
try {
const settings = await this.store.getSettings();
const timeoutMs = settings.taskStuckTimeoutMs;
if (!timeoutMs || timeoutMs <= 0) return 0;
const tasks = await this.store.listTasks({ column: "in-review" });
const candidates = tasks.filter((task) =>
task.column === "in-review" &&
Boolean(task.status && ACTIVE_MERGE_STATUSES.has(task.status)) &&
this.isPastInterruptedMergeGrace(task, timeoutMs),
);
if (candidates.length === 0) return 0;
log.warn(`Found ${candidates.length} stale merging task(s) in in-review`);
let recovered = 0;
for (const task of candidates) {
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,
mergeDetails,
});
await this.store.moveTask(task.id, "done");
await this.cleanupInterruptedMergeArtifacts(task);
await this.store.logEntry(
task.id,
`Auto-recovered: stale merge status finalized from landed commit ${landedCommit.sha.slice(0, 8)}`,
);
log.log(`Recovered interrupted merge ${task.id}: finalized landed commit ${landedCommit.sha.slice(0, 8)}`);
recovered++;
continue;
}
await this.store.updateTask(task.id, { status: null, error: null });
await this.store.logEntry(
task.id,
"Auto-recovered: stale merge status cleared; merge will be retried",
);
log.log(`Recovered interrupted merge ${task.id}: cleared stale status for retry`);
recovered++;
} catch (err: unknown) { const errorMessage = err instanceof Error ? err.message : String(err);
log.error(`Failed to recover interrupted merge ${task.id}: ${errorMessage}`);
}
}
if (recovered > 0) {
log.log(`Recovered ${recovered} interrupted merge task(s)`);
}
return recovered;
} catch (err: unknown) { const errorMessage = err instanceof Error ? err.message : String(err);
log.error(`Interrupted merge recovery failed: ${errorMessage}`);
return 0;
}
}
// ── Misclassified failure recovery ───────────────────────────────
/**