examples, docs, ergonomics and code review fixes

docs/current_roadmap.md

@@ -57,6 +57,8 @@ implementation state.
   subgraph boundary. Saved/deployed child resolution remains next. Wrapper
   helpers currently run child workflows as ordinary nodes; native
   `SubgraphNode` is now the graph-as-node path for prepared children.
+  `WorkflowBuilder.prepare_subgraph()` and `WorkflowBuilder.resume()` make the
+  local runnable/resumable path available without core-runtime plumbing.
 - **Concurrent foreach**: implemented in core with explicit scheduling,
   reducer/merge semantics, item error policy, async handler batching, and
   quiescent interrupt behavior. Remaining work is polish and future reuse of

docs/wf_core_architecture.md

@@ -140,14 +140,17 @@ limits and intended adapter seam.
   and lineage; child output commits only through declared boundary bindings and
   the parent routes by the child's terminal workflow outcome. Saved/deployed
   workflow resolution remains outside core and is not implemented at this
-  boundary yet.
+  boundary yet. For local authoring, `WorkflowBuilder.prepare_subgraph()`
+  registers a child builder and `WorkflowBuilder.resume()` continues a paused
+  prepared-child interrupt without requiring direct core-runtime calls.
 - The current `wf_authoring` wrapper helpers still run child workflows as
   ordinary sync or async nodes and therefore do not preserve native child
   state or resumable interrupts. Native `SubgraphNode` plus
   `PreparedSubgraph` is the first-class path: prepared child interrupts now
   bubble to the parent run and resume inside the original child scope. See
   `examples/authoring_workflow_as_node.py` for the compatibility wrapper shape
-  and `examples/authoring_native_subgraph.py` for the native path.
+  and `examples/authoring_native_subgraph.py` plus
+  `examples/authoring_native_subgraph_interrupt.py` for the native path.
 - Saved workflow-as-node execution with interrupts still requires platform
   resolution of artifact/deployment references into prepared child
   dependencies before core execution begins.

docs/workflow_artifacts.md

@@ -662,8 +662,10 @@ artifact/deployment resolution into prepared children before core can run it.
 
 See `examples/authoring_workflow_as_node.py` for the compatibility wrapper-node
 approach and `examples/authoring_native_subgraph.py` for native prepared-child
-execution. In the wrapper example the parent trace sees one node call; in the
-native example child trace entries remain in the parent run state.
+execution. `examples/authoring_native_subgraph_interrupt.py` demonstrates a
+native child pause and builder-driven resume. In the wrapper example the
+parent trace sees one node call; in the native examples child trace entries
+remain in the parent run state.
 
 Until that core upgrade exists, artifact tooling must not assume that an
 interrupting saved workflow can safely be used as a child node. Top-level saved

examples/authoring_native_subgraph.py

@@ -10,7 +10,7 @@ from wf_authoring import (
     output_to,
     state_path,
 )
-from wf_core import END, PreparedSubgraph, RunState, Workflow, execute_workflow
+from wf_core import END, RunState
 
 
 class ChildInput(BaseModel):
@@ -74,14 +74,15 @@ def build_child_workflow() -> WorkflowBuilder:
     return child
 
 
-def build_parent_workflow(child_workflow: Workflow) -> WorkflowBuilder:
-    """Build a parent graph with one native child-workflow boundary."""
+def build_parent_workflow(child: WorkflowBuilder) -> WorkflowBuilder:
+    """Build a parent graph with one registered native child workflow."""
     parent = WorkflowBuilder(
         name="native_parent",
         input_schema=ParentInput,
         state_schema=ParentState,
         output_schema=ParentOutput,
     )
+    child_workflow = parent.prepare_subgraph(child)
     run_child = parent.subgraph(
         workflow=child_workflow,
         id="run_child",
@@ -94,28 +95,16 @@ def build_parent_workflow(child_workflow: Workflow) -> WorkflowBuilder:
 
 
 def run_native_subgraph_example(prompt: str = "hello") -> RunState:
-    """Execute a native subgraph using its prepared local runtime dependency.
+    """Execute a native subgraph using builder-managed local dependencies.
 
     `WorkflowBuilder.subgraph()` records the child contract in the parent graph.
-    `PreparedSubgraph` separately supplies the Python handlers needed to execute
-    that contract; saved artifact/deployment resolution belongs above wf_core.
+    `prepare_subgraph()` separately registers the Python handlers needed to
+    execute that contract; saved artifact/deployment resolution belongs above
+    this local authoring helper.
     """
     child = build_child_workflow()
-    child_workflow = child.compile()
-    parent = build_parent_workflow(child_workflow)
-    return execute_workflow(
-        parent.compile(),
-        {"prompt": prompt},
-        parent.registry(),
-        reducers=parent.reducer_registry(),
-        subgraphs={
-            child_workflow.name: PreparedSubgraph(
-                workflow=child_workflow,
-                registry=child.registry(),
-                reducers=child.reducer_registry(),
-            )
-        },
-    )
+    parent = build_parent_workflow(child)
+    return parent.execute({"prompt": prompt})
 
 
 def main() -> None:

examples/authoring_native_subgraph_interrupt.py

@@ -0,0 +1,107 @@
+from __future__ import annotations
+
+from pydantic import BaseModel
+
+from wf_authoring import WorkflowBuilder, input_from, input_path, output_to, state_path
+from wf_core import END, InterruptRequest, RunState
+
+
+class ChildInput(BaseModel):
+    """Question forwarded into the child workflow."""
+
+    question: str
+
+
+class ChildState(BaseModel):
+    """Child-owned answer populated only after resume."""
+
+    answer: str = ""
+
+
+class ChildOutput(BaseModel):
+    """Child result projected through the parent subgraph boundary."""
+
+    answer: str
+
+
+class ParentInput(BaseModel):
+    """Parent input used to build the child interrupt request."""
+
+    question: str
+
+
+class ParentState(BaseModel):
+    """Parent state updated only after the child resumes and completes."""
+
+    result: str = ""
+
+
+class ParentOutput(BaseModel):
+    """Final result exposed by the parent workflow."""
+
+    result: str
+
+
+def build_interrupting_child() -> WorkflowBuilder:
+    """Build a child graph whose first step pauses for one supplied answer."""
+    child = WorkflowBuilder(
+        name="answer_child",
+        input_schema=ChildInput,
+        state_schema=ChildState,
+        output_schema=ChildOutput,
+    )
+    ask = child.interrupt(
+        id="ask",
+        kind="input",
+        request=[input_from(input_path("question"), "question")],
+        resume=[output_to("answer", state_path("answer"))],
+    )
+    child.set_entry_point(ask)
+    child.connect(ask, "submitted", END)
+    return child
+
+
+def build_interrupting_parent(child: WorkflowBuilder) -> WorkflowBuilder:
+    """Build a parent graph that invokes the interrupting child natively."""
+    parent = WorkflowBuilder(
+        name="answer_parent",
+        input_schema=ParentInput,
+        state_schema=ParentState,
+        output_schema=ParentOutput,
+    )
+    child_workflow = parent.prepare_subgraph(child)
+    request_answer = parent.subgraph(
+        workflow=child_workflow,
+        id="request_answer",
+        input=[input_from(input_path("question"), "question")],
+        output=[output_to("answer", state_path("result"))],
+    )
+    parent.set_entry_point(request_answer)
+    parent.connect(request_answer, "ok", END)
+    return parent
+
+
+def run_native_subgraph_interrupt_example() -> tuple[InterruptRequest, RunState]:
+    """Pause in a child workflow, then resume it through authoring helpers.
+
+    `RunState` is resumed in place, so the example retains the emitted request
+    before continuing the run.
+    """
+    parent = build_interrupting_parent(build_interrupting_child())
+    paused = parent.execute({"question": "What is your answer?"})
+    if paused.interrupt is None:
+        raise AssertionError("expected child workflow to interrupt")
+    request = paused.interrupt
+    resumed = parent.resume(paused, payload={"answer": "confirmed"})
+    return request, resumed
+
+
+def main() -> None:
+    """Run the native interrupt/resume example from the command line."""
+    request, resumed = run_native_subgraph_interrupt_example()
+    print("interrupt", request.payload)
+    print("result", resumed.output)
+
+
+if __name__ == "__main__":
+    main()

src/wf_authoring/builder/core.py

@@ -14,7 +14,9 @@ from wf_core import (
     ForeachItemErrorPolicy,
     ForeachNode,
     InterruptNode,
+    NodeHandler,
     NodeUse,
+    PreparedSubgraph,
     SchemaRef,
     StateSchema,
     SubgraphNode,
@@ -22,6 +24,7 @@ from wf_core import (
     WorkflowRef,
     RunState,
     execute_workflow,
+    resume_workflow,
 )
 from wf_core.errors import WorkflowExecutionError
 from wf_core.models.conditions import Condition as CoreCondition
@@ -208,6 +211,9 @@ class WorkflowBuilder:
     node_specs: dict[str, NodeSpec[Any, Any]] = field(default_factory=dict)
     nodes: list[Step] = field(default_factory=list)
     edges: list[Edge] = field(default_factory=list)
+    prepared_subgraphs: dict[str, PreparedSubgraph[NodeHandler]] = field(
+        default_factory=dict
+    )
 
     def __post_init__(self) -> None:
         """Normalize authoring-friendly schema declarations into core schemas."""
@@ -397,8 +403,9 @@ class WorkflowBuilder:
         """Add a native subgraph boundary using a compiled child workflow contract.
 
         Core executes local children when callers supply a matching
-        `PreparedSubgraph`; saved artifact resolution and child interrupt resume
-        remain higher-layer/future responsibilities.
+        `PreparedSubgraph`; call `prepare_subgraph()` when the child is another
+        local `WorkflowBuilder`. Saved artifact resolution remains a platform
+        responsibility.
         """
         node = subgraph_ref(
             id=id or self._next_step_id(_workflow_ref_base(workflow_ref, workflow)),
@@ -411,6 +418,21 @@ class WorkflowBuilder:
         self.nodes.append(node)
         return node
 
+    def prepare_subgraph(self, child: WorkflowBuilder) -> Workflow:
+        """Register a local child builder for native execution and return its graph.
+
+        The compiled child contract is passed to `subgraph()`, while its Python
+        registry/reducers stay in this parent builder's local execution
+        environment. Saved artifacts should be resolved above this helper.
+        """
+        workflow = child.compile()
+        self.prepared_subgraphs[workflow.name] = PreparedSubgraph(
+            workflow=workflow,
+            registry=child.registry(),
+            reducers=child.reducer_registry(),
+        )
+        return workflow
+
     def _next_step_id(self, base: str) -> str:
         """Return a stable unused step id based on the requested base name."""
         return next_step_id(base, self.nodes)
@@ -454,6 +476,25 @@ class WorkflowBuilder:
             workflow_input,
             self.registry(),
             reducers=self.reducer_registry(),
+            subgraphs=self.prepared_subgraphs,
+        )
+
+    def resume(
+        self,
+        run: RunState,
+        *,
+        payload: dict[str, Any] | None = None,
+        outcome: str = "submitted",
+    ) -> RunState:
+        """Resume a locally executed workflow, including prepared child interrupts."""
+        return resume_workflow(
+            self.compile(),
+            run,
+            self.registry(),
+            resume_payload=payload,
+            resume_outcome=outcome,
+            reducers=self.reducer_registry(),
+            subgraphs=self.prepared_subgraphs,
         )
 
     def condition(

src/wf_core/runtime/lineage.py

@@ -215,6 +215,10 @@ def _lineage_chain(
     run: RunState, *, scope_id: str, lineage_id: str
 ) -> Iterator[LineageState]:
     lineage = _lineage(run, scope_id=scope_id, lineage_id=lineage_id)
-    if lineage.parent_id is not None:
-        yield from _lineage_chain(run, scope_id=scope_id, lineage_id=lineage.parent_id)
-    yield lineage
+    reverse_chain: list[LineageState] = []
+    while True:
+        reverse_chain.append(lineage)
+        if lineage.parent_id is None:
+            break
+        lineage = _lineage(run, scope_id=scope_id, lineage_id=lineage.parent_id)
+    yield from reversed(reverse_chain)

src/wf_core/runtime/ops/state.py

@@ -53,8 +53,19 @@ class StatePatch:
 
         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`.
+        metadata/tests that predate `StateWrite`. If both are supplied, they
+        must describe identical incoming writes; otherwise trace and replay
+        semantics would disagree.
         """
+        if self.changes and self.writes:
+            derived_changes = {
+                str(write.path): write.incoming_value for write in self.writes
+            }
+            if self.changes != derived_changes:
+                raise ValueError(
+                    "StatePatch constructed with inconsistent changes and writes"
+                )
+            return
         if not self.changes and self.writes:
             self.changes = {
                 str(write.path): write.incoming_value for write in self.writes

src/wf_core/runtime/preparation.py

@@ -37,7 +37,13 @@ def prepare_resume(
     interrupted_workflow: Workflow | None = None,
     interrupted_reducers: Mapping[str, ReducerDefinition] | None = None,
 ) -> WorkflowIndex | None:
-    """Validate and normalize a run state before resume execution."""
+    """Validate and normalize a run state before resume execution.
+
+    `workflow` and `reducers` always describe the outer run. For a routed
+    child interrupt, `interrupted_workflow` and `interrupted_reducers` identify
+    the child scope that owns the interrupt step, so resume bindings commit
+    inside that scope before outer scheduling continues.
+    """
     if run.workflow_name != workflow.name:
         raise WorkflowExecutionError(
             f"run state belongs to workflow {run.workflow_name!r}, not {workflow.name!r}"

tests/authoring/test_native_subgraph_example.py

@@ -1,6 +1,13 @@
 from __future__ import annotations
 
-from examples.authoring_native_subgraph import run_native_subgraph_example
+from examples.authoring_native_subgraph import (
+    build_child_workflow,
+    build_parent_workflow,
+    run_native_subgraph_example,
+)
+from examples.authoring_native_subgraph_interrupt import (
+    run_native_subgraph_interrupt_example,
+)
 from wf_core import RunStatus
 
 
@@ -13,3 +20,20 @@ def test_native_subgraph_example_runs_child_in_parent_trace() -> None:
         entry.node_id == "uppercase" and entry.frame_id != "root" for entry in run.trace
     )
     assert run.trace[-1].step_type == "subgraph"
+
+
+def test_builder_executes_registered_prepared_subgraph() -> None:
+    child = build_child_workflow()
+    parent = build_parent_workflow(child)
+
+    run = parent.execute({"prompt": "hello"})
+
+    assert run.output["result"] == "HELLO"
+
+
+def test_native_subgraph_interrupt_example_resumes_through_builder() -> None:
+    request, resumed = run_native_subgraph_interrupt_example()
+
+    assert request.node_id == "request_answer"
+    assert request.payload["question"] == "What is your answer?"
+    assert resumed.output["result"] == "confirmed"

tests/core/test_atomic_state_patches.py

@@ -281,6 +281,21 @@ def test_build_and_commit_patch_matches_apply_output_bindings() -> None:
     assert state_from_apply["person"]["name"] == state_from_patch["person"]["name"]
 
 
+def test_state_patch_rejects_inconsistent_trace_changes_and_writes() -> None:
+    with pytest.raises(ValueError, match="inconsistent changes and writes"):
+        StatePatch(
+            changes={"state.value": "trace"},
+            writes=[
+                StateWrite(
+                    path=StatePath(("value",)),
+                    incoming_value="actual",
+                    visible_value="actual",
+                    reducer=ReducerRef(name="wf.std.replace"),
+                )
+            ],
+        )
+
+
 def test_barrier_rejects_sibling_same_path_writes_without_reducer() -> None:
     workflow = _workflow(fields={"value": StateField(type="string")})
 

tests/core/test_lineage_state.py

@@ -158,6 +158,24 @@ def test_state_view_for_frame_overlays_writes_onto_frame_scope_state() -> None:
     assert run.state["value"] == "root"
 
 
+def test_lineage_state_view_handles_deep_ancestry_without_recursion() -> None:
+    run = create_run_state(_minimal_workflow(), {"value": "root"})
+    parent_id = "root"
+    for index in range(1100):
+        lineage_id = f"child-{index}"
+        add_lineage(
+            run,
+            scope_id="root",
+            lineage_id=lineage_id,
+            parent_id=parent_id,
+        )
+        parent_id = lineage_id
+
+    state_view = lineage_state_view(run, scope_id="root", lineage_id=parent_id)
+
+    assert state_view["value"] == "root"
+
+
 def test_non_root_frame_node_writes_are_buffered_in_lineage() -> None:
     workflow = _write_value_workflow()
     run = create_run_state(workflow, {"value": "root"})