fix(workflows): align stages with board columns (#2378)
## Summary The Coding (Ideas) workflow now behaves like the board it presents: Ideas stays inert, Todo owns planning and plan review, In progress owns implementation, and In review owns code review and merge. The restored preset is intentionally limited to that five-stage path, while the existing Coding workflow remains unchanged. Workflow execution now suspends at Todo→In progress instead of running the implementation node early. A durable, single-owner continuation records the exact resume node and survives process restarts; the scheduler remains the only component allowed to admit the task into WIP. Disabled optional review groups traverse the same boundary without invoking a reviewer, avoiding the prior stuck-task behavior. Workflow validation also rejects capacity holds with no reachable WIP destination, so deterministic lifecycle deadlocks fail at authoring time rather than after a task is running. Session-settled decisions carried from planning: columns are execution invariants, scheduler-owned WIP admission is preserved, the existing Coding (Ideas) preset is restored and simplified, and invalid release topology is rejected (user-approved). ## Validation - `pnpm lint` - `pnpm verify:fast` - `pnpm test:gate` (296 engine, 128 PostgreSQL core, and 63 CI-shape tests) - Focused workflow lifecycle tests (106 assertions) - PostgreSQL regression coverage proves atomic continuation replacement and database rejection of a second active owner <!-- This is an auto-generated comment: release notes by coderabbit.ai --> ## Summary by CodeRabbit * **New Features** * Added durable, resumable workflow execution across capacity boundaries (including explicit suspend/resume at the correct node). * Introduced Todo “plan review” workflow continuations and automated planning/capacity draining. * Restored Coding (Ideas) as a selectable built-in and updated its lane placement; improved optional-step group enablement support. * **Bug Fixes** * User moves back to Todo now cancels active workflow continuations. * Rejected workflow boundary transitions now surface as errors (instead of silently continuing). * Workflows with undriveable capacity-hold configurations are now rejected. * **Tests / Data** * Expanded coverage for workflow suspension, continuations, and continuation replacement; updated database schema to persist continuation metadata and enforce single active continuation. <!-- end of auto-generated comment: release notes by coderabbit.ai -->
This commit is contained in:
gsxdsm
7
.changeset/truthful-workflow-lifecycle.md
Normal file
7
.changeset/truthful-workflow-lifecycle.md
Normal file
@@ -0,0 +1,7 @@
|
||||
---
|
||||
"@runfusion/fusion": minor
|
||||
---
|
||||
|
||||
summary: Add a simple Ideas-to-Done workflow with truthful, resumable column transitions.
|
||||
category: feature
|
||||
dev: Persists capacity-boundary continuations and resumes the graph at the deferred node after scheduler release.
|
||||
@@ -0,0 +1,376 @@
|
||||
---
|
||||
title: "Truthful, Resumable Workflow Lanes - Plan"
|
||||
type: fix
|
||||
date: 2026-07-21
|
||||
deepened: 2026-07-21
|
||||
artifact_contract: ce-unified-plan/v1
|
||||
artifact_readiness: implementation-ready
|
||||
product_contract_source: ce-plan-bootstrap
|
||||
execution: code
|
||||
---
|
||||
|
||||
# Truthful, Resumable Workflow Lanes - Plan
|
||||
|
||||
## Goal Capsule
|
||||
|
||||
Make a workflow's authored columns an executable contract: Ideas stores work without AI, Todo creates and reviews `PROMPT.md`, In-progress implements, In-review reviews and merges, and Done is terminal. The engine must stop before crossing a scheduler-owned hold-to-WIP boundary, persist the next graph node, let the scheduler move the card when capacity is available, and resume at that node only after the move commits.
|
||||
|
||||
The user's confirmed five-column behavior is the product authority. Existing `builtin:coding` projects remain compatible; the repaired five-column flow is offered through the existing `builtin:coding-ideas` identity rather than silently migrating defaults. Stop implementation if the design would require node work to run outside its assigned column, use an in-memory-only continuation, or weaken `autoMerge:false`, pause/cancel, dependency, capacity, merge-proof, or workflow-drift safeguards.
|
||||
|
||||
Execution is branch-scoped feature work: use a dedicated worktree, keep the primary checkout on `main`, and land the complete behavior as one coherent change. The implementation owner carries the work through scoped verification, the merge gate, and the required `@runfusion/fusion` changeset; publishing a release is operator-only and out of scope.
|
||||
|
||||
---
|
||||
|
||||
## Product Contract
|
||||
|
||||
### Summary
|
||||
|
||||
Today a graph can encounter a Todo hold to In-progress WIP boundary, decline to move the card, and then execute the In-progress node anyway. The scheduler simultaneously refuses release when an enabled pre-release Plan Review has not passed. That circular ownership makes the intuitive five-column workflow either stall in Todo or run Plan Review after the card enters In-progress.
|
||||
|
||||
The fix adds a durable pause-and-resume protocol between graph traversal and scheduling, aligns optional-step enablement at every gate, rejects structurally undriveable workflow definitions, and restores the existing five-column coding workflow as a supported preset whose node placement matches the user's mental model.
|
||||
|
||||
### Actors
|
||||
|
||||
- A1. Operator authors or selects a workflow and manually promotes an idea into Todo.
|
||||
- A2. Planning agent creates or revises `PROMPT.md` while the card is in Todo.
|
||||
- A3. Plan reviewer approves or rejects `PROMPT.md` while the card is in Todo.
|
||||
- A4. Scheduler admits approved work into the WIP budget and moves it to In-progress.
|
||||
- A5. Executor implements only while the card is in In-progress.
|
||||
- A6. Code reviewer and merger operate while the card is in In-review, subject to `autoMerge` policy.
|
||||
- A7. Recovery services reconcile interrupted handoffs without skipping work or duplicating agent sessions.
|
||||
|
||||
### Requirements
|
||||
|
||||
#### Authored lane behavior
|
||||
|
||||
- R1. Creating a task in the five-column workflow lands it in Ideas and starts no AI or background execution until an operator promotes it.
|
||||
- R2. Promotion from Ideas to Todo starts planning in Todo, persists a real `PROMPT.md`, and runs enabled Plan Review attempts in Todo.
|
||||
- R3. A Plan Review REVISE outcome keeps the card in Todo, runs the existing bounded replan path, and reviews the revised artifact before release.
|
||||
- R4. No node assigned to In-progress may execute until the scheduler has committed the card's move into In-progress.
|
||||
- R5. Implementation nodes run in In-progress; entering an In-review Code Review node moves the card to In-review before the reviewer starts.
|
||||
- R6. A Code Review REVISE outcome enters its remediation node in In-progress before fixes start, then returns to In-review for another review.
|
||||
- R7. Completion summary, merge gates, merge attempts, retry/manual-hold behavior, and human review remain in In-review; confirmed merge advances to Done.
|
||||
- R8. Done is terminal and starts no additional work.
|
||||
|
||||
#### Durable orchestration
|
||||
|
||||
- R9. A graph reaching any scheduler-owned hold-to-WIP boundary returns a non-terminal suspended disposition before executing the target node.
|
||||
- R10. Suspension durably records the workflow run, resolved IR identity, target node, source column, target column, and held/runnable state so restart recovery can resume exactly once.
|
||||
- R11. The scheduler is the sole mover across hold-to-WIP capacity boundaries; it releases only a task whose durable continuation is ready and whose normal dependency, capacity, overlap, node, pause, and planning gates pass.
|
||||
- R12. After a successful release, graph execution resumes at the recorded target node instead of replaying from `start`; a racing or duplicate dispatch cannot run the continuation twice.
|
||||
- R13. Workflow edits that invalidate a suspended target follow the existing drift-park-and-re-resolve contract; they never execute a deleted or rehomed node against a stale IR.
|
||||
- R14. User pause, user move from In-progress to Todo, soft delete, and workflow reassignment cancel or supersede active continuations consistently with existing hard-cancel and workflow-work-item rules.
|
||||
|
||||
#### Authoring, compatibility, and visibility
|
||||
|
||||
- R15. Optional-group enablement has one shared interpretation at graph execution and release readiness: an explicit array wins, including `[]`; an absent array uses `defaultOn`.
|
||||
- R16. Workflow create, update, import, AI-design, CLI/agent tools, and dashboard save paths reject a capacity hold with no unambiguous reachable WIP release target and surface an actionable validation error.
|
||||
- R17. The editor describes column assignment as runtime placement and uses its existing lifecycle-warning surface to explain hold-to-WIP pause/release behavior; it must not imply that node placement is cosmetic.
|
||||
- R18. `builtin:coding` keeps its current columns, optional-step defaults, and existing-task behavior. Existing selections of `builtin:coding-ideas` continue resolving by the same ID.
|
||||
- R19. `builtin:coding-ideas` becomes selectable again and implements the confirmed five-column contract with only its essential Plan Review and Code Review gates: Plan Review in Todo, implementation in In-progress, and Code Review plus merge in In-review.
|
||||
- R20. Run audit and task logs expose suspension, release, resume, reconciliation, and invalid-continuation outcomes using IDs/counts/outcomes-only metadata.
|
||||
|
||||
### Key Flows
|
||||
|
||||
- F1. Capture and start: operator creates in Ideas → no work starts → operator promotes to Todo → planner writes `PROMPT.md` in Todo.
|
||||
- F2. Pre-release approval: Todo Plan Review APPROVE → graph suspends before the first In-progress node → scheduler reserves capacity and moves the card → executor resumes at that node.
|
||||
- F3. Pre-release rework: Todo Plan Review REVISE → planner revises in Todo → reviewer rechecks in Todo → only an approved result can reach the release seam.
|
||||
- F4. Delivery: implementation completes in In-progress → card enters In-review → Code Review and merge run there → confirmed merge moves to Done.
|
||||
- F5. Review rework: Code Review REVISE in In-review → card enters In-progress before remediation → fixes complete → card returns to In-review for the next review.
|
||||
- F6. Restart at seam: process stops after suspension or after the scheduler move → startup reconciliation reconstructs one authoritative continuation → work resumes once in the correct column.
|
||||
- F7. Optional review disabled: explicit removal of Plan Review bypasses its body, reaches the same durable release seam, and never deadlocks; an absent toggle list honors the workflow's `defaultOn` value.
|
||||
|
||||
### Acceptance Examples
|
||||
|
||||
- AE1. Given a new `builtin:coding-ideas` task, when no operator action occurs, then it remains in Ideas with no planning, review, execution, or merge session.
|
||||
- AE2. Given the task is promoted to Todo with Plan Review enabled, when planning finishes, then `PROMPT.md` exists and the Plan Review session starts while the persisted card column is Todo.
|
||||
- AE3. Given Plan Review approves and WIP is full, when the graph reaches `parse`, then the card remains in Todo, `parse` has not run, and one held continuation identifies `parse` as the next node.
|
||||
- AE4. Given the same held continuation and capacity becomes available, when the scheduler releases it, then the move to In-progress commits before `parse` or implementation starts, and the continuation is claimed once.
|
||||
- AE5. Given Plan Review is explicitly disabled with `enabledWorkflowSteps: []`, when planning finishes, then the task reaches the same scheduler release seam without a review session and does not remain stuck.
|
||||
- AE6. Given `enabledWorkflowSteps` is absent and Plan Review has `defaultOn: true`, when planning finishes, then Plan Review runs and release waits for its passing result.
|
||||
- AE7. Given Code Review returns REVISE, when remediation starts, then the card is already back in In-progress; when remediation succeeds, it re-enters In-review before another Code Review.
|
||||
- AE8. Given `autoMerge:false`, when Code Review passes, then the task remains in In-review until a human merge action; no recovery path moves it backward or promotes it automatically.
|
||||
- AE9. Given the engine restarts with a held or newly released continuation, when reconciliation runs, then exactly one node execution resumes and no completed Plan Review or implementation node is replayed.
|
||||
- AE10. Given an author saves a capacity-hold workflow with no reachable WIP column, then dashboard, HTTP API, and agent-tool saves all reject the same IR with the same domain validation reason.
|
||||
|
||||
### Success Criteria
|
||||
|
||||
- The five-column workflow completes F1-F4 in an automated lifecycle test without a fake boundary wrapper or manual scheduler simulation.
|
||||
- Session/run evidence proves every agent-bearing node begins only after the task is persisted in that node's assigned column.
|
||||
- Kill/restart tests cover both sides of the hold-release commit and show at-most-once continuation claims with eventual progress.
|
||||
- Default coding characterization tests remain green, including absent versus explicit optional-step toggle cases.
|
||||
- Invalid lifecycle shapes fail at every authoring surface before persistence.
|
||||
|
||||
### Scope Boundaries
|
||||
|
||||
In scope: graph traversal, workflow task-runner results, workflow-work-item persistence and leases, triage-to-graph handoff, scheduler release, startup/self-healing reconciliation, optional-group resolution, workflow validation, dashboard editor guidance, agent/API parity, the `builtin:coding-ideas` preset, audit events, regression tests, documentation, and a changeset.
|
||||
|
||||
Out of scope: changing the default workflow selection for existing projects; replacing the planning prompt or review prompts; redesigning the workflow editor canvas; changing merge strategies; changing global concurrency semantics; publishing a release; and migrating `builtin:coding` node placement.
|
||||
|
||||
### Symptom Verification
|
||||
|
||||
- **Original symptom:** placing Plan Review in Todo either prevents release forever or causes the review to appear at the beginning of In-progress; node work can execute while the card remains in the preceding column.
|
||||
- **Exact reproduction:** create/select the five-column Ideas workflow, promote an idea to Todo, enable Plan Review, allow planning to create `PROMPT.md`, and drive the real graph plus scheduler to the Todo hold → In-progress WIP boundary.
|
||||
- **Assertion it is gone:** Plan Review starts and finishes with `task.column === "todo"`; the first In-progress handler has zero calls while the continuation is held; after scheduler release it starts with `task.column === "in-progress"`; the full task then reviews/merges in In-review and reaches Done.
|
||||
|
||||
### Surface Enumeration
|
||||
|
||||
| Surface | States and variants that must be covered |
|
||||
|---|---|
|
||||
| Graph boundary | Same-column entry; graph-owned cross-column move; hold→WIP suspension; failure edge; backward remediation; columnless end |
|
||||
| Durable continuation | Runnable, running, held, succeeded, failed, cancelled, expired lease; duplicate claim; stale IR; missing target node |
|
||||
| Dispatch | Initial Todo planning completion; scheduler sweep; explicit hold promotion; restart recovery; duplicate executor callback |
|
||||
| Optional steps | Explicit enabled list; explicit empty list; absent list with `defaultOn:true`; absent list with `defaultOn:false` |
|
||||
| Task controls | User pause; hard-cancel move In-progress→Todo; soft delete; workflow reassignment; `autoMerge:false` |
|
||||
| Workflow source | Restored built-in; custom dashboard workflow; API import/update; `fn_workflow_create`/`fn_workflow_update`; existing task selection |
|
||||
| Capacity/data | Free/full WIP; dependencies met/unmet; overlap blocked; SQLite-free PostgreSQL path; restart before move/after move/before claim |
|
||||
| UI | Desktop and mobile workflow editor; create/copy/select preset; server validation error; node column/phase presentation |
|
||||
| Lifecycle result | APPROVE; REVISE within cap; replan/review cap exhausted; provider unavailable; merge success; manual merge wait; terminal failure |
|
||||
|
||||
### Dependencies
|
||||
|
||||
No new third-party dependency is required. This plan builds on the graph-owned lifecycle cutover in `docs/plans/2026-07-18-001-refactor-ir-driven-lifecycle-cutover-plan.md` and the existing `workflow_work_items`, workflow IR pin, move-task, scheduler capacity, and self-healing facilities.
|
||||
|
||||
---
|
||||
|
||||
## Planning Contract
|
||||
|
||||
### Key Technical Decisions
|
||||
|
||||
- KTD-1. **Column placement is an execution invariant.** `(session-settled: user-approved — chosen over treating columns as presentation metadata: the reported workflow must run each role in the lane where it was authored.)` `onNodeEntry` must return an entered-or-suspended decision, and the executor must not invoke a node handler after a suspended decision.
|
||||
- KTD-2. **The scheduler remains the only hold-to-WIP mover.** `(session-settled: user-approved — chosen over letting the graph bypass capacity arbitration: Todo must finish planning/review before the existing scheduler admits implementation.)` The graph produces readiness; the scheduler reserves capacity and commits the move; the graph resumes afterward.
|
||||
- KTD-3. **Use `workflow_work_items` as the durable continuation checkpoint.** A kind `task` item identifies the next node; `held` means waiting for planning or release, `runnable` means eligible to claim, and terminal states preserve idempotency. Add nullable stable-workflow-run ID, continuation sequence, wait reason, source-column, target-column, and IR-hash fields plus a database-enforced single-active-task-continuation constraint. Each visit gets a fresh item `runId`, while the stable workflow-run ID keeps graph/branch context coherent across rework visits; this avoids illegally reactivating a terminal item under the existing `(runId, taskId, nodeId, kind)` uniqueness rule. Existing merge/retry/manual-hold rows need no backfill. Do not repurpose `blockedReason`, `transitionPending` (move-hook recovery), or `workflowIrPin*` (drift evidence) as continuation storage.
|
||||
- KTD-4. **Resume at an explicit node, not by replaying from `start`.** Add a graph entry/resume mode that validates the target against the pinned IR and begins at the suspended node. Existing per-node result and branch/foreach state remain the source of truth inside completed regions; replay is a recovery fallback only when reconciliation proves that no side effect can repeat.
|
||||
- KTD-5. **A durable work-item loop, not triage or scheduler, runs graph nodes.** After triage persists `PROMPT.md`, it enqueues the initial kind-`task` continuation and returns, releasing its planning admission slot. The in-process runtime's workflow-work processor leases that item and invokes the executor-backed graph runtime. The same loop consumes the continuation after scheduler release. Triage never starts a nested reviewer, and the scheduler never invokes review merely to make its own release gate pass.
|
||||
- KTD-6. **Release readiness is generic continuation state, not a Plan Review special case.** Replace `isPlanReviewPreReleaseGateUnpassed` as the release authority with the presence of the correct held continuation plus the ordinary unplanned/status guards. Plan Review affects readiness by graph traversal and its outcome, so any future pre-WIP node gains the same behavior.
|
||||
- KTD-7. **One optional-group resolver governs every consumer.** Centralize explicit-list-versus-`defaultOn` resolution in `@fusion/core`; graph dispatch, release/readiness checks, preset seeding, dashboard display, and agent tools consume it. This follows `docs/solutions/logic-errors/optional-group-toggle-id-remapped-by-step-materializer.md`: group IDs remain identity-stable through real create and update paths.
|
||||
- KTD-8. **Restore, simplify, and do not duplicate the five-column preset.** `(session-settled: user-approved — chosen over changing `builtin:coding`, adding a near-duplicate, or carrying optional Browser/Post-merge Verification stages: the user asked for the smallest Ideas→plan→code→review/merge pipeline.)` Remove `builtin:coding-ideas` from the deprecation registry, retain its ID, prune Browser Verification and Post-merge Verification from this derived graph, and place Code Review plus merge in In-review while remediation remains in In-progress.
|
||||
- KTD-9. **Reject deterministic lifecycle deadlocks at the domain boundary.** `(session-settled: user-approved — chosen over saving invalid graphs with warnings: operators should not discover an impossible release topology after a task stalls.)` Shared IR validation must reject a capacity hold without exactly one downstream WIP target reachable under the runtime's release rules. The dashboard may add explanatory guidance, but API and agent saves must enforce the same error.
|
||||
- KTD-10. **Reconciliation repairs state, never weakens gates.** Startup/self-healing may recreate a missing continuation only from durable proof (`PROMPT.md`, node results, task column, IR pin, and no live execution). Ambiguity parks visibly. This follows `docs/solutions/logic-errors/mission-autopilot-stalled-by-stranded-done-feature.md`: pair strict gates with a bounded recovery path.
|
||||
- KTD-11. **All trigger gates receive the compatibility matrix.** `autoMerge`, dependency, pause, ephemeral-agent, overlap, capacity, soft-delete, and workflow reassignment checks apply before both pre-release and resumed dispatch. This follows `docs/solutions/logic-errors/per-task-auto-merge-override-ignored-by-trigger-gates.md`: changing the action path without every trigger path creates starvation or policy bypass.
|
||||
- KTD-12. **Planning rework is a distinct durable wait, not release readiness.** A Plan Review REVISE terminalizes the current visit and creates a held continuation with wait reason `planning` at the next review visit. Triage consumes that wait to revise `PROMPT.md`, then marks the successor runnable. Only a held continuation with wait reason `capacity` and a WIP target is eligible for scheduler release.
|
||||
|
||||
### High-Level Technical Design
|
||||
|
||||
The following state machine is a design constraint, not a prescribed method signature:
|
||||
|
||||
```mermaid
|
||||
stateDiagram-v2
|
||||
[*] --> PreReleaseRunnable: Todo planning persisted PROMPT.md
|
||||
PreReleaseRunnable --> PreReleaseRunnable: Todo node / review / replan
|
||||
PreReleaseRunnable --> HeldAtBoundary: next node is in WIP
|
||||
HeldAtBoundary --> HeldAtBoundary: capacity or other dispatch gate blocks
|
||||
HeldAtBoundary --> ReleasedRunnable: scheduler reserves slot and moves task
|
||||
ReleasedRunnable --> Running: continuation lease claimed
|
||||
Running --> Running: graph-owned column transitions and nodes
|
||||
Running --> HeldAtBoundary: later hold-to-WIP boundary
|
||||
Running --> Succeeded: graph reaches success terminal
|
||||
Running --> Failed: terminal node failure
|
||||
HeldAtBoundary --> Cancelled: pause, delete, workflow replacement, hard cancel
|
||||
ReleasedRunnable --> Cancelled: task invalidated before claim
|
||||
```
|
||||
|
||||
```mermaid
|
||||
sequenceDiagram
|
||||
participant T as Triage / planning
|
||||
participant P as Workflow work processor
|
||||
participant G as Graph runner
|
||||
participant W as Workflow work item
|
||||
participant S as Scheduler
|
||||
participant M as moveTask
|
||||
participant E as Executor
|
||||
|
||||
T->>W: enqueue initial runnable item after PROMPT.md commit
|
||||
T-->>T: return and release planning slot
|
||||
P->>W: lease runnable item
|
||||
P->>G: continue graph through executor-backed runtime
|
||||
G->>G: run Plan Review/replan in Todo
|
||||
G->>W: persist held continuation at first WIP node
|
||||
G-->>P: suspended (non-terminal)
|
||||
S->>W: verify held continuation + task gates
|
||||
S->>S: reserve capacity/worktree/semaphore
|
||||
S->>M: Todo -> In-progress
|
||||
M-->>S: committed
|
||||
S->>W: held -> runnable
|
||||
S->>P: wake task work processing
|
||||
P->>W: lease continuation
|
||||
P->>E: invoke executor-backed resume
|
||||
E->>G: resume at recorded node
|
||||
G->>G: assert card is in node.column, then run handler
|
||||
```
|
||||
|
||||
If the process fails before `moveTask` commits, the item remains held. If it fails after the move but before the item becomes runnable, reconciliation observes the committed WIP column plus the matching held target and advances it. If it fails after the item is runnable or running, normal lease expiry and claim idempotency recover it. No recovery path infers approval from column alone.
|
||||
|
||||
### System-Wide Impact
|
||||
|
||||
- **Domain and persistence:** work-item transitions become part of the main coding workflow, not only extension/merge support. PostgreSQL transactions and uniqueness constraints must preserve at most one active task continuation per task.
|
||||
- **Engine ownership:** triage gains a post-planning graph handoff; scheduler consumes readiness rather than Plan Review results; executor accepts an explicit continuation; self-healing audits stranded seams.
|
||||
- **Cancellation:** extend the existing targeted work-item cancellation seams used by hard-cancel and task lifecycle mutations so held/runnable task continuations are cancelled when appropriate without cancelling unrelated merge work or the scheduler-owned continuation during its intended release move.
|
||||
- **Observability:** add deduplicated events such as `task:workflow-suspended`, `task:workflow-release-ready`, `task:workflow-resumed`, and `task:reconcile-workflow-continuation`; metadata contains task/run/work-item/node/column IDs and outcomes only.
|
||||
- **Authoring parity:** `parseWorkflowIr` remains the domain authority. Dashboard routes, dry-run validation, and `fn_workflow_*` tools must expose its exact lifecycle error rather than reimplementing a client-only rule.
|
||||
- **Performance:** scheduler sweeps should batch continuation reads with task/workflow resolution and avoid a per-task query loop. Held items are expected steady state and must not produce per-poll info-log noise.
|
||||
|
||||
### Data Integrity and Commit Boundaries
|
||||
|
||||
| Invariant | Commit boundary | Recovery proof |
|
||||
|---|---|---|
|
||||
| At most one active kind-`task` continuation exists per task | Retire the prior active item and insert/update its successor in one store transaction, backed by a partial uniqueness constraint | Constraint violation loses the race and reloads the winner; it never launches work |
|
||||
| A target node never runs before its column move | Persist `held` before returning suspended; only the scheduler changes it to `runnable` after `moveTask` commits | Held + source column waits; held + target column is the moved-before-runnable repair case |
|
||||
| A runnable node runs at most once at a time | Claim with the existing state-and-lease compare-and-set before graph resume | Live lease suppresses duplicates; expired lease is reclaimable once |
|
||||
| Workflow edits cannot redirect a continuation | Store the resolved IR hash with the target and compare it with the live workflow before claim | Mismatch parks through the existing drift audit/clear path; no target substitution |
|
||||
| Operator invalidation wins over recovery | Pause/delete/reassignment cancellation and continuation transition share the task mutation boundary or use a post-write compare-and-set | Recovery reloads the task immediately before action and emits a no-action outcome when control state changed |
|
||||
| Rework can revisit a node without reviving terminal work | Give each visit a monotonic continuation sequence and fresh item `runId`, linked by one stable workflow-run ID | Unique sequence and single-active constraints reject duplicate successors; completed visits remain immutable |
|
||||
|
||||
The migration is additive: new work-item columns are nullable, existing rows remain valid, and only newly suspended coding runs create kind-`task` continuations. No data backfill is required. The schema change must include the Drizzle snapshot, the next PostgreSQL migration and journal entry, async/sync row mappers, and migration tests. Rollback safety is behavioral rather than destructive: do not drop or rewrite existing rows; an older binary must ignore the additive columns, while upgrade reconciliation can cancel orphaned task continuations created by an interrupted newer run.
|
||||
|
||||
### Sequencing
|
||||
|
||||
1. Land the shared contracts and continuation persistence before changing any trigger.
|
||||
2. Make graph suspension and explicit resume correct in isolation.
|
||||
3. Wire post-planning enqueue, the project runtime work loop, scheduler release, and executor-backed resume as one vertical slice.
|
||||
4. Add recovery/cancellation before exposing the preset.
|
||||
5. Restore the preset and authoring guidance only after the runtime can honor it.
|
||||
6. Replace the synthetic benchmark with real end-to-end acceptance and remove obsolete special cases.
|
||||
|
||||
### Implementation Constraints
|
||||
|
||||
- Preserve the workflow-policy / runtime-primitive / engine-substrate split described in `docs/solutions/architecture-patterns/workflow-native-runtime-primitives.md`.
|
||||
- Enforce node/column invariants at the shared graph boundary, not only in individual handlers, following `docs/solutions/logic-errors/repo-root-task-worktree-requeue-loop.md`.
|
||||
- User-configured commands remain async and supervised; no new synchronous shellouts or real polling waits.
|
||||
- New tests use fakes/fake timers and file-scoped fixtures; do not add real-network or slow full-suite tests.
|
||||
- New dashboard styles reuse existing workflow components and design tokens; no global CSS or hardcoded visual values.
|
||||
- Run-audit metadata follows the repository's IDs/counts/outcomes-only rule.
|
||||
|
||||
### Risks & Mitigations
|
||||
|
||||
- **Duplicate side effects on resume:** explicit target-node continuation plus work-item leases and uniqueness prevent start-node replay; kill-window tests cover every commit boundary.
|
||||
- **Task/work-item split-brain:** make move/release transitions transactional where the store permits; where cross-step commits are unavoidable, encode both intermediate states and reconcile them deterministically.
|
||||
- **Default workflow regression:** characterize `builtin:coding` separately and do not rehome its nodes or change its default selection.
|
||||
- **Stale workflow edits:** validate the saved IR identity at claim time and use the existing drift park/clear/re-resolve cycle instead of guessing a replacement node.
|
||||
- **Hidden dispatch bypass:** enumerate scheduler, triage, executor retry, heartbeat, self-healing, and explicit promotion entry points; route all through the same continuation claim/admission helper.
|
||||
- **Preset creates a new support burden:** retain the old ID and existing task compatibility, update its tests and description, and keep the workflow compositional rather than creating bespoke executor branches.
|
||||
|
||||
### Sources & Research
|
||||
|
||||
- `packages/engine/src/workflow-column-boundary.ts` — currently detects hold→WIP but returns `void`, so traversal cannot observe the park.
|
||||
- `packages/engine/src/workflow-graph-executor.ts` — currently calls `onNodeEntry` and immediately executes the node.
|
||||
- `packages/engine/src/hold-release.ts` — currently special-cases an explicitly enabled pre-release Plan Review result, creating the circular gate.
|
||||
- `packages/engine/src/__tests__/benchmark-six-column-workflow.test.ts` — documents that the graph does not suspend and uses a test-only scheduler wrapper.
|
||||
- `packages/core/src/types/merge-queue.ts` and `packages/core/src/postgres/schema/project.ts` — existing work-item states, leases, node identity, and uniqueness provide the durable continuation substrate.
|
||||
- `packages/core/src/builtin-coding-ideas-workflow-ir.ts` — existing five-column preset already supplies Ideas intake and Todo planning placement but leaves Code Review in In-progress.
|
||||
- `docs/plans/2026-07-18-001-refactor-ir-driven-lifecycle-cutover-plan.md` — origin design selected a scheduler-owned ready-for-release seam; this plan completes the deferred suspension/resume behavior.
|
||||
- [Fusion PR #2335](https://github.com/Runfusion/Fusion/pull/2335) — landed graph-owned lifecycle work while deferring actual suspension at the ready-for-release seam.
|
||||
|
||||
---
|
||||
|
||||
## Implementation Units
|
||||
|
||||
### U1. Define shared lifecycle and optional-step contracts
|
||||
|
||||
- **Goal:** Make suspension, continuation, and optional-group enablement first-class domain concepts used by every layer.
|
||||
- **Requirements:** R9, R10, R15, R16
|
||||
- **Dependencies:** none
|
||||
- **Files:** `packages/core/src/types/merge-queue.ts`, `packages/core/src/workflow-optional-steps.ts`, `packages/core/src/workflow-ir.ts`, `packages/core/src/workflow-lifecycle.ts` (new), `packages/core/src/postgres/schema/project.ts`, `packages/core/src/postgres/migrations/0031_workflow_task_continuations.sql` (new), `packages/core/src/postgres/migrations/meta/_journal.json`, `packages/core/src/task-store/row-types.ts`, `packages/core/src/task-store/task-row-mappers.ts`, `packages/core/src/task-store/async-workflow-workitems.ts`, `packages/core/src/task-store/workflow-workitems-ops-2.ts`, `packages/core/src/index.ts`, `packages/core/src/index.gate.ts`, `packages/core/src/__tests__/workflow-ir-validation.test.ts`, `packages/core/src/__tests__/workflow-optional-steps.test.ts`, `packages/core/src/__tests__/postgres/schema-applier.test.ts`
|
||||
- **Approach:** Define the continuation metadata/state transition contract around existing kind-`task` workflow work items; add the nullable metadata and active-item uniqueness constraint; add one pure optional-group enablement helper; add a pure lifecycle topology validator for capacity holds and downstream WIP targets. Keep validation in `parseWorkflowIr` so persistence, API, tools, and UI inherit it. Extend both async and compatibility row mappings together so no backend reads a partial shape.
|
||||
- **Test scenarios:** explicit `[]` overrides `defaultOn:true`; absent list enables `defaultOn:true`; capacity hold with zero or multiple ambiguous WIP release targets fails; the Ideas→Todo→In-progress topology passes; backward review-remediation edges remain legal; migration preserves existing non-task rows; two active task continuations for one task are rejected atomically.
|
||||
- **Verification:** targeted core Vitest files pass and public export gates include the shared helpers/types.
|
||||
|
||||
### U2. Persist, suspend, and explicitly resume graph traversal
|
||||
|
||||
- **Goal:** Stop traversal before a scheduler-owned boundary and resume once at the recorded node.
|
||||
- **Requirements:** R4, R9, R10, R12, R13
|
||||
- **Dependencies:** U1
|
||||
- **Files:** `packages/engine/src/workflow-column-boundary.ts`, `packages/engine/src/workflow-graph-executor.ts`, `packages/engine/src/workflow-graph-task-runner.ts`, `packages/engine/src/workflow-task-runtime.ts`, `packages/core/src/task-store/workflow-workitems-ops-2.ts`, `packages/core/src/task-store/async-workflow-workitems.ts`, `packages/engine/src/__tests__/workflow-graph-column-moves.test.ts`, `packages/engine/src/__tests__/workflow-graph-task-runner.test.ts`, `packages/engine/src/__tests__/workflow-task-runtime.test.ts`
|
||||
- **Approach:** Return a typed boundary decision; propagate a suspended disposition without treating it as success/failure; upsert the held continuation before returning; add explicit graph entry at a validated target node; claim/terminalize the work item around resumed traversal. Keep `workflowIrPin*` drift checks and `transitionPending` move recovery independent.
|
||||
- **Test scenarios:** target handler is not called on suspension; a same-column chain does not suspend; resume starts at target without replaying earlier prompt/review handlers; duplicate resume loses the lease; missing/rehomed target parks via drift; node failure terminalizes the claimed item without fabricating completion.
|
||||
- **Verification:** targeted graph, runner, runtime, and work-item tests prove the state machine and at-most-once claim behavior.
|
||||
|
||||
### U3. Connect Todo planning, scheduler release, and executor resume
|
||||
|
||||
- **Goal:** Run pre-release graph nodes in Todo and hand approved work through the real capacity scheduler into In-progress.
|
||||
- **Requirements:** R2, R3, R4, R11, R12, R15
|
||||
- **Dependencies:** U2
|
||||
- **Files:** `packages/engine/src/triage.ts`, `packages/engine/src/runtimes/in-process-runtime.ts`, `packages/engine/src/workflow-work-processor.ts`, `packages/engine/src/workflow-work-scheduler.ts`, `packages/engine/src/scheduler.ts`, `packages/engine/src/hold-release.ts`, `packages/engine/src/executor.ts`, `packages/engine/src/workflow-planning-service.ts`, `packages/engine/src/__tests__/plan-review-single-owner.test.ts`, `packages/engine/src/__tests__/scheduler-trait-dispatch.test.ts`, `packages/engine/src/__tests__/workflow-graph-optional-group.test.ts`, `packages/engine/src/__tests__/workflow-work-processor.test.ts`, `packages/engine/src/__tests__/workflow-work-scheduler.test.ts`
|
||||
- **Approach:** After the planning artifact commits, enqueue the initial continuation and let triage release its semaphore slot. Wire a bounded project-runtime processor for kind-`task` items that delegates leased work to the executor-backed graph runtime, so it reuses the real planning/review/implementation/merge seams rather than the single-node test facade. A REVISE outcome creates a planning-wait successor that triage advances only after rewriting `PROMPT.md`; an approved path creates a capacity-wait successor at the first WIP node. Have every release surface require and advance only the matching capacity-wait item; wake the processor only after the move succeeds. Remove the Plan-Review-specific release gate after generic readiness covers it. Preserve admission checks, pre-held slot transfer, and cleanup on every early return.
|
||||
- **Test scenarios:** planning completion enqueues but does not synchronously run review under the triage slot; APPROVE with free/full capacity; REVISE then APPROVE in Todo; explicit disabled review; absent toggle list; move rejection leaves item held; move success then dispatch-metadata failure remains recoverable; duplicate scheduler/processor polls produce one claim and one move.
|
||||
- **Verification:** targeted triage, hold-release, scheduler, optional-group, and executor dispatch tests pass with no test-only boundary wrapper.
|
||||
|
||||
### U4. Reconcile interruptions and honor operator control
|
||||
|
||||
- **Goal:** Make every suspension/release crash window recoverable without bypassing human or lifecycle controls.
|
||||
- **Requirements:** R13, R14, R20
|
||||
- **Dependencies:** U3
|
||||
- **Files:** `packages/engine/src/self-healing.ts`, `packages/engine/src/active-session-registry.ts`, `packages/core/src/task-store/moves.ts`, `packages/core/src/task-store/remaining-ops-2.ts`, `packages/engine/src/__tests__/workflow-graph-paused-node-resume.test.ts`, `packages/engine/src/__tests__/workflow-ir-pin-wiring.test.ts`, `packages/engine/src/__tests__/workflow-merge-cancellation.test.ts`, `packages/engine/src/__tests__/self-healing.test.ts`
|
||||
- **Approach:** Add bounded reconciliation for held-in-Todo, moved-but-held, runnable-with-expired-lease, and stale-target states. Cancellation/reassignment paths cancel only matching active task continuations. Apply `userPaused`, hard-cancel, soft-delete, dependency, and `autoMerge:false` guards before any repair action; emit deduplicated audit events for action and no-action outcomes.
|
||||
- **Test scenarios:** restart before move; restart after move before runnable transition; expired running lease; user pause at each state; In-progress→Todo hard cancel; soft-delete race; workflow replacement invalidates old run; `autoMerge:false` in-review task is untouched.
|
||||
- **Verification:** targeted recovery/cancellation tests use fake time and assert both state and audit metadata.
|
||||
|
||||
### U5. Restore the simple five-column coding preset
|
||||
|
||||
- **Goal:** Offer the workflow the user expected without changing default coding projects.
|
||||
- **Requirements:** R1-R8, R18, R19
|
||||
- **Dependencies:** U3, U4
|
||||
- **Files:** `packages/core/src/builtin-coding-ideas-workflow-ir.ts`, `packages/core/src/builtin-workflows.ts`, `packages/core/src/types.ts`, `packages/core/src/__tests__/builtin-coding-ideas-workflow-ir.test.ts`, `packages/core/src/__tests__/builtin-workflows.test.ts`, `packages/core/src/__tests__/store-create-intake-column.test.ts`, `packages/engine/src/__tests__/builtin-workflows-lifecycle.test.ts`
|
||||
- **Approach:** Remove only `builtin:coding-ideas` from the deprecated set; retain its identity and Ideas/Todo traits; derive a minimal graph that omits Browser Verification and Post-merge Verification; place Plan Review/replan in Todo, parse/foreach execution in In-progress, Code Review/completion summary/merge in In-review, Code Review remediation in In-progress, and only the terminal node in Done. Keep `builtin:coding` unchanged and update preset copy to describe the actual lanes.
|
||||
- **Test scenarios:** preset catalog visibility/default-enabled policy; existing selected task resolves unchanged; no AI in Ideas; plan/review in Todo; implementation-only In-progress; Code Review REVISE round trip; merge/manual wait in In-review; Done terminal; default coding characterization unchanged.
|
||||
- **Verification:** targeted core and engine built-in lifecycle tests pass.
|
||||
|
||||
### U6. Make authoring feedback truthful across UI, API, and tools
|
||||
|
||||
- **Goal:** Prevent authors from saving undriveable graphs and explain the runtime meaning of lane placement.
|
||||
- **Requirements:** R16, R17, R19
|
||||
- **Dependencies:** U1, U5
|
||||
- **Files:** `packages/dashboard/src/routes/register-workflow-routes.ts`, `packages/dashboard/src/routes/__tests__/workflow-validate-route.test.ts`, `packages/dashboard/app/components/WorkflowNodeEditor.tsx`, `packages/dashboard/app/components/WorkflowColumnPanel.tsx`, `packages/dashboard/app/components/workflow-lifecycle-autofix.ts`, `packages/dashboard/app/components/WorkflowNodeEditor.css`, `packages/dashboard/app/components/__tests__/WorkflowNodeEditor.test.tsx`, `packages/dashboard/app/components/__tests__/WorkflowColumnPanel.test.tsx`, `packages/dashboard/app/components/__tests__/workflow-lifecycle-autofix.test.ts`, `packages/engine/src/agent-tools.ts`, `packages/engine/src/__tests__/agent-workflow-tools-exposure.test.ts`
|
||||
- **Approach:** Map shared lifecycle validation errors into the existing save/dry-run response and lifecycle-warning system; display them at the affected hold/column; add concise placement/release guidance using existing components and tokens; ensure create/copy/select surfaces offer the restored preset. Agent tools must call the same validator and return the same actionable reason. Do not add a second validation panel or a new canvas interaction mode.
|
||||
- **Test scenarios:** desktop and mobile invalid-save feedback; valid five-column save; copied built-in stays editable; preset selectable; API and agent create/update reject the same invalid topology; no empty control shells or orphaned labels after UI changes.
|
||||
- **Verification:** targeted dashboard route/component tests and agent-tool parity tests pass; inspect both desktop and mobile layouts.
|
||||
|
||||
### U7. Replace the synthetic benchmark with end-to-end lifecycle proof
|
||||
|
||||
- **Goal:** Prove the original symptom and all lane boundaries through production orchestration.
|
||||
- **Requirements:** R1-R20; F1-F7; AE1-AE10
|
||||
- **Dependencies:** U3-U6
|
||||
- **Files:** `packages/engine/src/__tests__/benchmark-six-column-workflow.test.ts`, `packages/engine/src/__tests__/builtin-workflows-lifecycle.test.ts`, `packages/engine/src/__tests__/workflow-work-scheduler.test.ts`, `packages/dashboard/app/components/__tests__/board-quickcreate-workflow-lane-visibility.test.tsx`, `docs/workflow-editor.md`, `.changeset/truthful-workflow-lifecycle.md` (new)
|
||||
- **Approach:** Delete the benchmark's fake scheduler boundary wrapper and drive real triage/graph/scheduler/executor seams with deterministic fakes. Assert task column at every handler start, persisted continuation transitions, session uniqueness, and restart recovery. Update operator docs with the five-column preset and authoring rules. Add a minor `@runfusion/fusion` changeset because a supported preset becomes newly selectable and workflow runtime behavior changes.
|
||||
- **Test scenarios:** full happy path; both review rework loops; capacity wait; optional-step matrix; restart windows; manual merge; invalid topology; default workflow non-regression.
|
||||
- **Verification:** all scenario assertions pass without real waits/network calls; the required changeset passes strict format validation.
|
||||
|
||||
---
|
||||
|
||||
## Verification Contract
|
||||
|
||||
Run the smallest affected test files after each unit. Before handoff, run the union of changed-file tests with package-scoped Vitest commands, then:
|
||||
|
||||
- `pnpm check:changesets --strict` — the published-package changeset has valid labeled fields.
|
||||
- `pnpm lint` — workflow, dashboard, and test changes meet repository lint rules.
|
||||
- `pnpm verify:fast` — changed packages typecheck/build and the CLI boot smoke passes without running the full suite.
|
||||
- `pnpm test:gate` — the trusted merge gate remains green.
|
||||
|
||||
Behavioral verification must prove, from persisted task/work-item/session evidence rather than log text alone:
|
||||
|
||||
- No node handler starts before `task.column` equals the node's assigned column.
|
||||
- A held continuation prevents downstream calls and survives restart.
|
||||
- A committed scheduler move precedes runnable transition and resumed execution.
|
||||
- Plan Review placement/toggles cannot deadlock release.
|
||||
- Code Review remediation visibly crosses In-review → In-progress → In-review.
|
||||
- `autoMerge:false`, pause, delete, dependency, capacity, and workflow-drift gates remain authoritative.
|
||||
- Dashboard, API, import/dry-run, and agent tools accept/reject the same workflow shapes.
|
||||
|
||||
Do not run `pnpm test:full` or `pnpm verify:workspace` for this task unless a genuinely untargetable cross-workspace failure requires it. If a test flakes without a corresponding product bug, follow the quarantine-on-sight rule rather than adding retries or timeouts.
|
||||
|
||||
---
|
||||
|
||||
## Definition of Done
|
||||
|
||||
- R1-R20 and AE1-AE10 are traced to passing automated coverage.
|
||||
- The original reported workflow can be selected or authored and completes Ideas → Todo → In-progress → In-review → Done with the promised work in each lane.
|
||||
- The graph returns a durable suspended state at hold-to-WIP boundaries and cannot execute the target node early.
|
||||
- Scheduler release and executor resume are idempotent across all tested crash windows.
|
||||
- Optional-group enablement is shared and consistent for explicit, empty, and absent task settings.
|
||||
- Invalid capacity-hold topologies cannot be persisted through any supported authoring surface.
|
||||
- `builtin:coding` compatibility tests show no behavioral migration; existing `builtin:coding-ideas` selections remain valid and the preset is selectable again.
|
||||
- Audit metadata contains no prompt, review prose, model IDs, or other forbidden free text.
|
||||
- Dashboard changes reuse existing components/tokens and pass desktop/mobile surface checks.
|
||||
- Scoped tests, strict changeset validation, lint, `pnpm verify:fast`, and `pnpm test:gate` pass.
|
||||
- The diff contains no abandoned continuation fields, duplicate Plan Review gate, synthetic scheduler wrapper, dead helper, stale copy, or experimental code from rejected approaches.
|
||||
- No release or publish command has been run.
|
||||
@@ -92,6 +92,20 @@ describe("builtin coding-ideas workflow ir", () => {
|
||||
expect(codeReview?.config?.defaultOn).toBe(true);
|
||||
});
|
||||
|
||||
it("keeps each activity in the column named for that activity", () => {
|
||||
const ir = BUILTIN_CODING_IDEAS_WORKFLOW_IR as WorkflowIrV2;
|
||||
const nodeColumn = (id: string) => ir.nodes.find((node) => node.id === id)?.column;
|
||||
expect(nodeColumn("plan")).toBe("todo");
|
||||
expect(nodeColumn("plan-review")).toBe("todo");
|
||||
expect(nodeColumn("parse")).toBe("in-progress");
|
||||
expect(nodeColumn("steps")).toBe("in-progress");
|
||||
expect(nodeColumn("code-review")).toBe("in-review");
|
||||
expect(nodeColumn("completion-summary")).toBe("in-review");
|
||||
expect(nodeColumn("merge-gate")).toBe("in-review");
|
||||
expect(ir.nodes.some((node) => node.id === "browser-verification")).toBe(false);
|
||||
expect(ir.nodes.some((node) => node.id === "post-merge-verification")).toBe(false);
|
||||
});
|
||||
|
||||
it("never leaves a node in a column the workflow does not declare", () => {
|
||||
const ir = BUILTIN_CODING_IDEAS_WORKFLOW_IR as WorkflowIrV2;
|
||||
const declared = new Set(ir.columns.map((c) => c.id));
|
||||
|
||||
@@ -177,6 +177,7 @@ describe("built-in workflows", () => {
|
||||
it("merge-capable built-ins expose a default-off post-merge verification node after merge proof", () => {
|
||||
for (const workflow of BUILTIN_WORKFLOWS) {
|
||||
if (workflow.kind === "fragment") continue;
|
||||
if (workflow.id === "builtin:coding-ideas") continue; // intentionally minimal pipeline
|
||||
const mergeNode = workflow.ir.nodes.find((node) => node.id === "merge-attempt" || node.id === "merge");
|
||||
if (!mergeNode) continue;
|
||||
|
||||
@@ -357,13 +358,13 @@ describe("built-in workflows", () => {
|
||||
);
|
||||
});
|
||||
|
||||
it("keeps deprecated builtin:coding-ideas resolvable while excluding it from defaults", () => {
|
||||
it("keeps builtin:coding-ideas selectable as the simple five-stage pipeline", () => {
|
||||
const codingIdeas = getBuiltinWorkflow("builtin:coding-ideas");
|
||||
expect(codingIdeas).toBeDefined();
|
||||
expect(codingIdeas!.kind).toBe("workflow");
|
||||
expect(() => parseWorkflowIr(codingIdeas!.ir)).not.toThrow();
|
||||
expect(isBuiltinWorkflowDeprecated("builtin:coding-ideas")).toBe(true);
|
||||
expect(defaultEnabledBuiltinWorkflowIds()).not.toContain("builtin:coding-ideas");
|
||||
expect(isBuiltinWorkflowDeprecated("builtin:coding-ideas")).toBe(false);
|
||||
expect(defaultEnabledBuiltinWorkflowIds()).toContain("builtin:coding-ideas");
|
||||
});
|
||||
|
||||
it("orders builtin:brainstorming's ask-user/exit-gate loop ahead of the plan/execute spine", () => {
|
||||
@@ -745,16 +746,16 @@ describe("built-in workflows", () => {
|
||||
expect(defaultEnabledBuiltinWorkflowIds()).toContain("builtin:marketing");
|
||||
expect(defaultEnabledBuiltinWorkflowIds()).not.toContain("builtin:compound-engineering");
|
||||
expect(defaultEnabledBuiltinWorkflowIds()).not.toContain("builtin:brainstorming");
|
||||
expect(defaultEnabledBuiltinWorkflowIds()).not.toContain("builtin:coding-ideas");
|
||||
expect(defaultEnabledBuiltinWorkflowIds()).toContain("builtin:coding-ideas");
|
||||
expect(defaultEnabledBuiltinWorkflowIds()).not.toContain("builtin:pr-workflow");
|
||||
expect(getBuiltinWorkflow("builtin:pr-workflow")!.kind).toBe("fragment");
|
||||
expect(defaultEnabledBuiltinWorkflowIds().length).toBeGreaterThanOrEqual(5);
|
||||
expect(defaultEnabledBuiltinWorkflowIds().slice(0, 5)).toEqual([
|
||||
"builtin:coding",
|
||||
"builtin:coding-ideas",
|
||||
"builtin:legacy-coding",
|
||||
"builtin:quick-fix",
|
||||
"builtin:review-heavy",
|
||||
"builtin:marketing",
|
||||
]);
|
||||
expect(defaultEnabledBuiltinWorkflowIds()).toContain("builtin:stepwise-coding");
|
||||
});
|
||||
@@ -764,7 +765,7 @@ describe("built-in workflows", () => {
|
||||
expect(isBuiltinWorkflowPluginGated("builtin:coding")).toBe(false);
|
||||
expect(isBuiltinWorkflowPluginGated("builtin:quick-fix")).toBe(false);
|
||||
expect(isBuiltinWorkflowDeprecated("builtin:brainstorming")).toBe(true);
|
||||
expect(isBuiltinWorkflowDeprecated("builtin:coding-ideas")).toBe(true);
|
||||
expect(isBuiltinWorkflowDeprecated("builtin:coding-ideas")).toBe(false);
|
||||
expect(isBuiltinWorkflowDeprecated("builtin:coding")).toBe(false);
|
||||
});
|
||||
|
||||
@@ -1069,7 +1070,7 @@ describe("built-in workflows", () => {
|
||||
});
|
||||
|
||||
it("hides deprecated built-ins from selection listings while preserving management and direct resolution", async () => {
|
||||
const deprecatedIds = ["builtin:brainstorming", "builtin:coding-ideas"];
|
||||
const deprecatedIds = ["builtin:brainstorming"];
|
||||
const selectionList = await store.listWorkflowDefinitions();
|
||||
for (const id of deprecatedIds) {
|
||||
expect(selectionList.some((workflow) => workflow.id === id)).toBe(false);
|
||||
|
||||
@@ -67,4 +67,22 @@ pgDescribe("Coding (Ideas) custom-column moves (workflow-columns graduation)", (
|
||||
const moved = await store.moveTask(task.id, "todo", { moveSource: "engine" });
|
||||
expect(moved.column).toBe("todo");
|
||||
});
|
||||
|
||||
it("cancels an active task continuation when a user sends implementation back to todo", async () => {
|
||||
const store = harness.store();
|
||||
const task = await store.createTask({ description: "idea", workflowId: "builtin:coding-ideas" });
|
||||
await store.moveTask(task.id, "todo", { moveSource: "user" });
|
||||
await store.moveTask(task.id, "in-progress", { moveSource: "user" });
|
||||
const continuation = await store.upsertWorkflowWorkItem({
|
||||
runId: `${task.id}:continuation:test`,
|
||||
taskId: task.id,
|
||||
nodeId: "steps",
|
||||
kind: "task",
|
||||
state: "running",
|
||||
});
|
||||
|
||||
await store.moveTask(task.id, "todo", { moveSource: "user" });
|
||||
|
||||
expect((await store.getWorkflowWorkItem(continuation.id))?.state).toBe("cancelled");
|
||||
});
|
||||
});
|
||||
|
||||
@@ -74,6 +74,7 @@ import {
|
||||
TASK_DECLARED_SYMBOLS_VERSION,
|
||||
PLANNING_ACTIVE_TIMING_VERSION,
|
||||
SQLITE_MIGRATION_RUNTIME_READ_VERSION,
|
||||
WORKFLOW_TASK_CONTINUATIONS_VERSION,
|
||||
} from "../../postgres/schema-applier.js";
|
||||
import { ProjectPartitionRekeyError, rekeyFallbackProjectPartition } from "../../postgres/migration-stamping.js";
|
||||
import type { PluginSchemaInitHook } from "../../postgres/plugin-schema-hook.js";
|
||||
@@ -1580,6 +1581,7 @@ pgDescribe("schema-applier: automation project-isolation upgrade", () => {
|
||||
TASK_DECLARED_SYMBOLS_VERSION,
|
||||
PLANNING_ACTIVE_TIMING_VERSION,
|
||||
SQLITE_MIGRATION_RUNTIME_READ_VERSION,
|
||||
WORKFLOW_TASK_CONTINUATIONS_VERSION,
|
||||
]);
|
||||
expect((await applySchemaBaseline(ctx.db, { pluginHooks: [] })).applied).toBe(false);
|
||||
});
|
||||
@@ -1636,6 +1638,7 @@ pgDescribe("schema-applier: automation project-isolation upgrade", () => {
|
||||
TASK_DECLARED_SYMBOLS_VERSION,
|
||||
PLANNING_ACTIVE_TIMING_VERSION,
|
||||
SQLITE_MIGRATION_RUNTIME_READ_VERSION,
|
||||
WORKFLOW_TASK_CONTINUATIONS_VERSION,
|
||||
]);
|
||||
});
|
||||
|
||||
@@ -1825,6 +1828,7 @@ pgDescribe("schema-applier: automation project-isolation upgrade", () => {
|
||||
TASK_DECLARED_SYMBOLS_VERSION,
|
||||
PLANNING_ACTIVE_TIMING_VERSION,
|
||||
SQLITE_MIGRATION_RUNTIME_READ_VERSION,
|
||||
WORKFLOW_TASK_CONTINUATIONS_VERSION,
|
||||
]);
|
||||
});
|
||||
|
||||
@@ -1895,6 +1899,7 @@ pgDescribe("schema-applier: automation project-isolation upgrade", () => {
|
||||
TASK_DECLARED_SYMBOLS_VERSION,
|
||||
PLANNING_ACTIVE_TIMING_VERSION,
|
||||
SQLITE_MIGRATION_RUNTIME_READ_VERSION,
|
||||
WORKFLOW_TASK_CONTINUATIONS_VERSION,
|
||||
]);
|
||||
});
|
||||
|
||||
@@ -1965,6 +1970,7 @@ pgDescribe("schema-applier: automation project-isolation upgrade", () => {
|
||||
TASK_DECLARED_SYMBOLS_VERSION,
|
||||
PLANNING_ACTIVE_TIMING_VERSION,
|
||||
SQLITE_MIGRATION_RUNTIME_READ_VERSION,
|
||||
WORKFLOW_TASK_CONTINUATIONS_VERSION,
|
||||
]);
|
||||
});
|
||||
});
|
||||
|
||||
@@ -37,6 +37,53 @@ pgTest("TaskStore.listTasks facade (PostgreSQL)", () => {
|
||||
expect(tasks).toEqual([]);
|
||||
});
|
||||
|
||||
it("atomically replaces the sole active task workflow continuation", async () => {
|
||||
const store = h.store();
|
||||
const task = await store.createTask({ description: "continuation owner", column: "todo" });
|
||||
const first = await store.replaceActiveTaskWorkflowContinuation({
|
||||
runId: `${task.id}:continuation:0`,
|
||||
taskId: task.id,
|
||||
nodeId: "plan-review",
|
||||
kind: "task",
|
||||
state: "runnable",
|
||||
stableWorkflowRunId: `${task.id}:workflow`,
|
||||
continuationSequence: 0,
|
||||
waitReason: "planning",
|
||||
sourceColumn: "todo",
|
||||
targetColumn: "todo",
|
||||
irHash: "ir-v1",
|
||||
});
|
||||
const second = await store.replaceActiveTaskWorkflowContinuation({
|
||||
runId: `${task.id}:continuation:1`,
|
||||
taskId: task.id,
|
||||
nodeId: "parse",
|
||||
kind: "task",
|
||||
state: "held",
|
||||
stableWorkflowRunId: `${task.id}:workflow`,
|
||||
continuationSequence: 1,
|
||||
waitReason: "capacity",
|
||||
sourceColumn: "todo",
|
||||
targetColumn: "in-progress",
|
||||
irHash: "ir-v1",
|
||||
});
|
||||
|
||||
expect((await store.getWorkflowWorkItem(first.id))?.state).toBe("succeeded");
|
||||
expect(second).toMatchObject({
|
||||
state: "held",
|
||||
waitReason: "capacity",
|
||||
sourceColumn: "todo",
|
||||
targetColumn: "in-progress",
|
||||
continuationSequence: 1,
|
||||
});
|
||||
await expect(store.upsertWorkflowWorkItem({
|
||||
runId: `${task.id}:continuation:2`,
|
||||
taskId: task.id,
|
||||
nodeId: "steps",
|
||||
kind: "task",
|
||||
state: "runnable",
|
||||
})).rejects.toThrow();
|
||||
});
|
||||
|
||||
it("returns all live tasks sorted by createdAt then numeric id suffix", async () => {
|
||||
const store = h.store();
|
||||
// Seed with explicit ascending timestamps so ordering is deterministic.
|
||||
|
||||
@@ -24,6 +24,30 @@ import { computeRemovedOccupiedColumns } from "../workflow-reconciliation.js";
|
||||
|
||||
// ── Save-time hard errors ─────────────────────────────────────────────────────
|
||||
|
||||
describe("workflow IR validation — capacity release topology (hard error)", () => {
|
||||
it("rejects a capacity hold that has no downstream processing column", () => {
|
||||
const ir: WorkflowIr = {
|
||||
version: "v2",
|
||||
name: "deadlocked-capacity-hold",
|
||||
columns: [
|
||||
{ id: "ideas", name: "Ideas", traits: [{ trait: "intake" }] },
|
||||
{ id: "todo", name: "Todo", traits: [{ trait: "hold", config: { release: "capacity" } }] },
|
||||
{ id: "done", name: "Done", traits: [{ trait: "complete" }] },
|
||||
],
|
||||
nodes: [
|
||||
{ id: "start", kind: "start", column: "ideas" },
|
||||
{ id: "plan", kind: "prompt", column: "todo", config: { prompt: "plan" } },
|
||||
{ id: "end", kind: "end", column: "done" },
|
||||
],
|
||||
edges: [
|
||||
{ from: "start", to: "plan" },
|
||||
{ from: "plan", to: "end", condition: "success" },
|
||||
],
|
||||
};
|
||||
expect(() => parseWorkflowIr(ir)).toThrow("capacity hold column 'todo' requires a downstream wip column");
|
||||
});
|
||||
});
|
||||
|
||||
describe("workflow IR validation — node → nonexistent column (hard error)", () => {
|
||||
it("rejects a node assigned to a column the workflow does not declare", () => {
|
||||
const ir: WorkflowIr = {
|
||||
|
||||
@@ -81,11 +81,36 @@ const RAW_BUILTIN_CODING_IDEAS_WORKFLOW_IR: WorkflowIr = (() => {
|
||||
node.column = "todo";
|
||||
continue;
|
||||
}
|
||||
if (node.id === "code-review" || node.id === "completion-summary" || node.id.startsWith("merge-")) {
|
||||
node.column = "in-review";
|
||||
continue;
|
||||
}
|
||||
if (node.id === "code-review-remediation") {
|
||||
node.column = "in-progress";
|
||||
continue;
|
||||
}
|
||||
if (!node.column || !knownColumnIds.has(node.column)) {
|
||||
node.column = "todo";
|
||||
}
|
||||
}
|
||||
|
||||
// Keep this preset intentionally small: planning + plan review in Todo,
|
||||
// implementation in In progress, then code review and merge in In review.
|
||||
// Browser and post-merge verification remain available in richer workflows.
|
||||
const removedNodeIds = new Set([
|
||||
"browser-verification",
|
||||
"browser-verification-remediation",
|
||||
"post-merge-verification",
|
||||
]);
|
||||
v2.nodes = v2.nodes.filter((node) => !removedNodeIds.has(node.id));
|
||||
v2.edges = v2.edges.filter((edge) => !removedNodeIds.has(edge.from) && !removedNodeIds.has(edge.to));
|
||||
if (!v2.edges.some((edge) => edge.from === "steps" && edge.to === "code-review")) {
|
||||
v2.edges.push({ from: "steps", to: "code-review", condition: "success" });
|
||||
}
|
||||
if (!v2.edges.some((edge) => edge.from === "merge-attempt" && edge.to === "end" && edge.condition === "success")) {
|
||||
v2.edges.push({ from: "merge-attempt", to: "end", condition: "success" });
|
||||
}
|
||||
|
||||
v2.settings = BUILTIN_WORKFLOW_SETTINGS;
|
||||
return ir;
|
||||
})();
|
||||
|
||||
@@ -252,6 +252,7 @@ export { BUILTIN_MARKETING_WORKFLOW_IR } from "./builtin-marketing-workflow-ir.j
|
||||
export {
|
||||
resolveWorkflowOptionalSteps,
|
||||
resolveDefaultOnOptionalGroupIds,
|
||||
isWorkflowOptionalGroupEnabled,
|
||||
} from "./workflow-optional-steps.js";
|
||||
export type { ResolvedWorkflowOptionalStep } from "./workflow-optional-steps.js";
|
||||
export {
|
||||
|
||||
@@ -282,6 +282,7 @@ export { BUILTIN_MARKETING_WORKFLOW_IR } from "./builtin-marketing-workflow-ir.j
|
||||
export {
|
||||
resolveWorkflowOptionalSteps,
|
||||
resolveDefaultOnOptionalGroupIds,
|
||||
isWorkflowOptionalGroupEnabled,
|
||||
} from "./workflow-optional-steps.js";
|
||||
export type { ResolvedWorkflowOptionalStep } from "./workflow-optional-steps.js";
|
||||
export {
|
||||
|
||||
@@ -545,6 +545,12 @@ CREATE TABLE IF NOT EXISTS project.workflow_work_items (
|
||||
lease_expires_at text,
|
||||
last_error text,
|
||||
blocked_reason text,
|
||||
stable_workflow_run_id text,
|
||||
continuation_sequence integer,
|
||||
wait_reason text,
|
||||
source_column text,
|
||||
target_column text,
|
||||
ir_hash text,
|
||||
created_at text NOT NULL,
|
||||
updated_at text NOT NULL,
|
||||
CONSTRAINT workflow_work_items_task_id_fkey
|
||||
@@ -2028,4 +2034,3 @@ CREATE INDEX IF NOT EXISTS "idxArchivedTasksCreatedAt"
|
||||
-- GIN index on the archive search_vector (VAL-SEARCH-005).
|
||||
CREATE INDEX IF NOT EXISTS "idxArchivedTasksSearchVector"
|
||||
ON archive.archived_tasks USING gin(search_vector);
|
||||
|
||||
|
||||
@@ -0,0 +1,32 @@
|
||||
ALTER TABLE project.workflow_work_items
|
||||
ADD COLUMN IF NOT EXISTS stable_workflow_run_id text,
|
||||
ADD COLUMN IF NOT EXISTS continuation_sequence integer,
|
||||
ADD COLUMN IF NOT EXISTS wait_reason text,
|
||||
ADD COLUMN IF NOT EXISTS source_column text,
|
||||
ADD COLUMN IF NOT EXISTS target_column text,
|
||||
ADD COLUMN IF NOT EXISTS ir_hash text;
|
||||
|
||||
-- Older builds could leave more than one active task work item. Preserve the
|
||||
-- newest continuation and retire the rest before enforcing single ownership.
|
||||
WITH ranked AS (
|
||||
SELECT project_id, id,
|
||||
row_number() OVER (
|
||||
PARTITION BY project_id, task_id
|
||||
ORDER BY updated_at DESC, id DESC
|
||||
) AS active_rank
|
||||
FROM project.workflow_work_items
|
||||
WHERE kind = 'task' AND state IN ('runnable', 'running', 'held', 'retrying')
|
||||
)
|
||||
UPDATE project.workflow_work_items AS item
|
||||
SET state = 'succeeded',
|
||||
lease_owner = NULL,
|
||||
lease_expires_at = NULL,
|
||||
updated_at = now()::text
|
||||
FROM ranked
|
||||
WHERE item.project_id = ranked.project_id
|
||||
AND item.id = ranked.id
|
||||
AND ranked.active_rank > 1;
|
||||
|
||||
CREATE UNIQUE INDEX IF NOT EXISTS idx_workflow_work_items_one_active_task_continuation
|
||||
ON project.workflow_work_items(project_id, task_id)
|
||||
WHERE kind = 'task' AND state IN ('runnable', 'running', 'held', 'retrying');
|
||||
@@ -37,11 +37,11 @@ FNXC:PostgresBigintCounters 2026-07-19-12:00:
|
||||
SCHEMA_BASELINE_VERSION advances to 0026 for the bigint counters migration.
|
||||
Per-migration identities above stay fixed; only this latest-version marker moves.
|
||||
|
||||
FNXC:MigrationStatusRuntimeRead 2026-07-20:
|
||||
SCHEMA_BASELINE_VERSION advances to 0030 for project-scoped runtime reads of
|
||||
the SQLite cutover ledger.
|
||||
FNXC:WorkflowTaskContinuations 2026-07-21:
|
||||
SCHEMA_BASELINE_VERSION advances to 0031 for durable, single-owner task
|
||||
continuations at workflow column boundaries.
|
||||
*/
|
||||
export const SCHEMA_BASELINE_VERSION = "0030";
|
||||
export const SCHEMA_BASELINE_VERSION = "0031";
|
||||
/** FNXC:SymbolLock 2026-07-31-10:00: upgrades need durable task declarations before admission resolves symbols. */
|
||||
export const TASK_DECLARED_SYMBOLS_VERSION = "0028";
|
||||
const INITIAL_SCHEMA_VERSION = "0000";
|
||||
@@ -132,6 +132,7 @@ export const BIGINT_COUNTERS_VERSION = "0026";
|
||||
export const PLANNING_ACTIVE_TIMING_VERSION = "0029";
|
||||
/** Dashboard health needs project-scoped, read-only runtime access to the SQLite cutover ledger. */
|
||||
export const SQLITE_MIGRATION_RUNTIME_READ_VERSION = "0030";
|
||||
export const WORKFLOW_TASK_CONTINUATIONS_VERSION = "0031";
|
||||
|
||||
/**
|
||||
* Thrown when the database was migrated by a NEWER Fusion binary than the one now
|
||||
@@ -324,6 +325,7 @@ const WORKFLOW_IR_PIN_AND_LEGACY_ADOPTION_MIGRATION_PATH = join(
|
||||
const PLANNING_ACTIVE_TIMING_MIGRATION_PATH = join(MIGRATIONS_DIR, "0029_planning_active_timing.sql");
|
||||
const TASK_DECLARED_SYMBOLS_MIGRATION_PATH = join(MIGRATIONS_DIR, "0028_task_declared_symbols.sql");
|
||||
const SQLITE_MIGRATION_RUNTIME_READ_PATH = join(MIGRATIONS_DIR, "0030_sqlite_migration_runtime_read.sql");
|
||||
const WORKFLOW_TASK_CONTINUATIONS_PATH = join(MIGRATIONS_DIR, "0031_workflow_task_continuations.sql");
|
||||
|
||||
/**
|
||||
* Ensure the migration bookkeeping table exists. Lives in the public schema so
|
||||
@@ -422,6 +424,7 @@ export async function applySchemaBaseline(
|
||||
const workflowIrPinAndLegacyAdoptionAlreadyApplied = applied.includes(WORKFLOW_IR_PIN_AND_LEGACY_ADOPTION_VERSION);
|
||||
const planningActiveTimingAlreadyApplied = applied.includes(PLANNING_ACTIVE_TIMING_VERSION);
|
||||
const sqliteMigrationRuntimeReadAlreadyApplied = applied.includes(SQLITE_MIGRATION_RUNTIME_READ_VERSION);
|
||||
const workflowTaskContinuationsAlreadyApplied = applied.includes(WORKFLOW_TASK_CONTINUATIONS_VERSION);
|
||||
assertBinaryNotOlderThanDatabase(applied);
|
||||
let schemaChanged = false;
|
||||
|
||||
@@ -871,6 +874,13 @@ export async function applySchemaBaseline(
|
||||
schemaChanged = true;
|
||||
}
|
||||
|
||||
if (!workflowTaskContinuationsAlreadyApplied) {
|
||||
const migrationSql = await readFile(WORKFLOW_TASK_CONTINUATIONS_PATH, "utf8");
|
||||
await tx.execute(sql.raw(migrationSql));
|
||||
await tx.execute(sql`INSERT INTO public.${sql.identifier(MIGRATION_BOOKKEEPING_TABLE)} (version) VALUES (${WORKFLOW_TASK_CONTINUATIONS_VERSION}) ON CONFLICT (version) DO NOTHING`);
|
||||
schemaChanged = true;
|
||||
}
|
||||
|
||||
return { applied: schemaChanged, pluginHooksRun: pluginHooks.length };
|
||||
});
|
||||
}
|
||||
|
||||
@@ -874,6 +874,12 @@ export const workflowWorkItems = projectSchema.table("workflow_work_items", {
|
||||
leaseExpiresAt: text("lease_expires_at"),
|
||||
lastError: text("last_error"),
|
||||
blockedReason: text("blocked_reason"),
|
||||
stableWorkflowRunId: text("stable_workflow_run_id"),
|
||||
continuationSequence: integer("continuation_sequence"),
|
||||
waitReason: text("wait_reason"),
|
||||
sourceColumn: text("source_column"),
|
||||
targetColumn: text("target_column"),
|
||||
irHash: text("ir_hash"),
|
||||
createdAt: text("created_at").notNull(),
|
||||
updatedAt: text("updated_at").notNull(),
|
||||
}, (t) => [
|
||||
@@ -884,6 +890,9 @@ export const workflowWorkItems = projectSchema.table("workflow_work_items", {
|
||||
index("idx_workflow_work_items_due").on(t.state, t.retryAfter, t.createdAt),
|
||||
index("idx_workflow_work_items_leaseExpiresAt").on(t.leaseExpiresAt),
|
||||
index("idx_workflow_work_items_task_run").on(t.taskId, t.runId),
|
||||
uniqueIndex("idx_workflow_work_items_one_active_task_continuation")
|
||||
.on(t.projectId, t.taskId)
|
||||
.where(sql`${t.kind} = 'task' AND ${t.state} IN ('runnable', 'running', 'held', 'retrying')`),
|
||||
]);
|
||||
|
||||
export const workflowRunBranches = projectSchema.table("workflow_run_branches", {
|
||||
|
||||
@@ -118,7 +118,7 @@ import { acquireSymbolLocksAsync, inspectSymbolLockConflictsAsync, reconcileStal
|
||||
import type { AcquireSymbolLocksResult, ReconcileStaleSymbolLocksResult, ReleaseSymbolLocksResult, RenewSymbolLocksResult, SymbolLockConflict, SymbolLockOwner } from "./symbol-lock-types.js";
|
||||
import { queryRunAuditEvents } from "./task-store/async-audit.js";
|
||||
import { isValidMergeRequestTransitionImpl, enqueueMergeQueueSyncInternalImpl, releaseMergeQueueLeaseImpl, collectMergeDetailsImpl, applyPrMergedTransitionImpl } from "./task-store/merge-queue-ops-2.js";
|
||||
import { upsertWorkflowWorkItemImpl, transitionWorkflowWorkItemImpl, acquireWorkflowWorkItemLeaseImpl } from "./task-store/workflow-workitems-ops-2.js";
|
||||
import { upsertWorkflowWorkItemImpl, replaceActiveTaskWorkflowContinuationImpl, transitionWorkflowWorkItemImpl, acquireWorkflowWorkItemLeaseImpl } from "./task-store/workflow-workitems-ops-2.js";
|
||||
import { getSettingsImpl, getSettingsFastImpl, getSettingsByScopeImpl, getSettingsByScopeFastImpl } from "./task-store/settings-ops-2.js";
|
||||
import { runPluginColumnTransitionHooksImpl, logEntryImpl } from "./task-store/audit-ops.js";
|
||||
import { clearWorkflowRunBranchesImpl, projectMergeRequestToWorkflowWorkItemImpl, createCompletionHandoffWorkflowWorkImpl } from "./task-store/workflow-workitems-ops.js";
|
||||
@@ -1511,6 +1511,9 @@ export class TaskStore extends EventEmitter<TaskStoreEvents> {
|
||||
async upsertWorkflowWorkItem(input: WorkflowWorkItemUpsertInput, tx?: import("./postgres/data-layer.js").DbTransaction): Promise<WorkflowWorkItem> {
|
||||
return upsertWorkflowWorkItemImpl(this, input, tx);
|
||||
}
|
||||
async replaceActiveTaskWorkflowContinuation(input: WorkflowWorkItemUpsertInput & { kind: "task" }): Promise<WorkflowWorkItem> {
|
||||
return replaceActiveTaskWorkflowContinuationImpl(this, input);
|
||||
}
|
||||
async transitionWorkflowWorkItem( id: string, state: WorkflowWorkItemState, patch: WorkflowWorkItemTransitionPatch = {}, tx?: import("./postgres/data-layer.js").DbTransaction, ): Promise<WorkflowWorkItem> {
|
||||
return transitionWorkflowWorkItemImpl(this, id, state, patch, tx);
|
||||
}
|
||||
|
||||
@@ -45,11 +45,18 @@ import type { WorkflowWorkItemRow } from "./row-types.js";
|
||||
* item from being silently resurrected.
|
||||
*/
|
||||
const TERMINAL_WORKFLOW_WORK_ITEM_STATES: ReadonlySet<string> = new Set([
|
||||
"completed",
|
||||
"succeeded",
|
||||
"failed",
|
||||
"cancelled",
|
||||
]);
|
||||
|
||||
const ACTIVE_TASK_CONTINUATION_STATES: WorkflowWorkItemState[] = [
|
||||
"runnable",
|
||||
"running",
|
||||
"held",
|
||||
"retrying",
|
||||
];
|
||||
|
||||
/**
|
||||
* Normalize a workflow-work-item state string. Unknown values default to
|
||||
* 'runnable' (the sync `normalizeWorkflowWorkItemState` behavior).
|
||||
@@ -81,6 +88,12 @@ export function rowToWorkflowWorkItem(row: WorkflowWorkItemRow): WorkflowWorkIte
|
||||
leaseExpiresAt: row.leaseExpiresAt,
|
||||
lastError: row.lastError,
|
||||
blockedReason: row.blockedReason,
|
||||
stableWorkflowRunId: row.stableWorkflowRunId,
|
||||
continuationSequence: row.continuationSequence,
|
||||
waitReason: row.waitReason === "planning" || row.waitReason === "capacity" ? row.waitReason : null,
|
||||
sourceColumn: row.sourceColumn,
|
||||
targetColumn: row.targetColumn,
|
||||
irHash: row.irHash,
|
||||
createdAt: row.createdAt,
|
||||
updatedAt: row.updatedAt,
|
||||
};
|
||||
@@ -171,6 +184,12 @@ export async function upsertWorkflowWorkItem(
|
||||
lastError: input.lastError === undefined ? existing?.lastError ?? null : input.lastError,
|
||||
blockedReason:
|
||||
input.blockedReason === undefined ? existing?.blockedReason ?? null : input.blockedReason,
|
||||
stableWorkflowRunId: input.stableWorkflowRunId === undefined ? existing?.stableWorkflowRunId ?? null : input.stableWorkflowRunId,
|
||||
continuationSequence: input.continuationSequence === undefined ? existing?.continuationSequence ?? null : input.continuationSequence,
|
||||
waitReason: input.waitReason === undefined ? existing?.waitReason ?? null : input.waitReason,
|
||||
sourceColumn: input.sourceColumn === undefined ? existing?.sourceColumn ?? null : input.sourceColumn,
|
||||
targetColumn: input.targetColumn === undefined ? existing?.targetColumn ?? null : input.targetColumn,
|
||||
irHash: input.irHash === undefined ? existing?.irHash ?? null : input.irHash,
|
||||
createdAt: existing?.createdAt ?? now,
|
||||
updatedAt: now,
|
||||
})
|
||||
@@ -193,6 +212,12 @@ export async function upsertWorkflowWorkItem(
|
||||
lastError: input.lastError === undefined ? existing?.lastError ?? null : input.lastError,
|
||||
blockedReason:
|
||||
input.blockedReason === undefined ? existing?.blockedReason ?? null : input.blockedReason,
|
||||
stableWorkflowRunId: input.stableWorkflowRunId === undefined ? existing?.stableWorkflowRunId ?? null : input.stableWorkflowRunId,
|
||||
continuationSequence: input.continuationSequence === undefined ? existing?.continuationSequence ?? null : input.continuationSequence,
|
||||
waitReason: input.waitReason === undefined ? existing?.waitReason ?? null : input.waitReason,
|
||||
sourceColumn: input.sourceColumn === undefined ? existing?.sourceColumn ?? null : input.sourceColumn,
|
||||
targetColumn: input.targetColumn === undefined ? existing?.targetColumn ?? null : input.targetColumn,
|
||||
irHash: input.irHash === undefined ? existing?.irHash ?? null : input.irHash,
|
||||
updatedAt: now,
|
||||
},
|
||||
});
|
||||
@@ -222,6 +247,40 @@ export async function upsertWorkflowWorkItem(
|
||||
return existingTx ? doWork(existingTx) : layer.transactionImmediate(doWork);
|
||||
}
|
||||
|
||||
/** Atomically retire the current task continuation and persist its successor. */
|
||||
export async function replaceActiveTaskWorkflowContinuation(
|
||||
layer: AsyncDataLayer,
|
||||
input: WorkflowWorkItemUpsertInput & { kind: "task" },
|
||||
): Promise<WorkflowWorkItem> {
|
||||
return layer.transactionImmediate(async (tx) => {
|
||||
const activeRows = await tx
|
||||
.select()
|
||||
.from(schema.project.workflowWorkItems)
|
||||
.where(
|
||||
and(
|
||||
eq(schema.project.workflowWorkItems.taskId, input.taskId),
|
||||
eq(schema.project.workflowWorkItems.kind, "task"),
|
||||
inArray(schema.project.workflowWorkItems.state, ACTIVE_TASK_CONTINUATION_STATES),
|
||||
),
|
||||
);
|
||||
|
||||
for (const row of activeRows as WorkflowWorkItemRow[]) {
|
||||
const isSameIdentity =
|
||||
row.runId === input.runId && row.nodeId === input.nodeId && row.kind === input.kind;
|
||||
if (isSameIdentity) continue;
|
||||
await transitionWorkflowWorkItem(
|
||||
layer,
|
||||
row.id,
|
||||
"succeeded",
|
||||
{ leaseOwner: null, leaseExpiresAt: null, lastError: null },
|
||||
tx,
|
||||
);
|
||||
}
|
||||
|
||||
return upsertWorkflowWorkItem(layer, input, tx);
|
||||
});
|
||||
}
|
||||
|
||||
/**
|
||||
* FNXC:TaskStoreWorkflowWorkItems 2026-06-24-08:45:
|
||||
* Transition a workflow work item to a new state INSIDE a transaction, with a
|
||||
|
||||
@@ -1183,6 +1183,25 @@ export async function moveTaskInternalImpl(store: TaskStore, id: string, toColum
|
||||
});
|
||||
void store.clearCompletionHandoffAcceptedMarker(id);
|
||||
}
|
||||
if (toColumn === "todo" && moveSource === "user" && (fromIsImplementation || fromColumn === "in-review")) {
|
||||
// FNXC:WorkflowTaskCancellation 2026-07-21-11:51:
|
||||
// The task move is already committed here. Continuation cleanup is
|
||||
// best-effort so a storage fault cannot suppress task:moved or strand
|
||||
// transitionPending after a successful operator hard-cancel.
|
||||
try {
|
||||
await store.cancelActiveWorkflowWorkItemsForTask(id, {
|
||||
kinds: ["task"],
|
||||
now: movedAt,
|
||||
lastError: "cancelled-by-user-hard-cancel",
|
||||
});
|
||||
} catch (err) {
|
||||
storeLog.warn("Failed to cancel active task workflow continuation on user todo move (degraded)", {
|
||||
phase: "moveTaskInternal:cancel-task-continuation",
|
||||
taskId: id,
|
||||
error: err instanceof Error ? err.message : String(err),
|
||||
});
|
||||
}
|
||||
}
|
||||
if (toColumn === "done") {
|
||||
// FNXC:RuntimeTaskOrchestrationAsync 2026-06-24-16:00:
|
||||
// Backend mode: clearLinkedAgentTaskIds is a sync SQLite operation; skip
|
||||
|
||||
@@ -189,6 +189,12 @@ export interface WorkflowWorkItemRow {
|
||||
leaseExpiresAt: string | null;
|
||||
lastError: string | null;
|
||||
blockedReason: string | null;
|
||||
stableWorkflowRunId: string | null;
|
||||
continuationSequence: number | null;
|
||||
waitReason: string | null;
|
||||
sourceColumn: string | null;
|
||||
targetColumn: string | null;
|
||||
irHash: string | null;
|
||||
createdAt: string;
|
||||
updatedAt: string;
|
||||
}
|
||||
|
||||
@@ -230,6 +230,12 @@ export function rowToWorkflowWorkItemImpl(store: TaskStore, row: WorkflowWorkIte
|
||||
leaseExpiresAt: row.leaseExpiresAt,
|
||||
lastError: row.lastError,
|
||||
blockedReason: row.blockedReason,
|
||||
stableWorkflowRunId: row.stableWorkflowRunId,
|
||||
continuationSequence: row.continuationSequence,
|
||||
waitReason: row.waitReason === "planning" || row.waitReason === "capacity" ? row.waitReason : null,
|
||||
sourceColumn: row.sourceColumn,
|
||||
targetColumn: row.targetColumn,
|
||||
irHash: row.irHash,
|
||||
createdAt: row.createdAt,
|
||||
updatedAt: row.updatedAt,
|
||||
};
|
||||
|
||||
@@ -13,7 +13,7 @@ import {and, eq, inArray} from "drizzle-orm";
|
||||
import type {WorkflowWorkItem, WorkflowWorkItemState, WorkflowWorkItemTransitionPatch, WorkflowWorkItemUpsertInput} from "../types.js";
|
||||
import "../builtin-traits.js";
|
||||
import {__setTaskActivityLogLimitsForTesting} from "../task-store/comments.js";
|
||||
import {upsertWorkflowWorkItem as upsertWorkflowWorkItemAsync, transitionWorkflowWorkItem as transitionWorkflowWorkItemAsync, getWorkflowWorkItem as getWorkflowWorkItemAsync} from "../task-store/async-workflow-workitems.js";
|
||||
import {replaceActiveTaskWorkflowContinuation as replaceActiveTaskWorkflowContinuationAsync, upsertWorkflowWorkItem as upsertWorkflowWorkItemAsync, transitionWorkflowWorkItem as transitionWorkflowWorkItemAsync, getWorkflowWorkItem as getWorkflowWorkItemAsync} from "../task-store/async-workflow-workitems.js";
|
||||
import type {WorkflowWorkItemRow} from "../task-store/row-types.js";
|
||||
import type {DbTransaction} from "../postgres/data-layer.js";
|
||||
|
||||
@@ -40,9 +40,10 @@ export async function upsertWorkflowWorkItemImpl(store: TaskStore, input: Workfl
|
||||
.prepare(
|
||||
`INSERT INTO workflow_work_items (
|
||||
id, runId, taskId, nodeId, kind, state, attempt, retryAfter,
|
||||
leaseOwner, leaseExpiresAt, lastError, blockedReason, createdAt, updatedAt
|
||||
leaseOwner, leaseExpiresAt, lastError, blockedReason, stableWorkflowRunId,
|
||||
continuationSequence, waitReason, sourceColumn, targetColumn, irHash, createdAt, updatedAt
|
||||
)
|
||||
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
|
||||
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
|
||||
ON CONFLICT(runId, taskId, nodeId, kind) DO UPDATE SET
|
||||
state = excluded.state,
|
||||
attempt = excluded.attempt,
|
||||
@@ -51,6 +52,12 @@ export async function upsertWorkflowWorkItemImpl(store: TaskStore, input: Workfl
|
||||
leaseExpiresAt = excluded.leaseExpiresAt,
|
||||
lastError = excluded.lastError,
|
||||
blockedReason = excluded.blockedReason,
|
||||
stableWorkflowRunId = excluded.stableWorkflowRunId,
|
||||
continuationSequence = excluded.continuationSequence,
|
||||
waitReason = excluded.waitReason,
|
||||
sourceColumn = excluded.sourceColumn,
|
||||
targetColumn = excluded.targetColumn,
|
||||
irHash = excluded.irHash,
|
||||
updatedAt = excluded.updatedAt`,
|
||||
)
|
||||
.run(
|
||||
@@ -66,6 +73,12 @@ export async function upsertWorkflowWorkItemImpl(store: TaskStore, input: Workfl
|
||||
input.leaseExpiresAt === undefined ? existing?.leaseExpiresAt ?? null : input.leaseExpiresAt,
|
||||
input.lastError === undefined ? existing?.lastError ?? null : input.lastError,
|
||||
input.blockedReason === undefined ? existing?.blockedReason ?? null : input.blockedReason,
|
||||
input.stableWorkflowRunId === undefined ? existing?.stableWorkflowRunId ?? null : input.stableWorkflowRunId,
|
||||
input.continuationSequence === undefined ? existing?.continuationSequence ?? null : input.continuationSequence,
|
||||
input.waitReason === undefined ? existing?.waitReason ?? null : input.waitReason,
|
||||
input.sourceColumn === undefined ? existing?.sourceColumn ?? null : input.sourceColumn,
|
||||
input.targetColumn === undefined ? existing?.targetColumn ?? null : input.targetColumn,
|
||||
input.irHash === undefined ? existing?.irHash ?? null : input.irHash,
|
||||
existing?.createdAt ?? now,
|
||||
now,
|
||||
);
|
||||
@@ -84,6 +97,33 @@ export async function upsertWorkflowWorkItemImpl(store: TaskStore, input: Workfl
|
||||
});
|
||||
}
|
||||
|
||||
export async function replaceActiveTaskWorkflowContinuationImpl(
|
||||
store: TaskStore,
|
||||
input: WorkflowWorkItemUpsertInput & { kind: "task" },
|
||||
): Promise<WorkflowWorkItem> {
|
||||
if (store.backendMode) {
|
||||
return replaceActiveTaskWorkflowContinuationAsync(store.asyncLayer!, input);
|
||||
}
|
||||
|
||||
// Compatibility path for legacy embedded stores. PostgreSQL is the
|
||||
// authoritative runtime and performs this replacement atomically above.
|
||||
return store.db.transactionImmediate(() => {
|
||||
const active = store.db.prepare(
|
||||
`SELECT id, runId, nodeId, kind FROM workflow_work_items
|
||||
WHERE taskId = ? AND kind = 'task' AND state IN ('runnable', 'running', 'held', 'retrying')`,
|
||||
).all(input.taskId) as Array<{ id: string; runId: string; nodeId: string; kind: string }>;
|
||||
for (const row of active) {
|
||||
if (row.runId === input.runId && row.nodeId === input.nodeId && row.kind === input.kind) continue;
|
||||
store.transitionWorkflowWorkItemSync(row.id, "succeeded", {
|
||||
leaseOwner: null,
|
||||
leaseExpiresAt: null,
|
||||
lastError: null,
|
||||
});
|
||||
}
|
||||
return upsertWorkflowWorkItemImpl(store, input);
|
||||
});
|
||||
}
|
||||
|
||||
export async function transitionWorkflowWorkItemImpl(store: TaskStore, id: string, state: WorkflowWorkItemState, patch: WorkflowWorkItemTransitionPatch = {}, tx?: DbTransaction,): Promise<WorkflowWorkItem> {
|
||||
if (store.backendMode) {
|
||||
const layer = store.asyncLayer!;
|
||||
@@ -166,4 +206,3 @@ export async function acquireWorkflowWorkItemLeaseImpl(store: TaskStore, id: str
|
||||
return store.rowToWorkflowWorkItem(row);
|
||||
});
|
||||
}
|
||||
|
||||
|
||||
@@ -46,7 +46,6 @@ export { validateMcpServerDefinitionDetailed, validateMcpServerDefinitionsDetail
|
||||
*/
|
||||
export const DEPRECATED_BUILTIN_WORKFLOW_IDS: ReadonlySet<string> = new Set([
|
||||
"builtin:brainstorming",
|
||||
"builtin:coding-ideas",
|
||||
]);
|
||||
|
||||
|
||||
|
||||
@@ -56,6 +56,12 @@ export interface WorkflowWorkItem {
|
||||
leaseExpiresAt: string | null;
|
||||
lastError: string | null;
|
||||
blockedReason: string | null;
|
||||
stableWorkflowRunId: string | null;
|
||||
continuationSequence: number | null;
|
||||
waitReason: "planning" | "capacity" | null;
|
||||
sourceColumn: string | null;
|
||||
targetColumn: string | null;
|
||||
irHash: string | null;
|
||||
createdAt: string;
|
||||
updatedAt: string;
|
||||
}
|
||||
@@ -73,6 +79,12 @@ export interface WorkflowWorkItemUpsertInput {
|
||||
leaseExpiresAt?: string | null;
|
||||
lastError?: string | null;
|
||||
blockedReason?: string | null;
|
||||
stableWorkflowRunId?: string | null;
|
||||
continuationSequence?: number | null;
|
||||
waitReason?: "planning" | "capacity" | null;
|
||||
sourceColumn?: string | null;
|
||||
targetColumn?: string | null;
|
||||
irHash?: string | null;
|
||||
now?: string;
|
||||
}
|
||||
|
||||
|
||||
@@ -1553,6 +1553,22 @@ function validateColumnAgent(column: WorkflowIrColumn): void {
|
||||
function validateV2(ir: WorkflowIrV2): void {
|
||||
validateColumns(ir);
|
||||
|
||||
// Capacity holds must have somewhere the scheduler can actually release
|
||||
// them. Failing authoring here avoids durable continuations that can never
|
||||
// become runnable.
|
||||
for (const [index, column] of ir.columns.entries()) {
|
||||
const hold = column.traits.find((trait) => trait.trait === "hold");
|
||||
if (hold?.config?.release !== "capacity") continue;
|
||||
const hasDownstreamCapacity = ir.columns
|
||||
.slice(index + 1)
|
||||
.some((candidate) => resolveColumnFlags(candidate).countsTowardWip === true);
|
||||
if (!hasDownstreamCapacity) {
|
||||
throw new WorkflowIrError(
|
||||
`Workflow IR capacity hold column '${column.id}' requires a downstream wip column`,
|
||||
);
|
||||
}
|
||||
}
|
||||
|
||||
const columnIds = new Set(ir.columns.map((c) => c.id));
|
||||
const nodeIds = new Set<string>();
|
||||
for (const node of ir.nodes) {
|
||||
|
||||
@@ -14,6 +14,21 @@ export interface ResolvedWorkflowOptionalStep {
|
||||
defaultOn: boolean;
|
||||
}
|
||||
|
||||
/** Resolve one optional group's effective state consistently across runtime and
|
||||
* readiness gates. An explicit list (including `[]`) is authoritative; only a
|
||||
* missing list falls back to the workflow-authored default.
|
||||
* FNXC:WorkflowOptionalSteps 2026-07-21-11:51: Persisted explicit selections,
|
||||
* including an empty list, must override workflow defaults across all gates. */
|
||||
export function isWorkflowOptionalGroupEnabled(
|
||||
enabledWorkflowSteps: readonly string[] | undefined,
|
||||
groupId: string,
|
||||
defaultOn = false,
|
||||
): boolean {
|
||||
return Array.isArray(enabledWorkflowSteps)
|
||||
? enabledWorkflowSteps.includes(groupId)
|
||||
: defaultOn;
|
||||
}
|
||||
|
||||
/*
|
||||
FNXC:WorkflowOptionalGroup 2026-06-21-14:05:
|
||||
Re-pointed the per-task optional-step toggle SOURCE from the execution-inert `ir.optionalSteps` declaration to v2 `optional-group` NODES (one resolved entry per group). The legacy `WorkflowOptionalStep` type + `optionalSteps` IR field are now REMOVED (FNXC:WorkflowOptionalGroup 2026-06-21-18:00); a legacy persisted `optionalSteps` key on an old v2 row is tolerated/ignored at parse.
|
||||
|
||||
@@ -0,0 +1,53 @@
|
||||
import { describe, expect, it } from "vitest";
|
||||
import type { WorkflowIr } from "@fusion/core";
|
||||
import { isUnplannedForExecution, resolvePreReleasePlanReviewNode } from "../hold-release.js";
|
||||
|
||||
function workflow(reviewColumn = "todo"): WorkflowIr {
|
||||
return {
|
||||
version: "v2",
|
||||
name: "pre-release-review",
|
||||
columns: [
|
||||
{ id: "todo", name: "Todo", traits: [{ trait: "hold", config: { release: "capacity" } }] },
|
||||
{ id: "in-progress", name: "In progress", traits: [{ trait: "wip" }] },
|
||||
{ id: "done", name: "Done", traits: [{ trait: "complete" }] },
|
||||
],
|
||||
nodes: [
|
||||
{ id: "start", kind: "start", column: "todo" },
|
||||
{ id: "plan-review", kind: "optional-group", column: reviewColumn, config: { defaultOn: true, template: { nodes: [], edges: [] } } },
|
||||
{ id: "end", kind: "end", column: "done" },
|
||||
],
|
||||
edges: [
|
||||
{ from: "start", to: "plan-review" },
|
||||
{ from: "plan-review", to: "end", condition: "success" },
|
||||
],
|
||||
};
|
||||
}
|
||||
|
||||
describe("pre-release Plan Review readiness", () => {
|
||||
it("traverses the pre-release boundary even when its optional review is disabled", () => {
|
||||
expect(resolvePreReleasePlanReviewNode(workflow())?.id).toBe("plan-review");
|
||||
});
|
||||
|
||||
it("does not classify a review already inside WIP as a pre-release gate", () => {
|
||||
expect(resolvePreReleasePlanReviewNode(workflow("in-progress"))).toBeUndefined();
|
||||
});
|
||||
|
||||
it("keys release readiness to the durable capacity continuation", async () => {
|
||||
const task = { id: "T-4", column: "todo" } as any;
|
||||
const item = {
|
||||
id: "continuation",
|
||||
taskId: task.id,
|
||||
kind: "task",
|
||||
state: "held",
|
||||
waitReason: "capacity",
|
||||
sourceColumn: "todo",
|
||||
};
|
||||
const store = {
|
||||
listWorkflowWorkItemsForTask: async () => [item],
|
||||
} as any;
|
||||
|
||||
await expect(isUnplannedForExecution(store, task, workflow())).resolves.toBe(false);
|
||||
item.waitReason = "planning";
|
||||
await expect(isUnplannedForExecution(store, task, workflow())).resolves.toBe(true);
|
||||
});
|
||||
});
|
||||
@@ -130,12 +130,20 @@ describe("WorkflowColumnBoundary controller", () => {
|
||||
});
|
||||
|
||||
// execute lives in the wip column; entering it from the hold column must NOT move.
|
||||
await c.onNodeEntry(ir.nodes.find((n) => n.id === "execute")!);
|
||||
const entry = await c.onNodeEntry(ir.nodes.find((n) => n.id === "execute")!);
|
||||
|
||||
expect(moves).toHaveLength(0);
|
||||
expect(audit).toHaveLength(0);
|
||||
expect(c.currentColumn()).toBe("todo");
|
||||
expect(warnings.some((w) => w.includes("hold→wip"))).toBe(true);
|
||||
expect(entry).toEqual({
|
||||
kind: "suspended",
|
||||
reason: "capacity",
|
||||
nodeId: "execute",
|
||||
fromColumn: "todo",
|
||||
toColumn: "in-progress",
|
||||
irHash: expect.any(String),
|
||||
});
|
||||
});
|
||||
|
||||
it("settles a repeated transition exactly once (scenario 4: kill/restart idempotency)", async () => {
|
||||
@@ -164,7 +172,7 @@ describe("WorkflowColumnBoundary controller", () => {
|
||||
rejectMove: (to) => to === "in-review",
|
||||
});
|
||||
|
||||
await c.onNodeEntry(ir.nodes.find((n) => n.id === "review")!);
|
||||
await expect(c.onNodeEntry(ir.nodes.find((n) => n.id === "review")!)).rejects.toThrow("rejected move to in-review");
|
||||
|
||||
expect(moves).toHaveLength(0);
|
||||
expect(audit).toHaveLength(0);
|
||||
@@ -212,6 +220,32 @@ describe("WorkflowColumnBoundary controller", () => {
|
||||
});
|
||||
|
||||
describe("WorkflowGraphExecutor × column boundary (integration)", () => {
|
||||
it("suspends before a wip node and resumes it explicitly after scheduler release", async () => {
|
||||
const ir = benchmarkSliceIr();
|
||||
const handler = vi.fn(async () => ({ outcome: "success" as const }));
|
||||
const heldBoundary = wiredController({ ir, initialColumn: "todo", moves: [], audit: [] });
|
||||
const heldExecutor = new WorkflowGraphExecutor({ handlers: { prompt: handler }, columnBoundary: heldBoundary });
|
||||
const task = { id: "T-1", column: "todo" } as TaskDetail;
|
||||
|
||||
const suspended = await heldExecutor.run(task, settingsOn(), ir);
|
||||
|
||||
expect(suspended.suspended).toMatchObject({ nodeId: "execute", toColumn: "in-progress" });
|
||||
expect(handler).not.toHaveBeenCalled();
|
||||
|
||||
const releasedBoundary = wiredController({ ir, initialColumn: "in-progress", moves: [], audit: [] });
|
||||
const resumedExecutor = new WorkflowGraphExecutor({ handlers: { prompt: handler }, columnBoundary: releasedBoundary });
|
||||
const resumed = await resumedExecutor.run(
|
||||
{ ...task, column: "in-progress" },
|
||||
settingsOn(),
|
||||
ir,
|
||||
suspended.suspended!.nodeId,
|
||||
);
|
||||
|
||||
expect(resumed.suspended).toBeUndefined();
|
||||
expect(resumed.visitedNodeIds[0]).toBe("execute");
|
||||
expect(handler).toHaveBeenCalled();
|
||||
});
|
||||
|
||||
it("drives one move per real boundary through traversal and no move to done on success-to-end (scenarios 1-3)", async () => {
|
||||
const ir = benchmarkSliceIr();
|
||||
const moves: MoveRecord[] = [];
|
||||
|
||||
@@ -19,7 +19,7 @@ import { finalizeProvenAutoMergeTask } from "./auto-merge-finalization.js";
|
||||
import { mergeEffectiveSettings } from "./effective-settings.js";
|
||||
import { generateFeatureVideo, type GenerateFeatureVideoOptions } from "./review-artifacts/feature-video.js";
|
||||
import { moveTaskToReplanColumn, resolveReplanTargetColumn } from "./replan-target.js";
|
||||
import type { TaskStep, WorkflowIr, WorkflowFieldDefinition, WorkflowColumnAgent, EffectiveAgentInput, WorkflowWorkEngineDispatchResult } from "@fusion/core";
|
||||
import type { TaskStep, WorkflowIr, WorkflowFieldDefinition, WorkflowColumnAgent, EffectiveAgentInput, WorkflowWorkEngineDispatchResult, WorkflowWorkItem } from "@fusion/core";
|
||||
import { WorkflowGraphTaskRunner, type WorkflowGraphTaskRunResult, type WorkflowColumnBoundaryHooks } from "./workflow-graph-task-runner.js";
|
||||
import { createStoreIrPinPersistence, type WorkflowIrPinStoreSurface } from "./workflow-column-boundary.js";
|
||||
import { ensureWorkflowCompletionSummary } from "./workflow-completion-summary.js";
|
||||
@@ -5643,9 +5643,10 @@ export class TaskExecutor {
|
||||
// pending U6/U7 trait re-key; safe now). Absent → the graph performs no
|
||||
// lifecycle moves (pre-cutover byte-identical); present → the controller
|
||||
// moves the card on each node-column boundary with all move-safety.
|
||||
columnBoundaryHooks: this.buildColumnBoundaryHooks(task),
|
||||
columnBoundaryHooks: this.buildColumnBoundaryHooks(task, resolvedRunId),
|
||||
});
|
||||
let result: WorkflowGraphTaskRunResult;
|
||||
let continuation: WorkflowWorkItem | undefined;
|
||||
try {
|
||||
const loadedDetail = await this.store.getTask(task.id);
|
||||
/*
|
||||
@@ -5655,8 +5656,30 @@ export class TaskExecutor {
|
||||
const detail: TaskDetail = loadedDetail?.id === task.id
|
||||
? loadedDetail
|
||||
: { ...task, prompt: task.prompt ?? task.description ?? "" };
|
||||
result = await runner.run(detail, settings);
|
||||
const workItems = await this.store.listWorkflowWorkItemsForTask?.(task.id, { kinds: ["task"] }) ?? [];
|
||||
for (let index = workItems.length - 1; index >= 0; index -= 1) {
|
||||
const candidate = workItems[index];
|
||||
if (["held", "runnable", "running", "retrying"].includes(candidate.state)) {
|
||||
continuation = candidate;
|
||||
break;
|
||||
}
|
||||
}
|
||||
if (continuation && continuation.state !== "running") {
|
||||
continuation = await this.store.transitionWorkflowWorkItem(continuation.id, "running", {
|
||||
leaseOwner: `executor:${task.id}`,
|
||||
leaseExpiresAt: null,
|
||||
lastError: null,
|
||||
});
|
||||
}
|
||||
result = await runner.run(detail, settings, continuation?.nodeId);
|
||||
} catch (err) {
|
||||
if (continuation) {
|
||||
await this.store.transitionWorkflowWorkItem(continuation.id, "failed", {
|
||||
leaseOwner: null,
|
||||
leaseExpiresAt: null,
|
||||
lastError: "workflow-continuation-dispatch-failed",
|
||||
}).catch(() => undefined);
|
||||
}
|
||||
executorLog.error(
|
||||
`[workflow-graph] ${task.id} interpreter threw — parking task as workflow failure: ${err instanceof Error ? err.message : String(err)}`,
|
||||
);
|
||||
@@ -5678,9 +5701,26 @@ export class TaskExecutor {
|
||||
});
|
||||
return;
|
||||
}
|
||||
if (result.disposition === "suspended") {
|
||||
return;
|
||||
}
|
||||
if (result.disposition === "failed") {
|
||||
if (continuation) {
|
||||
await this.store.transitionWorkflowWorkItem(continuation.id, "failed", {
|
||||
leaseOwner: null,
|
||||
leaseExpiresAt: null,
|
||||
lastError: "workflow-continuation-failed",
|
||||
});
|
||||
}
|
||||
await this.handleGraphFailure(task, result);
|
||||
} else if (result.disposition === "completed") {
|
||||
if (continuation) {
|
||||
await this.store.transitionWorkflowWorkItem(continuation.id, "succeeded", {
|
||||
leaseOwner: null,
|
||||
leaseExpiresAt: null,
|
||||
lastError: null,
|
||||
});
|
||||
}
|
||||
const live = await this.store.getTask(task.id).catch(() => task);
|
||||
if ((live as TaskDetail).mergeDetails?.mergeConfirmed === true && (live as TaskDetail).column !== "done") {
|
||||
await this.finalizeMergeConfirmedWorkflowGraphTask(task.id, "graph-completed");
|
||||
@@ -5900,7 +5940,7 @@ export class TaskExecutor {
|
||||
});
|
||||
}
|
||||
|
||||
private buildColumnBoundaryHooks(task: Pick<Task, "id">): WorkflowColumnBoundaryHooks {
|
||||
private buildColumnBoundaryHooks(task: Pick<Task, "id">, workflowRunId?: string): WorkflowColumnBoundaryHooks {
|
||||
// KTD-3 (U9b): store-backed durable IR pin. The cast is the same posture as
|
||||
// buildBranchPersistence — structural probe of the row surface so a store
|
||||
// lacking the pin fields degrades to the inert no-pin seam.
|
||||
@@ -5914,6 +5954,24 @@ export class TaskExecutor {
|
||||
// KTD-3 drift-park loop fix (PR #2342): detectDrift clears the stale pin
|
||||
// row fields so an ordinary requeue re-resolves the CURRENT IR fresh.
|
||||
clearPin: pinPersistence.clearPin,
|
||||
onSuspend: async (suspension) => {
|
||||
const items = await this.store.listWorkflowWorkItemsForTask(task.id, { kinds: ["task"] });
|
||||
const live = items.filter((item) => ["held", "runnable", "running", "retrying"].includes(item.state));
|
||||
if (live.some((item) => item.nodeId === suspension.nodeId)) return;
|
||||
await this.store.replaceActiveTaskWorkflowContinuation({
|
||||
runId: `${workflowRunId ?? `${task.id}:workflow`}:continuation:${suspension.nodeId}:${items.length}`,
|
||||
taskId: task.id,
|
||||
nodeId: suspension.nodeId,
|
||||
kind: "task",
|
||||
state: "held",
|
||||
stableWorkflowRunId: workflowRunId ?? `${task.id}:workflow`,
|
||||
continuationSequence: items.length,
|
||||
waitReason: "capacity",
|
||||
sourceColumn: suspension.fromColumn,
|
||||
targetColumn: suspension.toColumn,
|
||||
irHash: suspension.irHash,
|
||||
});
|
||||
},
|
||||
moveTask: async (toColumn, ctx) => {
|
||||
this.workflowLifecycleMovesInFlight.add(task.id);
|
||||
try {
|
||||
|
||||
@@ -49,6 +49,7 @@ import {
|
||||
type TaskStore,
|
||||
type Task,
|
||||
type WorkflowIr,
|
||||
type WorkflowIrNode,
|
||||
type WorkflowIrV2,
|
||||
type WorkflowIrColumn,
|
||||
} from "@fusion/core";
|
||||
@@ -141,29 +142,18 @@ function isHeldTask(ir: WorkflowIr, task: Task): boolean {
|
||||
* `promoteHeldTask`, `releaseHeldTaskByEvent`) enforces the same invariant.
|
||||
*/
|
||||
/**
|
||||
* U3 — a card is "unplanned for execution" via a PRE-RELEASE Plan Review gate
|
||||
* when: the workflow contains a plan-review node placed in a NON-wip (pre-release)
|
||||
* column, Plan Review is ENABLED for the task (`enabledWorkflowSteps` includes the
|
||||
* group), and no PASSED plan-review step result exists yet. Returns false when the
|
||||
* plan-review node sits in a wip column (post-release gate — builtin `in-progress`),
|
||||
* is absent, is disabled, or has already passed. Pure (no store/clock).
|
||||
* Locate a Plan Review node placed before WIP. Disabled optional groups still
|
||||
* traverse this node, allowing the graph to persist the same generic capacity
|
||||
* continuation without invoking a reviewer.
|
||||
*/
|
||||
function isPlanReviewPreReleaseGateUnpassed(task: Task, ir: WorkflowIr): boolean {
|
||||
export function resolvePreReleasePlanReviewNode(ir: WorkflowIr): WorkflowIrNode | undefined {
|
||||
const planReviewNode = ir.nodes.find((n) => n.id === PLAN_REVIEW_GROUP_ID);
|
||||
if (!planReviewNode?.column) return false;
|
||||
if (!planReviewNode?.column) return undefined;
|
||||
const column = findColumn(ir, planReviewNode.column);
|
||||
if (!column) return false;
|
||||
if (!column) return undefined;
|
||||
// Post-release gate (plan-review lives in a wip column): do not hold release.
|
||||
if (resolveColumnFlags(column).countsTowardWip === true) return false;
|
||||
// Disabled → releases without any reviewer (U3 scenario 5).
|
||||
if (!Array.isArray(task.enabledWorkflowSteps) || !task.enabledWorkflowSteps.includes(PLAN_REVIEW_GROUP_ID)) {
|
||||
return false;
|
||||
}
|
||||
// Already passed → planned; release.
|
||||
const passed = task.workflowStepResults?.some(
|
||||
(r) => r.workflowStepId === PLAN_REVIEW_GROUP_ID && r.status === "passed",
|
||||
);
|
||||
return !passed;
|
||||
if (resolveColumnFlags(column).countsTowardWip === true) return undefined;
|
||||
return planReviewNode;
|
||||
}
|
||||
|
||||
export async function isUnplannedForExecution(store: TaskStore, task: Task, ir: WorkflowIr): Promise<boolean> {
|
||||
@@ -172,14 +162,32 @@ export async function isUnplannedForExecution(store: TaskStore, task: Task, ir:
|
||||
The graph is the SOLE Plan Review owner (triage's out-of-graph gate is deleted).
|
||||
When a workflow places the plan-review node in a PRE-RELEASE column (the
|
||||
benchmark's Plan Review in the Todo hold column — i.e. NOT a wip column), the
|
||||
card must not release into execution until the graph's plan-review gate has
|
||||
PASSED — releasing first would skip the gate. This re-keys the old
|
||||
triage-`status:"planning"` hold onto workflow step state. It intentionally does
|
||||
NOT fire when plan-review is placed in a wip column (builtin: in-progress), where
|
||||
the gate runs post-release, nor when Plan Review is disabled — so a disabled or
|
||||
post-release plan-review workflow releases normally (never deadlocks).
|
||||
card must not release into execution until the graph has reached its durable
|
||||
capacity boundary. Releasing first would skip the gate. This does not fire
|
||||
when Plan Review already lives in a WIP column.
|
||||
*/
|
||||
if (isPlanReviewPreReleaseGateUnpassed(task, ir)) return true;
|
||||
const preReleaseReview = resolvePreReleasePlanReviewNode(ir);
|
||||
if (preReleaseReview) {
|
||||
// Compatibility for tasks planned before durable continuations existed and
|
||||
// for narrow store adapters that expose only the legacy review result.
|
||||
const legacyPassed = task.workflowStepResults?.some(
|
||||
(result) => result.workflowStepId === PLAN_REVIEW_GROUP_ID && result.status === "passed",
|
||||
);
|
||||
if (!legacyPassed) {
|
||||
if (typeof store.listWorkflowWorkItemsForTask !== "function") return true;
|
||||
const continuations = await store.listWorkflowWorkItemsForTask(task.id, { kinds: ["task"] });
|
||||
const active = continuations.filter((item) =>
|
||||
["held", "runnable", "running", "retrying"].includes(item.state),
|
||||
);
|
||||
// Readiness is represented by the graph's durable boundary continuation,
|
||||
// not by a special-case review result. Optional groups that are disabled
|
||||
// are still traversed and therefore reach the same capacity boundary.
|
||||
const readyAtCapacityBoundary = active.some(
|
||||
(item) => item.waitReason === "capacity" && item.sourceColumn === task.column,
|
||||
);
|
||||
if (!readyAtCapacityBoundary) return true;
|
||||
}
|
||||
}
|
||||
// Still-live triage/executor statuses (kept, not triage-plan-review-owned):
|
||||
// `planning` = triage is actively writing PROMPT.md; `needs-replan` = the
|
||||
// executor's graph replan rebound parked the card for another planning pass.
|
||||
|
||||
@@ -15,7 +15,13 @@ import type {
|
||||
CliSession,
|
||||
NotificationPayload,
|
||||
} from "@fusion/core";
|
||||
import { AsyncCentralClaimStore, ChatStore, isEphemeralAgent } from "@fusion/core";
|
||||
import {
|
||||
AsyncCentralClaimStore,
|
||||
ChatStore,
|
||||
computeWorkflowIrPin,
|
||||
isEphemeralAgent,
|
||||
resolveWorkflowIrForTask,
|
||||
} from "@fusion/core";
|
||||
import { Scheduler } from "../scheduler.js";
|
||||
import type { PrMonitor, PrComment } from "../pr-monitor.js";
|
||||
import type { PrInfo } from "@fusion/core";
|
||||
@@ -53,6 +59,7 @@ import { validateProjectNodeMapping } from "../node-dispatch-validation.js";
|
||||
import { attachAgentLinkSync } from "../task-agent-sync.js";
|
||||
import { createRunAuditor, generateSyntheticRunId } from "../run-audit.js";
|
||||
import { setImmediate as setImmediateCb } from "node:timers";
|
||||
import { resolvePreReleasePlanReviewNode } from "../hold-release.js";
|
||||
|
||||
const yieldEventLoop = (): Promise<void> => new Promise((resolve) => setImmediateCb(resolve));
|
||||
|
||||
@@ -228,6 +235,8 @@ export class InProcessRuntime
|
||||
private missionExecutionLoop?: MissionExecutionLoop;
|
||||
private missionAutopilot?: MissionAutopilot;
|
||||
private triageProcessor?: TriageProcessor;
|
||||
private workflowContinuationTimer?: ReturnType<typeof setInterval>;
|
||||
private workflowContinuationDrainActive = false;
|
||||
private messageStore?: MessageStore;
|
||||
private chatStore?: ChatStore;
|
||||
private detachAgentLinkSync?: () => void;
|
||||
@@ -980,6 +989,33 @@ export class InProcessRuntime
|
||||
onSpecifyComplete: (t) => {
|
||||
this.recordActivity();
|
||||
runtimeLog.log(`Specified ${t.id} → todo`);
|
||||
void (async () => {
|
||||
const live = await this.taskStore.getTask(t.id);
|
||||
if (!live || live.paused || live.userPaused) return;
|
||||
const ir = await resolveWorkflowIrForTask(this.taskStore, live.id);
|
||||
const planReview = resolvePreReleasePlanReviewNode(ir);
|
||||
if (!planReview || planReview.column !== live.column) return;
|
||||
|
||||
const active = await this.taskStore.listWorkflowWorkItemsForTask(live.id, { kinds: ["task"] });
|
||||
if (!active.some((item) => ["held", "runnable", "running", "retrying"].includes(item.state))) {
|
||||
await this.taskStore.replaceActiveTaskWorkflowContinuation({
|
||||
runId: `${live.id}:planning-continuation:${planReview.id}:${active.length}`,
|
||||
taskId: live.id,
|
||||
nodeId: planReview.id,
|
||||
kind: "task",
|
||||
state: "runnable",
|
||||
stableWorkflowRunId: `${live.id}:${ir.name}`,
|
||||
continuationSequence: active.length,
|
||||
waitReason: "planning",
|
||||
sourceColumn: live.column,
|
||||
targetColumn: live.column,
|
||||
irHash: computeWorkflowIrPin(ir, planReview.id).irHash,
|
||||
});
|
||||
}
|
||||
this.kickWorkflowContinuationProcessor();
|
||||
})().catch((error) => {
|
||||
runtimeLog.error(`Failed to start Todo plan review for ${t.id}:`, error);
|
||||
});
|
||||
},
|
||||
onSpecifyError: (t, e) => {
|
||||
runtimeLog.error(`Triage failed for ${t.id}: ${e.message}`);
|
||||
@@ -1218,6 +1254,11 @@ export class InProcessRuntime
|
||||
}
|
||||
|
||||
this.setStatus("active");
|
||||
this.workflowContinuationTimer = setInterval(() => {
|
||||
this.kickWorkflowContinuationProcessor();
|
||||
}, 2_000);
|
||||
this.workflowContinuationTimer.unref?.();
|
||||
this.kickWorkflowContinuationProcessor();
|
||||
runtimeLog.log(`InProcessRuntime started for project ${this.config.projectId}`);
|
||||
} catch (error) {
|
||||
const err = error instanceof Error ? error : new Error(String(error));
|
||||
@@ -1274,6 +1315,10 @@ export class InProcessRuntime
|
||||
this.backendShutdown = undefined;
|
||||
let stopError: Error | undefined;
|
||||
try {
|
||||
if (this.workflowContinuationTimer) {
|
||||
clearInterval(this.workflowContinuationTimer);
|
||||
this.workflowContinuationTimer = undefined;
|
||||
}
|
||||
// 1. Remove concurrency change listener (if we registered one)
|
||||
if (this.concurrencyChangedListener && typeof this.centralCore.off === "function") {
|
||||
this.centralCore.off("concurrency:changed", this.concurrencyChangedListener);
|
||||
@@ -1795,6 +1840,37 @@ export class InProcessRuntime
|
||||
return this.missionExecutionLoop;
|
||||
}
|
||||
|
||||
/** Wake the durable task-continuation consumer without nesting execution in triage. */
|
||||
private kickWorkflowContinuationProcessor(): void {
|
||||
queueMicrotask(() => {
|
||||
void this.drainWorkflowContinuations().catch((error) => {
|
||||
runtimeLog.error("Workflow continuation processor failed:", error);
|
||||
});
|
||||
});
|
||||
}
|
||||
|
||||
private async drainWorkflowContinuations(): Promise<void> {
|
||||
if (this.workflowContinuationDrainActive || this.status !== "active") return;
|
||||
this.workflowContinuationDrainActive = true;
|
||||
try {
|
||||
const items = await this.taskStore.listDueWorkflowWorkItems({
|
||||
kinds: ["task"],
|
||||
states: ["runnable", "retrying"],
|
||||
limit: 20,
|
||||
});
|
||||
for (const item of items) {
|
||||
if (item.waitReason !== "planning") continue;
|
||||
const task = await this.taskStore.getTask(item.taskId).catch(() => undefined);
|
||||
if (!task || task.paused || task.userPaused) continue;
|
||||
void this.executor.execute(task).catch((error) => {
|
||||
runtimeLog.error(`Workflow continuation ${item.id} failed:`, error);
|
||||
});
|
||||
}
|
||||
} finally {
|
||||
this.workflowContinuationDrainActive = false;
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Execute a heartbeat run for an agent.
|
||||
*
|
||||
|
||||
@@ -49,6 +49,7 @@ export type WorkflowColumnBoundaryAuditEvent =
|
||||
workflowId: string;
|
||||
fromColumn: string;
|
||||
toColumn: string;
|
||||
irHash: string;
|
||||
nodeId: string;
|
||||
}
|
||||
| {
|
||||
@@ -97,6 +98,8 @@ export interface WorkflowColumnBoundaryDeps {
|
||||
priorPin?: WorkflowIrPin;
|
||||
/** Optional diagnostics sink; never throws into the run. */
|
||||
onWarn?: (message: string, detail: Record<string, unknown>) => void;
|
||||
/** Persist a durable continuation before control returns to the scheduler. */
|
||||
onSuspend?: (suspension: Extract<WorkflowColumnBoundaryEntryResult, { kind: "suspended" }>) => void | Promise<void>;
|
||||
}
|
||||
|
||||
/** The seam the graph executor consumes. */
|
||||
@@ -104,12 +107,23 @@ export interface WorkflowColumnBoundary {
|
||||
/** The card's current lifecycle column (updated after each successful move). */
|
||||
currentColumn(): string;
|
||||
/** Cross into `node.column` when it differs from the current column. */
|
||||
onNodeEntry(node: WorkflowIrNode): Promise<void>;
|
||||
onNodeEntry(node: WorkflowIrNode): Promise<WorkflowColumnBoundaryEntryResult | void>;
|
||||
/** KTD-3 drift guard — run once at graph start. Returns true when the pinned
|
||||
* node/column is gone from the current IR (run must park, not traverse). */
|
||||
detectDrift(): Promise<boolean>;
|
||||
}
|
||||
|
||||
export type WorkflowColumnBoundaryEntryResult =
|
||||
| { kind: "entered" }
|
||||
| {
|
||||
kind: "suspended";
|
||||
reason: "capacity";
|
||||
nodeId: string;
|
||||
fromColumn: string;
|
||||
toColumn: string;
|
||||
irHash: string;
|
||||
};
|
||||
|
||||
/*
|
||||
FNXC:WorkflowIrPin 2026-07-19-18:30 (KTD-3 / U9b):
|
||||
Store-backed KTD-3 pin persistence. The U9b schema landed the durable pin as three
|
||||
@@ -259,10 +273,10 @@ export function createWorkflowColumnBoundary(
|
||||
return true;
|
||||
},
|
||||
|
||||
async onNodeEntry(node: WorkflowIrNode): Promise<void> {
|
||||
async onNodeEntry(node: WorkflowIrNode): Promise<WorkflowColumnBoundaryEntryResult> {
|
||||
const toColumn = node.column;
|
||||
// KTD-1: a columnless node (e.g. `end`) never moves the card.
|
||||
if (!toColumn) return;
|
||||
if (!toColumn) return { kind: "entered" };
|
||||
|
||||
// KTD-3: pin the resolved IR for this node-entry (durable seam).
|
||||
try {
|
||||
@@ -272,7 +286,7 @@ export function createWorkflowColumnBoundary(
|
||||
}
|
||||
|
||||
// Idempotent: a re-entered/rework node or a same-column node chain no-ops.
|
||||
if (toColumn === column) return;
|
||||
if (toColumn === column) return { kind: "entered" };
|
||||
|
||||
const fromColumn = column;
|
||||
|
||||
@@ -283,9 +297,19 @@ export function createWorkflowColumnBoundary(
|
||||
warn("hold→wip boundary parked at ready-for-release seam (scheduler-owned)", {
|
||||
fromColumn,
|
||||
toColumn,
|
||||
irHash: computeWorkflowIrPin(deps.ir, node.id).irHash,
|
||||
nodeId: node.id,
|
||||
});
|
||||
return;
|
||||
const suspension = {
|
||||
kind: "suspended",
|
||||
reason: "capacity",
|
||||
nodeId: node.id,
|
||||
fromColumn,
|
||||
toColumn,
|
||||
irHash: computeWorkflowIrPin(deps.ir, node.id).irHash,
|
||||
} as const;
|
||||
await deps.onSuspend?.(suspension);
|
||||
return suspension;
|
||||
}
|
||||
|
||||
// The single mover: the store's trait-hook moveTask path.
|
||||
@@ -302,7 +326,7 @@ export function createWorkflowColumnBoundary(
|
||||
nodeId: node.id,
|
||||
error: err instanceof Error ? err.message : String(err),
|
||||
});
|
||||
return;
|
||||
throw err;
|
||||
}
|
||||
}
|
||||
|
||||
@@ -314,6 +338,7 @@ export function createWorkflowColumnBoundary(
|
||||
workflowId: deps.workflowId,
|
||||
fromColumn,
|
||||
toColumn,
|
||||
irHash: computeWorkflowIrPin(deps.ir, node.id).irHash,
|
||||
nodeId: node.id,
|
||||
});
|
||||
} catch (err) {
|
||||
@@ -323,6 +348,7 @@ export function createWorkflowColumnBoundary(
|
||||
error: err instanceof Error ? err.message : String(err),
|
||||
});
|
||||
}
|
||||
return { kind: "entered" };
|
||||
},
|
||||
};
|
||||
}
|
||||
|
||||
@@ -9,7 +9,7 @@ import type {
|
||||
WorkflowNodeExtensionResult,
|
||||
WorkflowStepResult,
|
||||
} from "@fusion/core";
|
||||
import { BUILTIN_CODING_WORKFLOW_IR, PLAN_REVIEW_GROUP_ID, WorkflowIrError, getWorkflowExtensionRegistry, resolveMaxReworkCycles, isExperimentalFeatureEnabled, GRAPH_NATIVE_POST_MERGE_FLAG, isCompletionSummaryNode, classifyReviewLease } from "@fusion/core";
|
||||
import { BUILTIN_CODING_WORKFLOW_IR, PLAN_REVIEW_GROUP_ID, WorkflowIrError, getWorkflowExtensionRegistry, resolveMaxReworkCycles, isExperimentalFeatureEnabled, GRAPH_NATIVE_POST_MERGE_FLAG, isCompletionSummaryNode, classifyReviewLease, isWorkflowOptionalGroupEnabled } from "@fusion/core";
|
||||
import { isNonPlanDefectPlanReviewFailure } from "./transient-error-detector.js";
|
||||
|
||||
import {
|
||||
@@ -269,6 +269,19 @@ export interface WorkflowGraphExecutorResult {
|
||||
outcome: WorkflowNodeOutcome;
|
||||
context: Record<string, unknown>;
|
||||
visitedNodeIds: string[];
|
||||
suspended?: {
|
||||
reason: "capacity";
|
||||
nodeId: string;
|
||||
fromColumn: string;
|
||||
toColumn: string;
|
||||
irHash: string;
|
||||
};
|
||||
}
|
||||
|
||||
class WorkflowGraphSuspended extends Error {
|
||||
public constructor(public readonly suspension: NonNullable<WorkflowGraphExecutorResult["suspended"]>) {
|
||||
super(`Workflow suspended before node ${suspension.nodeId}`);
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
@@ -380,8 +393,11 @@ export class WorkflowGraphExecutor {
|
||||
task: TaskDetail,
|
||||
settings: (WorkflowNodeSettings & Partial<Pick<Settings, "autoMerge">>) | undefined,
|
||||
ir: WorkflowIr = BUILTIN_CODING_WORKFLOW_IR,
|
||||
startNodeId?: string,
|
||||
): Promise<WorkflowGraphExecutorResult> {
|
||||
const startNode = ir.nodes.find((node) => node.kind === "start");
|
||||
const startNode = startNodeId
|
||||
? ir.nodes.find((node) => node.id === startNodeId)
|
||||
: ir.nodes.find((node) => node.kind === "start");
|
||||
if (!startNode) throw new WorkflowIrError("Workflow IR missing start node");
|
||||
|
||||
const nodeMap = new Map(ir.nodes.map((node) => [node.id, node]));
|
||||
@@ -553,7 +569,8 @@ export class WorkflowGraphExecutor {
|
||||
// columnless node, a same-column node, or a hold→wip boundary produces no
|
||||
// move; the controller owns that decision. Runs BEFORE the node executes,
|
||||
// so an execute failure parks the card in the column it just entered.
|
||||
await this.deps.columnBoundary?.onNodeEntry(node);
|
||||
const boundary = await this.deps.columnBoundary?.onNodeEntry(node);
|
||||
if (boundary?.kind === "suspended") throw new WorkflowGraphSuspended(boundary);
|
||||
|
||||
if (node.kind === "split") {
|
||||
// Concurrent fan-out: branches run in parallel up to their join, which
|
||||
@@ -667,9 +684,11 @@ export class WorkflowGraphExecutor {
|
||||
* workflow-authored `defaultOn` so default Coding still runs Plan Review
|
||||
* before execution and Code Review before merge.
|
||||
*/
|
||||
const enabled = Array.isArray(task.enabledWorkflowSteps)
|
||||
? task.enabledWorkflowSteps.includes(node.id)
|
||||
: node.config?.defaultOn === true;
|
||||
const enabled = isWorkflowOptionalGroupEnabled(
|
||||
task.enabledWorkflowSteps,
|
||||
node.id,
|
||||
node.config?.defaultOn === true,
|
||||
);
|
||||
const requiresAutoMergeOff = node.config?.requiresAutoMergeOff === true;
|
||||
const autoMergeOff = task.autoMerge === false || (settings?.autoMerge === false && task.autoMerge !== true);
|
||||
/*
|
||||
@@ -1062,7 +1081,8 @@ export class WorkflowGraphExecutor {
|
||||
visitedNodeIds.push(mergeNode.id);
|
||||
// U1: the synthetic merge seam carries the merge-region's column; cross the
|
||||
// boundary on entry so a custom "Merging" column receives the card (KTD-1).
|
||||
await this.deps.columnBoundary?.onNodeEntry(mergeNode);
|
||||
const boundary = await this.deps.columnBoundary?.onNodeEntry(mergeNode);
|
||||
if (boundary?.kind === "suspended") throw new WorkflowGraphSuspended(boundary);
|
||||
const result = await this.executeNodeWithRetries(
|
||||
mergeNode,
|
||||
task,
|
||||
@@ -1145,12 +1165,32 @@ export class WorkflowGraphExecutor {
|
||||
break;
|
||||
}
|
||||
} catch (err) {
|
||||
if (err instanceof WorkflowGraphSuspended) throw err;
|
||||
this.deps.logTaskEntry?.(
|
||||
`[post-merge] traversal error: ${err instanceof Error ? err.message : String(err)}`,
|
||||
);
|
||||
}
|
||||
}
|
||||
if (aggregate.outcome === "failure") break;
|
||||
// A deliberately minimal merge workflow may route success straight
|
||||
// to end with no post-merge work node to enter the complete column.
|
||||
// Preserve end's handler-free terminal semantics while still entering
|
||||
// its authored column after merge proof.
|
||||
if (postMergeEntryNodeIds.length === 0) {
|
||||
const terminalEdge = [...outgoingMap.entries()].flatMap(([from, outgoing]) => {
|
||||
const fromNode = nodeMap.get(from);
|
||||
if (!fromNode || !isMergeRegionKind(fromNode.kind)) return [];
|
||||
return outgoing.filter((candidate) => {
|
||||
const terminal = nodeMap.get(candidate.to);
|
||||
return terminal?.kind === "end" && (!candidate.condition || candidate.condition === "success");
|
||||
});
|
||||
})[0];
|
||||
const terminalNode = terminalEdge ? nodeMap.get(terminalEdge.to) : undefined;
|
||||
if (terminalNode) {
|
||||
const boundary = await this.deps.columnBoundary?.onNodeEntry(terminalNode);
|
||||
if (boundary?.kind === "suspended") throw new WorkflowGraphSuspended(boundary);
|
||||
}
|
||||
}
|
||||
continue;
|
||||
}
|
||||
const child = await walk(edge.to);
|
||||
@@ -1175,7 +1215,19 @@ export class WorkflowGraphExecutor {
|
||||
return { executed: true, outcome: "failure", context, visitedNodeIds };
|
||||
}
|
||||
|
||||
const terminal = await walk(startNode.id);
|
||||
let terminal: WorkflowNodeResult | ReworkSignal;
|
||||
try {
|
||||
terminal = await walk(startNode.id);
|
||||
} catch (error) {
|
||||
if (!(error instanceof WorkflowGraphSuspended)) throw error;
|
||||
return {
|
||||
executed: true,
|
||||
outcome: "success",
|
||||
context,
|
||||
visitedNodeIds,
|
||||
suspended: error.suspension,
|
||||
};
|
||||
}
|
||||
if (isReworkSignal(terminal)) {
|
||||
// A rework edge whose target is not an enclosing head on the stack — i.e. a
|
||||
// rework edge pointing at a node never entered as a loop head. Malformed IR.
|
||||
|
||||
@@ -35,6 +35,7 @@ import type { PrNodeDeps } from "./pr-nodes.js";
|
||||
import type { WorkflowPrimitiveContext, WorkflowRuntimePrimitives } from "./runtime-primitives.js";
|
||||
import {
|
||||
type WorkflowColumnBoundary,
|
||||
type WorkflowColumnBoundaryDeps,
|
||||
type WorkflowColumnBoundaryAuditEvent,
|
||||
type WorkflowColumnMove,
|
||||
createWorkflowColumnBoundary,
|
||||
@@ -49,7 +50,7 @@ import type { WorkflowIrPin } from "@fusion/core";
|
||||
* - "fell-back" — the interpreter did not (or could not) own this task;
|
||||
* the caller must run the legacy pipeline instead.
|
||||
*/
|
||||
export type WorkflowGraphRunDisposition = "completed" | "failed" | "fell-back";
|
||||
export type WorkflowGraphRunDisposition = "completed" | "failed" | "fell-back" | "suspended";
|
||||
|
||||
export interface WorkflowGraphTaskRunResult {
|
||||
disposition: WorkflowGraphRunDisposition;
|
||||
@@ -63,6 +64,7 @@ export interface WorkflowGraphTaskRunResult {
|
||||
interruptedNodeId?: string;
|
||||
/** Typed abort provenance for the interrupted node; absent for genuine node failures. */
|
||||
interruptedAbortKind?: WorkflowNodeAbortKind;
|
||||
suspension?: NonNullable<import("./workflow-graph-executor.js").WorkflowGraphExecutorResult["suspended"]>;
|
||||
}
|
||||
|
||||
/** The minimal store surface the runner needs — keeps tests fake-friendly. */
|
||||
@@ -167,6 +169,7 @@ export interface WorkflowColumnBoundaryHooks {
|
||||
* fields when detectDrift fires so a requeue re-resolves the current IR. */
|
||||
clearPin?: () => void | Promise<void>;
|
||||
onWarn?: (message: string, detail: Record<string, unknown>) => void;
|
||||
onSuspend?: WorkflowColumnBoundaryDeps["onSuspend"];
|
||||
}
|
||||
|
||||
/**
|
||||
@@ -210,6 +213,7 @@ export class WorkflowGraphTaskRunner {
|
||||
public async run(
|
||||
task: TaskDetail,
|
||||
settings: Pick<Settings, "experimentalFeatures"> | undefined,
|
||||
startNodeId?: string,
|
||||
): Promise<WorkflowGraphTaskRunResult> {
|
||||
let selection: { workflowId: string; stepIds: string[] } | undefined;
|
||||
try {
|
||||
@@ -270,6 +274,7 @@ export class WorkflowGraphTaskRunner {
|
||||
priorPin,
|
||||
clearPin: hooks.clearPin,
|
||||
onWarn: hooks.onWarn,
|
||||
onSuspend: hooks.onSuspend,
|
||||
});
|
||||
}
|
||||
|
||||
@@ -403,10 +408,20 @@ export class WorkflowGraphTaskRunner {
|
||||
}
|
||||
},
|
||||
});
|
||||
const result = await executor.run(task, settings, validatedIr);
|
||||
const result = await executor.run(task, settings, validatedIr, startNodeId);
|
||||
if (!result.executed) {
|
||||
return this.fallBack(task.id, "not-executed");
|
||||
}
|
||||
if (result.suspended) {
|
||||
this.emit("terminal", task.id, `${definition.id}:suspended`);
|
||||
return {
|
||||
disposition: "suspended",
|
||||
outcome: result.outcome,
|
||||
visitedNodeIds: result.visitedNodeIds,
|
||||
context: result.context,
|
||||
suspension: result.suspended,
|
||||
};
|
||||
}
|
||||
const disposition: WorkflowGraphRunDisposition = result.outcome === "success" ? "completed" : "failed";
|
||||
this.emit("terminal", task.id, `${definition.id}:${disposition}`);
|
||||
const interruptedNodeId = typeof result.context[WORKFLOW_INTERRUPTED_NODE_ID_CONTEXT_KEY] === "string"
|
||||
|
||||
Reference in New Issue
Block a user