subgraph support in wf_authoring

docs/authoring_sketch.md

@@ -247,7 +247,17 @@ it.
 
 ### `Workflow -> NodeSpec`
 
-In the future, a compiled workflow or subgraph can be wrapped as a reusable
+Today there are two different authoring paths:
+
+- `subgraph_node` / `async_subgraph_node` wrap a compiled workflow as a normal
+  `NodeSpec`. The parent sees one node call and child frames/interrupts are not
+  native parent state.
+- `subgraph_ref` builds a native `SubgraphNode` contract from a compiled
+  `Workflow`: child input schema, output schema, and workflow outcomes are
+  copied into the boundary. Runtime execution still raises until native
+  subgraph scopes are implemented.
+
+In the future, a compiled workflow or subgraph can also be exposed as a reusable
 `NodeSpec`, likely by treating workflow input and output schemas as the node's
 input and output schemas.
 

docs/superpowers/specs/2026-05-24-native-subgraphs-design.md

@@ -303,6 +303,22 @@ resolved dependency set.
 `wf_authoring` should expose native subgraph use separately from wrapper-node
 composition.
 
+Current helper:
+
+```python
+child = subgraph_ref(
+    id="run_child",
+    workflow=child_builder.compile(),
+    input=[input_from(state_path("request"), "request")],
+    output=[output_to("summary", state_path("child_summary"))],
+)
+```
+
+This copies the compiled child workflow contract into a core `SubgraphNode` but
+does not make the child executable yet. `workflow` is still a string reference
+inside the core model; higher layers need a structural workflow reference before
+saved/deployed workflow dependencies become stable.
+
 Possible API:
 
 ```python

src/wf_authoring/__init__.py

@@ -66,7 +66,7 @@ from .nodes import (
 )
 from .reducers import AuthoredReducer, ReducerCatalog, reducer
 from .schemas import StateFieldMetadata, state_field
-from .subgraph import async_subgraph_node, subgraph_node
+from .subgraph import async_subgraph_node, subgraph_node, subgraph_ref
 
 __all__ = [
     "NodeCatalog",
@@ -136,5 +136,6 @@ __all__ = [
     "state_field",
     "state_path",
     "subgraph_node",
+    "subgraph_ref",
     "truthy",
 ]

src/wf_authoring/builder/refs.py

@@ -3,12 +3,27 @@ from __future__ import annotations
 from dataclasses import dataclass
 from typing import Any, TypeAlias, TypeGuard
 
-from wf_core import ConditionNode, ForeachNode, InterruptNode, JoinNode, NodeUse
+from wf_core import (
+    ConditionNode,
+    EndNode,
+    ForeachNode,
+    InterruptNode,
+    JoinNode,
+    NodeUse,
+    SubgraphNode,
+)
 
 from ..nodes import NodeSpec
 
 StepRef: TypeAlias = (
-    str | NodeUse | ConditionNode | ForeachNode | InterruptNode | JoinNode
+    str
+    | NodeUse
+    | SubgraphNode
+    | ConditionNode
+    | ForeachNode
+    | InterruptNode
+    | JoinNode
+    | EndNode
 )
 """A reference to a step, which can be either a string id or a node object
  that should be auto-used."""

src/wf_authoring/subgraph.py

@@ -5,7 +5,14 @@ from typing import Any, TypeVar
 
 from pydantic import BaseModel
 
-from wf_core import RuntimeContext, Workflow, execute_workflow, execute_workflow_async
+from wf_core import (
+    RuntimeContext,
+    SubgraphNode,
+    Workflow,
+    execute_workflow,
+    execute_workflow_async,
+)
+from wf_core.models.steps import InputBinding, OutputBinding
 
 from .nodes import NodeSpec
 
@@ -13,6 +20,35 @@ InputT = TypeVar("InputT", bound=BaseModel)
 OutputT = TypeVar("OutputT", bound=BaseModel)
 
 
+def subgraph_ref(
+    *,
+    id: str,
+    workflow: Workflow,
+    input: list[InputBinding] | None = None,
+    output: list[OutputBinding] | None = None,
+    workflow_ref: str | None = None,
+    desc: str | None = None,
+) -> SubgraphNode:
+    """Create a native subgraph boundary from a compiled child workflow contract.
+
+    This does not make the child executable yet. It copies the child workflow's
+    public contract into `SubgraphNode` so parent graphs can validate mappings
+    and route child workflow outcomes before native child-scope runtime support
+    lands.
+    """
+    return SubgraphNode(
+        id=id,
+        type="subgraph",
+        workflow=workflow_ref or workflow.name,
+        desc=desc,
+        input_schema=workflow.input_schema,
+        output_schema=workflow.output_schema,
+        input=input or [],
+        output=output or [],
+        outcomes=list(workflow.outcomes),
+    )
+
+
 def subgraph_node(
     *,
     name: str,

tests/authoring/test_subgraph.py

@@ -14,9 +14,17 @@ from wf_authoring import (
     node,
     output_to,
     state_path,
+    subgraph_ref,
     subgraph_node,
 )
-from wf_core import END, RunStatus, RuntimeContext, execute_workflow_async
+from wf_core import (
+    END,
+    Edge,
+    RunStatus,
+    RuntimeContext,
+    Workflow,
+    execute_workflow_async,
+)
 from examples.demo_workflow import build_demo_registry, build_demo_workflow
 from examples.authoring_workflow_as_node import (
     build_parent_workflow,
@@ -142,3 +150,53 @@ def test_workflow_as_node_example_compiles_to_normal_node_use() -> None:
     assert node.type == "node"
     assert node.node == wrapped_demo_workflow.name
     assert workflow.node_defs[0].name == wrapped_demo_workflow.name
+
+
+def test_subgraph_ref_copies_child_workflow_contract() -> None:
+    child = build_demo_workflow()
+
+    step = subgraph_ref(
+        id="run_child",
+        workflow=child,
+        input=[input_from(input_path("folder_id"), "folder_id")],
+        output=[output_to("summary", state_path("summary"))],
+    )
+
+    assert step.type == "subgraph"
+    assert step.workflow == child.name
+    assert step.input_schema == child.input_schema
+    assert step.output_schema == child.output_schema
+    assert step.outcomes == child.outcomes
+
+
+def test_subgraph_ref_contract_validates_in_parent_workflow() -> None:
+    child = build_demo_workflow()
+    step = subgraph_ref(
+        id="run_child",
+        workflow=child,
+        input=[input_from(input_path("folder_id"), "folder_id")],
+        output=[output_to("summary", state_path("summary"))],
+    )
+
+    parent = Workflow.model_validate(
+        {
+            "name": "native_subgraph_parent",
+            "input_schema": child.input_schema.model_dump(mode="json"),
+            "state_schema": {
+                "type": "object",
+                "properties": {"summary": {"type": "string"}},
+            },
+            "output_schema": {
+                "type": "object",
+                "properties": {"summary": {"type": "string"}},
+            },
+            "outcomes": ["ok"],
+            "start": "run_child",
+            "nodes": [step.model_dump(mode="json", by_alias=True)],
+            "edges": [
+                Edge.model_validate({"from": "run_child", "outcome": "ok", "to": END})
+            ],
+        }
+    )
+
+    assert parent.validate_structure().errors == []