deployment binds accounts to nodes

docs/workflow_artifacts.md

@@ -510,10 +510,19 @@ saved workflow state schema, such as `custom.multiply`. This keeps artifact
 plans stable while allowing deployments to choose concrete accounts or local
 reducer packages.
 
-Node specs are less abstract today: saved raw workflow plans still contain
-concrete node spec names such as `demo.personal.echo_tool`. Rebinding node specs
-through deployment aliases is a later migration. Reducers are the first runtime
-dependency family to use deployment-bound logical names end-to-end.
+Node specs may also use deployment-bound logical names. A saved plan can refer
+to `demo.echo_tool`, while the deployment binds `demo` to a concrete source such
+as `demo.personal`. At runtime the compiler builds node definitions from the
+concrete source but leaves the saved artifact immutable. Concrete node names
+such as `demo.personal.echo_tool` remain supported for raw local plans and older
+artifacts.
+
+Implementation note: these references are currently parsed from strings with
+dot-separated source and capability names. That keeps the wire format simple but
+leaks path logic into runtime code. A future cleanup should introduce typed
+reference objects, such as `CapabilityRef(logical_source, capability_name)` and
+`BoundCapabilityRef(concrete_source, capability_name)`, and leave dot-joined
+strings as presentation and serialization only.
 
 The first implementation should prefer artifact validation and dependency
 diagnostics before attempting persistent nested resume.

src/wf_mcp/broker/service/core.py

@@ -522,15 +522,27 @@ class WfMcpService:
             )
             raise
 
-    def compile_plan(self, plan: RawWorkflowPlan) -> Workflow:
+    def compile_plan(
+        self,
+        plan: RawWorkflowPlan,
+        node_name_bindings: dict[str, str] | None = None,
+    ) -> Workflow:
         node_defs: dict[str, Any] = {}
+        bindings = node_name_bindings or {}
         for step in plan.nodes:
             if not isinstance(step, NodeUse):
                 continue
-            qualified_name = step.node
+            qualified_name = bindings.get(step.node, step.node)
             spec = self._get_qualified_spec(qualified_name)
             node_defs[qualified_name] = spec.to_node_def()
 
+        nodes = []
+        for node in plan.nodes:
+            payload = node.model_dump(by_alias=True)
+            if isinstance(node, NodeUse):
+                payload["node"] = bindings.get(node.node, node.node)
+            nodes.append(payload)
+
         payload = {
             "name": plan.name,
             "input_schema": plan.input_schema,
@@ -538,7 +550,7 @@ class WfMcpService:
             "output_schema": plan.output_schema,
             "start": plan.start,
             "node_defs": [node.model_dump() for node in node_defs.values()],
-            "nodes": [node.model_dump(by_alias=True) for node in plan.nodes],
+            "nodes": nodes,
             "edges": [edge.model_dump(by_alias=True) for edge in plan.edges],
         }
         return Workflow.model_validate(payload)
@@ -557,9 +569,8 @@ class WfMcpService:
                 payload={"input_keys": sorted(workflow_input.keys())},
             )
         )
-        workflow = self.compile_plan(plan)
         plan_node_names = [
-            node.node for node in workflow.nodes if isinstance(node, NodeUse)
+            node.node for node in plan.nodes if isinstance(node, NodeUse)
         ]
         runtime_artifact = artifact or WorkflowArtifact(
             id=plan.name,
@@ -576,6 +587,7 @@ class WfMcpService:
             sources=self.capability_sources,
             plan_node_names=plan_node_names,
         )
+        workflow = self.compile_plan(plan, dependencies.node_name_bindings)
         run = await execute_workflow_async(
             workflow,
             workflow_input,

src/wf_mcp/workflow_surface/runtime_dependencies.py

@@ -15,6 +15,7 @@ class RuntimeDependencies:
     """Executable dependencies resolved for one workflow run."""
 
     node_specs: dict[str, NodeSpec[Any, Any]]
+    node_name_bindings: dict[str, str]
     node_registry: dict[str, AsyncRegistryHandler]
     reducers: dict[str, ReducerDefinition]
 
@@ -33,10 +34,16 @@ def resolve_runtime_dependencies(
     they resolve through deployment bindings and are registered under the
     logical reducer name used by the workflow state schema.
     """
-    node_specs = {
-        node_name: _find_node_spec(node_name, sources)
-        for node_name in dict.fromkeys(plan_node_names)
-    }
+    node_specs: dict[str, NodeSpec[Any, Any]] = {}
+    node_name_bindings: dict[str, str] = {}
+    for node_name in dict.fromkeys(plan_node_names):
+        concrete_name, spec = _resolve_node_spec(
+            node_name=node_name,
+            deployment=deployment,
+            sources=sources,
+        )
+        node_name_bindings[node_name] = concrete_name
+        node_specs[concrete_name] = spec
     reducers = _resolve_reducers(
         required_capabilities=artifact.required_capabilities,
         deployment=deployment,
@@ -44,20 +51,44 @@ def resolve_runtime_dependencies(
     )
     return RuntimeDependencies(
         node_specs=node_specs,
+        node_name_bindings=node_name_bindings,
         node_registry=build_async_registry(*node_specs.values()),
         reducers=reducers,
     )
 
 
+def _resolve_node_spec(
+    *,
+    node_name: str,
+    deployment: WorkflowDeployment | None,
+    sources: dict[str, CapabilitySource],
+) -> tuple[str, NodeSpec[Any, Any]]:
+    concrete = _find_node_spec(node_name, sources)
+    if concrete is not None:
+        return node_name, concrete
+
+    if deployment is not None:
+        logical_source, separator, capability_name = node_name.rpartition(".")
+        if separator:
+            bound_source_id = deployment.bindings.get(logical_source)
+            if bound_source_id is not None:
+                bound_name = f"{bound_source_id}.{capability_name}"
+                concrete = _find_node_spec(bound_name, sources)
+                if concrete is not None:
+                    return bound_name, concrete
+
+    raise KeyError(f"unknown node spec {node_name!r}")
+
+
 def _find_node_spec(
     node_name: str,
     sources: dict[str, CapabilitySource],
-) -> NodeSpec[Any, Any]:
+) -> NodeSpec[Any, Any] | None:
     for source in sources.values():
         spec = source.capabilities.node_specs.get(node_name)
         if spec is not None:
             return spec
-    raise KeyError(f"unknown node spec {node_name!r}")
+    return None
 
 
 def _resolve_reducers(

tests/wf_mcp/test_workflow_surface.py

@@ -174,6 +174,39 @@ def test_workflow_surface_runs_non_interrupting_deployment() -> None:
     assert payload["diagnostics"] == []
 
 
+def test_workflow_surface_runs_deployment_with_bound_node_spec_dependency() -> None:
+    artifact_store = FileWorkflowArtifactStore(local_temp_root() / "surface_bound_node")
+    artifact_store.save_artifact(_logical_echo_artifact())
+    artifact_store.save_deployment(
+        WorkflowDeployment(
+            id="echo.personal",
+            artifact_id="logical_echo",
+            artifact_version=1,
+            bindings={"demo": "demo.personal"},
+        )
+    )
+    service = WfMcpService(
+        store=FileStore(local_temp_root() / "surface_bound_node_mcp"),
+        artifact_store=artifact_store,
+    )
+    service.register_connection(
+        ConnectionConfig(id="demo.personal", server="demo", account="personal")
+    )
+    service.register_specs("demo.personal", echo_tool)
+    handlers = WorkflowSurfaceHandlers(service)
+
+    payload = asyncio.run(
+        handlers.run_deployment(
+            deployment_id="echo.personal",
+            workflow_input={"text": "hello"},
+        )
+    )
+
+    assert payload["status"] == "completed"
+    assert payload["output"]["echoed"] == "hello"
+    assert payload["diagnostics"] == []
+
+
 def test_workflow_surface_runs_deployment_with_bound_reducer_dependency() -> None:
     artifact_store = FileWorkflowArtifactStore(local_temp_root() / "surface_reducer")
     artifact_store.save_artifact(_custom_reducer_artifact())
@@ -328,6 +361,20 @@ def _echo_artifact() -> WorkflowArtifact:
     )
 
 
+def _logical_echo_artifact() -> WorkflowArtifact:
+    artifact = _echo_artifact()
+    plan = dict(artifact.plan)
+    nodes = [dict(node) for node in plan["nodes"]]
+    nodes[0]["node"] = "demo.echo_tool"
+    plan["nodes"] = nodes
+    return artifact.model_copy(
+        update={
+            "id": "logical_echo",
+            "plan": plan,
+        }
+    )
+
+
 def _custom_reducer_artifact() -> WorkflowArtifact:
     plan: dict[str, Any] = {
         "name": "multiply",