more map ->list moves: WorkflowDeployment / WorkflowArtifact

src/wf_artifacts/__init__.py

@@ -28,6 +28,7 @@ from .models import (
     DiagnosticSeverity,
     DriftPolicy,
     RequiredCapability,
+    SourceBinding,
     WorkflowArtifact,
     WorkflowDeployment,
 )
@@ -48,6 +49,7 @@ __all__ = [
     "FileDraftWorkspaceStore",
     "FileWorkflowArtifactStore",
     "RequiredCapability",
+    "SourceBinding",
     "WorkflowArtifact",
     "WorkflowArtifactCatalogEntry",
     "WorkflowCapabilityRef",

src/wf_artifacts/catalog.py

@@ -41,7 +41,7 @@ def artifact_catalog_entry(
     required_sources = sorted(
         {
             capability.logical_source
-            for capability in artifact.required_capabilities.values()
+            for capability in artifact.required_capability_map().values()
         }
     )
     return WorkflowArtifactCatalogEntry(

src/wf_artifacts/factory.py

@@ -30,6 +30,11 @@ def create_workflow_artifact_from_plan(
         observed_node_specs,
     )
     _validate_workflow_plan(normalized_plan)
+    required = {
+        **_required_reducers_from_plan(normalized_plan),
+        **node_requirements,
+        **dict(required_capabilities or {}),
+    }
     return WorkflowArtifact(
         id=artifact_id,
         version=version,
@@ -40,11 +45,7 @@ def create_workflow_artifact_from_plan(
         output_schema=_required_object_field(normalized_plan, "output_schema"),
         outcomes=outcomes,
         plan=normalized_plan,
-        required_capabilities={
-            **_required_reducers_from_plan(normalized_plan),
-            **node_requirements,
-            **dict(required_capabilities or {}),
-        },
+        required_capabilities=list(required.values()),
         created_from_catalog_version=created_from_catalog_version,
     )
 
@@ -103,8 +104,7 @@ def _required_reducers_from_plan(plan: JsonObject) -> dict[str, RequiredCapabili
         except ValueError:
             continue
         requirements[reducer.name] = RequiredCapability(
-            logical_source=str(reducer_ref.source),
-            capability_name=reducer_ref.name,
+            ref=reducer_ref,
             kind="reducer",
         )
     return requirements

src/wf_artifacts/models.py

@@ -3,10 +3,14 @@ from __future__ import annotations
 from enum import StrEnum
 from typing import Any, Literal
 
-from pydantic import BaseModel, Field
+from pydantic import BaseModel, Field, model_validator
+
+from wf_platform import CapabilityRef, SourceRef
 
 JsonObject = dict[str, Any]
 ArtifactKind = Literal["workflow", "wrapper"]
+SourceRefInput = SourceRef | str
+CapabilityRefInput = CapabilityRef | str
 
 
 class DriftPolicy(StrEnum):
@@ -27,16 +31,46 @@ class DiagnosticSeverity(StrEnum):
 class RequiredCapability(BaseModel):
     """Saved contract for one capability an artifact references."""
 
-    logical_source: str
-    capability_name: str
+    ref: CapabilityRefInput
     kind: Literal["tool", "resource", "prompt", "node_spec", "reducer", "workflow"]
     input_schema_hash: str | None = None
     input_schema_snapshot: JsonObject | None = None
     output_schema_hash: str | None = None
     output_schema_snapshot: JsonObject | None = None
-    observed_concrete_source: str | None = None
+    observed_concrete_source: SourceRefInput | None = None
     observed_at_epoch_ms: int | None = Field(default=None, ge=0)
 
+    @property
+    def logical_source(self) -> str:
+        """Compatibility accessor for callers migrating to `ref`."""
+        return str(self.capability_ref().source)
+
+    @property
+    def capability_name(self) -> str:
+        """Compatibility accessor for callers migrating to `ref`."""
+        return self.capability_ref().name
+
+    def capability_ref(self) -> CapabilityRef:
+        """Return the typed ref even when constructed from JSON-compatible input."""
+        return (
+            self.ref
+            if isinstance(self.ref, CapabilityRef)
+            else CapabilityRef.parse(self.ref)
+        )
+
+    @model_validator(mode="before")
+    @classmethod
+    def _coerce_legacy_ref_fields(cls, value: object) -> object:
+        if not isinstance(value, dict):
+            return value
+        data = dict(value)
+        if "ref" not in data:
+            logical_source = data.pop("logical_source", None)
+            capability_name = data.pop("capability_name", None)
+            if isinstance(logical_source, str) and isinstance(capability_name, str):
+                data["ref"] = f"{logical_source}.{capability_name}"
+        return data
+
 
 class AvailableCapability(BaseModel):
     """Current contract for one capability exposed by a bound source."""
@@ -55,6 +89,29 @@ class AvailableSource(BaseModel):
     capabilities: dict[str, AvailableCapability] = Field(default_factory=dict)
 
 
+class SourceBinding(BaseModel):
+    """Deployment-time mapping from artifact logical source to concrete source."""
+
+    logical_source: SourceRefInput
+    concrete_source: SourceRefInput
+
+    def logical_ref(self) -> SourceRef:
+        """Return the typed logical source ref for runtime lookup code."""
+        return (
+            self.logical_source
+            if isinstance(self.logical_source, SourceRef)
+            else SourceRef.parse(self.logical_source)
+        )
+
+    def concrete_ref(self) -> SourceRef:
+        """Return the typed concrete source ref for runtime lookup code."""
+        return (
+            self.concrete_source
+            if isinstance(self.concrete_source, SourceRef)
+            else SourceRef.parse(self.concrete_source)
+        )
+
+
 class DependencyDiagnostic(BaseModel):
     """Machine-readable reason a deployment is degraded or unrunnable."""
 
@@ -78,10 +135,66 @@ class WorkflowArtifact(BaseModel):
     output_schema: JsonObject
     outcomes: tuple[str, ...]
     plan: JsonObject
-    required_capabilities: dict[str, RequiredCapability] = Field(default_factory=dict)
+    required_capabilities: (
+        list[RequiredCapability] | dict[str, RequiredCapability | JsonObject]
+    ) = Field(default_factory=list)
     workflow_dependencies: dict[str, int] = Field(default_factory=dict)
     created_from_catalog_version: str | None = None
 
+    def required_capability_map(self) -> dict[str, RequiredCapability]:
+        """Return required capabilities keyed by dot-joined capability ref."""
+        return {
+            str(capability.capability_ref()): capability
+            for capability in self._required_capability_list()
+        }
+
+    @model_validator(mode="before")
+    @classmethod
+    def _coerce_legacy_required_capabilities(cls, value: object) -> object:
+        if not isinstance(value, dict):
+            return value
+        data = dict(value)
+        required = data.get("required_capabilities")
+        if not isinstance(required, dict):
+            return data
+
+        normalized: list[object] = []
+        for raw_ref, raw_capability in required.items():
+            if not isinstance(raw_capability, dict):
+                normalized.append(raw_capability)
+                continue
+            capability_data = dict(raw_capability)
+            capability_data.setdefault("ref", str(raw_ref))
+            normalized.append(capability_data)
+        data["required_capabilities"] = normalized
+        return data
+
+    @model_validator(mode="after")
+    def _reject_duplicate_required_capabilities(self) -> WorkflowArtifact:
+        self.required_capabilities = self._required_capability_list()
+        refs = [
+            str(capability.capability_ref())
+            for capability in self.required_capabilities
+        ]
+        duplicates = {ref for ref in refs if refs.count(ref) > 1}
+        if duplicates:
+            raise ValueError(
+                "duplicate required capability refs: " + ", ".join(sorted(duplicates))
+            )
+        return self
+
+    def _required_capability_list(self) -> list[RequiredCapability]:
+        if isinstance(self.required_capabilities, dict):
+            return [
+                RequiredCapability.model_validate(
+                    {"ref": ref, **capability}
+                    if isinstance(capability, dict)
+                    else capability
+                )
+                for ref, capability in self.required_capabilities.items()
+            ]
+        return self.required_capabilities
+
 
 class WorkflowDeployment(BaseModel):
     """One configured way to run an artifact version in an environment."""
@@ -89,5 +202,57 @@ class WorkflowDeployment(BaseModel):
     id: str
     artifact_id: str
     artifact_version: int = Field(ge=1)
-    bindings: dict[str, str] = Field(default_factory=dict)
+    bindings: list[SourceBinding | dict[str, str]] | dict[str, str] = Field(
+        default_factory=list
+    )
     drift_policy: DriftPolicy = DriftPolicy.BLOCK
+
+    def binding_map(self) -> dict[str, str]:
+        """Return bindings keyed by dot-joined logical source ref."""
+        return {
+            str(binding.logical_ref()): str(binding.concrete_ref())
+            for binding in self._binding_list()
+        }
+
+    @model_validator(mode="before")
+    @classmethod
+    def _coerce_legacy_binding_map(cls, value: object) -> object:
+        if not isinstance(value, dict):
+            return value
+        data = dict(value)
+        bindings = data.get("bindings")
+        if not isinstance(bindings, dict):
+            return data
+
+        data["bindings"] = [
+            {"logical_source": logical, "concrete_source": concrete}
+            for logical, concrete in bindings.items()
+        ]
+        return data
+
+    @model_validator(mode="after")
+    def _reject_duplicate_bindings(self) -> WorkflowDeployment:
+        normalized = self._binding_list()
+        self.bindings = [*normalized]
+        logical_sources = [str(binding.logical_ref()) for binding in normalized]
+        duplicates = {
+            source for source in logical_sources if logical_sources.count(source) > 1
+        }
+        if duplicates:
+            raise ValueError(
+                "duplicate deployment source bindings: " + ", ".join(sorted(duplicates))
+            )
+        return self
+
+    def _binding_list(self) -> list[SourceBinding]:
+        if isinstance(self.bindings, dict):
+            return [
+                SourceBinding(logical_source=logical, concrete_source=concrete)
+                for logical, concrete in self.bindings.items()
+            ]
+        return [
+            binding
+            if isinstance(binding, SourceBinding)
+            else SourceBinding.model_validate(binding)
+            for binding in self.bindings
+        ]

src/wf_artifacts/references.py

@@ -46,8 +46,7 @@ def normalize_plan_node_refs(
             else None
         )
         requirements[logical_ref] = RequiredCapability(
-            logical_source=logical_source,
-            capability_name=capability_name,
+            ref=CapabilityRef.parse(logical_ref),
             kind="node_spec",
             input_schema_hash=(
                 hash_json_schema(observed.input_schema)
@@ -65,7 +64,7 @@ def normalize_plan_node_refs(
             output_schema_snapshot=(
                 observed.output_schema if observed is not None else None
             ),
-            observed_concrete_source=concrete_source,
+            observed_concrete_source=SourceRef.parse(concrete_source),
         )
 
     return normalized, requirements

src/wf_artifacts/validation.py

@@ -19,10 +19,11 @@ def validate_deployment_dependencies(
 ) -> list[DependencyDiagnostic]:
     """Validate that a deployment can satisfy an artifact's required contracts."""
     sources_by_id = {source.id: source for source in sources}
+    bindings = deployment.binding_map()
     diagnostics: list[DependencyDiagnostic] = []
 
-    for logical_ref, required in artifact.required_capabilities.items():
-        bound_source_id = deployment.bindings.get(required.logical_source)
+    for logical_ref, required in artifact.required_capability_map().items():
+        bound_source_id = bindings.get(required.logical_source)
         if bound_source_id is None:
             diagnostics.append(
                 _diagnostic(