resolve runtime dependencies

docs/workflow_artifacts.md

@@ -502,6 +502,19 @@ Saved workflow execution eventually needs first-class runtime support for:
   direct artifact dependencies just like node specs or tools; reducer config is
   part of the state field contract, while the dependency key is the reducer name
 
+Runtime execution must resolve reducer dependencies, not just validate that they
+exist. A deployment binding maps the artifact's logical reducer source, such as
+`custom`, to a concrete source, such as `custom.default`. The runtime registers
+the concrete `ReducerDefinition` under the logical reducer name used in the
+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.
+
 The first implementation should prefer artifact validation and dependency
 diagnostics before attempting persistent nested resume.
 

src/wf_mcp/broker/service/builtins.py

@@ -121,6 +121,11 @@ def builtin_reducers() -> dict[str, ReducerSpec]:
     }
 
 
+def builtin_reducer_definitions():
+    """Return executable built-in reducers for trusted runtime dependency wiring."""
+    return dict(DEFAULT_REDUCER_DEFINITIONS)
+
+
 def mcp_specs(service: ToolCaller) -> dict[str, NodeSpec[Any, Any]]:
     """Return service-bound MCP utility specs available to raw plans."""
 
@@ -153,6 +158,7 @@ def builtin_sources(service: ToolCaller) -> dict[str, CapabilitySource]:
             capabilities=CapabilityBuckets(
                 node_specs=builtin_specs(),
                 reducers=builtin_reducers(),
+                reducer_definitions=builtin_reducer_definitions(),
             ),
             visibility=SourceVisibility(
                 planner=True,

src/wf_mcp/broker/service/capability_sources.py

@@ -5,6 +5,7 @@ from typing import Any, Literal
 
 from wf_authoring import NodeSpec
 from wf_core import ReducerSpec
+from wf_core.runtime.ops.merges import ReducerDefinition
 
 SourceKind = Literal["system", "connection"]
 
@@ -29,6 +30,7 @@ class CapabilityBuckets:
     tools: dict[str, Any] = field(default_factory=dict)
     node_specs: dict[str, NodeSpec[Any, Any]] = field(default_factory=dict)
     reducers: dict[str, ReducerSpec] = field(default_factory=dict)
+    reducer_definitions: dict[str, ReducerDefinition] = field(default_factory=dict)
     prompts: dict[str, Any] = field(default_factory=dict)
     resources: dict[str, Any] = field(default_factory=dict)
 

src/wf_mcp/broker/service/core.py

@@ -7,11 +7,12 @@ from typing import Any
 
 from pydantic import BaseModel
 
-from wf_authoring import NodeReturn, NodeSpec, build_async_registry
+from wf_authoring import NodeReturn, NodeSpec
 from wf_artifacts import (
     FileWorkflowArtifactStore,
     WorkflowArtifact,
     WorkflowArtifactCatalogEntry,
+    WorkflowDeployment,
     WorkflowArtifactStore,
     artifact_catalog_entry,
 )
@@ -33,6 +34,7 @@ from ...shared.errors import error_payload
 from ...shared.names import RESERVED_CONNECTION_IDS
 from ...storage import Store
 from ...workflow.wrappers import _model_from_schema
+from ...workflow_surface.runtime_dependencies import resolve_runtime_dependencies
 from ..catalog import CombinedCatalog, snapshot_from_specs
 from ..discovery import discover_connection_capabilities, specs_from_discovered_tools
 from ..admin_capabilities import admin_source
@@ -545,6 +547,8 @@ class WfMcpService:
         self,
         plan: RawWorkflowPlan,
         workflow_input: dict[str, Any],
+        deployment: WorkflowDeployment | None = None,
+        artifact: WorkflowArtifact | None = None,
     ):
         self._record_event(
             make_event(
@@ -554,13 +558,30 @@ class WfMcpService:
             )
         )
         workflow = self.compile_plan(plan)
-        specs = [
-            self._get_qualified_spec(node.node)
-            for node in workflow.nodes
-            if isinstance(node, NodeUse)
+        plan_node_names = [
+            node.node for node in workflow.nodes if isinstance(node, NodeUse)
         ]
-        registry = build_async_registry(*specs)
-        run = await execute_workflow_async(workflow, workflow_input, registry)
+        runtime_artifact = artifact or WorkflowArtifact(
+            id=plan.name,
+            version=1,
+            title=plan.name,
+            input_schema=plan.input_schema,
+            output_schema=plan.output_schema,
+            outcomes=("completed",),
+            plan=plan.model_dump(mode="json", by_alias=True),
+        )
+        dependencies = resolve_runtime_dependencies(
+            artifact=runtime_artifact,
+            deployment=deployment,
+            sources=self.capability_sources,
+            plan_node_names=plan_node_names,
+        )
+        run = await execute_workflow_async(
+            workflow,
+            workflow_input,
+            dependencies.node_registry,
+            reducers=dependencies.reducers,
+        )
         self._record_event(
             make_event(
                 "workflow_run_completed",

src/wf_mcp/workflow_surface/handlers.py

@@ -251,7 +251,12 @@ class WorkflowSurfaceHandlers:
             )
 
         plan = _raw_plan_from_artifact(artifact)
-        run = await self.service.run_workflow_from_plan(plan, workflow_input)
+        run = await self.service.run_workflow_from_plan(
+            plan,
+            workflow_input,
+            deployment=deployment,
+            artifact=artifact,
+        )
         return _run_payload(
             deployment=deployment,
             artifact=artifact,

src/wf_mcp/workflow_surface/runtime_dependencies.py

@@ -0,0 +1,99 @@
+from __future__ import annotations
+
+from dataclasses import dataclass
+from typing import Any
+
+from wf_artifacts import RequiredCapability, WorkflowArtifact, WorkflowDeployment
+from wf_authoring import AsyncRegistryHandler, NodeSpec, build_async_registry
+from wf_core.runtime.ops.merges import ReducerDefinition
+
+from wf_mcp.broker.service.capability_sources import CapabilitySource
+
+
+@dataclass(frozen=True, slots=True)
+class RuntimeDependencies:
+    """Executable dependencies resolved for one workflow run."""
+
+    node_specs: dict[str, NodeSpec[Any, Any]]
+    node_registry: dict[str, AsyncRegistryHandler]
+    reducers: dict[str, ReducerDefinition]
+
+
+def resolve_runtime_dependencies(
+    *,
+    artifact: WorkflowArtifact,
+    deployment: WorkflowDeployment | None,
+    sources: dict[str, CapabilitySource],
+    plan_node_names: list[str],
+) -> RuntimeDependencies:
+    """Resolve source-owned specs and reducers into runtime callables.
+
+    Node names in stored plans are still concrete today, so node specs resolve
+    by exact plan names. Reducers are already logical artifact dependencies, so
+    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)
+    }
+    reducers = _resolve_reducers(
+        required_capabilities=artifact.required_capabilities,
+        deployment=deployment,
+        sources=sources,
+    )
+    return RuntimeDependencies(
+        node_specs=node_specs,
+        node_registry=build_async_registry(*node_specs.values()),
+        reducers=reducers,
+    )
+
+
+def _find_node_spec(
+    node_name: str,
+    sources: dict[str, CapabilitySource],
+) -> NodeSpec[Any, Any]:
+    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}")
+
+
+def _resolve_reducers(
+    *,
+    required_capabilities: dict[str, RequiredCapability],
+    deployment: WorkflowDeployment | None,
+    sources: dict[str, CapabilitySource],
+) -> dict[str, ReducerDefinition]:
+    reducers: dict[str, ReducerDefinition] = {}
+    if deployment is None:
+        return reducers
+
+    for logical_ref, required in required_capabilities.items():
+        if required.kind != "reducer":
+            continue
+        bound_source_id = deployment.bindings.get(required.logical_source)
+        if bound_source_id is None:
+            continue
+        source = sources.get(bound_source_id)
+        if source is None:
+            continue
+        definition = _find_reducer_definition(
+            source=source,
+            capability_name=required.capability_name,
+        )
+        if definition is not None:
+            reducers[logical_ref] = definition
+    return reducers
+
+
+def _find_reducer_definition(
+    *,
+    source: CapabilitySource,
+    capability_name: str,
+) -> ReducerDefinition | None:
+    for reducer_name, definition in source.capabilities.reducer_definitions.items():
+        if reducer_name.rsplit(".", maxsplit=1)[-1] == capability_name:
+            return definition
+    return None

tests/wf_mcp/test_workflow_surface.py

@@ -3,13 +3,22 @@ from __future__ import annotations
 import asyncio
 from typing import Any
 
+from pydantic import BaseModel
+
 from wf_artifacts import (
     FileWorkflowArtifactStore,
     RequiredCapability,
     WorkflowArtifact,
     WorkflowDeployment,
 )
+from wf_authoring import node, reducer
 from wf_mcp.broker import WfMcpService
+from wf_mcp.broker.service.capability_sources import (
+    CapabilityBuckets,
+    CapabilitySource,
+    SourcePermissions,
+    SourceVisibility,
+)
 from wf_mcp.models import ConnectionConfig
 from wf_mcp.models import RawWorkflowPlan
 from wf_mcp.storage import FileStore
@@ -18,6 +27,24 @@ from wf_mcp.workflow_surface import WorkflowSurfaceHandlers
 from .test_support import echo_tool, local_temp_root
 
 
+class AmountInput(BaseModel):
+    amount: int
+
+
+class AmountOutput(BaseModel):
+    amount: int
+
+
+@node()
+async def amount_tool(payload: AmountInput) -> AmountOutput:
+    return AmountOutput(amount=payload.amount)
+
+
+@reducer(name="custom.default.multiply")
+def multiply(current: int | None, incoming: int) -> int:
+    return (current or 1) * incoming
+
+
 def test_workflow_surface_lists_artifact_catalog_entries() -> None:
     artifact_store = FileWorkflowArtifactStore(local_temp_root() / "surface_artifacts")
     artifact_store.save_artifact(_artifact())
@@ -147,6 +174,56 @@ def test_workflow_surface_runs_non_interrupting_deployment() -> None:
     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())
+    artifact_store.save_deployment(
+        WorkflowDeployment(
+            id="multiply.personal",
+            artifact_id="multiply",
+            artifact_version=1,
+            bindings={
+                "demo": "demo.personal",
+                "custom": "custom.default",
+            },
+        )
+    )
+    service = WfMcpService(
+        store=FileStore(local_temp_root() / "surface_reducer_mcp"),
+        artifact_store=artifact_store,
+    )
+    service.register_connection(
+        ConnectionConfig(id="demo.personal", server="demo", account="personal")
+    )
+    service.register_specs("demo.personal", amount_tool)
+    service.register_capability_source(
+        CapabilitySource(
+            id="custom.default",
+            kind="system",
+            capabilities=CapabilityBuckets(
+                reducers={multiply.definition.spec.name: multiply.definition.spec},
+                reducer_definitions={
+                    multiply.definition.spec.name: multiply.definition,
+                },
+            ),
+            visibility=SourceVisibility(planner=True),
+            permissions=SourcePermissions(safe_for_workflow=True),
+        )
+    )
+    handlers = WorkflowSurfaceHandlers(service)
+
+    payload = asyncio.run(
+        handlers.run_deployment(
+            deployment_id="multiply.personal",
+            workflow_input={"total": 2, "amount": 3},
+        )
+    )
+
+    assert payload["status"] == "completed"
+    assert payload["output"]["total"] == 6
+    assert payload["diagnostics"] == []
+
+
 def test_workflow_surface_calls_saved_wrapper_artifact() -> None:
     artifact_store = FileWorkflowArtifactStore(
         local_temp_root() / "surface_wrapper_call"
@@ -249,3 +326,62 @@ def _echo_artifact() -> WorkflowArtifact:
             )
         },
     )
+
+
+def _custom_reducer_artifact() -> WorkflowArtifact:
+    plan: dict[str, Any] = {
+        "name": "multiply",
+        "input_schema": {
+            "type": "object",
+            "properties": {
+                "total": {"type": "integer"},
+                "amount": {"type": "integer"},
+            },
+            "required": ["total", "amount"],
+        },
+        "state_schema": {
+            "fields": {
+                "total": {
+                    "type": "integer",
+                    "reducer": "custom.multiply",
+                }
+            }
+        },
+        "output_schema": {
+            "type": "object",
+            "properties": {"total": {"type": "integer"}},
+            "required": ["total"],
+        },
+        "start": "amount",
+        "nodes": [
+            {
+                "id": "amount",
+                "type": "node",
+                "node": "demo.personal.amount_tool",
+                "in_map": {"input.amount": "amount"},
+                "out_map": {"amount": "state.total"},
+            }
+        ],
+        "edges": [{"from": "amount", "outcome": "ok", "to": "__end__"}],
+    }
+    return WorkflowArtifact(
+        id="multiply",
+        version=1,
+        title="Multiply",
+        input_schema=plan["input_schema"],
+        output_schema=plan["output_schema"],
+        outcomes=("completed",),
+        plan=plan,
+        required_capabilities={
+            "demo.amount_tool": RequiredCapability(
+                logical_source="demo",
+                capability_name="amount_tool",
+                kind="node_spec",
+            ),
+            "custom.multiply": RequiredCapability(
+                logical_source="custom",
+                capability_name="multiply",
+                kind="reducer",
+            ),
+        },
+    )