From 06ea44433956fd8ae9b9984cc15580b84e286225 Mon Sep 17 00:00:00 2001 From: gsxdsm Date: Tue, 11 Aug 2026 02:02:51 -0700 Subject: [PATCH] fix: uncap workflow principals and auto-resume stranded continuations Two independent wedges kept cards silently stuck on the board. 1. Workflow principals were capped. `WorkflowAgentCapacity.acquire` enforced `settings.maxConcurrent` as a project session budget plus an optional per-agent `maxWorkflowSessions`, and `routeWorkflowPrincipal`'s availability test applied the same per-agent ceiling. The workflow roles stand in for STAGES, not workers, and there is typically one agent per role - so the cap serialized the entire board behind a single Workflow Executor regardless of maxConcurrent/maxWorktrees. Admission now always succeeds; the lease survives as bookkeeping (it is what activeSessions counts and what the renewal timer keeps warm). `maxProjectSessions` is removed from the input rather than defaulted, so it cannot be reintroduced without deleting the contract, and the agent-capacity re-route loops in triage and graph admission are deleted with the refusal they existed to work around. 2. Continuations that stop in `running` or `held` were never re-polled. The scheduler's due-poll takes only `runnable`/`retrying`; a row claimed through a path that leaves `leaseExpiresAt` NULL keeps `state: "running"` forever after its process dies, and `acquireWorkflowWorkItemLease` can only re-take a `held` row whose blockedReason matches workflow-principal-%. Observed live: seven cards `running` behind leases from a process that exited ~9h earlier, two `held` with a NULL blockedReason for 46h, none emitting a single run-audit row while stranded. A further 33 active-state rows belonged to archived+soft-deleted tasks (the FK cascade only fires on hard delete). New sweep `reconcileStrandedWorkflowContinuations` (startup + periodic) re-queues both stranded shapes and retires dead tasks' rows, gated by the canonical liveness triple, a 10-minute grace matching the capacity lease duration, and a compare-and-set on the scanned state so a real claim wins. The decision is the pure `evaluateStrandedContinuationReclaim`, shared with its tests so coverage cannot drift from behavior - the drift that let the FN-8923 sweep ship covering one ninth of this problem. Verified: pnpm lint, engine typecheck, pnpm test:gate (606 tests), verify:fast, and the new suite under mutation (removing either guard fails 3 cases). The two pre-existing failures in self-healing-orphaned-pending-step-results.test.ts reproduce identically at HEAD without these changes. Co-Authored-By: Claude Opus 5 --- ...low-principals-and-continuation-reclaim.md | 7 + ...r-ephemeral-disabled-dispatch-gate.test.ts | 8 +- ...ling-stranded-continuation-reclaim.test.ts | 194 ++++++++++++++++++ .../__tests__/workflow-agent-capacity.test.ts | 34 ++- .../src/agents/workflow-agent-capacity.ts | 39 ++-- .../src/agents/workflow-agent-router.ts | 20 +- .../workflow-principal-before-node.ts | 55 ++--- packages/engine/src/self-healing.ts | 141 +++++++++++++ packages/engine/src/triage.ts | 48 ++--- .../stranded-continuation-reclaim.ts | 116 +++++++++++ 10 files changed, 561 insertions(+), 101 deletions(-) create mode 100644 .changeset/uncapped-workflow-principals-and-continuation-reclaim.md create mode 100644 packages/engine/src/__tests__/self-healing-stranded-continuation-reclaim.test.ts create mode 100644 packages/engine/src/workflows/stranded-continuation-reclaim.ts diff --git a/.changeset/uncapped-workflow-principals-and-continuation-reclaim.md b/.changeset/uncapped-workflow-principals-and-continuation-reclaim.md new file mode 100644 index 0000000000..878e9cb443 --- /dev/null +++ b/.changeset/uncapped-workflow-principals-and-continuation-reclaim.md @@ -0,0 +1,7 @@ +--- +"@runfusion/fusion": minor +--- + +summary: Remove workflow principal session caps and auto-resume continuations stranded in running or held. +category: fix +dev: `WorkflowAgentCapacity.acquire` drops `maxProjectSessions`/`maxWorkflowSessions` (leases become bookkeeping only) and `routeWorkflowPrincipal`'s availability test is now eligibility-only, so the capacity re-route loops in `triage.ts` and `workflow-principal-before-node.ts` are deleted. New self-healing sweep `reconcileStrandedWorkflowContinuations` (startup + periodic) re-queues `running` rows with a dead/absent lease and `held` rows the claim predicate cannot re-take, and retires active-state rows belonging to deleted/archived tasks; decision logic is the pure `evaluateStrandedContinuationReclaim`. New run-audit types: `workflowWorkItem:reconcile-stranded-requeued`, `workflowWorkItem:reconcile-stranded-retired`. diff --git a/packages/engine/src/__tests__/executor-ephemeral-disabled-dispatch-gate.test.ts b/packages/engine/src/__tests__/executor-ephemeral-disabled-dispatch-gate.test.ts index fbc3b7d0f3..c6b1262a46 100644 --- a/packages/engine/src/__tests__/executor-ephemeral-disabled-dispatch-gate.test.ts +++ b/packages/engine/src/__tests__/executor-ephemeral-disabled-dispatch-gate.test.ts @@ -113,11 +113,17 @@ describe("executor routes workflow stages through durable principals", () => { expect(agentStore.listAgents).toHaveBeenCalledWith({ includeEphemeral: true }); expect(store.upsertWorkflowWorkItem).toHaveBeenCalled(); + /* + FNXC:WorkflowAgentRouting 2026-08-11-09:12: + `maxProjectSessions` is gone from this call by design — workflow principals have no execution cap, + so admission passes no limit at all (see `WorkflowAgentCapacity.acquire`). Asserted as an ABSENCE, + not merely dropped from the shape, so silently reintroducing the cap fails here. + */ expect(acquire).toHaveBeenCalledWith(expect.objectContaining({ projectId: "project-fn-8821", agent: expect.objectContaining({ id: "workflow-executor" }), - maxProjectSessions: 2, })); + expect(acquire.mock.calls[0]?.[0]).not.toHaveProperty("maxProjectSessions"); expect(release).toHaveBeenCalledOnce(); expect((TaskExecutor as unknown as { processWideGraphRouting: Set }).processWideGraphRouting).not.toContain(live.id); expect(store.upsertWorkflowWorkItem).toHaveBeenCalledWith(expect.objectContaining({ diff --git a/packages/engine/src/__tests__/self-healing-stranded-continuation-reclaim.test.ts b/packages/engine/src/__tests__/self-healing-stranded-continuation-reclaim.test.ts new file mode 100644 index 0000000000..9330d751e5 --- /dev/null +++ b/packages/engine/src/__tests__/self-healing-stranded-continuation-reclaim.test.ts @@ -0,0 +1,194 @@ +import { describe, expect, it, vi } from "vitest"; +import type { Settings, Task, TaskStore, WorkflowWorkItem } from "@fusion/core"; + +const { recordRunAuditEventMock, resolveTaskLifecycleColumnsMock } = vi.hoisted(() => ({ + recordRunAuditEventMock: vi.fn(async () => undefined), + resolveTaskLifecycleColumnsMock: vi.fn(), +})); +vi.mock("@fusion/core", async (importOriginal) => ({ + ...(await importOriginal()), + resolveTaskLifecycleColumns: resolveTaskLifecycleColumnsMock, +})); +vi.mock("../util/run-audit.js", async (importOriginal) => ({ + ...(await importOriginal()), + createRunAuditor: vi.fn(() => ({ database: recordRunAuditEventMock })), +})); + +import { SelfHealingManager } from "../self-healing.js"; +import { evaluateStrandedContinuationReclaim } from "../workflows/stranded-continuation-reclaim.js"; + +/* +FNXC:StrandedContinuationReclaim 2026-08-11-09:12: +The observed incident: the scheduler's due-poll takes only `runnable`/`retrying`, so a continuation that +stops in `running` or `held` is never re-examined by anything. Nine live cards on the Fusion board were +stranded that way — seven `running` behind leases whose process exited ~9h earlier (NULL `leaseExpiresAt`, +so they never aged out) and two `held` with a NULL `blockedReason` for 46h (the claim predicate only +re-takes `workflow-principal-%`/`workflow-named-principal-%`/`workflow-role-pool-%` holds). A further 33 +rows in active states belonged to tasks that were archived AND soft-deleted, the oldest a month prior: +the FK cascade only fires on a hard delete. + +Surface enumeration — the invariant is asserted across every state the wedge can wear, not just the +reported one: `running` with a NULL lease, `running` with an EXPIRED lease, `held` with a NULL reason, +`held` with a principal-routing reason, a retired task's row in each active state, and the negative +cases (live session, unexpired lease, operator pause, manual-hold kind, engine pause, too-fresh). +*/ + +const stale = (ms: number) => new Date(Date.now() - ms).toISOString(); +const GRACE_EXCEEDED = 20 * 60_000; + +function item(overrides: Partial = {}): WorkflowWorkItem { + return { + id: "wi-1", runId: "run-1", taskId: "FN-8932", nodeId: "plan-review", kind: "task", + state: "running", attempt: 0, retryAfter: null, + leaseOwner: "executor:FN-8932", leaseExpiresAt: null, lastError: null, blockedReason: null, + createdAt: stale(GRACE_EXCEEDED), updatedAt: stale(GRACE_EXCEEDED), + ...overrides, + } as unknown as WorkflowWorkItem; +} + +function harness( + items: WorkflowWorkItem[] = [item()], + taskOverrides: Partial = {}, + settings: Partial = {}, +) { + const task = { + id: "FN-8932", title: "Memory layer 4a", description: "", column: "todo", + dependencies: [], steps: [], currentStep: 0, log: [], + createdAt: stale(GRACE_EXCEEDED), updatedAt: stale(GRACE_EXCEEDED), + ...taskOverrides, + } as unknown as Task; + const transitions: Array<{ id: string; state: string; patch: Record }> = []; + const logged: string[] = []; + const store = { + getSettings: vi.fn(async () => ({ globalPause: false, enginePaused: false, ...settings } as Settings)), + listDueWorkflowWorkItems: vi.fn(async () => items), + getTask: vi.fn(async (id: string) => (id === task.id ? task : undefined)), + transitionWorkflowWorkItem: vi.fn(async (id: string, state: string, patch: Record) => { + transitions.push({ id, state, patch }); + return { ...items.find((entry) => entry.id === id)!, state }; + }), + logEntry: vi.fn(async (_id: string, message: string) => { logged.push(message); }), + getRootDir: vi.fn(() => "/repo"), + getTasksDir: vi.fn(() => ""), + } as unknown as TaskStore; + resolveTaskLifecycleColumnsMock.mockResolvedValue({ complete: "done", archived: "archived" }); + return { task, store, transitions, logged, manager: new SelfHealingManager(store, { rootDir: "/repo" }) }; +} + +describe("reconcileStrandedWorkflowContinuations", () => { + it("re-queues a running continuation whose lease has no owner left", async () => { + recordRunAuditEventMock.mockClear(); + const { manager, transitions, logged } = harness(); + + await expect(manager.reconcileStrandedWorkflowContinuations()).resolves.toBe(1); + + expect(transitions).toEqual([{ + id: "wi-1", + state: "runnable", + // Clearing BOTH is what makes the row claimable: a stale leaseOwner reads as owned, and a + // surviving blockedReason keeps the claim predicate out. + patch: expect.objectContaining({ leaseOwner: null, leaseExpiresAt: null, blockedReason: null, expectedState: "running" }), + }]); + expect(logged.join(" ")).toContain("re-queued"); + expect(recordRunAuditEventMock).toHaveBeenCalledWith(expect.objectContaining({ + type: "workflowWorkItem:reconcile-stranded-requeued", + metadata: expect.objectContaining({ taskId: "FN-8932", priorState: "running", reason: "dead-lease" }), + })); + }); + + /* Every stranded shape observed in the incident, plus the expired-lease variant. */ + it("re-queues each stranded shape a dispatcher would never re-poll", async () => { + for (const [label, overrides] of [ + ["running, null lease expiry", { state: "running", leaseExpiresAt: null }], + ["running, expired lease", { state: "running", leaseExpiresAt: stale(GRACE_EXCEEDED) }], + ["held, null blocked reason", { state: "held", leaseOwner: null, blockedReason: null }], + ["held, principal routing reason", { state: "held", leaseOwner: null, blockedReason: "workflow-principal-role-pool-exhausted:executor" }], + ] as Array<[string, Partial]>) { + const { manager, transitions } = harness([item(overrides)]); + await expect(manager.reconcileStrandedWorkflowContinuations(), label).resolves.toBe(1); + expect(transitions[0]?.state, label).toBe("runnable"); + } + }); + + it("retires rows whose task can never run them again", async () => { + for (const taskOverrides of [ + { column: "archived", deletedAt: stale(1) }, + { column: "archived" }, + { column: "done" }, + ] as Array>) { + for (const state of ["running", "held", "runnable", "retrying"] as const) { + const { manager, transitions } = harness([item({ state })], taskOverrides); + await expect(manager.reconcileStrandedWorkflowContinuations()).resolves.toBe(1); + expect(transitions[0]?.state).toBe("cancelled"); + } + } + }); + + it("retires a row whose task no longer resolves at all", async () => { + const { manager, transitions } = harness([item({ taskId: "FN-DELETED" })]); + await expect(manager.reconcileStrandedWorkflowContinuations()).resolves.toBe(1); + expect(transitions[0]?.state).toBe("cancelled"); + }); + + /* + The false positive that would be worse than the bug: re-queueing a row a live session still owns + double-dispatches the task. Each guard is asserted separately so none can be dropped silently. + */ + it("never touches a row that is live, owned, paused, operator-held, or too fresh", async () => { + const cases: Array<[string, Parameters]> = [ + ["unexpired lease", [[item({ leaseExpiresAt: new Date(Date.now() + 600_000).toISOString() })], {}, {}]], + ["too fresh", [[item({ updatedAt: new Date().toISOString() })], {}, {}]], + ["manual-hold kind", [[item({ state: "held", kind: "manual-hold" })], {}, {}]], + ["operator paused", [[item()], { userPaused: true } as Partial, {}]], + ["task paused", [[item()], { paused: true } as Partial, {}]], + ["engine paused", [[item()], {}, { enginePaused: true }]], + ["global pause", [[item()], {}, { globalPause: true }]], + ["scheduler-owned runnable", [[item({ state: "runnable", leaseOwner: null })], {}, {}]], + ]; + for (const [label, args] of cases) { + const { manager, transitions } = harness(...args); + await expect(manager.reconcileStrandedWorkflowContinuations(), label).resolves.toBe(0); + expect(transitions, label).toEqual([]); + } + }); + + /* + The write is compare-and-set fenced on the state the scan observed. A dispatcher that legitimately + claimed the row in between must win, and the sweep must not count or log a repair it did not make. + */ + it("does not count a repair when the compare-and-set loses to a real claim", async () => { + const { manager, store, logged } = harness(); + (store.transitionWorkflowWorkItem as ReturnType).mockResolvedValue({ ...item(), state: "running" }); + await expect(manager.reconcileStrandedWorkflowContinuations()).resolves.toBe(0); + expect(logged).toEqual([]); + }); +}); + +describe("evaluateStrandedContinuationReclaim", () => { + const base = { + item: { state: "running", kind: "task", leaseExpiresAt: null, blockedReason: null }, + taskTerminal: false, taskMissing: false, taskPaused: false, + live: false, enginePaused: false, stalenessMs: GRACE_EXCEEDED, graceMs: 600_000, now: Date.now(), + } as Parameters[0]; + + it("orders engine pause above every other disposition", () => { + // Even a retirable row waits: a paused engine must make no autonomous writes at all. + expect(evaluateStrandedContinuationReclaim({ ...base, enginePaused: true, taskMissing: true })) + .toEqual({ action: "none", reason: "engine-paused" }); + }); + + it("retires a dead task's row regardless of pause or liveness flags", () => { + // Retirement sits ABOVE the pause/liveness guards on purpose — a deleted task has no operator + // decision left to respect, and that ordering is why the month-old residue accumulated. + expect(evaluateStrandedContinuationReclaim({ ...base, taskTerminal: true, taskPaused: true, live: true })) + .toEqual({ action: "retire", reason: "task-terminal" }); + }); + + it("treats a future lease expiry as proof of a live claim even past the grace window", () => { + expect(evaluateStrandedContinuationReclaim({ + ...base, + item: { ...base.item, leaseExpiresAt: new Date(base.now + 60_000).toISOString() }, + stalenessMs: 24 * 3_600_000, + })).toEqual({ action: "none", reason: "lease-active" }); + }); +}); diff --git a/packages/engine/src/__tests__/workflow-agent-capacity.test.ts b/packages/engine/src/__tests__/workflow-agent-capacity.test.ts index d42d78cb0f..ca6aa12f15 100644 --- a/packages/engine/src/__tests__/workflow-agent-capacity.test.ts +++ b/packages/engine/src/__tests__/workflow-agent-capacity.test.ts @@ -4,13 +4,35 @@ import { WorkflowAgentCapacity } from "../agents/workflow-agent-capacity.js"; const agent = (id: string, maxWorkflowSessions?: number) => ({ id, runtimeConfig: { maxWorkflowSessions } }) as any; describe("WorkflowAgentCapacity", () => { - it("keeps workflow and heartbeat limits independent while enforcing project then agent limits", async () => { + /* + FNXC:WorkflowAgentRouting 2026-08-11-09:12: + Replaces the former "enforces project then agent limits" case. Workflow principals have NO execution + cap: one Workflow Executor must be able to hold many concurrent sessions, because a per-principal + ceiling serialized the whole board (there is typically exactly one agent per role) and its refusal + became a durable `held` row no dispatcher re-polled. + */ + it("never refuses a workflow principal, however many sessions it already holds", async () => { const capacity = new WorkflowAgentCapacity(); + // `maxWorkflowSessions: 1` is deliberately set and must be ignored — it is the exact config that + // used to serialize a single-executor board. const constrained = agent("executor", 1); - expect(await capacity.acquire({ projectId: "project-a", agent: constrained, attemptId: "one", maxProjectSessions: 2 })).toMatchObject({ status: "acquired" }); - expect(await capacity.acquire({ projectId: "project-a", agent: constrained, attemptId: "two", maxProjectSessions: 2 })).toEqual({ status: "held", reason: "agent-capacity" }); - expect(await capacity.acquire({ projectId: "project-a", agent: agent("reviewer"), attemptId: "three", maxProjectSessions: 2 })).toMatchObject({ status: "acquired" }); - expect(await capacity.acquire({ projectId: "project-a", agent: agent("merger"), attemptId: "four", maxProjectSessions: 2 })).toEqual({ status: "held", reason: "project-capacity" }); + for (const attemptId of ["one", "two", "three", "four", "five"]) { + expect(await capacity.acquire({ projectId: "project-a", agent: constrained, attemptId })).toMatchObject({ status: "acquired" }); + } + expect(capacity.activeSessions("executor", "project-a")).toBe(5); + // A second role on the same project is likewise uncapped. + expect(await capacity.acquire({ projectId: "project-a", agent: agent("reviewer"), attemptId: "six" })).toMatchObject({ status: "acquired" }); + }); + + it("passes no session limits to the durable store, so a cap cannot be reintroduced by the caller", async () => { + const calls: Array> = []; + const capacity = new WorkflowAgentCapacity({ + acquireWorkflowSessionCapacity: async (input) => { calls.push(input); return "acquired"; }, + releaseWorkflowSessionCapacity: async () => undefined, + }); + await capacity.acquire({ projectId: "project-a", agent: agent("executor", 1), attemptId: "uncapped" }); + expect(calls[0]).not.toHaveProperty("maxProjectSessions"); + expect(calls[0]).not.toHaveProperty("maxAgentSessions"); }); it("isolates matching agent and attempt IDs across projects", async () => { @@ -25,7 +47,7 @@ describe("WorkflowAgentCapacity", () => { it("allows a fenced attempt to reacquire and releases exactly once", async () => { const capacity = new WorkflowAgentCapacity(); - const input = { projectId: "project-a", agent: agent("executor", 1), attemptId: "attempt", maxProjectSessions: 1 }; + const input = { projectId: "project-a", agent: agent("executor", 1), attemptId: "attempt" }; const first = await capacity.acquire(input); expect(await capacity.acquire(input)).toEqual(first); expect(capacity.activeSessions("executor")).toBe(1); diff --git a/packages/engine/src/agents/workflow-agent-capacity.ts b/packages/engine/src/agents/workflow-agent-capacity.ts index 7de73f891b..97e944cd6d 100644 --- a/packages/engine/src/agents/workflow-agent-capacity.ts +++ b/packages/engine/src/agents/workflow-agent-capacity.ts @@ -42,33 +42,48 @@ export class WorkflowAgentCapacity { return `${projectId}\u0000${attemptId}`; } + /** + * FNXC:WorkflowAgentRouting 2026-08-11-09:12: + * Workflow principals have NO execution cap. Admission always succeeds; the lease is taken purely as + * bookkeeping — it is what `activeSessions` counts and what the renewal timer keeps warm — never as a + * gate. + * + * Why the caps went away: the workflow roles (triage, executor, reviewer, merger) stand in for STAGES, + * not for workers, and there are typically one of each. Capping their concurrent sessions therefore + * capped the whole board — a project with a single Workflow Executor serialized every implementation + * task behind one session regardless of what `maxConcurrent`/`maxWorktrees` allowed. The refusal was + * also invisible: it surfaced as a durable `held` row reading + * `workflow-principal-agent-capacity:executor`, and until the FN reclaim sweep shipped alongside this + * change, nothing re-polled a `held` row at all. Task parallelism is already bounded where it belongs + * (worktree + concurrency admission on the task itself); a second bound at the principal was pure + * serialization with no safety value. + * + * `maxProjectSessions` is deliberately REMOVED from the input rather than defaulted to undefined, so a + * caller cannot silently reintroduce the cap without first deleting this contract. + * `runtimeConfig.maxWorkflowSessions` is likewise no longer consulted, here or in + * `routeWorkflowPrincipal`'s availability test. + */ public async acquire(input: { projectId: string; agent: Pick; attemptId: string; - maxProjectSessions?: number; }): Promise { const attemptKey = this.attemptKey(input.projectId, input.attemptId); const existing = this.leases.get(attemptKey); if (existing) return { status: "acquired", lease: existing }; - const agentLimit = input.agent.runtimeConfig?.maxWorkflowSessions; - const maxAgentSessions = typeof agentLimit === "number" && Number.isFinite(agentLimit) - ? agentLimit - : undefined; if (this.leaseStore) { + /* + FNXC:WorkflowAgentRouting 2026-08-11-09:12: + No limits are passed, so the durable store records the lease and returns "acquired". The store + KEEPS its limit parameters: they remain the correct cross-process gate for a caller that genuinely + needs one, and the reclaim of expired rows there is still what frees leases after a crash. + */ const outcome = await this.leaseStore.acquireWorkflowSessionCapacity({ agentId: input.agent.id, attemptId: input.attemptId, - maxProjectSessions: input.maxProjectSessions, - maxAgentSessions, leaseDurationMs: WorkflowAgentCapacity.LEASE_DURATION_MS, }); if (outcome !== "acquired") return { status: "held", reason: outcome }; - } else { - // Unit-test/local fallback retains deterministic semantics without pretending to coordinate processes. - const local = [...this.leases.values()].filter((lease) => lease.projectId === input.projectId); - if (input.maxProjectSessions !== undefined && local.length >= input.maxProjectSessions) return { status: "held", reason: "project-capacity" }; - if (maxAgentSessions !== undefined && local.filter((lease) => lease.agentId === input.agent.id).length >= maxAgentSessions) return { status: "held", reason: "agent-capacity" }; } const lease = { projectId: input.projectId, agentId: input.agent.id, attemptId: input.attemptId }; this.leases.set(attemptKey, lease); diff --git a/packages/engine/src/agents/workflow-agent-router.ts b/packages/engine/src/agents/workflow-agent-router.ts index 33ff0d998d..34114c4251 100644 --- a/packages/engine/src/agents/workflow-agent-router.ts +++ b/packages/engine/src/agents/workflow-agent-router.ts @@ -154,13 +154,13 @@ export function validateFencedWorkflowPrincipal(input: { })) { return staleFence; } - return available(agent, input.activeSessions ?? new Map()) + return available(agent) ? { status: "routed", route: { agent, role: input.role, authority: input.authority } } : { status: "held", role: input.role, reason: "named-principal-unavailable" }; } /** A named principal is never silently replaced once a precedence branch names it. */ -function available(agent: Agent | undefined, activeSessions: ReadonlyMap): agent is Agent { +function available(agent: Agent | undefined): agent is Agent { /* * FNXC:WorkflowAgentRouting 2026-08-07-03:38: * Workflow-stage routing may only select durable operator-visible principals. @@ -172,8 +172,16 @@ function available(agent: Agent | undefined, activeSessions: ReadonlyMap !input.excludedPoolAgentIds?.has(agent.id) - && hasWorkflowRoleCapability(agent, role) && available(agent, activeSessions)) + && hasWorkflowRoleCapability(agent, role) && available(agent)) .sort((left, right) => (activeSessions.get(left.id) ?? 0) - (activeSessions.get(right.id) ?? 0) || left.createdAt.localeCompare(right.createdAt) || left.id.localeCompare(right.id)); const agent = pool[0]; diff --git a/packages/engine/src/executor/workflow-principal-before-node.ts b/packages/engine/src/executor/workflow-principal-before-node.ts index 3dd80a2ebf..c36b79292f 100644 --- a/packages/engine/src/executor/workflow-principal-before-node.ts +++ b/packages/engine/src/executor/workflow-principal-before-node.ts @@ -123,8 +123,6 @@ let routed = hasFencedPrincipal agents, activeSessions, }); -/** Cleared when a stale fence is discarded, so the pool-capacity retry below is no longer fenced off. */ -let fenceStillGoverns = Boolean(hasFencedPrincipal); /* * FNXC:WorkflowAgentRouting 2026-08-10-07:50: * A fence that no longer describes reality is not a wait — the principal is gone, lost the role, or had its @@ -144,8 +142,6 @@ if (routed.status === "held" && routed.staleFence) { agents, activeSessions, }); - // The fence is discarded, so a fresh role-pool route may take the cross-engine capacity retry below. - fenceStillGoverns = false; } if (routed.status === "unclassified") return undefined; /* @@ -268,49 +264,24 @@ if (routed.status === "held") { */ const attemptId = `${deps.resolvedRunId ?? `${nodeTask.id}:workflow`}:${nodeInstanceId}`; /* - * FNXC:WorkflowAgentRouting 2026-08-07-05:29: - * Workflow-stage admission consumes the project workflow budget, while - * an agent's heartbeat retains its separate maxConcurrentRuns budget. - * Passing the project limit here closes the direct-graph path, which - * otherwise enforced only optional per-agent limits. + * FNXC:WorkflowAgentRouting 2026-08-11-09:12: + * Workflow-stage admission no longer consumes a project workflow budget — workflow principals have NO + * execution cap (see `WorkflowAgentCapacity.acquire` for why: with one durable agent per role, capping + * principal sessions capped the entire board, and the refusal became a durable `held` row nothing + * re-polled). The lease is still taken, because it is what `activeSessions` counts and what the renewal + * timer keeps warm; it just cannot refuse. An agent's heartbeat keeps its separate `maxConcurrentRuns` + * budget, which was always a different thing. + * + * The `agent-capacity` re-route loop is DELETED with the cap that motivated it: it existed only to pick + * a different pool member after a durable refusal, and there is no longer a refusal to react to. The + * `held` branch below survives as a fail-closed guard for a store-level refusal, and still records the + * durable held item so the state is inspectable rather than terminalized. */ -let capacity = await deps.workflowAgentCapacity.acquire({ +const capacity = await deps.workflowAgentCapacity.acquire({ projectId: deps.options.agentStore.workflowProjectId ?? deps.store.getRootDir(), agent: routed.route.agent, attemptId, - maxProjectSessions: deps.settings.maxConcurrent, }); -/* - * FNXC:WorkflowAgentRouting 2026-08-07-07:32: - * A role-pool snapshot is process-local, while admission is durable - * across engines. If another engine filled the selected agent between - * selection and the atomic acquire, try the next eligible pool member. - * Fenced and named principals never take this fallback. - */ -if (capacity.status === "held" && capacity.reason === "agent-capacity" - && routed.route.authority === "role-pool" && !fenceStillGoverns) { - const excludedPoolAgentIds = new Set(); - while (capacity.status === "held" && capacity.reason === "agent-capacity" - && routed.route.authority === "role-pool") { - excludedPoolAgentIds.add(routed.route.agent.id); - const retryRoute = routeWorkflowPrincipal({ - task: nodeTask, - ir: deps.columnAgentIr, - node, - agents, - activeSessions, - excludedPoolAgentIds, - }); - if (retryRoute.status !== "routed" || retryRoute.route.authority !== "role-pool") break; - routed = retryRoute; - capacity = await deps.workflowAgentCapacity.acquire({ - projectId: deps.options.agentStore.workflowProjectId ?? deps.store.getRootDir(), - agent: routed.route.agent, - attemptId, - maxProjectSessions: deps.settings.maxConcurrent, - }); - } -} if (capacity.status === "held") { const reason = `workflow-principal-${capacity.reason}:${routed.route.role}`; await holdDirectPrincipalWorkItem(reason, routed.route.agent.id, routed.route.authority); diff --git a/packages/engine/src/self-healing.ts b/packages/engine/src/self-healing.ts index 4d8c71ca8c..61a9497e92 100644 --- a/packages/engine/src/self-healing.ts +++ b/packages/engine/src/self-healing.ts @@ -83,6 +83,7 @@ import { isTaskStillInPlanningStage } from "./execution/replan-target.js"; import { classifyPersistedPlanHandoff, LEGACY_NULL_PLAN_HANDOFF_STALE_MS } from "./planning-handoff-recovery.js"; import { getPromptPath } from "./execution/spec-staleness.js"; import { evaluateStrandedHoldContinuation, seedPreReleasePlanReviewContinuation } from "./plan-review-continuation.js"; +import { evaluateStrandedContinuationReclaim, RECLAIM_RETIRED_STATE } from "./workflows/stranded-continuation-reclaim.js"; /* FNXC:Workspace 2026-06-22-14:10 (Phase D review G — cycle dissolved): `isRepoLanded` is the CANONICAL per-repo landed predicate (Phase C, exported A6). It now lives in @@ -1744,6 +1745,14 @@ export class SelfHealingManager extends SelfHealingGitEvidence { // FNXC:PrincipalHeldPlanning 2026-08-10-08:20: a planning hold from principal routing has no other // retry owner, so it must be re-queued before the steps below classify the card as simply idle. { name: "reconcile-principal-held-planning", fn: () => this.reconcilePrincipalHeldPlanningContinuations().then(() => undefined) }, + /* + FNXC:StrandedContinuationReclaim 2026-08-11-09:12: + Runs AFTER the two narrow continuation sweeps above and before any step that classifies a card as + idle. Startup is the highest-yield moment for it: a killed process is exactly what leaves a + `running` row with a lease that now has no owner, and until this sweep existed those rows survived + every subsequent restart untouched. + */ + { name: "reconcile-stranded-workflow-continuations", fn: () => this.reconcileStrandedWorkflowContinuations().then(() => undefined) }, { name: "no-progress-no-task-done", fn: () => this.recoverNoProgressNoTaskDoneFailures().then(() => undefined) }, { name: "completed-tasks", fn: () => this.recoverCompletedTasks().then(() => undefined) }, { name: "recover-stranded-completed-todo", fn: () => this.recoverStrandedCompletedTodoTasks().then(() => undefined) }, @@ -2858,6 +2867,9 @@ export class SelfHealingManager extends SelfHealingGitEvidence { // no benefit. { name: "reconcile-stranded-hold-continuations", fn: () => this.reconcileStrandedHoldContinuations() }, { name: "reconcile-principal-held-planning", fn: () => this.reconcilePrincipalHeldPlanningContinuations() }, + // FNXC:StrandedContinuationReclaim 2026-08-11-09:12: steady-state half of the startup sweep — + // a session can die mid-run without a restart, and the grace window keeps live work untouched. + { name: "reconcile-stranded-workflow-continuations", fn: () => this.reconcileStrandedWorkflowContinuations() }, { name: "recover-mergeable-review", fn: () => this.recoverMergeableReviewTasks() }, // FNXC:Workspace 2026-06-22-09:30 (Phase D U1) — workspace-mode reconcilers. { name: "reconcile-workspace-partial-lands", fn: () => this.reconcileWorkspacePartialLands() }, @@ -7704,6 +7716,135 @@ export class SelfHealingManager extends SelfHealingGitEvidence { } } + /** + * FNXC:StrandedContinuationReclaim 2026-08-11-09:12: + * The general reclaim for continuations no dispatcher will ever look at again. See + * `workflows/stranded-continuation-reclaim.ts` for the full mechanism; in short, the due-poll selects + * only `runnable`/`retrying`, so a row that stops in `running` (dead lease, NULL expiry) or `held` + * (reason the claim predicate cannot re-take) is silently terminal, and a soft-deleted task's rows are + * never cascaded away. + * + * This is the SUPERSET sweep that `reconcilePrincipalHeldPlanningContinuations` is the narrow special + * case of: that one restores a planning SIGNAL (`needs-replan`) so triage re-routes a held triage-role + * plan node, which is the correct repair for its shape and is deliberately left alone here. This sweep + * repairs the ROW, moving it back to `runnable` so the ordinary drain claims it on the next tick. + * + * Scan shape: `listDueWorkflowWorkItems` already filters `leaseExpiresAt IS NULL OR <= now`, so the + * one query returns exactly the rows that could be abandoned — no per-task fan-out, and no unbounded + * table walk. Every write is compare-and-set fenced on the state the scan observed, so a row a real + * dispatcher claimed between scan and write is left untouched rather than reset under a live session. + * + * @returns Count of rows actually re-queued or retired. + */ + async reconcileStrandedWorkflowContinuations(): Promise { + try { + if (typeof this.store.listDueWorkflowWorkItems !== "function") return 0; + const settings = await this.store.getSettings(); + const enginePaused = settings.globalPause === true || settings.enginePaused === true; + if (enginePaused) return 0; + /* + FNXC:StrandedContinuationReclaim 2026-08-11-09:12: + Ten minutes matches `WorkflowAgentCapacity.LEASE_DURATION_MS`, the longest a healthy claim can go + without renewing its durable capacity row. Anything younger may be a live session mid-renewal. + */ + const graceMs = 10 * 60_000; + const now = Date.now(); + const nowIso = new Date(now).toISOString(); + const live = (taskId: string) => activeSessionRegistry.pathsForTask(taskId).some((path) => activeSessionRegistry.isPathActive(path)) + || executingTaskLock.has(taskId) + || this.options.isTaskActive?.(taskId) === true; + const due = await this.store.listDueWorkflowWorkItems({ + now: nowIso, + states: [...ACTIVE_WORKFLOW_WORK_ITEM_STATES], + }); + let repaired = 0; + for (const item of due) { + try { + /* + FNXC:StrandedContinuationReclaim 2026-08-11-09:12: + `getTask` must see soft-deleted and archived rows, because those are precisely the tasks whose + continuations need retiring. A reader that hides them reports `task-missing`, which retires the + row too — the same disposition, so the sweep stays correct either way. + */ + const task = await this.store.getTask(item.taskId); + const terminalColumns = await resolveTaskLifecycleColumns(this.store, item.taskId).catch(() => undefined); + const doneColumn = terminalColumns?.complete ?? "done"; + const archivedColumn = terminalColumns?.archived ?? "archived"; + const verdict = evaluateStrandedContinuationReclaim({ + item, + taskMissing: !task, + taskTerminal: !!task && ( + task.deletedAt != null + || task.column === archivedColumn + || task.column === doneColumn + || task.column === "archived" + || task.column === "done" + ), + taskPaused: task?.userPaused === true || task?.paused === true, + live: live(item.taskId), + enginePaused, + stalenessMs: Math.max(0, now - new Date(item.updatedAt).getTime()), + graceMs, + now, + }); + if (verdict.action === "none") continue; + const target = verdict.action === "retire" ? RECLAIM_RETIRED_STATE : "runnable"; + const written = await this.store.transitionWorkflowWorkItem(item.id, target, { + /* + FNXC:StrandedContinuationReclaim 2026-08-11-09:12: + Clearing the lease and the blocked reason is what makes the row claimable again — leaving + either behind reproduces the exact wedge this sweep exists to end (the claim predicate reads + `blockedReason`, and a stale `leaseOwner` makes the row look owned to every human reading it). + `lastError` is preserved: it is the only surviving evidence of why the row stopped. + */ + leaseOwner: null, + leaseExpiresAt: null, + blockedReason: null, + expectedState: item.state, + }); + // CAS lost: another writer moved the row between the scan and this write. Leave it to them. + if (written.state !== target) continue; + repaired += 1; + if (verdict.action === "requeue") { + await this.store.logEntry( + item.taskId, + `[recovery] workflow continuation re-queued — ${item.nodeId} was stranded in '${item.state}' (${verdict.reason})`, + ).catch(() => undefined); + } + await createRunAuditor(this.store, { + runId: generateSyntheticRunId("reconcile-stranded-continuation", item.taskId), + agentId: "self-healing", + taskId: item.taskId, + taskLineageId: task?.lineageId, + phase: "reconcile-stranded-continuation", + }).database({ + type: (verdict.action === "retire" + ? "workflowWorkItem:reconcile-stranded-retired" + : "workflowWorkItem:reconcile-stranded-requeued") as DatabaseMutationType, + target: item.id, + /* Ids/counts/outcomes only — never `lastError` prose or node config. */ + metadata: { + taskId: item.taskId, + workItemId: item.id, + nodeId: item.nodeId, + kind: item.kind, + priorState: item.state, + reason: verdict.reason, + stalenessMs: Math.max(0, now - new Date(item.updatedAt).getTime()), + }, + }); + } catch (error) { + log.warn(`reconcileStrandedWorkflowContinuations: failed for ${item.id}: ${error instanceof Error ? error.message : String(error)}`); + } + } + if (repaired > 0) log.log(`Reclaimed ${repaired} stranded workflow continuation(s)`); + return repaired; + } catch (error) { + log.warn(`reconcileStrandedWorkflowContinuations failed: ${error instanceof Error ? error.message : String(error)}`); + return 0; + } + } + /* FNXC:OrphanedPendingSteps 2026-07-22-16:20 (FN-8492 incident): Consumer of `resolveOrphanedPendingStepResults` — the U9b helper shipped with NO caller diff --git a/packages/engine/src/triage.ts b/packages/engine/src/triage.ts index d43de6a1b9..b7a98451bd 100644 --- a/packages/engine/src/triage.ts +++ b/packages/engine/src/triage.ts @@ -2631,7 +2631,9 @@ export class TriageProcessor { if (!Array.isArray(agents)) { assignedAgent = assignedAgent ?? null; } else { - let routed = routeWorkflowPrincipal({ + // FNXC:WorkflowAgentRouting 2026-08-11-09:12: `const` since the capacity re-route loop that + // used to reassign this is gone with the principal execution cap. + const routed = routeWorkflowPrincipal({ task: currentTask, ir: planningIr, node: planningNode, @@ -2679,43 +2681,21 @@ export class TriageProcessor { triageRunContext.agentId = assignedAgent.id; workflowCapacityAttemptId = `${triageRunContext.runId}:${planningNode.id}`; workflowCapacityProjectId = this.options.agentStore.workflowProjectId ?? this.rootDir; - let capacity = await this.workflowAgentCapacity.acquire({ + /* + FNXC:WorkflowAgentRouting 2026-08-11-09:12: + Capacity admission no longer refuses (see `WorkflowAgentCapacity.acquire`): workflow + principals have no execution cap, so the lease is bookkeeping and the acquire always + succeeds. The `agent-capacity` retry loop that used to re-route to another pool member on + refusal is DELETED with the cap it existed to work around — with no refusal there is no + contender to exclude. The `held` branch is kept as a fail-closed guard for a store-level + refusal (a future caller that does pass limits, or a durable-store error path); it parks the + card on `needs-replan`, which triage rediscovers, rather than dropping it. + */ + const capacity = await this.workflowAgentCapacity.acquire({ projectId: workflowCapacityProjectId, agent: assignedAgent, attemptId: workflowCapacityAttemptId, - maxProjectSessions: settings.maxConcurrent, }); - /* - FNXC:WorkflowAgentRouting 2026-08-07-07:32: - Role-pool selection is optimistic across engine processes. Retry a - different pool member only after durable agent-capacity rejects this - contender; task owners and other named principals remain fail-closed. - */ - if (capacity.status === "held" && capacity.reason === "agent-capacity" - && routed.route.authority === "role-pool") { - const excludedPoolAgentIds = new Set(); - while (capacity.status === "held" && capacity.reason === "agent-capacity" - && routed.route.authority === "role-pool") { - excludedPoolAgentIds.add(routed.route.agent.id); - const retryRoute = routeWorkflowPrincipal({ - task: currentTask, - ir: planningIr, - node: planningNode, - agents, - excludedPoolAgentIds, - }); - if (retryRoute.status !== "routed" || retryRoute.route.authority !== "role-pool") break; - routed = retryRoute; - assignedAgent = routed.route.agent; - triageRunContext.agentId = assignedAgent.id; - capacity = await this.workflowAgentCapacity.acquire({ - projectId: workflowCapacityProjectId, - agent: assignedAgent, - attemptId: workflowCapacityAttemptId, - maxProjectSessions: settings.maxConcurrent, - }); - } - } if (capacity.status === "held") { await this.store.logEntry(task.id, `Planning held: workflow-principal-${capacity.reason}:triage`); await this.updatePlanningStateIfStillCurrent(task, { status: "needs-replan" }); diff --git a/packages/engine/src/workflows/stranded-continuation-reclaim.ts b/packages/engine/src/workflows/stranded-continuation-reclaim.ts new file mode 100644 index 0000000000..968120b5f1 --- /dev/null +++ b/packages/engine/src/workflows/stranded-continuation-reclaim.ts @@ -0,0 +1,116 @@ +import { ACTIVE_WORKFLOW_WORK_ITEM_STATES, type WorkflowWorkItem } from "@fusion/core"; + +/* +FNXC:StrandedContinuationReclaim 2026-08-11-09:12: +The scheduler's due-poll (`in-process-runtime.ts` -> `drainDuePlanningContinuations`) selects ONLY +`runnable`/`retrying`. A continuation that stops in `running` or `held` is therefore never looked at +again by any dispatcher, and the two ways it can stop there are both ordinary: + + 1. `running` with a dead session. `acquireWorkflowWorkItemLease` is the only writer that sets + `leaseExpiresAt`; the upsert/transition paths that install a fence leave it NULL. So a row claimed + through those paths whose process then dies keeps `state: "running"` with a `leaseOwner` and NO + expiry — permanently indistinguishable from live work to every reader that filters on state alone. + 2. `held` with a blocked reason nothing retries. `acquireWorkflowWorkItemLease` can only re-take a + `held` row whose `blockedReason` matches `workflow-principal-%`, `workflow-named-principal-%`, or + `workflow-role-pool-%`; a NULL or non-matching reason never becomes claimable again. + +Observed on the Fusion board on 2026-08-11: seven cards (FN-8932/8950/8953/8954/8956/8957/8958) sat in +`running` behind leases from a process that exited ~9h earlier, and FN-8901/FN-8902 sat `held` with a +NULL `blockedReason` for 46h. None produced a single run-audit row while stranded — the same silent +shape as the FN-8923 incident, whose sweep only covers `held` + `workflowRole: "triage"` + +`blockedReason` starting `workflow-principal-`, and so caught none of these nine. + +A third case is pure residue: 33 rows in active states belonged to tasks that were archived AND +soft-deleted, the oldest from 2026-07-13. The FK is `ON DELETE CASCADE`, which only fires on a HARD +delete, so soft-deleting a task strands its continuations forever. They cannot run (their task is gone +from every scan) but they are counted by every `ACTIVE_WORKFLOW_WORK_ITEM_STATES` filter in the engine. + +This module is the pure decision shared by the sweep and its tests, so "what is reclaimable" cannot +drift from "what the sweep reclaims" — the drift that let the FN-8923 sweep ship covering one ninth of +the problem it was written for. +*/ + +/** Terminal disposition for a row whose task can never run it again. */ +export const RECLAIM_RETIRED_STATE = "cancelled" as const; + +export type StrandedContinuationAction = "requeue" | "retire" | "none"; + +export type StrandedContinuationReason = + | "engine-paused" + | "task-terminal" + | "task-missing" + | "operator-paused" + | "manual-hold" + | "live-session" + | "too-fresh" + | "lease-active" + | "scheduler-owned" + | "dead-lease" + | "unclaimable-hold"; + +export interface StrandedContinuationVerdict { + action: StrandedContinuationAction; + reason: StrandedContinuationReason; +} + +/** + * FNXC:StrandedContinuationReclaim 2026-08-11-09:12: + * Decide what to do with ONE continuation row. Ordering is deliberate and load-bearing: + * + * - `enginePaused` wins over everything. A paused engine must not silently re-queue work an operator + * stopped; the row is still stranded when the engine resumes and the next sweep sees it. + * - Retirement is tested BEFORE the pause and liveness gates. A soft-deleted or archived task has no + * operator decision left to respect and no session that could be live, and its row is exactly the + * residue that accumulates for a month when this check sits behind those guards. + * - `manual-hold` is an operator-owned kind (`WORKFLOW_WORK_ITEM_KINDS`), not an accident. Its whole + * purpose is to stop until a human acts, so automatic reclaim must never touch it. + * - The liveness proof is the caller's (the canonical `activeSessionRegistry` / `executingTaskLock` / + * `isTaskActive` triple). A live session with a NULL-expiry lease is the exact false positive that + * would double-dispatch a running task, so `live` outranks the dead-lease test below it. + * - `lease-active` still defers to a real unexpired lease even past the grace window: an expiry in the + * future is affirmative proof of a live claim, which staleness alone never is. + * + * @param input.item The continuation row under consideration. + * @param input.taskTerminal The owning task is deleted, archived, or otherwise past running this row. + * @param input.taskMissing No task row resolved for `item.taskId` at all. + * @param input.live Caller-proven live execution for the owning task. + * @param input.stalenessMs Age of the row's last update. + * @param input.graceMs Minimum age before a row is considered abandoned. + */ +export function evaluateStrandedContinuationReclaim(input: { + item: Pick; + taskTerminal: boolean; + taskMissing: boolean; + taskPaused: boolean; + live: boolean; + enginePaused: boolean; + stalenessMs: number; + graceMs: number; + now: number; +}): StrandedContinuationVerdict { + if (input.enginePaused) return { action: "none", reason: "engine-paused" }; + if (input.taskMissing) return { action: "retire", reason: "task-missing" }; + if (input.taskTerminal) return { action: "retire", reason: "task-terminal" }; + if (!ACTIVE_WORKFLOW_WORK_ITEM_STATES.includes(input.item.state)) { + return { action: "none", reason: "scheduler-owned" }; + } + if (input.item.kind === "manual-hold") return { action: "none", reason: "manual-hold" }; + if (input.taskPaused) return { action: "none", reason: "operator-paused" }; + if (input.live) return { action: "none", reason: "live-session" }; + /* + FNXC:StrandedContinuationReclaim 2026-08-11-09:12: + `runnable`/`retrying` rows are the dispatcher's own queue. They are only reachable here through the + retirement branches above (a dead task's row), never for reclaim — re-writing a live queue entry + would reset its scheduling position for no gain. + */ + if (input.item.state === "runnable" || input.item.state === "retrying") { + return { action: "none", reason: "scheduler-owned" }; + } + if (input.stalenessMs < input.graceMs) return { action: "none", reason: "too-fresh" }; + if (input.item.state === "running") { + const expiresAt = input.item.leaseExpiresAt ? Date.parse(input.item.leaseExpiresAt) : Number.NaN; + if (Number.isFinite(expiresAt) && expiresAt > input.now) return { action: "none", reason: "lease-active" }; + return { action: "requeue", reason: "dead-lease" }; + } + return { action: "requeue", reason: "unclaimable-hold" }; +}