From dd40691ca24fc56c75389c5dcd48c767062f46d5 Mon Sep 17 00:00:00 2001 From: gsxdsm Date: Fri, 7 Aug 2026 17:09:53 -0700 Subject: [PATCH] fix: make workflow continuation writes atomic, not conflict-recovery MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Code review of ef8828f14 found the continuation handover it introduced was a hand-rolled, non-atomic replacement for a primitive this repo already has, with six P1 defects — two of which recreated the very deadlock it was written to fix. The invariant: a task may hold ONE active kind="task" work item (idx_workflow_work_items_one_active_task_continuation), and that partial unique index is NOT what a plain upsert's ON CONFLICT targets. So a predecessor the run has already left makes the write RAISE. Every continuation write in the executor and triage now goes through replaceActiveTaskWorkflowContinuation, which retires non-matching active rows and installs the successor in ONE transaction under the task advisory lock: - Sibling foreach instances share the template nodeId and differ only by runId, so the old node-identity guard released nothing and instance #1 re-deadlocked. - Reacting to a FAILED write could not tell an index conflict from a transient database error, so it destroyed legitimate held continuations. - Read-then-write across separate transactions let a concurrent engine lose a live claim; the lock now serializes it. - A failed retry left the task with zero active rows and no error, because the hold then transitioned an already-terminal row and the throw was swallowed. - The same unguarded write existed on the executor's hold path and at both of triage's planning-continuation writes; a throw there degraded a recoverable availability hold into a terminal graph failure. Coverage moves from a fake store to the real index: the new PG suite proves the bare upsert raises and that replace handles a different node, a sibling foreach instance, a held predecessor, and re-entry, plus a drift guard tying the SQL predicate to ACTIVE_WORKFLOW_WORK_ITEM_STATES. The hand-rolled handover is tombstoned so it cannot return as a "conflict fix", and both new run-audit events are documented in the AGENTS.md inventory. Co-Authored-By: Claude Opus 5 --- ...fn-workflow-executor-agent-store-wiring.md | 2 +- AGENTS.md | 2 + .../workflow-continuation-slot.pg.test.ts | 179 ++++++++++ ...xecutor-continuation-slot-handover.test.ts | 166 --------- .../src/__tests__/legacy-tombstones.test.ts | 13 + packages/engine/src/executor.ts | 324 +++++++++--------- packages/engine/src/triage.ts | 43 ++- 7 files changed, 392 insertions(+), 337 deletions(-) create mode 100644 packages/core/src/__tests__/postgres/workflow-continuation-slot.pg.test.ts delete mode 100644 packages/engine/src/__tests__/executor-continuation-slot-handover.test.ts diff --git a/.changeset/fn-workflow-executor-agent-store-wiring.md b/.changeset/fn-workflow-executor-agent-store-wiring.md index a7d233d15e..cf672d0160 100644 --- a/.changeset/fn-workflow-executor-agent-store-wiring.md +++ b/.changeset/fn-workflow-executor-agent-store-wiring.md @@ -4,4 +4,4 @@ summary: Fix tasks stalling forever in progress with no session after the workflow role-agent rollout. category: fix -dev: Two deadlocks in FN-8764's role routing, both silent. (1) The in-process runtime never passed its AgentStore into `TaskExecutorOptions`, so routing failed closed at every role-classified node. (2) A resumed run keeps its continuation work item active until the interpreter returns, so the next node's principal-fence upsert hit `idx_workflow_work_items_one_active_task_continuation` (not its ON CONFLICT target) and raised; the run then re-suspended every dispatch until an operator bounced the card. The executor now supersedes an active work item for a node the run has already left and retries once. Adds the `task:workflow-run-suspended` and `task:workflow-continuation-superseded` run-audit events, and logs principal holds and fence-write errors instead of swallowing them. +dev: Two deadlocks in FN-8764's role routing, both silent. (1) The in-process runtime never passed its AgentStore into `TaskExecutorOptions`, so routing failed closed at every role-classified node. (2) Durable continuation writes used a bare `upsertWorkflowWorkItem`, whose ON CONFLICT target is not `idx_workflow_work_items_one_active_task_continuation`, so a predecessor the run had already left (the resumed continuation, or a sibling foreach instance sharing the template nodeId) made the write RAISE; the run then re-suspended every dispatch until an operator bounced the card. Every `kind:"task"` continuation write in the executor and triage now goes through the atomic `replaceActiveTaskWorkflowContinuation`. Adds the `task:workflow-run-suspended` run-audit event, logs principal holds and fence-write errors instead of swallowing them, pins the invariant against a real Postgres index, and ratchets the hand-rolled handover as a tombstone. diff --git a/AGENTS.md b/AGENTS.md index ee7cf2dde5..8920f8c7b5 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -310,6 +310,8 @@ Scoped exception (FN-5819): shared-branch-group members (`branchContext.assignme - FN-8305: durable symbol-lock operations emit `symbol-lock:acquired`, `symbol-lock:acquire-conflict`, `symbol-lock:renewed`, `symbol-lock:released`, `symbol-lock:reconcile-stale`, and deduplicated `symbol-lock:reconcile-stale-no-action`. Metadata is ids/counts/outcomes-only; normalized opaque symbol keys are permitted IDs, while raw symbol prose is not. - FN-8600: triage emits `task:plan-admission-throttled` when planning admission is withheld while eligible cards are waiting, recording the binding gate (`blockedBy`, now always `"running-agent cap"` — the cross-project semaphore that was the other value is deleted, and the four `semaphore*` fields went with it) plus `maxConcurrent`, `claimed`, `projectRoom`, `eligibleCount`, up to five `eligibleTaskIds`, `processingCount`, and up to five `processingTaskIds`. Metadata is ids/counts-only. Deduped on the gate signature INCLUDING the eligible task IDs, so a sustained stall collapses to one row while a new card's stall is never swallowed; the marker is set only after the write lands, so a failed write retries on the next poll. Purpose: before this event the binding gate existed only in a `planLog` line that is persisted nowhere, so "why did this card sit queued to plan?" was unanswerable after the fact. Reachable today by direct DB query only — the sole run-audit read route resolves through a durable agent's heartbeat run and this event uses a synthetic run id under `agentId:"triage"`. - FN-8592: startup and periodic self-healing emit `task:reconcile-stranded-hold-continuation` when an idle hold-column card with a real spec is re-seeded at its pre-release Plan Review, and deduped `task:reconcile-stranded-hold-continuation-no-action` for a candidate guard or race loss. Metadata stays ids/counts/outcomes-only (`taskId`, `column`, node/workflow identifiers, staleness or reason); healthy non-candidates are silent. The repair is insert-only and uses the shared per-task advisory transaction lock; global/engine pause and `autoMerge:false` defer to the operator. +- Workflow role routing: the executor emits `task:workflow-run-suspended` whenever a graph run ends in the `suspended` disposition, recording where and why the card is waiting (`nodeId`, fixed `reason` code, `fromColumn`/`toColumn`, and the resumed continuation's id/node/state). A suspend writes no task error by design, so without this row a card that re-suspends at the same node every dispatch leaves NO new state anywhere and is indistinguishable from a dead engine — that is how an unwired `agentStore` deadlocked every task unnoticed. Metadata is ids/outcomes-only; reason codes are a fixed enum, never prose. +- Workflow role routing: `replaceActiveTaskWorkflowContinuation` is the ONLY sanctioned way to write a `kind:"task"` continuation. `idx_workflow_work_items_one_active_task_continuation` permits one active row per task and is NOT the constraint a plain upsert's ON CONFLICT targets, so a bare upsert RAISES against a predecessor the run has already left (the resumed continuation, or a sibling foreach instance sharing the template `nodeId`). The primitive retires non-matching active rows and installs the successor in one transaction under the task advisory lock. Never re-add a recovery path that reacts to a failed write by terminalizing other rows: it cannot distinguish an index conflict from a transient database error, and it destroys legitimate holds. Ratcheted by `packages/engine/src/__tests__/legacy-tombstones.test.ts`; the invariant is pinned against a real index by `packages/core/src/__tests__/postgres/workflow-continuation-slot.pg.test.ts`. - FN-8492: the self-healing sweep `reconcile-orphaned-pending-step-results` (startup, right after legacy adoption, plus periodic maintenance) emits `task:reconcile-orphaned-pending-step-results` when it REWRITES `pending` workflow-step results with no live session behind them to `failed` (canonical liveness triple: `activeSessionRegistry` path, `executingTaskLock`, `isTaskActive`). It must never DELETE an orphaned entry — the merge gate blocks on pending/failed results, not on an enabled step with no result, so deletion silently satisfies the gate and the task merges with its review skipped; the `failed` rewrite keeps the gate closed and hands re-run/bypass to the failed-pre-merge-steps recovery and FN-7720 operator-bypass paths. `in-progress` rows are always skipped (executor-owned; resume is deferred at startup), the row is re-read immediately before the write, and user pauses are never disturbed. Metadata is ids/counts-only (`taskId`, `column`, `orphanedCount`, `resultCount`). - FN-8356: self-healing emits `task:reconcile-stale-duplicate-decision` when it clears a triage-marker duplicate-decision pause against a missing, deleted, done, or archived canonical. Metadata is ids/outcomes-only (`taskId`, `canonicalId`, `canonicalColumn`, `canonicalDeleted`, `priorPausedReason`); active canonical decisions and user pauses remain untouched. - U9b (R10/KTD-8): the self-healing STARTUP recovery step `adopt-legacy-task-rows` emits `task:reconcile-legacy-adoption` when it adopts a pre-cutover row through the KTD-8 adoption table (clearing a legacy `task.status` whose writer the cutover deleted so the graph re-enters at its owning node, and/or landing the one-time `reviewLevel` -> `enabledWorkflowSteps` preset backfill), and `task:reconcile-legacy-adoption-unmappable` when an UNKNOWN status parks the row `paused` for a human with its status deliberately left in place. Metadata is ids/counts/outcomes-only (`taskId`, `action`, `priorStatus`, `column`, `backfilledStepCount`, `reason`), where `reason` is a fixed adoption-table note and never row prose. Adoption runs FIRST in startup recovery (every later step reasons about `task.status`), stamps `task.legacyAdoptedAt` only on rows it actually mutates (so upgrade does not mass-write every `done` row), and never touches a user pause or a `preserve` gate. `planLegacyAdoption` in `packages/core/src/legacy-adoption.ts` is the single shared decision used by both this sweep and the store-open reconcile so the two cannot drift. diff --git a/packages/core/src/__tests__/postgres/workflow-continuation-slot.pg.test.ts b/packages/core/src/__tests__/postgres/workflow-continuation-slot.pg.test.ts new file mode 100644 index 0000000000..b5e30df60b --- /dev/null +++ b/packages/core/src/__tests__/postgres/workflow-continuation-slot.pg.test.ts @@ -0,0 +1,179 @@ +/* +FNXC:WorkflowAgentRouting 2026-08-07-23:50: +GROUND TRUTH for the single-active-continuation slot, against a REAL PostgreSQL +partial unique index. + +`idx_workflow_work_items_one_active_task_continuation` is + UNIQUE (project_id, task_id) WHERE kind = 'task' AND state IN ('runnable','running','held','retrying') +while `upsertWorkflowWorkItem`'s ON CONFLICT target is the DIFFERENT constraint +(project_id, run_id, task_id, node_id, kind). A row the graph has already left therefore does +not upsert — it RAISES. In production that raise deadlocked every task: the executor's routing +failed closed, the run re-suspended on every dispatch, and only an operator moving the card back +to the hold column cleared it. + +Every test here uses the real store and the real index, because a mock cannot have this bug: the +earlier regression suite exercised the repair's helper against a hand-rolled fake store and so +proved only that the helper branched as written. These pin the two facts the repair actually +rests on — that the bare upsert really does raise, and that +`replaceActiveTaskWorkflowContinuation` really does retire the predecessor and install the +successor atomically — and they pin the invariant across every shape that reaches it, not just +the one reported reproduction: + - a DIFFERENT node (the resumed continuation: `parse` -> `step-execute`) + - the SAME node id with a different runId (two materialized foreach instances, + `steps#0:step-execute` -> `steps#1:step-execute`) — the shape the first repair missed + - a `held` predecessor, not just a `running` one + - re-entering the same node twice (idempotent; must not retire itself) +*/ + +import { expect, it, beforeAll, beforeEach, afterEach, afterAll } from "vitest"; +import { + pgDescribe, + createSharedPgTaskStoreTestHarness, + type SharedPgTaskStoreHarness, +} from "../../__test-utils__/pg-test-harness.js"; +import { ACTIVE_WORKFLOW_WORK_ITEM_STATES } from "../../types.js"; +import type { WorkflowWorkItemState } from "../../types.js"; + +pgDescribe("single active task continuation slot", () => { + const h: SharedPgTaskStoreHarness = createSharedPgTaskStoreTestHarness({ + prefix: "fusion_continuation_slot", + }); + + beforeAll(h.beforeAll); + beforeEach(h.beforeEach); + afterEach(h.afterEach); + afterAll(h.afterAll); + + const fence = (taskId: string, runId: string, nodeId: string, extra: Record = {}) => ({ + runId, + taskId, + nodeId, + kind: "task" as const, + state: "running" as WorkflowWorkItemState, + leaseOwner: `executor:${taskId}`, + leaseExpiresAt: null, + nodeInstanceId: nodeId, + ...extra, + }); + + const activeRows = async (taskId: string) => { + const items = await h.store().listWorkflowWorkItemsForTask(taskId, { kinds: ["task"] }); + return items.filter((i) => ACTIVE_WORKFLOW_WORK_ITEM_STATES.includes(i.state)); + }; + + /* + The premise of the whole repair. If this ever stops raising — the ON CONFLICT target is + widened, or the partial index is dropped — the repair is solving a problem that no longer + exists and this test says so loudly instead of silently passing. + */ + it("a bare upsert RAISES when the task already holds an active continuation elsewhere", async () => { + const store = h.store(); + const task = await h.createTestTask(); + await store.upsertWorkflowWorkItem(fence(task.id, `${task.id}:run:parse`, "parse")); + + const err = await store + .upsertWorkflowWorkItem(fence(task.id, `${task.id}:run:step-execute`, "step-execute")) + .then(() => null, (e: unknown) => e as Error); + + expect(err, "a second active continuation must violate the partial unique index").toBeInstanceOf(Error); + expect((await activeRows(task.id)).map((i) => i.nodeId)).toEqual(["parse"]); + }); + + it("replace retires the resumed continuation and installs the successor atomically", async () => { + const store = h.store(); + const task = await h.createTestTask(); + await store.upsertWorkflowWorkItem(fence(task.id, `${task.id}:run:parse`, "parse")); + + const item = await store.replaceActiveTaskWorkflowContinuation( + fence(task.id, `${task.id}:run:step-execute`, "step-execute"), + ); + + const active = await activeRows(task.id); + expect(active.map((i) => i.nodeId)).toEqual(["step-execute"]); + expect(active[0]?.id).toBe(item.id); + // Never a window with zero active rows, and the predecessor is retired truthfully. + const all = await store.listWorkflowWorkItemsForTask(task.id, { kinds: ["task"] }); + expect(all.find((i) => i.nodeId === "parse")?.state).toBe("succeeded"); + }); + + /* + THE SHAPE THE FIRST REPAIR MISSED. Two materialized foreach instances share the template + `nodeId` ("step-execute") and differ only by runId/nodeInstanceId, so a node-identity guard + short-circuits and releases nothing — leaving instance #1 to deadlock exactly like the + original bug. Instance #0's fence is still active here because the interpreter terminalizes + fence rows only after it returns. + */ + it("replace retires a SIBLING FOREACH INSTANCE of the same template node", async () => { + const store = h.store(); + const task = await h.createTestTask(); + await store.upsertWorkflowWorkItem( + fence(task.id, `${task.id}:run:steps#0:step-execute`, "step-execute", { nodeInstanceId: "steps#0:step-execute" }), + ); + + await store.replaceActiveTaskWorkflowContinuation( + fence(task.id, `${task.id}:run:steps#1:step-execute`, "step-execute", { nodeInstanceId: "steps#1:step-execute" }), + ); + + const active = await activeRows(task.id); + expect(active).toHaveLength(1); + expect(active[0]?.nodeInstanceId).toBe("steps#1:step-execute"); + const all = await store.listWorkflowWorkItemsForTask(task.id, { kinds: ["task"] }); + expect(all.find((i) => i.nodeInstanceId === "steps#0:step-execute")?.state).toBe("succeeded"); + }); + + /* + A held predecessor is an ACTIVE row too, so it occupies the same slot. The availability-hold + write must be able to take the slot over from it rather than raising. + */ + it("replace takes the slot over from a HELD predecessor", async () => { + const store = h.store(); + const task = await h.createTestTask(); + await store.upsertWorkflowWorkItem( + fence(task.id, `${task.id}:run:plan-review`, "plan-review", { state: "held", leaseOwner: null }), + ); + + await store.replaceActiveTaskWorkflowContinuation( + fence(task.id, `${task.id}:run:step-execute`, "step-execute", { state: "held", leaseOwner: null, blockedReason: "workflow-principal-role-pool-exhausted:executor" }), + ); + + const active = await activeRows(task.id); + expect(active).toHaveLength(1); + expect(active[0]?.nodeId).toBe("step-execute"); + expect(active[0]?.state).toBe("held"); + }); + + /* Re-entering the same node must not retire the row it is about to write. */ + it("replace is idempotent for the same (runId, nodeId)", async () => { + const store = h.store(); + const task = await h.createTestTask(); + const input = fence(task.id, `${task.id}:run:step-execute`, "step-execute"); + + const first = await store.replaceActiveTaskWorkflowContinuation(input); + const second = await store.replaceActiveTaskWorkflowContinuation(input); + + expect(second.id).toBe(first.id); + const active = await activeRows(task.id); + expect(active).toHaveLength(1); + expect(active[0]?.state).toBe("running"); + }); + + /* + FNXC:WorkflowAgentRouting 2026-08-07-23:50 (cross-package drift guard): + The engine's handover is only correct because `ACTIVE_WORKFLOW_WORK_ITEM_STATES` matches the + state list written by hand into the index's raw `sql` predicate in another file. Nothing links + them, so adding a state to one side would silently un-fix the deadlock. Read the schema source + and assert they agree. + */ + it("the index predicate lists exactly ACTIVE_WORKFLOW_WORK_ITEM_STATES", async () => { + const { readFile } = await import("node:fs/promises"); + const { fileURLToPath } = await import("node:url"); + const schemaPath = fileURLToPath(new URL("../../postgres/schema/project.ts", import.meta.url)); + const source = await readFile(schemaPath, "utf8"); + + const predicate = /idx_workflow_work_items_one_active_task_continuation[\s\S]*?state\}? IN \(([^)]*)\)/.exec(source); + expect(predicate, "could not locate the partial unique index predicate").not.toBeNull(); + + const declared = [...predicate![1].matchAll(/'([^']+)'/g)].map((m) => m[1]).sort(); + expect(declared).toEqual([...ACTIVE_WORKFLOW_WORK_ITEM_STATES].sort()); + }); +}); diff --git a/packages/engine/src/__tests__/executor-continuation-slot-handover.test.ts b/packages/engine/src/__tests__/executor-continuation-slot-handover.test.ts deleted file mode 100644 index aae2f9f255..0000000000 --- a/packages/engine/src/__tests__/executor-continuation-slot-handover.test.ts +++ /dev/null @@ -1,166 +0,0 @@ -/* -FNXC:WorkflowAgentRouting 2026-08-07-23:35: -A task may hold only ONE active workflow work item -(`idx_workflow_work_items_one_active_task_continuation`: kind='task' AND state IN -runnable/running/held/retrying). A RESUMED graph run keeps the continuation it woke on active -until the interpreter returns — which is strictly AFTER a later role-classified node needs to -write its own durable principal fence. That fence upsert's ON CONFLICT target is -(run_id, task_id, node_id, kind), a DIFFERENT index, so the stale row does not upsert: the -write raises, routing returns `workflow-principal-fence-unavailable:`, and the run -suspends. Nothing about that state changes between dispatches, so the card deadlocked in its -wip column forever with no task error, and the only recovery was an operator manually bouncing -it back to the hold column (which clears the continuation). - -These tests pin the handover invariant and its two limits, because the repair is only safe if -it stays narrow: it must release a row for a node the run has LEFT, must never steal a live -claim on the node currently executing, and must report "nothing released" so the caller still -fails closed rather than looping on a conflict it cannot resolve. -*/ - -import "./executor-test-helpers.js"; -import { beforeEach, describe, expect, it, vi } from "vitest"; -import { TaskExecutor } from "../executor.js"; -import { resetExecutorMocks } from "./executor-test-helpers.js"; - -interface FakeWorkItem { - id: string; - taskId: string; - nodeId: string; - nodeInstanceId?: string; - kind: string; - state: string; -} - -function createStore(items: FakeWorkItem[]) { - const rows = [...items]; - const store = { - on: vi.fn(), - off: vi.fn(), - getRootDir: () => "/tmp/test", - listWorkflowWorkItemsForTask: vi.fn(async (_taskId: string, _opts?: unknown) => rows), - transitionWorkflowWorkItem: vi.fn(async (id: string, state: string) => { - const row = rows.find((candidate) => candidate.id === id); - if (!row) throw new Error(`unknown work item ${id}`); - row.state = state; - return row; - }), - recordRunAuditEvent: vi.fn().mockResolvedValue(undefined), - logEntry: vi.fn().mockResolvedValue(undefined), - } as any; - return { store, rows }; -} - -/** The private repair is reached through routing in production; call it directly here. */ -function supersede( - executor: TaskExecutor, - taskId: string, - nodeInstanceId: string, - nodeId: string, -): Promise { - return (executor as unknown as { - supersedeStaleActiveWorkItems: (t: string, i: string, n: string) => Promise; - }).supersedeStaleActiveWorkItems(taskId, nodeInstanceId, nodeId); -} - -describe("workflow continuation slot handover", () => { - beforeEach(() => { - resetExecutorMocks(); - }); - - it("releases the active continuation of a node the run has already left", async () => { - const { store, rows } = createStore([ - { id: "wi-parse", taskId: "FN-1", nodeId: "parse", kind: "task", state: "running" }, - ]); - const executor = new TaskExecutor(store, "/tmp/test"); - - const released = await supersede(executor, "FN-1", "steps#0:step-execute", "step-execute"); - - expect(released).toBe(1); - expect(rows.find((row) => row.id === "wi-parse")?.state).toBe("succeeded"); - // Truthful terminal state, and the lease must be dropped so nothing reads it as live. - expect(store.transitionWorkflowWorkItem).toHaveBeenCalledWith( - "wi-parse", - "succeeded", - expect.objectContaining({ leaseOwner: null, leaseExpiresAt: null }), - ); - }); - - /* - Every active state the partial unique index covers occupies the slot, so every one of them - must be releasable — a repair that only handled `running` would still deadlock on a card - parked `held` by an earlier availability hold. - */ - it.each(["running", "held", "runnable", "retrying"])( - "releases a stale continuation in the '%s' state", - async (state) => { - const { store, rows } = createStore([ - { id: "wi-stale", taskId: "FN-1", nodeId: "parse", kind: "task", state }, - ]); - const executor = new TaskExecutor(store, "/tmp/test"); - - expect(await supersede(executor, "FN-1", "steps#0:step-execute", "step-execute")).toBe(1); - expect(rows.find((row) => row.id === "wi-stale")?.state).toBe("succeeded"); - }, - ); - - it("never steals a claim on the node currently executing", async () => { - const { store, rows } = createStore([ - { - id: "wi-same-node", - taskId: "FN-1", - nodeId: "step-execute", - nodeInstanceId: "steps#0:step-execute", - kind: "task", - state: "running", - }, - ]); - const executor = new TaskExecutor(store, "/tmp/test"); - - // 0 released → the caller must fail closed instead of retrying the fence write. - expect(await supersede(executor, "FN-1", "steps#0:step-execute", "step-execute")).toBe(0); - expect(rows.find((row) => row.id === "wi-same-node")?.state).toBe("running"); - expect(store.transitionWorkflowWorkItem).not.toHaveBeenCalled(); - }); - - it("leaves already-terminal rows untouched", async () => { - const { store, rows } = createStore([ - { id: "wi-done", taskId: "FN-1", nodeId: "plan", kind: "task", state: "succeeded" }, - { id: "wi-failed", taskId: "FN-1", nodeId: "plan-review", kind: "task", state: "failed" }, - ]); - const executor = new TaskExecutor(store, "/tmp/test"); - - expect(await supersede(executor, "FN-1", "steps#0:step-execute", "step-execute")).toBe(0); - expect(store.transitionWorkflowWorkItem).not.toHaveBeenCalled(); - expect(rows.map((row) => row.state)).toEqual(["succeeded", "failed"]); - }); - - /* - Bookkeeping must never decide the run: an unreadable or unwritable work-item table reports - "nothing released" so routing fails closed, rather than throwing out of `beforeNodeExecution` - and terminalizing the task on a storage hiccup. - */ - it("reports nothing released when the work-item table cannot be read", async () => { - const { store } = createStore([]); - store.listWorkflowWorkItemsForTask = vi.fn().mockRejectedValue(new Error("db down")); - const executor = new TaskExecutor(store, "/tmp/test"); - - await expect(supersede(executor, "FN-1", "steps#0:step-execute", "step-execute")).resolves.toBe(0); - }); - - it("reports nothing released when the transition itself fails", async () => { - const { store, rows } = createStore([ - { id: "wi-parse", taskId: "FN-1", nodeId: "parse", kind: "task", state: "running" }, - ]); - store.transitionWorkflowWorkItem = vi.fn().mockRejectedValue(new Error("cas lost")); - const executor = new TaskExecutor(store, "/tmp/test"); - - await expect(supersede(executor, "FN-1", "steps#0:step-execute", "step-execute")).resolves.toBe(0); - expect(rows.find((row) => row.id === "wi-parse")?.state).toBe("running"); - }); - - it("degrades to no-op on a store without the work-item API", async () => { - const executor = new TaskExecutor({ on: vi.fn(), off: vi.fn(), getRootDir: () => "/tmp/test" } as any, "/tmp/test"); - - await expect(supersede(executor, "FN-1", "steps#0:step-execute", "step-execute")).resolves.toBe(0); - }); -}); diff --git a/packages/engine/src/__tests__/legacy-tombstones.test.ts b/packages/engine/src/__tests__/legacy-tombstones.test.ts index 8b244ef567..7bd1add9e4 100644 --- a/packages/engine/src/__tests__/legacy-tombstones.test.ts +++ b/packages/engine/src/__tests__/legacy-tombstones.test.ts @@ -74,6 +74,19 @@ const DELETED_SYMBOLS: Array<{ symbol: string; why: string }> = [ { symbol: "throwDeferredReviewerFatal", why: "deferred provider-error channel that only existed because a tool handler cannot throw" }, { symbol: "MAX_CODE_REVIEW_UNAVAILABLE_RETRIES", why: "UNAVAILABLE budget for the deleted in-session code review" }, /* + FNXC:WorkflowAgentRouting 2026-08-07-23:50: + The hand-rolled continuation handover. A durable continuation write that reacts to a FAILED + write by terminalizing other rows is the shape being tombstoned, not any particular name: it + cannot tell an index conflict from a transient database error, so it destroys legitimate holds; + it runs read-then-write across separate transactions, so a concurrent engine can lose its claim; + and it leaves a window with zero active rows, which strands the task silently. The repository + already has the correct primitive — `replaceActiveTaskWorkflowContinuation` retires + non-matching active rows and installs the successor under the task advisory lock in ONE + transaction. Reintroducing the recovery-shaped version reads as "handle the conflict", which is + exactly how it arrived the first time. + */ + { symbol: "supersedeStaleActiveWorkItems", why: "a non-atomic, ownership-blind handover; replaceActiveTaskWorkflowContinuation does it in one locked transaction" }, + /* FNXC:WorkflowCutover 2026-07-19-18:10 (U10b / R9): The legacy EXECUTE fallback. `maybeExecuteWorkflowGraph` returned a boolean meaning "did the graph claim this task", and `false` handed the run to a legacy implementation path — an diff --git a/packages/engine/src/executor.ts b/packages/engine/src/executor.ts index 8bc6263a84..3884a4ee00 100644 --- a/packages/engine/src/executor.ts +++ b/packages/engine/src/executor.ts @@ -6589,6 +6589,17 @@ export class TaskExecutor { * process-local map while scheduled continuations remain fenced in Postgres. */ const directWorkflowPrincipalWorkItemIds = new Set(); + /* + * FNXC:WorkflowAgentRouting 2026-08-07-23:50: + * The subset of the above that this run persisted as an availability HOLD. When it is + * non-empty the run already owns the task's single active continuation, parked `held` + * at the node that could not route — so the hold branch below must NOT also transition + * the row the run resumed on. That row was retired by the same atomic replace, and + * transitioning a terminal row throws (the store's terminal guard), which an earlier + * revision swallowed — leaving the task parked with ZERO active continuations, no + * error, and nothing scheduled to resume it. + */ + const directWorkflowPrincipalHeldWorkItemIds = new Set(); /* FNXC:GlobalConcurrencyControls 2026-07-14-18:30: The hold/release sweep may have already tryAcquired a global slot for this card before moving it to in-progress. Claim that pre-held slot for the full graph run so utilization stays honest between workflow nodes and triage cannot overfill the cap while this task is still graph-owned. @@ -6865,28 +6876,91 @@ export class TaskExecutor { activeSessions, }); if (routed.status === "unclassified") return undefined; + /* + * FNXC:WorkflowAgentRouting 2026-08-07-23:50: + * EVERY durable continuation write on this path goes through the atomic + * replace primitive, never a bare upsert. + * + * `idx_workflow_work_items_one_active_task_continuation` permits ONE active + * (`runnable`/`running`/`held`/`retrying`) `kind:"task"` row per task, and a + * plain upsert's ON CONFLICT target is a DIFFERENT constraint + * (run_id, task_id, node_id, kind). So a row this run has already left — the + * continuation it resumed on, or a previous foreach instance of the same + * template node, which shares `nodeId` and differs only by `runId` — does not + * upsert, it RAISES. That raise deadlocked the board: routing failed closed, + * the run re-suspended every dispatch, and only an operator bouncing the card + * cleared it. + * + * `replaceActiveTaskWorkflowContinuation` retires every active row that is not + * this exact (runId, nodeId, kind) and upserts the successor inside ONE + * transaction holding the task's advisory lock. That is what makes the handover + * atomic (no window with zero active rows), instance-aware (a sibling foreach + * instance has a different runId, so it is retired), and race-free against a + * concurrent engine (the lock serializes the read and the write). It is the + * repository's existing primitive for exactly this — `plan-review-continuation.ts` + * and `workflow-column-boundary-hooks.ts` already use it. + * + * Deliberately NOT an error-recovery path: an earlier revision reacted to a + * failed upsert by terminalizing other rows, which meant any transient database + * error destroyed a legitimate `held` continuation. Replacing unconditionally on + * the success path removes the need to classify errors at all. + */ + const writeContinuation = async ( + input: Parameters>[0] & { kind: "task" }, + ): Promise => { + if (typeof this.store.replaceActiveTaskWorkflowContinuation === "function") { + return await this.store.replaceActiveTaskWorkflowContinuation(input); + } + // Degradation for minimal/legacy stores without the atomic primitive: + // a bare upsert keeps the pre-primitive behavior rather than failing the run. + if (typeof this.store.upsertWorkflowWorkItem === "function") { + return await this.store.upsertWorkflowWorkItem(input); + } + return undefined; + }; + /* + * FNXC:WorkflowAgentRouting 2026-08-07-23:50: + * A hold write must NEVER throw out of `beforeNodeExecution`. Only + * `WorkflowGraphSuspended` is rethrown by the interpreter, so any other throw + * here degrades a recoverable availability hold into a terminal graph failure — + * the card is parked failed instead of waiting for its principal. Failing to + * RECORD the hold is bad; failing the task because we could not record it is + * worse. Log and continue: the refusal value still fails the node closed. + */ const holdDirectPrincipalWorkItem = async ( reason: string, principalAgentId: string | null, authorityKind: "task-assignee" | "review-node-override" | "column-binding" | "role-pool" | null, ): Promise => { - if (typeof this.store.upsertWorkflowWorkItem !== "function") return; - const item = await this.store.upsertWorkflowWorkItem({ - runId: `${resolvedRunId ?? `${nodeTask.id}:workflow`}:${nodeInstanceId}`, - taskId: nodeTask.id, - nodeId: node.id, - nodeInstanceId, - kind: "task", - state: "held", - leaseOwner: null, - leaseExpiresAt: null, - blockedReason: reason, - lastError: reason, - principalAgentId, - workflowRole: classifiedRole, - authorityKind, - }); - directWorkflowPrincipalWorkItemIds.add(item.id); + try { + const item = await writeContinuation({ + runId: `${resolvedRunId ?? `${nodeTask.id}:workflow`}:${nodeInstanceId}`, + taskId: nodeTask.id, + nodeId: node.id, + nodeInstanceId, + kind: "task", + state: "held", + leaseOwner: null, + leaseExpiresAt: null, + blockedReason: reason, + lastError: reason, + principalAgentId, + workflowRole: classifiedRole, + authorityKind, + }); + if (item) { + directWorkflowPrincipalWorkItemIds.add(item.id); + // The run now owns the task's single active continuation at THIS node, so + // the caller must not also transition the row it resumed on (that row is + // already retired, and transitioning a terminal row throws). + directWorkflowPrincipalHeldWorkItemIds.add(item.id); + } + } catch (holdErr) { + executorLog.error( + `[workflow-graph] ${nodeTask.id}: could not persist the availability hold for node '${node.id}' (${reason}): ` + + `${holdErr instanceof Error ? holdErr.message : String(holdErr)}`, + ); + } }; if (routed.status === "held") { const reviewerOverride = classifiedRole === "reviewer" ? node.reviewerAgentId : undefined; @@ -6971,29 +7045,40 @@ export class TaskExecutor { * fence as a claimed continuation. A persistence failure releases the * just-acquired capacity and fails closed rather than running ambient. */ - if (!durableWorkItemId && typeof this.store.upsertWorkflowWorkItem === "function") { - const writeFence = (): Promise => this.store.upsertWorkflowWorkItem!({ - runId: `${resolvedRunId ?? `${nodeTask.id}:workflow`}:${nodeInstanceId}`, - taskId: nodeTask.id, - nodeId: node.id, - kind: "task", - state: "running", - leaseOwner: `executor:${nodeTask.id}`, - leaseExpiresAt: null, - principalAgentId: routed.route.agent.id, - workflowRole: routed.route.role, - authorityKind: routed.route.authority, - nodeInstanceId, - }); + if (!durableWorkItemId) { /* - * FNXC:WorkflowAgentRouting 2026-08-07-23:12: - * Never swallow the fence-write error. Failing closed is correct — a session must - * not run without its durable principal record — but the original bare `catch {}` - * discarded the ONLY evidence of why, and this hold is not transient: the same - * write fails on every dispatch. The store wraps driver errors, so the actionable - * text (constraint name, NOT NULL column) is on `cause`, not `message`. + * FNXC:WorkflowAgentRouting 2026-08-07-23:50: + * The fence is written through the atomic replace primitive (see + * `writeContinuation` above), so the row this run already left — the resumed + * continuation, or a sibling foreach instance sharing this template `nodeId` — + * is retired in the SAME locked transaction that installs this fence. There is + * therefore no conflict to react to and no window in which the task has zero + * active continuations. + * + * A failure here still fails CLOSED: no session may start without its durable + * principal record. Release the just-acquired capacity and surface the store + * error, whose actionable text (constraint name, NOT NULL column) the store + * layer puts on `cause` rather than `message`. */ - const handleFenceFailure = async (fenceErr: unknown): Promise => { + try { + const item = await writeContinuation({ + runId: `${resolvedRunId ?? `${nodeTask.id}:workflow`}:${nodeInstanceId}`, + taskId: nodeTask.id, + nodeId: node.id, + kind: "task", + state: "running", + leaseOwner: `executor:${nodeTask.id}`, + leaseExpiresAt: null, + principalAgentId: routed.route.agent.id, + workflowRole: routed.route.role, + authorityKind: routed.route.authority, + nodeInstanceId, + }); + if (item) { + durableWorkItemId = item.id; + directWorkflowPrincipalWorkItemIds.add(item.id); + } + } catch (fenceErr) { const detail = fenceErr instanceof Error ? fenceErr.message : String(fenceErr); const cause = fenceErr instanceof Error && fenceErr.cause instanceof Error ? ` [cause: ${fenceErr.cause.message}]` @@ -7006,50 +7091,8 @@ export class TaskExecutor { nodeTask.id, `Workflow principal fence write failed at node '${node.id}' — ${detail.slice(0, 300)}${cause}`, ).catch(() => undefined); - void this.workflowAgentCapacity.release(attemptId, this.options.agentStore!.workflowProjectId ?? this.store.getRootDir()); + void this.workflowAgentCapacity.release(attemptId, this.options.agentStore.workflowProjectId ?? this.store.getRootDir()); return { outcome: "failure" as const, value: `workflow-principal-fence-unavailable:${routed.route.role}` }; - }; - try { - const item = await writeFence(); - durableWorkItemId = item.id; - directWorkflowPrincipalWorkItemIds.add(item.id); - } catch (firstFenceErr) { - /* - * FNXC:WorkflowAgentRouting 2026-08-07-23:20: - * HAND OVER the task's single active-continuation slot before failing closed. - * - * `idx_workflow_work_items_one_active_task_continuation` allows at most ONE - * active (`runnable`/`running`/`held`/`retrying`) task work item per task, and - * the fence upsert's ON CONFLICT target is (run_id, task_id, node_id, kind) — - * a different index — so a still-active row for a node this run has ALREADY - * LEFT does not upsert, it raises. That is exactly the steady state of a resumed - * run: the continuation the run woke up on stays active until the interpreter - * returns, which is after this node needs its fence. - * - * The result was a permanent deadlock with no trace: every dispatch re-resumed - * at the continuation node, walked forward to the first role-classified node, - * failed this write, suspended, and left the same rows behind — so the card sat - * in its wip column forever and the only recovery was an operator bouncing it - * back to the hold column (which clears the continuation). Superseding the - * stale row here is what makes that bounce unnecessary. - * - * Only rows for a DIFFERENT node are superseded, and only after the write has - * actually failed, so the invariant still holds and a genuine concurrent claim - * on this same node is never stolen. `succeeded` is truthful: traversal reached - * a later node, which is only reachable through that one. - */ - const superseded = await this.supersedeStaleActiveWorkItems(nodeTask.id, nodeInstanceId, node.id); - if (superseded > 0) { - try { - const item = await writeFence(); - durableWorkItemId = item.id; - directWorkflowPrincipalWorkItemIds.add(item.id); - } catch (retryErr) { - return await handleFenceFailure(retryErr); - } - } else { - return await handleFenceFailure(firstFenceErr); - } } } workflowCapacityAttemptIds.add(attemptId); @@ -7296,13 +7339,38 @@ export class TaskExecutor { executorLog.warn(holdMessage); } await this.store.logEntry(task.id, `Workflow stage held — ${principalHoldReason}`).catch(() => undefined); - if (continuation && typeof this.store.transitionWorkflowWorkItem === "function") { - await this.store.transitionWorkflowWorkItem(continuation.id, "held", { - leaseOwner: null, - leaseExpiresAt: null, - lastError: principalHoldReason, - blockedReason: principalHoldReason, - }).catch(() => undefined); + /* + * FNXC:WorkflowAgentRouting 2026-08-07-23:50: + * The task must end this run with EXACTLY ONE active continuation, and the hold + * write above may already be it. + * + * When routing persisted a `held` row, the atomic replace retired the row this run + * resumed on, so transitioning that resumed row here would (a) be redundant and + * (b) throw on the store's terminal guard. The previous `.catch(() => undefined)` + * hid exactly that throw and left the card parked with zero active rows, no error, + * and nothing to resume from — the same silent deadlock this whole change removes. + * + * So: skip when the hold is already durable. Otherwise (the fail-closed + * routing-unavailable path writes no row) fall back to parking the resumed + * continuation, and if even that fails, say so instead of swallowing it — a task + * with no durable continuation is stranded, and the stall watchdog only reports it. + */ + if (directWorkflowPrincipalHeldWorkItemIds.size === 0 + && continuation + && typeof this.store.transitionWorkflowWorkItem === "function") { + try { + await this.store.transitionWorkflowWorkItem(continuation.id, "held", { + leaseOwner: null, + leaseExpiresAt: null, + lastError: principalHoldReason, + blockedReason: principalHoldReason, + }); + } catch (holdErr) { + executorLog.error( + `[workflow-graph] ${task.id}: could not park the resumed continuation as held (${principalHoldReason}); ` + + `the task may have no active continuation to resume from: ${holdErr instanceof Error ? holdErr.message : String(holdErr)}`, + ); + } } return; } @@ -7700,86 +7768,6 @@ export class TaskExecutor { return undefined; } - /** - * FNXC:WorkflowAgentRouting 2026-08-07-23:20: - * Release the task's single active-continuation slot from rows this run has already - * advanced past, so the node currently executing can write its own durable principal - * fence. - * - * `idx_workflow_work_items_one_active_task_continuation` permits ONE active task work - * item per task. A resumed run keeps the continuation it woke on active until the - * interpreter returns, which is strictly after a later node needs its fence — so the - * fence write hits that partial unique index (NOT the upsert's ON CONFLICT target) and - * raises. Before this repair the run suspended there on every dispatch and the card - * deadlocked in its wip column until an operator bounced it back to the hold column. - * - * Deliberately narrow: called ONLY after a fence write has actually failed, and only rows - * for a DIFFERENT node/instance are superseded — a live claim on this same node is never - * stolen, so a genuine concurrent-owner conflict still fails closed. `succeeded` is the - * truthful terminal state: traversal reached a later node, which is reachable only - * through the one being superseded. - * - * @returns how many rows were released; 0 means the conflict was not a stale continuation - * and the caller must fail closed. - */ - private async supersedeStaleActiveWorkItems( - taskId: string, - currentNodeInstanceId: string, - currentNodeId: string, - ): Promise { - if (typeof this.store.listWorkflowWorkItemsForTask !== "function" - || typeof this.store.transitionWorkflowWorkItem !== "function") { - return 0; - } - let items: WorkflowWorkItem[]; - try { - items = await this.store.listWorkflowWorkItemsForTask(taskId, { kinds: ["task"] }); - } catch (err) { - executorLog.warn( - `[workflow-graph] ${taskId}: could not read work items to supersede a stale continuation: ${err instanceof Error ? err.message : String(err)}`, - ); - return 0; - } - const stale = items.filter((item) => - ACTIVE_WORKFLOW_WORK_ITEM_STATES.includes(item.state) - && item.nodeId !== currentNodeId - && (item.nodeInstanceId ?? item.nodeId) !== currentNodeInstanceId); - let released = 0; - for (const item of stale) { - try { - await this.store.transitionWorkflowWorkItem(item.id, "succeeded", { - leaseOwner: null, - leaseExpiresAt: null, - lastError: null, - }); - released += 1; - executorLog.log( - `[workflow-graph] ${taskId}: superseded stale active work item at node '${item.nodeId}' (state=${item.state}) so node '${currentNodeId}' can claim the continuation slot`, - ); - await this.store.recordRunAuditEvent?.({ - taskId, - agentId: "executor", - runId: generateSyntheticRunId("workflow-supersede-continuation", taskId), - domain: "database", - mutationType: "task:workflow-continuation-superseded", - target: taskId, - metadata: { - taskId, - supersededNodeId: item.nodeId, - supersededState: item.state, - currentNodeId, - currentNodeInstanceId, - }, - }).catch(() => undefined); - } catch (err) { - executorLog.warn( - `[workflow-graph] ${taskId}: failed to supersede stale work item at node '${item.nodeId}': ${err instanceof Error ? err.message : String(err)}`, - ); - } - } - return released; - } - private buildParseStepsDeps(runId?: string): ParseStepsHandlerDeps { return { readArtifact: (task, key): Promise => this.readTaskArtifact(task.id, key), diff --git a/packages/engine/src/triage.ts b/packages/engine/src/triage.ts index 081cf95933..70fd5d992f 100644 --- a/packages/engine/src/triage.ts +++ b/packages/engine/src/triage.ts @@ -7,6 +7,7 @@ import type { TaskAttachment, Settings, Agent, + WorkflowWorkItem, AgentPermissionPolicy, PermanentAgentGatingContext, WorkflowIr, @@ -2588,7 +2589,18 @@ export class TriageProcessor { returning so recovery and operator surfaces retain the fail-closed reason and no synthetic planner can silently retry around it. */ - await this.store.upsertWorkflowWorkItem({ + /* + FNXC:WorkflowAgentRouting 2026-08-07-23:50: + Write the planning continuation through the ATOMIC replace, not a bare upsert. + `idx_workflow_work_items_one_active_task_continuation` allows one active + `kind:"task"` row per task, and the upsert's ON CONFLICT target is a different + constraint — so an active continuation left at another node (a stranded resume, or + a hold from a prior run) makes this write RAISE instead of upserting. In the + executor that raise deadlocked the board; here it fails planning with + `workflow-principal-fence-unavailable:triage`. Replace retires the predecessor and + installs this row in one locked transaction, so planning takes the slot over. + */ + await this.writePlanningContinuation({ runId: triageRunContext.runId, taskId: task.id, nodeId: planningNode.id, @@ -2654,7 +2666,10 @@ export class TriageProcessor { return; } try { - const item = await this.store.upsertWorkflowWorkItem({ + // FNXC:WorkflowAgentRouting 2026-08-07-23:50: same atomic-replace contract as the + // held write above — a predecessor continuation at another node must be retired, + // not collided with. + const item = await this.writePlanningContinuation({ runId: triageRunContext.runId, taskId: task.id, nodeId: planningNode.id, @@ -4065,6 +4080,30 @@ export class TriageProcessor { * provider response. `updateTaskAtomic` holds the task lock across the live-row * predicate and patch, closing the scheduler-transition race. */ + /** + * FNXC:WorkflowAgentRouting 2026-08-07-23:50: + * Persist a planning continuation through the atomic replace primitive. + * + * A task may hold only ONE active (`runnable`/`running`/`held`/`retrying`) `kind:"task"` + * work item — enforced by the partial unique index + * `idx_workflow_work_items_one_active_task_continuation`, which is NOT the constraint a + * plain upsert's ON CONFLICT targets. `replaceActiveTaskWorkflowContinuation` retires every + * active row that is not this exact (runId, nodeId, kind) and installs the successor inside + * one transaction holding the task's advisory lock, so the handover has no conflict to + * recover from and no window with zero active rows. + * + * Falls back to a bare upsert only for a store without the primitive (minimal/legacy test + * adapters), which preserves the pre-primitive behavior rather than failing the run. + */ + private async writePlanningContinuation( + input: Parameters>[0] & { kind: "task" }, + ): Promise { + if (typeof this.store.replaceActiveTaskWorkflowContinuation === "function") { + return await this.store.replaceActiveTaskWorkflowContinuation(input); + } + return await this.store.upsertWorkflowWorkItem(input); + } + private async updatePlanningStateIfStillCurrent( task: Task, patch: Parameters[1] | ((live: Task) => Parameters[1]),