workflow as node looking alright

tests/test_demo_workflow.py

@@ -19,12 +19,14 @@ from wf_authoring import (
     bind_fields,
     bind_state,
     build_registry,
+    expr,
     exists,
     state,
+    state_path,
     context_path,
     input_path,
     node,
-    state_path,
+    subgraph_node,
 )
 
 
@@ -487,7 +489,9 @@ def test_node_decorator_detects_async_automatically() -> None:
 
 
 def test_condition_dsl_compiles_to_core_condition() -> None:
-    condition = state("should_email").eq(True) & exists(state("summary"))
+    condition = expr(state_path("should_email")).eq(True) & exists(
+        state_path("summary")
+    )
 
     assert condition.to_condition().model_dump() == {
         "op": "and",
@@ -503,3 +507,33 @@ def test_condition_dsl_compiles_to_core_condition() -> None:
             },
         ],
     }
+
+
+def test_subgraph_node_wraps_compiled_workflow() -> None:
+    class ChildInput(BaseModel):
+        folder_id: str
+        should_email: bool
+
+    class ChildOutput(BaseModel):
+        summary: str
+        email_status: str
+
+    child_workflow = build_demo_workflow()
+    child_registry = build_demo_registry()
+    wrapped = subgraph_node(
+        name="wrapped_demo",
+        workflow=child_workflow,
+        registry=child_registry,
+        input_model=ChildInput,
+        output_model=ChildOutput,
+    )
+
+    registry = build_registry(wrapped)
+    result = registry["wrapped_demo"](
+        {"folder_id": "demo-folder", "should_email": False},
+        RuntimeContext(current_node_id="parent"),
+    )
+
+    assert result["outcome"] == "ok"
+    assert result["output"]["email_status"] == "skipped"
+    assert "summary" in result["output"]

wf_authoring/__init__.py

@@ -1,9 +1,10 @@
 from .builder import WorkflowBuilder
 from .catalog import NodeCatalog, NodeCatalogEntry
-from .conditions import context, exists, input, state
+from .conditions import context, exists, expr, input, state
 from .mapping import bind_fields, bind_state, merge_maps
 from .paths import GraphPath, context_path, graph_path, input_path, state_path
 from .spec import NodeReturn, NodeSpec, build_registry, node
+from .subgraph import subgraph_node
 
 __all__ = [
     "NodeCatalog",
@@ -18,6 +19,7 @@ __all__ = [
     "merge_maps",
     "context",
     "context_path",
+    "expr",
     "exists",
     "graph_path",
     "input",
@@ -25,4 +27,5 @@ __all__ = [
     "node",
     "state",
     "state_path",
+    "subgraph_node",
 ]

wf_authoring/conditions.py

@@ -12,14 +12,23 @@ from wf_core.model import (
     PathOperand,
     VariadicCondition,
 )
+from .paths import GraphPath, context_path, input_path, state_path
 
 
 def _operand(value: object) -> PathOperand | LiteralOperand:
     if isinstance(value, PathExpr):
         return PathOperand(path=value.path)
+    if isinstance(value, GraphPath):
+        return PathOperand(path=value.value)
     return LiteralOperand(value=value)
 
 
+def _path_str(value: PathExpr | GraphPath) -> str:
+    if isinstance(value, PathExpr):
+        return value.path
+    return value.value
+
+
 @dataclass(frozen=True, slots=True)
 class Expr:
     condition: Condition
@@ -79,20 +88,26 @@ class PathExpr:
         return self.lt(other)
 
 
+def expr(value: PathExpr | GraphPath) -> PathExpr:
+    if isinstance(value, PathExpr):
+        return value
+    return PathExpr(path=value.value)
+
+
 def state(field: str) -> PathExpr:
-    return PathExpr(path=f"state.{field}")
+    return expr(state_path(field))
 
 
 def input(field: str) -> PathExpr:
-    return PathExpr(path=f"input.{field}")
+    return expr(input_path(field))
 
 
 def context(field: str) -> PathExpr:
-    return PathExpr(path=f"context.{field}")
+    return expr(context_path(field))
 
 
-def exists(value: PathExpr) -> Expr:
-    return Expr(ExistsCondition(op="exists", path=value.path))
+def exists(value: PathExpr | GraphPath) -> Expr:
+    return Expr(ExistsCondition(op="exists", path=_path_str(value)))
 
 
 def compile_condition(value: Condition | Expr) -> Condition:

wf_authoring/subgraph.py

@@ -0,0 +1,41 @@
+from __future__ import annotations
+
+from collections.abc import Mapping
+from typing import Any, TypeVar
+
+from pydantic import BaseModel
+
+from wf_core import RuntimeContext, Workflow, execute_workflow
+
+from .spec import NodeSpec
+
+InputT = TypeVar("InputT", bound=BaseModel)
+OutputT = TypeVar("OutputT", bound=BaseModel)
+
+
+def 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]:
+    def run_subgraph(payload: InputT, ctx: RuntimeContext) -> OutputT:
+        child_run = execute_workflow(
+            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"Subgraph wrapper for {workflow.name}",
+        is_async=False,
+    )