This commit is contained in:
lda
2026-05-22 11:43:05 +07:00 Verified
parent 9cd064d6ef
commit afafe40109
8 changed files with 139 additions and 127 deletions
+35 -32
View File
@@ -7,7 +7,7 @@ This context defines the core workflow runtime language used by `wf_core`,
**Scheduler Foundation**: **Scheduler Foundation**:
The runtime model that selects runnable frames and advances workflow execution without assuming there is only one active cursor. The runtime model that selects runnable frames and advances workflow execution without assuming there is only one active cursor.
_Avoid_: Concurrent foreach, parallel foreach foundation _Avoid_: Foreach feature, concurrent foreach implementation
**Frame**: **Frame**:
An execution cursor for one active portion of a workflow run. An execution cursor for one active portion of a workflow run.
@@ -26,7 +26,7 @@ The ordered list of runnable frame identifiers that defines deterministic schedu
_Avoid_: Frame scan, implicit dict order _Avoid_: Frame scan, implicit dict order
**Foreach Policy**: **Foreach Policy**:
The future runtime configuration that decides parallel foreach batching, item failure handling, and cancellation/drain behavior. The future runtime configuration that decides concurrent foreach admission, item failure handling, and quiescence behavior.
_Avoid_: Parallel flag _Avoid_: Parallel flag
**Blocked Frame**: **Blocked Frame**:
@@ -62,7 +62,7 @@ The committed state a frame may read based on workflow topology and completed up
_Avoid_: Shared live object _Avoid_: Shared live object
**Lineage Isolation**: **Lineage Isolation**:
The rule that a parallel child frame sees parent-visible state plus its own ancestor writes, but not sibling branch writes. The rule that a concurrent child frame sees parent-visible state plus its own ancestor writes, but not sibling branch writes.
_Avoid_: Global committed state _Avoid_: Global committed state
**Barrier**: **Barrier**:
@@ -92,15 +92,15 @@ _Avoid_: Job, invocation
the child/event it waits on wakes it. the child/event it waits on wakes it.
- A **Blocked Frame** should carry a **Block Reason** so deadlocks and wakeups - A **Blocked Frame** should carry a **Block Reason** so deadlocks and wakeups
are explainable. are explainable.
- Parallel foreach and native subgraphs depend on the **Scheduler Foundation**. - Concurrent foreach and native subgraphs depend on the **Scheduler Foundation**.
- Parallel foreach requires a **Foreach Policy** before public support is - Concurrent foreach requires a **Foreach Policy** before public support is
enabled. enabled.
- Future parallel foreach should be bounded by default. `max_active` and - Future concurrent foreach should be bounded by default. `max_active` and
`max_outstanding` prevent one workflow from spawning unbounded MCP, HTTP, `max_outstanding` prevent one workflow from spawning unbounded MCP, HTTP,
browser, or external service calls. browser, or external service calls.
- Future concurrency-specific foreach settings should live in a nested parallel - Future concurrency-specific foreach settings should live in a nested
policy object rather than expanding `ForeachNode` with many top-level fields. policy object rather than expanding `ForeachNode` with many top-level fields.
- Item error handling is foreach-wide, not parallel-only. Serial and parallel - Item error handling is foreach-wide, not concurrent-only. Serial and concurrent
foreach can both profit from `fail`, `skip`, or `collect` item failure policy. foreach can both profit from `fail`, `skip`, or `collect` item failure policy.
- `collect` item error policy must declare an explicit destination for - `collect` item error policy must declare an explicit destination for
structured item errors. Collected errors should be ordered by item index, not structured item errors. Collected errors should be ordered by item index, not
@@ -137,7 +137,7 @@ _Avoid_: Job, invocation
failure was collected. failure was collected.
- `skip` emits `completed_with_errors` when one or more item failures were - `skip` emits `completed_with_errors` when one or more item failures were
skipped. It emits `done` only when all items succeed. skipped. It emits `done` only when all items succeed.
- Parallel `fail` item policy should stop scheduling new items, drain already - Concurrent `fail` item policy should stop scheduling new items, drain already
started jobs to a quiescent point, capture their results safely, and then fail started jobs to a quiescent point, capture their results safely, and then fail
the run. It should not assume hard cancellation is safe. the run. It should not assume hard cancellation is safe.
- After `fail` trips, drained sibling results are for trace/observability and - After `fail` trips, drained sibling results are for trace/observability and
@@ -145,7 +145,7 @@ _Avoid_: Job, invocation
boundary. boundary.
- Foreach modes that continue after item failure, such as `collect` and `skip`, - Foreach modes that continue after item failure, such as `collect` and `skip`,
should buffer item state patches until the foreach barrier completes. Future should buffer item state patches until the foreach barrier completes. Future
parallel foreach should always use barrier-buffered commits. concurrent foreach should always use barrier-buffered commits.
- Serial `fail` may keep immediate commits for compatibility. Commit strategy - Serial `fail` may keep immediate commits for compatibility. Commit strategy
should be extracted into runtime helpers instead of being smeared through should be extracted into runtime helpers instead of being smeared through
foreach execution code. foreach execution code.
@@ -161,7 +161,7 @@ _Avoid_: Job, invocation
- Pending barrier results must live in resumable `RunState`/frame metadata, not - Pending barrier results must live in resumable `RunState`/frame metadata, not
only in trace. Trace records history; runtime state is what resume/checkpoint only in trace. Trace records history; runtime state is what resume/checkpoint
uses. uses.
- Parallel item frames need lineage-local pending state: later nodes in the same - Concurrent item frames need lineage-local pending state: later nodes in the same
item lineage can read earlier pending patches from that item, while sibling item lineage can read earlier pending patches from that item, while sibling
items cannot. The exact aggregate output API for committing item results to items cannot. The exact aggregate output API for committing item results to
parent state is deferred. parent state is deferred.
@@ -224,10 +224,10 @@ _Avoid_: Job, invocation
but structured foreach refs should be preferred for non-trivial authoring. but structured foreach refs should be preferred for non-trivial authoring.
- Future foreach policy shape should validate cross-field rules: `collect` - Future foreach policy shape should validate cross-field rules: `collect`
requires `collect_to`, non-collect actions forbid `collect_to`, requires `collect_to`, non-collect actions forbid `collect_to`,
`mode="parallel"` requires a parallel policy, and `mode="serial"` forbids a `mode="concurrent"` requires a concurrent policy, and `mode="serial"` forbids
parallel policy. Deprecated top-level `on_item_error` may parse into the a concurrent policy. Deprecated top-level `on_item_error` may parse into the
nested item error policy, but canonical dumps should use the nested shape. nested item error policy, but canonical dumps should use the nested shape.
- `ForeachParallelPolicy` should split limits into `max_active` and - `ForeachConcurrentPolicy` should split limits into `max_active` and
`max_outstanding`. Defaults are `max_active=4` and `max_outstanding=20`; `max_outstanding`. Defaults are `max_active=4` and `max_outstanding=20`;
validation requires `max_outstanding >= max_active`. Ready or running item validation requires `max_outstanding >= max_active`. Ready or running item
frames consume active capacity; blocked item frames consume outstanding frames consume active capacity; blocked item frames consume outstanding
@@ -260,7 +260,7 @@ _Avoid_: Job, invocation
backpressure inside the admitted node-call boundary, not workflow-level backpressure inside the admitted node-call boundary, not workflow-level
`BLOCKED` state that frees foreach active capacity. `BLOCKED` state that frees foreach active capacity.
- An **Interrupt** pauses the whole **Run**, even if the interrupted frame is a - An **Interrupt** pauses the whole **Run**, even if the interrupted frame is a
child of future parallel work. child of future concurrent work.
- A **Runtime Failure** is distinct from a node returning an `error` outcome. - A **Runtime Failure** is distinct from a node returning an `error` outcome.
Outcomes are graph control flow; runtime failures are scheduler stops unless Outcomes are graph control flow; runtime failures are scheduler stops unless
a policy handles them. a policy handles them.
@@ -269,32 +269,34 @@ _Avoid_: Job, invocation
**Runtime Failure**. **Runtime Failure**.
- Missing edges for declared outcomes are validation errors, not normal runtime - Missing edges for declared outcomes are validation errors, not normal runtime
branch policy. branch policy.
- Unsupported parallel foreach semantics should be rejected by validation before - Unsupported concurrent foreach semantics should be rejected by validation before
runtime. Runtime may stay defensive, but validation owns the user-facing gate. runtime. Runtime may stay defensive, but validation owns the user-facing gate.
- `on_item_error="collect"` and `"skip"` are future policy shapes unless runtime - `on_item_error="collect"` and `"skip"` are future policy shapes unless runtime
support is explicitly implemented. Current scheduler work should make official support is explicitly implemented. Current scheduler work should make official
support easier, not pretend it already exists. support easier, not pretend it already exists.
- A **Trace** records actual scheduler execution order; grouping or sorting by - A **Trace** records actual scheduler execution order; grouping or sorting by
foreach index is a presentation concern. foreach index is a presentation concern.
- Parallel child frames may write to the same state path only through a - Concurrent child frames may write to the same state path only through a
**Reducer**. **Reducer**.
- Future parallel frames should produce **State Patches** that the scheduler - Future concurrent frames should produce **State Patches** that the scheduler
commits atomically. commits atomically.
- A frame's **State Visibility** excludes uncommitted sibling writes. - A frame's **State Visibility** excludes uncommitted sibling writes.
- Future parallel branches require **Lineage Isolation**: sibling branch writes - Future concurrent branches require **Lineage Isolation**: sibling branch writes
are invisible unless the graph explicitly joins or merges them. are invisible unless the graph explicitly joins or merges them.
- A **Barrier** is the explicit merge boundary for parallel lineage patches. - A **Barrier** is the explicit merge boundary for concurrent lineage patches.
- Foreach may own an implicit **Barrier**; future graph-level convergence may use - Foreach may own an implicit **Barrier**; future graph-level convergence may use
an explicit barrier node. an explicit barrier node.
## Example Dialogue ## Example Dialogue
> **Dev:** "Are we implementing parallel foreach now?" > **Dev:** "Are we implementing concurrent foreach now?"
> **Domain expert:** "No. First we are implementing the **Scheduler Foundation** so a **Run** can eventually manage multiple runnable **Frames** safely." > **Domain expert:** "No. First we are implementing the **Scheduler Foundation** so a **Run** can eventually manage multiple runnable **Frames** safely."
## Flagged Ambiguities ## Flagged Ambiguities
- "concurrent foreach" was used for the next task, but the resolved scope is **Scheduler Foundation**: deterministic internal runtime prep before public parallel foreach support. - "concurrent foreach" is the future workflow mode. The resolved current scope is
**Scheduler Foundation**: deterministic internal runtime prep before public
concurrent foreach support.
- `current_frame_id` / `current_node_id` are compatibility fields for the selected cursor, not the source of truth for all runnable work. - `current_frame_id` / `current_node_id` are compatibility fields for the selected cursor, not the source of truth for all runnable work.
- `current_frame_id` remains persisted for compatibility, but means "the frame - `current_frame_id` remains persisted for compatibility, but means "the frame
currently selected by the scheduler", not "the only live frame". currently selected by the scheduler", not "the only live frame".
@@ -311,7 +313,7 @@ _Avoid_: Job, invocation
interrupt reason, because it carries externally visible resume semantics. interrupt reason, because it carries externally visible resume semantics.
- Scheduling order should be explicit through a **Ready Queue**, not inferred - Scheduling order should be explicit through a **Ready Queue**, not inferred
from frame dictionary iteration. from frame dictionary iteration.
- `mode="parallel"` stays unsupported until **Foreach Policy** and parent - `mode="concurrent"` stays unsupported until **Foreach Policy** and parent
completion semantics are explicit. completion semantics are explicit.
- A **Run** has at most one outstanding **Interrupt**. Resume wakes the frame - A **Run** has at most one outstanding **Interrupt**. Resume wakes the frame
referenced by that interrupt and scheduling continues from the **Ready Queue**. referenced by that interrupt and scheduling continues from the **Ready Queue**.
@@ -321,15 +323,16 @@ _Avoid_: Job, invocation
- Ready sibling frames are preserved during an **Interrupt**, but the resumed - Ready sibling frames are preserved during an **Interrupt**, but the resumed
frame is placed at the front of the **Ready Queue** before scheduling frame is placed at the front of the **Ready Queue** before scheduling
continues. continues.
- Future parallel execution should not assume in-flight node calls can be - Future concurrent execution should not assume in-flight node calls can be
safely cancelled. When an **Interrupt** occurs, scheduling should stop; already safely cancelled. When an **Interrupt** occurs, scheduling should stop; already
started jobs should drain to pending results before resume/commit policy started jobs should drain to pending results before resume/commit policy
decides what becomes visible. decides what becomes visible.
- Future parallel interrupts should return control to the caller only at a - Future concurrent interrupts should return control to the caller only at a
quiescent pause point: no new work is scheduled, already-started jobs have quiescent pause point: no new work is scheduled, already-started jobs have
drained, and their results are captured without unsafe commits. drained, and their results are captured without unsafe commits.
- Future async execution must protect append-only **Trace** writes, but should - Future async execution must protect append-only **Trace** writes, but should
not change trace semantics. not change trace semantics. Async runtime is an execution capability for
simultaneous async node handlers, not a separate foreach workflow mode.
- Scheduler block/wake events should not be added to the public **Trace** in - Scheduler block/wake events should not be added to the public **Trace** in
the first pass. Future observability, including OpenTelemetry-style spans, is the first pass. Future observability, including OpenTelemetry-style spans, is
a separate concern. a separate concern.
@@ -339,7 +342,7 @@ _Avoid_: Job, invocation
but raise for malformed metadata on the right kind. Corrupt runtime metadata but raise for malformed metadata on the right kind. Corrupt runtime metadata
is a runtime invariant failure. is a runtime invariant failure.
- First-pass **Block Reason** support only needs child-frame blocking; future - First-pass **Block Reason** support only needs child-frame blocking; future
interrupt, subgraph, and parallel barrier reasons can extend the same shape. interrupt, subgraph, and concurrent barrier reasons can extend the same shape.
- Frame creation must reject duplicate frame identifiers. The **Ready Queue** - Frame creation must reject duplicate frame identifiers. The **Ready Queue**
must contain only existing pending frames and must never contain duplicate must contain only existing pending frames and must never contain duplicate
frame identifiers. frame identifiers.
@@ -378,10 +381,10 @@ _Avoid_: Job, invocation
- Serial foreach blocks the parent on one iteration child at a time, so the - Serial foreach blocks the parent on one iteration child at a time, so the
child runs until completion/interruption/failure before the parent wakes. child runs until completion/interruption/failure before the parent wakes.
- Future async execution must protect shared state writes. Last-writer-wins is - Future async execution must protect shared state writes. Last-writer-wins is
not an acceptable merge policy for parallel child frames. not an acceptable merge policy for concurrent child frames.
- Serial execution may continue mutating state immediately until parallel - Serial execution may continue mutating state immediately until concurrent
execution needs scheduler-controlled **State Patch** commits. execution needs scheduler-controlled **State Patch** commits.
- Parallel sibling frames must not depend on observing each other's writes. Use - Concurrent sibling frames must not depend on observing each other's writes. Use
serial foreach or explicit graph structure for ordered dependencies. serial foreach or explicit graph structure for ordered dependencies.
- Missing reducers mean replace only within a single serial lineage. At a - Missing reducers mean replace only within a single serial lineage. At a
**Barrier**, multiple writes to the same path without a reducer are conflicts, **Barrier**, multiple writes to the same path without a reducer are conflicts,
@@ -390,10 +393,10 @@ _Avoid_: Job, invocation
repurposed later only through an explicit design pass. repurposed later only through an explicit design pass.
- The first **Scheduler Foundation** implementation should create ready/block/wake - The first **Scheduler Foundation** implementation should create ready/block/wake
seams only. **Lineage Isolation** and **Barrier** merge behavior are future seams only. **Lineage Isolation** and **Barrier** merge behavior are future
parallel semantics, not first-pass behavior. concurrent semantics, not first-pass behavior.
- First-pass scheduler code should add durable seams such as ready queue, - First-pass scheduler code should add durable seams such as ready queue,
block/wake helpers, and typed foreach metadata. It should not add public block/wake helpers, and typed foreach metadata. It should not add public
parallel policy, barrier, snapshot, or lineage-patch fields before those concurrent policy, barrier, snapshot, or lineage-patch fields before those
semantics are enforced. semantics are enforced.
- Scheduler rules are shared by sync and async runtime paths; only node handler - Scheduler rules are shared by sync and async runtime paths; only node handler
execution differs. execution differs.
+4 -3
View File
@@ -81,6 +81,7 @@ Two runtime features may come back after the platform/DX roadmap advances:
- Native subgraph execution with child run state, interrupt bubbling, and resume - Native subgraph execution with child run state, interrupt bubbling, and resume
back into the child workflow. back into the child workflow.
- Async parallel foreach with explicit scheduling, reducer/merge semantics, and - Concurrent foreach with explicit scheduling, reducer/merge semantics, and
failure policy. Do not implement this as plain `asyncio.gather` over sync failure policy. Sync runtime can interleave item frames deterministically;
handlers. async runtime can add simultaneous async node handler execution. Do not
implement this as plain `asyncio.gather` over sync handlers.
@@ -1,10 +1,10 @@
# Scheduler Foundation Before Parallel Foreach # Scheduler Foundation Before Concurrent Foreach
Status: accepted Status: accepted
We will introduce an explicit scheduler foundation in `wf_core` before enabling We will introduce an explicit scheduler foundation in `wf_core` before enabling
parallel foreach. The first implementation preserves deterministic serial concurrent foreach. The first implementation preserves deterministic serial
behavior while adding ready/block/wake semantics, because parallel foreach and behavior while adding ready/block/wake semantics, because concurrent foreach and
native subgraphs both require multiple runnable frames without treating native subgraphs both require multiple runnable frames without treating
`current_frame_id` as the whole runtime state. `current_frame_id` as the whole runtime state.
@@ -15,9 +15,9 @@ The current runtime behaves like a stack cursor. `RunState.current_frame_id` and
at a time before control collapses back to the parent. This works for serial at a time before control collapses back to the parent. This works for serial
execution, but it does not give the runtime a clear way to represent multiple execution, but it does not give the runtime a clear way to represent multiple
runnable frames, blocked parent frames, interrupt resume priority, or future runnable frames, blocked parent frames, interrupt resume priority, or future
parallel branch merge semantics. concurrent branch merge semantics.
Parallel foreach is not just `asyncio.gather` over node calls. It changes state Concurrent foreach is not just `asyncio.gather` over node calls. It changes state
visibility, error handling, interrupt behavior, trace ordering, reducer rules, visibility, error handling, interrupt behavior, trace ordering, reducer rules,
and parent completion semantics. Enabling it before those concepts exist would and parent completion semantics. Enabling it before those concepts exist would
make the runtime fragile. make the runtime fragile.
@@ -25,7 +25,7 @@ make the runtime fragile.
## Decision ## Decision
Add an internal scheduler foundation first. The first pass will keep serial Add an internal scheduler foundation first. The first pass will keep serial
workflow behavior intact and will not enable `foreach(mode="parallel")`. workflow behavior intact and will not enable `foreach(mode="concurrent")`.
The scheduler model is: The scheduler model is:
@@ -56,7 +56,7 @@ the next item or finish.
## Rejected Alternatives ## Rejected Alternatives
**Patch parallelism directly into foreach.** **Patch concurrent execution directly into foreach.**
This would keep the stack-cursor model and force foreach to own scheduling, This would keep the stack-cursor model and force foreach to own scheduling,
interrupt, merge, and failure policy. It would not help native subgraphs. interrupt, merge, and failure policy. It would not help native subgraphs.
@@ -74,22 +74,22 @@ refine this later.
The future barrier semantics are larger than the current join node. `JoinNode` The future barrier semantics are larger than the current join node. `JoinNode`
may be repurposed later, but not silently in this scheduler pass. may be repurposed later, but not silently in this scheduler pass.
## Future Parallel Semantics ## Future Concurrent Semantics
Parallel foreach and graph-level convergence require additional semantics before Concurrent foreach and graph-level convergence require additional semantics before
they are enabled: they are enabled:
- A future foreach policy must define batching, `max_concurrency`, item failure - A future foreach policy must define admission limits, item failure handling,
handling, and cancellation/drain behavior. and quiescence behavior.
- Parallel foreach should be bounded by default so workflows do not spawn - Concurrent foreach should be bounded by default so workflows do not spawn
unbounded MCP, HTTP, browser, or external service calls. unbounded MCP, HTTP, browser, or external service calls.
- A future barrier must wait for multiple child or upstream frames and merge - A future barrier must wait for multiple child or upstream frames and merge
their lineage patches. their lineage patches.
- Missing reducers mean replace only within one serial lineage. At a barrier, - Missing reducers mean replace only within one serial lineage. At a barrier,
multiple writes to the same path without a reducer are conflicts. multiple writes to the same path without a reducer are conflicts.
- Future parallel child frames should produce state patches that the scheduler - Future concurrent child frames should produce state patches that the scheduler
commits atomically. commits atomically.
- Parallel branches require lineage isolation: a child frame sees parent-visible - Concurrent branches require lineage isolation: a child frame sees parent-visible
state plus its own ancestor writes, not sibling branch writes. state plus its own ancestor writes, not sibling branch writes.
- Public trace remains append-only chronological execution history. Scheduler - Public trace remains append-only chronological execution history. Scheduler
block/wake events are not added to public trace in this pass. block/wake events are not added to public trace in this pass.
@@ -104,7 +104,7 @@ Resume wakes the interrupted frame, clears the run interrupt, and places the
resumed frame at the front of the ready queue. Ready sibling frames are preserved resumed frame at the front of the ready queue. Ready sibling frames are preserved
but do not run while the interrupt is outstanding. but do not run while the interrupt is outstanding.
Future parallel execution should not assume in-flight node calls can be safely Future concurrent execution should not assume in-flight node calls can be safely
cancelled. If one frame interrupts while sibling jobs are already started, the cancelled. If one frame interrupts while sibling jobs are already started, the
runtime should stop scheduling new work, let started jobs drain to pending runtime should stop scheduling new work, let started jobs drain to pending
results, and defer sibling state commits until resume/commit policy decides what results, and defer sibling state commits until resume/commit policy decides what
@@ -120,7 +120,7 @@ are graph control flow. Runtime failures stop scheduling unless a future policy
explicitly handles them. explicitly handles them.
In the first scheduler pass, any runtime failure fails the whole run. Future In the first scheduler pass, any runtime failure fails the whole run. Future
parallel foreach policies may support `collect` or `skip`, but those are policy concurrent foreach policies may support `collect` or `skip`, but those are policy
features and should not be treated as implemented until runtime support exists. features and should not be treated as implemented until runtime support exists.
## Compatibility Requirements ## Compatibility Requirements
@@ -1,36 +1,40 @@
# Parallel Foreach Policy and Barrier Commits # Concurrent Foreach Policy and Barrier Commits
Status: accepted Status: accepted
Parallel foreach will use bounded scheduling, lineage-local pending state, and Concurrent foreach will use bounded scheduling, lineage-local pending state, and
barrier-buffered commits. This keeps parallel item execution deterministic, barrier-buffered commits. This keeps item execution deterministic, resumable,
resumable, and safe around reducers, interrupts, and external tool calls. and safe around reducers, interrupts, and external tool calls.
## Context ## Context
The scheduler foundation makes multiple runnable frames possible, but parallel The scheduler foundation makes multiple runnable frames possible, but concurrent
foreach needs more than concurrent node calls. Each item lineage may execute foreach needs more than multiple node calls. Each item lineage may execute
several nodes, read its own pending writes, fail independently, block, or several nodes, read its own pending writes, fail independently, block, or
interrupt the whole run. Sibling item writes must not leak into each other, and interrupt the whole run. Sibling item writes must not leak into each other, and
state commits need deterministic merge behavior. state commits need deterministic merge behavior.
## Decision ## Decision
Parallel foreach will be async-runtime-only and bounded by policy. Concurrent foreach is a workflow mode and must be bounded by policy.
The future foreach model should separate item error behavior from concurrency The future foreach model should separate item error behavior from concurrency
behavior: behavior:
- `item_error` is foreach-wide and applies to serial and parallel foreach. - `item_error` is foreach-wide and applies to serial and concurrent foreach.
- `parallel` is a nested policy object used only when `mode="parallel"`. - `concurrent` is a nested policy object used only when
- `parallel.max_active` defaults to `4`. `mode="concurrent"`.
- `parallel.max_outstanding` defaults to `20`. - `concurrent.max_active` defaults to `4`.
- `concurrent.max_outstanding` defaults to `20`.
- Validation requires `max_outstanding >= max_active`. - Validation requires `max_outstanding >= max_active`.
- Ready or running item frames consume active capacity. - Ready or running item frames consume active capacity.
- Blocked item frames consume outstanding capacity but not active capacity. - Blocked item frames consume outstanding capacity but not active capacity.
`foreach(mode="parallel")` requires async execution. Sync execution should reject `foreach(mode="concurrent")` should be executable by the sync runtime as
it clearly rather than inventing thread/process semantics. deterministic frame interleaving: one admitted node handler call at a time. The
async runtime may additionally run admitted async node handler calls
simultaneously. Sync handlers should not be pushed into thread/process
parallelism by default.
## Item Error Policy ## Item Error Policy
@@ -71,7 +75,7 @@ failure as control flow, the outcome should be `failed`.
## Barrier-Buffered Commits ## Barrier-Buffered Commits
Parallel foreach item writes are buffered as pending state patches/results, not Concurrent foreach item writes are buffered as pending state patches/results, not
committed directly to `RunState.state`. committed directly to `RunState.state`.
The state model is: The state model is:
@@ -88,8 +92,8 @@ Completion-order merging may exist later as an explicit barrier merge policy,
not reducer behavior. not reducer behavior.
Patch creation and commit must extract/reuse the existing node output Patch creation and commit must extract/reuse the existing node output
validation, output binding, and reducer logic. Parallel foreach must not create validation, output binding, and reducer logic. Concurrent foreach must not
a second write system. create a second write system.
## Merge and Reducer Rules ## Merge and Reducer Rules
@@ -103,7 +107,7 @@ means item index order.
## Interrupt and Failure Quiescence ## Interrupt and Failure Quiescence
Future parallel execution should not assume in-flight node calls can be safely Future concurrent execution should not assume in-flight node calls can be safely
cancelled. cancelled.
If an interrupt or fail policy trips while sibling jobs are already started, the If an interrupt or fail policy trips while sibling jobs are already started, the
@@ -131,7 +135,7 @@ node calls, and those waits keep the frame `RUNNING` rather than `BLOCKED`.
## Context and Lineage ## Context and Lineage
Parallel item frames need lineage-local pending state. Later nodes in the same Concurrent item frames need lineage-local pending state. Later nodes in the same
item lineage can read earlier pending patches from that item, while siblings item lineage can read earlier pending patches from that item, while siblings
cannot. cannot.
@@ -157,5 +161,5 @@ Forks produce branch lineage tokens; gathers consume declared token sets and
produce merged tokens. This supports partial gathers such as merging `a+b` produce merged tokens. This supports partial gathers such as merging `a+b`
before later merging with `c`. before later merging with `c`.
Parallel foreach remains the nearer target because its implicit lineage tokens Concurrent foreach remains the nearer target because its implicit lineage tokens
are item indexes owned by one foreach activation. are item indexes owned by one foreach activation.
+11 -9
View File
@@ -40,15 +40,17 @@ implementation state.
## Runtime and Platform Roadmap ## Runtime and Platform Roadmap
- Scheduler foundation decision record: - Scheduler foundation decision record:
[ADR 0001](./adr/0001-scheduler-foundation-before-parallel-foreach.md). [ADR 0001](./adr/0001-scheduler-foundation-before-concurrent-foreach.md).
- Parallel foreach policy decision record: - Concurrent foreach policy decision record:
[ADR 0002](./adr/0002-parallel-foreach-policy-and-barrier-commits.md). [ADR 0002](./adr/0002-concurrent-foreach-policy-and-barrier-commits.md).
- **Native subgraphs / graph-as-node**: add child run state, child trace - **Native subgraphs / graph-as-node**: add child run state, child trace
preservation, interrupt bubbling, and resume back into the child workflow. preservation, interrupt bubbling, and resume back into the child workflow.
Wrapper artifacts currently execute as deployments and return run status; Wrapper artifacts currently execute as deployments and return run status;
true graph-as-node outcome propagation belongs here. true graph-as-node outcome propagation belongs here.
- **Async parallel foreach**: add explicit scheduling, reducer/merge semantics, - **Concurrent foreach**: add explicit scheduling, reducer/merge semantics, and
and failure policy. Do not model this as plain parallel calls over sync failure policy. Sync runtime can interleave admitted item frames one node at a
time; async runtime can additionally run admitted async node handlers
simultaneously. Do not model this as plain `asyncio.gather` over sync
handlers. handlers.
- **Persistent run history**: add a run store before adding stable `run_id`, - **Persistent run history**: add a run store before adding stable `run_id`,
`inspect_run`, or `read_run_trace(run_id, range)` APIs. Current traces are `inspect_run`, or `read_run_trace(run_id, range)` APIs. Current traces are
@@ -64,10 +66,10 @@ implementation state.
Frame stress points to solve before either feature: Frame stress points to solve before either feature:
- `RunState.current_frame_id` currently models one active execution cursor. - `RunState.current_frame_id` currently models the selected execution cursor.
Parallel foreach likely needs multiple runnable child frames. Concurrent foreach needs multiple runnable child frames.
- `ExecutionFrame.metadata` currently carries ad hoc foreach data. Subgraphs and - `ExecutionFrame.metadata` currently carries ad hoc foreach data. Subgraphs and
parallel foreach should get typed frame payloads or strongly bounded helper concurrent foreach should get typed frame payloads or strongly bounded helper
accessors before metadata grows more meanings. accessors before metadata grows more meanings.
- Subgraph frames need child workflow identity/version/deployment binding, not - Subgraph frames need child workflow identity/version/deployment binding, not
just a generic metadata dictionary. just a generic metadata dictionary.
@@ -79,6 +81,6 @@ Frame stress points to solve before either feature:
The MCP workflow authoring path is now usable enough for real testing. The next The MCP workflow authoring path is now usable enough for real testing. The next
bottleneck is runtime/platform correctness: resumable child execution, bottleneck is runtime/platform correctness: resumable child execution,
parallel scheduling, persistent run history, and protocol-native progress concurrent scheduling, persistent run history, and protocol-native progress
reporting. Those pieces should come before adding more high-level authoring reporting. Those pieces should come before adding more high-level authoring
sugar. sugar.
@@ -1,10 +1,10 @@
# Parallel Foreach Roadmap Implementation Plan # Concurrent Foreach Roadmap Implementation Plan
> **For agentic workers:** REQUIRED SUB-SKILL: Use superpowers:subagent-driven-development (recommended) or superpowers:executing-plans to implement this plan task-by-task. Steps use checkbox (`- [ ]`) syntax for tracking. > **For agentic workers:** REQUIRED SUB-SKILL: Use superpowers:subagent-driven-development (recommended) or superpowers:executing-plans to implement this plan task-by-task. Steps use checkbox (`- [ ]`) syntax for tracking.
**Goal:** Implement parallel foreach incrementally without breaking serial workflows or duplicating state-write logic. **Goal:** Implement concurrent foreach incrementally without breaking serial workflows or duplicating state-write logic.
**Architecture:** The work is split into four independently shippable layers: policy models, state patch extraction, barrier runtime state, and async parallel execution. Each layer preserves current serial behavior and adds tests before implementation. `foreach(mode="parallel")` remains unsupported until the final layer. **Architecture:** The work is split into four independently shippable layers: policy models, state patch extraction, barrier runtime state, and concurrent execution. Each layer preserves current serial behavior and adds tests before implementation. `foreach(mode="concurrent")` remains unsupported until the final layer. Sync runtime should support deterministic interleaving once the mode is enabled; async runtime can additionally run admitted async node handlers simultaneously.
**Tech Stack:** Python 3.14, Pydantic v2, dataclasses, pytest, basedpyright, ruff, existing `wf_core` scheduler/runtime modules. **Tech Stack:** Python 3.14, Pydantic v2, dataclasses, pytest, basedpyright, ruff, existing `wf_core` scheduler/runtime modules.
@@ -48,7 +48,7 @@ def test_serial_foreach_defaults_to_fail_item_policy() -> None:
assert node.mode == "serial" assert node.mode == "serial"
assert node.item_error.action == "fail" assert node.item_error.action == "fail"
assert node.item_error.collect_to is None assert node.item_error.collect_to is None
assert node.parallel is None assert node.concurrent is None
def test_collect_item_policy_requires_collect_to() -> None: def test_collect_item_policy_requires_collect_to() -> None:
@@ -64,20 +64,20 @@ def test_collect_item_policy_requires_collect_to() -> None:
) )
def test_parallel_policy_requires_parallel_mode() -> None: def test_concurrent_policy_requires_concurrent_mode() -> None:
with pytest.raises(ValidationError, match="parallel policy"): with pytest.raises(ValidationError, match="concurrent policy"):
ForeachNode.model_validate( ForeachNode.model_validate(
{ {
"id": "each", "id": "each",
"type": "foreach", "type": "foreach",
"over": {"root": "state", "parts": ["items"]}, "over": {"root": "state", "parts": ["items"]},
"as": "item", "as": "item",
"parallel": {"max_active": 4, "max_outstanding": 20}, "concurrent": {"max_active": 4, "max_outstanding": 20},
} }
) )
def test_parallel_policy_validates_capacity_order() -> None: def test_concurrent_policy_validates_capacity_order() -> None:
with pytest.raises(ValidationError, match="max_outstanding"): with pytest.raises(ValidationError, match="max_outstanding"):
ForeachNode.model_validate( ForeachNode.model_validate(
{ {
@@ -85,8 +85,8 @@ def test_parallel_policy_validates_capacity_order() -> None:
"type": "foreach", "type": "foreach",
"over": {"root": "state", "parts": ["items"]}, "over": {"root": "state", "parts": ["items"]},
"as": "item", "as": "item",
"mode": "parallel", "mode": "concurrent",
"parallel": {"max_active": 10, "max_outstanding": 4}, "concurrent": {"max_active": 10, "max_outstanding": 4},
} }
) )
``` ```
@@ -116,8 +116,8 @@ class ForeachItemErrorPolicy(BaseModel):
return self return self
class ForeachParallelPolicy(BaseModel): class ForeachConcurrentPolicy(BaseModel):
"""Concurrency policy for async parallel foreach execution.""" """Concurrency policy for foreach frame admission."""
model_config = ConfigDict(extra="forbid") model_config = ConfigDict(extra="forbid")
@@ -136,7 +136,7 @@ Update `ForeachNode`:
```python ```python
item_error: ForeachItemErrorPolicy = Field(default_factory=ForeachItemErrorPolicy) item_error: ForeachItemErrorPolicy = Field(default_factory=ForeachItemErrorPolicy)
parallel: ForeachParallelPolicy | None = None concurrent: ForeachConcurrentPolicy | None = None
on_item_error: Literal["fail", "collect", "skip"] | None = Field( on_item_error: Literal["fail", "collect", "skip"] | None = Field(
default=None, default=None,
exclude=True, exclude=True,
@@ -149,10 +149,10 @@ Add a `model_validator(mode="before")` that converts old `on_item_error` into `i
Add a `model_validator(mode="after")` that enforces: Add a `model_validator(mode="after")` that enforces:
```python ```python
if self.mode == "parallel" and self.parallel is None: if self.mode == "concurrent" and self.concurrent is None:
raise ValueError("parallel foreach requires parallel policy") raise ValueError("concurrent foreach requires concurrent policy")
if self.mode == "serial" and self.parallel is not None: if self.mode == "serial" and self.concurrent is not None:
raise ValueError("parallel policy is only valid when mode='parallel'") raise ValueError("concurrent policy is only valid when mode='concurrent'")
``` ```
- [ ] **Step 3: Update derived outcomes** - [ ] **Step 3: Update derived outcomes**
@@ -185,14 +185,14 @@ In `src/wf_core/runtime/ops/foreach.py`, keep:
```python ```python
if step.mode != "serial": if step.mode != "serial":
raise WorkflowExecutionError("parallel foreach execution is not implemented yet") raise WorkflowExecutionError("concurrent foreach execution is not implemented yet")
``` ```
Add a comment: Add a comment:
```python ```python
# Policy models are accepted before execution support so saved workflows can # Policy models are accepted before execution support so saved workflows can
# validate shape, but runtime must reject parallel until barrier commits exist. # validate shape, but runtime must reject concurrent until barrier commits exist.
``` ```
- [ ] **Step 6: Verify phase** - [ ] **Step 6: Verify phase**
@@ -316,7 +316,7 @@ Expected: all pass; full suite should still pass before moving on.
## Phase 3: Foreach Barrier Runtime State ## Phase 3: Foreach Barrier Runtime State
**Goal:** Add resumable barrier metadata and pending result structures without enabling async parallel execution. **Goal:** Add resumable barrier metadata and pending result structures without enabling concurrent execution.
**Files:** **Files:**
- Modify: `src/wf_core/runtime/scheduler.py` - Modify: `src/wf_core/runtime/scheduler.py`
@@ -422,9 +422,9 @@ Expected: pass.
--- ---
## Phase 4: Async Parallel Foreach ## Phase 4: Concurrent Foreach Execution
**Goal:** Enable `foreach(mode="parallel")` in async execution only, using policy limits, pending results, and barrier commits. **Goal:** Enable `foreach(mode="concurrent")` using policy limits, pending results, and barrier commits. Sync runtime interleaves admitted item frames one node call at a time; async runtime may run admitted async node handler calls simultaneously.
**Files:** **Files:**
- Modify: `src/wf_core/runtime/ops/foreach.py` - Modify: `src/wf_core/runtime/ops/foreach.py`
@@ -432,22 +432,22 @@ Expected: pass.
- Modify: `src/wf_core/runtime/engine.py` - Modify: `src/wf_core/runtime/engine.py`
- Modify: `src/wf_core/runtime/ops/nodes.py` - Modify: `src/wf_core/runtime/ops/nodes.py`
- Modify: `src/wf_core/runtime/foreach_state.py` - Modify: `src/wf_core/runtime/foreach_state.py`
- Test: `tests/core/test_parallel_foreach.py` - Test: `tests/core/test_concurrent_foreach.py`
- [ ] **Step 1: Add async-only rejection tests** - [ ] **Step 1: Add sync interleaving and async execution tests**
Create `tests/core/test_parallel_foreach.py` with: Create `tests/core/test_concurrent_foreach.py` with:
```python ```python
def test_sync_runtime_rejects_parallel_foreach() -> None: def test_sync_runtime_interleaves_concurrent_foreach() -> None:
... ...
async def test_async_runtime_accepts_parallel_foreach() -> None: async def test_async_runtime_accepts_concurrent_foreach() -> None:
... ...
``` ```
Expected before implementation: async test fails because runtime still rejects parallel. Expected before implementation: both tests fail because runtime still rejects concurrent mode.
- [ ] **Step 2: Add capacity tests** - [ ] **Step 2: Add capacity tests**
@@ -456,7 +456,7 @@ Use async node handlers that record start/completion order and block on `asyncio
Test: Test:
```python ```python
async def test_parallel_foreach_respects_max_active() -> None: async def test_concurrent_foreach_respects_max_active() -> None:
... ...
assert max_seen_active == 2 assert max_seen_active == 2
``` ```
@@ -481,11 +481,11 @@ This may require a small test-only node that blocks through the runtime-supporte
Tests: Tests:
```python ```python
async def test_parallel_collect_writes_ordered_errors_and_emits_completed_with_errors() -> None: async def test_concurrent_collect_writes_ordered_errors_and_emits_completed_with_errors() -> None:
... ...
async def test_parallel_skip_emits_completed_with_errors_without_hidden_state() -> None: async def test_concurrent_skip_emits_completed_with_errors_without_hidden_state() -> None:
... ...
``` ```
@@ -502,17 +502,17 @@ Use nodes that complete out of order but write list-like results.
Assert committed state is ordered by item index, not completion order. Assert committed state is ordered by item index, not completion order.
- [ ] **Step 6: Implement async parallel child scheduling** - [ ] **Step 6: Implement concurrent child scheduling**
In `step_foreach`, branch by mode: In `step_foreach`, branch by mode:
```python ```python
if step.mode == "serial": if step.mode == "serial":
return step_foreach_serial(...) return step_foreach_serial(...)
return step_foreach_parallel(...) return step_foreach_concurrent(...)
``` ```
`step_foreach_parallel` should: `step_foreach_concurrent` should:
- inspect `ForeachBarrierState` - inspect `ForeachBarrierState`
- start children while `active < max_active` and `outstanding < max_outstanding` - start children while `active < max_active` and `outstanding < max_outstanding`
- block parent when waiting for children - block parent when waiting for children
@@ -530,14 +530,14 @@ class RuntimeLimits:
max_active_node_calls: int = 16 max_active_node_calls: int = 16
``` ```
If this is too large for the first async parallel pass, leave global node-call budget as follow-up and rely on foreach `max_active`. If this is too large for the first concurrent pass, leave global node-call budget as follow-up and rely on foreach `max_active`.
- [ ] **Step 8: Verify phase** - [ ] **Step 8: Verify phase**
Run: Run:
```bash ```bash
uv run pytest tests/core/test_parallel_foreach.py tests/authoring/test_demo_workflow.py -q uv run pytest tests/core/test_concurrent_foreach.py tests/authoring/test_demo_workflow.py -q
uv run pytest -q uv run pytest -q
uvx ruff check src tests uvx ruff check src tests
uv run basedpyright --level error uv run basedpyright --level error
@@ -554,12 +554,12 @@ Ship these as separate commits/PRs:
1. Phase 1: policy shape and validation 1. Phase 1: policy shape and validation
2. Phase 2: patch extraction with no behavior change 2. Phase 2: patch extraction with no behavior change
3. Phase 3: barrier metadata with serial behavior unchanged 3. Phase 3: barrier metadata with serial behavior unchanged
4. Phase 4: async parallel execution 4. Phase 4: concurrent execution
Do not start Phase 4 until Phase 2 and Phase 3 are stable. Parallel foreach depends on patch extraction and resumable barrier state. Do not start Phase 4 until Phase 2 and Phase 3 are stable. Concurrent foreach depends on patch extraction and resumable barrier state.
## Self-Review ## Self-Review
- Spec coverage: ADR 0002 decisions are represented across the four phases. - Spec coverage: ADR 0002 decisions are represented across the four phases.
- Intentional gaps: explicit Fork/Gather, lineage-token graph nodes, OpenTelemetry, platform source/tool caps, and full run persistence are not included. - Intentional gaps: explicit Fork/Gather, lineage-token graph nodes, OpenTelemetry, platform source/tool caps, and full run persistence are not included.
- Risk control: phases 1-3 preserve serial behavior and keep `mode="parallel"` unsupported until phase 4. - Risk control: phases 1-3 preserve serial behavior and keep `mode="concurrent"` unsupported until phase 4.
@@ -2,7 +2,7 @@
> **For agentic workers:** REQUIRED SUB-SKILL: Use superpowers:subagent-driven-development (recommended) or superpowers:executing-plans to implement this plan task-by-task. Steps use checkbox (`- [ ]`) syntax for tracking. > **For agentic workers:** REQUIRED SUB-SKILL: Use superpowers:subagent-driven-development (recommended) or superpowers:executing-plans to implement this plan task-by-task. Steps use checkbox (`- [ ]`) syntax for tracking.
**Goal:** Add the internal scheduler foundation needed for future parallel foreach and native subgraphs while preserving current serial workflow behavior. **Goal:** Add the internal scheduler foundation needed for future concurrent foreach and native subgraphs while preserving current serial workflow behavior.
**Architecture:** `RunState.frames` remains the source of frame lifecycle state, and a new serialized `ready_frame_ids` queue defines deterministic scheduling order. A new internal `wf_core.runtime.scheduler` module owns frame creation, enqueue/select, block/wake, and no-ready-frame resolution. Existing sync and async engines both use scheduler selection before `prepare_step`; node execution stays split between sync and async paths. **Architecture:** `RunState.frames` remains the source of frame lifecycle state, and a new serialized `ready_frame_ids` queue defines deterministic scheduling order. A new internal `wf_core.runtime.scheduler` module owns frame creation, enqueue/select, block/wake, and no-ready-frame resolution. Existing sync and async engines both use scheduler selection before `prepare_step`; node execution stays split between sync and async paths.
@@ -734,7 +734,7 @@ Expected: ruff passes and basedpyright reports 0 errors.
Check: Check:
```bash ```bash
git diff -- CONTEXT.md docs/adr/0001-scheduler-foundation-before-parallel-foreach.md docs/current_roadmap.md git diff -- CONTEXT.md docs/adr/0001-scheduler-foundation-before-concurrent-foreach.md docs/current_roadmap.md
``` ```
Expected: docs remain aligned with implemented first-pass behavior. Expected: docs remain aligned with implemented first-pass behavior.
@@ -744,5 +744,5 @@ Expected: docs remain aligned with implemented first-pass behavior.
## Self-Review ## Self-Review
- Spec coverage: ADR decisions are covered by tasks for `BLOCKED`, ready queue, scheduler helpers, sync/async engine migration, serial foreach block/wake, interrupt resume priority, and verification. - Spec coverage: ADR decisions are covered by tasks for `BLOCKED`, ready queue, scheduler helpers, sync/async engine migration, serial foreach block/wake, interrupt resume priority, and verification.
- Intentional gaps: no `foreach(mode="parallel")`, no `ParallelForeachPolicy`, no lineage patches, no BarrierNode, and no public scheduler exports. - Intentional gaps: no `foreach(mode="concurrent")`, no `ForeachConcurrentPolicy`, no lineage patches, no BarrierNode, and no public scheduler exports.
- Type consistency: helper names are stable across tasks: `add_frame`, `enqueue_frame`, `select_next_frame`, `mark_frame_pending`, `block_frame_on_children`, `wake_frame`, `wake_parent_if_children_complete`, and `resolve_no_ready_frames`. - Type consistency: helper names are stable across tasks: `add_frame`, `enqueue_frame`, `select_next_frame`, `mark_frame_pending`, `block_frame_on_children`, `wake_frame`, `wake_parent_if_children_complete`, and `resolve_no_ready_frames`.
+8 -6
View File
@@ -77,8 +77,8 @@ enqueues the child. When the child reaches `END`, `wake_parent_if_children_compl
wakes the blocked parent so it can create the next iteration or emit `done`. wakes the blocked parent so it can create the next iteration or emit `done`.
This preserves current serial behavior while making the hidden parent/child This preserves current serial behavior while making the hidden parent/child
relationship explicit. `foreach(mode="parallel")` is still unsupported because relationship explicit. `foreach(mode="concurrent")` is still unsupported because
parallel execution needs policy, barrier, lineage, and state-patch semantics concurrent execution needs policy, barrier, lineage, and state-patch semantics
that are not implemented yet. that are not implemented yet.
## Validation Flow ## Validation Flow
@@ -118,7 +118,7 @@ limits and intended adapter seam.
state patch commits. The remaining mapping design notes for future reducer state patch commits. The remaining mapping design notes for future reducer
metadata are documented in metadata are documented in
[`core_state_mapping_and_merge.md`](core_state_mapping_and_merge.md). [`core_state_mapping_and_merge.md`](core_state_mapping_and_merge.md).
- Foreach is still serial-only. The scheduler foundation exists, but parallel - Foreach is still serial-only. The scheduler foundation exists, but concurrent
foreach still needs explicit policy, implicit barrier state, lineage-aware foreach still needs explicit policy, implicit barrier state, lineage-aware
patch commits, and quiescent interrupt handling. `ForeachNode.over` is typed patch commits, and quiescent interrupt handling. `ForeachNode.over` is typed
as a `GraphSourcePath`, but execution is still serial. as a `GraphSourcePath`, but execution is still serial.
@@ -137,9 +137,11 @@ limits and intended adapter seam.
upgrade: nested run state, child-frame trace preservation, interrupt bubbling upgrade: nested run state, child-frame trace preservation, interrupt bubbling
with path metadata, and resume back into the child workflow. with path metadata, and resume back into the child workflow.
- Frames are no longer only a serial execution stack: the runtime has a ready - Frames are no longer only a serial execution stack: the runtime has a ready
queue and `BLOCKED` frame state. Async parallel foreach and native subgraphs queue and `BLOCKED` frame state. Concurrent foreach and native subgraphs still
still need more work: lineage isolation, barrier merge semantics, pending need more work: lineage isolation, barrier merge semantics, pending child
child results, and explicit child workflow/deployment identity. results, and explicit child workflow/deployment identity. Async runtime can
later add simultaneous async node handler execution, but the workflow mode is
still concurrent foreach.
- Runtime errors are still ordinary exceptions plus failed run status. A richer - Runtime errors are still ordinary exceptions plus failed run status. A richer
error payload can be added later, but should be designed as part of trace/run error payload can be added later, but should be designed as part of trace/run
state rather than scattered exceptions. state rather than scattered exceptions.