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

FN-7220: harden workflow recovery lineage checks

Browse Source 
Harden workflow recovery so pause/resume parks and already-merged detection preserve task ownership lineage.

- Reject already-merged recovery candidates with foreign task or lineage trailers before accepting patch-id, tree, or branch-tip matches.
- Recover benign pause/resume abort parks back to todo when the task has no live executor/session state.
- Add real-git and executor regression coverage plus architecture docs and a changeset for recovery behavior.

Files changed:
 .changeset/fn-7220-workflow-recovery-lineage.md    |   7 +
 docs/architecture.md                               |   8 +-
 .../already-merged-detector.real-git.test.ts       |  69 ++++++++
 .../executor-paused-abort-todo-benign.test.ts      | 194 ++++++++++++++++++---
 .../self-healing-already-merged.real-git.test.ts   |  64 ++++++-
 packages/engine/src/already-merged-detector.ts     | 113 +++++++++---
 packages/engine/src/executor.ts                    |  84 +++++++++
 packages/engine/src/self-healing.ts                | 134 ++++++++++++--
 8 files changed, 609 insertions(+), 64 deletions(-)

Fusion-Task-Id: FN-7220

Fusion-Task-Lineage: 186c65bc-fe1a-457a-b90a-abbd73044252

Co-authored-by: Fusion (runfusion.ai) <noreply@runfusion.ai>
This commit is contained in:
gsxdsm
2026-06-28 22:43:02 -07:00
parent d0a369e34d
commit e963be4088
8 changed files with 609 additions and 64 deletions
Download Patch File Download Diff File Expand all files Collapse all files

7
.changeset/fn-7220-workflow-recovery-lineage.md Normal file
View File

@@ -0,0 +1,7 @@
---
"@runfusion/fusion": patch
---
summary: Harden workflow graph recovery against stale plan replays and foreign landed tips.
category: fix
dev: Classifies stale in-review plan pause/resume replays and verifies task ownership before already-merged recovery finalization.

8
docs/architecture.md
View File

@@ -683,7 +683,7 @@ Runtime action-gate flow (v1):
- `recoverGhostReviewTasks()` is a fallback only for idle, non-terminal `in-review` states. Terminal/actionable states (notably `status: "failed"`) are preserved and **not** auto-kicked back to `todo`.
- `recoverPausedAbortFailures()` clears executor pause/resume abort parks only when the durable row is safe to recover. `todo`/`in-progress` rows are requeued for normal scheduling, while clean `in-review` rows (completed steps, not paused/user-paused/executing, auto-merge eligible, no confirmed or terminal merge evidence) have `status`/`error` cleared in place so review progression can continue. User hard-cancel, global/user pause, `autoMerge:false`, terminal merge, and live-execution guards remain operator-actionable. Successful recovery emits `task:auto-recover-paused-abort-park` with `preservedInReview` metadata.
- Workflow graph pause/resume is node-reentrant for typed engine-internal interruptions. When `WorkflowGraphExecutor` sees the graph abort signal or a node returns `value: "aborted"`, it stamps the interrupted node and `engine-pause` abort kind into graph context. `TaskExecutor` then uses the existing bounded `graphResumeRetryCount` budget to clear the transient abort, suppress failure notification with an `Auto-recovered:` task log, and re-enter the graph/task only under the same safety guards: no user/active global pause, no merge/finalize provenance, no genuine node failure, no terminal merge value, no `autoMerge:false` protected review row, and no active execution owner. Global-pause provenance from the graph-controller abort is re-entrant once the global pause has been lifted because it represents the same in-flight node interruption. Legacy pause-abort parks without the typed node marker remain operator-action failures.
- Workflow graph pause/resume is node-reentrant for typed engine-internal interruptions. When `WorkflowGraphExecutor` sees the graph abort signal or a node returns `value: "aborted"`, it stamps the interrupted node and `engine-pause` abort kind into graph context. `TaskExecutor` then uses the existing bounded `graphResumeRetryCount` budget to clear the transient abort, suppress failure notification with an `Auto-recovered:` task log, and re-enter the graph/task only under the same safety guards: no user/active global pause, no merge/finalize provenance, no genuine node failure, no terminal merge value, no `autoMerge:false` protected review row, and no active execution owner. Global-pause provenance from the graph-controller abort is re-entrant once the global pause has been lifted because it represents the same in-flight node interruption. Generic legacy pause-abort parks without the typed node marker remain operator-action failures except for the narrow `in-review`/`plan` stale-replay shape: hard-cancel pause provenance, `node:plan:value === "aborted"`, no typed interrupted node, no active task/user/global pause, no terminal merge value, no confirmed merge, auto-merge eligibility, and only a clean row or the exact stale plan pause-abort failure. That path logs `stale replay ignored`, clears only the stale failure state when present, preserves `in-review`, and never re-enters planning or moves the task to `todo`.
- `reattach-orphaned-assigned-executions` is a forward-resume safety net for durable-agent assignments. During startup recovery and periodic maintenance, after orphaned-agent and stale-heartbeat-run repairs, self-healing finds `in-progress` tasks with an `assignedAgentId` whose agent has no active heartbeat run and no active executor session after the orphan grace window. It re-dispatches in place via `executor.resumeTaskForAgent(agentId)` (the same seam used by clean `HeartbeatMonitor.onRunCompleted` and guarded by executor double-execution checks), emits `task:reattach-orphaned-execution`, and never moves the task backward. This complements engine-start `executor.resumeOrphaned()` and leaves unassigned/role-based execution recovery to the existing startup/limbo/stuck-task paths.
- Durable `Agent.taskId` is a running assignment for parked `todo`/`triage` task rows only when the agent has live proof: a fresh active heartbeat run or an executor-active/tracked heartbeat signal. Scheduler overlap requeues, task move sync, self-healing, and Reports Health Check share this invariant: stale durable links are cleared or rendered as stale while `status: "queued"` and `overlapBlockedBy` remain on the task row so file-scope lease blocking is not weakened. `fn_list_agents` and `fn_agent_show` render the linked task column next to `Current Task` (for example `Current Task: FN-1234 (triage)` or `Current Task: FN-1234 (not active — done)`) so parked-column planning ownership is not misread as in-progress execution drift.
- Mission validation has a dedicated stale-run reaper: startup recovery and Batch 2 maintenance call `reapStaleMissionValidatorRuns()` when wired by the runtime, using `VALIDATOR_RUN_STALE_MAX_AGE_MS` (currently 6 hours). The sweep terminates ownerless `mission_validator_runs.status='running'` rows as `error`, writes the reap reason into `summary`, leaves `lastValidatorRunId` pointing at the now-terminal run, and emits run-audit telemetry with `mutationType: "mission:validator-run-reaped"` plus `runId`/`featureId`/`missionId`/`triggerType`/`elapsedMs` metadata. Active mission features move to `loopState="needs_fix"` + `lastValidatorStatus="error"` unless their parent mission is already `complete`/`archived`.
@@ -698,7 +698,7 @@ Active `fn_run_verification` subprocesses are a bounded progress signal (FN-6598
If loop recovery times out during compact-and-resume and the executor does not unwind within the bounded force-requeue grace window, `TaskExecutor.markStuckAborted()` now hard-cancels the hung task before clearing execution guards: spawned child agents are terminated, `awaitAbortInFlightTaskWork()` reaps API/step/workflow/configured-command/subagent/CLI surfaces, completed/in-progress steps are reconciled against committed branch state before any checkout deletion, the task worktree is removed with `RemovalReason.ExecutorStuckKilled`, stale in-memory worktree/loop/paused/stuck state is cleared, and then the task is moved back to `todo` with the configured `preserveProgressOnStuckRequeue` semantics. With preserve-progress enabled, committed step progress is retained; when the branch has no unique commits, affected steps are reset to `pending` before the worktree/branch are cleared so a retry cannot skip deleted uncommitted-only work. The path preserves the concurrent-recovery guard: if the latest task column is no longer `in-progress`, it only clears the execution guard and does not reap/remove resources that a self-healing recovery now owns. Task logs distinguish loop detection, compaction timeout, force-kill cleanup start, force-requeue, and cleanup completion/failure.
- `recoverMissingWorktreeReviewFailures()` is a narrow failed-review recovery: only `status: "failed"` `in-review` tasks with the explicit session-start signature `Refusing to start coding agent in missing worktree:` (from `assertValidWorktreeSession()`) are requeued. Recovery clears stale session metadata (`worktree`, `branch`, `sessionFile`, transient failure state), preserves valid step progress/retry counters, logs the auto-recovery reason, and moves the task back to `todo` for a clean retry.
- `recoverMergeableReviewTasks()` only re-enqueues truly eligible tasks; retry-exhausted review tasks are skipped to avoid re-enqueue/no-op loops that keep refreshing `updatedAt`.
- `recoverAlreadyMergedReviewTasks()` auto-finalizes retry-exhausted `in-review` tasks when self-healing can prove their work already landed on the merge target. On this landed-content path it clears soft blockers (`paused`, stale `status: "failed"`, and residual `error`) before moving to `done`; true hard blockers (for example incomplete steps, awaiting-user-review, or failed pre-merge workflow steps) still park the task in stable `in-review/failed` state with a blocker error instead of entering an auto-finalize loop.
- `recoverAlreadyMergedReviewTasks()` auto-finalizes retry-exhausted `in-review` tasks when self-healing can prove their work already landed on the merge target. On this landed-content path it clears soft blockers (`paused`, stale `status: "failed"`, and residual `error`) before moving to `done`; true hard blockers (for example incomplete steps, awaiting-user-review, or failed pre-merge workflow steps) still park the task in stable `in-review/failed` state with a blocker error instead of entering an auto-finalize loop. Already-merged/tip recovery must prove task ownership before setting `mergeDetails.mergeConfirmed` or moving to `done`: accepted evidence is a matching `Fusion-Task-Id`, matching `Fusion-Task-Lineage`, a task-ID anchored conventional subject, or a patch-id/tree-equal fallback from the canonical `fusion/<task-id>` branch whose tip and candidate commit are not explicitly attributed to another task/lineage. Foreign task tips (for example an FN-7143 row pointing at an FN-7187 tip) are rejected in place with `[recovery] already-merged rejected ... reason=foreign-task-tip` and `task:auto-recover-already-merged-rejected` audit metadata instead of finalizing the wrong task.
- `recoverTransientMergeFailures()` handles retry-exhausted `in-review` merge failures only when `classifyTransientMergeError()` returns a bounded transient class: `lease-handoff-target-not-queued`, `spurious-concurrent-advance-same-sha`, or `process-spawn-failure` (`spawn ENOTDIR`, `spawn … ENOENT`, or a clean-room path reported as `is not a working tree`). Recovery resets `mergeRetries`, clears transient `status`/`error`, increments `mergeDetails.transientRecoveryCount`, and requeues auto-merge so the next attempt recreates the AI-merge clean room. The budget stays capped by `MAX_TRANSIENT_MERGE_RECOVERIES`; exhausted tasks remain parked with the `merger:transient-failure-budget-exhausted` audit path so real structural failures cannot loop forever. FN-6278 makes this recovery mostly after-the-fact insurance for cwd spawn faults: the merge runner now preflights reuse integration roots and repairs/reacquires missing or de-registered task worktrees before the first git spawn, so a stale `task.worktree` should not consume the transient recovery budget by repeatedly producing `spawn git ENOENT`.
- `reconcileTaskWorktreeMetadata()` (FN-4962) reconciles stale `task.worktree`/`task.branch` rows against authoritative `git worktree list --porcelain` branch mappings during startup recovery, periodic maintenance, and completion fan-out. The stage must run before `reclaim-stale-active-branches`: stale rows rebound to live `fusion/<id>` worktrees emit `task:auto-recover-worktree-metadata-rebound`; stale rows with no live branch mapping are nulled (`worktree=null`, `branch=null`, `baseCommitSha` unchanged) and emit `task:auto-recover-worktree-metadata-cleared`.
- `recoverInProgressLimbo()` (FN-5219) is the safety net for stranded executor rows: reset/requeue paths must never leave a task in `in-progress` without a runnable execution context. After metadata reconcile, stale `in-progress` tasks with null branch, missing/cleared worktree metadata, no live executor claim, and all-pending steps are audited and moved back to `todo`.
@@ -707,7 +707,7 @@ If loop recovery times out during compact-and-resume and the executor does not u
`recoverOrphanOnlyScopeViolations()` handles the narrow FN-4350 shape without weakening the file-scope invariant: it runs only when all of these predicates hold — task is `column === "in-review"`; task is failed (`status === "failed"`, with engine/global pause both off); error evidence is a FileScopeViolation (`tool_error` agent-log payload from `formatFileScopeViolationAgentLog`, with `task.error` prefix fallback); `task.scopeOverride !== true`; task is not actively executing and `mergeDetails.mergeConfirmed !== true`. It then verifies the task's specific work is already on `main` using `findAlreadyMergedTaskCommit` (Fusion-Task-Id trailer / ancestry / patch-id / tree-equality proof). Only when staged files are orphan-only (no declared-scope overlap after excluding `.changeset/*`) and main-branch proof is positive does it finalize as a no-op (`resolutionStrategy: "orphan-discard-no-op"`), append an explicit auto-recovery log line, and tear down the task worktree so orphan staging is discarded.
Guardrails: this routine does **not** retry merges, does **not** apply to mixed/non-orphan staging, and does **not** run when no landed-work proof exists (FN-4280 class protection).
- FN-4285 decision: add a follow-up for a tree-equality recovery strategy (`rev-parse <base>^{tree}` == `<task-branch>^{tree}`) in `findAlreadyMergedTaskCommit`. This closes stranded already-merged branches that evade trailer/ancestry/patch-id matching, with guardrails limited to retry-exhausted review tasks to avoid false positives during transient post-rebase parity windows.
- FN-4285 decision: tree-equality recovery (`rev-parse <base>^{tree}` == `<task-branch>^{tree}`) in `findAlreadyMergedTaskCommit` closes stranded already-merged branches that evade trailer/ancestry/patch-id matching. FN-7220 tightens the guardrail: patch-id and tree-equal matches imply ownership only from the canonical task branch and are rejected when either the branch tip or candidate merge-target commit carries a foreign Fusion task/lineage trailer.
- No-`fn_task_done` recovery classification is normalized across executor, restart recovery, and self-healing: detection keys on executor-emitted `"without calling fn_task_done"` strings (while still tolerating legacy `task_done` wording), then applies the bounded ladder deterministically (in-session retries → bounded todo requeues with preserved progress when appropriate → terminal surfaced failure when budget is exhausted).
- `clearStaleBlockedBy()` clears `blockedBy` (and transient `status`) on todo tasks when their blocker is missing, done, archived, paused in-review, or failed in-review with merge retries exhausted. FN-3924 extends this with a dependency-integrity guard: if a task has explicit dependencies and `blockedBy` is not one of the currently unresolved deps, the stale marker is cleared. FN-4091 broadens the sweep to active `in-progress` and un-paused `in-review` tasks as well, but those repairs only null `blockedBy` (they do not rewrite scheduler-owned queued state). FN-5488 adds two fast paths: (1) failed in-review blockers at/above `MAX_AUTO_MERGE_RETRIES` always fan out unblock recovery with explicit reason codes, and (2) `status="merging"|"merging-pr"` blockers with no active merger owner are treated as unbacked after a short grace window (`unbackedMergingFanoutGraceMs`, default 60s) so manual retry/unpause `updatedAt` refreshes cannot deadlock downstream todos indefinitely. Recovery logs now use `Auto-recovered (FN-5488): ... reason=<code>` for auditability while preserving FN-4538 overlap-blocking invariants.
- FN-5624 suppresses transient worktree-local `.fusion/tasks/<id>/task.json` ENOENT session-start failures. When the missing file path is under `task.worktree`, executor routes through unusable-worktree auto-recovery, skips persisting `status: "failed"`/`error` on the task row, and emits `[transient-task-json-suppressed] ... reason=missing-task-json-under-worktree`. The corresponding self-healing `Auto-recovered:` log entry keeps notification suppression aligned with the existing `/^Auto-recovered:/` grace-window rule.
@@ -739,6 +739,8 @@ Guardrails: this routine does **not** retry merges, does **not** apply to mixed/
- `AgentLogger` (`agent-logger.ts`) — structured per-agent run logging
- `RunAudit` (`run-audit.ts`) — mutation audit tracking (DB/git/filesystem)
- FN-7214: `task:reenter-paused-aborted-workflow-node` records executor re-entry after a typed workflow graph node was interrupted by engine pause/resume. Metadata includes `nodeId`, `fromColumn`, retry `attempt`/`maxAttempts`, `abortProvenance`, whether the task was preserved in `in-review`, and the re-entry `mode`.
- FN-7220: `task:classify-stale-in-review-plan-pause-abort-replay` records executor classification of a stale generic `in-review` plan-node pause/resume replay. Metadata includes `nodeId`, `fromColumn`, `abortProvenance`, whether a stale failure was cleared, `graphResumeRetryCount`, and `mode: "preserved-in-review"`.
- FN-7220: `task:auto-recover-already-merged-rejected` records self-healing rejection of cross-task already-merged/tip metadata. Metadata includes `reason` (`foreign-task-tip`, `foreign-lineage-tip`, or `foreign-landed-commit`), `phase`, candidate SHA, candidate owner when known, task branch, and merge target branch.
- FN-6782/FN-6796: `task:auto-recover-paused-abort-park` records self-healing recovery of pause-abort operator parks. Metadata includes the source column and whether recovery preserved a clean `in-review` row instead of requeueing to `todo`.
- FN-7069: `task:reconcile-phantom-committed-reservation` records task-store startup or self-healing cleanup of committed-reservation-without-task phantoms. Metadata includes `reservationStatus: "committed"` plus pruned `activityLog` and `agents` counts; `runAuditEvents` and the committed reservation are intentionally retained for auditability and ID permanence.
- FN-7074: `task:reservation-commit-rolled-back` records preventive create-path rollback when a distributed reservation was committed with the task-row insert but a later create materialization step failed. Metadata includes `{ reservationId, nodeId, reason: "failed-create", error }`; the task row/partial directory are removed and the reservation is moved to `aborted` so FN-7069 should not need to clean up a new phantom.

69
packages/engine/src/__tests__/already-merged-detector.real-git.test.ts
View File

@@ -52,6 +52,29 @@ describeIfGit("findAlreadyMergedTaskCommit ownership anchoring (real git)", () =
expect(result).not.toBeNull();
expect(result!.sha).toBe(landedSha);
expect(result!.strategy).toBe("trailer");
expect(result!.ownershipProof).toBe("task-trailer");
});
it("attributes legacy task-id trailer even when the commit also carries lineage but task has none", async () => {
const repo = setupRepo();
mkdirSync(path.join(repo, "src"), { recursive: true });
writeFileSync(path.join(repo, "src", "legacy-lineage.txt"), "owned legacy lineage\n", "utf-8");
git(
repo,
"git add src/legacy-lineage.txt && git commit -m 'feat: legacy lineage landed' -m 'Fusion-Task-Id: FN-AMD-LEGACY' -m 'Fusion-Task-Lineage: LINEAGE-OPTIONAL'",
);
const landedSha = git(repo, "git rev-parse HEAD");
const result = await findAlreadyMergedTaskCommit({
taskId: "FN-AMD-LEGACY",
repoDir: repo,
baseBranch: "main",
});
expect(result).not.toBeNull();
expect(result!.sha).toBe(landedSha);
expect(result!.strategy).toBe("trailer");
expect(result!.ownershipProof).toBe("task-trailer");
});
it("attributes via lineage trailer when present", async () => {
@@ -71,6 +94,7 @@ describeIfGit("findAlreadyMergedTaskCommit ownership anchoring (real git)", () =
expect(result).not.toBeNull();
expect(result!.sha).toBe(landedSha);
expect(result!.strategy).toBe("trailer");
expect(result!.ownershipProof).toBe("lineage-trailer");
});
// Incident bug #2 regression: a commit that merely *mentions* the task ID in
@@ -124,6 +148,50 @@ describeIfGit("findAlreadyMergedTaskCommit ownership anchoring (real git)", () =
}
});
it("rejects a patch-id match when the landed candidate carries a foreign task trailer", async () => {
const repo = setupRepo();
git(repo, "git checkout -b fusion/fn-amd-foreign");
mkdirSync(path.join(repo, "src"), { recursive: true });
writeFileSync(path.join(repo, "src", "foreign-patch.txt"), "same-content\n", "utf-8");
git(repo, "git add src/foreign-patch.txt && git commit -m 'work without owner'");
const branchBase = git(repo, "git merge-base main fusion/fn-amd-foreign");
git(repo, "git checkout main");
mkdirSync(path.join(repo, "src"), { recursive: true });
writeFileSync(path.join(repo, "src", "foreign-patch.txt"), "same-content\n", "utf-8");
git(repo, "git add src/foreign-patch.txt && git commit -m 'feat: foreign landed' -m 'Fusion-Task-Id: FN-AMD-OTHER'");
const result = await findAlreadyMergedTaskCommit({
taskId: "FN-AMD-FOREIGN",
repoDir: repo,
baseBranch: "main",
taskBranch: "fusion/fn-amd-foreign",
baseCommitSha: branchBase,
});
expect(result).toBeNull();
});
it("rejects branch-fallback attribution when task metadata points at another task branch", async () => {
const repo = setupRepo();
git(repo, "git checkout -b fusion/fn-amd-other-tip");
mkdirSync(path.join(repo, "src"), { recursive: true });
writeFileSync(path.join(repo, "src", "other-tip.txt"), "other\n", "utf-8");
git(repo, "git add src/other-tip.txt && git commit -m 'feat: other tip' -m 'Fusion-Task-Id: FN-AMD-OTHER-TIP'");
git(repo, "git checkout main");
mkdirSync(path.join(repo, "src"), { recursive: true });
writeFileSync(path.join(repo, "src", "other-tip.txt"), "other\n", "utf-8");
git(repo, "git add src/other-tip.txt && git commit -m 'land equivalent other tip'");
const result = await findAlreadyMergedTaskCommit({
taskId: "FN-AMD-RECOVERED",
repoDir: repo,
baseBranch: "main",
taskBranch: "fusion/fn-amd-other-tip",
});
expect(result).toBeNull();
});
it("attributes via ancestry when the landed commit carries a conventional-subject anchor", async () => {
const repo = setupRepo();
@@ -147,5 +215,6 @@ describeIfGit("findAlreadyMergedTaskCommit ownership anchoring (real git)", () =
// Trailer path won't match (no trailer); ownership-anchored ancestry should.
const subject = git(repo, `git show -s --format=%s ${result!.sha}`);
expect(subject).toContain("FN-AMD-3");
expect(result!.ownershipProof).toBe("subject-anchor");
});
});

194
packages/engine/src/__tests__/executor-paused-abort-todo-benign.test.ts
View File

@@ -281,7 +281,7 @@ describe("pause-abort benign requeue-to-todo (FN-6782)", () => {
expect(logText(store)).toContain("operator action required");
});
it("auto-recovers an in-review paused-aborted in-flight workflow node without operator-action parking", async () => {
it("classifies an in-review typed plan interruption as stale without operator-action parking", async () => {
const { store, task, executor } = makeHarness({ column: "in-review" });
const graphSpy = vi.spyOn(executor as any, "maybeExecuteWorkflowGraph").mockResolvedValue(true);
@@ -300,29 +300,22 @@ describe("pause-abort benign requeue-to-todo (FN-6782)", () => {
(call: unknown[]) => (call[1] as { status?: string } | undefined)?.status === "failed",
);
expect(parkedFailed).toBe(false);
expect(logText(store)).toContain("Auto-recovered: re-entering paused-aborted workflow graph node 'plan'");
expect(logText(store)).toContain("stale replay ignored, in-review state preserved");
expect(logText(store)).not.toContain("operator action required");
const bumpedRetry = store.updateTask.mock.calls.some(
(call: unknown[]) => {
const patch = call[1] as { graphResumeRetryCount?: number; status?: unknown; error?: unknown } | undefined;
return patch?.graphResumeRetryCount === 1 && patch?.status === null && patch?.error === null;
},
);
expect(bumpedRetry).toBe(true);
expect(store.updateTask.mock.calls.some(
(call: unknown[]) => (call[1] as { graphResumeRetryCount?: number } | undefined)?.graphResumeRetryCount !== undefined,
)).toBe(false);
expect(store.recordRunAuditEvent).toHaveBeenCalledWith(expect.objectContaining({
mutationType: "task:reenter-paused-aborted-workflow-node",
mutationType: "task:classify-stale-in-review-plan-pause-abort-replay",
metadata: expect.objectContaining({
nodeId: "plan",
fromColumn: "in-review",
attempt: 1,
maxAttempts: 2,
abortProvenance: "hard-cancel",
preservedInReview: true,
mode: "preserved-in-review",
}),
}));
await flushScheduledRetry();
expect(graphSpy).toHaveBeenCalledTimes(1);
expect(graphSpy).not.toHaveBeenCalled();
});
it("fire-time guard skips in-review graph re-entry when a graph run is already active", async () => {
@@ -330,9 +323,9 @@ describe("pause-abort benign requeue-to-todo (FN-6782)", () => {
const graphSpy = vi.spyOn(executor as any, "maybeExecuteWorkflowGraph").mockResolvedValue(true);
await invokeGraphFailure(executor, task, {
interruptedNodeId: "plan",
interruptedNodeId: "execute",
interruptedAbortKind: "engine-pause",
context: { "node:plan:value": "aborted", "node:plan:abortKind": "engine-pause" },
context: { "node:execute:value": "aborted", "node:execute:abortKind": "engine-pause" },
});
(executor as any).activeWorkflowGraphAbortControllers.set(task.id, new AbortController());
@@ -390,6 +383,159 @@ describe("pause-abort benign requeue-to-todo (FN-6782)", () => {
expect(executeSpy).toHaveBeenCalledTimes(1);
});
it("classifies a stale in-review plan pause/resume replay as benign without re-entering planning", async () => {
const { store, task, executor } = makeHarness({ column: "in-review", graphResumeRetryCount: 2 });
(executor as any).addActiveWorktree(task.id, task.worktree);
const graphSpy = vi.spyOn(executor as any, "maybeExecuteWorkflowGraph").mockResolvedValue(true);
const executeSpy = vi.spyOn(executor as any, "execute").mockResolvedValue(undefined);
await invokeGraphFailure(executor, task, {
visitedNodeIds: ["plan"],
context: { "node:plan:value": "aborted" },
});
const parkedFailed = store.updateTask.mock.calls.some(
(call: unknown[]) => (call[1] as { status?: string } | undefined)?.status === "failed",
);
expect(parkedFailed).toBe(false);
expect(logText(store)).toContain("stale replay ignored, in-review state preserved");
expect(logText(store)).not.toContain("operator action required");
expect(store.updateTask.mock.calls.some(
(call: unknown[]) => (call[1] as { graphResumeRetryCount?: number } | undefined)?.graphResumeRetryCount !== undefined,
)).toBe(false);
expect((executor as any).pausedAborted.has(task.id)).toBe(false);
expect((executor as any).activeWorktrees.has(task.id)).toBe(false);
expect(store.recordRunAuditEvent).toHaveBeenCalledWith(expect.objectContaining({
mutationType: "task:classify-stale-in-review-plan-pause-abort-replay",
metadata: expect.objectContaining({
nodeId: "plan",
fromColumn: "in-review",
abortProvenance: "hard-cancel",
clearedStaleFailure: false,
graphResumeRetryCount: 2,
mode: "preserved-in-review",
}),
}));
await flushScheduledRetry();
expect(graphSpy).not.toHaveBeenCalled();
expect(executeSpy).not.toHaveBeenCalled();
});
it("clears a prior stale operator-action failure for an in-review plan replay", async () => {
const { store, task, executor } = makeHarness({
column: "in-review",
status: "failed",
error: "Workflow graph failure surfaced after paused engine abort during pause/resume in 'in-review' at node 'plan' — operator action required; retry or explicitly unpause/resume after inspecting the task",
});
const graphSpy = vi.spyOn(executor as any, "maybeExecuteWorkflowGraph").mockResolvedValue(true);
await invokeGraphFailure(executor, task, {
visitedNodeIds: ["plan"],
context: { "node:plan:value": "aborted" },
});
expect(logText(store)).toContain("Auto-recovered: cleared stale in-review plan pause/resume replay failure");
expect(store.updateTask.mock.calls.some(
(call: unknown[]) => {
const patch = call[1] as { status?: unknown; error?: unknown } | undefined;
return patch?.status === null && patch?.error === null;
},
)).toBe(true);
expect(store.updateTask.mock.calls.some(
(call: unknown[]) => (call[1] as { status?: string } | undefined)?.status === "failed",
)).toBe(false);
await flushScheduledRetry();
expect(graphSpy).not.toHaveBeenCalled();
});
it("keeps manual retry of a prior plan pause-abort park in review instead of fresh planning", async () => {
const { store, task, executor } = makeHarness({
column: "in-review",
status: null,
error: null,
log: [{
action: "Workflow graph failure surfaced after paused engine abort during pause/resume in 'in-review' at node 'plan' — operator action required; retry or explicitly unpause/resume after inspecting the task",
timestamp: now,
}],
});
const graphSpy = vi.spyOn(executor as any, "maybeExecuteWorkflowGraph").mockResolvedValue(true);
const executeSpy = vi.spyOn(executor as any, "execute").mockResolvedValue(undefined);
await invokeGraphFailure(executor, task, {
visitedNodeIds: ["plan"],
context: { "node:plan:value": "aborted" },
});
expect(logText(store)).toContain("stale replay ignored, in-review state preserved");
expect(store.updateTask.mock.calls.some(
(call: unknown[]) => (call[1] as { status?: string } | undefined)?.status === "failed",
)).toBe(false);
await flushScheduledRetry();
expect(graphSpy).not.toHaveBeenCalled();
expect(executeSpy).not.toHaveBeenCalled();
});
it("does NOT classify a global-pause generic plan replay while global pause remains active", async () => {
const { store, task, executor } = makeHarness({ column: "in-review" }, "global-pause");
store.getSettings.mockResolvedValue({
maxConcurrent: 2,
maxWorktrees: 4,
pollIntervalMs: 15000,
autoMerge: true,
globalPause: true,
maxAutoMergeRetries: 3,
});
await invokeGraphFailure(executor, task, {
visitedNodeIds: ["plan"],
context: { "node:plan:value": "aborted" },
});
expect(logText(store)).toContain("operator action required");
expect(store.recordRunAuditEvent).not.toHaveBeenCalledWith(expect.objectContaining({
mutationType: "task:classify-stale-in-review-plan-pause-abort-replay",
}));
});
it.each([
{ label: "explicit user pause", overrides: { userPaused: true }, value: "aborted" },
{ label: "task pause", overrides: { paused: true }, value: "aborted" },
{ label: "non-clean real failure", overrides: { status: "failed", error: "plugin handler failed" }, value: "aborted" },
{ label: "human-gated autoMerge:false row", overrides: { autoMerge: false }, value: "aborted" },
{ label: "terminal contamination value", overrides: {}, value: "foreign-branch-contamination" },
])("does NOT classify a $label as stale in-review plan replay", async ({ overrides, value }) => {
const { store, task, executor } = makeHarness({ column: "in-review", ...overrides });
await invokeGraphFailure(executor, task, {
visitedNodeIds: ["plan"],
context: { "node:plan:value": value },
});
expect(logText(store)).toContain("operator action required");
expect(logText(store)).not.toContain("stale replay ignored");
expect(store.recordRunAuditEvent).not.toHaveBeenCalledWith(expect.objectContaining({
mutationType: "task:classify-stale-in-review-plan-pause-abort-replay",
}));
});
it("keeps the completed in-review pause-abort classifier distinct from stale plan replay", async () => {
const { store, task, executor } = makeHarness({
column: "in-review",
steps: [{ name: "Implement", status: "done" }],
});
await invokeGraphFailure(executor, task, {
visitedNodeIds: ["plan"],
context: { "node:plan:value": "aborted" },
});
expect(logText(store)).toContain("Workflow graph run ended during engine pause/resume while already in-review");
expect(logText(store)).not.toContain("stale replay ignored");
expect(store.recordRunAuditEvent).not.toHaveBeenCalledWith(expect.objectContaining({
mutationType: "task:classify-stale-in-review-plan-pause-abort-replay",
}));
});
it("STILL parks a genuine in-review node failure with no paused-node audit", async () => {
const { store, task, executor } = makeHarness({ column: "in-review" });
const graphSpy = vi.spyOn(executor as any, "maybeExecuteWorkflowGraph").mockResolvedValue(true);
@@ -407,7 +553,7 @@ describe("pause-abort benign requeue-to-todo (FN-6782)", () => {
expect(graphSpy).not.toHaveBeenCalled();
});
it("auto-recovers a global-pause in-review interrupted node after global resume", async () => {
it("classifies a global-pause in-review plan interruption after global resume", async () => {
const { store, task, executor } = makeHarness({ column: "in-review" }, "global-pause");
const graphSpy = vi.spyOn(executor as any, "maybeExecuteWorkflowGraph").mockResolvedValue(true);
@@ -417,10 +563,10 @@ describe("pause-abort benign requeue-to-todo (FN-6782)", () => {
context: { "node:plan:value": "aborted", "node:plan:abortKind": "engine-pause" },
});
expect(logText(store)).toContain("Auto-recovered: re-entering paused-aborted workflow graph node 'plan'");
expect(logText(store)).toContain("stale replay ignored, in-review state preserved");
expect(logText(store)).not.toContain("operator action required");
expect(store.recordRunAuditEvent).toHaveBeenCalledWith(expect.objectContaining({
mutationType: "task:reenter-paused-aborted-workflow-node",
mutationType: "task:classify-stale-in-review-plan-pause-abort-replay",
metadata: expect.objectContaining({
nodeId: "plan",
fromColumn: "in-review",
@@ -429,10 +575,10 @@ describe("pause-abort benign requeue-to-todo (FN-6782)", () => {
}),
}));
await flushScheduledRetry();
expect(graphSpy).toHaveBeenCalledTimes(1);
expect(graphSpy).not.toHaveBeenCalled();
});
it("does NOT auto-recover a global-pause in-review failure without an interrupted-node marker", async () => {
it("classifies a global-pause in-review plan replay without an interrupted-node marker after global resume", async () => {
const { store, task, executor } = makeHarness({ column: "in-review" }, "global-pause");
const graphSpy = vi.spyOn(executor as any, "maybeExecuteWorkflowGraph").mockResolvedValue(true);
@@ -441,7 +587,7 @@ describe("pause-abort benign requeue-to-todo (FN-6782)", () => {
context: { "node:plan:value": "aborted" },
});
expect(logText(store)).toContain("operator action required");
expect(logText(store)).toContain("stale replay ignored, in-review state preserved");
expect(store.recordRunAuditEvent).not.toHaveBeenCalledWith(expect.objectContaining({
mutationType: "task:reenter-paused-aborted-workflow-node",
}));
@@ -492,9 +638,9 @@ describe("pause-abort benign requeue-to-todo (FN-6782)", () => {
const graphSpy = vi.spyOn(executor as any, "maybeExecuteWorkflowGraph").mockResolvedValue(true);
await invokeGraphFailure(executor, task, {
interruptedNodeId: "plan",
interruptedNodeId: "execute",
interruptedAbortKind: "engine-pause",
context: { "node:plan:value": "aborted", "node:plan:abortKind": "engine-pause" },
context: { "node:execute:value": "aborted", "node:execute:abortKind": "engine-pause" },
});
expect(logText(store)).toContain("operator action required");

64
packages/engine/src/__tests__/self-healing-already-merged.real-git.test.ts
View File

@@ -250,7 +250,7 @@ describeIfGit("SelfHealingManager recoverAlreadyMergedReviewTasks (real git)", (
const repo = setupRepo();
mkdirSync(path.join(repo, "src"), { recursive: true });
writeFileSync(path.join(repo, "src", "other.txt"), "other\n", "utf-8");
git(repo, "git add src/other.txt && git commit -m 'feat: unrelated' -m 'Fusion-Task-Id: FN-OTHER'");
git(repo, "git add src/other.txt && git commit -m 'feat: unrelated generic tip'");
const unrelatedSha = git(repo, "git rev-parse HEAD");
writeFileSync(path.join(repo, "src", "misbound.txt"), "landed\n", "utf-8");
@@ -280,6 +280,68 @@ describeIfGit("SelfHealingManager recoverAlreadyMergedReviewTasks (real git)", (
);
}, 20_000);
it("rejects already-merged recovery when the task branch tip belongs to a foreign task", async () => {
const repo = setupRepo();
mkdirSync(path.join(repo, "src"), { recursive: true });
writeFileSync(path.join(repo, "src", "foreign-tip.txt"), "foreign\n", "utf-8");
git(repo, "git add src/foreign-tip.txt && git commit -m 'feat: foreign landed' -m 'Fusion-Task-Id: FN-7187'");
const foreignSha = git(repo, "git rev-parse HEAD");
const worktreePath = path.join(repo, ".worktrees", "fn-7143");
mkdirSync(path.dirname(worktreePath), { recursive: true });
git(repo, `git branch fusion/fn-7143 ${foreignSha}`);
git(repo, `git worktree add ${JSON.stringify(worktreePath)} fusion/fn-7143`);
const tasks: TaskMap = new Map([
["FN-7143", makeTask({ id: "FN-7143", column: "in-review", status: "failed", mergeRetries: 3, paused: false, baseBranch: "main", branch: "fusion/fn-7143", worktree: worktreePath })],
]);
const store = createStore(tasks);
const manager = new SelfHealingManager(store, { rootDir: repo, getExecutingTaskIds: () => new Set() });
await (manager as any).runMaintenance();
const task = tasks.get("FN-7143")!;
expect(task.column).toBe("in-review");
expect(task.mergeDetails?.mergeConfirmed).not.toBe(true);
expect((store as any).moveTask).not.toHaveBeenCalledWith("FN-7143", "done");
expect((store.logEntry as any).mock.calls.some((call: unknown[]) => String(call[1]).includes("already-merged rejected FN-7143") && String(call[1]).includes("owner=FN-7187"))).toBe(true);
expect((store as any).recordRunAuditEvent).toHaveBeenCalledWith(expect.objectContaining({
mutationType: "task:auto-recover-already-merged-rejected",
target: "FN-7143",
metadata: expect.objectContaining({ reason: "foreign-task-tip", candidateOwner: "FN-7187" }),
}));
}, 20_000);
it("rejects branch-misbound finalization when the misbound tip belongs to a foreign task", async () => {
const repo = setupRepo();
mkdirSync(path.join(repo, "src"), { recursive: true });
writeFileSync(path.join(repo, "src", "other.txt"), "other\n", "utf-8");
git(repo, "git add src/other.txt && git commit -m 'feat: unrelated' -m 'Fusion-Task-Id: FN-OTHER'");
const foreignSha = git(repo, "git rev-parse HEAD");
writeFileSync(path.join(repo, "src", "owned.txt"), "owned\n", "utf-8");
git(repo, "git add src/owned.txt && git commit -m 'feat: landed' -m 'Fusion-Task-Id: FN-TEST-FOREIGN-MISBOUND'");
const worktreePath = path.join(repo, ".worktrees", "fn-test-foreign-misbound");
mkdirSync(path.dirname(worktreePath), { recursive: true });
git(repo, `git branch fusion/fn-test-foreign-misbound ${foreignSha}`);
git(repo, `git worktree add ${JSON.stringify(worktreePath)} fusion/fn-test-foreign-misbound`);
const tasks: TaskMap = new Map([
["FN-TEST-FOREIGN-MISBOUND", makeTask({ id: "FN-TEST-FOREIGN-MISBOUND", column: "in-review", status: "failed", paused: false, baseBranch: "main", branch: "fusion/fn-test-foreign-misbound", worktree: worktreePath })],
]);
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-FOREIGN-MISBOUND")!;
expect(task.column).toBe("in-review");
expect(task.mergeDetails?.mergeConfirmed).not.toBe(true);
expect((store as any).moveTask).not.toHaveBeenCalledWith("FN-TEST-FOREIGN-MISBOUND", "done");
expect((store.logEntry as any).mock.calls.some((call: unknown[]) => String(call[1]).includes("already-merged rejected FN-TEST-FOREIGN-MISBOUND") && String(call[1]).includes("owner=FN-OTHER"))).toBe(true);
}, 20_000);
it("is idempotent across two maintenance passes", async () => {
const repo = setupRepo();
mkdirSync(path.join(repo, "src"), { recursive: true });

113
packages/engine/src/already-merged-detector.ts
View File

@@ -1,7 +1,7 @@
import { exec, execSync } from "node:child_process";
import { promisify } from "node:util";
import { resolveTaskWorkingBranch } from "./worktree-names.js";
import { canonicalFusionBranchName, resolveTaskWorkingBranch } from "./worktree-names.js";
const execAsync = promisify(exec);
@@ -16,9 +16,25 @@ export interface AlreadyMergedLookupInput {
baseCommitSha?: string;
}
export type AlreadyMergedOwnershipProof =
| "task-trailer"
| "lineage-trailer"
| "subject-anchor"
| "canonical-branch-patch"
| "canonical-branch-tree";
export interface AlreadyMergedLookupResult {
sha: string;
strategy: AlreadyMergedDetectionStrategy;
ownershipProof?: AlreadyMergedOwnershipProof;
}
export interface CommitTaskOwnership {
owned: boolean;
proof?: Extract<AlreadyMergedOwnershipProof, "task-trailer" | "lineage-trailer" | "subject-anchor">;
ownerTaskId?: string;
ownerLineageId?: string;
rejectionReason?: "foreign-task" | "foreign-lineage";
}
interface DetectAlreadyLandedInput {
@@ -38,33 +54,40 @@ function escapeRegex(value: string): string {
return value.replace(/[.*+?^${}()|[\]\\]/g, "\\$&");
}
function firstTrailerValue(body: string, trailer: "Fusion-Task-Id" | "Fusion-Task-Lineage"): string | undefined {
const match = body.match(new RegExp(`(?:^|\\n)${trailer}:\\s*([^\\n]+?)\\s*(?:\\n|$)`));
return match?.[1]?.trim();
}
/**
* Ownership anchor shared with self-healing's `commitOwnedByTask`.
* Ownership anchor shared with self-healing's already-merged recovery guards.
*
* The 2026-05-23 lost-work incident (bug #2) was a `git log --grep=<taskId>`
* first-hit attribution: a commit whose body merely *mentioned* a task ID in
* prose was accepted as that task's landed commit, stranding/mis-attributing
* the real work. The trailer strategies above are already anchored; the
* ancestry strategy below uses a loose `--grep=<taskId>`, so its candidate must
* be ownership-verified here before it is accepted.
*
* Accept when ANY of:
* - `Fusion-Task-Lineage: <lineageId>` is a complete trailer line in the body
* - `Fusion-Task-Id: <taskId>` is a complete trailer line in the body
* - the subject is anchored on the task ID in conventional-commit form:
* `<type>(<taskId>...): …` or `<taskId>: …`
* the real work. The FN-7143/FN-7187 incident added the inverse guard: explicit
* foreign Fusion trailers are rejection evidence even when a stale branch tip,
* patch-id, or tree-equality fallback otherwise appears to match.
*/
function commitOwnedByTask(
export function getCommitTaskOwnership(
taskId: string,
lineageId: string | undefined,
subject: string,
body: string,
): boolean {
if (lineageId && new RegExp(`(?:^|\\n)Fusion-Task-Lineage: ${escapeRegex(lineageId)}\\s*(?:\\n|$)`).test(body)) {
return true;
): CommitTaskOwnership {
const ownerTaskId = firstTrailerValue(body, "Fusion-Task-Id");
const ownerLineageId = firstTrailerValue(body, "Fusion-Task-Lineage");
if (ownerTaskId && ownerTaskId !== taskId) {
return { owned: false, ownerTaskId, ownerLineageId, rejectionReason: "foreign-task" };
}
if (new RegExp(`(?:^|\\n)Fusion-Task-Id: ${escapeRegex(taskId)}\\s*(?:\\n|$)`).test(body)) {
return true;
if (lineageId && ownerLineageId && ownerLineageId !== lineageId) {
return { owned: false, ownerTaskId, ownerLineageId, rejectionReason: "foreign-lineage" };
}
if (ownerTaskId === taskId) {
return { owned: true, proof: "task-trailer", ownerTaskId, ownerLineageId };
}
if (lineageId && ownerLineageId === lineageId) {
return { owned: true, proof: "lineage-trailer", ownerTaskId, ownerLineageId };
}
// Subject anchor MUST mention the task ID — either inside a conventional
// scope (`<type>(<…taskId…>): …`) or as a leading `<taskId>: …`. The scope
@@ -74,7 +97,26 @@ function commitOwnedByTask(
const subjectAnchor = new RegExp(
`^(?:[A-Za-z]+\\([^)]*\\b${escapeRegex(taskId)}\\b[^)]*\\):|${escapeRegex(taskId)}:)`,
);
return subjectAnchor.test(subject);
if (subjectAnchor.test(subject)) {
return { owned: true, proof: "subject-anchor", ownerTaskId, ownerLineageId };
}
return { owned: false, ownerTaskId, ownerLineageId };
}
async function commitHasForeignTaskOwnership(
repoDir: string,
sha: string,
taskId: string,
lineageId: string | undefined,
): Promise<boolean> {
const { stdout } = await execAsync(`git show -s --format=%s%x1f%b ${shellQuote(sha)}`, {
cwd: repoDir,
timeout: 30_000,
maxBuffer: 1024 * 1024,
});
const [subject = "", body = ""] = stdout.split("\x1f");
const ownership = getCommitTaskOwnership(taskId, lineageId, subject, body);
return ownership.rejectionReason === "foreign-task" || ownership.rejectionReason === "foreign-lineage";
}
export async function findAlreadyMergedTaskCommit(
@@ -100,7 +142,7 @@ export async function findAlreadyMergedTaskCommit(
});
const lineageSha = lineage.stdout.trim();
if (lineageSha) {
return { sha: lineageSha, strategy: "trailer" };
return { sha: lineageSha, strategy: "trailer", ownershipProof: "lineage-trailer" };
}
}
@@ -120,7 +162,7 @@ export async function findAlreadyMergedTaskCommit(
});
const sha = stdout.trim();
if (sha) {
return { sha, strategy: "trailer" };
return { sha, strategy: "trailer", ownershipProof: "task-trailer" };
}
} catch {
// Fall through to ancestry/patch-id checks.
@@ -128,12 +170,21 @@ export async function findAlreadyMergedTaskCommit(
let branchTip: string | null = null;
const branchName = resolveTaskWorkingBranch({ id: taskId, branch: taskBranch });
const canonicalBranchName = canonicalFusionBranchName(taskId);
/*
FNXC:WorkflowRecovery 2026-06-28-21:36:
FN-7143/FN-7187 proved patch-id and tree-equal fallbacks need branch identity proof, not just content equivalence. Only the canonical task branch may imply ownership for fallback matches, and any explicit foreign Fusion trailer on the branch tip or candidate commit rejects the recovery.
*/
const hasCanonicalBranchIdentity = branchName === canonicalBranchName;
try {
branchTip = execSync(`git rev-parse --verify ${shellQuote(branchName)}`, {
cwd: repoDir,
encoding: "utf-8",
stdio: ["pipe", "pipe", "pipe"],
}).trim();
if (await commitHasForeignTaskOwnership(repoDir, branchTip, taskId, lineageId)) {
return null;
}
execSync(`git merge-base --is-ancestor ${shellQuote(branchTip)} ${shellQuote(baseBranch)}`, {
cwd: repoDir,
@@ -165,8 +216,9 @@ export async function findAlreadyMergedTaskCommit(
for (const record of records) {
const [candidateSha, candidateSubject = "", candidateBody = ""] = record.split("\x1f");
const sha = candidateSha?.trim();
if (sha && commitOwnedByTask(taskId, lineageId, candidateSubject, candidateBody)) {
return { sha, strategy: "ancestry" };
const ownership = getCommitTaskOwnership(taskId, lineageId, candidateSubject, candidateBody);
if (sha && ownership.owned) {
return { sha, strategy: "ancestry", ownershipProof: ownership.proof };
}
}
} catch {
@@ -174,12 +226,18 @@ export async function findAlreadyMergedTaskCommit(
}
try {
if (!hasCanonicalBranchIdentity) {
return null;
}
if (!branchTip) {
branchTip = execSync(`git rev-parse --verify ${shellQuote(branchName)}`, {
cwd: repoDir,
encoding: "utf-8",
stdio: ["pipe", "pipe", "pipe"],
}).trim();
if (await commitHasForeignTaskOwnership(repoDir, branchTip, taskId, lineageId)) {
return null;
}
}
let branchBase = baseCommitSha?.trim();
@@ -231,8 +289,8 @@ export async function findAlreadyMergedTaskCommit(
}
const matchedSha = basePatchMap.get(branchPatchId);
if (matchedSha) {
return { sha: matchedSha, strategy: "patch-id" };
if (matchedSha && !await commitHasForeignTaskOwnership(repoDir, matchedSha, taskId, lineageId)) {
return { sha: matchedSha, strategy: "patch-id", ownershipProof: "canonical-branch-patch" };
}
} catch {
// Fall through to null when patch-id detection fails.
@@ -240,6 +298,9 @@ export async function findAlreadyMergedTaskCommit(
try {
const treeBranchName = resolveTaskWorkingBranch({ id: taskId, branch: taskBranch });
if (treeBranchName !== canonicalBranchName) {
return null;
}
execSync(`git rev-parse --verify ${shellQuote(treeBranchName)}`, {
cwd: repoDir,
encoding: "utf-8",
@@ -266,8 +327,8 @@ export async function findAlreadyMergedTaskCommit(
maxBuffer: 1024 * 1024,
});
const baseHead = baseHeadStdout.trim();
if (baseHead) {
return { sha: baseHead, strategy: "tree-equal" };
if (baseHead && !await commitHasForeignTaskOwnership(repoDir, baseHead, taskId, lineageId)) {
return { sha: baseHead, strategy: "tree-equal", ownershipProof: "canonical-branch-tree" };
}
}
} catch {

84
packages/engine/src/executor.ts
View File

@@ -6923,6 +6923,86 @@ export class TaskExecutor {
return true;
}
private isStalePauseAbortParkFailure(live: TaskDetail): boolean {
return live.status === "failed"
&& typeof live.error === "string"
&& live.error.includes(PAUSE_ABORT_PARK_ERROR_MARKER)
&& live.error.includes("engine abort during pause/resume")
&& live.error.includes("at node 'plan'");
}
private async handleStaleInReviewPlanPauseAbortReplay(
live: TaskDetail,
result: WorkflowGraphTaskRunResult,
abortProvenance: "global-pause" | "merge-seam" | "hard-cancel" | "completion-finalize" | undefined,
pausedAborted: boolean,
userCanceled: boolean,
): Promise<boolean> {
/*
FNXC:WorkflowLifecycle 2026-06-28-21:05:
FN-7143 showed that a stale graph lifecycle replay can surface at `plan` after an in-review pause/resume even though planning is not actually running anymore. Plan is not a safe re-entry point for review rows, typed or generic, so this classifier is clear/log-only: preserve in-review, never route to triage/todo, and keep genuine user/global pauses plus real plan failures on the operator-action path.
*/
if (!pausedAborted) return false;
if (abortProvenance !== "hard-cancel" && abortProvenance !== "global-pause") return false;
if (userCanceled) return false;
if (live.column !== "in-review") return false;
if (live.paused || live.userPaused === true) return false;
if (live.autoMerge === false) return false;
if (live.mergeDetails?.mergeConfirmed === true) return false;
if (result.interruptedAbortKind && result.interruptedAbortKind !== WORKFLOW_NODE_ENGINE_PAUSE_ABORT_KIND) return false;
const failedNode = result.interruptedNodeId ?? result.visitedNodeIds[result.visitedNodeIds.length - 1];
if (failedNode !== "plan") return false;
if (this.isMergeGraphFailure(failedNode)) return false;
const failureValue = typeof result.context?.[`node:${failedNode}:value`] === "string"
? result.context[`node:${failedNode}:value`] as string
: this.graphFailureValue(result);
if (failureValue !== "aborted") return false;
if (this.isTerminalMergeGraphFailureValue(failureValue)) return false;
const cleanRow = live.status == null && live.error == null;
const staleParkedFailure = this.isStalePauseAbortParkFailure(live);
if (!cleanRow && !staleParkedFailure) return false;
let settings: Settings;
try {
settings = await this.store.getSettings();
} catch {
return false;
}
if (settings.globalPause === true || settings.enginePaused === true) return false;
if (!allowsAutoMergeProcessing(live, settings) && !isSharedBranchGroupMemberIntegration(live)) return false;
this.clearPausedAborted(live.id);
this.activeWorktrees.delete(live.id);
const message = "Workflow graph plan node pause/resume replay surfaced after task was already in-review — stale replay ignored, in-review state preserved";
executorLog.log(`${live.id}: ${message}`);
await this.store.logEntry(live.id, message, undefined, this.getRunContextFor(live.id));
if (staleParkedFailure) {
await this.store.updateTask(live.id, { status: null, error: null }, this.getRunContextFor(live.id));
await this.store.logEntry(live.id, "Auto-recovered: cleared stale in-review plan pause/resume replay failure — failure notification suppressed", undefined, this.getRunContextFor(live.id));
}
try {
await this.store.recordRunAuditEvent?.({
taskId: live.id,
agentId: "executor",
runId: generateSyntheticRunId("workflow-stale-plan-replay", live.id),
domain: "database",
mutationType: "task:classify-stale-in-review-plan-pause-abort-replay",
target: live.id,
metadata: {
nodeId: failedNode,
fromColumn: live.column,
abortProvenance,
clearedStaleFailure: staleParkedFailure,
graphResumeRetryCount: live.graphResumeRetryCount ?? 0,
mode: "preserved-in-review",
},
});
} catch (error) {
executorLog.warn(`${live.id}: failed to record stale plan replay audit: ${error instanceof Error ? error.message : String(error)}`);
}
await this.persistTokenUsage(live.id);
return true;
}
private async isReentrantPausedAbortedInFlightNode(
live: TaskDetail,
result: WorkflowGraphTaskRunResult,
@@ -6945,6 +7025,7 @@ export class TaskExecutor {
if (live.column === "done" || live.column === "archived") return false;
if (result.interruptedAbortKind !== WORKFLOW_NODE_ENGINE_PAUSE_ABORT_KIND) return false;
if (!result.interruptedNodeId) return false;
if (live.column === "in-review" && result.interruptedNodeId === "plan") return false;
if (this.isMergeGraphFailure(result.interruptedNodeId)) return false;
if (this.isTerminalMergeGraphFailureValue(this.graphFailureValue(result))) return false;
if ((live.graphResumeRetryCount ?? 0) >= MAX_TRANSIENT_GRAPH_RESUME_RETRIES) return false;
@@ -7165,6 +7246,9 @@ export class TaskExecutor {
await this.persistTokenUsage(task.id);
return;
}
if (genuinePauseAbort && await this.handleStaleInReviewPlanPauseAbortReplay(live, result, abortProvenance, pausedAborted, this.userCanceledTaskIds.has(task.id))) {
return;
}
if (genuinePauseAbort) {
/*
FNXC:WorkflowLifecycle 2026-06-15-01:45:

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

@@ -56,7 +56,7 @@ imported it from merger-ai while merger-ai imports `MIN_TEMP_WORKTREE_REAP_AGE_M
self-healing — a real import cycle. Importing from the predicate module breaks the cycle.
*/
import { isRepoLanded } from "./workspace-land-predicate.js";
import { findAlreadyMergedTaskCommit } from "./already-merged-detector.js";
import { findAlreadyMergedTaskCommit, getCommitTaskOwnership } from "./already-merged-detector.js";
import { isAiMergeContainerDir, resolveAiMergeRootPath, resolveLegacyAiMergeRootPath, resolveWorktreesDir } from "./worktree-paths.js";
import { canonicalFusionBranchName, resolveTaskWorkingBranch } from "./worktree-names.js";
import { resolveIntegrationBranch } from "./integration-branch.js";
@@ -1875,6 +1875,79 @@ export class SelfHealingManager {
return findAlreadyMergedTaskCommit(input);
}
private async readCommitTaskOwnership(sha: string, taskId: string, lineageId?: string) {
const { stdout } = await execAsync(`git show -s --format=%s%x1f%b ${shellQuote(sha)}`, {
cwd: this.options.rootDir,
timeout: 30_000,
maxBuffer: 1024 * 1024,
});
const [subject = "", body = ""] = stdout.split("\x1f");
return getCommitTaskOwnership(taskId, lineageId, subject, body);
}
private async rejectForeignAlreadyMergedCandidate(input: {
task: Pick<Task, "id" | "lineageId">;
candidateSha: string;
candidateOwner?: string;
taskBranch?: string | null;
baseBranch: string;
reason: "foreign-task-tip" | "foreign-lineage-tip" | "foreign-landed-commit";
phase: string;
}): Promise<void> {
const { task, candidateSha, candidateOwner, taskBranch, baseBranch, reason, phase } = input;
/*
FNXC:WorkflowRecovery 2026-06-28-21:32:
FN-7143 observed an already-merged tip that appeared to belong to FN-7187. Self-healing must make that cross-task proof visible and leave the review task alone; ambiguous or foreign tips are not safe evidence for mergeConfirmed/done finalization.
*/
await this.store.logEntry(
task.id,
`[recovery] already-merged rejected ${task.id} candidate=${candidateSha.slice(0, 12)} owner=${candidateOwner ?? "unknown"} branch=${taskBranch ?? "?"} base=${baseBranch} reason=${reason}`,
);
try {
await this.store.recordRunAuditEvent?.({
taskId: task.id,
agentId: "self-healing",
runId: generateSyntheticRunId("self-heal-already-merged-rejected", task.id),
domain: "database",
mutationType: "task:auto-recover-already-merged-rejected",
target: task.id,
metadata: {
reason,
phase,
candidateSha,
candidateOwner: candidateOwner ?? null,
taskBranch: taskBranch ?? null,
baseBranch,
},
});
} catch (err: unknown) {
log.warn(`Failed to record already-merged rejection audit for ${task.id}: ${err instanceof Error ? err.message : String(err)}`);
}
}
private async branchTipForeignOwnership(input: {
taskId: string;
lineageId?: string;
branch: string;
}): Promise<{ sha: string; owner?: string; reason: "foreign-task-tip" | "foreign-lineage-tip" } | null> {
const { taskId, lineageId, branch } = input;
const { stdout } = await execAsync(`git rev-parse ${shellQuote(branch)}`, {
cwd: this.options.rootDir,
timeout: 30_000,
maxBuffer: 1024 * 1024,
});
const sha = stdout.trim();
if (!sha) return null;
const ownership = await this.readCommitTaskOwnership(sha, taskId, lineageId);
if (ownership.rejectionReason === "foreign-task") {
return { sha, owner: ownership.ownerTaskId, reason: "foreign-task-tip" };
}
if (ownership.rejectionReason === "foreign-lineage") {
return { sha, owner: ownership.ownerLineageId, reason: "foreign-lineage-tip" };
}
return null;
}
private async resolveSelfHealingMergeTarget(
task: Task,
settings: Settings | undefined,
@@ -3007,6 +3080,19 @@ export class SelfHealingManager {
}
if (inspection.kind === "tip-already-merged") {
const branchName = task.branch;
const ownership = await this.readCommitTaskOwnership(inspection.tipSha, task.id, task.lineageId).catch(() => null);
if (ownership?.rejectionReason === "foreign-task" || ownership?.rejectionReason === "foreign-lineage") {
await this.rejectForeignAlreadyMergedCandidate({
task,
candidateSha: inspection.tipSha,
candidateOwner: ownership.rejectionReason === "foreign-task" ? ownership.ownerTaskId : ownership.ownerLineageId,
taskBranch: branchName,
baseBranch: inspection.integrationRef,
reason: ownership.rejectionReason === "foreign-task" ? "foreign-task-tip" : "foreign-lineage-tip",
phase: "tip-already-merged",
});
continue;
}
let reclaimedCleanly = false;
try {
if (inspection.livePath && existsSync(inspection.livePath)) {
@@ -8260,6 +8346,21 @@ export class SelfHealingManager {
const mergeTarget = await this.resolveSelfHealingMergeTarget(task, settings, "recover-already-merged-review");
const baseBranch = mergeTarget.branch;
if (!baseBranch) continue;
if (task.branch) {
const foreignTip = await this.branchTipForeignOwnership({ taskId: task.id, lineageId: task.lineageId, branch: task.branch }).catch(() => null);
if (foreignTip) {
await this.rejectForeignAlreadyMergedCandidate({
task,
candidateSha: foreignTip.sha,
candidateOwner: foreignTip.owner,
taskBranch: task.branch,
baseBranch,
reason: foreignTip.reason,
phase: "recover-already-merged-review",
});
continue;
}
}
const landed = await this.findAlreadyMergedTaskCommit({
taskId: task.id,
@@ -8564,22 +8665,23 @@ export class SelfHealingManager {
taskId: string;
lineageId?: string;
baseBranch: string;
}): Promise<{ misbound: boolean; branchTip: string; landed: Awaited<ReturnType<typeof findAlreadyMergedTaskCommit>> }> {
}): Promise<{ misbound: boolean; branchTip: string; landed: Awaited<ReturnType<typeof findAlreadyMergedTaskCommit>>; rejection?: { reason: "foreign-task-tip" | "foreign-lineage-tip"; owner?: string } }> {
const { branch, taskId, lineageId, baseBranch } = input;
const { stdout: bodyOut } = await execAsync(`git log -1 --format=%B ${shellQuote(branch)}`, {
cwd: this.options.rootDir,
timeout: 30_000,
maxBuffer: 1024 * 1024,
});
const body = bodyOut;
const hasTaskId = body.includes(`Fusion-Task-Id: ${taskId}`);
const hasLineage = lineageId ? body.includes(`Fusion-Task-Lineage: ${lineageId}`) : false;
const { stdout: tipOut } = await execAsync(`git rev-parse ${shellQuote(branch)}`, {
cwd: this.options.rootDir,
timeout: 30_000,
maxBuffer: 1024 * 1024,
});
const branchTip = tipOut.trim();
const ownership = await this.readCommitTaskOwnership(branchTip, taskId, lineageId);
if (ownership.rejectionReason === "foreign-task") {
return { misbound: false, branchTip, landed: null, rejection: { reason: "foreign-task-tip", owner: ownership.ownerTaskId } };
}
if (ownership.rejectionReason === "foreign-lineage") {
return { misbound: false, branchTip, landed: null, rejection: { reason: "foreign-lineage-tip", owner: ownership.ownerLineageId } };
}
const hasTaskId = ownership.ownerTaskId === taskId;
const hasLineage = lineageId ? ownership.ownerLineageId === lineageId : false;
const landed = await this.findAlreadyMergedTaskCommit({
taskId,
lineageId,
@@ -8627,6 +8729,18 @@ export class SelfHealingManager {
lineageId: task.lineageId,
baseBranch,
});
if (check.rejection) {
await this.rejectForeignAlreadyMergedCandidate({
task,
candidateSha: check.branchTip,
candidateOwner: check.rejection.owner,
taskBranch: branch,
baseBranch,
reason: check.rejection.reason,
phase: "recover-branch-misbound-in-review",
});
continue;
}
if (!check.misbound || !check.landed) continue;
const mergeDetails: MergeDetails = {