dep validation of deployment

core wf valication uses a &mut List, this returns a List, hmmmm

src/wf_artifacts/__init__.py

@@ -1,4 +1,6 @@
 from .models import (
+    AvailableCapability,
+    AvailableSource,
     DependencyDiagnostic,
     DiagnosticSeverity,
     DriftPolicy,
@@ -7,8 +9,11 @@ from .models import (
     WorkflowDeployment,
 )
 from .store import FileWorkflowArtifactStore, WorkflowArtifactStore
+from .validation import validate_deployment_dependencies
 
 __all__ = [
+    "AvailableCapability",
+    "AvailableSource",
     "DependencyDiagnostic",
     "DiagnosticSeverity",
     "DriftPolicy",
@@ -17,4 +22,5 @@ __all__ = [
     "WorkflowArtifact",
     "WorkflowArtifactStore",
     "WorkflowDeployment",
+    "validate_deployment_dependencies",
 ]

src/wf_artifacts/models.py

@@ -37,6 +37,23 @@ class RequiredCapability(BaseModel):
     observed_at_epoch_ms: int | None = Field(default=None, ge=0)
 
 
+class AvailableCapability(BaseModel):
+    """Current contract for one capability exposed by a bound source."""
+
+    name: str
+    kind: Literal["tool", "resource", "prompt", "node_spec", "workflow"]
+    input_schema_hash: str | None = None
+    output_schema_hash: str | None = None
+
+
+class AvailableSource(BaseModel):
+    """Provider-neutral source snapshot used by artifact dependency validation."""
+
+    id: str
+    enabled: bool = True
+    capabilities: dict[str, AvailableCapability] = Field(default_factory=dict)
+
+
 class DependencyDiagnostic(BaseModel):
     """Machine-readable reason a deployment is degraded or unrunnable."""
 

src/wf_artifacts/validation.py

@@ -0,0 +1,171 @@
+from __future__ import annotations
+
+from .models import (
+    AvailableSource,
+    DependencyDiagnostic,
+    DiagnosticSeverity,
+    DriftPolicy,
+    RequiredCapability,
+    WorkflowArtifact,
+    WorkflowDeployment,
+)
+
+
+def validate_deployment_dependencies(
+    *,
+    artifact: WorkflowArtifact,
+    deployment: WorkflowDeployment,
+    sources: list[AvailableSource],
+) -> list[DependencyDiagnostic]:
+    """Validate that a deployment can satisfy an artifact's required contracts."""
+    sources_by_id = {source.id: source for source in sources}
+    diagnostics: list[DependencyDiagnostic] = []
+
+    for logical_ref, required in artifact.required_capabilities.items():
+        bound_source_id = deployment.bindings.get(required.logical_source)
+        if bound_source_id is None:
+            diagnostics.append(
+                _diagnostic(
+                    code="binding_missing",
+                    logical_ref=logical_ref,
+                    required=required,
+                    message=(
+                        f"No binding exists for logical source "
+                        f"{required.logical_source!r}."
+                    ),
+                    repair_hint=(
+                        "Bind the logical source to a compatible concrete source."
+                    ),
+                )
+            )
+            continue
+
+        source = sources_by_id.get(bound_source_id)
+        if source is None:
+            diagnostics.append(
+                _diagnostic(
+                    code="source_missing",
+                    logical_ref=logical_ref,
+                    required=required,
+                    bound_source=bound_source_id,
+                    message=f"Bound source {bound_source_id!r} is not available.",
+                    repair_hint=(
+                        "Reconnect this source or bind the logical source to "
+                        "another compatible source."
+                    ),
+                )
+            )
+            continue
+
+        if not source.enabled:
+            diagnostics.append(
+                _diagnostic(
+                    code="source_disabled",
+                    logical_ref=logical_ref,
+                    required=required,
+                    bound_source=bound_source_id,
+                    message=f"Bound source {bound_source_id!r} is disabled.",
+                    repair_hint="Enable the source or choose another binding.",
+                )
+            )
+            continue
+
+        capability = source.capabilities.get(required.capability_name)
+        if capability is None:
+            diagnostics.append(
+                _diagnostic(
+                    code="capability_missing",
+                    logical_ref=logical_ref,
+                    required=required,
+                    bound_source=bound_source_id,
+                    message=(
+                        f"Bound source {bound_source_id!r} does not expose "
+                        f"capability {required.capability_name!r}."
+                    ),
+                    repair_hint=(
+                        "Refresh the source catalog or bind to another compatible "
+                        "source."
+                    ),
+                )
+            )
+            continue
+
+        if _schema_changed(required, capability):
+            drift = _schema_drift_diagnostic(
+                logical_ref=logical_ref,
+                required=required,
+                bound_source=bound_source_id,
+                policy=deployment.drift_policy,
+            )
+            if drift is not None:
+                diagnostics.append(drift)
+
+    return diagnostics
+
+
+def _schema_changed(
+    required: RequiredCapability,
+    available: object,
+) -> bool:
+    input_hash = getattr(available, "input_schema_hash", None)
+    output_hash = getattr(available, "output_schema_hash", None)
+    return (
+        required.input_schema_hash is not None
+        and input_hash is not None
+        and required.input_schema_hash != input_hash
+    ) or (
+        required.output_schema_hash is not None
+        and output_hash is not None
+        and required.output_schema_hash != output_hash
+    )
+
+
+def _schema_drift_diagnostic(
+    *,
+    logical_ref: str,
+    required: RequiredCapability,
+    bound_source: str,
+    policy: DriftPolicy,
+) -> DependencyDiagnostic | None:
+    if policy == DriftPolicy.ALLOW:
+        return None
+    severity = (
+        DiagnosticSeverity.WARNING
+        if policy == DriftPolicy.WARN
+        else DiagnosticSeverity.ERROR
+    )
+    return _diagnostic(
+        severity=severity,
+        code="schema_changed",
+        logical_ref=logical_ref,
+        required=required,
+        bound_source=bound_source,
+        message=(
+            f"Capability {required.capability_name!r} on source {bound_source!r} "
+            "has a different schema hash than the saved artifact contract."
+        ),
+        repair_hint=(
+            "Review the changed capability contract, then update the deployment "
+            "policy or migrate the workflow artifact."
+        ),
+    )
+
+
+def _diagnostic(
+    *,
+    code: str,
+    logical_ref: str,
+    required: RequiredCapability,
+    message: str,
+    severity: DiagnosticSeverity = DiagnosticSeverity.ERROR,
+    bound_source: str | None = None,
+    repair_hint: str | None = None,
+) -> DependencyDiagnostic:
+    return DependencyDiagnostic(
+        severity=severity,
+        code=code,
+        logical_ref=logical_ref,
+        bound_source=bound_source,
+        message=message,
+        repair_hint=repair_hint,
+    )

src/wf_mcp/broker/service/core.py

@@ -185,8 +185,7 @@ class WfMcpService:
                 source.id,
                 specs=source.capabilities.node_specs,
                 tool_display_names={
-                    entry.local_name: entry.title
+                    entry.local_name: entry.title for entry in stored_snapshot.nodes
-                    for entry in stored_snapshot.nodes
                 }
                 if stored_snapshot is not None
                 else None,

src/wf_mcp/proxy_validation.py

@@ -8,6 +8,8 @@ from .shared.names import RESERVED_CONNECTION_IDS
 
 _NAMESPACE_ID_RE = re.compile(r"^[A-Za-z0-9][A-Za-z0-9.-]*$")
 _SUPPORTED_TRANSPORTS = {"stdio", "http", "streamable-http", "streamable_http", "sse"}
+
+
 class ProxyConfigError(ValueError):
     """Raised when a broker config cannot safely run as a transparent proxy."""
 

tests/artifacts/test_validation.py

@@ -0,0 +1,178 @@
+from __future__ import annotations
+
+from wf_artifacts import (
+    AvailableCapability,
+    AvailableSource,
+    DriftPolicy,
+    RequiredCapability,
+    WorkflowArtifact,
+    WorkflowDeployment,
+    validate_deployment_dependencies,
+)
+
+
+def required_capability(
+    *,
+    logical_source: str = "context7",
+    capability_name: str = "query-docs",
+    input_hash: str = "sha256:input",
+    output_hash: str = "sha256:output",
+) -> RequiredCapability:
+    return RequiredCapability(
+        logical_source=logical_source,
+        capability_name=capability_name,
+        kind="tool",
+        input_schema_hash=input_hash,
+        input_schema_snapshot={"type": "object", "properties": {}},
+        output_schema_hash=output_hash,
+        output_schema_snapshot={"type": "object", "properties": {}},
+    )
+
+
+def artifact_with(capability: RequiredCapability) -> WorkflowArtifact:
+    return WorkflowArtifact(
+        id="summarize_docs",
+        version=1,
+        title="Summarize Docs",
+        input_schema={"type": "object", "properties": {}},
+        output_schema={"type": "object", "properties": {}},
+        outcomes=("done",),
+        plan={"name": "summarize_docs", "nodes": [], "edges": []},
+        required_capabilities={
+            f"{capability.logical_source}.{capability.capability_name}": capability
+        },
+    )
+
+
+def deployment(
+    *,
+    bindings: dict[str, str] | None = None,
+    drift_policy: DriftPolicy = DriftPolicy.BLOCK,
+) -> WorkflowDeployment:
+    return WorkflowDeployment(
+        id="summarize_docs.personal",
+        artifact_id="summarize_docs",
+        artifact_version=1,
+        bindings={"context7": "context7.personal"} if bindings is None else bindings,
+        drift_policy=drift_policy,
+    )
+
+
+def source(
+    *,
+    id: str = "context7.personal",
+    enabled: bool = True,
+    capability_name: str = "query-docs",
+    input_hash: str = "sha256:input",
+    output_hash: str = "sha256:output",
+) -> AvailableSource:
+    return AvailableSource(
+        id=id,
+        enabled=enabled,
+        capabilities={
+            capability_name: AvailableCapability(
+                name=capability_name,
+                kind="tool",
+                input_schema_hash=input_hash,
+                output_schema_hash=output_hash,
+            )
+        },
+    )
+
+
+def test_validate_deployment_accepts_matching_bound_capability() -> None:
+    diagnostics = validate_deployment_dependencies(
+        artifact=artifact_with(required_capability()),
+        deployment=deployment(),
+        sources=[source()],
+    )
+
+    assert diagnostics == []
+
+
+def test_validate_deployment_reports_missing_binding() -> None:
+    diagnostics = validate_deployment_dependencies(
+        artifact=artifact_with(required_capability()),
+        deployment=deployment(bindings={}),
+        sources=[source()],
+    )
+
+    assert len(diagnostics) == 1
+    assert diagnostics[0].severity == "error"
+    assert diagnostics[0].code == "binding_missing"
+    assert diagnostics[0].logical_ref == "context7.query-docs"
+    assert diagnostics[0].bound_source is None
+
+
+def test_validate_deployment_reports_missing_source() -> None:
+    diagnostics = validate_deployment_dependencies(
+        artifact=artifact_with(required_capability()),
+        deployment=deployment(),
+        sources=[],
+    )
+
+    assert len(diagnostics) == 1
+    assert diagnostics[0].severity == "error"
+    assert diagnostics[0].code == "source_missing"
+    assert diagnostics[0].logical_ref == "context7.query-docs"
+    assert diagnostics[0].bound_source == "context7.personal"
+
+
+def test_validate_deployment_reports_disabled_source() -> None:
+    diagnostics = validate_deployment_dependencies(
+        artifact=artifact_with(required_capability()),
+        deployment=deployment(),
+        sources=[source(enabled=False)],
+    )
+
+    assert len(diagnostics) == 1
+    assert diagnostics[0].severity == "error"
+    assert diagnostics[0].code == "source_disabled"
+    assert diagnostics[0].bound_source == "context7.personal"
+
+
+def test_validate_deployment_reports_missing_capability() -> None:
+    diagnostics = validate_deployment_dependencies(
+        artifact=artifact_with(required_capability()),
+        deployment=deployment(),
+        sources=[source(capability_name="other-tool")],
+    )
+
+    assert len(diagnostics) == 1
+    assert diagnostics[0].severity == "error"
+    assert diagnostics[0].code == "capability_missing"
+    assert diagnostics[0].logical_ref == "context7.query-docs"
+
+
+def test_validate_deployment_blocks_changed_schema_by_default() -> None:
+    diagnostics = validate_deployment_dependencies(
+        artifact=artifact_with(required_capability()),
+        deployment=deployment(),
+        sources=[source(input_hash="sha256:changed")],
+    )
+
+    assert len(diagnostics) == 1
+    assert diagnostics[0].severity == "error"
+    assert diagnostics[0].code == "schema_changed"
+
+
+def test_validate_deployment_warns_for_changed_schema_when_policy_warns() -> None:
+    diagnostics = validate_deployment_dependencies(
+        artifact=artifact_with(required_capability()),
+        deployment=deployment(drift_policy=DriftPolicy.WARN),
+        sources=[source(output_hash="sha256:changed")],
+    )
+
+    assert len(diagnostics) == 1
+    assert diagnostics[0].severity == "warning"
+    assert diagnostics[0].code == "schema_changed"
+
+
+def test_validate_deployment_allows_changed_schema_when_policy_allows() -> None:
+    diagnostics = validate_deployment_dependencies(
+        artifact=artifact_with(required_capability()),
+        deployment=deployment(drift_policy=DriftPolicy.ALLOW),
+        sources=[source(input_hash="sha256:changed")],
+    )
+
+    assert diagnostics == []

tests/wf_mcp/test_broker_server.py

@@ -79,9 +79,7 @@ def test_create_broker_server_exposes_tools_resources_and_prompts() -> None:
         server.call_tool("get_planner_catalog", {})
     )
     planner_catalog = cast(dict[str, Any], cast(object, planner_catalog_raw))
-    planner_names = [
+    planner_names = [node["qualified_name"] for node in planner_catalog["nodes"]]
-        node["qualified_name"] for node in planner_catalog["nodes"]
-    ]
     assert "demo.personal.echo_tool" in planner_names
     assert "wf.mcp.call_tool" in planner_names
     assert "wf.std.runtime_error" in planner_names
@@ -106,9 +104,10 @@ def test_broker_admin_tools_are_backed_by_wf_admin_source() -> None:
 
     assert "list_spec_sources" in tool_names
     assert "get_planner_catalog" in tool_names
-    assert "wf.admin.list_sources" in service.capability_sources[
+    assert (
-        "wf.admin"
+        "wf.admin.list_sources"
-    ].capabilities.tools
+        in service.capability_sources["wf.admin"].capabilities.tools
+    )
 
 
 def test_build_service_from_config_registers_connections() -> None:

tests/wf_mcp/test_service.py

@@ -462,12 +462,8 @@ def test_service_excludes_disabled_connection_specs_from_planner_catalog() -> No
     service.capability_sources["demo.personal"].enabled = False
 
     planner_payload = service.get_planner_catalog().as_payload()
-    planner_names = [
+    planner_names = [node["qualified_name"] for node in planner_payload["nodes"]]
-        node["qualified_name"] for node in planner_payload["nodes"]
+    available_names = [entry.qualified_name for entry in service.list_available_specs()]
-    ]
-    available_names = [
-        entry.qualified_name for entry in service.list_available_specs()
-    ]
 
     assert "demo.personal.echo_tool" not in planner_names
     assert "demo.personal.echo_tool" not in available_names
@@ -501,7 +497,9 @@ def test_service_preserves_disabled_connection_source_on_reregistration() -> Non
     assert "demo.personal.echo_tool" not in source.capabilities.node_specs
 
 
-def test_service_excludes_planner_hidden_connection_specs_from_planner_catalog() -> None:
+def test_service_excludes_planner_hidden_connection_specs_from_planner_catalog() -> (
+    None
+):
     service = WfMcpService(
         store=FileStore(local_temp_root() / "hidden_connection_spec_store")
     )
@@ -516,12 +514,8 @@ def test_service_excludes_planner_hidden_connection_specs_from_planner_catalog()
     )
 
     planner_payload = service.get_planner_catalog().as_payload()
-    planner_names = [
+    planner_names = [node["qualified_name"] for node in planner_payload["nodes"]]
-        node["qualified_name"] for node in planner_payload["nodes"]
+    available_names = [entry.qualified_name for entry in service.list_available_specs()]
-    ]
-    available_names = [
-        entry.qualified_name for entry in service.list_available_specs()
-    ]
 
     assert "demo.personal.echo_tool" not in planner_names
     assert "demo.personal.echo_tool" not in available_names