fmt, fix

docs/superpowers/plans/2026-05-22-concurrent-foreach-barrier-write-semantics.md

@@ -333,6 +333,8 @@ Add:
 def validate_barrier_writes(
     item_patches: Sequence[StatePatch],
     state_fields: Mapping[StatePath, StateFieldDecl],
+    *,
+    reducers: Mapping[str, ReducerDefinition] | None = None,
 ) -> None:
     """Reject ambiguous sibling writes before replaying a foreach barrier.
 
@@ -369,7 +371,7 @@ In `build_barrier_patch(...)`, after:
 add:
 
 ```python
-    validate_barrier_writes(item_patches, state_fields)
+    validate_barrier_writes(item_patches, state_fields, reducers=reducers)
 ```
 
 - [ ] **Step 7: Verify focused unit tests**

docs/superpowers/plans/2026-05-22-concurrent-foreach-item-overlays.md

@@ -135,7 +135,11 @@ def test_sync_concurrent_foreach_item_reads_own_buffered_write() -> None:
     assert run.state["seen"] == ["scratch:a", "scratch:b", "scratch:c"]
 ```
 
-Important detail: update the existing `record` `NodeUse` in `_workflow(...)` or inside this test so it writes `scratch` to `state.scratch` for this workflow. If `_workflow(...)` is too fixed for that, create a small dedicated helper for this test instead of making `_workflow(...)` harder to read.
+Important detail: do not mutate the generic `_workflow(...)` helper for this test.
+Create a dedicated helper such as `_multi_step_overlay_workflow()` whose `record`
+`NodeUse` writes `scratch` to `state.scratch`. The test is specifically about an
+item-local write followed by an item-local read, so the workflow shape should be
+self-contained and obvious.
 
 - [ ] **Step 2: Add a sibling isolation test**
 
@@ -498,7 +502,8 @@ def test_sync_concurrent_foreach_allows_multi_step_item_body_with_overlay() -> N
     # Assert the workflow completes and output contains all expected values.
 ```
 
-Prefer not duplicating the full workflow; extract a helper:
+Prefer not duplicating the full workflow; reuse the dedicated helper introduced
+for the overlay read/write tests:
 
 ```python
 def _multi_step_overlay_workflow() -> Workflow:

docs/superpowers/plans/2026-05-22-concurrent-foreach-roadmap.md

@@ -4,7 +4,7 @@
 
 **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 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.
+**Architecture:** The work was 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. Sync runtime supports deterministic interleaving for concurrent foreach; 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.
 
@@ -19,10 +19,10 @@
 - Phase 3 is implemented: `wf_core.runtime.foreach_state` owns typed barrier
   metadata and serial foreach progress now uses that metadata instead of ad hoc
   `foreach_progress`.
-- Phase 4 is not implemented: `foreach(mode="concurrent")` still validates as a
+- Phase 4 is implemented: `foreach(mode="concurrent")` supports sync
-  model shape but runtime execution rejects it until concurrent scheduling,
+  interleaving, async item-node batching, barrier commits, item error policies,
-  barrier commits, and item failure handling are implemented.
+  and quiescent interrupt handling.
-- Phase 4 is expanded into a dedicated roadmap:
+- Phase 4 details live in the dedicated roadmap:
   [`2026-05-22-concurrent-foreach-phase4-roadmap.md`](2026-05-22-concurrent-foreach-phase4-roadmap.md).
   Start with
   [`2026-05-22-concurrent-foreach-v1-sync-fail-only.md`](2026-05-22-concurrent-foreach-v1-sync-fail-only.md).
@@ -466,7 +466,8 @@ async def test_async_runtime_accepts_concurrent_foreach() -> None:
     ...
 ```
 
-Expected before implementation: both tests fail because runtime still rejects concurrent mode.
+Historical expectation before Phase 4: both tests failed because runtime rejected
+concurrent mode. Current implementation status: these tests should pass.
 
 - [ ] **Step 2: Add capacity tests**
 
@@ -575,10 +576,10 @@ Ship these as separate commits/PRs:
 3. Phase 3: barrier metadata with serial behavior unchanged
 4. Phase 4: concurrent execution
 
-Do not start Phase 4 until Phase 2 and Phase 3 are stable. Concurrent foreach depends on patch extraction and resumable barrier state.
+Phase 4 depended on Phase 2 and Phase 3 because concurrent foreach needs patch extraction and resumable barrier state. Keep future fork/gather or native subgraph work layered on top of those runtime primitives instead of replacing them.
 
 ## 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="concurrent"` unsupported until phase 4.
+- Risk control: phases 1-3 preserved serial behavior until phase 4 enabled `mode="concurrent"`.

docs/wf_core_architecture.md

@@ -118,10 +118,11 @@ limits and intended adapter seam.
   state patch commits. The remaining mapping design notes for future reducer
   metadata are documented in
   [`core_state_mapping_and_merge.md`](core_state_mapping_and_merge.md).
-- Foreach is still serial-only. The scheduler foundation exists, but concurrent
+- Foreach supports serial and concurrent execution. Concurrent foreach uses
-  foreach still needs explicit policy, implicit barrier state, lineage-aware
+  explicit policy, typed barrier state, lineage-aware patch commits, item error
-  patch commits, and quiescent interrupt handling. `ForeachNode.over` is typed
+  policy, and quiescent interrupt handling. Remaining gaps are higher-level
-  as a `GraphSourcePath`, but execution is still serial.
+  graph constructs such as native subgraphs, explicit fork/gather nodes, and
+  advanced conflict strategies beyond exact-path mergeable reducers.
 - Interrupt lifecycle is still node-level and run-state-level. Long-lived
   external subscriptions or notification streams need a separate lifecycle
   design. Interrupt `request` and `resume` are canonical binding lists; nested

src/wf_authoring/builder/core.py

@@ -400,9 +400,9 @@ class WorkflowBuilder:
     ) -> ForeachNode:
         """Add a foreach step.
 
-        Concurrent mode is intentionally model-only for now: it validates saved
+        Concurrent mode is supported by the runtime with deterministic barrier
-        shape, but runtime execution still rejects it until barrier commits are
+        commits, item error policies, and async item-node batching. See ADR 0002
-        implemented.
+        for the exact merge and interrupt semantics.
         """
         node = ForeachNode.model_validate(
             {

src/wf_core/runtime/engine.py

@@ -77,8 +77,7 @@ def resume_workflow(
         return run
 
     while True:
-        frame = select_next_frame(run)
+        if select_next_frame(run) is None:
-        if frame is None:
             status = resolve_no_ready_frames(run)
             if status == RunStatus.COMPLETED:
                 break
@@ -119,8 +118,7 @@ async def resume_workflow_async(
         return run
 
     while True:
-        frame = select_next_frame(run)
+        if select_next_frame(run) is None:
-        if frame is None:
             status = resolve_no_ready_frames(run)
             if status == RunStatus.COMPLETED:
                 break

src/wf_core/runtime/step.py

@@ -241,17 +241,19 @@ async def _step_async_foreach_item_batch(
     deterministically.
     """
     frames = [first_frame, *_claim_matching_async_item_frames(run, index, first_frame)]
-    tasks = [
+    tasks = []
-        invoke_node_use_async_for_frame(
+    for frame in frames:
-            workflow,
+        node = _node_use_for_frame(index, frame)
-            run,
+        tasks.append(
-            frame,
+            invoke_node_use_async_for_frame(
-            _node_use_for_frame(index, frame),
+                workflow,
-            index.node_defs[_node_use_for_frame(index, frame).node],
+                run,
-            registry,
+                frame,
+                node,
+                index.node_defs[node.node],
+                registry,
+            )
         )
-        for frame in frames
-    ]
     results = await asyncio.gather(*tasks, return_exceptions=True)
     for frame, result in zip(frames, results, strict=True):
         run.current_frame_id = frame.id

src/wf_core/validation/steps.py

@@ -161,6 +161,11 @@ def validate_foreach_node(
         return
     collect_to = node.item_error.collect_to
     if collect_to is None:
+        report.add(
+            ValidationIssueCode.INVALID_FOREACH_COLLECT_DESTINATION,
+            f"nodes[{index}].item_error.collect_to",
+            "collect_to is required when item_error.action is 'collect'",
+        )
         return
     destination_root = _state_destination_root(collect_to)
     state_fields = workflow.state_schema.field_index()

tests/authoring/test_path_inputs.py

@@ -16,13 +16,19 @@ def test_single_string_path_input_uses_toml_dotted_key_syntax() -> None:
     assert coerce_state_path("person.name") == StatePath(("person", "name"))
     assert coerce_state_path('"person.name"') == StatePath(("person.name",))
     assert coerce_state_path('person."three and four"') == StatePath(
-        ("person", "three and four")
+        (
+            "person",
+            "three and four",
+        )
     )
 
 
 def test_vararg_path_input_treats_parts_as_literal_segments() -> None:
     assert coerce_state_path("person.name", "email address") == StatePath(
-        ("person.name", "email address")
+        (
+            "person.name",
+            "email address",
+        )
     )
 
 

tests/core/test_concurrent_foreach.py

@@ -200,7 +200,11 @@ def _workflow(
     if include_completed_with_errors:
         edges.append(
             Edge.model_validate(
-                {"from": "each", "outcome": "completed_with_errors", "to": END}
+                {
+                    "from": "each",
+                    "outcome": "completed_with_errors",
+                    "to": END,
+                }
             )
         )
     return Workflow(
@@ -338,10 +342,18 @@ def _multi_step_overlay_workflow() -> Workflow:
         ],
         edges=[
             Edge.model_validate(
-                {"from": "each", "outcome": "loop", "to": "stage_scratch"}
+                {
+                    "from": "each",
+                    "outcome": "loop",
+                    "to": "stage_scratch",
+                }
             ),
             Edge.model_validate(
-                {"from": "stage_scratch", "outcome": "ok", "to": "read_scratch"}
+                {
+                    "from": "stage_scratch",
+                    "outcome": "ok",
+                    "to": "read_scratch",
+                }
             ),
             Edge.model_validate({"from": "read_scratch", "outcome": "ok", "to": END}),
             Edge.model_validate({"from": "each", "outcome": "done", "to": END}),
@@ -404,7 +416,11 @@ def _same_path_replace_workflow() -> Workflow:
         ],
         edges=[
             Edge.model_validate(
-                {"from": "each", "outcome": "loop", "to": "write_winner"}
+                {
+                    "from": "each",
+                    "outcome": "loop",
+                    "to": "write_winner",
+                }
             ),
             Edge.model_validate({"from": "write_winner", "outcome": "ok", "to": END}),
             Edge.model_validate({"from": "each", "outcome": "done", "to": END}),

tests/core/test_concurrent_foreach_errors.py

@@ -147,7 +147,11 @@ def _workflow(*, item_error: dict[str, object]) -> Workflow:
             Edge.model_validate({"from": "record", "outcome": "ok", "to": END}),
             Edge.model_validate({"from": "each", "outcome": "done", "to": END}),
             Edge.model_validate(
-                {"from": "each", "outcome": "completed_with_errors", "to": END}
+                {
+                    "from": "each",
+                    "outcome": "completed_with_errors",
+                    "to": END,
+                }
             ),
         ],
     )

tests/core/test_concurrent_foreach_interrupts.py

@@ -142,7 +142,11 @@ def _workflow() -> Workflow:
             Edge.model_validate({"from": "each", "outcome": "loop", "to": "route"}),
             Edge.model_validate({"from": "route", "outcome": "ok", "to": END}),
             Edge.model_validate(
-                {"from": "route", "outcome": "needs_input", "to": "ask"}
+                {
+                    "from": "route",
+                    "outcome": "needs_input",
+                    "to": "ask",
+                }
             ),
             Edge.model_validate({"from": "ask", "outcome": "submitted", "to": END}),
             Edge.model_validate({"from": "each", "outcome": "done", "to": END}),

tests/core/test_foreach_barrier_state.py

@@ -142,7 +142,11 @@ def test_pending_item_result_reports_missing_required_field() -> None:
 def test_pending_item_result_rejects_negative_index() -> None:
     with pytest.raises(WorkflowExecutionError, match="index"):
         PendingItemResult.from_metadata(
-            {"index": -1, "frame_id": "child", "status": "succeeded"}
+            {
+                "index": -1,
+                "frame_id": "child",
+                "status": "succeeded",
+            }
         )