native subgraph execution

src/wf_core/runtime/__init__.py

@@ -13,6 +13,7 @@ from .engine import (
     resume_workflow,
     resume_workflow_async,
 )
+from .subgraphs import PreparedSubgraph
 from .step import complete_step, step_workflow, step_workflow_async
 
 __all__ = [
@@ -29,4 +30,5 @@ __all__ = [
     "resume_workflow_async",
     "step_workflow",
     "step_workflow_async",
+    "PreparedSubgraph",
 ]

src/wf_core/runtime/engine.py

@@ -3,17 +3,19 @@ from __future__ import annotations
 from collections.abc import Mapping
 from typing import Any
 
+from wf_core.errors import WorkflowExecutionError
 from wf_core.models.workflow import Workflow
 from wf_core.runtime.ops.flow import finalize_run
 from wf_core.runtime.ops.merges import ReducerDefinition
 from wf_core.runtime.ops.nodes import AsyncNodeHandler, NodeHandler
 from wf_core.runtime.ops.runs import create_run_state
 from wf_core.runtime.scheduler import resolve_no_ready_frames, select_next_frame
-from wf_core.run_state import RunState, RunStatus
+from wf_core.run_state import ROOT_SCOPE_ID, RunState, RunStatus
 from wf_core.tokens import END
 
 from .preparation import prepare_new_run, prepare_resume
 from .step import step_workflow, step_workflow_async
+from .subgraphs import PreparedSubgraph, resolve_prepared_subgraph
 
 
 def execute_workflow(
@@ -22,13 +24,20 @@ def execute_workflow(
     registry: Mapping[str, NodeHandler],
     *,
     reducers: Mapping[str, ReducerDefinition] | None = None,
+    subgraphs: Mapping[str, PreparedSubgraph[NodeHandler]] | None = None,
 ) -> RunState:
     """Create a run and execute a workflow synchronously until it stops."""
     run = create_run_state(workflow, workflow_input)
 
     try:
         prepare_new_run(workflow, workflow_input, run)
-        return resume_workflow(workflow, run, registry, reducers=reducers)
+        return resume_workflow(
+            workflow,
+            run,
+            registry,
+            reducers=reducers,
+            subgraphs=subgraphs,
+        )
     except Exception as exc:
         run.status = RunStatus.FAILED
         run.error = str(exc)
@@ -41,13 +50,20 @@ async def execute_workflow_async(
     registry: Mapping[str, AsyncNodeHandler],
     *,
     reducers: Mapping[str, ReducerDefinition] | None = None,
+    subgraphs: Mapping[str, PreparedSubgraph[AsyncNodeHandler]] | None = None,
 ) -> RunState:
     """Create a run and execute a workflow asynchronously until it stops."""
     run = create_run_state(workflow, workflow_input)
 
     try:
         prepare_new_run(workflow, workflow_input, run)
-        return await resume_workflow_async(workflow, run, registry, reducers=reducers)
+        return await resume_workflow_async(
+            workflow,
+            run,
+            registry,
+            reducers=reducers,
+            subgraphs=subgraphs,
+        )
     except Exception as exc:
         run.status = RunStatus.FAILED
         run.error = str(exc)
@@ -62,6 +78,7 @@ def resume_workflow(
     resume_payload: dict[str, Any] | None = None,
     resume_outcome: str = "submitted",
     reducers: Mapping[str, ReducerDefinition] | None = None,
+    subgraphs: Mapping[str, PreparedSubgraph[NodeHandler]] | None = None,
 ) -> RunState:
     """Resume a synchronous run from its current state."""
     index = prepare_resume(
@@ -77,17 +94,22 @@ def resume_workflow(
         return run
 
     while True:
-        if select_next_frame(run) is None:
+        frame = select_next_frame(run)
+        if frame is None:
             status = resolve_no_ready_frames(run)
             if status == RunStatus.COMPLETED:
                 break
             return run
+        active_workflow, active_registry, active_reducers = _sync_execution_target(
+            workflow, registry, reducers, run, subgraphs
+        )
         step_workflow(
-            workflow,
+            active_workflow,
             run,
-            registry,
-            index=index,
-            reducers=reducers,
+            active_registry,
+            index=index if frame.scope_id == ROOT_SCOPE_ID else None,
+            reducers=active_reducers,
+            subgraphs=subgraphs,
         )
         if run.status == RunStatus.INTERRUPTED:
             return run
@@ -103,6 +125,7 @@ async def resume_workflow_async(
     resume_payload: dict[str, Any] | None = None,
     resume_outcome: str = "submitted",
     reducers: Mapping[str, ReducerDefinition] | None = None,
+    subgraphs: Mapping[str, PreparedSubgraph[AsyncNodeHandler]] | None = None,
 ) -> RunState:
     """Resume an async run from its current state."""
     index = prepare_resume(
@@ -118,19 +141,68 @@ async def resume_workflow_async(
         return run
 
     while True:
-        if select_next_frame(run) is None:
+        frame = select_next_frame(run)
+        if frame is None:
             status = resolve_no_ready_frames(run)
             if status == RunStatus.COMPLETED:
                 break
             return run
+        active_workflow, active_registry, active_reducers = _async_execution_target(
+            workflow, registry, reducers, run, subgraphs
+        )
         await step_workflow_async(
-            workflow,
+            active_workflow,
             run,
-            registry,
-            index=index,
-            reducers=reducers,
+            active_registry,
+            index=index if frame.scope_id == ROOT_SCOPE_ID else None,
+            reducers=active_reducers,
+            subgraphs=subgraphs,
         )
         if run.status == RunStatus.INTERRUPTED:
             return run
 
     return finalize_run(workflow, run)
+
+
+def _sync_execution_target(
+    root_workflow: Workflow,
+    root_registry: Mapping[str, NodeHandler],
+    root_reducers: Mapping[str, ReducerDefinition] | None,
+    run: RunState,
+    subgraphs: Mapping[str, PreparedSubgraph[NodeHandler]] | None,
+) -> tuple[Workflow, Mapping[str, NodeHandler], Mapping[str, ReducerDefinition] | None]:
+    """Return the workflow dependencies owned by the selected frame scope."""
+    frame = run.current_frame()
+    if frame.scope_id == ROOT_SCOPE_ID:
+        return root_workflow, root_registry, root_reducers
+    scope = run.scopes.get(frame.scope_id)
+    if scope is None or scope.workflow_ref is None:
+        raise WorkflowExecutionError(
+            f"child frame {frame.id!r} has no prepared workflow scope"
+        )
+    child = resolve_prepared_subgraph(scope.workflow_ref, subgraphs)
+    return child.workflow, child.registry, child.reducers
+
+
+def _async_execution_target(
+    root_workflow: Workflow,
+    root_registry: Mapping[str, AsyncNodeHandler],
+    root_reducers: Mapping[str, ReducerDefinition] | None,
+    run: RunState,
+    subgraphs: Mapping[str, PreparedSubgraph[AsyncNodeHandler]] | None,
+) -> tuple[
+    Workflow,
+    Mapping[str, AsyncNodeHandler],
+    Mapping[str, ReducerDefinition] | None,
+]:
+    """Return async workflow dependencies owned by the selected frame scope."""
+    frame = run.current_frame()
+    if frame.scope_id == ROOT_SCOPE_ID:
+        return root_workflow, root_registry, root_reducers
+    scope = run.scopes.get(frame.scope_id)
+    if scope is None or scope.workflow_ref is None:
+        raise WorkflowExecutionError(
+            f"child frame {frame.id!r} has no prepared workflow scope"
+        )
+    child = resolve_prepared_subgraph(scope.workflow_ref, subgraphs)
+    return child.workflow, child.registry, child.reducers

src/wf_core/runtime/lineage.py

@@ -7,10 +7,9 @@ from typing import Any
 
 from wf_core.errors import WorkflowExecutionError
 from wf_core.run_state import ExecutionFrame, LineageState, RunState, StateWrite
-from wf_core.run_state import ROOT_LINEAGE_ID, ROOT_SCOPE_ID
 from wf_core.runtime.foreach_state import ForeachBarrierState, item_frame_owner
 from wf_core.runtime.ops.state import StatePatch
-from wf_core.runtime.ops.state import safe_set_nested_value
+from wf_core.runtime.ops.state import commit_state_patch, safe_set_nested_value
 
 
 @dataclass(slots=True)
@@ -79,14 +78,34 @@ def lineage_writes_for_frame(
     return pending.patch.writes
 
 
-def is_root_lineage_frame(frame: ExecutionFrame) -> bool:
-    """Return whether a frame currently commits directly to root run state.
+def is_scope_root_lineage_frame(run: RunState, frame: ExecutionFrame) -> bool:
+    """Return whether writes from this frame commit to its scope state root."""
+    lineage = run.lineages.get(frame.lineage_id)
+    return (
+        lineage is not None
+        and lineage.scope_id == frame.scope_id
+        and lineage.parent_id is None
+    )
 
-    This is a migration shortcut, not the final commit policy. Once native
-    subgraphs can complete, direct commits should be decided by an explicit
-    scope/lineage commit target rather than only by root ids.
+
+def commit_patch_for_frame(
+    run: RunState, frame: ExecutionFrame, patch: StatePatch
+) -> dict[str, Any]:
+    """Commit at a scope root or buffer writes in the frame lineage.
+
+    Child workflow root frames own a committed child-state root just like the
+    top-level root frame owns `RunState.state`. Descendant branch/item frames
+    remain isolated until an explicit barrier or future gather commits them.
     """
-    return frame.scope_id == ROOT_SCOPE_ID and frame.lineage_id == ROOT_LINEAGE_ID
+    if is_scope_root_lineage_frame(run, frame):
+        return commit_state_patch(scope_state_for_frame(run, frame), patch)
+    append_lineage_writes(
+        run,
+        scope_id=frame.scope_id,
+        lineage_id=frame.lineage_id,
+        writes=patch.writes,
+    )
+    return {}
 
 
 def scope_state_for_frame(run: RunState, frame: ExecutionFrame) -> dict[str, Any]:
@@ -97,6 +116,14 @@ def scope_state_for_frame(run: RunState, frame: ExecutionFrame) -> dict[str, Any
     return scope.committed_state
 
 
+def scope_input_for_frame(run: RunState, frame: ExecutionFrame) -> dict[str, Any]:
+    """Return the invocation input associated with the frame's workflow scope."""
+    scope = run.scopes.get(frame.scope_id)
+    if scope is None:
+        raise ValueError(f"unknown scope {frame.scope_id!r}")
+    return scope.workflow_input
+
+
 def add_lineage(
     run: RunState,
     *,

src/wf_core/runtime/ops/flow.py

@@ -80,6 +80,10 @@ def advance_frame(
     frame.activated_incoming_edge = frame.node_id
     frame.node_id = next_node_id
     if next_node_id == END:
+        if frame.kind in {"workflow", "subgraph_root"}:
+            # Legacy terminal routing emits the workflow-level `ok` outcome.
+            # Explicit EndNode execution stores its declared outcome first.
+            frame.metadata.setdefault("workflow_outcome", "ok")
         frame.status = FrameStatus.COMPLETED
         frame.finished_at_node_id = END
         wake_parent_for_child_progress(run, frame.id)

src/wf_core/runtime/ops/foreach.py

@@ -14,18 +14,18 @@ from wf_core.runtime.foreach_state import (
 )
 from wf_core.runtime.lineage import (
     add_lineage,
-    append_lineage_writes,
-    is_root_lineage_frame,
+    commit_patch_for_frame,
     lineage_patch,
+    scope_input_for_frame,
 )
 from wf_core.runtime.ops.flow import advance_frame, append_step_result_trace
 from wf_core.runtime.ops.frames import frame_context_values
 from wf_core.runtime.ops.index import WorkflowIndex
 from wf_core.runtime.ops.merges import ReducerDefinition
+from wf_core.runtime.ops.overlays import state_view_for_frame
 from wf_core.runtime.ops.state import (
     StatePatch,
     build_barrier_patch,
-    commit_state_patch,
 )
 from wf_core.runtime.scheduler import (
     ForeachIterationMetadata,
@@ -182,8 +182,8 @@ def _resolve_foreach_iterable(
 ) -> list[object]:
     iterable = safe_resolve_path(
         str(step.over),
-        state=run.state,
-        workflow_input=run.workflow_input,
+        state=state_view_for_frame(run, frame),
+        workflow_input=scope_input_for_frame(run, frame),
         context=frame_context_values(frame),
     )
     if not isinstance(iterable, list):
@@ -346,19 +346,10 @@ def _finish_concurrent_foreach(
     combined = build_barrier_patch(
         workflow,
         item_patches,
-        run.state,
+        state_view_for_frame(run, frame),
         reducers=reducers,
     )
-    if is_root_lineage_frame(frame):
-        state_changes = commit_state_patch(run.state, combined)
-    else:
-        append_lineage_writes(
-            run,
-            scope_id=frame.scope_id,
-            lineage_id=frame.lineage_id,
-            writes=combined.writes,
-        )
-        state_changes = {}
+    state_changes = commit_patch_for_frame(run, frame, combined)
     append_step_result_trace(
         run,
         frame_id=frame.id,

src/wf_core/runtime/ops/handlers.py

@@ -3,9 +3,11 @@ from __future__ import annotations
 from wf_core.conditions import eval_condition
 from wf_core.models.steps import ConditionNode, InterruptNode
 from wf_core.run_state import FrameStatus, RunState, RunStatus, StepExecutionResult
+from wf_core.runtime.lineage import scope_input_for_frame
 from wf_core.runtime.ops.flow import append_trace
 from wf_core.runtime.ops.frames import frame_context_values
 from wf_core.runtime.ops.interrupts import build_interrupt_request
+from wf_core.runtime.ops.overlays import state_view_for_frame
 
 
 def handle_condition_step(
@@ -15,8 +17,8 @@ def handle_condition_step(
     frame = run.current_frame()
     predicate = eval_condition(
         step.check,
-        run.state,
-        run.workflow_input,
+        state_view_for_frame(run, frame),
+        scope_input_for_frame(run, frame),
         frame.prior_outcome,
     )
     outcome = "true" if predicate else "false"

src/wf_core/runtime/ops/nodes.py

@@ -18,12 +18,16 @@ from wf_core.run_state import (
     StepExecutionResult,
 )
 from wf_core.runtime.foreach_state import ForeachBarrierState, item_frame_owner
-from wf_core.runtime.lineage import append_lineage_writes, is_root_lineage_frame
+from wf_core.runtime.lineage import (
+    append_lineage_writes,
+    commit_patch_for_frame,
+    scope_input_for_frame,
+)
 from wf_core.runtime.ops.frames import frame_context_values
 from wf_core.runtime.ops.merges import ReducerDefinition
 from wf_core.runtime.ops.overlays import state_view_for_frame
 from wf_core.runtime.ops.schemas import validate_payload_against_schema
-from wf_core.runtime.ops.state import StatePatch, build_output_patch, commit_state_patch
+from wf_core.runtime.ops.state import StatePatch, build_output_patch
 
 NodeHandler = Callable[[dict[str, Any], RuntimeContext], NodeResult | dict[str, Any]]
 AsyncNodeHandler = Callable[
@@ -62,7 +66,7 @@ def _resolve_node_execution(
             value = safe_resolve_path(
                 str(binding.path),
                 state=state_view,
-                workflow_input=run.workflow_input,
+                workflow_input=scope_input_for_frame(run, frame),
                 context=context_values,
             )
         else:
@@ -121,18 +125,7 @@ def _finalize_node_execution(
     )
     owner = item_frame_owner(frame)
     if owner is None:
-        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,
-                lineage_id=frame.lineage_id,
-                writes=patch.writes,
-            )
-            state_changes = {}
+        state_changes = commit_patch_for_frame(run, frame, patch)
     else:
         parent_frame_id, foreach_node_id, item_index = owner
         parent_frame = run.frames[parent_frame_id]
@@ -155,7 +148,7 @@ def _finalize_node_execution(
             barrier.save_to_frame(parent_frame, foreach_node_id)
             state_changes = {}
         else:
-            state_changes = commit_state_patch(run.state, patch)
+            state_changes = commit_patch_for_frame(run, parent_frame, patch)
     return StepExecutionResult(
         outcome=result.outcome,
         resolved_input=resolved_input,

src/wf_core/runtime/ops/runs.py

@@ -18,12 +18,20 @@ from wf_core.run_state import (
 from wf_core.runtime.scheduler import add_frame
 
 
-def create_run_state(workflow: Workflow, workflow_input: dict[str, object]) -> RunState:
+def initial_state(
+    workflow: Workflow, workflow_input: dict[str, object]
+) -> dict[str, object]:
+    """Create one scope's committed state from defaults plus workflow input."""
     state: dict[str, object] = {}
     for field in workflow.state_schema.fields:
         if field.default is not None:
             set_nested_value(state, list(field.path.parts), deepcopy(field.default))
     state.update(dict(workflow_input))
+    return state
+
+
+def create_run_state(workflow: Workflow, workflow_input: dict[str, object]) -> RunState:
+    state = initial_state(workflow, workflow_input)
     run = RunState(
         workflow_name=workflow.name,
         status=RunStatus.PENDING,
@@ -33,6 +41,7 @@ def create_run_state(workflow: Workflow, workflow_input: dict[str, object]) -> R
             ROOT_SCOPE_ID: RuntimeScope(
                 id=ROOT_SCOPE_ID,
                 workflow_name=workflow.name,
+                workflow_input=dict(workflow_input),
                 committed_state=state,
             )
         },

src/wf_core/runtime/step.py

@@ -38,7 +38,9 @@ from wf_core.runtime.scheduler import (
     select_next_frame,
     wake_parent_for_child_progress,
 )
+from wf_core.runtime.subgraphs import PreparedSubgraph, step_subgraph
 from wf_core.run_state import ExecutionFrame, FrameStatus, RunState, StepExecutionResult
+from wf_core.run_state import ROOT_SCOPE_ID
 from wf_core.tokens import END
 
 from .preparation import prepare_step
@@ -85,7 +87,10 @@ def complete_end_step(
 ) -> RunState:
     """Record an explicit workflow terminal and complete the active frame."""
     result = StepExecutionResult(outcome=outcome)
-    run.outcome = outcome
+    frame = run.frames[frame_id]
+    frame.metadata["workflow_outcome"] = outcome
+    if frame.parent_frame_id is None:
+        run.outcome = outcome
     append_step_result_trace(
         run,
         frame_id=frame_id,
@@ -96,7 +101,7 @@ def complete_end_step(
     )
     advance_frame(
         run,
-        run.frames[frame_id],
+        frame,
         outcome=outcome,
         next_node_id=END,
     )
@@ -110,6 +115,7 @@ def step_workflow(
     *,
     index: WorkflowIndex | None = None,
     reducers: Mapping[str, ReducerDefinition] | None = None,
+    subgraphs: Mapping[str, PreparedSubgraph[NodeHandler]] | None = None,
 ) -> RunState:
     """Execute at most one synchronous workflow step."""
     frame = run.current_frame() if run.current_frame_id is not None else None
@@ -149,14 +155,23 @@ def step_workflow(
             outcome=step.outcome,
         )
     elif isinstance(step, InterruptNode):
+        if frame.kind == "subgraph_root" or frame.scope_id != ROOT_SCOPE_ID:
+            raise WorkflowExecutionError(
+                "child interrupts are not supported until native subgraph resume routing exists"
+            )
         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"
+        step_result = step_subgraph(
+            workflow,
+            run,
+            step,
+            subgraphs=subgraphs,
+            reducers=reducers,
         )
+        if step_result is None:
+            return run
     else:
         raise WorkflowExecutionError(
             f"unsupported step type {getattr(step, 'type', type(step).__name__)!r}"
@@ -205,6 +220,7 @@ async def step_workflow_async(
     *,
     index: WorkflowIndex | None = None,
     reducers: Mapping[str, ReducerDefinition] | None = None,
+    subgraphs: Mapping[str, PreparedSubgraph[AsyncNodeHandler]] | None = None,
 ) -> RunState:
     """Execute at most one async workflow step."""
     frame = run.current_frame() if run.current_frame_id is not None else None
@@ -255,14 +271,23 @@ async def step_workflow_async(
             outcome=step.outcome,
         )
     elif isinstance(step, InterruptNode):
+        if frame.kind == "subgraph_root" or frame.scope_id != ROOT_SCOPE_ID:
+            raise WorkflowExecutionError(
+                "child interrupts are not supported until native subgraph resume routing exists"
+            )
         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"
+        step_result = step_subgraph(
+            workflow,
+            run,
+            step,
+            subgraphs=subgraphs,
+            reducers=reducers,
         )
+        if step_result is None:
+            return run
     else:
         raise WorkflowExecutionError(
             f"unsupported step type {getattr(step, 'type', type(step).__name__)!r}"

src/wf_core/runtime/subgraphs.py

@@ -0,0 +1,273 @@
+from __future__ import annotations
+
+from collections.abc import Callable, Mapping, Sequence
+from dataclasses import dataclass
+from typing import Any, Generic, TypeVar
+
+from wf_core.conditions import safe_resolve_path
+from wf_core.errors import WorkflowExecutionError
+from wf_core.local_paths import LocalPathError, set_local_value
+from wf_core.models.steps import (
+    InputBinding,
+    InputPathBinding,
+    InputValueBinding,
+    SubgraphNode,
+)
+from wf_core.models.workflow import Workflow
+from wf_core.models.workflow_refs import WorkflowRef
+from wf_core.run_state import (
+    ExecutionFrame,
+    FrameStatus,
+    LineageState,
+    RunState,
+    RuntimeScope,
+    StepExecutionResult,
+)
+from wf_core.runtime.lineage import commit_patch_for_frame
+from wf_core.runtime.ops.frames import frame_context_values
+from wf_core.runtime.ops.merges import ReducerDefinition
+from wf_core.runtime.ops.overlays import state_view_for_frame
+from wf_core.runtime.ops.runs import initial_state
+from wf_core.runtime.ops.schemas import validate_payload_against_schema
+from wf_core.runtime.ops.state import build_output_patch, project_output
+from wf_core.runtime.scheduler import add_frame, block_frame_on_children
+
+HandlerT = TypeVar("HandlerT", bound=Callable[..., object])
+_ACTIVATION_KEY = "subgraph_activation"
+
+
+@dataclass(slots=True, frozen=True)
+class PreparedSubgraph(Generic[HandlerT]):
+    """Executable local child dependency supplied by the caller.
+
+    Core owns child execution semantics but does not load artifacts or resolve
+    deployment/source bindings. Higher layers must resolve those concerns into
+    this prepared dependency before a run starts.
+    """
+
+    workflow: Workflow
+    registry: Mapping[str, HandlerT]
+    reducers: Mapping[str, ReducerDefinition] | None = None
+
+
+@dataclass(slots=True, frozen=True)
+class SubgraphActivation:
+    """Runtime ownership record for one in-flight subgraph boundary."""
+
+    workflow_ref: WorkflowRef
+    scope_id: str
+    lineage_id: str
+    child_frame_id: str
+    child_input: dict[str, Any]
+
+    @classmethod
+    def from_frame(cls, frame: ExecutionFrame) -> SubgraphActivation | None:
+        raw = frame.metadata.get(_ACTIVATION_KEY)
+        if raw is None:
+            return None
+        if not isinstance(raw, Mapping):
+            raise WorkflowExecutionError(
+                f"malformed subgraph activation for frame {frame.id!r}"
+            )
+        try:
+            return cls(
+                workflow_ref=WorkflowRef.model_validate(raw["workflow_ref"]),
+                scope_id=str(raw["scope_id"]),
+                lineage_id=str(raw["lineage_id"]),
+                child_frame_id=str(raw["child_frame_id"]),
+                child_input=dict(raw["child_input"]),
+            )
+        except (KeyError, TypeError, ValueError) as exc:
+            raise WorkflowExecutionError(
+                f"malformed subgraph activation for frame {frame.id!r}"
+            ) from exc
+
+    def save_to_frame(self, frame: ExecutionFrame) -> None:
+        frame.metadata[_ACTIVATION_KEY] = {
+            "workflow_ref": self.workflow_ref.model_dump(mode="json"),
+            "scope_id": self.scope_id,
+            "lineage_id": self.lineage_id,
+            "child_frame_id": self.child_frame_id,
+            "child_input": dict(self.child_input),
+        }
+
+
+def resolve_prepared_subgraph(
+    ref: WorkflowRef,
+    subgraphs: Mapping[str, PreparedSubgraph[HandlerT]] | None,
+) -> PreparedSubgraph[HandlerT]:
+    """Resolve a prepared local child; artifact loading is not a core concern."""
+    if ref.name is None:
+        raise WorkflowExecutionError(
+            f"saved child workflow reference {ref.display!r} is not prepared for core execution"
+        )
+    prepared = None if subgraphs is None else subgraphs.get(ref.name)
+    if prepared is None:
+        raise WorkflowExecutionError(
+            f"no prepared child workflow registered for {ref.name!r}"
+        )
+    return prepared
+
+
+def resolve_input_bindings(
+    bindings: Sequence[InputBinding],
+    *,
+    state: Mapping[str, Any],
+    workflow_input: Mapping[str, Any],
+    context: Mapping[str, Any],
+    label: str,
+) -> dict[str, Any]:
+    """Build a local input payload from canonical value/path bindings."""
+    payload: dict[str, Any] = {}
+    for binding in bindings:
+        if isinstance(binding, InputValueBinding):
+            value = binding.value
+        elif isinstance(binding, InputPathBinding):
+            value = safe_resolve_path(
+                str(binding.path),
+                state=state,
+                workflow_input=workflow_input,
+                context=context,
+            )
+        else:
+            raise WorkflowExecutionError(f"unsupported input binding for {label}")
+        try:
+            set_local_value(payload, binding.target, value)
+        except LocalPathError as exc:
+            raise WorkflowExecutionError(str(exc)) from exc
+    return payload
+
+
+def step_subgraph(
+    workflow: Workflow,
+    run: RunState,
+    step: SubgraphNode,
+    *,
+    subgraphs: Mapping[str, PreparedSubgraph[HandlerT]] | None,
+    reducers: Mapping[str, ReducerDefinition] | None,
+) -> StepExecutionResult | None:
+    """Start or finish one native child activation.
+
+    Returning ``None`` means the parent frame is blocked while child frames run.
+    Returning a result means child execution completed and the parent boundary
+    can advance normally through the child's terminal workflow outcome.
+    """
+    frame = run.current_frame()
+    activation = SubgraphActivation.from_frame(frame)
+    prepared = resolve_prepared_subgraph(step.workflow, subgraphs)
+    if activation is None:
+        _start_subgraph(run, frame, step, prepared)
+        return None
+    return _finish_subgraph(workflow, run, frame, step, activation, prepared, reducers)
+
+
+def _start_subgraph(
+    run: RunState,
+    frame: ExecutionFrame,
+    step: SubgraphNode,
+    prepared: PreparedSubgraph[HandlerT],
+) -> None:
+    prepared.workflow.validate_structure().raise_for_errors()
+    parent_scope = run.scopes[frame.scope_id]
+    child_input = resolve_input_bindings(
+        step.input,
+        state=state_view_for_frame(run, frame),
+        workflow_input=parent_scope.workflow_input,
+        context=frame_context_values(frame),
+        label=f"subgraph {step.id!r}",
+    )
+    validate_payload_against_schema(
+        step.input_schema, child_input, f"subgraph input for {step.id}"
+    )
+    validate_payload_against_schema(
+        prepared.workflow.input_schema,
+        child_input,
+        f"child workflow input for {step.id}",
+    )
+
+    scope_id = f"{frame.id}:subgraph:{step.id}"
+    lineage_id = f"{scope_id}:root"
+    child_frame_id = f"{scope_id}:frame"
+    if scope_id in run.scopes or lineage_id in run.lineages:
+        raise WorkflowExecutionError(
+            f"duplicate subgraph activation identifiers for step {step.id!r}"
+        )
+    run.scopes[scope_id] = RuntimeScope(
+        id=scope_id,
+        workflow_name=prepared.workflow.name,
+        workflow_input=dict(child_input),
+        committed_state=initial_state(prepared.workflow, child_input),
+        workflow_ref=step.workflow,
+    )
+    run.lineages[lineage_id] = LineageState(id=lineage_id, scope_id=scope_id)
+    add_frame(
+        run,
+        ExecutionFrame(
+            id=child_frame_id,
+            kind="subgraph_root",
+            node_id=prepared.workflow.start,
+            status=FrameStatus.PENDING,
+            parent_frame_id=frame.id,
+            scope_id=scope_id,
+            lineage_id=lineage_id,
+        ),
+        ready=True,
+    )
+    SubgraphActivation(
+        workflow_ref=step.workflow,
+        scope_id=scope_id,
+        lineage_id=lineage_id,
+        child_frame_id=child_frame_id,
+        child_input=child_input,
+    ).save_to_frame(frame)
+    block_frame_on_children(run, frame.id, (child_frame_id,))
+
+
+def _finish_subgraph(
+    workflow: Workflow,
+    run: RunState,
+    frame: ExecutionFrame,
+    step: SubgraphNode,
+    activation: SubgraphActivation,
+    prepared: PreparedSubgraph[HandlerT],
+    reducers: Mapping[str, ReducerDefinition] | None,
+) -> StepExecutionResult:
+    child_frame = run.frames.get(activation.child_frame_id)
+    if child_frame is None or child_frame.status != FrameStatus.COMPLETED:
+        raise WorkflowExecutionError(
+            f"subgraph step {step.id!r} resumed before its child completed"
+        )
+    child_scope = run.scopes.get(activation.scope_id)
+    if child_scope is None:
+        raise WorkflowExecutionError(
+            f"subgraph step {step.id!r} is missing child scope {activation.scope_id!r}"
+        )
+    child_outcome = child_frame.metadata.get("workflow_outcome")
+    if not isinstance(child_outcome, str):
+        raise WorkflowExecutionError(
+            f"subgraph step {step.id!r} child completed without a workflow outcome"
+        )
+    child_output = project_output(prepared.workflow, child_scope.committed_state)
+    validate_payload_against_schema(
+        prepared.workflow.output_schema,
+        child_output,
+        f"child workflow output for {step.id}",
+    )
+    validate_payload_against_schema(
+        step.output_schema, child_output, f"subgraph output for {step.id}"
+    )
+    patch = build_output_patch(
+        workflow,
+        step.output,
+        child_output,
+        state_view_for_frame(run, frame),
+        reducers=reducers,
+        missing_field_message="subgraph output did not include required field {field}",
+    )
+    state_changes = commit_patch_for_frame(run, frame, patch)
+    return StepExecutionResult(
+        outcome=child_outcome,
+        resolved_input=activation.child_input,
+        output=child_output,
+        state_changes=state_changes,
+    )