lda.chat/lda-wf
2
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

docs

Browse Source
This commit is contained in:
lda
2026-05-22 04:40:15 +07:00 Verified
parent 4fa0beab8c
commit 9cd064d6ef
7 changed files with 906 additions and 21 deletions
Download Patch File Download Diff File Expand all files Collapse all files
docs/adr/0001-scheduler-foundation-before-parallel-foreach.md
+2
View File
@@ -81,6 +81,8 @@ they are enabled:
- A future foreach policy must define batching, `max_concurrency`, item failure
handling, and cancellation/drain behavior.
- Parallel foreach should be bounded by default so workflows do not spawn
unbounded MCP, HTTP, browser, or external service calls.
- A future barrier must wait for multiple child or upstream frames and merge
their lineage patches.
- Missing reducers mean replace only within one serial lineage. At a barrier,
docs/adr/0002-parallel-foreach-policy-and-barrier-commits.md
+161
View File
@@ -0,0 +1,161 @@
# Parallel Foreach Policy and Barrier Commits
Status: accepted
Parallel foreach will use bounded scheduling, lineage-local pending state, and
barrier-buffered commits. This keeps parallel item execution deterministic,
resumable, and safe around reducers, interrupts, and external tool calls.
## Context
The scheduler foundation makes multiple runnable frames possible, but parallel
foreach needs more than concurrent node calls. Each item lineage may execute
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
state commits need deterministic merge behavior.
## Decision
Parallel foreach will be async-runtime-only and bounded by policy.
The future foreach model should separate item error behavior from concurrency
behavior:
- `item_error` is foreach-wide and applies to serial and parallel foreach.
- `parallel` is a nested policy object used only when `mode="parallel"`.
- `parallel.max_active` defaults to `4`.
- `parallel.max_outstanding` defaults to `20`.
- Validation requires `max_outstanding >= max_active`.
- Ready or running item frames consume active capacity.
- Blocked item frames consume outstanding capacity but not active capacity.
`foreach(mode="parallel")` requires async execution. Sync execution should reject
it clearly rather than inventing thread/process semantics.
## Item Error Policy
Item error policy handles runtime failures inside an item frame, not normal graph
control-flow outcomes. A node returning an `error` outcome is still ordinary
graph routing if the graph declares that edge.
Supported policy actions:
- `fail`: stop scheduling new items, drain already-started jobs to a quiescent
point, capture results for trace/observability, then fail the run.
- `skip`: mark the failed item frame failed, record no hidden state, continue,
and emit `completed_with_errors` if any item was skipped.
- `collect`: mark the failed item frame failed, collect a structured item error
record, continue, and emit `completed_with_errors` if any item was collected.
`collect` must declare an explicit state array destination. The foreach barrier
writes the full ordered error list once. On clean completion, `collect` writes an
empty list and emits `done`.
Collected error records include item index, frame id, failing node id, error
type/message, and the item value when it can be represented safely.
## Aggregate Outcomes
Foreach aggregate outcomes are policy-derived:
- `done`: all items completed cleanly.
- `completed_with_errors`: all items completed but one or more item failures
were handled by `skip` or `collect`.
Validation should require a routable `completed_with_errors` edge whenever the
policy can emit it. Users may route it to the same target as `done` explicitly.
Aggregate result vocabulary should be shared with future barrier-like graph
steps. `fail` is a policy action; if a future aggregate node emits graph-level
failure as control flow, the outcome should be `failed`.
## Barrier-Buffered Commits
Parallel foreach item writes are buffered as pending state patches/results, not
committed directly to `RunState.state`.
The state model is:
- `RunState.state` remains committed parent/global state.
- Each item lineage reads a visible state view built from committed state plus
its own lineage-local patch overlay.
- Sibling item patches are invisible to each other.
- Pending barrier results live in resumable run state/frame metadata, not trace.
- Trace `state_changes` means committed state changes only.
The barrier commits item results in deterministic item-index order by default.
Completion-order merging may exist later as an explicit barrier merge policy,
not reducer behavior.
Patch creation and commit must extract/reuse the existing node output
validation, output binding, and reducer logic. Parallel foreach must not create
a second write system.
## Merge and Reducer Rules
At a barrier, missing reducer means default replace only for single-writer
paths. Multiple sibling lineages writing the same state path require an explicit
reducer. Ancestor/descendant overlapping writes across lineages are conflicts
unless an explicit merge strategy covers them.
Reducers apply incrementally in deterministic lineage order. For foreach, that
means item index order.
## Interrupt and Failure Quiescence
Future parallel execution should not assume in-flight node calls can be safely
cancelled.
If an interrupt or fail policy trips while sibling jobs are already started, the
runtime should:
1. stop scheduling new work
2. let already-started jobs drain to a quiescent point
3. capture their results safely
4. defer or discard commits according to the policy boundary
5. return control only after no hidden background jobs continue mutating state
For `fail`, drained sibling results are for observability/cleanup only and
should not commit normal state progress after the failure boundary.
## Capacity and Runtime Limits
Foreach capacity is local correctness policy, not total process protection.
Future runtime should also support a basic global node-call budget in `wf_core`.
Source-, account-, or tool-specific limits belong in the platform layer.
Global node-call limits count admitted node handler calls, not all frames.
Control-flow frames are scheduler work; node calls are the expensive boundary.
Node calls waiting on platform/source/tool semaphores still count as active
node calls, and those waits keep the frame `RUNNING` rather than `BLOCKED`.
## Context and Lineage
Parallel item frames need lineage-local pending state. Later nodes in the same
item lineage can read earlier pending patches from that item, while siblings
cannot.
Patch overlays conceptually belong to lineage tokens, not execution frames. A
scoped foreach implementation may begin by storing them in foreach metadata, but
future Fork/Gather needs first-class lineage ownership.
Future foreach metadata should evolve into inherited structured lineage context:
- core runtime context is structural, not alias-first
- foreach structured context keys are foreach node ids
- `as_` remains authoring sugar and human-readable trace/docs context
- Python `RuntimeContext` should eventually expose typed context such as
`ctx.foreach["docs"].index`
- normal node authors should receive foreach values through mapped input;
inspecting runtime context is an advanced escape hatch
## Deferred Work
Explicit Fork/Gather is deferred. A future `GatherNode` should expose explicit
barrier semantics, but arbitrary graph convergence needs lineage tokens first.
Forks produce branch lineage tokens; gathers consume declared token sets and
produce merged tokens. This supports partial gathers such as merging `a+b`
before later merging with `c`.
Parallel foreach remains the nearer target because its implicit lineage tokens
are item indexes owned by one foreach activation.
docs/current_roadmap.md
+2
View File
@@ -41,6 +41,8 @@ implementation state.
- Scheduler foundation decision record:
[ADR 0001](./adr/0001-scheduler-foundation-before-parallel-foreach.md).
- Parallel foreach policy decision record:
[ADR 0002](./adr/0002-parallel-foreach-policy-and-barrier-commits.md).
- **Native subgraphs / graph-as-node**: add child run state, child trace
preservation, interrupt bubbling, and resume back into the child workflow.
Wrapper artifacts currently execute as deployments and return run status;
docs/superpowers/plans/2026-05-22-parallel-foreach-roadmap.md
+565
View File
@@ -0,0 +1,565 @@
# Parallel 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.
**Goal:** Implement parallel 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.
**Tech Stack:** Python 3.14, Pydantic v2, dataclasses, pytest, basedpyright, ruff, existing `wf_core` scheduler/runtime modules.
---
## Phase 1: Foreach Policy Models
**Goal:** Add the future policy shape while keeping runtime behavior serial-only.
**Files:**
- Modify: `src/wf_core/models/steps.py`
- Modify: `src/wf_core/validation/outcomes.py`
- Modify: `src/wf_core/validation/steps.py`
- Modify: `src/wf_authoring/builder/core.py`
- Test: `tests/core/test_foreach_policy.py`
- Test: `tests/authoring/test_builder.py`
- [ ] **Step 1: Add failing model tests**
Create `tests/core/test_foreach_policy.py`:
```python
from __future__ import annotations
import pytest
from pydantic import ValidationError
from wf_core.models.steps import ForeachNode
def test_serial_foreach_defaults_to_fail_item_policy() -> None:
node = ForeachNode.model_validate(
{
"id": "each",
"type": "foreach",
"over": {"root": "state", "parts": ["items"]},
"as": "item",
}
)
assert node.mode == "serial"
assert node.item_error.action == "fail"
assert node.item_error.collect_to is None
assert node.parallel is None
def test_collect_item_policy_requires_collect_to() -> None:
with pytest.raises(ValidationError, match="collect_to"):
ForeachNode.model_validate(
{
"id": "each",
"type": "foreach",
"over": {"root": "state", "parts": ["items"]},
"as": "item",
"item_error": {"action": "collect"},
}
)
def test_parallel_policy_requires_parallel_mode() -> None:
with pytest.raises(ValidationError, match="parallel policy"):
ForeachNode.model_validate(
{
"id": "each",
"type": "foreach",
"over": {"root": "state", "parts": ["items"]},
"as": "item",
"parallel": {"max_active": 4, "max_outstanding": 20},
}
)
def test_parallel_policy_validates_capacity_order() -> None:
with pytest.raises(ValidationError, match="max_outstanding"):
ForeachNode.model_validate(
{
"id": "each",
"type": "foreach",
"over": {"root": "state", "parts": ["items"]},
"as": "item",
"mode": "parallel",
"parallel": {"max_active": 10, "max_outstanding": 4},
}
)
```
- [ ] **Step 2: Add policy models**
In `src/wf_core/models/steps.py`, add:
```python
from typing import Self
class ForeachItemErrorPolicy(BaseModel):
"""Policy for runtime failures inside one foreach item lineage."""
model_config = ConfigDict(extra="forbid")
action: Literal["fail", "skip", "collect"] = "fail"
collect_to: StatePath | None = None
@model_validator(mode="after")
def _validate_collect_to(self) -> Self:
if self.action == "collect" and self.collect_to is None:
raise ValueError("collect item error policy requires collect_to")
if self.action != "collect" and self.collect_to is not None:
raise ValueError("collect_to is only valid when action='collect'")
return self
class ForeachParallelPolicy(BaseModel):
"""Concurrency policy for async parallel foreach execution."""
model_config = ConfigDict(extra="forbid")
max_active: int = Field(default=4, ge=1)
max_outstanding: int = Field(default=20, ge=1)
interrupt: Literal["quiesce"] = "quiesce"
@model_validator(mode="after")
def _validate_capacity(self) -> Self:
if self.max_outstanding < self.max_active:
raise ValueError("max_outstanding must be >= max_active")
return self
```
Update `ForeachNode`:
```python
item_error: ForeachItemErrorPolicy = Field(default_factory=ForeachItemErrorPolicy)
parallel: ForeachParallelPolicy | None = None
on_item_error: Literal["fail", "collect", "skip"] | None = Field(
default=None,
exclude=True,
description="Deprecated parse-only shorthand; use item_error.action.",
)
```
Add a `model_validator(mode="before")` that converts old `on_item_error` into `item_error.action`.
Add a `model_validator(mode="after")` that enforces:
```python
if self.mode == "parallel" and self.parallel is None:
raise ValueError("parallel foreach requires parallel policy")
if self.mode == "serial" and self.parallel is not None:
raise ValueError("parallel policy is only valid when mode='parallel'")
```
- [ ] **Step 3: Update derived outcomes**
In `src/wf_core/validation/outcomes.py`, update foreach outcome derivation:
```python
if step.type == "foreach":
outcomes = {"loop", "done"}
if step.item_error.action in {"skip", "collect"}:
outcomes.add("completed_with_errors")
return outcomes
```
- [ ] **Step 4: Validate collect destination schema**
In `src/wf_core/validation/steps.py`, when `node.item_error.action == "collect"`:
```python
destination_root = _state_destination_root(node.item_error.collect_to)
if destination_root is None or destination_root not in state_root_fields:
report.add(...)
```
Add a follow-up test that collect-to unknown state root reports a validation issue.
- [ ] **Step 5: Keep runtime unsupported**
In `src/wf_core/runtime/ops/foreach.py`, keep:
```python
if step.mode != "serial":
raise WorkflowExecutionError("parallel foreach execution is not implemented yet")
```
Add a comment:
```python
# Policy models are accepted before execution support so saved workflows can
# validate shape, but runtime must reject parallel until barrier commits exist.
```
- [ ] **Step 6: Verify phase**
Run:
```bash
uv run pytest tests/core/test_foreach_policy.py tests/authoring/test_builder.py -q
uvx ruff check src tests
uv run basedpyright --level error
```
Expected: all pass.
---
## Phase 2: State Patch Extraction
**Goal:** Split current node output writes into reusable “build patch” and “commit patch” operations without changing current serial behavior.
**Files:**
- Modify: `src/wf_core/runtime/ops/state.py`
- Modify: `src/wf_core/runtime/ops/nodes.py`
- Test: `tests/core/test_atomic_state_patches.py`
- Test: `tests/core/test_nested_state_paths.py`
- [ ] **Step 1: Add state patch model**
In `src/wf_core/runtime/ops/state.py`, add:
```python
from dataclasses import dataclass, field
from typing import Any
@dataclass(slots=True)
class StatePatch:
"""Validated state writes produced by one step before commit."""
changes: dict[str, Any] = field(default_factory=dict)
```
- [ ] **Step 2: Extract patch builder**
Refactor existing `apply_output_bindings(...)` into:
```python
def build_output_patch(
workflow: Workflow,
bindings: Sequence[OutputBinding],
output: Mapping[str, Any],
state: MutableMapping[str, Any],
*,
reducers: Mapping[str, ReducerDefinition] | None = None,
missing_field_message: str = "node output is missing required field {field}",
) -> StatePatch:
...
```
This function should:
- validate source paths
- validate destination paths
- calculate reducer-aware changes
- not mutate `state`
- [ ] **Step 3: Extract patch committer**
Add:
```python
def commit_state_patch(
state: MutableMapping[str, Any],
patch: StatePatch,
) -> dict[str, Any]:
"""Commit a validated patch to state and return committed changes."""
for path, value in patch.changes.items():
set_nested_value(state, split_state_path(path), value)
return dict(patch.changes)
```
Use the existing typed path helpers; do not reintroduce dotted-string parsing if a typed path helper exists.
- [ ] **Step 4: Preserve old API**
Keep `apply_output_bindings(...)` as a wrapper:
```python
patch = build_output_patch(...)
return commit_state_patch(state, patch)
```
Existing callers should keep working.
- [ ] **Step 5: Add equivalence tests**
Add tests that compare:
```python
old_changes = apply_output_bindings(...)
patch = build_output_patch(...)
new_changes = commit_state_patch(state2, patch)
assert old_changes["state.some_path"] == new_changes["state.some_path"]
assert state1["some_path"] == state2["some_path"]
```
Do not assert whole dict equality unless the test intentionally owns the full structure.
- [ ] **Step 6: Verify phase**
Run:
```bash
uv run pytest tests/core/test_atomic_state_patches.py tests/core/test_nested_state_paths.py tests/authoring/test_demo_workflow.py -q
uvx ruff check src tests
uv run basedpyright --level error
```
Expected: all pass; full suite should still pass before moving on.
---
## Phase 3: Foreach Barrier Runtime State
**Goal:** Add resumable barrier metadata and pending result structures without enabling async parallel execution.
**Files:**
- Modify: `src/wf_core/runtime/scheduler.py`
- Create: `src/wf_core/runtime/foreach_state.py`
- Modify: `src/wf_core/runtime/ops/foreach.py`
- Test: `tests/core/test_foreach_barrier_state.py`
- [ ] **Step 1: Add pending result dataclasses**
Create `src/wf_core/runtime/foreach_state.py`:
```python
from __future__ import annotations
from dataclasses import dataclass, field
from typing import Any, Literal
from wf_core.runtime.ops.state import StatePatch
@dataclass(slots=True)
class ItemErrorRecord:
"""Structured runtime failure record for one foreach item."""
index: int
frame_id: str
node_id: str
error_type: str
message: str
item: Any = None
@dataclass(slots=True)
class PendingItemResult:
"""Buffered item result waiting for foreach barrier commit."""
index: int
frame_id: str
status: Literal["succeeded", "failed"]
patch: StatePatch = field(default_factory=StatePatch)
error: ItemErrorRecord | None = None
```
- [ ] **Step 2: Add foreach barrier state**
In the same file:
```python
@dataclass(slots=True)
class ForeachBarrierState:
"""Resumable state owned by one foreach parent frame."""
next_index: int = 0
active_frame_ids: tuple[str, ...] = ()
outstanding_frame_ids: tuple[str, ...] = ()
pending_results: dict[int, PendingItemResult] = field(default_factory=dict)
```
Add `to_metadata()` / `from_frame()` helpers. Wrong frame kind returns `None`; malformed metadata for a foreach parent raises `WorkflowExecutionError`.
- [ ] **Step 3: Move serial progress into typed state**
Current serial foreach uses:
```python
progress_map = frame.metadata.setdefault("foreach_progress", {})
```
Replace with typed barrier state, but keep behavior equivalent:
```python
barrier = ForeachBarrierState.from_frame(frame) or ForeachBarrierState()
loop_index = barrier.next_index
barrier.next_index += 1
frame.metadata["foreach_barrier"] = barrier.to_metadata()
```
- [ ] **Step 4: Add serialization tests**
Test:
```python
def test_foreach_barrier_state_round_trips_through_frame_metadata() -> None:
...
```
Assert specific fields:
```python
assert loaded.next_index == 2
assert loaded.outstanding_frame_ids == ("child-1",)
```
- [ ] **Step 5: Keep serial behavior passing**
Run:
```bash
uv run pytest tests/core/test_foreach_barrier_state.py tests/authoring/test_demo_workflow.py -q
```
Expected: pass.
---
## Phase 4: Async Parallel Foreach
**Goal:** Enable `foreach(mode="parallel")` in async execution only, using policy limits, pending results, and barrier commits.
**Files:**
- Modify: `src/wf_core/runtime/ops/foreach.py`
- Modify: `src/wf_core/runtime/step.py`
- Modify: `src/wf_core/runtime/engine.py`
- Modify: `src/wf_core/runtime/ops/nodes.py`
- Modify: `src/wf_core/runtime/foreach_state.py`
- Test: `tests/core/test_parallel_foreach.py`
- [ ] **Step 1: Add async-only rejection tests**
Create `tests/core/test_parallel_foreach.py` with:
```python
def test_sync_runtime_rejects_parallel_foreach() -> None:
...
async def test_async_runtime_accepts_parallel_foreach() -> None:
...
```
Expected before implementation: async test fails because runtime still rejects parallel.
- [ ] **Step 2: Add capacity tests**
Use async node handlers that record start/completion order and block on `asyncio.Event`.
Test:
```python
async def test_parallel_foreach_respects_max_active() -> None:
...
assert max_seen_active == 2
```
Use `max_active=2`.
- [ ] **Step 3: Add outstanding tests**
Use a node that blocks internally through a future block helper or controlled async wait.
Test:
```python
async def test_blocked_items_count_against_max_outstanding_not_active() -> None:
...
```
This may require a small test-only node that blocks through the runtime-supported internal wait. If internal blocking is not implemented yet, defer this test to subgraph/internal-wait work and keep `max_outstanding` tested through queued children.
- [ ] **Step 4: Add collect/skip tests**
Tests:
```python
async def test_parallel_collect_writes_ordered_errors_and_emits_completed_with_errors() -> None:
...
async def test_parallel_skip_emits_completed_with_errors_without_hidden_state() -> None:
...
```
Assert:
```python
assert run.state["document_errors"][0]["index"] == 1
assert run.trace[-1].outcome == "completed_with_errors"
```
- [ ] **Step 5: Add barrier commit ordering test**
Use nodes that complete out of order but write list-like results.
Assert committed state is ordered by item index, not completion order.
- [ ] **Step 6: Implement async parallel child scheduling**
In `step_foreach`, branch by mode:
```python
if step.mode == "serial":
return step_foreach_serial(...)
return step_foreach_parallel(...)
```
`step_foreach_parallel` should:
- inspect `ForeachBarrierState`
- start children while `active < max_active` and `outstanding < max_outstanding`
- block parent when waiting for children
- finish when all items terminal
- commit barrier patches in item index order
- emit `done` or `completed_with_errors`
- [ ] **Step 7: Add async node-call budget seam**
Add execution option shape only if needed by implementation:
```python
@dataclass(slots=True)
class RuntimeLimits:
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`.
- [ ] **Step 8: Verify phase**
Run:
```bash
uv run pytest tests/core/test_parallel_foreach.py tests/authoring/test_demo_workflow.py -q
uv run pytest -q
uvx ruff check src tests
uv run basedpyright --level error
```
Expected: all pass.
---
## Implementation Order Recommendation
Ship these as separate commits/PRs:
1. Phase 1: policy shape and validation
2. Phase 2: patch extraction with no behavior change
3. Phase 3: barrier metadata with serial behavior unchanged
4. Phase 4: async parallel execution
Do not start Phase 4 until Phase 2 and Phase 3 are stable. Parallel foreach depends on patch extraction and resumable barrier state.
## Self-Review
- 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.
- Risk control: phases 1-3 preserve serial behavior and keep `mode="parallel"` unsupported until phase 4.
docs/superpowers/plans/2026-05-22-scheduler-foundation.md
+3 -3
View File
@@ -26,7 +26,7 @@
- Modify: `src/wf_core/runtime/ops/frames.py`
- Keep context helpers; demote stack-collapse usage or leave compatibility wrappers that call scheduler helpers.
- Modify: `src/wf_core/runtime/preparation.py`
- Stop relying on `collapse_completed_frames()` for selection.
- Stop relying on stack-style frame collapse for selection.
- Resume interrupt by waking/enqueueing the resumed frame.
- Modify: `src/wf_core/runtime/engine.py`
- Use scheduler loop for sync and async resume.
@@ -527,9 +527,9 @@ while True:
Use the same selection flow in the async loop before `await step_workflow_async(...)`.
- [ ] **Step 3: Remove stack collapse from step preparation**
- [ ] **Step 3: Remove stack-style frame collapse from step preparation**
In `prepare_step`, remove `collapse_completed_frames(run)` and keep it focused on resolving the selected frame’s node. It should still return `None` for interrupted/end states.
In `prepare_step`, remove old stack-style frame collapse and keep it focused on resolving the selected frame’s node. It should still return `None` for interrupted/end states.
- [ ] **Step 4: Verify serial workflows**