workflow capabilityref: more models

src/wf_artifacts/__init__.py

@@ -15,6 +15,7 @@ from .models import (
     WorkflowArtifact,
     WorkflowDeployment,
 )
+from .refs import WorkflowCapabilityRef
 from .store import FileWorkflowArtifactStore, WorkflowArtifactStore
 from .validation import validate_deployment_dependencies
 from .references import logical_ref_for_concrete_ref, normalize_plan_node_refs
@@ -30,6 +31,7 @@ __all__ = [
     "RequiredCapability",
     "WorkflowArtifact",
     "WorkflowArtifactCatalogEntry",
+    "WorkflowCapabilityRef",
     "WorkflowArtifactStore",
     "WorkflowDeployment",
     "artifact_catalog_entry",

src/wf_artifacts/catalog.py

@@ -3,6 +3,7 @@ from __future__ import annotations
 from pydantic import BaseModel, Field
 
 from .models import DependencyDiagnostic, JsonObject, WorkflowArtifact
+from .refs import WorkflowCapabilityRef
 
 
 class WorkflowArtifactCatalogEntry(BaseModel):
@@ -23,7 +24,12 @@ class WorkflowArtifactCatalogEntry(BaseModel):
 
 def artifact_node_name(artifact: WorkflowArtifact) -> str:
     """Return the stable planner name for an artifact version."""
-    return f"workflow.{artifact.id}.v{artifact.version}"
+    return str(
+        WorkflowCapabilityRef(
+            artifact_id=artifact.id,
+            version=artifact.version,
+        )
+    )
 
 
 def artifact_catalog_entry(

src/wf_artifacts/refs.py

@@ -0,0 +1,33 @@
+from __future__ import annotations
+
+from dataclasses import dataclass
+
+
+@dataclass(frozen=True, slots=True)
+class WorkflowCapabilityRef:
+    """Stable public capability name for one saved workflow artifact version."""
+
+    artifact_id: str
+    version: int
+
+    def __post_init__(self) -> None:
+        if not self.artifact_id:
+            raise ValueError("workflow capability ref requires an artifact id")
+        if self.version < 1:
+            raise ValueError("workflow capability ref requires version >= 1")
+
+    @classmethod
+    def parse(cls, value: str) -> WorkflowCapabilityRef:
+        """Parse `workflow.<artifact_id>.v<version>` into first-class fields."""
+        prefix = "workflow."
+        if not value.startswith(prefix):
+            raise ValueError("workflow capability ref must use the workflow namespace")
+        artifact_part, separator, version_part = value[len(prefix) :].rpartition(".v")
+        if not separator or not artifact_part or not version_part.isdecimal():
+            raise ValueError(
+                "workflow capability ref must be workflow.<artifact_id>.v<version>"
+            )
+        return cls(artifact_id=artifact_part, version=int(version_part))
+
+    def __str__(self) -> str:
+        return f"workflow.{self.artifact_id}.v{self.version}"

src/wf_mcp/workflow_surface/handlers.py

@@ -11,6 +11,7 @@ from wf_artifacts import (
     DiagnosticSeverity,
     RequiredCapability,
     WorkflowArtifact,
+    WorkflowCapabilityRef,
     WorkflowDeployment,
     create_workflow_artifact_from_plan as build_workflow_artifact_from_plan,
     validate_deployment_dependencies,
@@ -356,20 +357,21 @@ def _capability_name(qualified_name: str) -> str | None:
 
 def _artifact_capability_id(artifact: WorkflowArtifact) -> str:
     """Use the same stable name shape as workflow artifact catalog entries."""
-    return f"workflow.{artifact.id}.v{artifact.version}"
+    return str(
+        WorkflowCapabilityRef(
+            artifact_id=artifact.id,
+            version=artifact.version,
+        )
+    )
 
 
 def _parse_artifact_capability_id(qualified_name: str) -> tuple[str, int] | None:
     """Parse the stable `workflow.<artifact_id>.v<version>` capability name."""
-    prefix = "workflow."
-    if not qualified_name.startswith(prefix):
+    try:
+        ref = WorkflowCapabilityRef.parse(qualified_name)
+    except ValueError:
         return None
-    artifact_part, separator, version_part = qualified_name[len(prefix) :].rpartition(
-        ".v"
-    )
-    if not separator or not artifact_part or not version_part.isdecimal():
-        return None
-    return artifact_part, int(version_part)
+    return ref.artifact_id, ref.version
 
 
 def _raw_plan_from_artifact(artifact: WorkflowArtifact) -> RawWorkflowPlan: