workflow ref

docs/authoring_sketch.md

@@ -254,8 +254,10 @@ Today there are two different authoring paths:
   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
+  copied into the boundary. The workflow reference is structural: local
-  subgraph scopes are implemented.
+  workflows use `{"name": "child"}`, saved artifacts can use
+  `{"artifact_id": "child", "version": 1}`. Runtime execution still raises
+  until native subgraph scopes are implemented.
 - `WorkflowBuilder.subgraph(...)` is the builder-facing version of
   `subgraph_ref`: it appends the native boundary step and returns it as a
   `StepRef` for `connect()` / `set_entry_point()`.

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

@@ -64,8 +64,11 @@ class SubgraphNode(BaseModel):
 ```
 
 Current implementation status: `wf_core` has a first placeholder
-`SubgraphNode`, but `workflow` is still a plain string reference. The placeholder
+`SubgraphNode`. Its `workflow` field is a structural `WorkflowRef`: local
-also carries `input_schema` and `output_schema` so validation can check parent
+compiled workflows use `{"name": "child"}`, while saved artifacts can use
+`{"artifact_id": "child", "version": 1}`. Legacy strings still parse as input,
+but saved graphs should persist the structural shape. The placeholder also
+carries `input_schema` and `output_schema` so validation can check parent
 bindings before native execution exists. Runtime execution intentionally raises
 until a later slice adds child scope/frame execution.
 
@@ -316,10 +319,10 @@ child = parent.subgraph(
 
 This copies the compiled child workflow contract into a core `SubgraphNode`,
 appends it to the builder, and returns the step for normal routing. It does not
-make the child executable yet. `workflow` is still a string reference inside the
+make the child executable yet. The core `workflow` field is structural, but
-core model; higher layers need a structural workflow reference before
+higher layers still need dependency resolution before saved/deployed workflow
-saved/deployed workflow dependencies become stable. The lower-level
+refs can run. The lower-level `subgraph_ref(...)` helper exists for code that
-`subgraph_ref(...)` helper exists for code that wants only the core step object.
+wants only the core step object.
 
 Possible API:
 

src/wf_authoring/builder/core.py

@@ -19,6 +19,7 @@ from wf_core import (
     StateSchema,
     SubgraphNode,
     Workflow,
+    WorkflowRef,
     RunState,
     execute_workflow,
 )
@@ -34,6 +35,7 @@ from wf_core.models.steps import (
 )
 from wf_core.paths import GraphSourcePath, LocalPath, StatePath
 from wf_core.runtime.ops.merges import ReducerDefinition
+from wf_platform import CapabilityRef
 
 from ..dsl import Expr, GraphPath, PathArg, PathExpr, compile_condition
 from ..nodes.callables import SyncRegistryHandler
@@ -77,6 +79,30 @@ def _condition_base(condition: CoreCondition) -> str:
     return "condition"
 
 
+def _workflow_ref_base(
+    workflow_ref: WorkflowRef | str | Mapping[str, object] | None,
+    workflow: Workflow,
+) -> str:
+    """Return an auto-id base for a subgraph ref without parsing display names."""
+    if isinstance(workflow_ref, WorkflowRef):
+        return slug_id(workflow_ref.name or workflow_ref.artifact_id or workflow.name)
+    if isinstance(workflow_ref, Mapping):
+        raw_name = workflow_ref.get("name") or workflow_ref.get("artifact_id")
+        return slug_id(str(raw_name or workflow.name))
+    return slug_id(workflow_ref or workflow.name)
+
+
+def _capability_ref_name(name: str | CapabilityRef) -> str:
+    """Return the current core NodeUse string for an authored capability ref.
+
+    `NodeUse.node` is still a string boundary because node definitions and
+    registries are keyed by string today. Accepting `CapabilityRef` here keeps
+    authoring code structural while confining display-name conversion to this
+    compatibility seam.
+    """
+    return str(name)
+
+
 def _canonical_input_bindings(
     in_map: Mapping[GraphSourcePath, LocalPath],
     input_values: Mapping[LocalPath, Any],
@@ -281,7 +307,7 @@ class WorkflowBuilder:
     @overload
     def use_ref(
         self,
-        name: str,
+        name: str | CapabilityRef,
         *,
         id: str | None = None,
         input: Sequence[InputBindingArg] | None = None,
@@ -293,7 +319,7 @@ class WorkflowBuilder:
     @deprecated("use input/output canonical binding lists instead")
     def use_ref(
         self,
-        name: str,
+        name: str | CapabilityRef,
         *,
         id: str | None = None,
         in_map: MapArg | None = None,
@@ -304,7 +330,7 @@ class WorkflowBuilder:
 
     def use_ref(
         self,
-        name: str,
+        name: str | CapabilityRef,
         *,
         id: str | None = None,
         input: Sequence[InputBindingArg] | None = None,
@@ -346,10 +372,11 @@ class WorkflowBuilder:
             if output is not None
             else _canonical_output_bindings(normalize_output_mapping(out_map))
         )
+        node_name = _capability_ref_name(name)
         node = NodeUse(
-            id=id or self._next_step_id(slug_id(name)),
+            id=id or self._next_step_id(slug_id(node_name)),
             type="node",
-            node=name,
+            node=node_name,
             desc=desc,
             input=node_input,
             output=node_output,
@@ -364,7 +391,7 @@ class WorkflowBuilder:
         id: str | None = None,
         input: Sequence[InputBindingArg] | None = None,
         output: Sequence[OutputBindingArg] | None = None,
-        workflow_ref: str | None = None,
+        workflow_ref: WorkflowRef | str | Mapping[str, object] | None = None,
         desc: str | None = None,
     ) -> SubgraphNode:
         """Add a native subgraph boundary using a compiled child workflow contract.
@@ -373,7 +400,7 @@ class WorkflowBuilder:
         until wf_core grows child workflow scope/frame execution.
         """
         node = subgraph_ref(
-            id=id or self._next_step_id(slug_id(workflow_ref or workflow.name)),
+            id=id or self._next_step_id(_workflow_ref_base(workflow_ref, workflow)),
             workflow=workflow,
             input=normalize_input_bindings(input),
             output=normalize_output_bindings(output),

src/wf_authoring/subgraph.py

@@ -9,8 +9,10 @@ from wf_core import (
     RuntimeContext,
     SubgraphNode,
     Workflow,
+    WorkflowRef,
     execute_workflow,
     execute_workflow_async,
+    workflow_ref_from,
 )
 from wf_core.models.steps import InputBinding, OutputBinding
 
@@ -26,7 +28,7 @@ def subgraph_ref(
     workflow: Workflow,
     input: list[InputBinding] | None = None,
     output: list[OutputBinding] | None = None,
-    workflow_ref: str | None = None,
+    workflow_ref: WorkflowRef | str | Mapping[str, object] | None = None,
     desc: str | None = None,
 ) -> SubgraphNode:
     """Create a native subgraph boundary from a compiled child workflow contract.
@@ -39,7 +41,7 @@ def subgraph_ref(
     return SubgraphNode(
         id=id,
         type="subgraph",
-        workflow=workflow_ref or workflow.name,
+        workflow=workflow_ref_from(workflow_ref or {"name": workflow.name}),
         desc=desc,
         input_schema=workflow.input_schema,
         output_schema=workflow.output_schema,

src/wf_core/__init__.py

@@ -18,6 +18,8 @@ from .models import (
     StateSchema,
     SubgraphNode,
     Workflow,
+    WorkflowRef,
+    workflow_ref_from,
 )
 from .runtime import (
     AsyncNodeHandler,
@@ -84,6 +86,7 @@ __all__ = [
     "ValidationIssueCode",
     "ValidationReport",
     "Workflow",
+    "WorkflowRef",
     "WorkflowExecutionError",
     "coerce_node_result",
     "execute_workflow_async",
@@ -93,4 +96,5 @@ __all__ = [
     "step_workflow_async",
     "step_workflow",
     "validate_workflow",
+    "workflow_ref_from",
 ]

src/wf_core/models/__init__.py

@@ -24,6 +24,7 @@ from wf_core.models.steps import (
     SubgraphNode,
 )
 from wf_core.models.workflow import Edge, Workflow
+from wf_core.models.workflow_refs import WorkflowRef, workflow_ref_from
 
 __all__ = [
     "BinaryCondition",
@@ -54,4 +55,6 @@ __all__ = [
     "SubgraphNode",
     "VariadicCondition",
     "Workflow",
+    "WorkflowRef",
+    "workflow_ref_from",
 ]

src/wf_core/models/steps.py

@@ -7,6 +7,7 @@ from pydantic import BaseModel, ConfigDict, Field, model_validator
 
 from wf_core.models.conditions import Condition
 from wf_core.models.schemas import SchemaRef
+from wf_core.models.workflow_refs import WorkflowRef
 from wf_core.paths import GraphSourcePath, LocalPath, StatePath
 
 
@@ -167,7 +168,7 @@ class SubgraphNode(BaseModel):
 
     id: str
     type: Literal["subgraph"]
-    workflow: str = Field(
+    workflow: WorkflowRef = Field(
         description=(
             "Reference to the child workflow artifact or registry key. The core "
             "does not resolve this reference until native subgraph runtime "

src/wf_core/models/workflow_refs.py

@@ -0,0 +1,91 @@
+from __future__ import annotations
+
+from collections.abc import Mapping
+from typing import Self
+
+from pydantic import BaseModel, ConfigDict, Field, model_serializer, model_validator
+
+
+class WorkflowRef(BaseModel):
+    """Structural reference to a child workflow contract.
+
+    Core needs a workflow reference for graph shape and validation, but it must
+    not depend on artifact stores or MCP capability naming. Saved workflow
+    artifacts can use ``artifact_id`` plus ``version``; local compiled workflows
+    can use ``name``. Legacy strings parse as ``name`` unless they use the old
+    ``workflow.<artifact>.v<version>`` display format.
+    """
+
+    model_config = ConfigDict(extra="forbid")
+
+    name: str | None = Field(default=None, description="Local workflow registry name.")
+    artifact_id: str | None = Field(
+        default=None,
+        description="Saved workflow artifact id, when referencing an immutable artifact.",
+    )
+    version: int | None = Field(
+        default=None,
+        ge=1,
+        description="Saved workflow artifact version. Requires artifact_id.",
+    )
+
+    @model_validator(mode="before")
+    @classmethod
+    def _coerce_legacy_strings(cls, value: object) -> object:
+        if not isinstance(value, str):
+            return value
+        try:
+            artifact_id, version = _parse_legacy_workflow_capability(value)
+        except ValueError:
+            return {"name": value}
+        return {"artifact_id": artifact_id, "version": version}
+
+    @model_validator(mode="after")
+    def _validate_one_ref_kind(self) -> Self:
+        has_name = self.name is not None
+        has_artifact = self.artifact_id is not None or self.version is not None
+        if has_name == has_artifact:
+            raise ValueError(
+                "workflow ref requires exactly one of name or artifact_id/version"
+            )
+        if has_artifact and (self.artifact_id is None or self.version is None):
+            raise ValueError(
+                "workflow artifact ref requires both artifact_id and version"
+            )
+        if self.name is not None and not self.name.strip():
+            raise ValueError("workflow ref name must not be empty")
+        if self.artifact_id is not None and not self.artifact_id.strip():
+            raise ValueError("workflow artifact id must not be empty")
+        return self
+
+    @property
+    def display(self) -> str:
+        """Return a human-readable compatibility name; do not parse this for meaning."""
+        if self.name is not None:
+            return self.name
+        return f"workflow.{self.artifact_id}.v{self.version}"
+
+    @model_serializer(mode="wrap")
+    def _serialize_without_none_fields(self, handler: object) -> dict[str, object]:
+        """Persist only the active ref shape."""
+        if not callable(handler):
+            raise TypeError("workflow ref serializer handler must be callable")
+        data = handler(self)
+        if not isinstance(data, dict):
+            raise TypeError("workflow ref serializer expected a dict")
+        return {key: value for key, value in data.items() if value is not None}
+
+
+def workflow_ref_from(value: WorkflowRef | str | Mapping[str, object]) -> WorkflowRef:
+    """Normalize public helper inputs into the core workflow ref model."""
+    return WorkflowRef.model_validate(value)
+
+
+def _parse_legacy_workflow_capability(value: str) -> tuple[str, int]:
+    prefix = "workflow."
+    if not value.startswith(prefix):
+        raise ValueError("not a workflow artifact display ref")
+    artifact_part, separator, version_part = value[len(prefix) :].rpartition(".v")
+    if not separator or not artifact_part or not version_part.isdecimal():
+        raise ValueError("not a workflow artifact display ref")
+    return artifact_part, int(version_part)

tests/authoring/test_builder.py

@@ -17,6 +17,7 @@ from wf_authoring import (
 from wf_core import END, RunStatus, WorkflowExecutionError
 from wf_core.models.steps import InputPathBinding, InputValueBinding
 from wf_core.paths import GraphSourcePath, LocalPath, StatePath
+from wf_platform import CapabilityRef
 
 from tests.authoring.helpers import (
     AutoBindInput,
@@ -376,6 +377,23 @@ def test_builder_use_ref_accepts_canonical_binding_dicts() -> None:
     assert step.output[0].target == StatePath.of("echoed")
 
 
+def test_builder_use_ref_accepts_structural_capability_ref() -> None:
+    builder = WorkflowBuilder(
+        name="external_ref_structural",
+        input_schema={},
+        state_schema={"fields": {}},
+        output_schema={},
+    )
+
+    step = builder.use_ref(
+        CapabilityRef.parse("demo.personal.echo"),
+        input=[input_from(input_path("text"), "text")],
+    )
+
+    assert step.node == "demo.personal.echo"
+    assert step.id == "demo_personal_echo"
+
+
 def test_builder_warns_when_explicit_deprecated_maps_are_used() -> None:
     builder = WorkflowBuilder(
         name="deprecated_maps",

tests/authoring/test_subgraph.py

@@ -163,7 +163,7 @@ def test_subgraph_ref_copies_child_workflow_contract() -> None:
     )
 
     assert step.type == "subgraph"
-    assert step.workflow == child.name
+    assert step.workflow.name == child.name
     assert step.input_schema == child.input_schema
     assert step.output_schema == child.output_schema
     assert step.outcomes == child.outcomes
@@ -202,6 +202,21 @@ def test_subgraph_ref_contract_validates_in_parent_workflow() -> None:
     assert parent.validate_structure().errors == []
 
 
+def test_subgraph_ref_accepts_structural_saved_workflow_reference() -> None:
+    child = build_demo_workflow()
+
+    step = subgraph_ref(
+        id="run_child",
+        workflow=child,
+        workflow_ref={"artifact_id": "demo_child", "version": 2},
+    )
+
+    dumped = step.model_dump(mode="json")
+    assert dumped["workflow"]["artifact_id"] == "demo_child"
+    assert dumped["workflow"]["version"] == 2
+    assert "name" not in dumped["workflow"]
+
+
 def test_workflow_builder_subgraph_adds_native_subgraph_node() -> None:
     class ParentInput(BaseModel):
         folder_id: str

tests/refs/test_workflow_refs.py

@@ -1,8 +1,14 @@
 from __future__ import annotations
 
-from pydantic import BaseModel
+import pytest
+from pydantic import BaseModel, ValidationError
 
 from wf_artifacts import WorkflowCapabilityRef
+from wf_core import WorkflowRef
+
+
+class CoreWorkflowRefPayload(BaseModel):
+    ref: WorkflowRef
 
 
 def test_workflow_capability_ref_round_trips() -> None:
@@ -60,3 +66,38 @@ def test_workflow_capability_ref_rejects_other_namespaces() -> None:
         assert "workflow" in str(exc)
     else:
         raise AssertionError("expected non-workflow ref to be rejected")
+
+
+def test_core_workflow_ref_accepts_local_name_string() -> None:
+    payload = CoreWorkflowRefPayload.model_validate({"ref": "child_workflow"})
+
+    assert payload.ref.name == "child_workflow"
+    assert payload.ref.artifact_id is None
+    assert payload.model_dump(mode="json")["ref"]["name"] == "child_workflow"
+
+
+def test_core_workflow_ref_accepts_legacy_saved_artifact_display_string() -> None:
+    payload = CoreWorkflowRefPayload.model_validate({"ref": "workflow.echo.wrapper.v2"})
+
+    assert payload.ref.name is None
+    assert payload.ref.artifact_id == "echo.wrapper"
+    assert payload.ref.version == 2
+    assert payload.model_dump(mode="json")["ref"]["artifact_id"] == "echo.wrapper"
+    assert payload.model_dump(mode="json")["ref"]["version"] == 2
+
+
+def test_core_workflow_ref_accepts_structural_saved_artifact_ref() -> None:
+    payload = CoreWorkflowRefPayload.model_validate(
+        {"ref": {"artifact_id": "echo_wrapper", "version": 2}}
+    )
+
+    assert payload.ref.artifact_id == "echo_wrapper"
+    assert payload.ref.version == 2
+    assert payload.ref.display == "workflow.echo_wrapper.v2"
+
+
+def test_core_workflow_ref_rejects_mixed_name_and_artifact_ref() -> None:
+    with pytest.raises(ValidationError, match="exactly one"):
+        CoreWorkflowRefPayload.model_validate(
+            {"ref": {"name": "child", "artifact_id": "echo", "version": 1}}
+        )