conversion fns

docs/workflow_artifacts.md

@@ -632,6 +632,19 @@ Saved workflow artifact names use a separate grammar and ref type:
 `workflow.<artifact_id>.v<version>`. Artifact ids may contain dots, so this
 must not be parsed as a generic `CapabilityRef`.
 
+When an artifact is used as a child workflow dependency, convert it to the core
+`WorkflowRef` shape instead of reusing display strings:
+
+```python
+workflow_ref_from_artifact(artifact)
+workflow_ref_from_capability(WorkflowCapabilityRef("echo_wrapper", 1))
+workflow_capability_ref_from_workflow_ref(ref)  # only for artifact-backed refs
+```
+
+This keeps the three identities separate: `WorkflowArtifact` is the saved
+document, `WorkflowCapabilityRef` is the public callable capability name, and
+`WorkflowRef` is the core subgraph dependency pointer.
+
 The first implementation should prefer artifact validation and dependency
 diagnostics before attempting persistent nested resume.
 

src/wf_artifacts/__init__.py

@@ -32,7 +32,12 @@ from .models import (
     WorkflowArtifact,
     WorkflowDeployment,
 )
-from .refs import WorkflowCapabilityRef
+from .refs import (
+    WorkflowCapabilityRef,
+    workflow_capability_ref_from_workflow_ref,
+    workflow_ref_from_artifact,
+    workflow_ref_from_capability,
+)
 from .store import FileWorkflowArtifactStore, WorkflowArtifactStore
 from .validation import validate_deployment_dependencies
 from .references import logical_ref_for_concrete_ref, normalize_plan_node_refs
@@ -70,4 +75,7 @@ __all__ = [
     "summarize_draft_workspace",
     "validate_deployment_dependencies",
     "validate_workflow_draft",
+    "workflow_capability_ref_from_workflow_ref",
+    "workflow_ref_from_artifact",
+    "workflow_ref_from_capability",
 ]

src/wf_artifacts/refs.py

@@ -5,6 +5,10 @@ from typing import Any
 
 from pydantic_core import core_schema
 
+from wf_core import WorkflowRef
+
+from .models import WorkflowArtifact
+
 
 @dataclass(frozen=True, slots=True)
 class WorkflowCapabilityRef:
@@ -70,3 +74,22 @@ class WorkflowCapabilityRef:
     def _serialize(value: WorkflowCapabilityRef) -> dict[str, int | str]:
         """Serialize canonical saved workflow refs without a display-name parser."""
         return {"artifact_id": value.artifact_id, "version": value.version}
+
+
+def workflow_ref_from_artifact(artifact: WorkflowArtifact) -> WorkflowRef:
+    """Return the core child-workflow ref for one saved artifact version."""
+    return WorkflowRef(artifact_id=artifact.id, version=artifact.version)
+
+
+def workflow_ref_from_capability(ref: WorkflowCapabilityRef) -> WorkflowRef:
+    """Return the core child-workflow ref for a workflow capability identity."""
+    return WorkflowRef(artifact_id=ref.artifact_id, version=ref.version)
+
+
+def workflow_capability_ref_from_workflow_ref(
+    ref: WorkflowRef,
+) -> WorkflowCapabilityRef:
+    """Return the public capability ref for an artifact-backed workflow ref."""
+    if ref.artifact_id is None or ref.version is None:
+        raise ValueError("workflow capability refs require an artifact workflow ref")
+    return WorkflowCapabilityRef(artifact_id=ref.artifact_id, version=ref.version)

tests/refs/test_workflow_refs.py

@@ -3,7 +3,13 @@ from __future__ import annotations
 import pytest
 from pydantic import BaseModel, ValidationError
 
-from wf_artifacts import WorkflowCapabilityRef
+from wf_artifacts import (
+    WorkflowArtifact,
+    WorkflowCapabilityRef,
+    workflow_capability_ref_from_workflow_ref,
+    workflow_ref_from_artifact,
+    workflow_ref_from_capability,
+)
 from wf_core import WorkflowRef
 
 
@@ -101,3 +107,51 @@ def test_core_workflow_ref_rejects_mixed_name_and_artifact_ref() -> None:
         CoreWorkflowRefPayload.model_validate(
             {"ref": {"name": "child", "artifact_id": "echo", "version": 1}}
         )
+
+
+def test_workflow_ref_from_artifact_uses_artifact_identity() -> None:
+    ref = workflow_ref_from_artifact(_artifact("echo_wrapper", 2))
+
+    assert ref.artifact_id == "echo_wrapper"
+    assert ref.version == 2
+    assert ref.name is None
+
+
+def test_workflow_ref_from_capability_uses_saved_workflow_identity() -> None:
+    ref = workflow_ref_from_capability(
+        WorkflowCapabilityRef(artifact_id="echo_wrapper", version=2)
+    )
+
+    assert ref.artifact_id == "echo_wrapper"
+    assert ref.version == 2
+
+
+def test_workflow_capability_ref_from_workflow_ref_requires_artifact_ref() -> None:
+    capability = workflow_capability_ref_from_workflow_ref(
+        WorkflowRef(artifact_id="echo_wrapper", version=2)
+    )
+
+    assert capability == WorkflowCapabilityRef(artifact_id="echo_wrapper", version=2)
+
+    with pytest.raises(ValueError, match="artifact"):
+        workflow_capability_ref_from_workflow_ref(WorkflowRef(name="local_child"))
+
+
+def _artifact(artifact_id: str, version: int) -> WorkflowArtifact:
+    return WorkflowArtifact(
+        id=artifact_id,
+        version=version,
+        title="Echo Wrapper",
+        input_schema={"type": "object"},
+        output_schema={"type": "object"},
+        outcomes=("ok",),
+        plan={
+            "name": "echo_wrapper",
+            "input_schema": {"type": "object"},
+            "state_schema": {"type": "object", "properties": {}},
+            "output_schema": {"type": "object"},
+            "start": "end",
+            "nodes": [{"id": "end", "type": "end", "outcome": "ok"}],
+            "edges": [],
+        },
+    )