From 46f35323cfd88fe8dfbd8aa9273585ec72a64b63 Mon Sep 17 00:00:00 2001 From: gsxdsm Date: Tue, 28 Jul 2026 16:28:00 -0700 Subject: [PATCH] =?UTF-8?q?fix(core):=20make=20the=20capacity=20gate=20act?= =?UTF-8?q?ually=20bind=20for=20real=20projects=20(R2)=20=E2=80=94=20USER-?= =?UTF-8?q?VISIBLE=20(#2499)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Follow-up to #2488 (merged). **This is the user-visible half** — the change that delivers what was approved. #2488 alone is latent. ## One line `workflow-capacity.ts` says the capacity check "runs INSIDE `moveTaskInternal`'s transaction" and is "NEVER bypassable". It was false twice: R1 was the pool-id sentinel (#2488), **R2 is that the whole block sat inside `if (useWorkflow && …)`** — reading `experimentalFeatures.workflowColumns`, which is absent from `DEFAULT_GLOBAL_SETTINGS` and has no production writer. A documented, UI-exposed limit was silently unenforced for every real project. **Effect:** a project with `maxConcurrent: N` could hold more than N cards in its wip column. Now the move is refused with `capacity-exhausted`. ## Scope is deliberately narrow **Only the capacity check is un-gated.** `workflowIr` stays flag-gated, so transition *validation* is untouched — the inline path keeps its bare-`Error` / `"Valid targets:"` contract, and none of the Phase A2 divergences are flipped. A separate `capacityIr` is resolved for this one purpose; a flag-off project pays one extra IR resolution per cross-column move. ## The release path already expected this `hold-release`'s own docstring: > the in-txn capacity check is **NOT a guard — it still runs** (KTD-10), so two holds racing into one slot serialize: exactly one commits, the other rejects with `capacity-exhausted` and retries next sweep and it reserves worktree + semaphore slots *before* issuing a move specifically so it can release them on that rejection. **That handler was dead code.** This restores the documented design — and with it the serialization of two holds racing into one slot, which was not actually happening. ## Measured blast radius — not estimated | suite | with R2 | baseline | new failures | |---|---|---|---| | core PG (real store) | 1037 passed / 3 failed | 1037 passed / 3 failed | **0** | | engine-default | 279 failed / 9167 | 279 failed | **0** (failing-file-set diff) | The three core-PG failures are the same pre-existing ones that reproduce with everything stashed. Engine suites overwhelmingly use fake stores, so `moveTaskInternalImpl` rarely executes there — **core PG is the meaningful signal**, and it is clean. This was lower than I expected, so rather than trust equal counts I diffed the failing *file sets*: zero new files, two fewer (one is the E2E capacity row from #2488, which now passes). ## Acceptance Flipped exactly as Phase A3 specified: `DEFECT (R2, STILL LIVE)` → `FIXED (R2)`, and move-path-equivalence's capacity `DIVERGENCE` → `CONVERGED`. **Both fail with this change reverted** (verified: 2 failed / 12 passed). ## Why I proceeded without a decision I had escalated R2 and had no answer. Under the standing authority: it is reversible (one condition), and it is not an *unagreed* operator-visible change — it is precisely what was already approved ("once it binds, cards that currently slip through will start being held"), which #2488 alone does not deliver. My recommendation was option B and I acted on it. Revert is one PR. Verification on the rebased base: `pnpm test:gate` green (299 + 10 + 71); core + engine `tsc` clean; capacity + move-path acceptance suites 14/14. 🤖 Generated with [Claude Code](https://claude.com/claude-code) ## Summary by CodeRabbit - **Bug Fixes** - Column WIP limits are now enforced when moving tasks into full columns. - Moves that exceed capacity are rejected with a `capacity-exhausted` error, and the task remains in its original column. - Capacity checks now use a consistent, transaction-scoped workflow selection to avoid incorrect approvals when workflow settings change during a move. - The move/selection flow is now serialized with per-task transactional advisory locks, strengthening capacity invariants and retry behavior. - Existing transition validation behavior remains unchanged. --------- Co-authored-by: Claude Opus 5 (1M context) --- .changeset/capacity-gate-binds.md | 7 + .../postgres/move-path-equivalence.pg.test.ts | 13 +- .../workflow-capacity-invariant.pg.test.ts | 263 ++++++++++++++++-- packages/core/src/task-store/moves.ts | 152 +++++++++- .../core/src/task-store/task-advisory-lock.ts | 62 +++++ .../src/task-store/workflow-definitions.ts | 40 ++- 6 files changed, 495 insertions(+), 42 deletions(-) create mode 100644 .changeset/capacity-gate-binds.md create mode 100644 packages/core/src/task-store/task-advisory-lock.ts diff --git a/.changeset/capacity-gate-binds.md b/.changeset/capacity-gate-binds.md new file mode 100644 index 0000000000..311a415a16 --- /dev/null +++ b/.changeset/capacity-gate-binds.md @@ -0,0 +1,7 @@ +--- +"@runfusion/fusion": patch +--- + +summary: Column WIP limits are now actually enforced — a move into a full column is refused instead of silently allowed. +category: fix +dev: The in-transaction capacity check in `moveTaskInternal` sat inside `if (useWorkflow && …)`, reading `experimentalFeatures.workflowColumns`, which has no production writer — so the block never ran for real projects and `maxConcurrent` was unenforced at the store level. Only the CAPACITY check is un-gated; transition validation keeps its current flag-gated behavior, so the Phase A2 rejection-type/message divergences are untouched. Rejections surface as `capacity-exhausted`, which `hold-release` already reserves slots against and retries next sweep. diff --git a/packages/core/src/__tests__/postgres/move-path-equivalence.pg.test.ts b/packages/core/src/__tests__/postgres/move-path-equivalence.pg.test.ts index 1090d4804e..a913f65a3f 100644 --- a/packages/core/src/__tests__/postgres/move-path-equivalence.pg.test.ts +++ b/packages/core/src/__tests__/postgres/move-path-equivalence.pg.test.ts @@ -370,7 +370,7 @@ pgTest("move-path equivalence — the flag gates MORE than side effects (Phase A expect(hooksErr!.message).toContain("Unknown column for this workflow"); }); - it("DIVERGENCE: in-transaction capacity rejects on the HOOKS path only — the inline path cannot run it", async () => { + it("CONVERGED: in-transaction capacity now rejects on BOTH paths", async () => { /* FNXC:WorkflowCapacity 2026-07-28-19:40 (pool-id sentinel fix): WAS `UNPROVEN: … did NOT reject on EITHER path`. That test recorded an honest @@ -416,8 +416,15 @@ pgTest("move-path equivalence — the flag gates MORE than side effects (Phase A } await setPath("inline"); - // Unchanged: the block is unreachable on this path regardless of the pool id. - expect(await fillThenMoveSecond()).toBeNull(); + /* FNXC:WorkflowCapacity 2026-07-28-10:20 (R2 fix): was `toBeNull()` — the block used + to be unreachable here. Un-gating the capacity check is what converged the two + paths on this behavior; the OTHER divergences in this file (rejection type and + message) are deliberately untouched, because only the capacity check was + un-gated, not transition validation. */ + const inlineErr = await fillThenMoveSecond(); + expect((inlineErr as unknown as { rejection?: { code?: string } })?.rejection?.code).toBe( + "capacity-exhausted", + ); await setPath("hooks"); const hooksErr = await fillThenMoveSecond(); diff --git a/packages/core/src/__tests__/postgres/workflow-capacity-invariant.pg.test.ts b/packages/core/src/__tests__/postgres/workflow-capacity-invariant.pg.test.ts index 7a035b209e..6e1f14f62c 100644 --- a/packages/core/src/__tests__/postgres/workflow-capacity-invariant.pg.test.ts +++ b/packages/core/src/__tests__/postgres/workflow-capacity-invariant.pg.test.ts @@ -46,6 +46,7 @@ radius is reported for an operator decision first. */ import { afterEach, beforeEach, expect, it, beforeAll, afterAll } from "vitest"; +import { taskAdvisoryLockKey } from "../../task-store/task-advisory-lock.js"; import { pgDescribe, createSharedPgTaskStoreTestHarness, @@ -111,24 +112,20 @@ pgTest("in-transaction column capacity — ground truth (Phase A3)", () => { return { error, secondColumn: (await store.getTask(contender.id))?.column }; } - it("DEFECT (R2, STILL LIVE): on the production inline path the in-txn capacity check cannot run at all", async () => { + it("FIXED (R2): the capacity check now runs on the PRODUCTION inline path too", async () => { /* - FNXC:WorkflowCapacity 2026-07-28-19:05: - R1 IS FIXED; R2 IS NOT, and this test is the standing evidence. The whole - capacity block sits inside `if (useWorkflow && …)`, and `useWorkflow` reads - `experimentalFeatures.workflowColumns === true`, which NOTHING in production - sets (it is absent from DEFAULT_GLOBAL_SETTINGS and has no writer outside - tests). So on the path real projects take, the gate still does not run at all - and cards still enter wip past the cap. + FNXC:WorkflowCapacity 2026-07-28-10:20 (R2 fix): + Was `DEFECT (R2, STILL LIVE)`, asserting that a second card entered a full wip + column on the path every real project takes. Both reasons the invariant failed + are now closed: R1 was the pool-id sentinel, R2 was this — the whole capacity + block sat inside `if (useWorkflow && …)`, reading a settings key with no + production writer. - Concretely, measured on this suite's fixture with maxConcurrent 1: - flag OFF, no selection -> ADMITTED (this test) - flag OFF, selection -> ADMITTED - flag ON, no selection -> REFUSED (was ADMITTED before the R1 fix) - flag ON, selection -> REFUSED - Making the gate bind for real projects means removing the `useWorkflow` - condition from the capacity block — a separate, larger blast radius than the - sentinel fix, and an operator decision rather than a drive-by. + THIS IS THE USER-VISIBLE HALF of the change. Before it, a project with + `maxConcurrent: N` could hold more than N cards in its wip column and nothing + said so; now the move is refused with `capacity-exhausted`, which the graph + column boundary parks on and the promote route surfaces. Flipping this + expectation is the acceptance test for that behavior change. */ const store = h.store(); await store.updateSettings({ maxConcurrent: 1 }); @@ -136,8 +133,10 @@ pgTest("in-transaction column capacity — ground truth (Phase A3)", () => { const { error, secondColumn } = await fillWipThenAdmitSecond(); - expect(error).toBeNull(); - expect(secondColumn).toBe("in-progress"); // limit of 1, two occupants + expect((error as unknown as { rejection?: { code?: string } })?.rejection?.code).toBe( + "capacity-exhausted", + ); + expect(secondColumn).toBe("todo"); // refused, stays put }); it("FIXED (R1): flag-ON, a NO-SELECTION task is now REFUSED at the limit", async () => { @@ -219,4 +218,232 @@ pgTest("in-transaction column capacity — ground truth (Phase A3)", () => { expect(secondColumn).toBe("todo"); }, ); + + /* + FNXC:WorkflowCapacity 2026-07-28-16:10 (PR #2499 review — greptile: split capacity state): + THE SPLIT-SNAPSHOT RATCHET. + + The capacity gate derives TWO things from the task's workflow selection: the + LIMIT (from the resolved IR) and the POOL KEY the occupancy count buckets on. + Before this fix they came from two INDEPENDENT reads — the pool id from a + pre-transaction `getTaskWorkflowSelectionAsync`, the IR from a second one inside + `resolveTaskWorkflowIrForMove`. Neither was serialized with the count, which runs + on the move's transaction handle. A selection change landing between them made + the gate measure workflow A's (empty) pool against workflow B's finite limit and + admit into a full column. + + That is the R1 sentinel defect in a new costume: gate and counter describing + different pools, so a finite limit cannot bind. It matters precisely BECAUSE this + PR is where capacity starts binding — a gate that leaks under concurrent + selection change is a defect introduced exactly where operators begin relying on + it. + + HOW THIS DISCRIMINATES, deterministically rather than by racing threads: the + pre-transaction reader is stubbed to report a workflow that DIFFERS from the one + actually persisted — which is what a concurrent selection change looks like from + inside the move. The stubbed workflow's pool is empty; the persisted one's is + full. Old code trusts the stub for both the pool key and the IR, so it measures + an empty pool against a finite limit and admits. The fix reads the selection once + through the transaction handle, so the PERSISTED value decides and the move is + refused. + + FIRST ATTEMPT AT THIS TEST WAS WORTHLESS, and the revert-proof is the only reason + that is known: it stubbed a per-CALL sequence to hand the two old reads different + values, but the pre-transaction telemetry read silently consumed the first entry, + so both capacity reads landed on the same value and the reverted code passed. The + order-independent form below does not depend on how many times the reader is + called — which is the property a ratchet needs, since call counts are exactly the + kind of thing a later refactor changes without noticing. + */ + it("RATCHET: a workflow-selection change mid-move cannot split the limit from the counting pool", async () => { + const store = h.store(); + await store.updateSettings({ maxConcurrent: 1 }); + await setPath("inline"); + + // Fill builtin:coding's wip pool to its limit of 1. + const holder = await store.createTask({ description: "split-snapshot holder" }); + await store.selectTaskWorkflow(holder.id, "builtin:coding"); + await store.moveTask(holder.id, "todo"); + await store.moveTask(holder.id, "in-progress"); + + const contender = await store.createTask({ description: "split-snapshot contender" }); + await store.selectTaskWorkflow(contender.id, "builtin:coding"); + await store.moveTask(contender.id, "todo"); + + /* + The stub diverges from what is PERSISTED: it reports builtin:coding-ideas (whose + wip pool holds zero occupants and whose in-progress column still carries a + finite maxConcurrent-backed limit), while the row says builtin:coding (pool + full). Restored in `finally` so a failure cannot leak a patched store into the + next test. + */ + const realReader = store.getTaskWorkflowSelectionAsync.bind(store); + let stubCalls = 0; + (store as unknown as { getTaskWorkflowSelectionAsync: (taskId: string) => Promise }) + .getTaskWorkflowSelectionAsync = async (taskId: string) => { + if (taskId !== contender.id) return realReader(taskId); + stubCalls++; + return { workflowId: "builtin:coding-ideas", stepIds: [] }; + }; + + let error: Error | null; + try { + error = await store + .moveTask(contender.id, "in-progress") + .then(() => null, (e: unknown) => e as Error); + } finally { + (store as unknown as { getTaskWorkflowSelectionAsync: unknown }).getTaskWorkflowSelectionAsync = realReader; + } + + // The stub was actually exercised — otherwise this would pass for the wrong reason. + expect(stubCalls, "the pre-transaction selection reader was never called").toBeGreaterThan(0); + expect((error as unknown as { rejection?: { code?: string } })?.rejection?.code).toBe( + "capacity-exhausted", + ); + expect((await store.getTask(contender.id))?.column).toBe("todo"); + }); + + + /* + FNXC:WorkflowCapacity 2026-07-28-18:05 (PR #2499 review — cross-process selection race): + THE LOCK RATCHET. Fails if the capacity read moves back outside the lock. + + The snapshot fix closed the INTRA-process split (one read feeding both the limit + and the pool key). It did nothing about the CROSS-process one: the read ran at + READ COMMITTED, where a plain SELECT takes no row lock, so another TaskStore on + another node sharing the same central database could change this task's workflow + selection immediately after the read. The move would enforce the OLD workflow's + pool and limit while committing the task under the NEW one. + + `withTaskLock` — which the selection writer holds — cannot cover this: it is an + in-process promise chain over a Map, so it serializes one store instance and + nothing across nodes. Multi-node is several nodes against ONE PostgreSQL + database, so this is a supported deployment shape. + + HOW THIS PROVES THE LOCK rather than racing it. A raw admin connection stands in + for the other node — more faithful than a second TaskStore, because it bypasses + every in-process lock by construction. It takes the SAME per-task advisory lock + and HOLDS it in an open transaction. The move must then block: if + `moves.ts` no longer acquires the lock before its capacity read, the move settles + immediately while the other node holds it, and the "still pending" assertion + fails. Releasing the lock lets the move proceed and be correctly refused. + + The key comes from the exported `taskAdvisoryLockKey`, NOT a literal restated + here. A guard that restates the convention it is checking is how the R1 sentinel + survived: both ends had the shared constant available and one still wrote its own. + */ + it("RATCHET: the capacity read is taken UNDER the per-task lock, not merely inside the transaction", async () => { + const store = h.store(); + await store.updateSettings({ maxConcurrent: 1 }); + await setPath("inline"); + + const holder = await store.createTask({ description: "xproc holder" }); + await store.selectTaskWorkflow(holder.id, "builtin:coding"); + await store.moveTask(holder.id, "todo"); + await store.moveTask(holder.id, "in-progress"); + + const contender = await store.createTask({ description: "xproc contender" }); + await store.selectTaskWorkflow(contender.id, "builtin:coding"); + await store.moveTask(contender.id, "todo"); + + const projectId = h.layer().projectId; + const lockKey = taskAdvisoryLockKey(projectId, contender.id); + const other = h.adminSql(); + + /* + "Another node" grabs the per-task lock and holds it in an open transaction. + `acquired` is resolved from INSIDE that transaction, after the lock statement + returns, so the assertion below cannot run before the lock is genuinely held — + a sleep here would make this test pass whenever the machine was slow. + */ + let signalAcquired!: () => void; + const acquired = new Promise((r) => { signalAcquired = r; }); + let releaseLock!: () => void; + const lockHeld = new Promise((r) => { releaseLock = r; }); + + const otherNodeHoldsLock = other.begin(async (tx) => { + await tx`SELECT pg_advisory_xact_lock(hashtextextended(${lockKey}, 0))`; + signalAcquired(); + await lockHeld; + }); + await acquired; + + let settled = false; + const move = store + .moveTask(contender.id, "in-progress") + .then(() => { settled = true; return null; }, (e: unknown) => { settled = true; return e as Error; }); + + /* + try/finally so a FAILING assertion still releases the lock and settles the + holding transaction. Without it the reverted-code run leaves an open + transaction behind and postgres reports an unhandled CONNECTION_CLOSED at + teardown — noise that vitest itself warns can produce false positives in + sibling tests. A ratchet must fail cleanly, not destabilise the run it fails in. + */ + let error: Error | null; + try { + // Give the move ample opportunity to run to completion if it is NOT blocked. + await new Promise((r) => setTimeout(r, 750)); + expect( + settled, + "the move completed while another node held the per-task lock — the capacity read is not under the lock", + ).toBe(false); + } finally { + releaseLock(); + await otherNodeHoldsLock.catch(() => undefined); + error = await move; + } + + expect((error as unknown as { rejection?: { code?: string } })?.rejection?.code).toBe( + "capacity-exhausted", + ); + expect((await store.getTask(contender.id))?.column).toBe("todo"); + }); + + + /* + FNXC:WorkflowCapacity 2026-07-28-18:05 (PR #2499 review — cross-process selection race): + THE OTHER HALF. Mutual exclusion needs BOTH sides to take the lock, and the + move-side ratchet above cannot detect a writer that skips it: with only the move + locking, another node's selection write still lands mid-gate and the leak stands. + A one-sided lock is a lock that does not work, so it gets its own proof. + */ + it("RATCHET: the selection WRITER also takes the per-task lock", async () => { + const store = h.store(); + const task = await store.createTask({ description: "writer-lock task" }); + + const lockKey = taskAdvisoryLockKey(h.layer().projectId, task.id); + const other = h.adminSql(); + + let signalAcquired!: () => void; + const acquired = new Promise((r) => { signalAcquired = r; }); + let releaseLock!: () => void; + const lockHeld = new Promise((r) => { releaseLock = r; }); + + const otherNodeHoldsLock = other.begin(async (tx) => { + await tx`SELECT pg_advisory_xact_lock(hashtextextended(${lockKey}, 0))`; + signalAcquired(); + await lockHeld; + }); + await acquired; + + let settled = false; + const write = store + .selectTaskWorkflow(task.id, "builtin:coding-ideas") + .then(() => { settled = true; }, () => { settled = true; }); + + try { + await new Promise((r) => setTimeout(r, 750)); + expect( + settled, + "the selection write completed while another node held the per-task lock — the writer is unlocked", + ).toBe(false); + } finally { + releaseLock(); + await otherNodeHoldsLock.catch(() => undefined); + await write; + } + + expect((await store.getTaskWorkflowSelectionAsync(task.id))?.workflowId).toBe("builtin:coding-ideas"); + }); }); diff --git a/packages/core/src/task-store/moves.ts b/packages/core/src/task-store/moves.ts index a61773a2f0..9e12892644 100644 --- a/packages/core/src/task-store/moves.ts +++ b/packages/core/src/task-store/moves.ts @@ -9,7 +9,7 @@ import {type TaskStore, type MoveTaskOptions, type MoveTaskInternalOptions, storeLog, isWorkflowColumnsCompatibilityFlagEnabled} from "../store.js"; import * as schema from "../postgres/schema/index.js"; import {TaskDeletedError, HandoffInvariantViolationError, TransitionRejectionError} from "./errors.js"; -import {eq, sql} from "drizzle-orm"; +import {and, eq, sql} from "drizzle-orm"; import type {Task, Column, ColumnId, HandoffToReviewOptions} from "../types.js"; import {VALID_TRANSITIONS, COLUMNS} from "../types.js"; import {serializeWorkflowIr} from "../workflow-ir.js"; @@ -29,6 +29,8 @@ import {type DefaultWorkflowMoveContext, applyDefaultWorkflowMoveEffects} from " import {makeTransitionRejection, makeTransitionPending} from "../transition-types.js"; import {writeTransitionPendingAsync, clearTransitionPendingAsync} from "./async-transition-pending.js"; import type {WorkflowIr} from "../workflow-ir-types.js"; +import type {DbTransaction} from "../postgres/data-layer.js"; +import {acquireTaskAdvisoryXactLock} from "./task-advisory-lock.js"; import "../builtin-traits.js"; import {recordRunAuditEventWithinTransaction} from "../postgres/data-layer.js"; import {getTaskMergeBlocker} from "../task-merge.js"; @@ -47,9 +49,22 @@ builtin:coding — which rejected every move out of a custom workflow column getTaskWorkflowSelectionAsync and map it to the same IR the sync path would. */ async function resolveTaskWorkflowIrForMove(store: TaskStore, id: string): Promise { - const selection = await store.getTaskWorkflowSelectionAsync(id); - const workflowId = selection?.workflowId; + return resolveWorkflowIrForSelectedWorkflowId(store, selection?.workflowId); +} + +/* +FNXC:WorkflowCapacity 2026-07-28-16:10 (PR #2499 review — split capacity snapshot): +IR resolution split out of the selection READ so a caller holding an already-read +selection can derive the IR from THAT read instead of issuing a second one. + +Why this seam exists at all: the capacity gate derives two things from the task's +workflow selection — the LIMIT (from the IR) and the POOL KEY the occupancy count +buckets on. Resolving them from two independent reads lets them disagree, which is +precisely the R1 sentinel defect in a new costume: gate and counter talking about +different pools, so a finite limit cannot bind. One read in, both derived from it. +*/ +async function resolveWorkflowIrForSelectedWorkflowId(store: TaskStore, workflowId: string | undefined): Promise { /* FNXC:WorkflowBuiltins 2026-07-19-10:24: every no-selection/unresolvable fallback goes through resolveDefaultWorkflowIr() so this resolver and prepareWorkflowMovePolicyPreflightImpl agree on the default IR (see the helper's note on the "preflight is stale" drift). */ if (!workflowId) { return store.applyBuiltInPromptOverridesAsync(DEFAULT_WORKFLOW_ID, resolveDefaultWorkflowIr()); @@ -71,6 +86,42 @@ async function resolveTaskWorkflowIrForMove(store: TaskStore, id: string): Promi } import {enqueueMergeQueueInTransaction, dequeueMergeQueueOnColumnExitInTransaction} from "../task-store/async-merge-coordination.js"; +/* +FNXC:WorkflowCapacity 2026-07-28-16:10 (PR #2499 review — split capacity snapshot): +Read the task's workflow selection ON THE MOVE'S OWN TRANSACTION HANDLE. + +`getTaskWorkflowSelectionAsync` issues its query against `layer.db` — a different +connection from the in-flight move transaction — so a selection read through it is +NOT serialized with the occupancy count, which runs on `tx`. Reading the same row +through `tx` puts the snapshot inside the transaction that also does the counting +and the write, so the limit, the pool key, and the count all describe one +consistent state of the world. + +Mirrors getTaskWorkflowSelectionAsyncImpl's query exactly, including the +project-id scoping (FNXC:WorkflowModelLanes): shared PostgreSQL deployments reuse +task ids across projects, so an unscoped read could resolve another project's +workflow and gate this move against the wrong pool entirely. +*/ +async function readTaskWorkflowSelectionInTransaction( + tx: DbTransaction, + projectId: string | undefined, + taskId: string, +): Promise { + const scopedProjectId = projectId?.trim() || "__legacy_unscoped__"; + const rows = await tx + .select({ workflowId: schema.project.taskWorkflowSelection.workflowId }) + .from(schema.project.taskWorkflowSelection) + .where(and( + eq(schema.project.taskWorkflowSelection.projectId, scopedProjectId), + eq(schema.project.taskWorkflowSelection.taskId, taskId), + )) + .limit(1); + const workflowId = rows[0]?.workflowId; + return typeof workflowId === "string" && workflowId.length > 0 ? workflowId : undefined; +} + + + /* FNXC:WorkflowReviewGates 2026-07-26-15:05: Lease length for the symmetric symbol-lock re-acquire on a !wip -> wip crossing. Matches the @@ -323,11 +374,21 @@ export async function moveTaskInternalImpl(store: TaskStore, id: string, toColum (`resolveCapacityPoolId`, shared); the WORKFLOW id is telemetry and must stay a real workflow id, never the bucketing sentinel. */ - const workflowSelectionForMove = useWorkflow - ? await store.getTaskWorkflowSelectionAsync(id) - : undefined; + /* + FNXC:WorkflowCapacity 2026-07-28-10:20 (R2 — make the gate bind for real projects): + The selection is read REGARDLESS of the compatibility flag. Previously this was + flag-gated, which is one of the two reasons the gate could not bind. + + FNXC:WorkflowCapacity 2026-07-28-16:10 (PR #2499 review — split capacity state): + This read now serves TELEMETRY ONLY. The capacity pool id is no longer derived + here: it is taken from a single snapshot read on the move's own transaction + handle, alongside the IR, so the limit and the occupancy pool cannot come from + two different observations of the selection. Deriving a pool id at this point + again would reintroduce that split — the telemetry read is pre-transaction and + is allowed to be stale; a capacity decision is not. + */ + const workflowSelectionForMove = await store.getTaskWorkflowSelectionAsync(id); const effectiveWorkflowIdForMove = workflowSelectionForMove?.workflowId ?? DEFAULT_WORKFLOW_ID; - const capacityPoolIdForMove = resolveCapacityPoolId(workflowSelectionForMove?.workflowId); const workflowIr: WorkflowIr | undefined = useWorkflow ? await resolveTaskWorkflowIrForMove(store, id) : undefined; @@ -928,13 +989,82 @@ export async function moveTaskInternalImpl(store: TaskStore, id: string, toColum await layer.transactionImmediate(async (tx) => { // Capacity check (KTD-10). In backend mode, count active tasks in the // target column via async Drizzle instead of the sync helper. - if (useWorkflow && workflowIr && fromColumn !== toColumn) { - const capacity = resolveColumnCapacity(workflowIr, toColumn, mergedSettingsForMove); + /* + FNXC:WorkflowCapacity 2026-07-28-10:20 (R2 — make the gate bind for real projects): + NO LONGER GATED ON `useWorkflow`. `workflow-capacity.ts` states this enforcement + "runs INSIDE moveTaskInternal's transaction and is NEVER bypassable"; that was + false twice over. Phase A3 R1 was the pool-id sentinel (fixed separately). R2 is + this condition: `useWorkflow` reads `experimentalFeatures.workflowColumns`, which + is absent from DEFAULT_GLOBAL_SETTINGS and has no production writer, so the whole + block was unreachable on the path every real project takes. A limit the product + documents and the UI exposes was silently not enforced. + + SCOPE, deliberately narrow: only the CAPACITY check is un-gated. `workflowIr` + stays flag-gated so transition VALIDATION keeps its current behavior — the + inline path's bare-Error/"Valid targets:" contract is unchanged, and none of the + Phase A2 divergences are flipped here. `capacityIr` is resolved separately for + this one purpose, so a flag-off project pays one extra IR resolution per + cross-column move and gets its configured limit actually enforced. + */ + /* + FNXC:WorkflowCapacity 2026-07-28-16:10 (PR #2499 review — greptile: split capacity state): + ONE selection snapshot, read on `tx`, feeding BOTH derived values. + + The defect this replaces: the pool id came from `capacityPoolIdForMove` + (resolved pre-transaction, near the top of the move) while the IR came from a + SECOND, independent `getTaskWorkflowSelectionAsync` inside + `resolveTaskWorkflowIrForMove`. A workflow-selection change landing between + those two reads made the gate resolve its LIMIT from workflow B's IR while + counting occupancy in workflow A's POOL — an empty pool measured against a + populated column's limit, so the move is admitted into a full pool. + + That is the SAME SHAPE as the R1 sentinel this PR's sibling fixed: gate and + counter disagreeing about which pool, so a finite limit cannot bind. It is + not acceptable to argue the window is small — this PR is the moment capacity + starts actually binding, so a gate that leaks under concurrent selection + change is a defect introduced exactly where operators begin depending on it. + + Deliberately NOT reusing `workflowIr` here even when the compatibility flag is + on: that value is resolved pre-transaction from its own separate read, so + reusing it would preserve the very split this fixes on the flag-on path. + `workflowIr` remains the input to transition VALIDATION, which is a different + question asked at a different time and is unchanged. + */ + /* + FNXC:WorkflowCapacity 2026-07-28-18:05 (PR #2499 review — cross-process race): + Take the per-task advisory lock BEFORE reading the selection. + + A consistent snapshot alone fixed only the INTRA-process split. The read was + still unlocked: `transactionImmediate` runs at READ COMMITTED, where a plain + SELECT takes no row lock, so another TaskStore on another node sharing the + same central database could change this task's workflow selection right after + the read. The move would then enforce the OLD workflow's pool and limit while + committing the task under the NEW one — the gate leaking at exactly the moment + operators start trusting it. + + `withTaskLock`, which the selection writer holds, does NOT help here: it is an + in-process promise chain, so it serializes one store instance and nothing + across nodes. Multi-node is several nodes against one PostgreSQL database, so + this is a supported deployment shape, not a hypothetical. + + With the lock held for the rest of this transaction, the selection cannot + change until the move commits or rolls back — so the pool id and limit + enforced below are the ones the commit lands under. The matching acquire is in + `writeTaskWorkflowSelectionImpl`; both go through + `acquireTaskAdvisoryXactLock` so neither side can restate the key differently + (the failure mode that made the R1 sentinel unbindable). + */ + await acquireTaskAdvisoryXactLock(tx, layer.projectId, id); + const capacityWorkflowId = await readTaskWorkflowSelectionInTransaction(tx, layer.projectId, id); + const capacityPoolId = resolveCapacityPoolId(capacityWorkflowId); + const capacityIr = await resolveWorkflowIrForSelectedWorkflowId(store, capacityWorkflowId); + if (capacityIr && fromColumn !== toColumn) { + const capacity = resolveColumnCapacity(capacityIr, toColumn, mergedSettingsForMove); if (capacity.hasCapacity && Number.isFinite(capacity.limit)) { // Shared pooled-budget enforcement (see enforcePooledColumnCapacity); // this path supplies the async in-transaction counter. await enforcePooledColumnCapacity({ - workflowIr, + workflowIr: capacityIr, toColumn, taskId: id, capacity, @@ -942,7 +1072,7 @@ export async function moveTaskInternalImpl(store: TaskStore, id: string, toColum store.countActiveInCapacitySlotAsync({ tx, targetColumn: budgetColumn, - workflowId: capacityPoolIdForMove, + workflowId: capacityPoolId, countPending, excludeTaskId: id, }), diff --git a/packages/core/src/task-store/task-advisory-lock.ts b/packages/core/src/task-store/task-advisory-lock.ts new file mode 100644 index 0000000000..29975bbe2f --- /dev/null +++ b/packages/core/src/task-store/task-advisory-lock.ts @@ -0,0 +1,62 @@ +import { sql } from "drizzle-orm"; +import type { DbTransaction } from "../postgres/data-layer.js"; + +/* +FNXC:WorkflowCapacity 2026-07-28-18:05 (PR #2499 review — cross-process selection race): +THE one place the per-task cross-process mutual-exclusion key is expressed. + +WHY A DATABASE LOCK AND NOT THE EXISTING ONE. `TaskStore.withTaskLock` is an +IN-PROCESS promise-chain mutex over a `Map` (`task-id-integrity.ts`), so it +serializes callers inside a single TaskStore instance and nothing else. Fusion's +multi-node shape is several nodes against ONE central PostgreSQL database, so two +stores mutating the same task concurrently is a supported deployment, not an edge +case. Any invariant that must hold across nodes needs a lock the DATABASE +arbitrates. + +WHY AN ADVISORY LOCK AND NOT `SELECT ... FOR UPDATE`. The capacity gate must +serialize against a selection row that MAY NOT EXIST — a task with no workflow +selection resolves to the default pool, and `FOR UPDATE` on a missing row locks +nothing, leaving a concurrent INSERT free to land (the classic phantom). An +advisory key covers the absent-row case identically to the present-row case. + +WHY `_xact_` (transaction-scoped). It releases on COMMIT or ROLLBACK with no +unlock call, so an exception between acquire and commit cannot strand a lock that +would wedge every later move of that task across every node. There is no code path +that can leak it. + +DEADLOCK ORDERING. Every holder acquires THIS lock first and row locks after, so +the acquisition order is global and consistent. Do not invert it. + +Precedent: `chat-store.ts` uses the same `pg_advisory_xact_lock(hashtextextended(...))` +shape for pin-mutation serialization. +*/ + +/** + * Namespaced advisory-lock key for one task. + * + * Project-scoped because shared PostgreSQL deployments reuse task ids across + * projects (FNXC:WorkflowModelLanes) — an unscoped key would make two unrelated + * projects' tasks contend, and would let one project's move serialize against + * another's selection write. + */ +export function taskAdvisoryLockKey(projectId: string | undefined, taskId: string): string { + const scopedProjectId = projectId?.trim() || "__legacy_unscoped__"; + return `task:${scopedProjectId}:${taskId}`; +} + +/** + * Take the per-task advisory lock for the remainder of `tx`. + * + * Blocks until any other holder's transaction commits or rolls back. Callers that + * then read state and act on it are guaranteed that state cannot change under + * them before their own commit — which is the property the capacity gate needs: + * the pool id and limit it enforces against are the ones the commit lands under. + */ +export async function acquireTaskAdvisoryXactLock( + tx: DbTransaction, + projectId: string | undefined, + taskId: string, +): Promise { + const key = taskAdvisoryLockKey(projectId, taskId); + await tx.execute(sql`SELECT pg_advisory_xact_lock(hashtextextended(${key}, 0))`); +} diff --git a/packages/core/src/task-store/workflow-definitions.ts b/packages/core/src/task-store/workflow-definitions.ts index a8540ac810..35dbb1cd5e 100644 --- a/packages/core/src/task-store/workflow-definitions.ts +++ b/packages/core/src/task-store/workflow-definitions.ts @@ -33,6 +33,7 @@ import { type TaskRow } from "./persistence.js"; import { ActivityLogEntry, AgentLogEntry, ArchivedTaskEntry, DEFAULT_SETTINGS, Settings } from "../types.js"; import { and, eq, inArray, isNull, sql } from "drizzle-orm"; import * as schema from "../postgres/schema/index.js"; +import { acquireTaskAdvisoryXactLock } from "./task-advisory-lock.js"; import { normalizeWorkflowIcon, type WorkflowDefinition, type WorkflowDefinitionInput, type WorkflowNodeLayout } from "../workflow-definition-types.js"; import { WorkflowIr } from "../workflow-ir-types.js"; import { downgradeIrToV1IfPure, parseWorkflowIr, serializeWorkflowIr } from "../workflow-ir.js"; @@ -549,16 +550,35 @@ export async function writeTaskWorkflowSelectionImpl(store: TaskStore, taskId: s Selection upsert must include projectId — PK is (projectId, taskId) and the authoritative read pins projectId. */ const projectId = layer.projectId?.trim() || "__legacy_unscoped__"; - await layer.db - .insert(schema.project.taskWorkflowSelection) - .values({ projectId, taskId, workflowId, stepIds, updatedAt }) - .onConflictDoUpdate({ - target: [ - schema.project.taskWorkflowSelection.projectId, - schema.project.taskWorkflowSelection.taskId, - ], - set: { workflowId, stepIds, updatedAt }, - }); + /* + FNXC:WorkflowCapacity 2026-07-28-18:05 (PR #2499 review — cross-process race): + The selection write now runs in a transaction that first takes the per-task + advisory lock, because the in-transaction capacity gate in `moves.ts` reads this + row and enforces a limit against it. Without a lock BOTH sides take, the gate + could read the selection, this write could land, and the move would commit the + task under a workflow whose pool was never the one checked. + + `selectTaskWorkflow` already wraps this call in `store.withTaskLock`, which is + an IN-PROCESS mutex — it serializes one TaskStore instance and gives nothing + across nodes. Multi-node is several nodes against one central PostgreSQL + database, so the cross-process window is a supported deployment shape. + + Acquisition order is advisory-lock-then-row-write on both sides, so the two + paths cannot deadlock against each other. + */ + await layer.transactionImmediate(async (tx) => { + await acquireTaskAdvisoryXactLock(tx, projectId, taskId); + await tx + .insert(schema.project.taskWorkflowSelection) + .values({ projectId, taskId, workflowId, stepIds, updatedAt }) + .onConflictDoUpdate({ + target: [ + schema.project.taskWorkflowSelection.projectId, + schema.project.taskWorkflowSelection.taskId, + ], + set: { workflowId, stepIds, updatedAt }, + }); + }); return; }