add the async twin of Ts

docs/wf_core_architecture.md

@@ -136,9 +136,11 @@ limits and intended adapter seam.
   includes node, condition, foreach, join, and interrupt steps; `Workflow` does
   not contain nested workflow/subgraph steps.
 - Nested subgraph interruption is not first-class yet. The current
-  `wf_authoring` subgraph helper wraps a child workflow as an ordinary node and
-  validates the child output; it does not preserve a child run state that can
-  interrupt, bubble to the parent, and later resume inside the child.
+  `wf_authoring` subgraph helpers wrap a child workflow as an ordinary sync or
+  async node and validate the child output; they do not preserve a child run
+  state that can interrupt, bubble to the parent, and later resume inside the
+  child. See `examples/authoring_workflow_as_node.py` for the current
+  wrapper-node shape.
 - Saved workflow-as-node execution with interrupts requires a core runtime
   upgrade: nested run state, child-frame trace preservation, interrupt bubbling
   with path metadata, and resume back into the child workflow.

docs/workflow_artifacts.md

@@ -637,11 +637,17 @@ diagnostics before attempting persistent nested resume.
 
 Native subgraphs are not in `wf_core` yet. The current core `Step` model has
 node, condition, foreach, join, and interrupt steps, but no subgraph step. The
-current `wf_authoring.subgraph_node` helper executes a child workflow as a plain
-node and validates the child output. Future saved-workflow-as-node execution
+current `wf_authoring.subgraph_node` and `async_subgraph_node` helpers execute
+a child workflow as a plain node and validate the child output. The async helper
+is explicit because hiding `asyncio.run()` inside the sync wrapper would break
+inside already-running event loops. Future saved-workflow-as-node execution
 needs a real child run state if child interrupts should pause the parent and
 later resume the child.
 
+See `examples/authoring_workflow_as_node.py` for the current wrapper-node
+approach. In that example the parent trace sees one node call; the child
+workflow's internal trace is not embedded 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
 workflows with interrupt nodes are valid, but nested interrupting workflows

examples/authoring_workflow_as_node.py

@@ -0,0 +1,112 @@
+from __future__ import annotations
+
+import sys
+from pathlib import Path
+
+from pydantic import BaseModel
+
+# Keep this example runnable as either `python -m examples...` or a direct file.
+if __package__ in {None, ""}:
+    sys.path.insert(0, str(Path(__file__).resolve().parents[1]))
+
+from wf_authoring import (
+    WorkflowBuilder,
+    input_from,
+    output_to,
+    state_path,
+    subgraph_node,
+)
+from wf_core import END
+from wf_core.run_state import RunState
+
+from examples.demo_workflow import build_demo_registry, build_demo_workflow
+
+
+class ChildInput(BaseModel):
+    """Input accepted by the wrapped child workflow."""
+
+    folder_id: str
+    should_email: bool
+
+
+class ChildOutput(BaseModel):
+    """Output exposed by the wrapped child workflow."""
+
+    summary: str
+    email_status: str
+
+
+class ParentInput(BaseModel):
+    """Input accepted by the parent workflow."""
+
+    folder_id: str
+    should_email: bool
+
+
+class ParentState(BaseModel):
+    """Parent state stores only the child workflow output fields it needs."""
+
+    summary: str = ""
+    email_status: str = ""
+
+
+class ParentOutput(BaseModel):
+    """Parent output projected from state after the wrapped child completes."""
+
+    summary: str
+    email_status: str
+
+
+wrapped_demo_workflow = subgraph_node(
+    name="example.wrapped_demo_workflow",
+    workflow=build_demo_workflow(),
+    registry=build_demo_registry(),
+    input_model=ChildInput,
+    output_model=ChildOutput,
+    description=(
+        "Runs the demo child workflow as one parent node. Current wrapper "
+        "semantics do not embed the child trace or support child interrupts."
+    ),
+)
+
+
+def build_parent_workflow() -> WorkflowBuilder:
+    """Build a parent workflow that treats a whole child workflow as one node."""
+    graph = WorkflowBuilder(
+        name="workflow_as_node_parent",
+        input_schema=ParentInput,
+        state_schema=ParentState,
+        output_schema=ParentOutput,
+    )
+    child = graph.use(
+        wrapped_demo_workflow,
+        id="run_child",
+        input=[
+            input_from("input.folder_id", "folder_id"),
+            input_from("input.should_email", "should_email"),
+        ],
+        output=[
+            output_to("summary", state_path("summary")),
+            output_to("email_status", state_path("email_status")),
+        ],
+    )
+    graph.set_entry_point(child)
+    graph.connect(child, "ok", END)
+    return graph
+
+
+def run_parent_workflow() -> RunState:
+    """Run the parent workflow around the wrapped child workflow."""
+    return build_parent_workflow().execute(
+        {"folder_id": "demo-folder", "should_email": False}
+    )
+
+
+def main() -> None:
+    """Run the example directly from the command line."""
+    run = run_parent_workflow()
+    print(run.status.value, run.output)
+
+
+if __name__ == "__main__":
+    main()

src/wf_authoring/__init__.py

@@ -66,12 +66,13 @@ from .nodes import (
 )
 from .reducers import AuthoredReducer, ReducerCatalog, reducer
 from .schemas import StateFieldMetadata, state_field
-from .subgraph import subgraph_node
+from .subgraph import async_subgraph_node, subgraph_node
 
 __all__ = [
     "NodeCatalog",
     "NodeCatalogEntry",
     "AuthoredReducer",
+    "async_subgraph_node",
     "BoolOutput",
     "CoalesceInput",
     "ConstantInput",

src/wf_authoring/subgraph.py

@@ -5,7 +5,7 @@ from typing import Any, TypeVar
 
 from pydantic import BaseModel
 
-from wf_core import RuntimeContext, Workflow, execute_workflow
+from wf_core import RuntimeContext, Workflow, execute_workflow, execute_workflow_async
 
 from .nodes import NodeSpec
 
@@ -22,6 +22,14 @@ def subgraph_node(
     output_model: type[OutputT],
     description: str | None = None,
 ) -> NodeSpec[InputT, OutputT]:
+    """Wrap a compiled workflow as a sync authoring node.
+
+    This is deliberately only a wrapper-node compatibility helper: the parent
+    trace sees one node call, and child interrupts/frames are not promoted into
+    native parent subgraph state. Use `async_subgraph_node` when the wrapped
+    workflow registry contains async handlers.
+    """
+
     def run_subgraph(payload: InputT, ctx: RuntimeContext) -> OutputT:
         child_run = execute_workflow(
             workflow,
@@ -39,3 +47,38 @@ def subgraph_node(
         description=description or f"Subgraph wrapper for {workflow.name}",
         is_async=False,
     )
+
+
+def async_subgraph_node(
+    *,
+    name: str,
+    workflow: Workflow,
+    registry: Mapping[str, Any],
+    input_model: type[InputT],
+    output_model: type[OutputT],
+    description: str | None = None,
+) -> NodeSpec[InputT, OutputT]:
+    """Wrap a compiled workflow with async handlers as an async authoring node.
+
+    This keeps async explicit instead of calling `asyncio.run()` from the sync
+    wrapper, which would break inside already-running event loops. It is still
+    wrapper-node composition, not native subgraph execution.
+    """
+
+    async def run_subgraph(payload: InputT, ctx: RuntimeContext) -> OutputT:
+        child_run = await execute_workflow_async(
+            workflow,
+            payload.model_dump(),
+            registry,
+        )
+        return output_model.model_validate(child_run.output)
+
+    return NodeSpec(
+        name=name,
+        input_model=input_model,
+        output_model=output_model,
+        outcomes=("ok",),
+        fn=run_subgraph,
+        description=description or f"Async subgraph wrapper for {workflow.name}",
+        is_async=True,
+    )

tests/authoring/test_subgraph.py

@@ -1,10 +1,28 @@
 from __future__ import annotations
 
+import asyncio
+
 from pydantic import BaseModel
 
-from wf_authoring import build_registry, subgraph_node
-from wf_core import RuntimeContext
+from wf_authoring import (
+    WorkflowBuilder,
+    async_subgraph_node,
+    build_async_registry,
+    build_registry,
+    input_from,
+    input_path,
+    node,
+    output_to,
+    state_path,
+    subgraph_node,
+)
+from wf_core import END, RunStatus, RuntimeContext, execute_workflow_async
 from examples.demo_workflow import build_demo_registry, build_demo_workflow
+from examples.authoring_workflow_as_node import (
+    build_parent_workflow,
+    run_parent_workflow,
+    wrapped_demo_workflow,
+)
 
 
 def test_subgraph_node_wraps_compiled_workflow() -> None:
@@ -35,3 +53,92 @@ def test_subgraph_node_wraps_compiled_workflow() -> None:
     assert result["outcome"] == "ok"
     assert result["output"]["email_status"] == "skipped"
     assert "summary" in result["output"]
+
+
+def test_async_subgraph_node_wraps_async_compiled_workflow() -> None:
+    class ChildInput(BaseModel):
+        text: str
+
+    class ChildState(BaseModel):
+        text: str
+
+    class ChildOutput(BaseModel):
+        text: str
+
+    @node(name="child.async_upper", input_model=ChildInput, output_model=ChildOutput)
+    async def async_upper(payload: ChildInput) -> ChildOutput:
+        return ChildOutput(text=payload.text.upper())
+
+    child = WorkflowBuilder(
+        name="async_child",
+        input_schema=ChildInput,
+        state_schema=ChildState,
+        output_schema=ChildOutput,
+    )
+    upper = child.use(
+        async_upper,
+        id="upper",
+        input=[input_from(input_path("text"), "text")],
+        output=[output_to("text", state_path("text"))],
+    )
+    child.set_entry_point(upper)
+    child.connect(upper, "ok", END)
+
+    wrapped = async_subgraph_node(
+        name="wrapped_async_child",
+        workflow=child.compile(),
+        registry=build_async_registry(async_upper),
+        input_model=ChildInput,
+        output_model=ChildOutput,
+    )
+    parent = WorkflowBuilder(
+        name="async_parent",
+        input_schema=ChildInput,
+        state_schema=ChildState,
+        output_schema=ChildOutput,
+    )
+    step = parent.use(
+        wrapped,
+        id="run_async_child",
+        input=[input_from(input_path("text"), "text")],
+        output=[output_to("text", state_path("text"))],
+    )
+    parent.set_entry_point(step)
+    parent.connect(step, "ok", END)
+
+    run = asyncio.run(
+        execute_workflow_async(
+            parent.compile(),
+            {"text": "hello"},
+            build_async_registry(wrapped),
+        )
+    )
+
+    assert run.status == RunStatus.COMPLETED
+    assert run.output["text"] == "HELLO"
+
+
+def test_workflow_as_node_example_runs_child_inside_parent_workflow() -> None:
+    run = run_parent_workflow()
+
+    assert run.status == RunStatus.COMPLETED
+    assert run.output["email_status"] == "skipped"
+    assert "Summary of demo-folder/meeting-notes.md" in run.output["summary"]
+
+
+def test_workflow_as_node_example_parent_trace_is_one_node_call() -> None:
+    run = run_parent_workflow()
+
+    assert len(run.trace) == 1
+    assert run.trace[0].node_id == "run_child"
+    assert run.trace[0].step_type == "node"
+    assert run.trace[0].outcome == "ok"
+
+
+def test_workflow_as_node_example_compiles_to_normal_node_use() -> None:
+    workflow = build_parent_workflow().compile()
+    node = workflow.nodes[0]
+
+    assert node.type == "node"
+    assert node.node == wrapped_demo_workflow.name
+    assert workflow.node_defs[0].name == wrapped_demo_workflow.name