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 <noreply@anthropic.com>
This commit is contained in:
gsxdsm
@@ -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`.
|
||||
@@ -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<string> }).processWideGraphRouting).not.toContain(live.id);
|
||||
expect(store.upsertWorkflowWorkItem).toHaveBeenCalledWith(expect.objectContaining({
|
||||
|
||||
@@ -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<typeof import("@fusion/core")>()),
|
||||
resolveTaskLifecycleColumns: resolveTaskLifecycleColumnsMock,
|
||||
}));
|
||||
vi.mock("../util/run-audit.js", async (importOriginal) => ({
|
||||
...(await importOriginal<typeof import("../util/run-audit.js")>()),
|
||||
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> = {}): 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<Task> = {},
|
||||
settings: Partial<Settings> = {},
|
||||
) {
|
||||
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<string, unknown> }> = [];
|
||||
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<string, unknown>) => {
|
||||
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<WorkflowWorkItem>]>) {
|
||||
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<Partial<Task>>) {
|
||||
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<typeof harness>]> = [
|
||||
["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>, {}]],
|
||||
["task paused", [[item()], { paused: true } as Partial<Task>, {}]],
|
||||
["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<typeof vi.fn>).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<typeof evaluateStrandedContinuationReclaim>[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" });
|
||||
});
|
||||
});
|
||||
@@ -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<Record<string, unknown>> = [];
|
||||
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);
|
||||
|
||||
@@ -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<Agent, "id" | "runtimeConfig">;
|
||||
attemptId: string;
|
||||
maxProjectSessions?: number;
|
||||
}): Promise<WorkflowAgentCapacityResult> {
|
||||
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);
|
||||
|
||||
@@ -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<string, number>): 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<string,
|
||||
// (isWorkflowPrincipalEligible) so "what provisioning must produce" and "what routing accepts"
|
||||
// cannot drift apart again — that drift is what left every built-in owner unroutable.
|
||||
if (!agent || isEphemeralAgent(agent) || !isWorkflowPrincipalEligible(agent)) return false;
|
||||
const max = agent.runtimeConfig?.maxWorkflowSessions;
|
||||
return typeof max !== "number" || activeSessions.get(agent.id) === undefined || activeSessions.get(agent.id)! < max;
|
||||
/*
|
||||
FNXC:WorkflowAgentRouting 2026-08-11-09:12:
|
||||
Availability is now a pure ELIGIBILITY test — is this a durable principal that may run this role — with
|
||||
no session-count ceiling. `runtimeConfig.maxWorkflowSessions` used to gate here; a role agent at its
|
||||
ceiling reported `role-pool-exhausted`/`named-principal-unavailable` and the node became a durable
|
||||
`held` row that no dispatcher re-polled. With one agent per role that ceiling serialized the entire
|
||||
board. See `WorkflowAgentCapacity.acquire` for the full rationale — the two must stay consistent, since
|
||||
a cap in either place reproduces the same wedge on its own.
|
||||
*/
|
||||
return true;
|
||||
}
|
||||
|
||||
/**
|
||||
@@ -225,7 +233,7 @@ export function routeWorkflowPrincipal(input: {
|
||||
if (!agent || !hasWorkflowRoleCapability(agent, role, { allowEngineerAsExecutor: authority === "task-assignee" })) {
|
||||
return undefined;
|
||||
}
|
||||
return available(agent, activeSessions)
|
||||
return available(agent)
|
||||
? { status: "routed", route: { agent, role, authority } }
|
||||
: { status: "held", role, reason: "named-principal-unavailable" };
|
||||
};
|
||||
@@ -247,7 +255,7 @@ export function routeWorkflowPrincipal(input: {
|
||||
if (bound) return bound;
|
||||
const pool = input.agents
|
||||
.filter((agent) => !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];
|
||||
|
||||
@@ -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<string>();
|
||||
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);
|
||||
|
||||
@@ -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<number> {
|
||||
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
|
||||
|
||||
@@ -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<string>();
|
||||
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" });
|
||||
|
||||
116
packages/engine/src/workflows/stranded-continuation-reclaim.ts
Normal file
116
packages/engine/src/workflows/stranded-continuation-reclaim.ts
Normal file
@@ -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<WorkflowWorkItem, "state" | "kind" | "leaseExpiresAt" | "blockedReason">;
|
||||
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" };
|
||||
}
|
||||
Reference in New Issue
Block a user