subgraph model & validation

src/wf_core/runtime/lineage.py

@@ -55,6 +55,9 @@ def lineage_writes_for_frame(
             )
         )
 
+    # Compatibility fallback: concurrent foreach used barrier-local patches
+    # before `RunState.lineages` became the primary write store. Keep reading
+    # those patches so old serialized runs and direct barrier tests still work.
     owner = item_frame_owner(frame)
     if owner is None:
         return ()
@@ -127,7 +130,11 @@ def lineage_patch(
     scope_id: str,
     lineage_id: str,
 ) -> StatePatch:
-    """Return a replayable patch for one lineage's pending writes."""
+    """Return a replayable patch for one lineage's pending writes.
+
+    Barrier/gather code should consume this instead of reconstructing a patch
+    from visible state. Incoming values are the replay source of truth.
+    """
     lineage = _lineage(run, scope_id=scope_id, lineage_id=lineage_id)
     return StatePatch(writes=list(lineage.writes))
 

src/wf_core/runtime/ops/foreach.py

@@ -340,6 +340,8 @@ def _finish_concurrent_foreach(
             raise WorkflowExecutionError(
                 "collect item error policy requires collect_to"
             )
+        # Collect-error records are generated by the barrier itself, not by an
+        # item lineage, so they still enter as a compatibility `changes` patch.
         item_patches.append(StatePatch(changes={str(collect_to): error_records}))
     combined = build_barrier_patch(
         workflow,

src/wf_core/runtime/ops/nodes.py

@@ -124,6 +124,8 @@ def _finalize_node_execution(
         if is_root_lineage_frame(frame):
             state_changes = commit_state_patch(run.state, patch)
         else:
+            # Non-root frames are future subgraph/fork branch execution: writes
+            # become lineage-local until an explicit boundary/barrier commits.
             append_lineage_writes(
                 run,
                 scope_id=frame.scope_id,
@@ -136,6 +138,8 @@ def _finalize_node_execution(
         parent_frame = run.frames[parent_frame_id]
         barrier = ForeachBarrierState.from_frame(parent_frame, foreach_node_id)
         if barrier is not None and barrier.mode == "concurrent":
+            # New concurrent foreach stores writes in the child lineage; the
+            # barrier keeps only result metadata plus old patch fallback.
             append_lineage_writes(
                 run,
                 scope_id=frame.scope_id,

src/wf_core/runtime/ops/state.py

@@ -49,7 +49,12 @@ class StatePatch:
     _staged_state: dict[str, Any] = dataclass_field(default_factory=dict, repr=False)
 
     def __post_init__(self) -> None:
-        """Keep legacy `StatePatch(changes=...)` usable during migration."""
+        """Keep legacy `StatePatch(changes=...)` usable during migration.
+
+        New runtime code should prefer ordered `writes`. `changes` stays as the
+        public trace-facing view and as parse compatibility for old barrier
+        metadata/tests that predate `StateWrite`.
+        """
         if not self.changes and self.writes:
             self.changes = {
                 str(write.path): write.incoming_value for write in self.writes

src/wf_core/runtime/step.py

@@ -11,6 +11,7 @@ from wf_core.models.steps import (
     InterruptNode,
     JoinNode,
     NodeUse,
+    SubgraphNode,
 )
 from wf_core.models.workflow import Workflow
 from wf_core.runtime.ops.flow import advance_frame, append_step_result_trace
@@ -115,6 +116,11 @@ def step_workflow(
         return handle_interrupt_step(run, step)
     elif isinstance(step, ForeachNode):
         return step_foreach(workflow, run, step, index, reducers=reducers)
+    elif isinstance(step, SubgraphNode):
+        raise WorkflowExecutionError(
+            f"subgraph step {step.id!r} references {step.workflow!r}, "
+            "but native subgraph execution is not implemented yet"
+        )
     else:
         raise WorkflowExecutionError(
             f"unsupported step type {getattr(step, 'type', type(step).__name__)!r}"
@@ -209,6 +215,11 @@ async def step_workflow_async(
         return handle_interrupt_step(run, step)
     elif isinstance(step, ForeachNode):
         return step_foreach(workflow, run, step, index, reducers=reducers)
+    elif isinstance(step, SubgraphNode):
+        raise WorkflowExecutionError(
+            f"subgraph step {step.id!r} references {step.workflow!r}, "
+            "but native subgraph execution is not implemented yet"
+        )
     else:
         raise WorkflowExecutionError(
             f"unsupported step type {getattr(step, 'type', type(step).__name__)!r}"