alright uhh

src/wf_artifacts/__init__.py

@@ -17,6 +17,7 @@ from .models import (
 )
 from .store import FileWorkflowArtifactStore, WorkflowArtifactStore
 from .validation import validate_deployment_dependencies
+from .references import logical_ref_for_concrete_ref, normalize_plan_node_refs
 
 __all__ = [
     "AvailableCapability",
@@ -34,5 +35,7 @@ __all__ = [
     "artifact_catalog_entry",
     "artifact_node_name",
     "create_workflow_artifact_from_plan",
+    "logical_ref_for_concrete_ref",
+    "normalize_plan_node_refs",
     "validate_deployment_dependencies",
 ]

src/wf_artifacts/factory.py

@@ -5,6 +5,7 @@ from collections.abc import Mapping
 from wf_core import ReducerRef, Workflow
 
 from .models import ArtifactKind, JsonObject, RequiredCapability, WorkflowArtifact
+from .references import normalize_plan_node_refs
 
 
 def create_workflow_artifact_from_plan(
@@ -17,22 +18,28 @@ def create_workflow_artifact_from_plan(
     kind: ArtifactKind = "workflow",
     description: str | None = None,
     required_capabilities: Mapping[str, RequiredCapability] | None = None,
+    source_bindings: Mapping[str, str] | None = None,
     created_from_catalog_version: str | None = None,
 ) -> WorkflowArtifact:
     """Create an immutable artifact from a declarative workflow plan."""
-    _validate_workflow_plan(plan)
+    normalized_plan, node_requirements = normalize_plan_node_refs(
+        plan,
+        source_bindings or {},
+    )
+    _validate_workflow_plan(normalized_plan)
     return WorkflowArtifact(
         id=artifact_id,
         version=version,
         title=title,
         kind=kind,
         description=description,
-        input_schema=_required_object_field(plan, "input_schema"),
-        output_schema=_required_object_field(plan, "output_schema"),
+        input_schema=_required_object_field(normalized_plan, "input_schema"),
+        output_schema=_required_object_field(normalized_plan, "output_schema"),
         outcomes=outcomes,
-        plan=plan,
+        plan=normalized_plan,
         required_capabilities={
-            **_required_reducers_from_plan(plan),
+            **_required_reducers_from_plan(normalized_plan),
+            **node_requirements,
             **dict(required_capabilities or {}),
         },
         created_from_catalog_version=created_from_catalog_version,

src/wf_artifacts/references.py

@@ -0,0 +1,73 @@
+from __future__ import annotations
+
+from collections.abc import Mapping
+from copy import deepcopy
+
+from .models import JsonObject, RequiredCapability
+
+
+def normalize_plan_node_refs(
+    plan: JsonObject,
+    source_bindings: Mapping[str, str],
+) -> tuple[JsonObject, dict[str, RequiredCapability]]:
+    """Rewrite concrete node refs in a plan to deployment-bound logical refs.
+
+    `source_bindings` uses the same direction as `WorkflowDeployment.bindings`:
+    logical source -> concrete source. This lets artifact creation accept the
+    source mapping authors already need at deployment time while saving portable
+    plan refs such as `demo.echo_tool` instead of `demo.personal.echo_tool`.
+    """
+    if not source_bindings:
+        return deepcopy(plan), {}
+
+    normalized = deepcopy(plan)
+    requirements: dict[str, RequiredCapability] = {}
+    nodes = normalized.get("nodes")
+    if not isinstance(nodes, list):
+        return normalized, requirements
+
+    for node in nodes:
+        if not isinstance(node, dict):
+            continue
+        raw_node_ref = node.get("node")
+        if not isinstance(raw_node_ref, str):
+            continue
+        replacement = logical_ref_for_concrete_ref(raw_node_ref, source_bindings)
+        if replacement is None:
+            continue
+        logical_ref, logical_source, capability_name, concrete_source = replacement
+        node["node"] = logical_ref
+        requirements[logical_ref] = RequiredCapability(
+            logical_source=logical_source,
+            capability_name=capability_name,
+            kind="node_spec",
+            observed_concrete_source=concrete_source,
+        )
+
+    return normalized, requirements
+
+
+def logical_ref_for_concrete_ref(
+    concrete_ref: str,
+    source_bindings: Mapping[str, str],
+) -> tuple[str, str, str, str] | None:
+    """Return a logical ref for one concrete capability ref when bound.
+
+    The longest concrete source match wins so `demo.personal.pro.echo` can be
+    handled safely when both `demo.personal` and `demo.personal.pro` exist.
+    """
+    matches = sorted(
+        source_bindings.items(),
+        key=lambda item: len(item[1]),
+        reverse=True,
+    )
+    for logical_source, concrete_source in matches:
+        prefix = f"{concrete_source}."
+        if not concrete_ref.startswith(prefix):
+            continue
+        capability_name = concrete_ref[len(prefix) :]
+        if not capability_name:
+            continue
+        logical_ref = f"{logical_source}.{capability_name}"
+        return logical_ref, logical_source, capability_name, concrete_source
+    return None