From cb4058eb0a70f676b437c23a58fc9af6c52c9984 Mon Sep 17 00:00:00 2001 From: Alberto Perdomo Date: Wed, 2 Sep 2026 11:01:00 +0100 Subject: [PATCH] feat(flows): execute fork branches in parallel with join semantics 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..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 --- .../lib/flows/__tests__/editor-graph.test.ts | 24 +++ .../src/lib/flows/__tests__/runner.test.ts | 92 ++++++++++++ .../lib/flows/__tests__/validation.test.ts | 139 +++++++++++++++++ apps/web/src/lib/flows/editor-graph.ts | 4 + apps/web/src/lib/flows/node-executor-utils.ts | 8 +- apps/web/src/lib/flows/runner.ts | 142 ++++++++++++++++-- apps/web/src/lib/flows/types.ts | 8 + apps/web/src/lib/flows/validation.ts | 98 +++++++++++- .../parallel-flow-execution/.openspec.yaml | 2 + .../changes/parallel-flow-execution/design.md | 53 +++++++ .../parallel-flow-execution/proposal.md | 20 +++ .../specs/flow-execution/spec.md | 56 +++++++ .../changes/parallel-flow-execution/tasks.md | 20 +++ 13 files changed, 651 insertions(+), 15 deletions(-) create mode 100644 openspec/changes/parallel-flow-execution/.openspec.yaml create mode 100644 openspec/changes/parallel-flow-execution/design.md create mode 100644 openspec/changes/parallel-flow-execution/proposal.md create mode 100644 openspec/changes/parallel-flow-execution/specs/flow-execution/spec.md create mode 100644 openspec/changes/parallel-flow-execution/tasks.md diff --git a/apps/web/src/lib/flows/__tests__/editor-graph.test.ts b/apps/web/src/lib/flows/__tests__/editor-graph.test.ts index a553a26b1..7082b4f6b 100644 --- a/apps/web/src/lib/flows/__tests__/editor-graph.test.ts +++ b/apps/web/src/lib/flows/__tests__/editor-graph.test.ts @@ -93,6 +93,30 @@ describe('flow editor graph helpers', () => { }) }) + it('remaps a fork join when the join node is renamed', () => { + const flow: FlowDefinition = { + edges: [ + { id: 'edge-1', sourceNodeId: 'fork-1', targetNodeId: 'agent-2' }, + { id: 'edge-2', sourceNodeId: 'agent-2', targetNodeId: 'merge-1' }, + ], + layout: undefined, + nodes: [ + { id: 'fork-1', joinNodeId: 'merge-1', name: 'Fan out', type: 'fork' }, + { compactOutput: false, id: 'agent-2', name: 'Agent', promptTemplate: 'Work', targetAgentId: null, type: 'agent' }, + { id: 'merge-1', name: 'Collect', type: 'merge' }, + ], + startNodeId: 'fork-1', + version: 1, + } + const mergeNode = flow.nodes[2]! + + const result = updateFlowDefinitionNode(flow, { ...mergeNode, name: 'Collect results' }) + + expect(result?.nodeId).toBe('collect-results') + const fork = result?.definition.nodes[0] + expect(fork?.type === 'fork' && fork.joinNodeId).toBe('collect-results') + }) + it('renames condition targets and template references across node types', () => { const flow: FlowDefinition = { edges: [ diff --git a/apps/web/src/lib/flows/__tests__/runner.test.ts b/apps/web/src/lib/flows/__tests__/runner.test.ts index 070ce08bc..1bba12ab8 100644 --- a/apps/web/src/lib/flows/__tests__/runner.test.ts +++ b/apps/web/src/lib/flows/__tests__/runner.test.ts @@ -351,6 +351,98 @@ describe('triggerFlowNow', () => { expect(mocks.markRunSucceeded).toHaveBeenCalledWith('run-1', expect.objectContaining({ openCodeSessionId: 'session-1' })) }) + describe('fork execution', () => { + function createForkFlowDefinition() { + return { + edges: [ + { id: 'edge-1', sourceNodeId: 'agent-1', targetNodeId: 'fork-1' }, + { id: 'edge-2', sourceNodeId: 'fork-1', targetNodeId: 'agent-2' }, + { id: 'edge-3', sourceNodeId: 'fork-1', targetNodeId: 'agent-3' }, + { id: 'edge-4', sourceNodeId: 'agent-2', targetNodeId: 'merge-1' }, + { id: 'edge-5', sourceNodeId: 'agent-3', targetNodeId: 'merge-1' }, + { id: 'edge-6', sourceNodeId: 'merge-1', targetNodeId: 'agent-4' }, + ], + nodes: [ + { compactOutput: false, id: 'agent-1', name: 'Orient', promptTemplate: 'Start', targetAgentId: null, type: 'agent' }, + { id: 'fork-1', joinNodeId: 'merge-1', name: 'Fan out', type: 'fork' }, + { compactOutput: false, id: 'agent-2', name: 'Hunt bugs', promptTemplate: 'Branch A', targetAgentId: null, type: 'agent' }, + { compactOutput: false, id: 'agent-3', name: 'Hunt perf', promptTemplate: 'Branch B', targetAgentId: null, type: 'agent' }, + { id: 'merge-1', name: 'Collect', type: 'merge' }, + { compactOutput: false, id: 'agent-4', name: 'Verify', promptTemplate: 'Merged: {{steps.agent-2.output}} / {{steps.agent-3.output}}', targetAgentId: null, type: 'agent' }, + ], + startNodeId: 'agent-1', + version: 1, + } + } + + function createForkFlow() { + const flow = createClaimedFlow() + flow.definition = createForkFlowDefinition() + return flow + } + + function promptedPrompts() { + return mocks.runFlowPromptAndReadOutput.mock.calls.map((call) => (call[0] as { prompt: string }).prompt) + } + + it('runs fork branches and exposes branch outputs after the join', async () => { + mocks.userFindByIdSelect.mockResolvedValue({ slug: 'alice' }) + mocks.createRun.mockResolvedValue(createRunRecord()) + mocks.runFlowPromptAndReadOutput.mockImplementation(async (params: { prompt: string }) => ({ + ok: true, + output: params.prompt, + })) + + await runClaimedFlow(createForkFlow(), FlowRunTrigger.manual) + + const prompts = promptedPrompts() + expect(prompts).toContain('Branch A') + expect(prompts).toContain('Branch B') + expect(prompts.at(-1)).toBe('Merged: Branch A / Branch B') + expect(mocks.markRunSucceeded).toHaveBeenCalledWith('run-1', expect.objectContaining({ openCodeSessionId: 'session-1' })) + expect(mocks.markRunFailed).not.toHaveBeenCalled() + }) + + it('creates a separate session per fork branch', async () => { + mocks.userFindByIdSelect.mockResolvedValue({ slug: 'alice' }) + mocks.createRun.mockResolvedValue(createRunRecord()) + mocks.runFlowPromptAndReadOutput.mockImplementation(async (params: { prompt: string }) => ({ + ok: true, + output: params.prompt, + })) + const createSession = vi.fn() + .mockResolvedValueOnce({ data: { id: 'session-primary' } }) + .mockResolvedValueOnce({ data: { id: 'session-branch-a' } }) + .mockResolvedValueOnce({ data: { id: 'session-branch-b' } }) + mocks.createInstanceClient.mockResolvedValue({ session: { create: createSession } }) + + await runClaimedFlow(createForkFlow(), FlowRunTrigger.manual) + + expect(mocks.markRunSucceeded).toHaveBeenCalled() + const titles = createSession.mock.calls.map((call) => (call[0] as { title: string }).title) + expect(titles).toHaveLength(3) + expect(titles.some((title) => title.includes('Hunt bugs'))).toBe(true) + expect(titles.some((title) => title.includes('Hunt perf'))).toBe(true) + }) + + it('fails the run without executing the join when a branch fails', async () => { + mocks.userFindByIdSelect.mockResolvedValue({ slug: 'alice' }) + mocks.createRun.mockResolvedValue(createRunRecord()) + mocks.runFlowPromptAndReadOutput.mockImplementation(async (params: { prompt: string }) => { + if (params.prompt === 'Branch B') { + return { ok: false, type: 'failed', error: 'flow_step_failed' } + } + return { ok: true, output: params.prompt } + }) + + await runClaimedFlow(createForkFlow(), FlowRunTrigger.manual) + + expect(promptedPrompts()).not.toContain('Merged: Branch A / Branch B') + expect(mocks.markRunFailed).toHaveBeenCalledWith('run-1', expect.objectContaining({ error: 'flow_step_failed' })) + expect(mocks.markRunSucceeded).not.toHaveBeenCalled() + }) + }) + it('keeps the flow run and lease active when runtime termination is unconfirmed', async () => { mocks.userFindByIdSelect.mockResolvedValue({ slug: 'alice' }) mocks.runFlowPromptAndReadOutput.mockResolvedValue({ diff --git a/apps/web/src/lib/flows/__tests__/validation.test.ts b/apps/web/src/lib/flows/__tests__/validation.test.ts index ec86c2ab2..87bb5fc3b 100644 --- a/apps/web/src/lib/flows/__tests__/validation.test.ts +++ b/apps/web/src/lib/flows/__tests__/validation.test.ts @@ -259,3 +259,142 @@ describe('validateFlowDefinition', () => { } }) }) + +describe('fork topology validation', () => { + function agentNode(id: string, name: string, promptTemplate = 'Work') { + return { compactOutput: false, id, name, promptTemplate, targetAgentId: null, type: 'agent' as const } + } + + function forkNode(id: string, joinNodeId: string, name = 'Fan out') { + return { id, joinNodeId, name, type: 'fork' as const } + } + + function createForkDefinition(): FlowDefinition { + return { + edges: [ + { id: 'edge-1', sourceNodeId: 'agent-1', targetNodeId: 'fork-1' }, + { id: 'edge-2', sourceNodeId: 'fork-1', targetNodeId: 'agent-2' }, + { id: 'edge-3', sourceNodeId: 'fork-1', targetNodeId: 'agent-3' }, + { id: 'edge-4', sourceNodeId: 'agent-2', targetNodeId: 'merge-1' }, + { id: 'edge-5', sourceNodeId: 'agent-3', targetNodeId: 'merge-1' }, + ], + nodes: [ + agentNode('agent-1', 'Orient'), + forkNode('fork-1', 'merge-1'), + agentNode('agent-2', 'Hunt bugs'), + agentNode('agent-3', 'Hunt perf'), + { id: 'merge-1', name: 'Collect', type: 'merge' }, + ], + startNodeId: 'agent-1', + version: 1, + } + } + + it('accepts a well-formed fork/join flow and keeps the join reference', () => { + const result = validateFlowDefinition(createForkDefinition()) + + expect(result.ok).toBe(true) + if (!result.ok) return + const fork = result.definition.nodes.find((node) => node.id === 'fork-1') + expect(fork?.type === 'fork' && fork.joinNodeId).toBe('merge-1') + }) + + it('rejects fork nodes without a join reference', () => { + const cases: unknown[] = [ + { ...forkNode('fork-1', 'merge-1'), joinNodeId: '' }, + { ...forkNode('fork-1', 'merge-1'), joinNodeId: ' ' }, + { id: 'fork-1', name: 'Fan out', type: 'fork' }, + ] + + for (const fork of cases) { + const definition = createForkDefinition() + definition.nodes = [fork as FlowDefinition['nodes'][number], ...definition.nodes.slice(1)] + expect(validateFlowDefinition(definition)).toEqual({ ok: false, error: 'invalid_flow_nodes' }) + } + }) + + it('rejects non-condition nodes with more than one outgoing edge', () => { + const definition = createForkDefinition() + definition.edges.push({ id: 'edge-6', sourceNodeId: 'agent-1', targetNodeId: 'merge-1' }) + + expect(validateFlowDefinition(definition)).toEqual({ ok: false, error: 'multiple_outgoing_edges:agent-1' }) + }) + + it('rejects fork joins that are missing or not merge nodes', () => { + const missing = createForkDefinition() + missing.nodes[1] = forkNode('fork-1', 'missing') + expect(validateFlowDefinition(missing)).toEqual({ ok: false, error: 'fork_unknown_join:fork-1' }) + + const notMerge = createForkDefinition() + notMerge.nodes[1] = forkNode('fork-1', 'agent-2') + expect(validateFlowDefinition(notMerge)).toEqual({ ok: false, error: 'fork_join_not_merge:fork-1' }) + }) + + it('rejects two forks declaring the same join', () => { + const definition = createForkDefinition() + definition.nodes.push(forkNode('fork-2', 'merge-1', 'Fan out again')) + + expect(validateFlowDefinition(definition)).toEqual({ ok: false, error: 'fork_join_shared' }) + }) + + it('rejects forks with fewer than two branches or a direct fork-to-join edge', () => { + const single = createForkDefinition() + single.edges = single.edges.filter((edge) => edge.id !== 'edge-3' && edge.id !== 'edge-5') + expect(validateFlowDefinition(single)).toEqual({ ok: false, error: 'fork_without_branches:fork-1' }) + + const direct = createForkDefinition() + direct.edges.push({ id: 'edge-6', sourceNodeId: 'fork-1', targetNodeId: 'merge-1' }) + expect(validateFlowDefinition(direct)).toEqual({ ok: false, error: 'fork_branch_empty:fork-1' }) + }) + + it('rejects branches that dead-end before reaching the join', () => { + const deadEnd = createForkDefinition() + deadEnd.edges = deadEnd.edges.filter((edge) => edge.id !== 'edge-4') + expect(validateFlowDefinition(deadEnd)).toEqual({ ok: false, error: 'fork_branch_dead_end:agent-2' }) + }) + + it('rejects human and slack nodes inside a branch region', () => { + const definition = createForkDefinition() + definition.nodes[2] = { id: 'agent-2', instructions: 'Review', name: 'Hunt bugs', required: true, type: 'human' } + + expect(validateFlowDefinition(definition)).toEqual({ ok: false, error: 'fork_branch_unsupported_node:agent-2' }) + }) + + it('rejects join inputs from outside the branch region', () => { + const definition = createForkDefinition() + definition.nodes.push(agentNode('agent-4', 'Side chain')) + definition.edges.push({ id: 'edge-6', sourceNodeId: 'agent-4', targetNodeId: 'merge-1' }) + + expect(validateFlowDefinition(definition)).toEqual({ ok: false, error: 'fork_join_external_input:fork-1' }) + }) + + it('accepts nested forks with their own joins', () => { + const definition: FlowDefinition = { + edges: [ + { id: 'edge-1', sourceNodeId: 'agent-1', targetNodeId: 'fork-1' }, + { id: 'edge-2', sourceNodeId: 'fork-1', targetNodeId: 'agent-2' }, + { id: 'edge-3', sourceNodeId: 'fork-1', targetNodeId: 'fork-2' }, + { id: 'edge-4', sourceNodeId: 'agent-2', targetNodeId: 'merge-1' }, + { id: 'edge-5', sourceNodeId: 'fork-2', targetNodeId: 'agent-4' }, + { id: 'edge-6', sourceNodeId: 'fork-2', targetNodeId: 'agent-5' }, + { id: 'edge-7', sourceNodeId: 'agent-4', targetNodeId: 'merge-2' }, + { id: 'edge-8', sourceNodeId: 'agent-5', targetNodeId: 'merge-2' }, + { id: 'edge-9', sourceNodeId: 'merge-2', targetNodeId: 'merge-1' }, + ], + nodes: [ + agentNode('agent-1', 'Orient'), + forkNode('fork-1', 'merge-1'), + agentNode('agent-2', 'Hunt bugs'), + forkNode('fork-2', 'merge-2', 'Fan out inner'), + agentNode('agent-4', 'Hunt perf'), + agentNode('agent-5', 'Hunt improve'), + { id: 'merge-2', name: 'Collect inner', type: 'merge' }, + { id: 'merge-1', name: 'Collect', type: 'merge' }, + ], + startNodeId: 'agent-1', + version: 1, + } + + expect(validateFlowDefinition(definition).ok).toBe(true) + }) +}) diff --git a/apps/web/src/lib/flows/editor-graph.ts b/apps/web/src/lib/flows/editor-graph.ts index 90e5cabcc..0cf3e58ee 100644 --- a/apps/web/src/lib/flows/editor-graph.ts +++ b/apps/web/src/lib/flows/editor-graph.ts @@ -68,6 +68,10 @@ function updateNodeReferences(node: FlowNode, previousId: string, nextId: string return { ...node, promptTemplate: replaceNodeVariableReferences(node.promptTemplate, previousId, nextId) } } + if (node.type === 'fork') { + return { ...node, joinNodeId: node.joinNodeId === previousId ? nextId : node.joinNodeId } + } + return node } diff --git a/apps/web/src/lib/flows/node-executor-utils.ts b/apps/web/src/lib/flows/node-executor-utils.ts index c5a5882f3..bd5d39cd1 100644 --- a/apps/web/src/lib/flows/node-executor-utils.ts +++ b/apps/web/src/lib/flows/node-executor-utils.ts @@ -3,10 +3,14 @@ import { FlowNodeType as PrismaFlowNodeType } from '@prisma/client' import { FLOW_RUN_CANCELLED_ERROR, } from '@/lib/flows/session-executor' -import type { FlowNode } from '@/lib/flows/types' +import type { FlowNode, ForkFlowNode } from '@/lib/flows/types' import type { FlowRunStepRecord } from '@/lib/services/flow' -export function nodeTypeToPrisma(node: FlowNode): PrismaFlowNodeType { +// Fork nodes never record run steps (they are handled by the runner loop), +// so the step-record node mapping excludes them. +export type StepRecordFlowNode = Exclude + +export function nodeTypeToPrisma(node: StepRecordFlowNode): PrismaFlowNodeType { switch (node.type) { case 'agent': return PrismaFlowNodeType.agent diff --git a/apps/web/src/lib/flows/runner.ts b/apps/web/src/lib/flows/runner.ts index a7c6911df..187599235 100644 --- a/apps/web/src/lib/flows/runner.ts +++ b/apps/web/src/lib/flows/runner.ts @@ -9,6 +9,7 @@ import { formatFlowRunDate } from '@/lib/flows/cron' import { dispatchFlowExecution } from '@/lib/flows/execution-dispatcher' import { getFlowNodeById, getFlowOutgoingTargets } from '@/lib/flows/graph' import { executeFlowNode } from '@/lib/flows/node-executors' +import { errorMessage, replaceStep } from '@/lib/flows/node-executor-utils' import { planFlowRetry } from '@/lib/flows/retry-policy' import { validateFlowSlackNodeAccess } from '@/lib/flows/route-auth' import { serializeFlowRun } from '@/lib/flows/serializers' @@ -16,7 +17,7 @@ import { createFlowLeaseOwner, FLOW_LEASE_MS, } from '@/lib/flows/session-executor' -import type { FlowDefinition } from '@/lib/flows/types' +import type { FlowDefinition, ForkFlowNode } from '@/lib/flows/types' import { validateFlowDefinition } from '@/lib/flows/validation' import { createInstanceClient } from '@/lib/opencode/client' import { @@ -110,7 +111,12 @@ async function hasActiveFlowLease(flowId: string, leaseOwner: string): Promise { + // Branch fibers stop when they reach the fork's join node; the join is + // executed once by the parent cursor. + stopBeforeNodeId?: string + // `currentNodeId` is ambiguous while branches run concurrently, so fibers + // leave it untouched and only single-cursor segments track it. + trackCurrentNode?: boolean +} + +type FlowNodesResult = FlowExecutionOutcome & { steps: FlowRunStepRecord[] } + +async function runFlowNodes(params: FlowNodesParams): Promise { let currentNodeId = params.startNodeId let previousOutput = params.previousOutput let steps = params.steps @@ -130,24 +146,46 @@ async function executeFlowNodes(params: { while (currentNodeId) { if (visitedNodeIds.has(currentNodeId)) { - return { status: 'failed', error: 'cyclic_flow' } + return { status: 'failed', error: 'cyclic_flow', steps } + } + + if (params.stopBeforeNodeId && currentNodeId === params.stopBeforeNodeId) { + return { status: 'succeeded', steps } + } + + if (params.branchState?.aborted) { + return { status: 'cancelled', steps } } if (await isRunCancelled(params.run.id)) { - return { status: 'cancelled' } + return { status: 'cancelled', steps } } if (!await hasActiveFlowLease(params.flow.id, params.leaseOwner)) { - return { status: 'failed', error: 'flow_lease_lost' } + return { status: 'failed', error: 'flow_lease_lost', steps } } visitedNodeIds.add(currentNodeId) const node = getFlowNodeById(params.definition, currentNodeId) if (!node) { - return { status: 'failed', error: 'flow_node_not_found' } + return { status: 'failed', error: 'flow_node_not_found', steps } + } + + if (params.trackCurrentNode !== false) { + await flowService.updateRunCurrentNode(params.run.id, node.id) } - await flowService.updateRunCurrentNode(params.run.id, node.id) + if (node.type === 'fork') { + const forkResult = await executeForkBranches({ ...params, node, steps }) + steps = forkResult.steps + if (forkResult.status !== 'succeeded') return forkResult + + // Branch outputs are referenced through {{steps..output}}; the + // single cursor continues from the join without a branch-scoped output. + previousOutput = null + currentNodeId = node.joinNodeId + continue + } const result = await executeFlowNode({ client: params.client, @@ -164,17 +202,97 @@ async function executeFlowNodes(params: { }) steps = result.steps - if (result.status === 'cancelled') return { status: 'cancelled' } - if (result.status === 'failed') return { status: 'failed', error: result.error } + if (result.status === 'cancelled') return { status: 'cancelled', steps } + if (result.status === 'failed') return { status: 'failed', error: result.error, steps } if (result.status === 'termination_unconfirmed') { - return { status: 'termination_unconfirmed', cause: result.cause } + return { status: 'termination_unconfirmed', cause: result.cause, steps } } - if (result.status === 'waiting_for_human') return { status: 'waiting_for_human', nodeId: result.nodeId } + if (result.status === 'waiting_for_human') return { status: 'waiting_for_human', nodeId: result.nodeId, steps } previousOutput = result.previousOutput currentNodeId = result.nextNodeId } + return { status: 'succeeded', steps } +} + +// Each branch runs against its own OpenCode session: a workspace session +// executes one prompt at a time, so sharing the parent session would make +// concurrent branches fail spuriously with session_busy. +async function executeForkBranches( + params: FlowNodesParams & { node: ForkFlowNode; steps: FlowRunStepRecord[] }, +): Promise { + const branchStarts = getFlowOutgoingTargets(params.definition, params.node.id) + const branchState: FlowBranchState = { aborted: false } + const baseTitle = buildFlowSessionTitle(params.flow, params.run.scheduledFor) + + const results = await Promise.all(branchStarts.map(async (branchStartId): Promise => { + try { + const branchNode = getFlowNodeById(params.definition, branchStartId) + const sessionResult = await params.client.session.create( + { title: `${baseTitle} · ${branchNode?.name ?? branchStartId}` }, + { throwOnError: true }, + ) + if (!sessionResult.data) { + branchState.aborted = true + return { status: 'failed', error: 'flow_branch_session_create_failed', steps: params.steps } + } + + const result = await runFlowNodes({ + ...params, + branchState, + sessionId: sessionResult.data.id, + startNodeId: branchStartId, + steps: [...params.steps], + stopBeforeNodeId: params.node.joinNodeId, + trackCurrentNode: false, + }) + if (result.status === 'failed' || result.status === 'termination_unconfirmed') { + branchState.aborted = true + } + return result + } catch (error) { + branchState.aborted = true + return { status: 'failed', error: errorMessage(error, 'flow_branch_failed'), steps: params.steps } + } + })) + + let steps = params.steps + for (const result of results) { + for (const step of result.steps) { + steps = replaceStep(steps, step) + } + } + + const firstFailure = results.find((result) => result.status === 'termination_unconfirmed') + ?? results.find((result) => result.status === 'failed') + ?? results.find((result) => result.status === 'cancelled') + ?? results.find((result) => result.status === 'waiting_for_human') + + if (!firstFailure) { + return { status: 'succeeded', steps } + } + + if (firstFailure.status === 'termination_unconfirmed') { + return { status: 'termination_unconfirmed', cause: firstFailure.cause, steps } + } + if (firstFailure.status === 'failed') { + return { status: 'failed', error: firstFailure.error, steps } + } + if (firstFailure.status === 'cancelled') { + return { status: 'cancelled', steps } + } + // Validation forbids human nodes inside branches; a waiting branch cannot + // pause its siblings, so it surfaces as a failure. + return { status: 'failed', error: 'flow_branch_pause_unsupported', steps } +} + +async function executeFlowNodes(params: FlowNodesParams): Promise { + const result = await runFlowNodes(params) + if (result.status === 'failed') return { status: 'failed', error: result.error } + if (result.status === 'cancelled') return { status: 'cancelled' } + if (result.status === 'termination_unconfirmed') return { status: 'termination_unconfirmed', cause: result.cause } + if (result.status === 'waiting_for_human') return { status: 'waiting_for_human', nodeId: result.nodeId } return { status: 'succeeded' } } diff --git a/apps/web/src/lib/flows/types.ts b/apps/web/src/lib/flows/types.ts index eca34c0d3..1b2b55abf 100644 --- a/apps/web/src/lib/flows/types.ts +++ b/apps/web/src/lib/flows/types.ts @@ -97,6 +97,13 @@ export type MergeFlowNode = { name: string } +export type ForkFlowNode = { + id: string + type: 'fork' + name: string + joinNodeId: string +} + export type CompactionFlowNode = { id: string type: 'compaction' @@ -111,6 +118,7 @@ export type FlowNode = | SlackFlowNode | MergeFlowNode | CompactionFlowNode + | ForkFlowNode export type FlowEdge = { id: string diff --git a/apps/web/src/lib/flows/validation.ts b/apps/web/src/lib/flows/validation.ts index ff55f5fdc..47d3902cd 100644 --- a/apps/web/src/lib/flows/validation.ts +++ b/apps/web/src/lib/flows/validation.ts @@ -1,4 +1,4 @@ -import { getFlowTraversalTargets } from '@/lib/flows/graph' +import { getFlowNodeById, getFlowTraversalTargets } from '@/lib/flows/graph' import { isRecord } from '@/lib/records' import type { @@ -14,6 +14,7 @@ import type { FlowNode, FlowSlackMessageMode, FlowSlackTarget, + ForkFlowNode, HumanFlowNode, MergeFlowNode, SlackFlowNode, @@ -192,6 +193,13 @@ function parseNode(value: unknown): FlowNode | null { return { id, name, type } satisfies MergeFlowNode } + if (type === 'fork') { + const joinNodeId = readString(value, 'joinNodeId') + if (!joinNodeId) return null + + return { id, joinNodeId, name, type } satisfies ForkFlowNode + } + if (type === 'compaction') { const promptTemplate = readString(value, 'promptTemplate') if (!promptTemplate) return null @@ -275,6 +283,89 @@ function hasCycle(definition: FlowDefinition): boolean { return definition.nodes.some((node) => visit(node.id)) } +// Nodes that pause a run (human) or notify asynchronously (slack) are not +// supported inside parallel branches: a paused branch has no resume semantics +// while its siblings keep executing. +const FORK_UNSUPPORTED_BRANCH_NODE_TYPES = new Set(['human', 'slack']) + +function computeForkBranchRegion(definition: FlowDefinition, fork: ForkFlowNode): Set { + const region = new Set() + const queue = getFlowTraversalTargets(definition, fork.id) + .filter((nodeId) => nodeId !== fork.joinNodeId) + + while (queue.length > 0) { + const nodeId = queue.pop()! + if (nodeId === fork.joinNodeId || region.has(nodeId)) continue + + region.add(nodeId) + for (const target of getFlowTraversalTargets(definition, nodeId)) { + if (target !== fork.joinNodeId) queue.push(target) + } + } + + return region +} + +// Enforces the single-cursor contract for non-condition nodes (extra outgoing +// edges are silently dropped at runtime) and the fork/join topology rules. +// The graph is acyclic by the time this runs, so "every branch edge stays in +// the branch region or targets the join" guarantees every branch path reaches +// the join. +function validateFlowGraph(definition: FlowDefinition): string | null { + for (const node of definition.nodes) { + if (node.type === 'condition' || node.type === 'fork') continue + if (getFlowTraversalTargets(definition, node.id).length > 1) { + return `multiple_outgoing_edges:${node.id}` + } + } + + const declaredJoinIds = new Set() + for (const node of definition.nodes) { + if (node.type !== 'fork') continue + if (declaredJoinIds.has(node.joinNodeId)) return 'fork_join_shared' + declaredJoinIds.add(node.joinNodeId) + } + + for (const node of definition.nodes) { + if (node.type !== 'fork') continue + + const join = getFlowNodeById(definition, node.joinNodeId) + if (!join) return `fork_unknown_join:${node.id}` + if (join.type !== 'merge') return `fork_join_not_merge:${node.id}` + + const branchStarts = getFlowTraversalTargets(definition, node.id) + if (branchStarts.length < 2) return `fork_without_branches:${node.id}` + if (branchStarts.some((branchStart) => branchStart === node.joinNodeId)) { + return `fork_branch_empty:${node.id}` + } + + const region = computeForkBranchRegion(definition, node) + for (const nodeId of region) { + const branchNode = getFlowNodeById(definition, nodeId) + if (!branchNode) continue + + if (FORK_UNSUPPORTED_BRANCH_NODE_TYPES.has(branchNode.type)) { + return `fork_branch_unsupported_node:${nodeId}` + } + + // The region is the reachable set, so every branch edge already lands in + // the region or on the join; only a dead end can stop a path from + // reaching the join in an acyclic graph. + if (getFlowTraversalTargets(definition, nodeId).length === 0) { + return `fork_branch_dead_end:${nodeId}` + } + } + + for (const edge of definition.edges) { + if (edge.targetNodeId === node.joinNodeId && !region.has(edge.sourceNodeId)) { + return `fork_join_external_input:${node.id}` + } + } + } + + return null +} + export function validateFlowDefinition(value: unknown): FlowDefinitionValidationResult { if (!isRecord(value) || value.version !== 1) { return { ok: false, error: 'invalid_definition_version' } @@ -372,6 +463,11 @@ export function validateFlowDefinition(value: unknown): FlowDefinitionValidation return { ok: false, error: 'cyclic_flow' } } + const graphError = validateFlowGraph(definition) + if (graphError) { + return { ok: false, error: graphError } + } + return { ok: true, definition } } diff --git a/openspec/changes/parallel-flow-execution/.openspec.yaml b/openspec/changes/parallel-flow-execution/.openspec.yaml new file mode 100644 index 000000000..032461ff9 --- /dev/null +++ b/openspec/changes/parallel-flow-execution/.openspec.yaml @@ -0,0 +1,2 @@ +schema: spec-driven +created: 2026-09-02 diff --git a/openspec/changes/parallel-flow-execution/design.md b/openspec/changes/parallel-flow-execution/design.md new file mode 100644 index 000000000..ee8b36dc3 --- /dev/null +++ b/openspec/changes/parallel-flow-execution/design.md @@ -0,0 +1,53 @@ +## Context + +`executeFlowNodes` walks one `currentNodeId`; `nextEdge()` takes the first outgoing edge; `executeMergeNode` records a pass-through step. Fork/join adds a second execution mode between a `fork` and its paired `merge` while single-cursor execution stays the default everywhere else. + +## Goals + +- Parallel branch execution with real concurrency for LLM-bound steps. +- Failure/cancellation semantics that never leave the run in an inconsistent state. +- Deterministic validation so mis-wired forks fail at save/import/run time instead of at runtime. + +## Non-goals + +- Editor authoring UI, human/slack nodes inside branches, dynamic joins (see proposal). + +## Decisions + +### Branches get their own OpenCode sessions + +The design sketch assumed branches share the parent session. They cannot: a workspace session runs one prompt at a time, and `createActiveRunAfterRuntimeStateCheck` returns `session_busy` (or steals a stale lock) when two runs race on the same `(slug, sessionId)`. Each branch fiber therefore creates its own session via `client.session.create` with the title ` · `. Branch sessions are not attached to the run record (`attachRunSession` stays reserved for the primary session); branch steps are recorded per node as usual, which is what the template context and run timeline consume. + +### Steps: per-fiber copies, merged per nodeId + +`replaceStep` is immutable and keyed by `nodeId`, so concurrent fibers mutating one array would lose updates. Each fiber starts from a copy of the parent steps and returns its own array; the parent folds fiber results through `replaceStep` after `Promise.all`. Merge order does not matter because keys are unique per node. + +### Fail-fast via a shared abort flag + +Promises cannot be cancelled, and relying on run-level cancellation alone would keep sibling branches burning tokens after a failure. Each fork creates a `branchState = { aborted }` object shared by its fibers; the fiber that fails (or that cannot confirm termination) sets the flag, and every fiber checks it at the top of its loop and returns `cancelled`. The fork's outcome aggregation gives `termination_unconfirmed` precedence over `failed`, then `cancelled`, then `waiting_for_human` (which validation forbids inside branches and surfaces as `flow_branch_pause_unsupported` if it is ever reached). Run-level cancellation still wins through the existing per-iteration `isRunCancelled` check. + +### `currentNodeId` tracks the single cursor only + +Fibers run with `trackCurrentNode: false`, so `updateRunCurrentNode` records the fork before branches start and the join/merge afterwards. A retry after a fork failure resumes from the fork and re-runs the whole region; `upsertRunStep` is keyed by `(runId, nodeId)`, so re-run branch steps overwrite cleanly. + +### Validation: region containment instead of literal fan-in equality + +The design sketch required "the merge has exactly as many incoming edges as the fork has outgoing edges". That is too strict to be sound: a condition node inside a branch may legitimately route several rule targets into the join, breaking equality for a correct graph. The implemented rules express the same guarantee structurally: the branch region is the set of nodes reachable from the branch starts without passing through the join (so containment holds by construction), no region node may dead-end, the join accepts no incoming edges from outside the region, and the graph is acyclic — together these prove every branch path reaches the join. A direct fork→join edge (`fork_branch_empty`) is rejected, so the join is always fed by real branch work. + +### Additional structural rule: single outgoing edge outside condition/fork + +Nodes other than `condition` and `fork` with more than one outgoing edge are rejected (`multiple_outgoing_edges:`). Today such edges are silently dropped, which is the bug this change fixes; flagging these graphs is intentionally breaking for flows that were already broken at runtime. + +### Prisma schema untouched + +Fork nodes never record run steps (the runner intercepts them before `executeFlowNode`), so the `FlowNodeType` Prisma enum — and therefore the database — needs no migration. `nodeTypeToPrisma` is typed to exclude fork nodes so the compiler keeps it that way. + +## Risks / Trade-offs + +- Concurrent branch prompts multiply simultaneous provider load for the workspace (multiple sessions, same credentials) — accepted; per-provider rate limits apply as for interactive use. +- A fork flow edited in the flow editor renders the fork as an unknown node type (no inspector UI yet); definitions round-trip unchanged, and validation keeps edited flows saveable only while the fork topology stays intact. +- Nested forks work by recursion (each with its own join), but nothing in the UI communicates depth limits; recursion depth is bounded by the acyclic validation. + +## Open Questions + +- Should the run timeline surface branch sessions (currently only branch steps appear)? Defer to a follow-up with the editor UI work. diff --git a/openspec/changes/parallel-flow-execution/proposal.md b/openspec/changes/parallel-flow-execution/proposal.md new file mode 100644 index 000000000..d794632ef --- /dev/null +++ b/openspec/changes/parallel-flow-execution/proposal.md @@ -0,0 +1,20 @@ +## Why + +The flow runner executes nodes with a single cursor. When a node has multiple outgoing edges, only the first edge is followed — `nextEdge()` in `apps/web/src/lib/flows/node-executors.ts` takes `getFlowOutgoingTargets(...)[0]` and the rest are silently dropped. `merge` is a pass-through, not a join: it neither waits for incoming branches nor aggregates their outputs. A flow that fans out to three hunter agents then merges into a verify step actually executes only the first hunter; the others never run, and `{{steps..output}}` references downstream resolve to `null`, failing the run with `unknown_template_variable`. Validation accepts these graphs, so the breakage appears only at runtime. + +## What Changes + +- Add a `fork` flow node type (`{ id, name, type: 'fork', joinNodeId }`) whose outgoing edges each start an independent parallel branch. `joinNodeId` names the `merge` node where the branches reconverge, making the pairing explicit. +- Execute forks in the runner: when the single cursor reaches a `fork`, the runner spawns one execution fiber per outgoing branch — each an independent `runFlowNodes` traversal with its own visited set, initial `previousOutput`, and a shared fail-fast flag — and runs them via `Promise.all`. When every branch reaches the fork's join node, the parent cursor executes the `merge` node once and continues single-cursor execution. +- Give branches their own OpenCode sessions. A workspace session executes one prompt at a time (`createActiveRunAfterRuntimeStateCheck` fails with `session_busy` while busy), so concurrent branches sharing the parent session would fail spuriously. Each branch session is created from the workspace client with a derived title and is used only for that branch. +- Make branch failures fail the fork: the first branch failure (or runtime-termination uncertainty) sets the shared abort flag so sibling branches exit on their next loop iteration, and the fork reports the failure upstream. If the run is cancelled, all branches exit through the existing cancellation check. +- Merge branch steps into the parent template context: each fiber works on a copy of the step list and the parent merges results per `nodeId` after the branches settle, so `{{steps..output}}` resolves in nodes after the join. `previousOutput` after a fork is `null` — authors reference branch outputs explicitly. +- Validate the topology at save/import/run time (`validateFlowDefinition`): a node other than `condition` or `fork` may have at most one outgoing edge (extra edges are dropped at runtime today — this is a correctness fix that flags silently broken flows); fork joins must reference an existing `merge`, be unique per fork, declare at least two branches with no direct fork→join edge; every branch must reach the join (no dead ends); `human` and `slack` nodes are not allowed inside a branch region (no pause/notify semantics under parallelism yet); every incoming edge of a join must come from inside its branch region. +- Preserve backward compatibility: definitions without `fork` nodes execute exactly as before, and `merge` nodes without a paired fork remain pass-throughs. + +## Non-goals + +- No editor UI for authoring fork nodes (palette entry, join picker, canvas rendering). Fork flows are expressible via templates and the API in this change; editor authoring is a follow-up. +- No `human` or `slack` nodes inside branch regions — pausing one branch while siblings run needs branch-state persistence and a fork-aware resume path. +- No dynamic joins (conditional branch counts) and no mid-branch provider token refresh coordination beyond what the per-slug sync lock already provides. +- No change to template resolution: `buildFlowTemplateContext` already exposes every completed step, and `validateFlowPayload` already validates template references against all node ids at save time. diff --git a/openspec/changes/parallel-flow-execution/specs/flow-execution/spec.md b/openspec/changes/parallel-flow-execution/specs/flow-execution/spec.md new file mode 100644 index 000000000..c5abecde8 --- /dev/null +++ b/openspec/changes/parallel-flow-execution/specs/flow-execution/spec.md @@ -0,0 +1,56 @@ +## Purpose + +Defines the behavioral contract for parallel flow execution: how `fork` nodes fan work out into concurrent branches, how branches run and reconverge at their paired `merge` node, which graph topologies are valid, and how failure and cancellation propagate through a fork. + +## ADDED Requirements + +### Requirement: Fork nodes fan out into parallel branches + +A `fork` node SHALL start one independent branch per outgoing edge. Each branch SHALL execute as its own traversal with its own session on the workspace instance, so branch steps run concurrently. The fork SHALL record no run step of its own, and execution SHALL continue from the fork's paired join node only after every branch has reached it. + +#### Scenario: Two agent branches run concurrently + +- **WHEN** a flow reaches a `fork` with two outgoing edges into agent nodes that reconverge at the fork's join +- **THEN** both agent steps execute on separate sessions and the merge node executes once after both have finished + +#### Scenario: Branch outputs are available after the join + +- **WHEN** a node after the join references `{{steps..output}}` +- **THEN** the template resolves with the corresponding branch output + +### Requirement: Branch failures fail the fork without running the join + +If any branch fails, or cannot confirm runtime termination, the fork SHALL report that outcome upstream, sibling branches SHALL stop on their next execution step, and the join node SHALL NOT execute. A cancelled run SHALL cause all branches to exit. The merge node of a failed fork SHALL NOT be recorded. + +#### Scenario: One branch fails while a sibling is running + +- **WHEN** a branch step fails while a sibling branch is still executing +- **THEN** the sibling branch stops at its next loop iteration and the run fails with the first branch failure's error + +#### Scenario: The run is cancelled during a fork + +- **WHEN** the run is cancelled while branches are executing +- **THEN** every branch exits through the cancellation check and the run settles as cancelled + +### Requirement: Fork topology is validated before execution + +Flow definition validation SHALL reject: nodes other than `condition` and `fork` with more than one outgoing edge; a fork whose `joinNodeId` does not reference an existing `merge` node; two forks declaring the same join; a fork with fewer than two branches or with a direct fork→join edge; a branch that dead-ends before reaching the join; `human` or `slack` nodes inside a branch region; and incoming edges into a join from outside the fork's branch region. + +#### Scenario: Multi-edge node is rejected + +- **WHEN** a definition gives an `agent` node two outgoing edges +- **THEN** validation fails with `multiple_outgoing_edges` naming the node, instead of silently dropping the second edge at runtime + +#### Scenario: Dangling branch is rejected + +- **WHEN** a fork branch reaches a node whose edges do not lead back to the fork's join +- **THEN** validation fails instead of executing a branch that never reconverges + +### Requirement: Definitions without forks are unaffected + +A definition with no `fork` nodes SHALL execute exactly as before the change, and a `merge` node that no fork targets SHALL remain a pass-through. + +#### Scenario: Existing flow keeps its execution order + +- **WHEN** a definition that contains no `fork` nodes runs after the change +- **THEN** its steps execute in the same order and with the same outcomes as before diff --git a/openspec/changes/parallel-flow-execution/tasks.md b/openspec/changes/parallel-flow-execution/tasks.md new file mode 100644 index 000000000..b4bb01f4a --- /dev/null +++ b/openspec/changes/parallel-flow-execution/tasks.md @@ -0,0 +1,20 @@ +## 1. Fork node type + +- [x] 1.1 Add `ForkFlowNode` (`{ id, type: 'fork', name, joinNodeId }`) to the `FlowNode` union in `apps/web/src/lib/flows/types.ts`. +- [x] 1.2 Parse and validate `fork` nodes in `apps/web/src/lib/flows/validation.ts` (`joinNodeId` required non-empty string); narrow `nodeTypeToPrisma` in `apps/web/src/lib/flows/node-executor-utils.ts` to exclude fork so step records stay type-safe without a Prisma migration. + +## 2. Fork/join topology validation + +- [x] 2.1 Add `validateFlowGraph` to `validation.ts`: at most one outgoing edge for nodes other than `condition`/`fork`; fork joins must exist, be `merge` nodes, and be unique per fork; at least two branches and no direct fork→join edge; branch region containment (every branch edge stays in the region or targets the join, no dead ends); no `human`/`slack` nodes inside a branch region; no incoming join edges from outside the region. Verify with `validation.test.ts` cases for each rule plus backward-compatible definitions. +- [x] 2.2 Remap `joinNodeId` when the join node is renamed in `apps/web/src/lib/flows/editor-graph.ts` (`updateNodeReferences`). Verify with an `editor-graph.test.ts` case. + +## 3. Parallel execution in the runner + +- [x] 3.1 Split `executeFlowNodes` in `apps/web/src/lib/flows/runner.ts` into an internal `runFlowNodes` that carries the step list through its outcome and a wrapper preserving the public outcome type; add `stopBeforeNodeId`, `trackCurrentNode`, and `branchState` params. +- [x] 3.2 Handle `fork` nodes in the loop: spawn one fiber per outgoing branch (own OpenCode session, own steps copy, own visited set, shared abort flag, stopping at the join), aggregate outcomes with `termination_unconfirmed` > `failed` > `cancelled` precedence, merge branch steps into the parent context, continue single-cursor execution from the join with `previousOutput` reset to `null`. +- [x] 3.3 Verify with `runner.test.ts`: branches run and both outputs reach a downstream `{{steps..output}}` template; a branch failure fails the run without executing the merge; branch steps appear in the merged context; branch sessions are created per branch; definitions without forks execute unchanged. + +## 4. OpenSpec and checks + +- [x] 4.1 `openspec validate parallel-flow-execution --strict` passes. +- [x] 4.2 eslint clean on changed files; affected vitest suites pass.