feat(flows): execute fork branches in parallel with join semantics - #501
Open
albertoperdomo2 wants to merge 2 commits into
Open
albertoperdomo2 wants to merge 2 commits into
albertoperdomo2 wants to merge 2 commits into
Conversation
The flow runner walked a single cursor and followed only the first
outgoing edge of every node, silently dropping the rest; `merge` was a
pass-through that neither waited for branches nor aggregated outputs.
Fan-out flows parsed and validated but lost branches at runtime, and
downstream `{{steps.<nodeId>.output}}` references to the dropped nodes
resolved to null.
- Add a `fork` node type (`joinNodeId` names its paired `merge`) and
execute branches as concurrent fibers in the runner, each with its
own OpenCode session (a workspace session runs one prompt at a time,
so sharing the parent session would fail spuriously with
session_busy), own visited set, and a shared fail-fast flag.
- Merge branch steps into the parent template context after the
branches settle, so post-join nodes can reference branch outputs;
the join executes once on the parent cursor and `previousOutput`
resets to null after a fork.
- First branch failure (or unconfirmed termination) aborts sibling
branches on their next loop iteration and fails the fork without
executing the join; run cancellation still exits through the
existing per-iteration check.
- Validate topology up front: single outgoing edge outside
condition/fork (the silent-drop bug, now an error), join must be an
existing unique merge, at least two real branches, no branch dead
ends, no human/slack nodes inside branches, and no join inputs from
outside the branch region. Definitions without forks are unchanged.
Fork nodes record no run step, so the Prisma FlowNodeType enum and the
database are untouched. Editor authoring UI for forks is a follow-up.
Co-Authored-By: Claude Code <noreply@anthropic.com>
6 tasks
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Summary
Implements the fork/join design from
PARALLEL-FLOW-EXECUTION.md. Today the runner walks a single cursor and follows only the first outgoing edge of every node (nextEdge()takesgetFlowOutgoingTargets(...)[0]), silently dropping the rest;mergeis a pass-through, not a join. A fan-out flow that looks correct in the editor executes only its first branch, and downstream{{steps.<dropped-node>.output}}references resolve tonulland fail the run withunknown_template_variable.Changes
forknode type —{ id, name, type: 'fork', joinNodeId };joinNodeIdexplicitly names the pairedmerge, avoiding ambiguity about which join collects which branches.fork, the runner spawns one execution fiber per outgoing edge and runs them viaPromise.all. Each fiber is an independent traversal (own visited set, own step-list copy, stops at the fork's join viastopBeforeNodeId). When all branches arrive, the parent cursor executes themergeonce and continues single-cursor from its outgoing edge.createActiveRunAfterRuntimeStateCheckfails withsession_busyon a busy session), so branches sharing the parent session would fail spuriously or interleave. Each branch creates its own session (Flow | <name> | <date> · <branch node>); branch steps are recorded per node as usual, which is what the template context consumes.termination_unconfirmed) sets it, siblings exit on their next loop iteration, and the fork reports the failure without executing the join. Run cancellation still exits through the existing per-iteration check. Outcome precedence:termination_unconfirmed>failed>cancelled.nodeIdinto the parent after the branches settle, so{{steps.<branchNode>.output}}works in post-join nodes;previousOutputresets tonullafter a fork (branch outputs are referenced explicitly).validateFlowDefinition) — new rules: at most one outgoing edge for nodes other thancondition/fork(this is the silent-drop bug, now an error — intentionally flags currently-broken flows); join must reference an existingmergeand be unique per fork; ≥2 branches and no direct fork→join edge; no branch dead-ends; nohuman/slackinside a branch region; no join inputs from outside the region.fork.joinNodeId(join fields previously fell through unmapped).Design decisions documented in the OpenSpec change
human/slacknodes are rejected inside branches for now: pausing one branch while siblings run needs branch-state persistence and a fork-aware resume path.FlowNodeTypeenum and the database need no migration;nodeTypeToPrismais typed to exclude fork so the compiler keeps that invariant.currentNodeIdtracks the single cursor only; a retry after a fork failure resumes from the fork and re-runs the region.OpenSpec
Adds
parallel-flow-executionchange with proposal, design, tasks, and aflow-executionspec delta (parallel branch execution, join failure semantics, topology validation, backward compatibility).openspec validate parallel-flow-execution --strictpasses.Test plan
validation.test.ts— well-formed fork accepted (nested forks too); one test per rejection rule; multi-edge rejectionrunner.test.ts— branches run and both outputs render in the post-join template; per-branch session creation; branch failure fails the run without executing the joineditor-graph.test.ts— join rename remapsjoinNodeIdtsc --noEmit: no new errors (only the pre-existingtests/providers-*local-env noise, untouched by this diff)Notes / follow-ups
executeFlowNodes(between-step token refresh), so whichever merges second will need a rebase — the changes are complementary, not conflicting in intent.