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

feat(FN-5627): self-heal transient merge failures stuck at mergeRetries=3

Browse Source 
After the FN-5627 merger fix (b2d547eae, 230f6f45b) landed, two in-review
tasks (FN-5628, FN-5632) remained stuck at mergeRetries=3 with
status=failed because the merger correctly identified transient failure
classes but had no auto-recovery path \u2014 the AUTO_MERGE_COOLDOWN_MS reset
takes hours and gives up too easily.

Failure classes covered:
- lease-handoff-failed: target-not-queued (FN-5353/FN-5363 race where the
  merge queue lease was cleared between enqueue and handoff acquisition).
- Legacy same-SHA spurious 'Integration branch X advanced concurrently
  (expected SHA, observed SHA)' errors from pre-FN-5627 code paths.

Implementation:
- New MergeDetails.transientRecoveryCount field tracks per-task recovery
  attempts, bounded by MAX_TRANSIENT_MERGE_RECOVERIES = 2.
- New classifyTransientMergeError() string matcher in self-healing.ts
  identifies recoverable classes by error pattern. Returns null for
  genuine merge failures (verification, conflicts, real concurrent
  advances with different SHAs).
- SelfHealingManager.recoverTransientMergeFailures() sweep finds
  matching in-review tasks, resets mergeRetries=0, clears status/error,
  increments recovery count, re-enqueues via requeueForAutoMerge.
- Wired into BOTH startup recovery and periodic Batch 2 maintenance loop.
- Emits merger:transient-failure-auto-recovered (recovered) and
  merger:transient-failure-budget-exhausted (terminal) audit events.

No-op when autoMerge=false, requeueForAutoMerge not wired, or pause
active. Repeat-suppression on budget-exhausted emit via error marker
[transient-recovery-budget-exhausted] to prevent log spam.

Tests (6 new):
- target-not-queued recovery path
- spurious-concurrent-advance-same-sha recovery path (legacy)
- genuine concurrent-advance (different SHAs) NOT recovered
- non-transient failures NOT recovered (verification, conflicts)
- budget exhaustion emits marker once, no further requeue
- autoMerge=false no-op

Engine suite: 6157 tests pass (6 new).

In-flight: FN-5628 and FN-5632 were manually reset via SQL so the
already-running engine (which has the FN-5627 merger fix) can re-attempt
their merges before this self-healing path lands and reloads. Future
occurrences self-recover.

Fusion-Task-Id: FN-5627
This commit is contained in:
gsxdsm
2026-05-28 13:53:15 -07:00
parent 5b5da2c24f
commit 5768d5ec45
5 changed files with 461 additions and 1 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -1757,6 +1757,206 @@ describe("SelfHealingManager", () => {
});
});
describe("FN-5627: recoverTransientMergeFailures", () => {
function setupTransientRecoveryStore(opts: {
tasks: Array<Record<string, unknown>>;
settings?: Record<string, unknown>;
}): TaskStore & EventEmitter {
const taskMap = new Map(opts.tasks.map((t) => [t.id as string, t]));
return createMockStore({
getSettings: vi.fn().mockResolvedValue({
autoMerge: true,
globalPause: false,
enginePaused: false,
...(opts.settings ?? {}),
} as unknown as Settings),
listTasks: vi.fn().mockResolvedValue(opts.tasks),
getTask: vi.fn((id: string) => Promise.resolve(taskMap.get(id) as Task | undefined)),
updateTask: vi.fn(async (id: string, updates: Partial<Task>) => {
const existing = taskMap.get(id) ?? {};
const merged = { ...existing, ...updates };
if (updates.mergeDetails !== undefined) {
merged.mergeDetails = updates.mergeDetails;
}
taskMap.set(id, merged);
return merged as Task;
}),
});
}
it("resets mergeRetries and re-enqueues lease-handoff-target-not-queued failures", async () => {
const transientStore = setupTransientRecoveryStore({
tasks: [
{
id: "FN-5628",
column: "in-review",
paused: false,
status: "failed",
mergeRetries: 3,
error: "Merge handoff refused (lease-handoff-failed): target-not-queued",
mergeDetails: undefined,
},
],
});
const requeueForAutoMerge = vi.fn();
const mgr = new SelfHealingManager(transientStore, {
rootDir: "/tmp/test-project",
requeueForAutoMerge,
});
const recovered = await mgr.recoverTransientMergeFailures();
expect(recovered).toBe(1);
expect(requeueForAutoMerge).toHaveBeenCalledWith("FN-5628");
const updateCalls = (transientStore.updateTask as ReturnType<typeof vi.fn>).mock.calls as unknown as Array<[string, Partial<Task>]>;
const recoveryCall = updateCalls.find((call) => call[0] === "FN-5628" && call[1].status === null);
expect(recoveryCall).toBeDefined();
expect(recoveryCall![1].mergeRetries).toBe(0);
expect(recoveryCall![1].error).toBeNull();
expect((recoveryCall![1] as { mergeDetails?: { transientRecoveryCount?: number } }).mergeDetails?.transientRecoveryCount).toBe(1);
mgr.stop();
});
it("recovers same-SHA spurious concurrent-advance failures (pre-FN-5627 legacy)", async () => {
const transientStore = setupTransientRecoveryStore({
tasks: [
{
id: "FN-5632",
column: "in-review",
paused: false,
status: "failed",
mergeRetries: 3,
error: "Integration branch main advanced concurrently (expected 5b5da2c24fa006b46139ce4566b764126c6b84ca, observed 5b5da2c24fa006b46139ce4566b764126c6b84ca) while applying 283b290aec527f9ba4244f2935700a2823dd106b for FN-5632",
mergeDetails: undefined,
},
],
});
const requeueForAutoMerge = vi.fn();
const mgr = new SelfHealingManager(transientStore, {
rootDir: "/tmp/test-project",
requeueForAutoMerge,
});
const recovered = await mgr.recoverTransientMergeFailures();
expect(recovered).toBe(1);
expect(requeueForAutoMerge).toHaveBeenCalledWith("FN-5632");
mgr.stop();
});
it("does NOT recover genuine concurrent-advance failures (different SHAs)", async () => {
const transientStore = setupTransientRecoveryStore({
tasks: [
{
id: "FN-genuine",
column: "in-review",
paused: false,
status: "failed",
mergeRetries: 3,
// Different SHAs — a real concurrent advance happened. Don't auto-recover.
error: "Integration branch main advanced concurrently (expected aaa1111aaa1111aaa1111aaa1111aaa1111aaaa, observed bbb2222bbb2222bbb2222bbb2222bbb2222bbbb) while applying ccc3333ccc3333ccc3333ccc3333ccc3333cccc for FN-genuine",
},
],
});
const requeueForAutoMerge = vi.fn();
const mgr = new SelfHealingManager(transientStore, {
rootDir: "/tmp/test-project",
requeueForAutoMerge,
});
const recovered = await mgr.recoverTransientMergeFailures();
expect(recovered).toBe(0);
expect(requeueForAutoMerge).not.toHaveBeenCalled();
mgr.stop();
});
it("does NOT recover non-transient merge failures (verification, conflict, etc.)", async () => {
const transientStore = setupTransientRecoveryStore({
tasks: [
{
id: "FN-verify",
column: "in-review",
paused: false,
status: "failed",
mergeRetries: 3,
error: "Verification failed: pnpm test exit 1",
},
],
});
const requeueForAutoMerge = vi.fn();
const mgr = new SelfHealingManager(transientStore, {
rootDir: "/tmp/test-project",
requeueForAutoMerge,
});
const recovered = await mgr.recoverTransientMergeFailures();
expect(recovered).toBe(0);
expect(requeueForAutoMerge).not.toHaveBeenCalled();
mgr.stop();
});
it("parks task as failed once budget is exhausted (transientRecoveryCount >= 2)", async () => {
const transientStore = setupTransientRecoveryStore({
tasks: [
{
id: "FN-exhausted",
column: "in-review",
paused: false,
status: "failed",
mergeRetries: 3,
error: "Merge handoff refused (lease-handoff-failed): target-not-queued",
mergeDetails: { transientRecoveryCount: 2 },
},
],
});
const requeueForAutoMerge = vi.fn();
const mgr = new SelfHealingManager(transientStore, {
rootDir: "/tmp/test-project",
requeueForAutoMerge,
});
const recovered = await mgr.recoverTransientMergeFailures();
expect(recovered).toBe(0);
expect(requeueForAutoMerge).not.toHaveBeenCalled();
// updateTask called to add budget-exhausted marker to error
const updateCalls = (transientStore.updateTask as ReturnType<typeof vi.fn>).mock.calls as unknown as Array<[string, Partial<Task>]>;
const markerCall = updateCalls.find((call) => call[0] === "FN-exhausted" && typeof call[1].error === "string" && (call[1].error as string).includes("[transient-recovery-budget-exhausted]"));
expect(markerCall).toBeDefined();
mgr.stop();
});
it("is a no-op when autoMerge is disabled", async () => {
const transientStore = setupTransientRecoveryStore({
tasks: [
{
id: "FN-no-automerge",
column: "in-review",
paused: false,
status: "failed",
mergeRetries: 3,
error: "Merge handoff refused (lease-handoff-failed): target-not-queued",
},
],
settings: { autoMerge: false },
});
const requeueForAutoMerge = vi.fn();
const mgr = new SelfHealingManager(transientStore, {
rootDir: "/tmp/test-project",
requeueForAutoMerge,
});
const recovered = await mgr.recoverTransientMergeFailures();
expect(recovered).toBe(0);
expect(requeueForAutoMerge).not.toHaveBeenCalled();
mgr.stop();
});
});
describe("recoverStrandedCompletedTodoTasks", () => {
it("promotes completed todo tasks and calls recover fn once per qualifying task", async () => {
const recoverFn = vi.fn().mockResolvedValue(true);

17
packages/engine/src/run-audit.ts
View File

@@ -565,7 +565,22 @@ export type DatabaseMutationType =
* `merger:fast-path-blocked-foreign-commit` and parks as failed.
* Metadata: { taskId, commitSha, integrationBranch, reason, diagnostic, mergeRetries, maxRetries }
*/
| "merger:fast-path-auto-recovered";
| "merger:fast-path-auto-recovered"
/**
* FN-5627 follow-up: self-healing recovered an `in-review` task that was
* stuck at `mergeRetries >= MAX_AUTO_MERGE_RETRIES` with `status='failed'`
* due to a TRANSIENT merge failure class (`target-not-queued` lease
* handoff race, or spurious same-SHA concurrent-advance left over from
* pre-FN-5627 code). The sweep reset `mergeRetries` to 0, cleared
* `status`/`error`, incremented `mergeDetails.transientRecoveryCount`,
* and re-enqueued the task via `requeueForAutoMerge`. Bounded by
* `MAX_TRANSIENT_MERGE_RECOVERIES` (2). Once exhausted, the task stays
* parked as `failed` for manual review.
* Metadata: { taskId, transientClass, mergeRetries, recoveryCount, errorSnippet }
*/
| "merger:transient-failure-auto-recovered"
/** Metadata: { taskId, transientClass, recoveryCount, maxRecoveries, errorSnippet } */
| "merger:transient-failure-budget-exhausted";
// ── Filesystem mutation types ─────────────────────────────────────────────────

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

@@ -311,6 +311,44 @@ const ORPHANED_WITH_WORKTREE_GRACE_MS = 300_000;
const MAX_TASK_DONE_RETRIES = 3;
export const MAX_WORKTREE_SESSION_RETRIES = 3;
export const MAX_AUTO_MERGE_RETRIES = 3;
/**
* FN-5627 follow-up: bounded budget for self-healing transient-merge-failure
* recovery. After this many cycles of `recoverTransientMergeFailures` resetting
* `mergeRetries` and re-enqueueing the same task, the task is considered
* genuinely stuck and stays parked as `failed` for manual review. Tracked via
* `task.mergeDetails.transientRecoveryCount`.
*/
export const MAX_TRANSIENT_MERGE_RECOVERIES = 2;
/**
* FN-5627 follow-up: classify an in-review failed-task error message as a
* recoverable transient merge failure. Returns a stable class label when the
* error matches a known transient pattern; null otherwise.
*
* Recognized classes:
* - `lease-handoff-target-not-queued`: the merge queue lease acquisition saw
* the task drop out of the queue between enqueue and handoff. Race with
* self-healing sweeps that clean stale `mergeQueue` rows (FN-5353/FN-5363).
* - `spurious-concurrent-advance-same-sha`: the merger reported
* `Integration branch X advanced concurrently (expected SHA, observed SHA)`
* with identical SHA on both sides — the integration ref didn't actually
* move. Pre-FN-5627 misclassification in `merger-ref-update-advance.ts`
* routed real ref-update-refusal failures (lock contention, hook rejection)
* through `IntegrationBranchConcurrentAdvanceError`. The current code
* classifies these as `ref-update-refused`, so this class is here to
* recover legacy stuck rows from before the fix landed.
*/
export function classifyTransientMergeError(error: string | null | undefined): string | null {
if (!error) return null;
if (/lease-handoff-failed[^a-z]+target-not-queued/i.test(error)) {
return "lease-handoff-target-not-queued";
}
const sameSha = error.match(/advanced concurrently \(expected ([0-9a-f]{7,40}),\s+observed ([0-9a-f]{7,40})\)/i);
if (sameSha && sameSha[1].toLowerCase() === sameSha[2].toLowerCase()) {
return "spurious-concurrent-advance-same-sha";
}
return null;
}
const MAX_STARVATION_DROPS = 3;
const DEADLOCK_RECOVERY_COOLDOWN_MS = 15 * 60_000;
const DEFAULT_STALE_MERGING_STATUS_MIN_AGE_MS = 5 * 60_000;
@@ -782,6 +820,7 @@ export class SelfHealingManager {
{ name: "stale-incomplete-review", fn: () => this.recoverStaleIncompleteReviewTasks().then(() => undefined) },
{ name: "failed-pre-merge-steps", fn: () => this.recoverReviewTasksWithFailedPreMergeSteps().then(() => undefined) },
{ name: "interrupted-merging", fn: () => this.recoverInterruptedMergingTasks().then(() => undefined) },
{ name: "transient-merge-failures", fn: () => this.recoverTransientMergeFailures().then(() => undefined) },
{ name: "done-merge-metadata", fn: () => this.recoverDoneTaskMergeMetadata().then(() => undefined) },
{ name: "reconcile-done-task-integrity", fn: () => this.reconcileDoneTaskIntegrity().then(() => undefined) },
// FN-5092: must run BEFORE any merger pickup path so the merger queue is
@@ -1437,6 +1476,7 @@ export class SelfHealingManager {
{ name: "recover-stale-incomplete-review", fn: () => this.recoverStaleIncompleteReviewTasks() },
{ name: "recover-failed-pre-merge-steps", fn: () => this.recoverReviewTasksWithFailedPreMergeSteps() },
{ name: "recover-interrupted-merging", fn: () => this.recoverInterruptedMergingTasks() },
{ name: "recover-transient-merge-failures", fn: () => this.recoverTransientMergeFailures() },
{ name: "recover-done-merge-metadata", fn: () => this.recoverDoneTaskMergeMetadata() },
{ name: "recover-stale-merging-status", fn: () => this.recoverStaleMergingStatus() },
{ name: "finalize-noop-review", fn: () => this.finalizeNoOpReviewTasks() },
@@ -5107,6 +5147,170 @@ export class SelfHealingManager {
* No-op when `settings.autoMerge === false` — PR-based review flow owns lifecycle until human merge.
* @returns Number of tasks finalized or unblocked
*/
/**
* FN-5627 follow-up: recover in-review tasks that are stuck with
* `mergeRetries >= MAX_AUTO_MERGE_RETRIES` and `status='failed'` due to a
* TRANSIENT merge failure class (e.g., `target-not-queued` lease handoff
* race, legacy same-SHA spurious concurrent-advance). These tasks are
* NOT really stuck — they just hit a race or a misclassified ref-update
* failure that would have cleared on a fresh attempt. Without this sweep,
* the only path forward is manual intervention (the
* `AUTO_MERGE_COOLDOWN_MS`-based reset takes hours).
*
* For each matching task:
* - Reset `mergeRetries=0`, clear `status` and `error`.
* - Increment `mergeDetails.transientRecoveryCount`.
* - Re-enqueue via `requeueForAutoMerge`.
* - Bounded by `MAX_TRANSIENT_MERGE_RECOVERIES`; exhausted tasks stay
* parked as failed and emit `merger:transient-failure-budget-exhausted`
* once for diagnostic visibility.
*
* No-op when `settings.autoMerge === false`, no `requeueForAutoMerge`
* callback is wired, or global/engine pause is active.
*
* @returns Number of tasks recovered
*/
async recoverTransientMergeFailures(): Promise<number> {
const requeue = this.options.requeueForAutoMerge;
if (!requeue) return 0;
try {
const settings = await this.store.getSettings();
if (settings.autoMerge === false) return 0;
if (settings.globalPause || settings.enginePaused) return 0;
const slim = await this.store.listTasks({ column: "in-review", slim: true });
const candidates = slim.filter((t) =>
t.column === "in-review"
&& t.status === "failed"
&& (t.mergeRetries ?? 0) >= MAX_AUTO_MERGE_RETRIES
&& typeof t.error === "string"
&& t.error.length > 0
&& classifyTransientMergeError(t.error) !== null,
);
if (candidates.length === 0) return 0;
log.warn(
`Found ${candidates.length} in-review task(s) with transient merge failures stuck at mergeRetries=${MAX_AUTO_MERGE_RETRIES}; attempting auto-recovery`,
);
let recovered = 0;
for (const slimTask of candidates) {
const task = await this.store.getTask(slimTask.id).catch(() => null);
if (!task) continue;
// Re-check selector on the full row — the slim snapshot is best-effort
// and may be stale once we await.
if (
task.column !== "in-review"
|| task.status !== "failed"
|| (task.mergeRetries ?? 0) < MAX_AUTO_MERGE_RETRIES
) {
continue;
}
const errorText = task.error ?? "";
const transientClass = classifyTransientMergeError(errorText);
if (!transientClass) continue;
const currentCount = task.mergeDetails?.transientRecoveryCount ?? 0;
const errorSnippet = errorText.slice(0, 200);
const audit = createRunAuditor(this.store, {
runId: generateSyntheticRunId("self-heal-transient-merge-recovery", task.id),
agentId: "self-healing",
taskId: task.id,
phase: "recover-transient-merge-failures",
});
if (currentCount >= MAX_TRANSIENT_MERGE_RECOVERIES) {
// Budget exhausted — emit once for diagnostic visibility, leave parked.
// Repeat-suppression: if the failure error has already accreted the
// exhaustion marker, skip emit to avoid log spam.
if (!errorText.includes("[transient-recovery-budget-exhausted]")) {
await this.store.logEntry(
task.id,
`[FN-5627] Transient merge failure auto-recovery budget exhausted (${transientClass}, ${currentCount}/${MAX_TRANSIENT_MERGE_RECOVERIES}). Task remains parked in in-review for manual review.`,
);
await this.store.updateTask(task.id, {
error: `${errorText} [transient-recovery-budget-exhausted]`,
});
try {
await audit.database({
type: "merger:transient-failure-budget-exhausted",
target: task.id,
metadata: {
taskId: task.id,
transientClass,
recoveryCount: currentCount,
maxRecoveries: MAX_TRANSIENT_MERGE_RECOVERIES,
errorSnippet,
},
});
} catch (auditErr) {
log.warn(
`recoverTransientMergeFailures: audit emit failed for ${task.id}: ${
auditErr instanceof Error ? auditErr.message : String(auditErr)
}`,
);
}
}
continue;
}
const nextCount = currentCount + 1;
await this.store.logEntry(
task.id,
`[FN-5627] Auto-recovering transient merge failure (${transientClass}); resetting mergeRetries=0 and re-enqueueing (recovery ${nextCount}/${MAX_TRANSIENT_MERGE_RECOVERIES}).`,
);
await this.store.updateTask(task.id, {
mergeRetries: 0,
status: null,
error: null,
mergeDetails: {
...task.mergeDetails,
transientRecoveryCount: nextCount,
},
});
try {
await audit.database({
type: "merger:transient-failure-auto-recovered",
target: task.id,
metadata: {
taskId: task.id,
transientClass,
mergeRetries: task.mergeRetries ?? 0,
recoveryCount: nextCount,
errorSnippet,
},
});
} catch (auditErr) {
log.warn(
`recoverTransientMergeFailures: audit emit failed for ${task.id}: ${
auditErr instanceof Error ? auditErr.message : String(auditErr)
}`,
);
}
try {
await requeue(task.id);
recovered++;
} catch (requeueErr) {
log.warn(
`recoverTransientMergeFailures: requeue failed for ${task.id}: ${
requeueErr instanceof Error ? requeueErr.message : String(requeueErr)
}`,
);
}
}
if (recovered > 0) {
log.log(`recoverTransientMergeFailures: re-enqueued ${recovered} stuck in-review task(s)`);
}
return recovered;
} catch (err) {
log.warn(
`recoverTransientMergeFailures sweep failed: ${err instanceof Error ? err.message : String(err)}`,
);
return 0;
}
}
async recoverInterruptedMergingTasks(): Promise<number> {
try {
const settings = await this.store.getSettings();