From 8f42098cc329f78fdd051f95f36bf9eea3c69867 Mon Sep 17 00:00:00 2001 From: gsxdsm Date: Mon, 8 Jun 2026 19:13:14 -0700 Subject: [PATCH] feat(FN-6035): route execution through workflow primitives Fusion-Task-Id: FN-6035 --- .../workflow-native-runtime-primitives.md | 5 + CONCEPTS.md | 12 + ...big-bang-workflow-native-execution-plan.md | 372 ++++++++++++++++++ docs/workflow-steps.md | 40 +- .../core/src/builtin-coding-workflow-ir.ts | 17 +- packages/core/src/workflow-compiler.ts | 4 +- .../src/__tests__/runtime-primitives.test.ts | 57 +++ .../workflow-graph-executor-parity.test.ts | 22 +- .../__tests__/workflow-task-runtime.test.ts | 100 +++-- packages/engine/src/executor.ts | 311 ++++++++++++++- packages/engine/src/index.ts | 26 ++ packages/engine/src/runtime-primitives.ts | 238 +++++++++++ .../src/workflow-authoritative-driver.ts | 60 ++- .../engine/src/workflow-graph-executor.ts | 14 +- .../engine/src/workflow-graph-task-runner.ts | 17 + packages/engine/src/workflow-node-handlers.ts | 175 +++++++- packages/engine/src/workflow-task-runtime.ts | 9 +- 17 files changed, 1383 insertions(+), 96 deletions(-) create mode 100644 .changeset/workflow-native-runtime-primitives.md create mode 100644 docs/plans/2026-06-09-001-refactor-big-bang-workflow-native-execution-plan.md create mode 100644 packages/engine/src/__tests__/runtime-primitives.test.ts create mode 100644 packages/engine/src/runtime-primitives.ts diff --git a/.changeset/workflow-native-runtime-primitives.md b/.changeset/workflow-native-runtime-primitives.md new file mode 100644 index 0000000000..96ddba80a5 --- /dev/null +++ b/.changeset/workflow-native-runtime-primitives.md @@ -0,0 +1,5 @@ +--- +"@runfusion/fusion": patch +--- + +Route task execution through workflow-native runtime primitives and make the built-in coding workflow explicitly own planning before execute/review/merge. diff --git a/CONCEPTS.md b/CONCEPTS.md index dc7dbcbfc1..41e69af8a9 100644 --- a/CONCEPTS.md +++ b/CONCEPTS.md @@ -79,6 +79,18 @@ A Feature's position in the execution loop (being implemented, awaiting or under ### Task The core board entity: a unit of work that moves through columns (triage, todo, in-progress, in-review, done, archived) and is executed by agents. A Task carries its own per-task settings that can override project-level defaults. +### Workflow Runtime +The authoritative task lifecycle runtime. It resolves a Task to workflow IR, walks the graph, routes node outcomes, and invokes runtime primitives for side effects. The engine substrate still owns scheduling, routing claims, persistence, concurrency, process supervision, storage, and audit plumbing; lifecycle policy lives in workflow nodes and built-in workflow IR. + +### Runtime Primitive +A named, injected operation a workflow node can call to perform side effects without depending on `executor.ts` lifecycle branches. Examples include planning session, coding session, step execution/reset, review, verification, workflow step, transition, merge request, abort, and audit. Primitives are the boundary between workflow policy and engine substrate. + +### Built-in Lifecycle Node +A node in a built-in workflow that expresses default Fusion behavior, such as planning, execute, review, merge, parse-steps, step-review, or PR lifecycle actions. Built-in lifecycle nodes are the compatibility layer for existing behavior: changing default execution means changing the built-in workflow and its primitive wiring, not adding hidden imperative branches. + +### Recovery Event +A workflow-observable condition that requires recovery policy, such as implementation incomplete, review unavailable, merge timeout, manual merge required, integration conflict, or hard cancel. Recovery may use engine primitives for aborting processes, writing audit entries, resetting steps, or parking tasks, but the routing decision belongs to workflow logic. + ### Auto-merge The named process that automatically lands a completed Task's branch onto its merge target once the Task reaches In-review and passes its merge blockers. Gated twice: a project-level setting enables it globally, and each Task may carry an explicit per-task override. diff --git a/docs/plans/2026-06-09-001-refactor-big-bang-workflow-native-execution-plan.md b/docs/plans/2026-06-09-001-refactor-big-bang-workflow-native-execution-plan.md new file mode 100644 index 0000000000..c28ac0ddec --- /dev/null +++ b/docs/plans/2026-06-09-001-refactor-big-bang-workflow-native-execution-plan.md @@ -0,0 +1,372 @@ +--- +title: "refactor: Big-bang workflow-native execution cutover" +type: refactor +status: active +date: 2026-06-09 +depth: deep +origin: none (solo planning bootstrap; supersedes docs/plans/2026-06-07-001-refactor-workflow-runtime-cutover-plan.md with big-bang posture) +--- + +# refactor: Big-bang workflow-native execution cutover + +## Summary + +Replace Fusion's legacy executor/triage lifecycle with a workflow-native execution model in one coordinated cutover. Built-in workflows become the complete expression of default engine behavior, nodes own execution and recovery policy, and the engine remains the runtime substrate for scheduling, routing, persistence, concurrency, process supervision, storage, and audit plumbing. + +--- + +## Problem Frame + +The workflow stack already has a graph executor, built-in workflow IR, step inversion, workflow-defined columns, workflow settings, node handlers, custom nodes, PR nodes, and plugin workflow extensions. But executable tasks still pass through `TaskExecutor.execute()` and `TriageProcessor.specifyTask()` for most lifecycle policy. The current graph path uses seams and completion interceptors to call back into legacy executor logic, so the system has two control planes: workflow IR says it owns sequencing, while `executor.ts`, `triage.ts`, and self-healing sweeps still own planning, implementation, verification, review, merge routing, and recovery decisions. + +This plan intentionally changes the migration posture. The target is not a parity-gated rollout with old-engine fallback. The target is a single big-bang replacement branch where workflow runtime becomes the only task lifecycle driver before merge. The implementation can be internally sequenced and heavily characterized, but the shipped state must not contain a production alternate path that reruns legacy executor orchestration. + +--- + +## Requirements + +- R1. All executable task lifecycle policy is represented by selected workflow IR, built-in workflow IR, node handlers, or workflow extension nodes. +- R2. Built-in workflows fully express the current default coding, stepwise coding, planning/specification, verification, review, merge, PR response, and recovery behavior. +- R3. The engine keeps scheduler dispatch, node routing, persistence adapters, concurrency/semaphore control, process supervision, storage access, and audit/log sinks. +- R4. The engine does not keep lifecycle policy hidden in `TaskExecutor.execute()`, `TriageProcessor.specifyTask()`, self-healing mutation sweeps, or graph-to-legacy completion interceptors. +- R5. Runtime mechanics are extracted as typed primitives that nodes can call: worktree preparation, session execution, command execution, artifact IO, task transition, merge operation, reviewer invocation, process abort, and audit emission. +- R6. Recovery is workflow/node-owned policy. Sweeps and timers may wake workflow recovery nodes, but they must not directly decide lifecycle repair behavior for task execution. +- R7. Existing invariants remain non-configurable: user move `in-progress -> todo` is hard cancel, `autoMerge:false` is terminal until human merge except scoped shared-branch member integration, file-scope and squash overlap guards remain authoritative, worktree ownership remains enforced, and branch-group merge target rules remain intact. +- R8. Effective execution principal resolution applies across every workflow execution surface, including custom nodes, planning nodes, coding nodes, step-execute, review nodes, heartbeat serialization, resume wakeups, permissions, and hot-swap/change detection. +- R9. A workflow run is durable and inspectable: current node, side-effect boundary, active sessions, branch/foreach instances, terminal outcome, recovery decision, and audit correlation are persisted or derivable through one runtime state model. +- R10. The shipped cutover has no production fallback to old lifecycle orchestration. If workflow runtime fails after merge, failure routes through workflow-defined recovery/parking behavior, not a legacy executor retry path. +- R11. Tests assert the invariant across all known execution surfaces: default workflow, stepwise workflow, custom workflow, plugin work engine extension, PR nodes, triage/planning, pause/cancel, auto-merge off, branch groups, workflow settings/test mode, cross-node runtime routing, and restart recovery. + +--- + +## Scope Boundaries + +### In Scope + +- Promote `WorkflowTaskRuntime` into the only executable task runtime. +- Replace production `WorkflowLegacySeams` with first-class node handlers backed by typed runtime primitives. +- Move built-in triage/spec generation into workflow nodes and a built-in planning region. +- Move deterministic verification, pre-merge workflow steps, revision/fix routing, review handoff, merge/manual-required behavior, and completed-task recovery into built-in workflow IR. +- Convert self-healing execution repair into workflow recovery wakeups plus node-authored recovery branches. +- Preserve existing stores, task rows, task documents, reviewer implementation, merger implementation, agent runtimes, plugin runtime, and scheduler infrastructure as mechanics. +- Delete or demote old lifecycle orchestration so no production path depends on graph re-entry into `TaskExecutor.execute()`. + +### Out of Scope + +- Replacing SQLite storage or the task model identity. +- Rewriting git merge algorithms or GitHub PR operations. +- Rewriting agent runtime protocols. +- Redesigning dashboard workflow authoring UI beyond what is necessary to expose new built-in node kinds. +- Removing workflow authoring surfaces or plugin workflow extension surfaces. +- Changing published CLI release process. + +### Deferred to Follow-Up Work + +- Cosmetic cleanup of old file names after the cutover if the implementation leaves mechanically named modules like `executor-primitives.ts`. +- Broad dashboard UX redesign for workflow debugging beyond required runtime state display. +- New plugin APIs unrelated to execution/recovery ownership. + +--- + +## Key Technical Decisions + +- KTD-1. Big-bang means no production fallback. Implementation may use characterization tests and temporary adapters while the branch is under construction, but the final branch must remove old lifecycle dispatch and old fallback flags before merge. +- KTD-2. Workflow IR expresses policy; runtime primitives express mechanics. A node may call `prepareWorktree`, `runAgentSession`, `runReviewer`, or `requestMerge`, but the decision to call them, retry them, recover from them, or park the task belongs to the workflow graph. +- KTD-3. Built-in workflows are source of truth. `builtin:coding`, `builtin:stepwise-coding`, and any new built-in planning/recovery/PR workflows must encode the current engine logic rather than relying on comments or hidden `executor.ts` branches. +- KTD-4. Recovery becomes routable graph behavior. Timers and sweeps identify stale runtime state and enqueue recovery work; recovery nodes decide whether to resume, reset, requeue, park, request human review, or no-op. +- KTD-5. Preserve centralized invariants as substrate guards. File-scope enforcement, worktree ownership checks, transition validation, merge target resolution, auto-merge processing predicates, and process supervision stay centralized and are invoked by nodes as non-bypassable guards. +- KTD-6. Remove mutable per-task graph seam slots. Existing `graphCompletionInterceptors`, `graphStepRunOnce`, `graphSeamGoverningNodeId`, and similar slots are symptoms of graph-to-executor re-entry. Runtime state should be explicit per workflow run and per node/session. +- KTD-7. Effective principal is resolved once per node invocation and carried through all identity-sensitive subsystems. The column-agent blast-radius checklist applies to every node that can run a session or wake a session. +- KTD-8. Cross-node execution remains a runtime routing concern. A remote node executes the workflow runtime in its own process; the central node routes, monitors, and receives events. It does not execute task lifecycle policy locally for remote work. +- KTD-9. Characterization coverage is allowed; compatibility code is not. Tests may compare old behavior during implementation, but final production code should not keep old orchestration for rollback. + +--- + +## High-Level Technical Design + +### Ownership Boundary + +```mermaid +flowchart TB + Scheduler[Scheduler / task event / recovery timer] --> Claim[Workflow runtime claim] + Claim --> Router{Local or remote node} + Router -->|local| Runtime[WorkflowTaskRuntime] + Router -->|remote| Remote[Remote Fusion API executes WorkflowTaskRuntime] + + Runtime --> Resolve[Resolve workflow IR + effective settings + principal] + Resolve --> Graph[WorkflowGraphExecutor] + + Graph --> Nodes[Workflow node handlers] + Nodes --> Primitives[Runtime primitives] + Primitives --> Store[(TaskStore / CentralCore)] + Primitives --> Git[git / worktrees] + Primitives --> Agents[agent sessions / CLI sessions] + Primitives --> Merge[merger / PR operations] + Primitives --> Audit[run audit / logs] + + Graph --> RunState[(workflow run state)] + Scheduler --> RunState +``` + +The engine owns when a run may start, where it runs, how many concurrent units may execute, how subprocesses are supervised, and where state is stored. The workflow owns what happens next. + +### Built-In Workflow Shape + +```mermaid +flowchart TB + Intake[start/intake] --> Plan[planning/spec node] + Plan --> PlanReview[spec review node] + PlanReview -->|approve| Hold[todo hold/capacity] + PlanReview -->|revise| Plan + Hold --> Prepare[prepare worktree] + Prepare --> Implement[implementation session or foreach steps] + Implement --> Verify[deterministic verification + workflow checks] + Verify -->|approve| Review[review handoff] + Verify -->|revise/fix| Implement + Review --> MergeGate[merge/manual-required gate] + MergeGate -->|auto/processable| Merge[merge operation] + MergeGate -->|manual-required| Manual[in-review hold] + Merge -->|success| Done[done] + Merge -->|failure| Recovery[recovery router] + Recovery --> Implement + Recovery --> Review + Recovery --> Manual + Recovery --> Hold +``` + +The exact node names are implementation details, but every box above must be represented in built-in workflow IR or a built-in reusable subgraph. No box should be hidden inside a monolithic executor function. + +### Recovery Model + +```mermaid +stateDiagram-v2 + [*] --> Running + Running --> Paused: user/global pause + Running --> Aborting: hard cancel + Running --> FailedNode: node failure + Running --> Stale: timer/sweep detects stale run + FailedNode --> RecoveryNode + Stale --> RecoveryNode + RecoveryNode --> Running: resume/retry + RecoveryNode --> TodoHold: clean requeue + RecoveryNode --> ReviewHold: needs human review + RecoveryNode --> ManualMergeHold: autoMerge false/manual required + RecoveryNode --> Done: already completed/finalized + Aborting --> TodoHold + Paused --> Running: unpause +``` + +Sweeps do not mutate task lifecycle directly. They materialize a workflow recovery event or wake the recovery node for a task/run. The recovery node then uses the same transition primitives and invariants as ordinary execution. + +--- + +## Implementation Units + +### U1. Runtime Primitive Boundary + +- **Goal:** Extract lifecycle mechanics from `executor.ts`, `triage.ts`, and self-healing into typed primitives that workflow nodes can call without invoking legacy orchestration. +- **Requirements:** R3, R5, R7. +- **Dependencies:** None. +- **Files:** `packages/engine/src/runtime-primitives.ts` (new), `packages/engine/src/executor.ts`, `packages/engine/src/triage.ts`, `packages/engine/src/step-session-executor.ts`, `packages/engine/src/reviewer.ts`, `packages/engine/src/merger.ts`, `packages/engine/src/worktree-acquisition.ts`, `packages/engine/src/__tests__/runtime-primitives.test.ts`, `packages/engine/src/__tests__/executor-worktree.test.ts`, `packages/engine/src/__tests__/executor-step-session.test.ts`, `packages/engine/src/__tests__/base-commit-capture.real-git.test.ts`. +- **Approach:** Define primitive interfaces for worktree preparation, base/fork-point capture, task document/artifact IO, planning session execution, coding session execution, single-step execution, reviewer invocation, deterministic command execution, workflow-step execution, transition writes, merge request/merge operation, process abort, and audit emission. Move code behind those interfaces in small extractions, but keep behavior identical. Primitives must accept a workflow run context and node context so audit and effective principal are never inferred from task fields alone. +- **Execution note:** Characterization-first. Preserve old behavior with tests before extraction, especially worktree liveness, local-first base capture, file-scope checks, token persistence, and abort cleanup. +- **Patterns to follow:** `step-runner.ts` DI style, `worktree-acquisition.ts`, `base-commit-capture.ts`, `createRunAuditor`, and existing real-git tests. +- **Test scenarios:** worktree preparation refuses invalid/unowned worktrees; base capture uses local integration branch before origin; deterministic commands use async execution with timeout; abort cancels active sessions and subprocesses; reviewer primitive maps verdicts without mutating unrelated state; merge primitive honors branch-group targets; primitive audit includes run id, node id, and effective principal. +- **Verification:** Primitive tests pass independently of workflow graph tests; old executor characterization tests still pass while extraction is in progress. + +### U2. Workflow Run State Authority + +- **Goal:** Create one durable runtime state model for active workflow runs, side-effect boundaries, node progress, active sessions, recovery events, branch rows, and foreach instances. +- **Requirements:** R6, R9, R10. +- **Dependencies:** U1. +- **Files:** `packages/core/src/db.ts`, `packages/core/src/store.ts`, `packages/core/src/types.ts`, `packages/engine/src/workflow-task-runtime.ts`, `packages/engine/src/workflow-graph-executor.ts`, `packages/engine/src/workflow-graph-branches.ts`, `packages/engine/src/workflow-graph-foreach.ts`, `packages/core/src/__tests__/central-db.test.ts`, `packages/core/src/__tests__/store-workflow-runtime.test.ts` (new), `packages/engine/src/__tests__/workflow-task-runtime.test.ts`. +- **Approach:** Add or consolidate top-level workflow run records around the existing branch and foreach persistence. Persist run id, workflow id, current node, side-effect-started marker, active primitive/session ids, effective principal, terminal outcome, recovery event queue, and last activity timestamp. Keep branch and step-instance tables as detail tables keyed by the same run id. +- **Patterns to follow:** `workflow_run_branches`, `workflow_run_step_instances`, `runAuditEvents`, `transitionPending`. +- **Test scenarios:** a run starts once for a task; current node updates idempotently; active session ids clear on terminal outcomes; stale run rows prune without deleting current run rows; side-effect-started prevents fallback reruns; recovery events are append-only or idempotently consumed; restart can reconstruct active branches and foreach instances. +- **Verification:** Store/runtime tests prove workflow run state is sufficient to decide resume, abort, or park without inspecting `TaskExecutor` in-memory sets. + +### U3. Built-In Workflow IR Full Lifecycle Expression + +- **Goal:** Expand built-in workflows so they encode all default task lifecycle policy currently hidden in executor, triage, workflow-step orchestration, review handoff, merge handling, and recovery. +- **Requirements:** R1, R2, R4, R6, R7. +- **Dependencies:** U1, U2. +- **Files:** `packages/core/src/builtin-coding-workflow-ir.ts`, `packages/core/src/builtin-stepwise-coding-workflow-ir.ts`, `packages/core/src/builtin-pr-workflow-ir.ts`, `packages/core/src/workflow-ir-types.ts`, `packages/core/src/workflow-ir.ts`, `packages/core/src/builtin-workflows.ts`, `packages/core/src/__tests__/builtin-coding-workflow-ir.test.ts` (new), `packages/core/src/__tests__/builtin-stepwise-coding-workflow-ir.test.ts`, `packages/core/src/__tests__/workflow-ir.test.ts`, `packages/core/src/__tests__/workflow-ir-loop.test.ts`, `packages/core/src/__tests__/workflow-ir-foreach.test.ts`. +- **Approach:** Add built-in node kinds or reserved node configs for planning/spec generation, spec review, todo capacity hold, stale-spec handling, dependency gate, worktree preparation, implementation session, deterministic verification, pre-merge workflow checks, revision/fix loop, review handoff, merge/manual-required routing, post-merge checks, and recovery router. Prefer explicit built-in node kinds for non-authorable engine primitives; use prompt/script/code nodes for authorable behavior. +- **Patterns to follow:** Existing `parse-steps`, `foreach`, `loop`, `step-review`, `code`, PR node validation, and workflow setting declarations. +- **Test scenarios:** built-in coding workflow validates; null workflow selection resolves to the built-in workflow; default workflow includes every legacy phase in a phase-map test; stepwise workflow includes parse/foreach/review/rework; `executionMode: "fast"` bypasses pre-merge checks but not post-merge checks; `autoMerge:false` routes to manual-required hold; branch-group member integration routes to member landing; malformed built-in IR fails tests at parse time. +- **Verification:** Core tests make built-in IR the explicit compatibility contract before runtime dispatch changes. + +### U4. Node Handler Runtime + +- **Goal:** Replace production legacy seam handlers with node handlers that call runtime primitives directly. +- **Requirements:** R1, R4, R5, R8, R10. +- **Dependencies:** U1, U2, U3. +- **Files:** `packages/engine/src/workflow-node-handlers.ts`, `packages/engine/src/workflow-task-runtime.ts`, `packages/engine/src/workflow-graph-executor.ts`, `packages/engine/src/pr-nodes.ts`, `packages/engine/src/code-node-runner.ts`, `packages/engine/src/__tests__/workflow-node-handlers.test.ts`, `packages/engine/src/__tests__/workflow-node-handler-extensions.test.ts`, `packages/engine/src/__tests__/workflow-verdict-provider-extensions.test.ts`, `packages/engine/src/__tests__/code-node.test.ts`. +- **Approach:** Implement handlers for the new built-in lifecycle nodes and rework existing prompt/script/gate behavior so seam configuration is no longer a production path. Custom prompt/script/gate/code nodes continue to use custom-node machinery. Handler results should produce explicit outcome values like `spec-approved`, `capacity-blocked`, `implementation-complete`, `revision-requested`, `workflow-check-failed`, `manual-required`, `merge-timeout`, `recover-resume`, and `recover-park-review`. +- **Patterns to follow:** `createParseStepsHandler`, `createStepReviewHandler`, `createPrNodeHandlers`, plugin node-handler fallback/degradation. +- **Test scenarios:** planning node writes expected artifacts and step projection; implementation node invokes coding primitive once; verification node routes revise vs approve; workflow-check node handles gate/advisory modes; merge node honors manual-required; recovery node consumes recovery event and routes to the correct edge; missing primitive wiring fails closed; plugin node handler degradation never bypasses built-in guard nodes. +- **Verification:** Node handler tests cover success/failure for every built-in lifecycle node kind without using `TaskExecutor.execute()`. + +### U5. Planning And Triage Workflow Nodes + +- **Goal:** Move built-in triage/specification behavior out of `TriageProcessor` lifecycle orchestration and into planning workflow nodes. +- **Requirements:** R1, R2, R4, R6, R11. +- **Dependencies:** U1, U2, U3, U4. +- **Files:** `packages/engine/src/triage.ts`, `packages/engine/src/workflow-node-handlers.ts`, `packages/engine/src/workflow-task-runtime.ts`, `packages/engine/src/agent-tools.ts`, `packages/engine/src/__tests__/triage-core.test.ts`, `packages/engine/src/__tests__/triage-soft-delete-abort.test.ts`, `packages/engine/src/__tests__/workflow-planning-nodes.test.ts` (new), `packages/engine/src/__tests__/reliability-interactions/ghost-bug-preflight.test.ts`. +- **Approach:** Extract triage session mechanics into planning primitives, then create planning/spec-review nodes in built-in workflows. The scheduler can still identify intake tasks and respect capacity, but planning behavior is a workflow region. Tools like task creation, task document writing, spec review, ghost-bug preflight, duplicate detection, and starved refinement recovery must be called by nodes or recovery branches. +- **Patterns to follow:** Existing prompt layer construction in `triage.ts`, `fn_task_create` tool surface, task document APIs, ghost-bug preflight tests. +- **Test scenarios:** intake task runs planning node and writes `PROMPT.md`; spec review APPROVE routes to todo hold; REVISE routes back to planning; split-into-subtasks path creates child tasks and parks parent according to existing behavior; soft-deleted task aborts planning; orphaned planning session recovery wakes recovery node instead of directly clearing status. +- **Verification:** Triage behavior tests pass through workflow runtime, and no production path calls `TriageProcessor.specifyTask()` as lifecycle owner. + +### U6. Scheduler, Routing, And Dispatch Cutover + +- **Goal:** Route all executable work through `WorkflowTaskRuntime` while keeping scheduling, node routing, concurrency, and remote runtime boundaries in the engine. +- **Requirements:** R1, R3, R8, R10, R11. +- **Dependencies:** U2, U3, U4, U5. +- **Files:** `packages/engine/src/scheduler.ts`, `packages/engine/src/runtimes/in-process-runtime.ts`, `packages/engine/src/runtimes/remote-node-runtime.ts`, `packages/engine/src/runtimes/remote-node-client.ts`, `packages/engine/src/project-manager.ts`, `packages/engine/src/hybrid-executor.ts`, `packages/engine/src/executor.ts`, `packages/engine/src/__tests__/scheduler-node-routing.test.ts`, `packages/engine/src/__tests__/hybrid-executor-multi-node-routing.test.ts`, `packages/engine/src/runtimes/__tests__/remote-node-runtime.test.ts`, `packages/engine/src/__tests__/cross-node-claim-mutex.integration.test.ts`. +- **Approach:** Replace `TaskExecutor.execute()` dispatch ownership with runtime dispatch ownership. Keep remote nodes as remote workflow runtimes: local central routing sends execute requests/events, but the remote side resolves and executes the workflow. Concurrency limits are acquired by the runtime before side-effecting nodes and released by terminal/paused outcomes. Existing task-moved and heartbeat wake paths should call the runtime facade. +- **Patterns to follow:** `ProjectRuntime` boundary, `RemoteNodeRuntime`, `executingTaskLock`, node routing policy tests. +- **Test scenarios:** local task dispatch starts one workflow run; remote task dispatch calls remote execute API and streams events; duplicate dispatch is claimed once; unavailable node policy still blocks or falls back according to settings; heartbeat completion wakes tasks by effective principal; workflow runtime respects global pause and engine pause before starting new side effects. +- **Verification:** Scheduler and runtime tests show no dispatch path invokes legacy executor lifecycle. + +### U7. Recovery As Workflow Policy + +- **Goal:** Convert execution self-healing and restart recovery from direct lifecycle mutations into workflow recovery events handled by recovery nodes. +- **Requirements:** R6, R7, R9, R10, R11. +- **Dependencies:** U2, U4, U6. +- **Files:** `packages/engine/src/self-healing.ts`, `packages/engine/src/restart-recovery-coordinator.ts`, `packages/engine/src/recovery-policy.ts`, `packages/engine/src/auto-recovery.ts`, `packages/engine/src/auto-recovery-handlers/branch-worktree.ts`, `packages/engine/src/auto-recovery-handlers/contamination.ts`, `packages/engine/src/auto-recovery-handlers/file-scope.ts`, `packages/engine/src/workflow-task-runtime.ts`, `packages/engine/src/__tests__/self-healing.test.ts`, `packages/engine/src/__tests__/workflow-recovery-nodes.test.ts` (new), `packages/engine/src/__tests__/reliability-interactions/workflow-interpreter-cutover.test.ts`, `packages/engine/src/__tests__/reliability-interactions/active-worktree-removal-liveness.test.ts`. +- **Approach:** Keep sweep timing and detection in the engine, but change detected conditions into typed recovery events on the workflow run. Built-in recovery nodes handle completed-but-stranded work, failed pre-merge checks, no-progress no-task-done, partial-progress no-task-done, in-progress limbo, missing worktree review failure, contamination recovery, file-scope recovery, branch conflict recovery, stale planning, and orphaned sessions. Some non-task agent/heartbeat recovery can remain engine-owned when it is not task lifecycle policy. +- **Patterns to follow:** `recovery-policy.ts`, run audit events, `allowsAutoMergeProcessing`, branch-group recovery tests. +- **Test scenarios:** completed stranded task routes through recovery node to review handoff; failed pre-merge check routes to fix loop; no-progress failure cleanly requeues through todo hold; partial-progress failure preserves progress; contamination recovery routes to human review when unique commits remain; `autoMerge:false` in-review tasks are inspected but not moved backward; branch-group member landing uses group target; recovery events are idempotent across repeated sweeps. +- **Verification:** Reliability interaction tests prove sweeps no longer mutate execution lifecycle directly and recovery outcomes are authored by workflow nodes. + +### U8. Effective Principal And Permission Surface Audit + +- **Goal:** Re-key every execution identity, permission, serialization, wakeup, and hot-swap reader to use the node's effective principal. +- **Requirements:** R8, R11. +- **Dependencies:** U4, U6. +- **Files:** `packages/engine/src/executor.ts`, `packages/engine/src/agent-heartbeat.ts`, `packages/engine/src/workflow-task-runtime.ts`, `packages/engine/src/session-skill-context.ts`, `packages/engine/src/agent-tools.ts`, `packages/engine/src/action-gate.ts` or adjacent action gate modules, `packages/core/src/workflow-ir-resolver.ts`, `packages/core/src/workflow-columns-settings.ts`, `packages/engine/src/__tests__/agent-workflow-tools-exposure.test.ts`, `packages/engine/src/__tests__/agent-tools-workflow-settings.test.ts`, `packages/engine/src/__tests__/executor-column-agent-custom-node.test.ts`, `packages/engine/src/__tests__/effective-node.test.ts`. +- **Approach:** Centralize effective principal resolution for each node invocation. Pass the resolved identity through session setup, prompt/persona/memory, permission gates, heartbeat serialization in both directions, resume queries, change detection, audit attribution, and missing-agent fallback. Search all `assignedAgentId` execution readers and either re-key them or document why they are task metadata readers only. +- **Execution note:** Apply the full checklist from `docs/solutions/architecture-patterns/per-entity-execution-principal-override-blast-radius.md`. +- **Patterns to follow:** Existing column-agent resolver, `resolveEffectiveAgent`, heartbeat serialization guards. +- **Test scenarios:** override column agent governs custom prompt node, planning node, coding node, step-execute, and review node; defer mode loses to own agent/model settings; missing agent degrades; heartbeat does not run concurrently with its effective task session; resume wakes by effective principal; permission gate sees the effective agent. +- **Verification:** Identity matrix tests cover mode x surface x own-settings x missing-agent. + +### U9. Old Orchestration Deletion + +- **Goal:** Remove production lifecycle orchestration from `TaskExecutor`, `TriageProcessor`, graph seams, and self-healing callbacks after workflow runtime and built-ins cover the behavior. +- **Requirements:** R4, R10, R11. +- **Dependencies:** U1, U2, U3, U4, U5, U6, U7, U8. +- **Files:** `packages/engine/src/executor.ts`, `packages/engine/src/triage.ts`, `packages/engine/src/workflow-graph-task-runner.ts`, `packages/engine/src/workflow-authoritative-driver.ts`, `packages/engine/src/workflow-task-runtime.ts`, `packages/engine/src/workflow-node-handlers.ts`, `packages/engine/src/__tests__/workflow-graph-entry.test.ts` if present, `packages/engine/src/__tests__/stepwise-workflow-parity.test.ts`, `packages/engine/src/__tests__/executor-core.test.ts`. +- **Approach:** Delete `maybeExecuteWorkflowGraph`, `runImplementationPhase`, `graphCompletionInterceptors`, production `WorkflowLegacySeams`, authoritative-driver fallback, dual-observe gating for production authority, triage lifecycle loops, and self-healing callbacks that directly call executor/triage recovery methods. Keep or rename files only when they contain primitives. Any remaining `executor.ts` code should be mechanical session/worktree/tool support, not lifecycle owner. +- **Patterns to follow:** Search-based deletion gates used in existing no-nohup and lazy-view inventory tests. +- **Test scenarios:** search tests fail if production code references `graphCompletionInterceptors`, `runImplementationPhase`, `maybeExecuteWorkflowGraph`, `WorkflowAuthoritativeDriver`, or production `WorkflowLegacySeams`; runtime tests prove null workflow selection executes built-in workflow; no tests need legacy fallback to pass. +- **Verification:** A grep audit plus tests show no production lifecycle path bypasses workflow runtime. + +### U10. Surface Enumeration And Gate Test Matrix + +- **Goal:** Build the regression matrix required for a big-bang cutover across every known execution surface. +- **Requirements:** R7, R8, R10, R11. +- **Dependencies:** U3, U4, U5, U6, U7, U8, U9. +- **Files:** `packages/engine/src/__tests__/workflow-runtime-cutover.matrix.test.ts` (new), `packages/engine/src/__tests__/workflow-work-engine-dispatch.test.ts`, `packages/engine/src/__tests__/pr-workflow-e2e.test.ts`, `packages/engine/src/__tests__/workflow-step-integration-cwd.test.ts`, `packages/engine/src/__tests__/reliability-interactions/workflow-and-file-scope.test.ts`, `packages/engine/src/__tests__/reliability-interactions/owning-node-unavailable-interactions.test.ts`, `packages/engine/src/__tests__/reliability-interactions/cross-node-assignment-wake.test.ts`, `packages/engine/src/__tests__/reliability-interactions/workflow-interpreter-cutover.test.ts`, `packages/engine/vitest.config.ts`. +- **Approach:** Enumerate surfaces explicitly and add targeted tests rather than relying on old executor suites. Cover default workflow, stepwise workflow, custom prompt/script/gate/code nodes, plugin work-engine extension, PR nodes, planning nodes, recovery nodes, local runtime, remote runtime, pause/cancel, branch-group integration, auto-merge global/per-task matrix, test mode/mock provider, workflow settings fallback, and file-scope/squash guards. +- **Patterns to follow:** `docs/testing.md` Surface Enumeration checklist, existing reliability-interactions tests, thin trusted gate constraints. +- **Test scenarios:** every surface above has at least one happy path and one failure/recovery path; gate suite gets only high-signal cutover invariants; slow/full-suite coverage remains opt-in unless evidence supports gate admission. +- **Verification:** `pnpm test:gate`, `pnpm lint`, `pnpm build`, and targeted engine/core suites pass without relying on old executor fallback. + +### U11. Documentation And Concept Refresh + +- **Goal:** Update docs and vocabulary so the repository describes workflow-native execution as the actual architecture. +- **Requirements:** R1, R2, R3, R6, R9. +- **Dependencies:** U9, U10. +- **Files:** `docs/architecture.md`, `docs/workflow-steps.md`, `docs/testing.md`, `docs/agents.md`, `docs/settings-reference.md`, `CONCEPTS.md`, `packages/engine/CHANGELOG.md`, `packages/engine/src/__tests__/workflow-settings-fallback-alignment.test.ts`. +- **Approach:** Replace interpreter-scaffold and parity-gated rollout language with the new ownership boundary. Add Concepts entries for Workflow runtime, Runtime primitive, Recovery event, and Built-in lifecycle node. Document what remains engine-owned and what is workflow-owned. Update testing docs with the new cutover matrix and recovery-node pattern. +- **Patterns to follow:** Existing Concepts style, Architecture package responsibility tables, workflow-steps IR sections. +- **Test scenarios:** documentation inventory tests still pass; lazy-heavy view inventory is unaffected; workflow settings docs describe built-in workflow settings as runtime-owned defaults. +- **Verification:** Docs use current code names and no longer describe legacy executor/reviewer/merger/scheduler flow as authoritative for task lifecycle. + +--- + +## Surface Enumeration + +The cutover must cover these surfaces before it can be considered complete: + +| Surface | Required coverage | +|---|---| +| Default coding workflow | planning, implementation, verification, review, merge, failure, recovery | +| Stepwise coding workflow | parse-steps, foreach, step-execute, step-review, rework, integration conflict | +| Custom workflows | prompt, script, gate, code, split/join, loop, plugin node handler fallback | +| Planning/triage | new task intake, refinement, split subtasks, spec review, duplicate/ghost preflight | +| Recovery | completed stranded, failed pre-merge, no-progress, partial-progress, limbo, contamination, branch conflict | +| Merge | global auto-merge on/off, per-task override, manual-required, branch-group member landing, group promotion | +| Runtime routing | local runtime, remote node runtime, unavailable-node policy, cross-node assignment wake | +| Identity | assigned agent, column-agent defer, column-agent override, missing agent fallback, heartbeat serialization | +| Safety guards | hard cancel, file-scope guard, squash overlap, worktree ownership, process abort | +| Settings | workflow settings defaults, task overrides, test mode/mock provider, model lane resolution | +| Plugins | workflow work engine extension, node-handler extension degradation, PR node dependencies | + +--- + +## System-Wide Impact + +This is a high-risk architectural replacement. It changes the mental model for every developer-facing execution path: instead of learning `scheduler -> executor -> workflow steps -> review -> merge -> self-healing`, readers should learn `scheduler/router -> workflow runtime -> graph nodes -> primitives`. The user-visible behavior should remain recognizable, but the authorship surface changes: default behavior is inspectable in built-in workflows, and recovery decisions are visible as graph edges or recovery nodes. + +Stakeholders affected: + +- Developers need new runtime primitive and node-handler patterns. +- Reviewers need to inspect built-in workflow IR as executable policy. +- Operators/support gain durable workflow run state but lose the old executor-log-only mental model. +- Plugin authors must treat workflow extension handlers as first-class execution participants. +- Users should see fewer stranded tasks because recovery decisions become explicit and inspectable. + +--- + +## Risks & Mitigations + +| Risk | Mitigation | +|---|---| +| The big-bang branch is too large to reason about | Keep implementation units independently testable, but require final deletion of legacy orchestration before merge. | +| Hidden lifecycle behavior in `executor.ts` is missed | U3 phase-map tests and U10 surface enumeration must map every current branch before deletion. | +| Recovery becomes less safe during the cutover | U2 run state and U7 recovery events land before old sweeps are demoted. | +| Auto-merge/manual-required regresses | Preserve `allowsAutoMergeProcessing`, branch-group merge target resolution, and merger authority as substrate guards; cross global x per-task tests. | +| Effective principal is partially re-keyed | U8 applies the full blast-radius checklist and grep audit for old identity readers. | +| Built-in workflow IR becomes unreadable | Use reusable built-in subgraphs or explicit node kinds for engine-owned primitives; do not encode everything as opaque prompt nodes. | +| Tests become slow or flaky | Put only high-signal cutover invariants in the gate; keep broader matrix targeted and avoid real polling/network. | + +--- + +## Acceptance Examples + +- AE1. Given a task with no workflow selection, when the scheduler dispatches it, then `WorkflowTaskRuntime` resolves `builtin:coding` and no production code invokes legacy executor lifecycle fallback. +- AE2. Given a new task in intake, when planning starts, then a workflow planning node produces/reviews the spec and routes to todo hold without `TriageProcessor` owning lifecycle control. +- AE3. Given a coding task completes implementation, when verification fails with revision feedback, then workflow edges route back to the implementation/fix node and preserve existing progress semantics. +- AE4. Given `settings.autoMerge:false` and no per-task override, when the merge gate is reached, then the workflow routes to manual-required/in-review hold and self-healing does not move the task backward. +- AE5. Given `settings.autoMerge:false` and `task.autoMerge:true`, when the merge gate is reached, then trigger processing is allowed and the merge node proceeds through the normal merge primitive. +- AE6. Given a user moves an active task from `in-progress` to `todo`, when any node-owned session or command is active, then the runtime aborts it and parks the task with user-paused hard-cancel semantics. +- AE7. Given a crashed run with side effects already started, when restart recovery runs, then the scheduler wakes a workflow recovery event and does not rerun old executor orchestration. +- AE8. Given a column-agent override on a workflow column, when any node in that column runs, then session identity, permission gates, heartbeat serialization, and audit attribution all use the effective principal. +- AE9. Given a remote node owns a project, when central dispatches a task, then the remote node executes the workflow runtime and central only observes/routes events. + +--- + +## Documentation / Operational Notes + +- The final branch changes architecture docs, not just code. `docs/architecture.md` must describe workflow runtime as authoritative. +- `docs/workflow-steps.md` must stop describing `workflowGraphExecutor` and `workflowInterpreterAuthoritative` as rollout flags for production authority once the cutover lands. +- `CONCEPTS.md` should add Workflow runtime, Runtime primitive, Recovery event, and Built-in lifecycle node. +- This change affects private packages, but if behavior in published `@runfusion/fusion` changes through CLI/runtime behavior, the implementation work should add an appropriate changeset. + +--- + +## Sources & Research + +- `docs/plans/2026-06-07-001-refactor-workflow-runtime-cutover-plan.md` is the closest prior plan and is superseded here by the user's big-bang requirement. +- `docs/plans/2026-06-03-002-feat-workflow-interpreter-cutover-plan.md` documents the earlier seam/fallback interpreter path that this plan removes. +- `docs/plans/2026-06-04-001-feat-step-inversion-workflow-modelable-steps-plan.md` documents parse-steps, foreach, step-review, code nodes, and step-instance persistence. +- `docs/workflow-steps.md` documents current workflow IR, step inversion, workflow steps, parity flags, and reliability invariants. +- `CONCEPTS.md` defines Task, Default workflow, Step instance, In-review, Merge queue, Manual-required, Self-healing sweep, Column agent, and Effective agent. +- `docs/solutions/architecture-patterns/per-entity-execution-principal-override-blast-radius.md` supplies the identity blast-radius checklist used by U8. +- `docs/solutions/logic-errors/per-task-auto-merge-override-ignored-by-trigger-gates.md` supplies the additive trigger-gate lesson used by merge/recovery tests. +- `docs/solutions/logic-errors/files-changed-inflated-by-origin-first-base-commit.md` supplies the local-first fork-point invariant used by U1. +- `packages/engine/src/workflow-task-runtime.ts`, `workflow-graph-executor.ts`, `workflow-node-handlers.ts`, `workflow-graph-foreach.ts`, and `workflow-graph-loop.ts` are the existing workflow runtime foundation. +- `packages/engine/src/executor.ts`, `triage.ts`, `self-healing.ts`, `scheduler.ts`, and `runtimes/in-process-runtime.ts` are the legacy ownership surfaces to subsume. diff --git a/docs/workflow-steps.md b/docs/workflow-steps.md index d78c37b364..fd954f4bfa 100644 --- a/docs/workflow-steps.md +++ b/docs/workflow-steps.md @@ -28,23 +28,21 @@ Out of scope for v1: - Execution history/runtime traces - Migration tooling for future schema versions (future versions should use explicit `schemaVersion` migrations) -### Workflow Graph Executor (interpreter scaffold) +### Workflow Runtime -FN-5766 adds a **flagged-off** interpreter scaffold in `@fusion/engine` (`WorkflowGraphExecutor`) plus a built-in coding lifecycle IR in `@fusion/core` (`BUILTIN_CODING_WORKFLOW_IR`). +The workflow runtime is the authoritative execution path for task lifecycle work. `WorkflowGraphExecutor` owns graph traversal and routing; node handlers call runtime primitives supplied by `TaskExecutor` for side-effecting operations such as planning, coding sessions, review, step execution/reset, merge requests, transitions, and audit. -- Feature flag key: `experimentalFeatures.workflowGraphExecutor` -- Default: **OFF** -- OFF behavior is strict no-op (no task mutations, no session/git side effects), so the legacy imperative pipeline remains authoritative. +The engine remains the substrate for scheduler dispatch, routing claims, persistence, concurrency limits, process supervision, storage, and audit plumbing. Lifecycle policy belongs in built-in or custom workflows. The default built-in catalog entry `builtin:coding` is backed by the canonical `BUILTIN_CODING_WORKFLOW_IR`, which is also the resolver/runtime fallback for tasks with no workflow selection, an explicit default selection, or a missing/corrupt custom selection. That IR currently encodes the legacy lifecycle path as graph stages: -- `triage` → `execute` → `review` → `merge` → `end` +- `triage/planning` → `execute` → `review` → `merge` → `end` -`builtin:stepwise-coding` is a separate opt-in graph-mode variant backed by `BUILTIN_STEPWISE_CODING_WORKFLOW_IR`; it keeps the same lifecycle columns/traits while modeling per-step parse/execute/review as authored graph structure, and requires the `workflowGraphExecutor` flag at runtime. +`builtin:stepwise-coding` is a separate graph variant backed by `BUILTIN_STEPWISE_CODING_WORKFLOW_IR`; it keeps the same lifecycle columns/traits while modeling per-step parse/execute/review/rework as authored graph structure. -#### Interpreter-parity gating criterion +#### Runtime invariant criterion -Interpreter authority is gated on parity: interpreter-driven coding runs must match legacy behavior for observable task transitions and reliability invariants (file-scope guards including `FileScopeViolationError`, squash/merge contract, self-healing expectations, `autoMerge:false` terminal-until-merged, and `moveTask(in-progress→todo)` hard-cancel semantics). +Workflow-driven coding runs must preserve observable task transitions and reliability invariants: file-scope guards including `FileScopeViolationError`, squash/merge contract, recovery expectations, `autoMerge:false` terminal-until-merged, and `moveTask(in-progress→todo)` hard-cancel semantics. For grouped branch flows (`branch_groups`), auto-merge precedence is split: per-task `autoMerge` controls member→group-integration landing, while group `autoMerge` controls group→default-branch promotion eligibility. @@ -485,13 +483,14 @@ Prompt-mode workflow agents should emit a trailing JSON object: - Backward compatibility remains for legacy prose-only responses via heuristic fallback (`REQUEST REVISION` and approval keywords). - If neither structured JSON nor fallback prose can be interpreted, output is recorded as `malformed` (no inferable verdict) instead of hard-failing the task. -## Workflow Graph Executor (interpreter) +## Workflow Graph Executor -The experimental `workflowGraphExecutor` path remains **default OFF** and only runs when `settings.experimentalFeatures.workflowGraphExecutor = true`. +Workflow graph execution is the task lifecycle runtime. `TaskExecutor` pins `workflowGraphExecutor` for the run and unselected tasks resolve to `builtin:coding`. -When enabled, interpreter nodes dispatch through DI-backed legacy seams: -- `prompt` / `script` nodes with `config.seam` dispatch to `execute`, `review`, `merge`, or `schedule` -- `gate` nodes evaluate context-key expectations and return success/failure outcomes +Default node dispatch: +- `prompt` / `script` nodes with `config.seam` dispatch through workflow runtime primitives (`planning`, `execute`, `review`, `merge`, `schedule`, `step-execute`) +- `step-review`, `parse-steps`, `code`, and PR nodes use their dedicated primitive/dependency adapters +- `gate` nodes evaluate context-key expectations or run configured executable checks Traversal semantics: - edge with no condition or `success` routes on success @@ -500,18 +499,13 @@ Traversal semantics: - unsupported conditions throw `WorkflowIrError` - per-node retries are bounded and deterministic -Parity coverage includes flag-OFF no-op behavior, lifecycle ordering parity vs legacy seams, merge/file-scope-like failure routing, and downstream halt behavior for hard-cancel/self-healing style failures. +Coverage includes lifecycle ordering, primitive invocation, merge/file-scope failure routing, and downstream halt behavior for hard-cancel/recovery style failures. -### Interpreter-authoritative cutover +### Workflow-native Cutover -A second default-OFF flag, `experimentalFeatures.workflowInterpreterAuthoritative`, promotes the interpreter from shadow/selected-workflow sequencing to the **authoritative** lifecycle driver for default coding tasks. +`TaskExecutor.execute()` gives graph routing first claim. The graph runtime resolves a workflow selection, degrading missing/unselected workflows to `builtin:coding`, and parks interpreter failures as workflow failures instead of re-running the old imperative lifecycle. -The cutover stays opt-in, guarded, and reversible: -- **Default OFF:** legacy executor/reviewer/merger/scheduler flow remains authoritative. -- **Guarded ON:** the engine only routes through the authoritative driver when `evaluateInterpreterCutoverReadiness(...)` reports ready. The guard consumes explicit rollout evidence (cutover flag enabled, dual-observe enabled, non-empty parity observations, zero unresolved drift). -- **Rollback:** turning `workflowInterpreterAuthoritative` back OFF immediately restores the legacy path; no migration or cleanup step is required. - -When the guard passes, the runtime binds real DI seams from `TaskExecutor` into the built-in coding IR and drives `BUILTIN_CODING_WORKFLOW_IR` through `WorkflowGraphExecutor`. The interpreter does **not** reimplement lifecycle behavior: it delegates execute/review/merge to the same legacy seams already used by the imperative path. +The legacy seam adapter remains as a compatibility layer for older tests and callers, but authoritative node execution uses `WorkflowRuntimePrimitives`. The built-in coding workflow now includes explicit planning before execute/review/merge. Reliability invariants preserved under authoritative mode: - file-scope enforcement including `FileScopeViolationError` diff --git a/packages/core/src/builtin-coding-workflow-ir.ts b/packages/core/src/builtin-coding-workflow-ir.ts index d4515a1a97..a4f663cddd 100644 --- a/packages/core/src/builtin-coding-workflow-ir.ts +++ b/packages/core/src/builtin-coding-workflow-ir.ts @@ -18,9 +18,10 @@ import { builtinPromptConfig } from "./builtin-workflow-prompts.js"; * done = complete * archived = archived * - * The seam nodes (execute/review/merge) are placed in their columns; the graph - * walk (edges) is byte-identical to the prior v1 coding pipeline, so the graph - * executor continues to drive execute → review → merge unchanged. + * The lifecycle seam nodes are placed in their columns. Planning is explicit so + * the built-in workflow owns the specification phase rather than relying on + * triage code that runs outside the graph; execute/review/merge keep the same + * observable pipeline and failure routing. */ const RAW_BUILTIN_CODING_WORKFLOW_IR: WorkflowIr = { version: "v2", @@ -47,6 +48,12 @@ const RAW_BUILTIN_CODING_WORKFLOW_IR: WorkflowIr = { ], nodes: [ { id: "start", kind: "start", column: "triage" }, + { + id: "planning", + kind: "prompt", + column: "triage", + config: builtinPromptConfig("planning", "Plan / specify"), + }, { id: "execute", kind: "prompt", @@ -58,10 +65,12 @@ const RAW_BUILTIN_CODING_WORKFLOW_IR: WorkflowIr = { { id: "end", kind: "end", column: "done" }, ], edges: [ - { from: "start", to: "execute" }, + { from: "start", to: "planning" }, + { from: "planning", to: "execute", condition: "success" }, { from: "execute", to: "review", condition: "success" }, { from: "review", to: "merge", condition: "success" }, { from: "merge", to: "end", condition: "success" }, + { from: "planning", to: "end", condition: "failure" }, { from: "execute", to: "end", condition: "failure" }, { from: "review", to: "end", condition: "failure" }, { from: "merge", to: "end", condition: "failure" }, diff --git a/packages/core/src/workflow-compiler.ts b/packages/core/src/workflow-compiler.ts index dcf7ce0363..b920782a3c 100644 --- a/packages/core/src/workflow-compiler.ts +++ b/packages/core/src/workflow-compiler.ts @@ -17,7 +17,7 @@ export class WorkflowCompileError extends Error { /** Seam anchor kinds, encoded on IR nodes as `config.seam`. These map to the * fixed execute → review → merge pipeline and are not emitted as steps. */ -const SEAM_NAMES = new Set(["execute", "review", "merge"]); +const SEAM_NAMES = new Set(["planning", "execute", "review", "merge"]); function seamOf(node: WorkflowIrNode): string | undefined { const seam = node.config?.seam; @@ -111,7 +111,7 @@ export function validateLinearity(ir: WorkflowIr): WorkflowCompileError | null { // seams as a fixed execute → review → merge boundary (merge flips pre- to // post-merge), so out-of-order or duplicate seams would compile inconsistently // with the runtime contract. - const expectedSeamOrder = ["execute", "review", "merge"] as const; + const expectedSeamOrder = ["planning", "execute", "review", "merge"] as const; const seenSeams = new Set(); let nextExpectedSeamIndex = 0; const visited = new Set(); diff --git a/packages/engine/src/__tests__/runtime-primitives.test.ts b/packages/engine/src/__tests__/runtime-primitives.test.ts new file mode 100644 index 0000000000..8b9810a4fe --- /dev/null +++ b/packages/engine/src/__tests__/runtime-primitives.test.ts @@ -0,0 +1,57 @@ +import { describe, expect, it } from "vitest"; + +import { markSideEffectsStarted, primitiveNodeContext } from "../runtime-primitives.js"; + +describe("runtime primitives", () => { + it("creates a workflow primitive context from a run and node", () => { + const run = { + runId: "run-1", + taskId: "FN-1", + workflowId: "coding", + }; + const node = { + id: "execute", + kind: "prompt" as const, + column: "in-progress", + config: { prompt: "implement" }, + }; + + const ctx = primitiveNodeContext(run, node, { + effectivePrincipalId: "agent:builder", + attempt: 2, + context: { priorOutcome: "revise" }, + }); + + expect(ctx).toEqual({ + run, + node: { + node, + effectivePrincipalId: "agent:builder", + attempt: 2, + context: { priorOutcome: "revise" }, + }, + }); + }); + + it("marks side effects on an immutable context copy", () => { + const ctx = primitiveNodeContext( + { + runId: "run-1", + taskId: "FN-1", + workflowId: "coding", + }, + { id: "execute", kind: "prompt" as const }, + ); + + const marked = markSideEffectsStarted(ctx); + + expect(marked).toEqual({ + ...ctx, + run: { + ...ctx.run, + sideEffectsStarted: true, + }, + }); + expect(ctx.run.sideEffectsStarted).toBeUndefined(); + }); +}); diff --git a/packages/engine/src/__tests__/workflow-graph-executor-parity.test.ts b/packages/engine/src/__tests__/workflow-graph-executor-parity.test.ts index 3d7ad33a18..50263fedec 100644 --- a/packages/engine/src/__tests__/workflow-graph-executor-parity.test.ts +++ b/packages/engine/src/__tests__/workflow-graph-executor-parity.test.ts @@ -1,7 +1,7 @@ // ───────────────────────────────────────────────────────────────────────────── // PARITY SUBJECT (test-file ownership, U7 / KTD-9): // This suite owns DEFAULT-WORKFLOW BYTE-IDENTITY parity — it proves the graph -// executor reproduces the legacy monolithic execute → review → merge seam +// executor reproduces the workflow-native planning → execute → review → merge seam // sequence exactly (the parity ORACLE per KTD-1). It deliberately does NOT // cover per-step / updateStep-trajectory parity. // @@ -26,6 +26,9 @@ const task = { id: "FN-5767" } as TaskDetail; function runLegacy(seams: WorkflowLegacySeams) { return async () => { const events: string[] = []; + const planning = await seams.planning(task, {}); + events.push(`planning:${planning.outcome}`); + if (planning.outcome !== "success") return events; const execute = await seams.execute(task, {}); events.push(`execute:${execute.outcome}`); if (execute.outcome !== "success") return events; @@ -47,7 +50,7 @@ describe("WorkflowGraphExecutor interpreter-parity", () => { expect(prompt).not.toHaveBeenCalled(); }); - it("matches legacy execute-review-merge success path", async () => { + it("matches default planning-execute-review-merge success path", async () => { const events: string[] = []; const seams: WorkflowLegacySeams = { planning: async () => ({ outcome: "success" }), @@ -82,7 +85,7 @@ describe("WorkflowGraphExecutor interpreter-parity", () => { const executor = new WorkflowGraphExecutor({ seams }); const result = await executor.run(task, { experimentalFeatures: { workflowGraphExecutor: true } }); expect(result.outcome).toBe("failure"); - expect(legacyEvents).toEqual(["execute:success", "review:success", "merge:failure"]); + expect(legacyEvents).toEqual(["planning:success", "execute:success", "review:success", "merge:failure"]); }); it("preserves autoMerge:false terminal in-review semantics via review failure", async () => { @@ -182,13 +185,16 @@ describe("column-agent feature is invisible when unbound (U7 / R9)", () => { // Bind the invariant to actual executor behavior (PR #1432 review): the // observation below derives from the run-captured seam sequence, so seam // drift fails here instead of being masked by a hard-coded literal. - expect(stages).toEqual(["execute", "review", "merge"]); + expect(stages).toEqual(["planning", "execute", "review", "merge"]); // Legacy authoritative observation: a clean run that lands in `done`/merged. - const legacyObs = buildWorkflowObservationFromTask( - { column: "done", status: "done", review: { verdict: "approve" } }, - { columnSequence: ["todo", "in-progress", "in-review", "done"] }, - ); + const legacyObs = buildWorkflowObservation({ + stageTransitions: ["triage", "planning", "execute", "review", "merge"], + terminalColumn: "done", + terminalStatus: "done", + reviewVerdict: "approve", + mergeOutcome: "merged", + }); // Interpreter (binding-free) observation assembled from the same run. const interpreterObs = buildWorkflowObservation({ stageTransitions: ["triage", ...stages] as WorkflowStage[], diff --git a/packages/engine/src/__tests__/workflow-task-runtime.test.ts b/packages/engine/src/__tests__/workflow-task-runtime.test.ts index 5c73270f2b..8e3b4ba2d2 100644 --- a/packages/engine/src/__tests__/workflow-task-runtime.test.ts +++ b/packages/engine/src/__tests__/workflow-task-runtime.test.ts @@ -3,7 +3,7 @@ import type { Settings, TaskDetail, WorkflowIr } from "@fusion/core"; import { WorkflowTaskRuntime, type WorkflowTaskRuntimeDeps } from "../workflow-task-runtime.js"; import type { WorkflowNodeResult } from "../workflow-graph-executor.js"; -import type { WorkflowLegacySeams } from "../workflow-node-handlers.js"; +import type { PreparedWorktree, WorkflowRuntimePrimitives } from "../runtime-primitives.js"; const task = { id: "FN-9002" } as TaskDetail; const flagOff = { experimentalFeatures: {} } as unknown as Pick; @@ -27,31 +27,69 @@ function selectedIr(): WorkflowIr { }; } -function recordingSeams(calls: string[], overrides: Partial> = {}): WorkflowLegacySeams { - const seam = (name: keyof WorkflowLegacySeams) => async (): Promise => { - calls.push(name); - return overrides[name] ?? { outcome: "success" }; - }; +function recordingPrimitives( + calls: string[], + overrides: Partial> = {}, +): WorkflowRuntimePrimitives { + const prepared: PreparedWorktree = { worktreePath: "/tmp/fusion-worktree" }; return { - planning: seam("planning"), - execute: seam("execute"), - review: seam("review"), - merge: seam("merge"), - schedule: seam("schedule"), + prepareWorktree: async () => { + calls.push("prepare-worktree"); + return { outcome: "success", data: prepared }; + }, + readArtifact: async () => undefined, + writeArtifact: async (_ctx, _task, key) => ({ outcome: "success", data: { key } }), + runPlanningSession: async () => { + calls.push("planning"); + return { outcome: "success", data: { approved: true, artifactKeys: [] } }; + }, + runCodingSession: async () => { + calls.push("execute"); + const override = overrides.execute; + return { + outcome: override?.outcome ?? "success", + value: override?.value ?? "implemented", + contextPatch: override?.contextPatch, + data: { taskDone: override?.outcome !== "failure", modifiedFiles: [] }, + }; + }, + runTaskStep: async () => ({ outcome: "success" }), + resetTaskStep: async () => ({ ok: true }), + runReview: async (_ctx, _task, input) => { + calls.push(input.stepIndex === undefined ? "review" : "step-review"); + return { + outcome: "success", + value: input.stepIndex === undefined ? "in-review" : "approve", + data: { verdict: "APPROVE" }, + }; + }, + runVerification: async () => ({ outcome: "success", data: { verdict: "skipped" } }), + runWorkflowStep: async () => ({ outcome: "success", data: { allPassed: true } }), + updateSteps: async (_ctx, _task, steps) => ({ outcome: "success", data: { count: steps.length } }), + transitionTask: async () => { + calls.push("schedule"); + return { outcome: "success" }; + }, + requestMerge: async () => { + calls.push("merge"); + return { outcome: "success", value: "merged", data: { status: "merged" } }; + }, + abortRun: async () => ({ outcome: "success" }), + audit: () => undefined, }; } describe("WorkflowTaskRuntime", () => { it("requires execution wiring at the type boundary", () => { - // @ts-expect-error WorkflowTaskRuntime is an execution entry point, so seams are required. - const missingSeams: WorkflowTaskRuntimeDeps = { + // @ts-expect-error WorkflowTaskRuntime is an execution entry point, so primitives are required. + const missingPrimitives: WorkflowTaskRuntimeDeps = { store: { getTaskWorkflowSelection: () => undefined, getWorkflowDefinition: async () => undefined, }, runCustomNode: async () => ({ outcome: "success" }), }; - expect(missingSeams).toBeDefined(); + expect(missingPrimitives).toBeDefined(); }); it("runs a selected workflow through the graph engine", async () => { @@ -65,7 +103,7 @@ describe("WorkflowTaskRuntime", () => { }, getWorkflowDefinition: async () => ({ ir: selectedIr() }), }, - seams: recordingSeams(calls), + primitives: recordingPrimitives(calls), runCustomNode: async (node) => { calls.push(`custom:${node.id}`); return { outcome: "success" }; @@ -75,7 +113,7 @@ describe("WorkflowTaskRuntime", () => { const result = await runtime.run(task, flagOff); expect(result.disposition).toBe("completed"); - expect(calls).toEqual(["custom:prepare", "execute"]); + expect(calls).toEqual(["custom:prepare", "prepare-worktree", "execute"]); expect(result.visitedNodeIds).toEqual(["start", "prepare", "execute"]); expect(workflowSelectionReads).toBe(1); }); @@ -87,7 +125,7 @@ describe("WorkflowTaskRuntime", () => { getTaskWorkflowSelection: () => undefined, getWorkflowDefinition: async () => undefined, }, - seams: recordingSeams(calls), + primitives: recordingPrimitives(calls), runCustomNode: async (node) => { calls.push(`custom:${node.id}`); return { outcome: "success" }; @@ -97,8 +135,8 @@ describe("WorkflowTaskRuntime", () => { const result = await runtime.run(task, flagOff); expect(result.disposition).toBe("completed"); - expect(calls).toEqual(["execute", "review", "merge"]); - expect(result.visitedNodeIds).toEqual(["start", "execute", "review", "merge"]); + expect(calls).toEqual(["planning", "prepare-worktree", "execute", "review", "merge"]); + expect(result.visitedNodeIds).toEqual(["start", "planning", "execute", "review", "merge"]); }); it("turns selected workflow lookup failures into the built-in workflow target", async () => { @@ -109,7 +147,7 @@ describe("WorkflowTaskRuntime", () => { getTaskWorkflowSelection: () => ({ workflowId: "WF-MISSING", stepIds: [] }), getWorkflowDefinition: async () => undefined, }, - seams: recordingSeams(calls), + primitives: recordingPrimitives(calls), runCustomNode: async () => ({ outcome: "success" }), branchPersistence: { loadBranchStates: (_taskId, runId) => { @@ -122,7 +160,7 @@ describe("WorkflowTaskRuntime", () => { const result = await runtime.run(task, flagOff); expect(result.disposition).toBe("completed"); - expect(calls).toEqual(["execute", "review", "merge"]); + expect(calls).toEqual(["planning", "prepare-worktree", "execute", "review", "merge"]); expect(observedRunIds).toContain("FN-9002:builtin:coding"); expect(observedRunIds).not.toContain("FN-9002:WF-MISSING"); }); @@ -135,7 +173,7 @@ describe("WorkflowTaskRuntime", () => { getTaskWorkflowSelection: () => ({ workflowId: "WF-CORRUPT", stepIds: [] }), getWorkflowDefinition: async () => ({ ir: "not a workflow ir" }), }, - seams: recordingSeams(calls), + primitives: recordingPrimitives(calls), runCustomNode: async () => ({ outcome: "success" }), branchPersistence: { loadBranchStates: (_taskId, runId) => { @@ -148,7 +186,7 @@ describe("WorkflowTaskRuntime", () => { const result = await runtime.run(task, flagOff); expect(result.disposition).toBe("completed"); - expect(calls).toEqual(["execute", "review", "merge"]); + expect(calls).toEqual(["planning", "prepare-worktree", "execute", "review", "merge"]); expect(observedRunIds).toContain("FN-9002:builtin:coding"); expect(observedRunIds).not.toContain("FN-9002:WF-CORRUPT"); }); @@ -160,7 +198,7 @@ describe("WorkflowTaskRuntime", () => { getTaskWorkflowSelection: () => ({ workflowId: "WF-001", stepIds: [] }), getWorkflowDefinition: async () => ({ ir: selectedIr() }), }, - seams: recordingSeams([]), + primitives: recordingPrimitives([]), runCustomNode: async () => ({ outcome: "success" }), handlers: { prompt: async (_node, context) => { @@ -189,7 +227,7 @@ describe("WorkflowTaskRuntime", () => { getTaskWorkflowSelection: () => ({ workflowId: "WF-001", stepIds: [] }), getWorkflowDefinition: async () => ({ ir: selectedIr() }), }, - seams: recordingSeams([]), + primitives: recordingPrimitives([]), runCustomNode: async () => ({ outcome: "success" }), branchPersistence: { loadBranchStates: (_taskId, runId) => { @@ -211,7 +249,7 @@ describe("WorkflowTaskRuntime", () => { getTaskWorkflowSelection: () => undefined, getWorkflowDefinition: async () => undefined, }, - seams: recordingSeams([]), + primitives: recordingPrimitives([]), runCustomNode: async () => ({ outcome: "success" }), branchPersistence: { loadBranchStates: (_taskId, runId) => { @@ -233,7 +271,7 @@ describe("WorkflowTaskRuntime", () => { getTaskWorkflowSelection: () => ({ workflowId: "WF-001", stepIds: [] }), getWorkflowDefinition: async () => ({ ir: selectedIr() }), }, - seams: recordingSeams(calls, { execute: { outcome: "failure", value: "implementation-incomplete" } }), + primitives: recordingPrimitives(calls, { execute: { outcome: "failure", value: "implementation-incomplete" } }), runCustomNode: async (node) => { calls.push(`custom:${node.id}`); return { outcome: "success" }; @@ -244,7 +282,7 @@ describe("WorkflowTaskRuntime", () => { expect(result.disposition).toBe("failed"); expect(result.outcome).toBe("failure"); - expect(calls).toEqual(["custom:prepare", "execute"]); + expect(calls).toEqual(["custom:prepare", "prepare-worktree", "execute"]); }); it("converts interpreter throws into workflow-engine failures", async () => { @@ -262,7 +300,7 @@ describe("WorkflowTaskRuntime", () => { getTaskWorkflowSelection: () => ({ workflowId: "WF-001", stepIds: [] }), getWorkflowDefinition: async () => ({ ir: badIr }), }, - seams: recordingSeams([]), + primitives: recordingPrimitives([]), runCustomNode: async () => ({ outcome: "success" }), }); @@ -292,7 +330,7 @@ describe("WorkflowTaskRuntime", () => { getTaskWorkflowSelection: () => ({ workflowId: "WF-001", stepIds: [] }), getWorkflowDefinition: async () => ({ ir: cyclicIr }), }, - seams: recordingSeams([]), + primitives: recordingPrimitives([]), runCustomNode: async () => ({ outcome: "success" }), }); @@ -309,7 +347,7 @@ describe("WorkflowTaskRuntime", () => { getTaskWorkflowSelection: () => ({ workflowId: "WF-001", stepIds: [] }), getWorkflowDefinition: async () => ({ ir: selectedIr() }), }, - seams: recordingSeams([]), + primitives: recordingPrimitives([]), runCustomNode: async () => ({ outcome: "success" }), onEvent: () => { throw new Error("diagnostics failed"); diff --git a/packages/engine/src/executor.ts b/packages/engine/src/executor.ts index 982b694951..600f705a8e 100644 --- a/packages/engine/src/executor.ts +++ b/packages/engine/src/executor.ts @@ -9,7 +9,7 @@ import { delimiter, isAbsolute, join, relative, resolve as resolvePath } from "n import { existsSync, realpathSync } from "node:fs"; import { readFile, rm, writeFile } from "node:fs/promises"; import type { TaskStore, Task, TaskDetail, TaskTokenUsage, StepStatus, Settings, WorkflowStep, MissionStore, Slice, AgentState, AgentCapability, RunMutationContext, AgentHeartbeatConfig, Agent, AgentMemoryInclusionMode, ProjectSettings, MergeResult, WorkflowIrNode } from "@fusion/core"; -import { RetryStormError, TaskDeletedError, serializeRetryStormError, isExperimentalFeatureEnabled, isWorkflowColumnsEnabled, resolveWorkflowIrForTask, resolveColumnAgentBinding, resolveEffectiveAgent, instanceNodeId, getWorkflowExtensionRegistry } from "@fusion/core"; +import { RetryStormError, TaskDeletedError, serializeRetryStormError, isExperimentalFeatureEnabled, isWorkflowColumnsEnabled, resolveWorkflowIrForTask, resolveColumnAgentBinding, resolveEffectiveAgent, instanceNodeId, getWorkflowExtensionRegistry, getBuiltinWorkflow } from "@fusion/core"; import { mergeEffectiveSettings } from "./effective-settings.js"; import type { TaskStep, WorkflowIr, WorkflowFieldDefinition, WorkflowColumnAgent, EffectiveAgentInput, WorkflowWorkEngineDispatchResult } from "@fusion/core"; import { @@ -30,10 +30,17 @@ import { observeWorkflowParity, WORKFLOW_INTERPRETER_DUAL_OBSERVE_FLAG } from ". import { FOREACH_ACTIVE_CONTEXT_KEY, SEAM_GOVERNING_NODE_CONTEXT_KEY, + SPLIT_ACTIVE_CONTEXT_KEY, type ForeachActiveContext, type WorkflowLegacySeams, } from "./workflow-node-handlers.js"; import type { WorkflowNodeResult } from "./workflow-graph-executor.js"; +import type { + AuditPrimitiveInput, + PreparedWorktree, + WorkflowPrimitiveContext, + WorkflowRuntimePrimitives, +} from "./runtime-primitives.js"; import { ApprovalRequestStore, buildExecutionMemoryInstructions, @@ -3649,19 +3656,32 @@ export class TaskExecutor { let settings: Settings; try { settings = await this.store.getSettings(); - } catch { - return false; + } catch (err) { + await this.handleGraphFailure(task, { + disposition: "failed", + outcome: "failure", + reason: `settings-load-failed: ${err instanceof Error ? err.message : String(err)}`, + visitedNodeIds: [], + }); + return true; } - if (!isExperimentalFeatureEnabled(settings, "workflowGraphExecutor")) return false; - if (typeof this.store.getTaskWorkflowSelection !== "function") return false; - + const hasWorkflowResolver = typeof this.store.getTaskWorkflowSelection === "function"; + const explicitlyEnabled = isExperimentalFeatureEnabled(settings, "workflowGraphExecutor"); + if (!hasWorkflowResolver && !explicitlyEnabled) return false; + settings = { + ...settings, + experimentalFeatures: { + ...(settings.experimentalFeatures ?? {}), + workflowGraphExecutor: true, + }, + }; let selection: { workflowId: string; stepIds: string[] } | undefined; try { - selection = this.store.getTaskWorkflowSelection(task.id); + selection = this.store.getTaskWorkflowSelection?.(task.id); } catch { - return false; + selection = undefined; } - if (!selection) return false; + selection ??= { workflowId: "builtin:coding", stepIds: [] }; // Resolve the production run id ONCE, here, so it is the single source of // truth shared by the runner AND the executor-side persistence deps @@ -3673,7 +3693,9 @@ export class TaskExecutor { // prior behavior — so this never strands a task. let resolvedRunId: string | undefined; try { - const definition = await this.store.getWorkflowDefinition?.(selection.workflowId); + const definition = selection.workflowId === "builtin:coding" + ? { id: "builtin:coding" } + : await this.store.getWorkflowDefinition?.(selection.workflowId); if (definition) resolvedRunId = `${task.id}:${definition.id}`; } catch { // Definition load failure — leave undefined; deps/runner use fallbacks. @@ -3710,8 +3732,15 @@ export class TaskExecutor { } const runner = new WorkflowGraphTaskRunner({ - store: this.store, + store: { + ...this.store, + getTaskWorkflowSelection: (taskId: string) => + this.store.getTaskWorkflowSelection?.(taskId) ?? { workflowId: "builtin:coding", stepIds: [] }, + getWorkflowDefinition: async (id: string) => + (await this.store.getWorkflowDefinition?.(id)) ?? getBuiltinWorkflow("builtin:coding"), + }, runId: resolvedRunId, + primitives: this.createAuthoritativeWorkflowPrimitives(settings), seams: this.createAuthoritativeWorkflowSeams(settings), runCustomNode: (node, nodeTask) => this.runGraphCustomNode(node, nodeTask, settings, resolveBindingForNode(node.id)), @@ -3758,16 +3787,26 @@ export class TaskExecutor { const detail = await this.store.getTask(task.id); result = await runner.run(detail, settings); } catch (err) { - // A thrown interpreter error must not strand the task in-progress: fall - // back to the legacy pipeline so the normal executor lock + flow runs. executorLog.error( - `[workflow-graph] ${task.id} interpreter threw — falling back to legacy pipeline: ${err instanceof Error ? err.message : String(err)}`, + `[workflow-graph] ${task.id} interpreter threw — parking task as workflow failure: ${err instanceof Error ? err.message : String(err)}`, ); - return false; + await this.handleGraphFailure(task, { + disposition: "failed", + outcome: "failure", + reason: `interpreter-error: ${err instanceof Error ? err.message : String(err)}`, + visitedNodeIds: [], + }); + return true; } if (result.disposition === "fell-back") { - executorLog.log(`[workflow-graph] ${task.id} fell back to legacy pipeline: ${result.reason}`); - return false; + executorLog.warn(`[workflow-graph] ${task.id} could not resolve workflow — parking task instead of legacy fallback: ${result.reason}`); + await this.handleGraphFailure(task, { + ...result, + disposition: "failed", + outcome: "failure", + reason: result.reason ?? "workflow-resolution-failed", + }); + return true; } if (result.disposition === "failed") { await this.handleGraphFailure(task, result); @@ -4553,6 +4592,240 @@ export class TaskExecutor { /** Public authoritative-driver seam factory: exposes the same real lifecycle * seams the internal graph runner uses, without changing legacy behavior. */ + public createAuthoritativeWorkflowPrimitives(settings: Settings): WorkflowRuntimePrimitives { + const logAudit = async (taskId: string | undefined, input: AuditPrimitiveInput): Promise => { + if (!taskId) return; + try { + await this.store.logEntry(taskId, input.message, input.metadata ? JSON.stringify(input.metadata) : undefined); + } catch { + // Audit is diagnostic-only and must not affect workflow execution. + } + }; + + return { + prepareWorktree: async (_ctx, task) => { + const live = await this.store.getTask(task.id); + const prepared: PreparedWorktree = { + worktreePath: live.worktree || this.rootDir, + branchName: live.branch, + }; + return { outcome: "success", value: "worktree-ready", data: prepared }; + }, + readArtifact: async (_ctx, task, key) => { + const deps = this.buildParseStepsDeps(`${task.id}:artifact-read`); + return deps.readArtifact(task, key); + }, + writeArtifact: async (ctx, task, key, content) => { + const writer = (this.store as unknown as { + writeTaskDocument?: (taskId: string, key: string, content: string) => Promise; + }).writeTaskDocument; + if (!writer) { + await logAudit(task.id, { + type: "artifact-write-unavailable", + message: `Workflow node ${ctx.node.node.id} could not write artifact ${key}: store writer unavailable`, + }); + return { outcome: "failure", value: "artifact-write-unavailable" }; + } + await writer.call(this.store, task.id, key, content); + return { outcome: "success", value: "artifact-written", data: { key } }; + }, + runPlanningSession: async () => ({ outcome: "success", value: "pre-specified", data: { + approved: true, + artifactKeys: [], + } }), + runCodingSession: async (ctx, task) => { + const governingNodeId = ctx.node.context?.[SEAM_GOVERNING_NODE_CONTEXT_KEY]; + if (typeof governingNodeId === "string") { + this.graphSeamGoverningNodeId.set(task.id, governingNodeId); + } + let result: { taskDone: boolean; modifiedFiles: string[] }; + try { + result = await this.runImplementationPhase(task); + } finally { + this.graphSeamGoverningNodeId.delete(task.id); + } + if (result.taskDone) { + return { outcome: "success", value: "implemented", data: result }; + } + let paused = this.pausedAborted.has(task.id); + if (!paused) { + try { + paused = Boolean((await this.store.getTask(task.id)).paused); + } catch { + // Best-effort pause probe; fall through to the failure value. + } + } + return { + outcome: "failure", + value: paused ? "implementation-paused" : "implementation-incomplete", + data: result, + }; + }, + runTaskStep: async (ctx, task, stepIndex) => { + const context = ctx.node.context ?? {}; + const active = context[FOREACH_ACTIVE_CONTEXT_KEY] as ForeachActiveContext | undefined; + if (!active || typeof active.stepIndex !== "number") { + return { outcome: "failure" }; + } + const live = await this.store.getTask(task.id); + const worktreePath = active.worktreePath || live.worktree || this.rootDir; + this.graphStepActiveContext.set(this.graphActiveContextKey(task.id, active.instanceId), active); + const stepGoverningNodeId = context[SEAM_GOVERNING_NODE_CONTEXT_KEY]; + return await runTaskStep( + { + store: this.store, + worktreePath, + runStep: (idx) => + this.runGraphTaskStep( + task, + idx, + active.instanceId, + typeof stepGoverningNodeId === "string" ? stepGoverningNodeId : undefined, + ), + }, + { id: task.id, steps: live.steps }, + stepIndex, + { markDoneOnSuccess: active.deferDoneToReview !== true }, + ); + }, + resetTaskStep: async (ctx, task, stepIndex, baselineSha, checkpointId) => { + const active = ctx.node.context?.[FOREACH_ACTIVE_CONTEXT_KEY] as ForeachActiveContext | undefined; + const branchScoped = typeof active?.worktreePath === "string" && active.worktreePath.length > 0; + let worktreePath = active?.worktreePath ?? this.rootDir; + if (!branchScoped) { + try { + worktreePath = (await this.store.getTask(task.id)).worktree || this.rootDir; + } catch { + // Best-effort worktree resolution; fall back to rootDir. + } + } + const liveSteps = await this.store.getTask(task.id).then((t) => t.steps).catch(() => []); + return await resetStepToBaseline( + { + store: this.store, + worktreePath, + sessionRef: { current: null }, + reviewType: "code", + blastRadiusGuard: branchScoped + ? undefined + : makeAncestryBlastRadiusGuard({ + worktreePath, + task: { id: task.id, steps: liveSteps }, + stepIndex, + }), + }, + { id: task.id, steps: liveSteps }, + stepIndex, + baselineSha, + checkpointId, + ); + }, + runReview: async (ctx, task, input) => { + if (typeof input.stepIndex === "number") { + const context = ctx.node.context ?? {}; + const active = context[FOREACH_ACTIVE_CONTEXT_KEY] as ForeachActiveContext | undefined; + if (!active || typeof active.stepIndex !== "number") { + return { + outcome: "success", + value: "unavailable", + data: { verdict: "UNAVAILABLE", review: "no active step instance" }, + }; + } + const config = { + type: input.type, + advisory: context[SPLIT_ACTIVE_CONTEXT_KEY] === true, + } as const; + const seamResult = await this.createAuthoritativeWorkflowSeams(settings).stepReview?.( + task, + context, + config, + ); + return { + outcome: "success", + value: seamResult?.verdict === "APPROVE" ? "approve" : seamResult?.verdict === "REVISE" ? "revise" : seamResult?.verdict === "RETHINK" ? "rethink" : "unavailable", + data: seamResult ?? { verdict: "UNAVAILABLE", review: "step review unavailable" }, + }; + } + const live = await this.store.getTask(task.id); + await this.persistTokenUsage(task.id); + await this.handoffTaskToReview(live, "workflow-graph-review"); + return { + outcome: "success", + value: "in-review", + data: { verdict: "APPROVE", summary: "Task handed off for merge review" }, + }; + }, + runVerification: async () => ({ outcome: "success", value: "verification-skipped", data: { + verdict: "skipped", + } }), + runWorkflowStep: async () => ({ outcome: "success", value: "workflow-step-skipped", data: { + allPassed: true, + } }), + updateSteps: async (_ctx, task, steps) => { + await this.store.updateTask(task.id, { steps }); + return { outcome: "success", value: "steps-updated", data: { count: steps.length } }; + }, + transitionTask: async (_ctx, task, input) => { + const patch: Partial = {}; + if (input.column !== undefined) patch.column = input.column; + if (input.status !== undefined && input.status !== null) patch.status = input.status; + if (Object.keys(patch).length > 0) { + await this.store.updateTask(task.id, patch); + } + return { outcome: "success", value: input.reason }; + }, + requestMerge: async (ctx, task) => { + if (!this.mergeRequester) { + return { outcome: "failure", value: "merge-unavailable", data: { status: "failed", reason: "merge-unavailable" } }; + } + const GRAPH_MERGE_TIMEOUT_MS = 30 * 60 * 1000; + let timeoutHandle: ReturnType | undefined; + const timeout = new Promise<"timeout">((resolve) => { + timeoutHandle = setTimeout(() => resolve("timeout"), GRAPH_MERGE_TIMEOUT_MS); + timeoutHandle.unref?.(); + }); + try { + const result = await Promise.race([this.mergeRequester(task.id), timeout]); + if (result === "timeout") { + executorLog.warn(`${task.id}: workflow merge primitive timed out after ${GRAPH_MERGE_TIMEOUT_MS}ms`); + return { outcome: "failure", value: "merge-timeout", data: { status: "timeout" } }; + } + if (result.merged || result.noOp) { + return { + outcome: "success", + value: result.noOp ? "merge-noop" : "merged", + data: { status: "merged", noOp: result.noOp }, + }; + } + return { + outcome: "failure", + value: result.reason ?? result.error ?? "merge-failed", + data: { status: "failed", reason: result.reason ?? result.error ?? "merge-failed" }, + }; + } finally { + if (timeoutHandle) clearTimeout(timeoutHandle); + await logAudit(task.id, { + type: "merge-requested", + message: `Workflow node ${ctx.node.node.id} requested merge`, + }); + } + }, + abortRun: async (_ctx, task, input) => { + if (input.hardCancel) { + this.pausedAborted.add(task.id); + } + await this.store.updateTask(task.id, { + paused: true, + pausedReason: input.reason, + } as Partial); + return { outcome: "success", value: "aborted" }; + }, + audit: async (ctx: WorkflowPrimitiveContext, input) => { + await logAudit(ctx.run.taskId, input); + }, + }; + } + public createAuthoritativeWorkflowSeams(_settings: Settings): WorkflowLegacySeams { return { // Built-in triage/spec generation runs upstream of the interpreter today, @@ -5700,10 +5973,10 @@ export class TaskExecutor { executorLog.log(`execute() called for ${task.id} while graph routing is active — skipping duplicate`); return; } - const authoritativeOwned = await this.options.workflowAuthoritativeDispatch?.(task); - if (authoritativeOwned) return; const graphOwned = await this.maybeExecuteWorkflowGraph(task); if (graphOwned) return; + const authoritativeOwned = await this.options.workflowAuthoritativeDispatch?.(task); + if (authoritativeOwned) return; } // FN-4811 follow-up (FN-4814/FN-4809/FN-4811 production failure): claim a diff --git a/packages/engine/src/index.ts b/packages/engine/src/index.ts index 1110472ac9..76826ac8ae 100644 --- a/packages/engine/src/index.ts +++ b/packages/engine/src/index.ts @@ -40,10 +40,14 @@ export { } from "./workflow-graph-branches.js"; export { createDefaultNodeHandlers, + createPrimitivePromptLikeHandler, + createPrimitiveStepReviewHandler, createNoopLegacySeams, createParseStepsHandler, createCodeNodeHandler, PARSE_STEPS_DEFAULT_ARTIFACT, + WORKFLOW_ID_CONTEXT_KEY, + WORKFLOW_RUN_ID_CONTEXT_KEY, type WorkflowCustomNodeRunner, type WorkflowLegacySeams, type WorkflowSeamName, @@ -51,6 +55,28 @@ export { type CodeNodeRunner, type DefaultNodeHandlerDeps, } from "./workflow-node-handlers.js"; +export { + markSideEffectsStarted, + primitiveNodeContext, + type RuntimePrimitiveName, + type WorkflowRuntimeRunContext, + type WorkflowRuntimeNodeContext, + type WorkflowPrimitiveContext, + type RuntimePrimitiveResult, + type PreparedWorktree, + type PlanningSessionResult, + type CodingSessionResult, + type ReviewPrimitiveResult, + type VerificationPrimitiveResult, + type WorkflowStepPrimitiveInput, + type WorkflowStepPrimitiveResult, + type TransitionPrimitiveInput, + type MergePrimitiveInput, + type MergePrimitiveResult, + type AbortPrimitiveInput, + type AuditPrimitiveInput, + type WorkflowRuntimePrimitives, +} from "./runtime-primitives.js"; export { createPrNodeHandlers, createAutoMergeGateHandler, diff --git a/packages/engine/src/runtime-primitives.ts b/packages/engine/src/runtime-primitives.ts new file mode 100644 index 0000000000..2136b05069 --- /dev/null +++ b/packages/engine/src/runtime-primitives.ts @@ -0,0 +1,238 @@ +import type { TaskDetail, TaskStep, WorkflowIrNode } from "@fusion/core"; + +import type { PrMergeCallResult } from "./pr-nodes.js"; +import type { RunTaskStepResult, ResetStepResult } from "./step-runner.js"; +import type { WorkflowNodeOutcome } from "./workflow-graph-executor.js"; + +export type RuntimePrimitiveName = + | "prepare-worktree" + | "read-artifact" + | "write-artifact" + | "planning-session" + | "coding-session" + | "step-session" + | "reset-step" + | "review" + | "verification" + | "workflow-step" + | "transition" + | "merge" + | "abort" + | "audit"; + +export interface WorkflowRuntimeRunContext { + runId: string; + taskId: string; + workflowId: string; + /** True after any primitive with task/git/session side effects starts. */ + sideEffectsStarted?: boolean; + recoveryEventId?: string; +} + +export interface WorkflowRuntimeNodeContext { + node: Pick; + effectivePrincipalId?: string; + attempt?: number; + context?: Record; +} + +export interface WorkflowPrimitiveContext { + run: WorkflowRuntimeRunContext; + node: WorkflowRuntimeNodeContext; +} + +export interface RuntimePrimitiveResult { + outcome: WorkflowNodeOutcome; + value?: string; + data?: TValue; + contextPatch?: Record; +} + +export interface PreparedWorktree { + worktreePath: string; + branchName?: string; + baseCommitSha?: string; + modifiedFiles?: string[]; +} + +export interface PlanningSessionResult { + approved: boolean; + artifactKeys: string[]; + createdTaskIds?: string[]; + feedback?: string; +} + +export interface CodingSessionResult { + taskDone: boolean; + modifiedFiles: string[]; + summary?: string; +} + +export interface ReviewPrimitiveResult { + verdict: "APPROVE" | "REVISE" | "RETHINK" | "UNAVAILABLE"; + review?: string; + summary?: string; +} + +export interface VerificationPrimitiveResult { + verdict: "approve" | "revise" | "failed" | "advisory-failed" | "skipped"; + feedback?: string; + stepName?: string; +} + +export interface WorkflowStepPrimitiveInput { + phase: "pre-merge" | "post-merge"; + stepId?: string; + worktreePath?: string; +} + +export interface WorkflowStepPrimitiveResult { + allPassed: boolean; + revisionRequested?: boolean; + feedback?: string; + stepName?: string; +} + +export interface TransitionPrimitiveInput { + column?: string; + status?: string | null; + reason: string; + preserveProgress?: boolean; +} + +export interface MergePrimitiveInput { + expectedHeadOid?: string; + manualAllowed?: boolean; +} + +export type MergePrimitiveResult = + | { status: "merged"; noOp?: boolean } + | { status: "manual-required"; reason?: string } + | { status: "failed"; reason: string } + | { status: "timeout" } + | PrMergeCallResult; + +export interface AbortPrimitiveInput { + reason: string; + hardCancel?: boolean; +} + +export interface AuditPrimitiveInput { + type: string; + message: string; + metadata?: Record; +} + +export interface WorkflowRuntimePrimitives { + prepareWorktree( + ctx: WorkflowPrimitiveContext, + task: TaskDetail, + ): Promise>; + + readArtifact( + ctx: WorkflowPrimitiveContext, + task: TaskDetail, + key: string, + ): Promise; + + writeArtifact( + ctx: WorkflowPrimitiveContext, + task: TaskDetail, + key: string, + content: string, + ): Promise>; + + runPlanningSession( + ctx: WorkflowPrimitiveContext, + task: TaskDetail, + ): Promise>; + + runCodingSession( + ctx: WorkflowPrimitiveContext, + task: TaskDetail, + prepared: PreparedWorktree, + ): Promise>; + + runTaskStep( + ctx: WorkflowPrimitiveContext, + task: TaskDetail, + stepIndex: number, + ): Promise; + + resetTaskStep( + ctx: WorkflowPrimitiveContext, + task: TaskDetail, + stepIndex: number, + baselineSha?: string, + checkpointId?: string, + ): Promise; + + runReview( + ctx: WorkflowPrimitiveContext, + task: TaskDetail, + input: { type: "plan" | "code"; stepIndex?: number; baselineSha?: string }, + ): Promise>; + + runVerification( + ctx: WorkflowPrimitiveContext, + task: TaskDetail, + prepared: PreparedWorktree, + ): Promise>; + + runWorkflowStep( + ctx: WorkflowPrimitiveContext, + task: TaskDetail, + input: WorkflowStepPrimitiveInput, + ): Promise>; + + updateSteps( + ctx: WorkflowPrimitiveContext, + task: TaskDetail, + steps: TaskStep[], + ): Promise>; + + transitionTask( + ctx: WorkflowPrimitiveContext, + task: TaskDetail, + input: TransitionPrimitiveInput, + ): Promise; + + requestMerge( + ctx: WorkflowPrimitiveContext, + task: TaskDetail, + input?: MergePrimitiveInput, + ): Promise>; + + abortRun( + ctx: WorkflowPrimitiveContext, + task: TaskDetail, + input: AbortPrimitiveInput, + ): Promise; + + audit(ctx: WorkflowPrimitiveContext, input: AuditPrimitiveInput): Promise | void; +} + +export function markSideEffectsStarted(ctx: WorkflowPrimitiveContext): WorkflowPrimitiveContext { + return { + ...ctx, + run: { + ...ctx.run, + sideEffectsStarted: true, + }, + }; +} + +export function primitiveNodeContext( + run: WorkflowRuntimeRunContext, + node: WorkflowRuntimeNodeContext["node"], + extras: Omit = {}, +): WorkflowPrimitiveContext { + return { + run, + node: { + ...extras, + node, + }, + }; +} + diff --git a/packages/engine/src/workflow-authoritative-driver.ts b/packages/engine/src/workflow-authoritative-driver.ts index c017f5ff47..6ec2fa4fb1 100644 --- a/packages/engine/src/workflow-authoritative-driver.ts +++ b/packages/engine/src/workflow-authoritative-driver.ts @@ -14,6 +14,8 @@ import type { TaskExecutor } from "./executor.js"; import { executorLog } from "./logger.js"; import { WORKFLOW_INTERPRETER_DUAL_OBSERVE_FLAG } from "./workflow-parity-observer.js"; import { WorkflowGraphTaskRunner, type WorkflowGraphTaskRunResult } from "./workflow-graph-task-runner.js"; +import type { WorkflowLegacySeams } from "./workflow-node-handlers.js"; +import type { PreparedWorktree, WorkflowRuntimePrimitives } from "./runtime-primitives.js"; const AUTHORITATIVE_WORKFLOW_ID = "workflow-interpreter-authoritative"; @@ -26,7 +28,7 @@ export interface WorkflowAuthoritativeDriverStore { export interface WorkflowAuthoritativeDriverDeps { store: WorkflowAuthoritativeDriverStore; - executor: Pick; + executor: Pick & Partial>; minimumObservedRuns?: number; } @@ -49,6 +51,58 @@ function buildAuthoritativeSettings(settings: Settings): Settings { }; } +function primitivesFromLegacySeams(seams: WorkflowLegacySeams): WorkflowRuntimePrimitives { + const prepared: PreparedWorktree = { worktreePath: "" }; + return { + prepareWorktree: async () => ({ outcome: "success", data: prepared }), + readArtifact: async () => undefined, + writeArtifact: async (_ctx, _task, key) => ({ outcome: "success", data: { key } }), + runPlanningSession: async (ctx, task) => { + const result = await seams.planning(task, ctx.node.context ?? {}); + return { ...result, data: { approved: result.outcome === "success", artifactKeys: [] } }; + }, + runCodingSession: async (ctx, task) => { + const result = await seams.execute(task, ctx.node.context ?? {}); + return { + ...result, + data: { taskDone: result.outcome === "success", modifiedFiles: [] }, + }; + }, + runTaskStep: async (ctx, task) => { + const result = await seams.stepExecute?.(task, ctx.node.context ?? {}); + return { outcome: result?.outcome ?? "failure" }; + }, + resetTaskStep: async () => ({ ok: true }), + runReview: async (ctx, task, input) => { + if (typeof input.stepIndex === "number") { + const result = await seams.stepReview?.(task, ctx.node.context ?? {}, { type: input.type }); + return { + outcome: "success", + value: result?.verdict === "APPROVE" ? "approve" : result?.verdict === "REVISE" ? "revise" : result?.verdict === "RETHINK" ? "rethink" : "unavailable", + data: result ?? { verdict: "UNAVAILABLE" }, + }; + } + const result = await seams.review(task, ctx.node.context ?? {}); + return { ...result, data: { verdict: result.outcome === "success" ? "APPROVE" : "REVISE" } }; + }, + runVerification: async () => ({ outcome: "success", data: { verdict: "skipped" } }), + runWorkflowStep: async () => ({ outcome: "success", data: { allPassed: true } }), + updateSteps: async (_ctx, _task, steps) => ({ outcome: "success", data: { count: steps.length } }), + transitionTask: async (ctx, task) => seams.schedule(task, ctx.node.context ?? {}), + requestMerge: async (ctx, task) => { + const result = await seams.merge(task, ctx.node.context ?? {}); + return { + ...result, + data: result.outcome === "success" + ? { status: "merged" as const } + : { status: "failed" as const, reason: result.value ?? "merge-failed" }, + }; + }, + abortRun: async () => ({ outcome: "success" }), + audit: () => undefined, + }; +} + export class WorkflowAuthoritativeDriver { public constructor(private readonly deps: WorkflowAuthoritativeDriverDeps) {} @@ -114,6 +168,7 @@ export class WorkflowAuthoritativeDriver { readinessReasons: [], }; } + const seams = this.deps.executor.createAuthoritativeWorkflowSeams(settings); const runner = new WorkflowGraphTaskRunner({ store: { getTaskWorkflowSelection: () => ({ workflowId: AUTHORITATIVE_WORKFLOW_ID, stepIds: [] }), @@ -123,7 +178,8 @@ export class WorkflowAuthoritativeDriver { ir: BUILTIN_CODING_WORKFLOW_IR, } satisfies Pick as WorkflowDefinition), }, - seams: this.deps.executor.createAuthoritativeWorkflowSeams(settings), + primitives: this.deps.executor.createAuthoritativeWorkflowPrimitives?.(settings) ?? primitivesFromLegacySeams(seams), + seams, runCustomNode: async (node) => { throw new Error(`unexpected custom node in builtin authoritative workflow: ${node.id}`); }, diff --git a/packages/engine/src/workflow-graph-executor.ts b/packages/engine/src/workflow-graph-executor.ts index 1ce6ee51f8..193eecacce 100644 --- a/packages/engine/src/workflow-graph-executor.ts +++ b/packages/engine/src/workflow-graph-executor.ts @@ -5,12 +5,15 @@ import { createDefaultNodeHandlers, createNoopLegacySeams, SPLIT_ACTIVE_CONTEXT_KEY, + WORKFLOW_ID_CONTEXT_KEY, + WORKFLOW_RUN_ID_CONTEXT_KEY, type CodeNodeRunner, type ForeachActiveContext, type ParseStepsHandlerDeps, type WorkflowCustomNodeRunner, type WorkflowLegacySeams, } from "./workflow-node-handlers.js"; +import type { WorkflowRuntimePrimitives } from "./runtime-primitives.js"; import type { PrNodeDeps } from "./pr-nodes.js"; import { runSplitJoin, @@ -48,6 +51,9 @@ export type WorkflowNodeHandler = (node: WorkflowIrNode, context: WorkflowNodeEx export interface WorkflowGraphExecutorDeps { handlers?: Partial>; + /** Workflow-native runtime primitives. When present, default nodes call these + * directly instead of legacy executor/reviewer/merge seams. */ + primitives?: WorkflowRuntimePrimitives; seams?: WorkflowLegacySeams; /** Executes custom (non-seam) prompt/script/gate nodes. */ runCustomNode?: WorkflowCustomNodeRunner; @@ -150,6 +156,7 @@ export class WorkflowGraphExecutor { this.maxRetriesPerNode = Math.max(1, Math.floor(deps.maxRetriesPerNode ?? 2)); this.handlers = { ...createDefaultNodeHandlers(deps.seams ?? createNoopLegacySeams(), deps.runCustomNode, { + primitives: deps.primitives, parseSteps: deps.parseStepsDeps, runCode: deps.runCode, prNodes: deps.prNodes, @@ -181,10 +188,13 @@ export class WorkflowGraphExecutor { outgoingMap.set(edge.from, list); } - const context: Record = {}; + const runId = this.deps.runId ?? `${task.id}:run`; + const context: Record = { + [WORKFLOW_RUN_ID_CONTEXT_KEY]: runId, + [WORKFLOW_ID_CONTEXT_KEY]: ir.name || "unknown", + }; const visitedNodeIds: string[] = []; const inStack = new Set(); - const runId = this.deps.runId ?? `${task.id}:run`; // Bounded-rework generalization (U6). A `kind: "rework"` edge is the only // legal cycle: it loops back to a "rework region head" (the edge's `to` node). diff --git a/packages/engine/src/workflow-graph-task-runner.ts b/packages/engine/src/workflow-graph-task-runner.ts index 57c37d6401..2cb8cedda7 100644 --- a/packages/engine/src/workflow-graph-task-runner.ts +++ b/packages/engine/src/workflow-graph-task-runner.ts @@ -16,6 +16,7 @@ import type { } from "./workflow-graph-branches.js"; import type { ForeachEnvironment, WorkflowStepInstancePersistence } from "./workflow-graph-foreach.js"; import type { PrNodeDeps } from "./pr-nodes.js"; +import type { WorkflowPrimitiveContext, WorkflowRuntimePrimitives } from "./runtime-primitives.js"; // (Both types are also used as values in the side-effect tracking wrappers below.) /** @@ -46,6 +47,7 @@ export interface WorkflowGraphRunnerStore { export interface WorkflowGraphTaskRunnerDeps { store: WorkflowGraphRunnerStore; seams: WorkflowLegacySeams; + primitives?: WorkflowRuntimePrimitives; runCustomNode: WorkflowCustomNodeRunner; maxRetriesPerNode?: number; /** Optional diagnostics hook (audit/log emission). Never throws into the run. */ @@ -186,10 +188,25 @@ export class WorkflowGraphTaskRunner { invoked.push(node.id); return this.deps.runCustomNode(node, t, c); }; + const wrappedPrimitives = this.deps.primitives + ? new Proxy(this.deps.primitives, { + get: (target, prop, receiver) => { + const value = Reflect.get(target, prop, receiver); + if (typeof value !== "function") return value; + return (...args: unknown[]) => { + sideEffectsRan = true; + const ctx = args[0] as WorkflowPrimitiveContext | undefined; + invoked.push(ctx?.node?.node?.id ?? String(prop)); + return value.apply(target, args); + }; + }, + }) as WorkflowRuntimePrimitives + : undefined; try { const executor = new WorkflowGraphExecutor({ seams: wrappedSeams, + primitives: wrappedPrimitives, runCustomNode: wrappedRunCustomNode, maxRetriesPerNode: this.deps.maxRetriesPerNode, branchPersistence: this.deps.branchPersistence, diff --git a/packages/engine/src/workflow-node-handlers.ts b/packages/engine/src/workflow-node-handlers.ts index c2fb07b387..a8d86fecfa 100644 --- a/packages/engine/src/workflow-node-handlers.ts +++ b/packages/engine/src/workflow-node-handlers.ts @@ -3,6 +3,11 @@ import type { TaskDetail, TaskStep, WorkflowIrNode } from "@fusion/core"; import type { WorkflowNodeHandler, WorkflowNodeResult } from "./workflow-graph-executor.js"; import { createPrNodeHandlers, createAutoMergeGateHandler, type PrNodeDeps } from "./pr-nodes.js"; +import { + primitiveNodeContext, + type WorkflowPrimitiveContext, + type WorkflowRuntimePrimitives, +} from "./runtime-primitives.js"; export type WorkflowSeamName = "planning" | "execute" | "review" | "merge" | "schedule" | "step-execute"; @@ -104,6 +109,12 @@ export const SPLIT_ACTIVE_CONTEXT_KEY = "split:active"; */ export const INTEGRATION_CONFLICT_CONTEXT_KEY = "integration:conflict"; +/** Reserved graph context key for the current workflow run id. */ +export const WORKFLOW_RUN_ID_CONTEXT_KEY = "workflow:run-id"; + +/** Reserved graph context key for the current workflow id. */ +export const WORKFLOW_ID_CONTEXT_KEY = "workflow:id"; + /** Shape of the value stored under {@link FOREACH_ACTIVE_CONTEXT_KEY}. */ export interface ForeachActiveContext { foreachNodeId: string; @@ -149,6 +160,34 @@ export type WorkflowCustomNodeRunner = ( context: Record, ) => Promise; +function primitiveContextForNode( + node: WorkflowIrNode, + task: TaskDetail, + context: Record, + attempt?: number, +): WorkflowPrimitiveContext { + return primitiveNodeContext( + { + runId: typeof context[WORKFLOW_RUN_ID_CONTEXT_KEY] === "string" + ? context[WORKFLOW_RUN_ID_CONTEXT_KEY] + : `${task.id}:workflow`, + taskId: task.id, + workflowId: typeof context[WORKFLOW_ID_CONTEXT_KEY] === "string" + ? context[WORKFLOW_ID_CONTEXT_KEY] + : "unknown", + }, + node, + { + attempt, + context, + effectivePrincipalId: + typeof context["workflow:effective-principal-id"] === "string" + ? context["workflow:effective-principal-id"] + : undefined, + }, + ); +} + /** Resolve a node's seam name, or undefined for custom (non-seam) nodes. */ export function resolveSeamName(node: { config?: Record }): WorkflowSeamName | undefined { const seam = node.config?.seam; @@ -220,6 +259,83 @@ export function createPromptLikeHandler( }; } +export function createPrimitivePromptLikeHandler( + primitives: WorkflowRuntimePrimitives, + runCustomNode?: WorkflowCustomNodeRunner, +): WorkflowNodeHandler { + return async (node, context) => { + const seam = resolveSeamName(node); + if (seam === "step-execute") { + const active = context.context[FOREACH_ACTIVE_CONTEXT_KEY] as + | ForeachActiveContext + | undefined; + if (!active || typeof active.stepIndex !== "number") { + throw new WorkflowIrError( + `step-execute node '${node.id}' reached without an active foreach instance context`, + ); + } + context.context[SEAM_GOVERNING_NODE_CONTEXT_KEY] = instanceNodeId( + active.foreachNodeId, + active.stepIndex, + node.id, + ); + const result = await primitives.runTaskStep( + primitiveContextForNode(node, context.task, context.context), + context.task, + active.stepIndex, + ); + active.baselineSha = result.baselineSha; + active.checkpointId = result.checkpointId; + return { + outcome: result.outcome, + value: result.outcome === "success" ? "step-done" : "step-failed", + contextPatch: { + [FOREACH_ACTIVE_CONTEXT_KEY]: active, + }, + }; + } + if (seam) { + context.context[SEAM_GOVERNING_NODE_CONTEXT_KEY] = node.id; + const primitiveCtx = primitiveContextForNode(node, context.task, context.context); + if (seam === "planning") { + const result = await primitives.runPlanningSession(primitiveCtx, context.task); + return { outcome: result.outcome, value: result.value, contextPatch: result.contextPatch }; + } + if (seam === "execute") { + const prepared = await primitives.prepareWorktree(primitiveCtx, context.task); + if (prepared.outcome !== "success" || !prepared.data) { + return { + outcome: prepared.outcome, + value: prepared.value ?? "prepare-worktree-failed", + contextPatch: prepared.contextPatch, + }; + } + const result = await primitives.runCodingSession(primitiveCtx, context.task, prepared.data); + return { outcome: result.outcome, value: result.value, contextPatch: result.contextPatch }; + } + if (seam === "review") { + const result = await primitives.runReview(primitiveCtx, context.task, { type: "code" }); + return { outcome: result.outcome, value: result.value, contextPatch: result.contextPatch }; + } + if (seam === "merge") { + const result = await primitives.requestMerge(primitiveCtx, context.task); + return { outcome: result.outcome, value: result.value, contextPatch: result.contextPatch }; + } + if (seam === "schedule") { + const result = await primitives.transitionTask(primitiveCtx, context.task, { + reason: "workflow-schedule", + preserveProgress: true, + }); + return { outcome: result.outcome, value: result.value, contextPatch: result.contextPatch }; + } + } + if (!runCustomNode) { + throw new WorkflowIrError(`No custom-node runner registered for node: ${node.id}`); + } + return runCustomNode(node, context.task, context.context); + }; +} + /** * Gate handler. Two forms: * - Context gate (original scaffold contract): `config.expect` compared against @@ -337,6 +453,55 @@ export function createStepReviewHandler(seams: WorkflowLegacySeams): WorkflowNod }; } +export function createPrimitiveStepReviewHandler(primitives: WorkflowRuntimePrimitives): WorkflowNodeHandler { + return async (node, ctx) => { + const active = ctx.context[FOREACH_ACTIVE_CONTEXT_KEY] as ForeachActiveContext | undefined; + if (!active || typeof active.stepIndex !== "number") { + throw new WorkflowIrError( + `step-review node '${node.id}' reached without an active foreach instance context`, + ); + } + + const advisory = ctx.context[SPLIT_ACTIVE_CONTEXT_KEY] === true; + const config = resolveStepReviewConfig(node, advisory); + let result: StepReviewSeamResult = { + verdict: "UNAVAILABLE", + }; + for (let attempt = 0; attempt <= STEP_REVIEW_UNAVAILABLE_RETRY_CAP; attempt++) { + const primitiveResult = await primitives.runReview( + primitiveContextForNode(node, ctx.task, ctx.context, attempt + 1), + ctx.task, + { + type: config.type, + stepIndex: active.stepIndex, + baselineSha: config.type === "code" ? active.baselineSha : undefined, + }, + ); + result = primitiveResult.data ?? { verdict: "UNAVAILABLE" as const }; + if (result.verdict !== "UNAVAILABLE") break; + } + + if (!advisory) { + active.verdict = result.verdict; + } + const patch: Record = { + [FOREACH_ACTIVE_CONTEXT_KEY]: active, + [`node:${node.id}:verdict`]: result.verdict, + }; + + const value = + result.verdict === "APPROVE" + ? "approve" + : result.verdict === "REVISE" + ? "revise" + : result.verdict === "RETHINK" + ? "rethink" + : "unavailable"; + + return { outcome: "success", value, contextPatch: patch }; + }; +} + // ── parse-steps node (U12, KTD-12) ────────────────────────────────────────── /** The implicit default step-source artifact when a workflow declares no @@ -542,6 +707,8 @@ export function createCodeNodeHandler(runCode?: CodeNodeRunner): WorkflowNodeHan } export interface DefaultNodeHandlerDeps { + /** Workflow-native runtime primitives. When present they replace legacy seams. */ + primitives?: WorkflowRuntimePrimitives; /** parse-steps node deps (U12). When absent, a parse-steps node fails cleanly. */ parseSteps?: ParseStepsHandlerDeps; /** code node runner (U14). When absent, a code node fails cleanly. */ @@ -566,7 +733,9 @@ export function createDefaultNodeHandlers( | "pr-merge", WorkflowNodeHandler > { - const promptLike = createPromptLikeHandler(seams, runCustomNode); + const promptLike = deps?.primitives + ? createPrimitivePromptLikeHandler(deps.primitives, runCustomNode) + : createPromptLikeHandler(seams, runCustomNode); // parse-steps without deps fails closed (would otherwise have no handler at // all and throw "No handler registered"); a clean failure is the safe posture. const parseSteps: WorkflowNodeHandler = deps?.parseSteps @@ -596,7 +765,9 @@ export function createDefaultNodeHandlers( prompt: promptLike, script: promptLike, gate, - "step-review": createStepReviewHandler(seams), + "step-review": deps?.primitives + ? createPrimitiveStepReviewHandler(deps.primitives) + : createStepReviewHandler(seams), "parse-steps": parseSteps, code: createCodeNodeHandler(deps?.runCode), ...prNodes, diff --git a/packages/engine/src/workflow-task-runtime.ts b/packages/engine/src/workflow-task-runtime.ts index 8ca46f8aac..daf6aa5644 100644 --- a/packages/engine/src/workflow-task-runtime.ts +++ b/packages/engine/src/workflow-task-runtime.ts @@ -15,9 +15,10 @@ import { } from "./workflow-graph-executor.js"; import { createDefaultNodeHandlers, + createNoopLegacySeams, type WorkflowCustomNodeRunner, - type WorkflowLegacySeams, } from "./workflow-node-handlers.js"; +import type { WorkflowRuntimePrimitives } from "./runtime-primitives.js"; export type WorkflowTaskRuntimeDisposition = "completed" | "failed"; @@ -31,7 +32,7 @@ export interface WorkflowTaskRuntimeResult { export interface WorkflowTaskRuntimeDeps extends Omit { store: WorkflowIrResolverStore; - seams: WorkflowLegacySeams; + primitives: WorkflowRuntimePrimitives; runCustomNode: WorkflowCustomNodeRunner; onEvent?: (event: { type: "start" | "terminal"; taskId: string; detail: string }) => void; } @@ -80,6 +81,7 @@ export class WorkflowTaskRuntime { const invoked: string[] = []; const executor = new WorkflowGraphExecutor({ ...this.deps, + primitives: this.deps.primitives, handlers: this.recordingHandlers(invoked), // WorkflowTaskRuntime is the execution engine, so internally the graph // executor is authoritative even before the old feature flag plumbing is @@ -140,7 +142,8 @@ export class WorkflowTaskRuntime { } private recordingHandlers(invoked: string[]): Partial> { - const defaultHandlers = createDefaultNodeHandlers(this.deps.seams, this.deps.runCustomNode, { + const defaultHandlers = createDefaultNodeHandlers(createNoopLegacySeams(), this.deps.runCustomNode, { + primitives: this.deps.primitives, parseSteps: this.deps.parseStepsDeps, runCode: this.deps.runCode, prNodes: this.deps.prNodes,