more examples of concurrent foreach

docs/wf_authoring_control_flow.md

@@ -233,6 +233,8 @@ See `examples/authoring_concurrent_foreach.py` for a runnable example covering:
 - sync concurrent foreach with `item_error={"action": "collect", ...}`;
 - async item-node batching with deterministic output order;
 - `item_error` as a string, mapping, or `ForeachItemErrorPolicy` object.
+- the replace-conflict case when sibling item writes target a non-mergeable
+  state path.
 
 ## Deprecated `route`
 

docs/wf_core_architecture.md

@@ -69,17 +69,22 @@ When the ready queue is empty, the scheduler classifies the run as completed,
 interrupted, failed, or deadlocked. This replaces the older assumption that
 `current_node_id == END` alone is enough to decide runtime completion.
 
-## Serial Foreach
+## Foreach
 
-Foreach remains serial-only. A serial foreach parent frame creates one iteration
+Serial foreach creates one iteration child frame, records typed
-child frame, records typed `ForeachIterationMetadata`, blocks on that child, and
+`ForeachIterationMetadata`, blocks on that child, and enqueues the child. When
-enqueues the child. When the child reaches `END`, `wake_parent_if_children_complete`
+the child reaches `END`, `wake_parent_if_children_complete` wakes the blocked
-wakes the blocked parent so it can create the next iteration or emit `done`.
+parent so it can create the next iteration or emit `done`.
 
-This preserves current serial behavior while making the hidden parent/child
+Concurrent foreach uses the same frame machinery but admits multiple item
-relationship explicit. `foreach(mode="concurrent")` is still unsupported because
+lineages according to `ForeachConcurrentPolicy`. Each item lineage reads through
-concurrent execution needs policy, barrier, lineage, and state-patch semantics
+its own overlay, successful item patches are buffered, and the parent foreach
-that are not implemented yet.
+commits the barrier in item-index order. Sibling writes to the same state path
+require a mergeable reducer on that exact path; ancestor/descendant sibling
+writes are rejected until an explicit deep merge policy exists.
+
+See `examples/raw_concurrent_foreach.py` for the canonical raw workflow shape and
+`examples/authoring_concurrent_foreach.py` for the authoring-layer shape.
 
 ## Validation Flow
 

examples/authoring_concurrent_foreach.py

@@ -20,6 +20,7 @@ from wf_core import (
     END,
     ForeachConcurrentPolicy,
     ForeachItemErrorPolicy,
+    WorkflowExecutionError,
     execute_workflow_async,
 )
 from wf_core.paths import StatePath
@@ -132,6 +133,47 @@ def run_collected_errors_example() -> RunState:
     return builder.execute({"items": ["a", "bad", "c"]})
 
 
+def run_replace_conflict_example() -> None:
+    """Demonstrate the exact-path reducer rule for sibling item writes.
+
+    Concurrent sibling writes to the same state path need a mergeable reducer.
+    `errors` has the default replace semantics, so writing every successful item
+    to that path is rejected at the barrier.
+    """
+    builder = WorkflowBuilder(
+        name="authoring_concurrent_foreach_replace_conflict",
+        input_schema=ItemsInput,
+        state_schema=ConcurrentForeachState,
+        output_schema=ConcurrentForeachOutput,
+    )
+    each = builder.foreach(
+        id="each",
+        over=state_path("items"),
+        as_="item",
+        mode="concurrent",
+        item_error="fail",
+        concurrent={"max_active": 2, "max_outstanding": 2},
+    )
+    record = builder.use(
+        record_item,
+        id="record",
+        input=[
+            input_from(context_path("item"), "value"),
+            input_from(context_path("item"), "seen"),
+        ],
+        output=[output_to("seen", state_path("errors"))],
+    )
+    builder.set_entry_point(each)
+    builder.connect(each, "loop", record)
+    builder.connect(record, "ok", END)
+    builder.connect(each, "done", END)
+    try:
+        builder.execute({"items": ["a", "b"]})
+    except WorkflowExecutionError:
+        return
+    raise AssertionError("expected same-path sibling replace writes to fail")
+
+
 async def run_async_ordered_example() -> RunState:
     """Run the async example; barrier commits still preserve item order."""
     builder = build_concurrent_foreach_workflow(record_item_async)

examples/raw_concurrent_foreach.py

@@ -0,0 +1,138 @@
+from __future__ import annotations
+
+from wf_core import END, RuntimeContext, Workflow, execute_workflow
+from wf_core.run_state import RunState
+
+
+def build_raw_concurrent_foreach_workflow() -> Workflow:
+    """Build the canonical JSON/Pydantic shape for concurrent foreach.
+
+    This example is intentionally raw `wf_core`: it is useful for MCP/LLM-facing
+    authoring surfaces that need to emit workflow JSON without Python builder
+    helpers.
+    """
+    return Workflow.model_validate(
+        {
+            "name": "raw_concurrent_foreach",
+            "input_schema": {
+                "type": "object",
+                "properties": {"items": {"type": "array"}},
+                "required": ["items"],
+            },
+            "state_schema": {
+                "type": "object",
+                "properties": {
+                    "items": {"type": "array"},
+                    "seen": {
+                        "type": "array",
+                        "reducer": "wf.std.append",
+                    },
+                    "errors": {"type": "array"},
+                },
+            },
+            "output_schema": {
+                "type": "object",
+                "properties": {
+                    "seen": {"type": "array"},
+                    "errors": {"type": "array"},
+                },
+            },
+            "node_defs": [
+                {
+                    "name": "record_item",
+                    "input_schema": {
+                        "type": "object",
+                        "properties": {
+                            "value": {},
+                            "seen": {},
+                        },
+                        "required": ["value", "seen"],
+                    },
+                    "output_schema": {
+                        "type": "object",
+                        "properties": {"seen": {}},
+                        "required": ["seen"],
+                    },
+                    "outcomes": ["ok"],
+                }
+            ],
+            "start": "each",
+            "nodes": [
+                {
+                    "id": "each",
+                    "type": "foreach",
+                    "over": {"root": "state", "parts": ["items"]},
+                    "as": "item",
+                    "mode": "concurrent",
+                    "concurrent": {
+                        "max_active": 2,
+                        "max_outstanding": 2,
+                    },
+                    "item_error": {
+                        "action": "collect",
+                        "collect_to": {"root": "state", "parts": ["errors"]},
+                    },
+                },
+                {
+                    "id": "record",
+                    "type": "node",
+                    "node": "record_item",
+                    "input": [
+                        {
+                            "target": {"root": "local", "parts": ["value"]},
+                            "path": {"root": "context", "parts": ["item"]},
+                        },
+                        {
+                            "target": {"root": "local", "parts": ["seen"]},
+                            "path": {"root": "context", "parts": ["item"]},
+                        },
+                    ],
+                    "output": [
+                        {
+                            "source": {"root": "local", "parts": ["seen"]},
+                            "target": {"root": "state", "parts": ["seen"]},
+                        }
+                    ],
+                },
+            ],
+            "edges": [
+                {"from": "each", "outcome": "loop", "to": "record"},
+                {"from": "record", "outcome": "ok", "to": END},
+                {"from": "each", "outcome": "done", "to": END},
+                {"from": "each", "outcome": "completed_with_errors", "to": END},
+            ],
+        }
+    )
+
+
+def build_raw_concurrent_foreach_registry():
+    """Return handlers for the raw concurrent foreach example."""
+
+    def record_item(
+        payload: dict[str, object],
+        _context: RuntimeContext,
+    ) -> dict[str, object]:
+        if payload["value"] == "bad":
+            raise ValueError("bad item")
+        return {"outcome": "ok", "output": {"seen": payload["seen"]}}
+
+    return {"record_item": record_item}
+
+
+def run_raw_concurrent_foreach_example() -> RunState:
+    """Run raw concurrent foreach with collected item errors."""
+    return execute_workflow(
+        build_raw_concurrent_foreach_workflow(),
+        {"items": ["a", "bad", "c"]},
+        build_raw_concurrent_foreach_registry(),
+    )
+
+
+def main() -> None:
+    """Run the example directly from the command line."""
+    run = run_raw_concurrent_foreach_example()
+    print(run.status.value, run.output)
+
+
+if __name__ == "__main__":
+    main()

tests/authoring/test_concurrent_foreach_examples.py

@@ -13,6 +13,7 @@ from examples.authoring_concurrent_foreach import (
     record_item,
     run_async_ordered_example,
     run_collected_errors_example,
+    run_replace_conflict_example,
 )
 from wf_authoring import WorkflowBuilder, state_path
 from wf_core import ForeachConcurrentPolicy
@@ -74,6 +75,10 @@ def test_authoring_foreach_accepts_concurrent_policy_object() -> None:
     assert foreach.model_dump(mode="json")["concurrent"]["max_outstanding"] == 3
 
 
+def test_authoring_concurrent_foreach_example_documents_replace_conflict() -> None:
+    run_replace_conflict_example()
+
+
 def test_authoring_foreach_deprecated_on_item_error_warns() -> None:
     builder = WorkflowBuilder(
         name="deprecated_item_error",

tests/core/test_raw_canonical_workflow_example.py

@@ -6,6 +6,10 @@ from examples.raw_canonical_workflow import (
     build_raw_canonical_workflow,
     run_raw_canonical_example,
 )
+from examples.raw_concurrent_foreach import (
+    build_raw_concurrent_foreach_workflow,
+    run_raw_concurrent_foreach_example,
+)
 
 
 def test_raw_canonical_workflow_runs() -> None:
@@ -36,3 +40,31 @@ def test_raw_canonical_workflow_serializes_new_shape() -> None:
     assert node["output"][0]["target"] == {"root": "state", "parts": ["message"]}
     assert message_schema["type"] == "string"
     assert message_schema["reducer"] == "wf.std.replace"
+
+
+def test_raw_concurrent_foreach_workflow_runs() -> None:
+    run = run_raw_concurrent_foreach_example()
+
+    assert run.status == RunStatus.COMPLETED
+    assert run.output["seen"] == ["a", "c"]
+    assert len(run.output["errors"]) == 1
+    error = run.output["errors"][0]
+    assert error["index"] == 1
+    assert error["node_id"] == "record"
+    assert error["error_type"] == "ValueError"
+    assert error["message"] == "bad item"
+
+
+def test_raw_concurrent_foreach_serializes_canonical_policy_shape() -> None:
+    workflow = build_raw_concurrent_foreach_workflow()
+    dumped = workflow.model_dump(mode="json")
+    foreach = dumped["nodes"][0]
+
+    assert foreach["mode"] == "concurrent"
+    assert foreach["concurrent"]["max_active"] == 2
+    assert foreach["item_error"]["action"] == "collect"
+    assert foreach["item_error"]["collect_to"] == {
+        "root": "state",
+        "parts": ["errors"],
+    }
+    assert "on_item_error" not in foreach