From fbe7eb5c5af74f5a2100d45d4242cdc26560ba66 Mon Sep 17 00:00:00 2001 From: gsxdsm Date: Mon, 27 Jul 2026 21:39:21 -0700 Subject: [PATCH] U7 PR1: the manual plan-approval gate was bypassable (3 planning-lane surfaces, 8/13 revert-proof) (#2491) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit ## What this is The first slice of **U7 — the graph owns planning**. Characterizing the planning lane's dual ownership turned up a live defect in the exact seam the unit exists to remove, so this PR fixes that first and reports the measured map of what U7 still has to move. ## The defect The manual plan-approval gate parks a card by writing `status: "awaiting-approval"` and **returning early** from `finalizeApprovedTask` — before the release move. `specifyTask` then calls `onSpecifyComplete` **unconditionally** afterwards. Three automated surfaces went on to advance the parked card, each having re-derived its own weaker "may I advance this?" check from `paused`/`userPaused` alone. `isTaskBlockedOnApproval` (`packages/core/src/task-merge.ts`) already declares itself *"the single shared predicate core and engine code must consult before rebounding, requeuing, resuming, re-planning, or otherwise advancing a task"*. **Measured: it had exactly one production consumer** (`overseer-human-control-policy.ts`). Now four. Reachable end to end for a **plan-in-place** card — one whose column already equals the plan-review node's column (Coding (Ideas), or any `needs-replan` revision resting in the default workflow's `todo`): ``` park at awaiting-approval → onSpecifyComplete fires anyway → a runnable plan-review continuation is seeded → the drain dispatches it → Plan Review runs on a plan the operator never approved → its evidence satisfies isUnplannedForExecution → the capacity sweep releases the card into In progress ``` Blast radius: projects that have manual plan approval switched on. `planApprovalMode` defaults to auto-approve (FN-7557), so unset projects have no gate to skip — but the operator who turns it on is precisely the one who cares. ## Surface enumeration Per AGENTS.md — fix the invariant, not the repro. | # | Surface | Fix | |---|---|---| | 1 | `issueRelease` — the choke point for the sweep, `promoteHeldTask`, `releaseHeldTaskByEvent`, and the scheduler's `reserveSlot` guard | Guarded there rather than inside `isUnplannedForExecution`, because an approval-held card is not "unplanned". Guarded **again** inside the `moveTaskIf` predicate so a park landing mid-sweep cannot lose the race (R6 — only the in-txn check is authoritative). Operator force-promote (`allowUnplanned`) still waives it: that *is* a human decision about this card. | | 2 | **Both** continuation seeders — `seedPreReleasePlanReviewContinuation` (normal completion) and `evaluateStrandedHoldContinuation` (FN-8592 self-healing re-seed) | Guard at the seam, not in the callers: the seeder itself checked nothing, and its two callers each pre-checked a different subset. | | 3 | `resolvePlanningContinuationCandidate` (drain classifier) | **Skip, never orphan.** Cancelling terminalizes the item, so an approval landing a minute later would have nothing left to resume and would need a second repair to come back. | ## Measured, not assumed The two hold shapes `isTaskBlockedOnApproval` accepts were **not equally broken**. The `paused` + `pausedReason` shape was already refused by the sweep and the drain — they happen to test `paused` — so it was refused *for the wrong stated reason*, not advanced. Every genuine advance gap is on the **status-only** shape, which is exactly what the gate writes. Both are covered anyway, plus an `ORDINARY_PAUSE` counter-case so the new check cannot quietly become a catch-all for every operator park. ## Revert proof With the three production files reverted: **8 of 13 tests fail.** The 5 that still pass are the 3 controls and the 2 pause-shape rows the pre-existing `paused` checks already covered. ``` ·x··xxxxx·xx· → Tests 8 failed | 5 passed (13) ``` ## Verification | Check | Result | |---|---| | new suite | 13/13 | | hold-release (×2) + plan-review (×3) + pre-release-plan-review + promote-force-unplanned | 43/43 | | stranded-hold-continuation (×2) + continuation-selection + planning-finished-wake + planning-service | 27/27 | | scheduler-trait-dispatch | 9/9 | | `pnpm --filter @fusion/engine exec tsc --noEmit` | clean | | `pnpm lint` | clean | | `pnpm test:gate` | green | | `pnpm check:changesets` | clean | ## Two findings for the coordinator **1. `triage.ts` is absent from the Phase B census.** The plan's per-file table (535 sites) covers `self-healing.ts` (U4), the executor/scheduler cluster (U5), and the core policy modules (U6). `triage.ts` appears in none of them, so its lifecycle-column literals are unowned scope — U7 absorbs them. Measured with the plan's own methodology (block and line comments stripped, code lines only): a naive quoted-literal grep of `triage.ts` reports **50** sites, but **35 of those are the agent *role* string `"triage"`**, not the column. The genuine lifecycle-column surface is **15 sites**, of which 12 are planning-lane and 3 are `column !== "done"` in duplicate search. The 50 figure would over-count by 3.3×. **2. The graph's planning seam is a rubber stamp, in triplicate.** `createAuthoritativeWorkflowSeams().planning` returns `{ outcome: "success", value: "pre-specified" }`; `WorkflowPlanningService.runPlanningSession` returns the same; `createNoopLegacySeams().planning` is a bare success. The real specification is ~1,000 lines of `triage.specifyTask`, entirely outside the graph. That is the flip U7's remaining slices have to make, and it is the reason the planning lane has two owners at all. ## Deliberately not in this PR Triage's unconditional `onSpecifyComplete` call. That is the **ownership** half — `finalizeApprovedTask` must report whether it released, and the reaction must key on that outcome — and it belongs with the seam flip, where finalize's outcome becomes the graph's edge condition anyway, rather than as a half-measure now. With the three guards above in place, the downstream damage is already contained; what remains is a reaction firing for a non-event and an operator-visible log line (`Specified X → todo`) that is untrue for a parked card. 🤖 Generated with [Claude Code](https://claude.com/claude-code) ## Summary by CodeRabbit * **Bug Fixes** * Tasks awaiting manual plan approval are no longer automatically planned, reviewed, started, or released into active work. * Approval-held items are consistently skipped across planning continuations and related workflows. * Approval-held due work is deferred to prevent starvation while waiting, and operator force-promotion still bypasses the gate. * **Tests** * Added regression coverage to ensure the manual approval hold behavior remains invariant across multiple continuation scenarios. --------- Co-authored-by: Claude Opus 5 (1M context) --- .changeset/plan-approval-hold-invariant.md | 7 + .changeset/workflow-work-item-cas.md | 7 + .../__tests__/workflow-work-item-cas.test.ts | 130 ++++ .../task-store/async-workflow-workitems.ts | 12 + .../src/task-store/branch-and-pr-entities.ts | 5 + packages/core/src/types/merge-queue.ts | 14 + .../plan-approval-hold-invariant.test.ts | 559 ++++++++++++++++++ packages/engine/src/hold-release.ts | 42 +- .../engine/src/plan-review-continuation.ts | 42 +- .../engine/src/runtimes/in-process-runtime.ts | 248 ++++++-- 10 files changed, 1024 insertions(+), 42 deletions(-) create mode 100644 .changeset/plan-approval-hold-invariant.md create mode 100644 .changeset/workflow-work-item-cas.md create mode 100644 packages/core/src/__tests__/workflow-work-item-cas.test.ts create mode 100644 packages/engine/src/__tests__/plan-approval-hold-invariant.test.ts diff --git a/.changeset/plan-approval-hold-invariant.md b/.changeset/plan-approval-hold-invariant.md new file mode 100644 index 0000000000..bfc4e39482 --- /dev/null +++ b/.changeset/plan-approval-hold-invariant.md @@ -0,0 +1,7 @@ +--- +"@runfusion/fusion": patch +--- + +summary: A task awaiting manual plan approval is no longer planned, reviewed, or started before you approve it. +category: fix +dev: U7 (workflow-owned lifecycle) — `isTaskBlockedOnApproval` is now consulted by the three planning-lane advance surfaces that re-derived their own weaker check from `paused`/`userPaused`: `issueRelease` (plus its in-txn `moveTaskIf` predicate), both plan-review continuation seeders (`seedPreReleasePlanReviewContinuation`, `evaluateStrandedHoldContinuation`), and the drain classifier `resolvePlanningContinuationCandidate` (skip, never orphan). The gap was the status-only hold shape (`status: "awaiting-approval"`, no pause flag) the manual gate writes. Operator force-promote (`allowUnplanned`) still waives it. diff --git a/.changeset/workflow-work-item-cas.md b/.changeset/workflow-work-item-cas.md new file mode 100644 index 0000000000..9cdde3c29c --- /dev/null +++ b/.changeset/workflow-work-item-cas.md @@ -0,0 +1,7 @@ +--- +"@runfusion/fusion": patch +--- + +summary: A task queued for plan review no longer waits behind other tasks that are parked awaiting your approval. +category: fix +dev: U7 (PR #2491 review). The planning-continuation drain polls a bounded FIFO batch and a skipped item stayed `runnable` and due, so cards parked on approval re-filled every batch and starved newer plan-review work. Skipped operator-parks now get their `retryAfter` pushed out (`PARKED_CONTINUATION_DEFER_MS`, 60s) instead of a state change, so idleness predicates over `ACTIVE_WORKFLOW_WORK_ITEM_STATES` are unaffected. The write is a compare-and-set via the new `WorkflowWorkItemTransitionPatch.expectedState`, so a claim another node took between the poll and the write is never reset (`running` was not covered by the pre-existing terminal-state check). The drain loop moved to the exported `drainDuePlanningContinuations` so the wiring is testable without constructing a runtime. diff --git a/packages/core/src/__tests__/workflow-work-item-cas.test.ts b/packages/core/src/__tests__/workflow-work-item-cas.test.ts new file mode 100644 index 0000000000..c3f1042731 --- /dev/null +++ b/packages/core/src/__tests__/workflow-work-item-cas.test.ts @@ -0,0 +1,130 @@ +/* +FNXC:WorkflowWorkItemCas 2026-07-27-22:10 (U7, PR #2491 review — greptile P1): + +`WorkflowWorkItemTransitionPatch.expectedState` is a compare-and-set guard: the +transition applies only if the row's state read INSIDE the transaction still +equals it, and is otherwise a no-op returning the row untouched. + +WHY IT EXISTS: a caller that decided from a due-poll SNAPSHOT and then writes +unconditionally can clobber a newer state another node reached in between. The +concrete case is the planning drain's fairness deferral — it pushes an +operator-parked item's `retryAfter` forward so the item stops re-filling the FIFO +due batch. Written blind, that would reset a `running` claim back to `runnable` +and let the item be claimed twice. The pre-existing terminal-state check already +refuses cancelled/succeeded/failed (it throws), so `running` was the one +unguarded state, and it is the one a live worker holds. + +Losing the CAS is an ordinary outcome for a snapshot-driven caller, not an error — +hence a silent no-op rather than a throw. A throw would push every caller into a +try/catch whose only correct body is "do nothing". + +These run against real PostgreSQL through the shared harness, so the guard is +proven where it actually lives: inside the transaction that re-reads the row. +*/ +import { afterAll, afterEach, beforeAll, beforeEach, expect, it } from "vitest"; +import { createSharedPgTaskStoreTestHarness, pgDescribe, type SharedPgTaskStoreHarness } from "../__test-utils__/pg-test-harness.js"; + +function continuation(taskId: string) { + return { + runId: `${taskId}:continuation:cas`, + taskId, + nodeId: "plan-review", + kind: "task" as const, + state: "runnable" as const, + stableWorkflowRunId: `${taskId}:workflow`, + continuationSequence: 0, + waitReason: "planning" as const, + sourceColumn: "todo", + targetColumn: "todo", + irHash: "ir-test", + }; +} + +const LATER = "2026-07-27T12:01:00.000Z"; + +pgDescribe("workflow work-item transition compare-and-set", () => { + const h: SharedPgTaskStoreHarness = createSharedPgTaskStoreTestHarness({ prefix: "fusion_workitem_cas" }); + beforeAll(h.beforeAll); + afterAll(h.afterAll); + beforeEach(h.beforeEach); + afterEach(h.afterEach); + + it("applies the patch when the observed state still matches (the control)", async () => { + const store = h.store(); + const task = await store.createTask({ description: "cas match", column: "todo" }); + const item = await store.upsertWorkflowWorkItem(continuation(task.id)); + + const result = await store.transitionWorkflowWorkItem(item.id, "runnable", { + expectedState: "runnable", + retryAfter: LATER, + }); + + expect(result.state).toBe("runnable"); + expect(result.retryAfter).toBe(LATER); + }); + + it("is a silent NO-OP when another writer claimed the item first — the claim is not reset", async () => { + const store = h.store(); + const task = await store.createTask({ description: "cas claim race", column: "todo" }); + const item = await store.upsertWorkflowWorkItem(continuation(task.id)); + + // Another node claims it after our snapshot said `runnable`. + const claimed = await store.transitionWorkflowWorkItem(item.id, "running", { + leaseOwner: "other-node", + }); + expect(claimed.state).toBe("running"); + + // Our snapshot-driven deferral now loses the CAS. + const result = await store.transitionWorkflowWorkItem(item.id, "runnable", { + expectedState: "runnable", + retryAfter: LATER, + }); + + // The live claim survives untouched: state, owner, and retryAfter all unchanged. + expect(result.state).toBe("running"); + expect(result.leaseOwner).toBe("other-node"); + expect(result.retryAfter).not.toBe(LATER); + + // And the persisted row agrees — not just the returned value. + const persisted = (await store.listWorkflowWorkItemsForTask(task.id)).find((i) => i.id === item.id); + expect(persisted?.state).toBe("running"); + expect(persisted?.leaseOwner).toBe("other-node"); + expect(persisted?.retryAfter).not.toBe(LATER); + }); + + it("is a NO-OP rather than a throw for a terminalized item", async () => { + const store = h.store(); + const task = await store.createTask({ description: "cas terminal race", column: "todo" }); + const item = await store.upsertWorkflowWorkItem(continuation(task.id)); + await store.transitionWorkflowWorkItem(item.id, "cancelled", {}); + + // Without the CAS this same call throws (terminal state); with it the guard + // fires FIRST, so a snapshot-driven caller needs no try/catch to be correct. + const result = await store.transitionWorkflowWorkItem(item.id, "runnable", { + expectedState: "runnable", + retryAfter: LATER, + }); + + expect(result.state).toBe("cancelled"); + expect(result.retryAfter).not.toBe(LATER); + }); + + it("without expectedState the pre-existing unconditional behavior is unchanged", async () => { + const store = h.store(); + const task = await store.createTask({ description: "cas omitted", column: "todo" }); + const item = await store.upsertWorkflowWorkItem(continuation(task.id)); + await store.transitionWorkflowWorkItem(item.id, "running", { leaseOwner: "other-node" }); + + // No guard requested → the write lands, exactly as before this field existed. + const result = await store.transitionWorkflowWorkItem(item.id, "runnable", { retryAfter: LATER }); + + expect(result.state).toBe("runnable"); + expect(result.retryAfter).toBe(LATER); + + // And a terminal row still THROWS when no guard is requested. + await store.transitionWorkflowWorkItem(item.id, "cancelled", {}); + await expect( + store.transitionWorkflowWorkItem(item.id, "runnable", { retryAfter: LATER }), + ).rejects.toThrow(/terminal/); + }); +}); diff --git a/packages/core/src/task-store/async-workflow-workitems.ts b/packages/core/src/task-store/async-workflow-workitems.ts index 4a2e151f68..0a4fa156e7 100644 --- a/packages/core/src/task-store/async-workflow-workitems.ts +++ b/packages/core/src/task-store/async-workflow-workitems.ts @@ -361,6 +361,18 @@ export async function transitionWorkflowWorkItem( if (!existing) throw new Error(`Workflow work item ${id} not found`); const fromState = normalizeWorkflowWorkItemState(existing.state); + /* + FNXC:WorkflowWorkItemCas 2026-07-27-22:10 (U7, PR #2491 review — greptile P1): + Compare-and-set no-op. The state is re-read inside this transaction, so an + `expectedState` mismatch means another writer moved the row after the caller's + snapshot. Return it untouched rather than overwriting: a blind write here would + reset a `running` claim to `runnable` and let the item be claimed twice. + Not an error — losing this race is an ordinary outcome for a snapshot-driven + caller, and the correct response is to leave the newer state alone. + */ + if (patch.expectedState !== undefined && fromState !== patch.expectedState) { + return rowToWorkflowWorkItem(existing); + } if (isTerminalWorkflowWorkItemState(fromState) && fromState !== state) { throw new Error( `Workflow work item ${id} is terminal (${fromState}) and cannot transition to ${state}`, diff --git a/packages/core/src/task-store/branch-and-pr-entities.ts b/packages/core/src/task-store/branch-and-pr-entities.ts index 7b992d6dc8..e2e7847dfe 100644 --- a/packages/core/src/task-store/branch-and-pr-entities.ts +++ b/packages/core/src/task-store/branch-and-pr-entities.ts @@ -1099,6 +1099,11 @@ export function transitionWorkflowWorkItemSyncImpl(store: TaskStore, const existing = store.db.prepare("SELECT * FROM workflow_work_items WHERE id = ?").get(id) as WorkflowWorkItemRow | undefined; if (!existing) throw new Error(`Workflow work item ${id} not found`); const fromState = store.normalizeWorkflowWorkItemState(existing.state); + // FNXC:WorkflowWorkItemCas 2026-07-27-22:10 (U7, PR #2491 review — greptile P1): + // Mirrors the async path's compare-and-set no-op so the two cannot drift. + if (patch.expectedState !== undefined && fromState !== patch.expectedState) { + return store.rowToWorkflowWorkItem(existing); + } if (store.isTerminalWorkflowWorkItemState(fromState) && fromState !== state) { throw new Error(`Workflow work item ${id} is terminal (${fromState}) and cannot transition to ${state}`); } diff --git a/packages/core/src/types/merge-queue.ts b/packages/core/src/types/merge-queue.ts index 99d7ac77a3..e6b15e36bf 100644 --- a/packages/core/src/types/merge-queue.ts +++ b/packages/core/src/types/merge-queue.ts @@ -105,6 +105,20 @@ export interface WorkflowWorkItemTransitionPatch { lastError?: string | null; blockedReason?: string | null; now?: string; + /* + FNXC:WorkflowWorkItemCas 2026-07-27-22:10 (U7, PR #2491 review — greptile P1): + Compare-and-set guard. When set, the transition is applied ONLY if the row's + state read INSIDE the transaction still equals this value; otherwise it is a + no-op that returns the current row unchanged (no write, no run-audit row). + + Why this exists: a caller that decided from a due-poll SNAPSHOT and then writes + unconditionally can clobber a newer state another node reached in between — + resetting a `running` claim back to `runnable` (double-claim) is the concrete + case. The terminal-state check below already refuses cancelled/succeeded/failed, + so `running` was the unguarded gap. Callers that legitimately force a state + (the executor's own lifecycle writes) simply omit this and behave as before. + */ + expectedState?: WorkflowWorkItemState; } export interface WorkflowWorkItemDueFilter { diff --git a/packages/engine/src/__tests__/plan-approval-hold-invariant.test.ts b/packages/engine/src/__tests__/plan-approval-hold-invariant.test.ts new file mode 100644 index 0000000000..e568995503 --- /dev/null +++ b/packages/engine/src/__tests__/plan-approval-hold-invariant.test.ts @@ -0,0 +1,559 @@ +/* +FNXC:PlanApprovalHold 2026-07-27-19:30 (U7 / R4, R12 — workflow-owned lifecycle): + +THE INVARIANT: while a task is blocked on a pending human approval decision, no +AUTOMATED path advances it — not the capacity release, not the plan-review +continuation seed, not the continuation drain. `isTaskBlockedOnApproval` +(`packages/core/src/task-merge.ts`) already declares itself "the single shared +predicate core and engine code must consult before rebounding, requeuing, +resuming, re-planning, or otherwise advancing a task". Before this suite it had +exactly ONE production consumer (`overseer-human-control-policy.ts`), and the +planning lane's three advance surfaces each re-derived their own weaker version +of "may I advance this card" from `paused`/`userPaused` alone. + +WHY THE PLANNING LANE IS WHERE THIS BIT: the manual plan-approval gate parks the +card by writing `status: "awaiting-approval"` and RETURNING EARLY from +`finalizeApprovedTask` — before the release move. `specifyTask` then calls +`onSpecifyComplete` unconditionally (triage.ts), so the parked card is announced +as specified. For a PLAN-IN-PLACE card — one whose column already equals the +plan-review node's column (Coding (Ideas), or any `needs-replan` revision +resting in the default workflow's `todo`) — the seed's `node.column === +task.column` precondition holds, so a runnable plan-review continuation is +written for a plan the operator has not approved. The drain then dispatches it, +Plan Review runs, its evidence lands, and the capacity sweep releases the card +into implementation. The operator's gate is skipped end to end. + +Surface enumeration (AGENTS.md — "Fix the invariant, not the repro"). Every +automated surface that can advance a held card, and where each is covered: + 1. capacity release — `issueRelease`, the single choke point every release + surface funnels through (sweep, `promoteHeldTask`, `releaseHeldTaskByEvent`, + the scheduler's `reserveSlot` guard) — guarded there rather than inside + `isUnplannedForExecution`, because an approval-held card is not "unplanned". + Guarded again inside the `moveTaskIf` predicate so a park landing mid-sweep + cannot lose the race. Operator force-promote (`allowUnplanned`) deliberately + still passes: that IS a human decision about this card. [describe #1] + 2. continuation seed — `seedPreReleasePlanReviewContinuation` (normal + completion) and `evaluateStrandedHoldContinuation` (FN-8592 self-healing + re-seed). Both seeders, not just the one on the reported path. [describe #2] + 3. continuation drain— `resolvePlanningContinuationCandidate`, the classifier + that decides whether a due work item is dispatched. Skipped (HELD), never + cancelled: the operator may still approve, and a cancelled item would need + a second repair to come back. [describe #3] +Both approval HOLD SHAPES are exercised, because `isTaskBlockedOnApproval` +accepts either and a status-only check would silently miss the other: +`status: "awaiting-approval"` and `paused` + `pausedReason:"awaiting-approval"`. + +Each test below FAILS on the pre-fix code — the fix is a guard, and a guard that +cannot be shown to fail on the original defect is not a guard. + +────────────────────────────────────────────────────────────────────────────── +ADDED IN REVIEW ROUND 1 (PR #2491), because correctly HOLDING a card is not free: + + 4. `resolveParkedContinuationDeferral` — the due poll is a FIFO batch, and a + skipped item stays `runnable` and due, so it re-fills a slot every pass. + Before the guard an approval-held item was DISPATCHED and so never + accumulated; parking it correctly means 20 parked cards would starve every + newer plan-review continuation. This starvation is INTRODUCED by the guard, + so it is fixed here, not filed. [describe #4] + 5. `drainDuePlanningContinuations` — the pass that APPLIES the deferral. The + loop was extracted from a private runtime method for this test to exist at + all; deleting the `defer(...)` call now fails. The deferral also carries the + state the poll OBSERVED as a compare-and-set, so it cannot reset a claim + another node took mid-pass (`running` was not covered by the store's + terminal-state check). The store-side guard is proven against real + PostgreSQL in `packages/core/src/__tests__/workflow-work-item-cas.test.ts`. + [describe #5] +*/ +import { beforeEach, describe, expect, it, vi } from "vitest"; +import type { Task, TaskStore, WorkflowIr, WorkflowWorkItem } from "@fusion/core"; +import { AWAITING_APPROVAL_PAUSE_REASON, PLAN_REVIEW_GROUP_ID } from "@fusion/core"; + +import { runHoldReleaseSweep, resetHoldReleaseInstrumentation } from "../hold-release.js"; +import { + evaluateStrandedHoldContinuation, + seedPreReleasePlanReviewContinuation, +} from "../plan-review-continuation.js"; +import { + drainDuePlanningContinuations, + PARKED_CONTINUATION_DEFER_MS, + resolveParkedContinuationDeferral, + resolvePlanningContinuationCandidate, + type DuePlanningContinuationDrainDeps, +} from "../runtimes/in-process-runtime.js"; +import { schedulerLog } from "../logger.js"; + +const WF = "custom:planning-lane"; + +function task(over: Partial = {}): Task { + return { + id: "FN-1", + title: "t", + description: "", + column: "todo", + status: null, + dependencies: [], + steps: [], + currentStep: 0, + log: [], + createdAt: "2026-01-01T00:00:00.000Z", + updatedAt: "2026-01-01T00:00:00.000Z", + columnMovedAt: "2026-01-01T00:00:00.000Z", + ...over, + } as Task; +} + +/** + * The two hold shapes `isTaskBlockedOnApproval` accepts. Driven as a table so a + * fix that checks only `status` fails the second row rather than passing by + * covering the reported case alone. + * + * MEASURED, not assumed: the two shapes were NOT equally broken. The + * `paused`-flag shape was already refused by the release sweep and the drain, + * because both happen to test `paused` — it was refused for the wrong stated + * reason, not advanced. Every genuine ADVANCE gap is on the STATUS-only shape, + * which is precisely the one the plan-approval gate writes + * (`updatePlanningStateIfStillCurrent(task, { status: "awaiting-approval" })`, + * no pause flag). Both shapes are asserted anyway: a fix that covered only the + * reported shape would leave the predicate's other half unexercised. + * + * Attribution: an approval hold is reported as `awaiting-approval` on both + * shapes — for the pause-flag shape the `pausedReason` says so outright, so + * "paused" would be the less specific answer. The ORDINARY_PAUSE row below is + * the counter-case that keeps the new check from swallowing every pause. + */ +const APPROVAL_HOLDS: ReadonlyArray<{ label: string; fields: Partial }> = [ + { label: "status: awaiting-approval", fields: { status: "awaiting-approval" } }, + { + label: "paused + pausedReason: awaiting-approval", + fields: { paused: true, pausedReason: AWAITING_APPROVAL_PAUSE_REASON }, + }, +]; + +/** A pause that has nothing to do with approval. Must still be attributed to the + * pause, so the approval check cannot become a catch-all for operator parks. */ +const ORDINARY_PAUSE: Partial = { paused: true, pausedReason: "usage-limit" }; + +// ───────────────────────────────────────────────────────────────────────────── +// #1 — the capacity release surface +// ───────────────────────────────────────────────────────────────────────────── + +/** A hold column with a capacity release and a downstream wip column. No + * plan-review node: the release must be refused on the approval hold ALONE, + * not as a side effect of the pre-release plan-review gate. */ +function releaseIr(): WorkflowIr { + return { + version: "v2", + id: WF, + nodes: [], + edges: [], + columns: [ + { id: "todo", label: "Todo", traits: [{ trait: "hold", config: { release: "capacity" } }] }, + { id: "in-progress", label: "In progress", traits: [{ trait: "wip", config: { limitSetting: "maxConcurrent" } }] }, + { id: "done", label: "Done", traits: [{ trait: "complete" }] }, + ], + } as unknown as WorkflowIr; +} + +/** + * `onLockedRead` models the live row as the task lock sees it, which is NOT + * necessarily the snapshot `runHoldReleaseSweep` loaded at the top of the pass. + * Without it the fake would move unconditionally and the in-transaction recheck + * would be dead code that no test could distinguish from its absence. + */ +function releaseStore(tasks: Task[], onLockedRead?: (live: Task) => Task): TaskStore { + const selection = { workflowId: WF, stepIds: [] }; + const ir = releaseIr(); + return { + getSettings: vi.fn(async () => ({ maxConcurrent: 2 })), + listTasks: vi.fn(async () => tasks), + getTask: vi.fn(async (id: string) => tasks.find((t) => t.id === id) ?? null), + moveTaskIf: vi.fn(async ( + id: string, + column: string, + predicate: (live: Task) => boolean | Promise, + ) => { + const cur = tasks.find((t) => t.id === id)!; + // The predicate is the authoritative guard; the fake must consult it or the + // in-lock recheck is untested (PR #2491 review — greptile P2 / CodeRabbit). + const live = onLockedRead ? onLockedRead(cur) : cur; + if (!(await predicate(live))) return { task: cur, moved: false }; + cur.column = column; + return { task: cur, moved: true }; + }), + logEntry: vi.fn(async () => undefined), + recordRunAuditEvent: vi.fn(async () => undefined), + getCompletionHandoffAcceptedMarker: vi.fn(async () => null), + listWorkflowWorkItemsForTask: vi.fn(async () => []), + getTaskWorkflowSelection: vi.fn(() => selection), + getTaskWorkflowSelectionAsync: vi.fn(async () => selection), + getWorkflowDefinition: vi.fn(async () => ({ ir })), + } as unknown as TaskStore; +} + +describe("#1 the capacity release never advances a card blocked on approval", () => { + beforeEach(() => { + resetHoldReleaseInstrumentation(); + vi.restoreAllMocks(); + vi.spyOn(schedulerLog, "log").mockImplementation(() => {}); + vi.spyOn(schedulerLog, "debug").mockImplementation(() => {}); + vi.spyOn(schedulerLog, "warn").mockImplementation(() => {}); + }); + + it("releases an ordinary held card (the control — proves the fixture can release at all)", async () => { + const held = task({ id: "OK" }); + const result = await runHoldReleaseSweep(releaseStore([held]), { now: () => 1_000_000 }); + + expect(result.released).toContain("OK"); + expect(held.column).toBe("in-progress"); + }); + + for (const hold of APPROVAL_HOLDS) { + it(`holds a card parked on approval (${hold.label}) instead of releasing it into wip`, async () => { + const held = task({ id: "HOLD", ...hold.fields }); + const result = await runHoldReleaseSweep(releaseStore([held]), { now: () => 1_000_000 }); + + // The operator has not decided yet: the card must still be in the hold column. + expect(held.column).toBe("todo"); + expect(result.released).not.toContain("HOLD"); + }); + + it(`refuses IN THE LOCK when the hold (${hold.label}) lands after the sweep's snapshot`, async () => { + // The snapshot is clean, so the pre-check passes and the sweep proceeds to + // the move — the only thing that can still stop it is the predicate under + // the task lock. This is the race the in-txn half exists for: a plan gate + // (or an operator) parking the card mid-pass must win. + const held = task({ id: "RACE" }); + const store = releaseStore([held], (live) => ({ ...live, ...hold.fields })); + + const result = await runHoldReleaseSweep(store, { now: () => 1_000_000 }); + + expect(held.column).toBe("todo"); + expect(result.released).not.toContain("RACE"); + }); + } +}); + +// ───────────────────────────────────────────────────────────────────────────── +// #2 — the two continuation seeders +// ───────────────────────────────────────────────────────────────────────────── + +/** Plan-in-place shape: the plan-review node sits in the SAME column the card + * rests in, which is the precondition both seeders require. */ +function planInPlaceIr(): WorkflowIr { + return { + version: "v2", + id: WF, + name: WF, + columns: [ + { id: "todo", label: "Todo", traits: [{ trait: "hold", config: { release: "capacity" } }] }, + { id: "in-progress", label: "In progress", traits: [{ trait: "wip", config: { limitSetting: "maxConcurrent" } }] }, + ], + nodes: [ + { id: "start", kind: "start", column: "todo" }, + { + id: PLAN_REVIEW_GROUP_ID, + kind: "optional-group", + column: "todo", + config: { name: "Plan Review", defaultOn: true, template: { nodes: [], edges: [] } }, + }, + { id: "execute", kind: "prompt", column: "in-progress", config: {} }, + ], + edges: [ + { from: "start", to: PLAN_REVIEW_GROUP_ID }, + { from: PLAN_REVIEW_GROUP_ID, to: "execute", condition: "success" }, + ], + } as unknown as WorkflowIr; +} + +function seedStore(): { store: TaskStore; seeded: () => number } { + let seeds = 0; + const store = { + listWorkflowWorkItemsForTask: vi.fn(async () => [] as WorkflowWorkItem[]), + replaceActiveTaskWorkflowContinuation: vi.fn(async () => { + seeds += 1; + return { id: "wi-1" } as WorkflowWorkItem; + }), + seedStrandedPlanReviewContinuation: vi.fn(async () => { + seeds += 1; + return { seeded: true, workItemId: "wi-1" }; + }), + } as unknown as TaskStore; + return { store, seeded: () => seeds }; +} + +describe("#2 neither continuation seeder arms a run for a card blocked on approval", () => { + it("seeds for an ordinary specified card (the control)", async () => { + const { store, seeded } = seedStore(); + const result = await seedPreReleasePlanReviewContinuation(store, task(), planInPlaceIr()); + + expect(result.seeded).toBe(true); + expect(seeded()).toBe(1); + }); + + for (const hold of APPROVAL_HOLDS) { + it(`refuses the normal-completion seed (${hold.label})`, async () => { + const { store, seeded } = seedStore(); + const result = await seedPreReleasePlanReviewContinuation( + store, + task({ ...hold.fields }), + planInPlaceIr(), + ); + + expect(result.seeded).toBe(false); + expect(result.reason).toBe("awaiting-approval"); + expect(seeded()).toBe(0); + }); + + it(`refuses the FN-8592 self-healing re-seed (${hold.label})`, () => { + const verdict = evaluateStrandedHoldContinuation({ + task: task({ ...hold.fields }), + columnFlags: { hold: true }, + ir: planInPlaceIr(), + continuations: [], + stepResults: [], + effectiveSettings: {}, + enginePaused: false, + promptContent: "# FN-1 real spec\n\nA genuine plan.\n", + live: false, + stalenessMs: 60 * 60 * 1000, + graceMs: 1000, + }); + + // Not "stranded": the card is exactly where the operator's decision left it. + expect(verdict.stranded).toBe(false); + expect(verdict.reason).toBe("awaiting-approval"); + }); + } + + it("still attributes an ORDINARY pause to the pause, not to approval", async () => { + const { store, seeded } = seedStore(); + const seed = await seedPreReleasePlanReviewContinuation( + store, + task({ ...ORDINARY_PAUSE }), + planInPlaceIr(), + ); + expect(seed.seeded).toBe(false); + expect(seed.reason).toBe("paused"); + expect(seeded()).toBe(0); + + const verdict = evaluateStrandedHoldContinuation({ + task: task({ ...ORDINARY_PAUSE }), + columnFlags: { hold: true }, + ir: planInPlaceIr(), + continuations: [], + stepResults: [], + effectiveSettings: {}, + enginePaused: false, + promptContent: "# FN-1 real spec\n\nA genuine plan.\n", + live: false, + stalenessMs: 60 * 60 * 1000, + graceMs: 1000, + }); + expect(verdict.reason).toBe("paused"); + }); +}); + +// ───────────────────────────────────────────────────────────────────────────── +// #3 — the continuation drain +// ───────────────────────────────────────────────────────────────────────────── + +const dueItem = (over: Partial = {}): WorkflowWorkItem => + ({ + id: "wi-1", + taskId: "FN-1", + nodeId: PLAN_REVIEW_GROUP_ID, + kind: "task", + state: "runnable", + waitReason: "planning", + ...over, + } as WorkflowWorkItem); + +describe("#3 the continuation drain holds, and never cancels, an approval-blocked item", () => { + it("dispatches for an ordinary card (the control)", () => { + expect(resolvePlanningContinuationCandidate(dueItem(), task()).kind).toBe("actionable"); + }); + + for (const hold of APPROVAL_HOLDS) { + it(`skips dispatch (${hold.label}) and leaves the item claimable for after the decision`, () => { + const resolved = resolvePlanningContinuationCandidate(dueItem(), task({ ...hold.fields })); + + expect(resolved.kind).toBe("skip"); + // Deliberately NOT "orphan": cancelling would terminalize the item, so an + // approval landing later would have nothing left to resume. + expect(resolved.kind === "skip" && resolved.reason).toBe("awaiting-approval"); + }); + } + + it("still attributes an ORDINARY pause to the pause, not to approval", () => { + const resolved = resolvePlanningContinuationCandidate(dueItem(), task({ ...ORDINARY_PAUSE })); + + expect(resolved.kind).toBe("skip"); + expect(resolved.kind === "skip" && resolved.reason).toBe("paused"); + }); +}); + +// ───────────────────────────────────────────────────────────────────────────── +// #4 — the starvation the guard would otherwise introduce +// ───────────────────────────────────────────────────────────────────────────── + +/* +FNXC:PlanApprovalHold 2026-07-27-21:30 (U7, PR #2491 review — greptile P1): +Skipping is not free. The due poll is a FIFO batch (`limit: 20`) and a skipped item +stays `runnable` and due, so it re-occupies a slot every pass. Before the approval +guard an approval-held item was DISPATCHED and therefore never accumulated — so +this starvation is a consequence the guard INTRODUCES, and it is fixed here rather +than noted. A parked item is pushed out of the due window instead. +*/ +describe("#4 an operator-parked item leaves the due window instead of starving the batch", () => { + const NOW = Date.parse("2026-07-27T12:00:00.000Z"); + + for (const hold of APPROVAL_HOLDS) { + it(`defers the approval-parked item (${hold.label})`, () => { + const resolved = resolvePlanningContinuationCandidate(dueItem(), task({ ...hold.fields })); + const deferral = resolveParkedContinuationDeferral(resolved, NOW); + + expect(deferral?.itemId).toBe("wi-1"); + // Pushed strictly into the future, so the next due poll does not return it. + expect(Date.parse(deferral!.retryAfter)).toBe(NOW + PARKED_CONTINUATION_DEFER_MS); + expect(Date.parse(deferral!.retryAfter)).toBeGreaterThan(NOW); + }); + } + + it("defers an ordinary pause too — the same open-ended human wait", () => { + const resolved = resolvePlanningContinuationCandidate(dueItem(), task({ ...ORDINARY_PAUSE })); + + expect(resolveParkedContinuationDeferral(resolved, NOW)?.itemId).toBe("wi-1"); + }); + + it("never defers an ACTIONABLE item — deferring work that is ready would stall the lane", () => { + const resolved = resolvePlanningContinuationCandidate(dueItem(), task()); + + expect(resolved.kind).toBe("actionable"); + expect(resolveParkedContinuationDeferral(resolved, NOW)).toBeNull(); + }); + + it("never defers a non-planning item — that item belongs to a different drain", () => { + const resolved = resolvePlanningContinuationCandidate( + dueItem({ waitReason: "capacity" }), + task(), + ); + + expect(resolved.kind === "skip" && resolved.reason).toBe("not-planning"); + expect(resolveParkedContinuationDeferral(resolved, NOW)).toBeNull(); + }); + + it("never defers an ORPHAN — a cancelled item is terminal and must not be resurrected as runnable", () => { + const resolved = resolvePlanningContinuationCandidate(dueItem(), null); + + expect(resolved.kind).toBe("orphan"); + expect(resolveParkedContinuationDeferral(resolved, NOW)).toBeNull(); + }); +}); + +// ───────────────────────────────────────────────────────────────────────────── +// #5 — the drain PASS itself: the deferral is applied, and it is a compare-and-set +// ───────────────────────────────────────────────────────────────────────────── + +/* +FNXC:PlanApprovalHold 2026-07-27-22:10 (U7, PR #2491 review — CodeRabbit + greptile P1): +#4 tests the deferral DECISION; these test that the pass actually applies it, which +is the half that was unprovable while the loop lived inside a private method of a +class whose construction attaches to the real project registry. Deleting the +`defer(...)` call from the pass now fails here. + +The CAS case is the greptile P1 half: a blind write would reset a claim another node +took between the due poll and the write. The store's terminal-state check already +refuses cancelled/succeeded/failed, so `running` was the unguarded gap — deferral is +a fairness optimization and must never disturb live work to get it. +*/ +function drainHarness( + items: WorkflowWorkItem[], + tasks: Record, +): { + deps: DuePlanningContinuationDrainDeps; + dispatched: string[]; + deferred: Array<{ itemId: string; expectedState: string; retryAfter: string }>; + cancelled: Array<{ itemId: string; reason: string }>; +} { + const dispatched: string[] = []; + const deferred: Array<{ itemId: string; expectedState: string; retryAfter: string }> = []; + const cancelled: Array<{ itemId: string; reason: string }> = []; + return { + dispatched, + deferred, + cancelled, + deps: { + listDue: async () => items, + getTask: async (taskId) => tasks[taskId], + cancelOrphan: async (item, reason) => { cancelled.push({ itemId: item.id, reason }); }, + defer: async (d) => { deferred.push(d); }, + dispatch: (task, item) => { dispatched.push(`${task.id}@${item.id}`); }, + nowMs: () => Date.parse("2026-07-27T12:00:00.000Z"), + warn: () => {}, + }, + }; +} + +describe("#5 the drain pass applies the deferral and never starves a ready card", () => { + it("defers the approval-parked item, dispatches the actionable one, and does not cancel either", async () => { + const parked = dueItem({ id: "wi-parked", taskId: "FN-PARKED" }); + const ready = dueItem({ id: "wi-ready", taskId: "FN-READY" }); + const h = drainHarness([parked, ready], { + "FN-PARKED": task({ id: "FN-PARKED", status: "awaiting-approval" }), + "FN-READY": task({ id: "FN-READY" }), + }); + + await drainDuePlanningContinuations(h.deps); + + // The parked card leaves the due window; the ready card behind it still runs. + expect(h.deferred.map((d) => d.itemId)).toEqual(["wi-parked"]); + expect(h.dispatched).toEqual(["FN-READY@wi-ready"]); + expect(h.cancelled).toEqual([]); + }); + + it("carries the OBSERVED state as the compare-and-set guard, so a claim taken mid-pass is not reset", async () => { + // The due poll returns `runnable`; the write must be conditional on exactly + // that, so a concurrent `running` claim makes the store's CAS a no-op. + const parked = dueItem({ id: "wi-parked", taskId: "FN-PARKED", state: "runnable" }); + const h = drainHarness([parked], { + "FN-PARKED": task({ id: "FN-PARKED", status: "awaiting-approval" }), + }); + + await drainDuePlanningContinuations(h.deps); + + expect(h.deferred).toHaveLength(1); + expect(h.deferred[0].expectedState).toBe("runnable"); + expect(Date.parse(h.deferred[0].retryAfter)).toBe( + Date.parse("2026-07-27T12:00:00.000Z") + PARKED_CONTINUATION_DEFER_MS, + ); + }); + + it("cancels an orphan WITHOUT deferring it — a terminal item must not be written back as runnable", async () => { + const orphan = dueItem({ id: "wi-orphan", taskId: "FN-GONE" }); + const h = drainHarness([orphan], {}); + + await drainDuePlanningContinuations(h.deps); + + expect(h.cancelled).toEqual([{ itemId: "wi-orphan", reason: "task-not-found" }]); + expect(h.deferred).toEqual([]); + expect(h.dispatched).toEqual([]); + }); + + it("a getTask throw is an orphan, not an aborted pass — later items still dispatch (FN-8470/FN-8471)", async () => { + const bad = dueItem({ id: "wi-bad", taskId: "FN-THROWS" }); + const ready = dueItem({ id: "wi-ready", taskId: "FN-READY" }); + const h = drainHarness([bad, ready], { "FN-READY": task({ id: "FN-READY" }) }); + const deps: DuePlanningContinuationDrainDeps = { + ...h.deps, + getTask: async (taskId) => { + if (taskId === "FN-THROWS") throw new Error("soft-deleted, no archive snapshot"); + return task({ id: "FN-READY" }); + }, + }; + + await drainDuePlanningContinuations(deps); + + expect(h.cancelled).toEqual([{ itemId: "wi-bad", reason: "task-not-found" }]); + expect(h.dispatched).toEqual(["FN-READY@wi-ready"]); + }); +}); diff --git a/packages/engine/src/hold-release.ts b/packages/engine/src/hold-release.ts index 876e4fb49a..eef0fa6813 100644 --- a/packages/engine/src/hold-release.ts +++ b/packages/engine/src/hold-release.ts @@ -49,6 +49,7 @@ import { isUnplannedSeedPrompt, isWorkflowOptionalGroupEnabled, resolveEffectiveAutoMerge, + isTaskBlockedOnApproval, type TaskStore, type Task, type WorkflowIr, @@ -685,6 +686,35 @@ async function issueRelease( automatic surfaces — the sweep and the webhook release — never pass it, so FN-7648's invariant still holds for every non-operator release. */ + /* + FNXC:PlanApprovalHold 2026-07-27-19:30 (U7 / R4): + A card blocked on a pending human approval decision must never be released into + a processing column by an AUTOMATED surface. `isTaskBlockedOnApproval` is core's + declared "single shared predicate ... before rebounding, requeuing, resuming, + re-planning, or otherwise advancing a task" (task-merge.ts), and this release + path did not consult it. + + How it was reachable: the manual plan-approval gate parks the card by writing + `status: "awaiting-approval"` with NO pause flag, so the pre-existing + `task.paused || task.userPaused` skip in `runHoldReleaseSweep` did not match, + and `isUnplannedForExecution` enumerates only `planning` / `needs-replan`. Once + the plan-review gate's evidence landed, the sweep released a card whose plan the + operator had never approved — the gate was skipped end to end. + + Checked HERE rather than in each caller because `issueRelease` is the single + choke point every release surface funnels through (sweep, `promoteHeldTask`, + `releaseHeldTaskByEvent`, and the scheduler's `reserveSlot` guard). Gated on + `!options.allowUnplanned` for the same reason the unplanned check is: an + explicit operator force-promote IS a human decision about this card, so it + waives the human-decision gate, while no automatic surface can. + */ + if (targetIsProcessing && !options.allowUnplanned && isTaskBlockedOnApproval(task)) { + schedulerLog.debug( + `Hold release for ${task.id} blocked — awaiting a human approval decision (status=${task.status ?? "null"}, pausedReason=${task.pausedReason ?? "null"})`, + ); + return false; + } + if (targetIsProcessing && !options.allowUnplanned && (await isUnplannedForExecution(store, task, ir))) { /* FNXC:StrandedHoldContinuation 2026-07-26-14:15: @@ -749,11 +779,21 @@ async function issueRelease( /* FNXC:UserPausedDispatch 2026-07-21-21:45: Hold release must test the source column and both pause flags under the same task lock as the move. This makes an operator pause win atomically against scheduler dispatch and also replaces event-identity inference for concurrent release attempts. + + FNXC:PlanApprovalHold 2026-07-27-19:30 (U7 / R6): + The approval hold is tested here too, not only in the pre-check above. The + pre-check reads a task snapshot the sweep loaded earlier in the pass, so a plan + gate (or an operator) that parks the card between that read and this move would + otherwise lose the race and the card would release unapproved. Only the + predicate under the task lock is authoritative; the pre-check is an early exit. + `allowUnplanned` is carried through so an explicit operator force-promote waives + the gate here exactly as it does above — one waiver, not two policies. */ const result = await store.moveTaskIf( task.id, target, - (live) => live.column === originalColumn && live.paused !== true && live.userPaused !== true, + (live) => live.column === originalColumn && live.paused !== true && live.userPaused !== true + && !(targetIsProcessing && !options.allowUnplanned && isTaskBlockedOnApproval(live)), { moveSource: "scheduler", allocateWorktree: diff --git a/packages/engine/src/plan-review-continuation.ts b/packages/engine/src/plan-review-continuation.ts index 4896761580..ddab512ac1 100644 --- a/packages/engine/src/plan-review-continuation.ts +++ b/packages/engine/src/plan-review-continuation.ts @@ -1,6 +1,7 @@ import { ACTIVE_WORKFLOW_WORK_ITEM_STATES, computeWorkflowIrPin, + isTaskBlockedOnApproval, isUnplannedSeedPrompt, PLAN_REVIEW_GROUP_ID, type Task, @@ -13,6 +14,7 @@ import { resolvePreReleasePlanReviewNode } from "./hold-release.js"; export type StrandedHoldContinuationReason = | "not-hold-column" | "no-pre-release-review" | "active-continuation" | "plan-review-passed" | "seed-prompt" | "prompt-missing" | "triage-owned" + | "awaiting-approval" | "paused" | "engine-paused" | "live" | "too-fresh" | "auto-merge-off" | "ready"; /** @@ -27,9 +29,33 @@ export async function seedPreReleasePlanReviewContinuation( task: Task, ir: WorkflowIr, options: { atomic?: boolean } = {}, -): Promise<{ seeded: boolean; reason?: "active-continuation" | "plan-review-passed"; workItemId?: string }> { +): Promise<{ + seeded: boolean; + reason?: "active-continuation" | "plan-review-passed" | "awaiting-approval" | "paused"; + workItemId?: string; +}> { const node = resolvePreReleasePlanReviewNode(ir); if (!node || node.column !== task.column) return { seeded: false }; + /* + FNXC:PlanApprovalHold 2026-07-27-19:30 (U7 / R4): + Arming a runnable continuation is starting AI work on this card, so the parks + belong here at the seam rather than in each caller. Both callers already + pre-checked something — the runtime's `onSpecifyComplete` reaction checks the + pause flags, FN-8592's self-healing sweep checks its own fuller predicate — but + NEITHER checked the approval hold, and the seeder itself checked nothing. + + Why that mattered: the manual plan-approval gate parks the card by RETURNING + EARLY from `finalizeApprovedTask` after writing `status: "awaiting-approval"`, + while `specifyTask` announces completion unconditionally afterwards. For a + plan-in-place card (`node.column === task.column` — Coding (Ideas), or a + `needs-replan` revision resting in the default workflow's `todo`) the guard + above passes, so a Plan Review run was armed for a plan the operator had not + approved yet. The pause check is added alongside it because a seam that starts + work must refuse every operator park, not the subset its callers happened to + filter. + */ + if (isTaskBlockedOnApproval(task)) return { seeded: false, reason: "awaiting-approval" }; + if (task.paused === true || task.userPaused === true) return { seeded: false, reason: "paused" }; const items = await store.listWorkflowWorkItemsForTask(task.id); const active = items.filter((item) => ACTIVE_WORKFLOW_WORK_ITEM_STATES.includes(item.state)); if (active.length > 0) return { seeded: false, reason: "active-continuation" }; @@ -65,7 +91,7 @@ export async function seedPreReleasePlanReviewContinuation( * tokens represent an audit-worthy race loss from the conditional store op. */ export function evaluateStrandedHoldContinuation(input: { - task: Pick; + task: Pick; columnFlags: { hold?: boolean }; ir: WorkflowIr; continuations: WorkflowWorkItem[]; @@ -85,6 +111,18 @@ export function evaluateStrandedHoldContinuation(input: { if (input.promptContent === null) return { stranded: false, candidate: false, reason: "prompt-missing" }; if (isUnplannedSeedPrompt(input.promptContent, input.task.id, input.task.title, input.task.description)) return { stranded: false, candidate: false, reason: "seed-prompt" }; if (input.task.status === "planning" || input.task.status === "needs-replan") return { stranded: false, candidate: false, reason: "triage-owned" }; + /* + FNXC:PlanApprovalHold 2026-07-27-19:30 (U7 / R4): + An approval-held card is not stranded — it is exactly where the operator's + pending decision left it, so re-seeding would run Plan Review on an unapproved + plan. Reported as a CANDIDATE (like `paused`, unlike `triage-owned`) because the + block is an operator park that clears on its own: once the decision lands the + card becomes repairable again, and the audit distinction between "quiet + non-candidate" and "race loss" should still apply to it. + Placed before `paused` so the STATUS-only hold shape — the one the plan-approval + gate actually writes, with no pause flag — is matched at all. + */ + if (isTaskBlockedOnApproval(input.task)) return { stranded: false, candidate: true, reason: "awaiting-approval" }; if (input.task.paused || input.task.userPaused) return { stranded: false, candidate: true, reason: "paused" }; if (input.enginePaused) return { stranded: false, candidate: true, reason: "engine-paused" }; if (input.live) return { stranded: false, candidate: true, reason: "live" }; diff --git a/packages/engine/src/runtimes/in-process-runtime.ts b/packages/engine/src/runtimes/in-process-runtime.ts index a16ad7699e..c3caffaf67 100644 --- a/packages/engine/src/runtimes/in-process-runtime.ts +++ b/packages/engine/src/runtimes/in-process-runtime.ts @@ -16,11 +16,13 @@ import type { CliSession, NotificationPayload, WorkflowWorkItem, + WorkflowWorkItemState, } from "@fusion/core"; import { AsyncCentralClaimStore, ChatStore, isEphemeralAgent, + isTaskBlockedOnApproval, resolveWorkflowIrForTask, } from "@fusion/core"; import { Scheduler } from "../scheduler.js"; @@ -125,7 +127,7 @@ export function isPlanningContinuationTaskDispatchable( /** Outcome of resolving one due work item for the planning-continuation drain. */ export type PlanningContinuationResolution = | { kind: "actionable"; item: WorkflowWorkItem; task: Task } - | { kind: "skip"; item: WorkflowWorkItem; reason: "not-planning" | "paused" } + | { kind: "skip"; item: WorkflowWorkItem; reason: "not-planning" | "paused" | "awaiting-approval" } | { kind: "orphan"; item: WorkflowWorkItem; @@ -152,6 +154,21 @@ export function resolvePlanningContinuationCandidate( if (item.waitReason !== "planning") { return { kind: "skip", item, reason: "not-planning" }; } + /* + FNXC:PlanApprovalHold 2026-07-27-19:30 (U7 / R4): + Dispatching a planning continuation starts a Plan Review run, so a card blocked + on a pending human approval decision must not be dispatched. The status-only + hold shape the plan-approval gate writes (`status: "awaiting-approval"`, no + pause flag) fell straight through the pause check below and was dispatched. + + SKIP, never `orphan`: an orphan is cancelled and terminalized, so an approval + landing a minute later would find nothing left to resume and would need a second + repair (FN-8592's sweep) to come back. Skipping leaves the item due and + claimable, which is what "the operator has not decided yet" actually means. + */ + if (isTaskBlockedOnApproval(task)) { + return { kind: "skip", item, reason: "awaiting-approval" }; + } if (task.paused === true || task.userPaused === true) { return { kind: "skip", item, reason: "paused" }; } @@ -161,6 +178,141 @@ export function resolvePlanningContinuationCandidate( return { kind: "actionable", item, task }; } +/** + * FNXC:PlanApprovalHold 2026-07-27-21:30 (U7, PR #2491 review — greptile P1): + * How long a park-skipped continuation leaves the due window for. + * + * The due poll is a FIFO batch (`limit: 20`) and a skipped item stays `runnable` + * and due, so it re-fills a batch slot on every pass. Before the approval guard + * an approval-held item was DISPATCHED, so it never accumulated; now that it is + * correctly skipped, 20 cards parked on approval would starve every newer + * plan-review continuation until enough humans decided. That is a real + * consequence of the guard, not a pre-existing one. + * + * Deferral, not a state change: the item stays `runnable`, so every "is the graph + * idle?" predicate that reasons over ACTIVE_WORKFLOW_WORK_ITEM_STATES behaves + * exactly as before — moving it to `held` would remove it from the due set too, + * but nothing requeues a `held` planning item, which trades starvation for a + * permanent strand. `retryAfter` is a pure due-window filter, so the worst case + * is bounded latency instead. + * + * 60s is chosen against HUMAN latency: the park it defers is waiting on a person, + * who has already taken minutes or hours, so an extra minute after the decision + * is invisible — while occupancy of the shared batch drops from every poll (~2s) + * to at most one slot per minute per parked card. + */ +export const PARKED_CONTINUATION_DEFER_MS = 60_000; + +/** + * FNXC:PlanApprovalHold 2026-07-27-21:30 (U7, PR #2491 review — greptile P1): + * Decide whether a skipped due item should be pushed out of the due window. + * + * Only the OPERATOR-PARK skips qualify (`awaiting-approval`, `paused`): those are + * open-ended waits on a human, which is what makes them able to accumulate. + * `not-planning` is deliberately excluded — that item belongs to a different + * drain, and deferring another owner's work would be this drain reaching outside + * its own lane. + * + * Pure and separately exported so the deferral is testable without constructing a + * runtime, matching why `resolvePlanningContinuationCandidate` is exported. + */ +export function resolveParkedContinuationDeferral( + resolution: PlanningContinuationResolution, + nowMs: number, + deferMs: number = PARKED_CONTINUATION_DEFER_MS, +): { itemId: string; expectedState: WorkflowWorkItemState; retryAfter: string } | null { + if (resolution.kind !== "skip") return null; + if (resolution.reason !== "awaiting-approval" && resolution.reason !== "paused") return null; + return { + itemId: resolution.item.id, + /* + FNXC:WorkflowWorkItemCas 2026-07-27-22:10 (U7, PR #2491 review — greptile P1): + The state as the due poll SAW it, carried so the write is a compare-and-set. + A blind write would reset a claim another node took between the poll and here: + the store's terminal-state check refuses cancelled/succeeded/failed, but + `running` is not terminal, so `running -> runnable` would have succeeded and + let the item be claimed twice. Deferral is a fairness optimization — it must + never be able to disturb live work to achieve it. + */ + expectedState: resolution.item.state, + retryAfter: new Date(nowMs + deferMs).toISOString(), + }; +} + +/** The FIFO due-poll batch size. Named because the starvation the deferral above + * prevents is a property of this bound, so the two belong in one place. */ +export const DUE_PLANNING_CONTINUATION_BATCH_LIMIT = 20; + +/** Everything the drain pass touches, injected so the pass is exercisable without + * constructing a runtime (which would attach to the real project registry). */ +export interface DuePlanningContinuationDrainDeps { + listDue: () => Promise; + getTask: (taskId: string) => Promise; + cancelOrphan: ( + item: WorkflowWorkItem, + reason: "task-not-found" | "task-terminal", + ) => Promise; + defer: ( + deferral: { itemId: string; expectedState: WorkflowWorkItemState; retryAfter: string }, + ) => Promise; + /** `item` is passed only so the caller's failure log can keep naming the work + * item verbatim; the extraction is otherwise a byte-for-byte body move. */ + dispatch: (task: Task, item: WorkflowWorkItem) => void; + nowMs: () => number; + warn: (message: string) => void; +} + +/** + * FNXC:WorkflowScheduling 2026-07-21-12:20: + * A single runtime drain owns selection at a time. Concurrent wakeups collapse + * behind the caller's guard and the recurring processor supplies the next pass. + * + * FNXC:WorkflowScheduling 2026-07-21-22:31: + * Per-item task loads must not abort the pass. getTask throws for soft-deleted + * rows without an archive snapshot; one orphan earlier in created_at FIFO used + * to prevent every later planning continuation from dispatching (FN-8470 → FN-8471). + * Cancel orphaned work items so they leave the due set and free the batch window. + * + * FNXC:PlanApprovalHold 2026-07-27-22:10 (U7, PR #2491 review — CodeRabbit): + * EXTRACTED verbatim from `InProcessRuntime.drainWorkflowContinuations` with no + * behavior change, because the deferral wiring was unprovable where it lived: the + * method is private on a class whose construction attaches to the real project + * registry, so no test could tell "the drain applies the deferral" from "the drain + * ignores it". The re-entry guard and `status === "active"` check stay with the + * caller — those are runtime lifecycle, not pass logic. + */ +export async function drainDuePlanningContinuations( + deps: DuePlanningContinuationDrainDeps, +): Promise { + const items = await deps.listDue(); + for (const item of items) { + let task: Task | undefined; + let taskLookupFailed = false; + try { + task = await deps.getTask(item.taskId); + } catch (error) { + taskLookupFailed = true; + deps.warn( + `Workflow continuation ${item.id}: getTask(${item.taskId}) failed — treating as orphan: ${ + error instanceof Error ? error.message : String(error) + }`, + ); + } + const resolved = resolvePlanningContinuationCandidate(item, task, { taskLookupFailed }); + if (resolved.kind === "orphan") { + await deps.cancelOrphan(resolved.item, resolved.reason); + continue; + } + // FNXC:PlanApprovalHold 2026-07-27-21:30 (U7, PR #2491 review — greptile P1): + // push an operator-parked item out of the FIFO due window so it cannot starve + // newer actionable continuations while a human decides. + const deferral = resolveParkedContinuationDeferral(resolved, deps.nowMs()); + if (deferral) await deps.defer(deferral); + if (resolved.kind !== "actionable") continue; + deps.dispatch(resolved.task, resolved.item); + } +} + /** * FNXC:WorkflowScheduling 2026-07-21-12:30: * Select due planning continuations whose task remains dispatchable. @@ -2028,54 +2180,72 @@ export class InProcessRuntime }); } + /** + * FNXC:WorkflowScheduling 2026-07-21-12:20: + * A single runtime drain owns selection at a time. Concurrent wakeups collapse + * behind this guard and the recurring processor supplies the next bounded pass. + * + * The pass itself lives in `drainDuePlanningContinuations` (see its header for + * the FN-8470/FN-8471 orphan rationale and the deferral); this method is the + * runtime-lifecycle wrapper — re-entry guard, active-status check, and the + * adapters that bind the pass to this runtime's store and executor. + */ private async drainWorkflowContinuations(): Promise { - /* - FNXC:WorkflowScheduling 2026-07-21-12:20: - A single runtime drain owns selection at a time. Concurrent wakeups collapse - behind this guard and the recurring processor supplies the next bounded pass. - - FNXC:WorkflowScheduling 2026-07-21-22:31: - Per-item task loads must not abort the pass. getTask throws for soft-deleted - rows without an archive snapshot; one orphan earlier in created_at FIFO used - to prevent every later planning continuation from dispatching (FN-8470 → FN-8471). - Cancel orphaned work items so they leave the due set and free the limit:20 window. - */ if (this.workflowContinuationDrainActive || this.status !== "active") return; this.workflowContinuationDrainActive = true; try { - const items = await this.taskStore.listDueWorkflowWorkItems({ - kinds: ["task"], - states: ["runnable", "retrying"], - limit: 20, + await drainDuePlanningContinuations({ + listDue: () => this.taskStore.listDueWorkflowWorkItems({ + kinds: ["task"], + states: ["runnable", "retrying"], + limit: DUE_PLANNING_CONTINUATION_BATCH_LIMIT, + }), + getTask: (taskId) => Promise.resolve(this.taskStore.getTask(taskId)), + cancelOrphan: (item, reason) => this.cancelOrphanedWorkflowWorkItem(item, reason), + defer: (deferral) => this.deferParkedWorkflowWorkItem(deferral), + dispatch: (task, item) => { + void this.executor.execute(task).catch((error) => { + runtimeLog.error(`Workflow continuation ${item.id} failed:`, error); + }); + }, + nowMs: () => Date.now(), + warn: (message) => runtimeLog.warn(message), }); - for (const item of items) { - let task: Task | undefined; - let taskLookupFailed = false; - try { - task = await this.taskStore.getTask(item.taskId); - } catch (error) { - taskLookupFailed = true; - runtimeLog.warn( - `Workflow continuation ${item.id}: getTask(${item.taskId}) failed — treating as orphan: ${ - error instanceof Error ? error.message : String(error) - }`, - ); - } - const resolved = resolvePlanningContinuationCandidate(item, task, { taskLookupFailed }); - if (resolved.kind === "orphan") { - await this.cancelOrphanedWorkflowWorkItem(resolved.item, resolved.reason); - continue; - } - if (resolved.kind !== "actionable") continue; - void this.executor.execute(resolved.task).catch((error) => { - runtimeLog.error(`Workflow continuation ${resolved.item.id} failed:`, error); - }); - } } finally { this.workflowContinuationDrainActive = false; } } + /** + * FNXC:PlanApprovalHold 2026-07-27-21:30 (U7, PR #2491 review — greptile P1): + * Push an operator-parked item's `retryAfter` forward so it leaves the due + * window. State stays `runnable` on purpose (see + * `PARKED_CONTINUATION_DEFER_MS`). + * + * Fail-soft: if the write loses, the item simply stays due and is re-skipped + * next pass — exactly the pre-deferral behavior — so a store hiccup degrades to + * the old starvation risk rather than dropping the card's continuation. + */ + private async deferParkedWorkflowWorkItem( + deferral: { itemId: string; expectedState: WorkflowWorkItemState; retryAfter: string }, + ): Promise { + if (typeof this.taskStore.transitionWorkflowWorkItem !== "function") return; + try { + // `expectedState` makes this a compare-and-set: if another node claimed or + // terminalized the item since the due poll, the store returns it untouched. + await this.taskStore.transitionWorkflowWorkItem(deferral.itemId, deferral.expectedState, { + expectedState: deferral.expectedState, + retryAfter: deferral.retryAfter, + }); + } catch (error) { + runtimeLog.warn( + `Failed to defer parked workflow work item ${deferral.itemId}: ${ + error instanceof Error ? error.message : String(error) + }`, + ); + } + } + /** * FNXC:WorkflowScheduling 2026-07-21-22:31: * Terminalize a due work item whose task can no longer host graph work so the