live check opt-in

src/wf_mcp/workflow_surface/handlers.py

@@ -1,9 +1,15 @@
 from __future__ import annotations
 
+import asyncio
 from collections.abc import Sequence
 from dataclasses import asdict
 from typing import TYPE_CHECKING, Any
 
+import anyio
+import httpx
+from mcp.client.streamable_http import StreamableHTTPError
+from mcp.shared.exceptions import McpError
+
 from wf_artifacts import (
     ArtifactKind,
     AvailableCapability,
@@ -41,6 +47,7 @@ from wf_core.models.steps import (
 )
 from wf_core.paths import GraphSourcePath, LocalPath, StatePath
 
+from ..broker.service.adapters import require_adapter
 from ..events import make_event
 from ..models import RawWorkflowPlan
 from ..shared import matches_query, paged_list_payload
@@ -74,6 +81,19 @@ from .wrapper_hints import (
     wrapper_hints_for_capability,
 )
 
+LIVE_SOURCE_CHECK_TIMEOUT_SECONDS = 8.0
+_LIVE_SOURCE_CHECK_FAILURES = (
+    KeyError,
+    TimeoutError,
+    OSError,
+    anyio.ClosedResourceError,
+    anyio.EndOfStream,
+    anyio.BrokenResourceError,
+    httpx.HTTPError,
+    McpError,
+    StreamableHTTPError,
+)
+
 if TYPE_CHECKING:
     from wf_core import RunState
 
@@ -933,10 +953,23 @@ class WorkflowSurfaceHandlers:
             "saved": True,
         }
 
-    async def validate_deployment(self, *, deployment_id: str) -> dict[str, Any]:
+    async def validate_deployment(
-        deployment, artifact, diagnostics, _tree = self._deployment_validation(
+        self,
+        *,
+        deployment_id: str,
+        live_check: bool = False,
+    ) -> dict[str, Any]:
+        deployment, artifact, diagnostics, tree = self._deployment_validation(
             deployment_id
         )
+        if live_check:
+            diagnostics.extend(
+                await _live_source_diagnostics(
+                    self.service,
+                    deployment=deployment,
+                    artifacts=[artifact, *tree.artifacts_by_ref.values()],
+                )
+            )
         return {
             "deployment_id": deployment.id,
             "artifact_id": artifact.id,
@@ -1214,6 +1247,70 @@ def _available_sources(service: WfMcpService) -> list[AvailableSource]:
     return sources
 
 
+async def _live_source_diagnostics(
+    service: WfMcpService,
+    *,
+    deployment: WorkflowDeployment,
+    artifacts: Sequence[WorkflowArtifact],
+) -> list[DependencyDiagnostic]:
+    """Return opt-in diagnostics for bound upstream sources that cannot answer.
+
+    Static deployment validation only checks the last known source catalog.
+    This probe intentionally performs live upstream I/O, so MCP tools keep it
+    disabled by default and only run it when the caller asks for liveness.
+    """
+    diagnostics: list[DependencyDiagnostic] = []
+    for source_id, logical_ref in _required_live_sources(deployment, artifacts).items():
+        source = service.capability_sources.get(source_id)
+        if (
+            source is None
+            or not source.enabled
+            or not source.permissions.calls_upstream
+        ):
+            continue
+        try:
+            connection = service.connections.get(source_id)
+            adapter = require_adapter(connection, service.adapters)
+            auth = service.load_auth(source_id)
+            await asyncio.wait_for(
+                adapter.list_tools(connection, auth),
+                timeout=LIVE_SOURCE_CHECK_TIMEOUT_SECONDS,
+            )
+        except _LIVE_SOURCE_CHECK_FAILURES as exc:
+            diagnostics.append(
+                DependencyDiagnostic(
+                    severity=DiagnosticSeverity.ERROR,
+                    code="source_unreachable",
+                    logical_ref=logical_ref,
+                    bound_source=source_id,
+                    message=(
+                        f"Live check for upstream source {source_id!r} failed: "
+                        f"{type(exc).__name__}: {exc}"
+                    ),
+                    repair_hint=(
+                        "Start or reconnect the source, fix its transport/auth "
+                        "configuration, or bind this deployment to another source."
+                    ),
+                )
+            )
+    return diagnostics
+
+
+def _required_live_sources(
+    deployment: WorkflowDeployment,
+    artifacts: Sequence[WorkflowArtifact],
+) -> dict[str, str]:
+    """Return concrete upstream source ids to live-check, with one logical ref."""
+    bindings = deployment.binding_map()
+    required: dict[str, str] = {}
+    for artifact in artifacts:
+        for logical_ref, capability in artifact.required_capability_map().items():
+            source_id = bindings.get(capability.logical_source)
+            if source_id is not None:
+                required.setdefault(source_id, logical_ref)
+    return required
+
+
 def _required_capabilities_for_plan(
     plan: dict[str, Any],
     *,

src/wf_mcp/workflow_surface/tools.py

@@ -596,8 +596,26 @@ def register_workflow_tools(server: FastMCP[Any], service: WfMcpService) -> None
         title="Validate Workflow Deployment",
         description="Check whether a deployment_id can run with currently enabled sources.",
     )
-    async def validate_deployment(deployment_id: str) -> dict[str, Any]:
+    async def validate_deployment(
-        return await handlers.validate_deployment(deployment_id=deployment_id)
+        deployment_id: Annotated[
+            str,
+            Field(description="Saved workflow deployment id to validate."),
+        ],
+        live_check: Annotated[
+            bool,
+            Field(
+                description=(
+                    "When true, also contact each required upstream source to "
+                    "verify it is reachable. Defaults false because this may "
+                    "spawn stdio servers or perform network I/O."
+                )
+            ),
+        ] = False,
+    ) -> dict[str, Any]:
+        return await handlers.validate_deployment(
+            deployment_id=deployment_id,
+            live_check=live_check,
+        )
 
     @server.tool(
         name="wf.workflow.run_deployment",

tests/wf_mcp/server/test_tools.py

@@ -128,6 +128,7 @@ def test_workflow_tools_have_human_metadata() -> None:
             tools = await client.list_tools()
             by_name = {tool.name: tool for tool in tools}
             list_artifacts = by_name["wf.workflow.list_artifacts"]
+            validate_deployment = by_name["wf.workflow.validate_deployment"]
             run_deployment = by_name["wf.workflow.run_deployment"]
             inspect_run = by_name["wf.workflow.inspect_run"]
             read_run_trace = by_name["wf.workflow.read_run_trace"]
@@ -138,6 +139,10 @@ def test_workflow_tools_have_human_metadata() -> None:
             assert "kind" in list_artifacts.inputSchema["properties"]
             assert "cursor" in list_artifacts.inputSchema["properties"]
             assert "limit" in list_artifacts.inputSchema["properties"]
+            live_check_schema = validate_deployment.inputSchema["properties"][
+                "live_check"
+            ]
+            assert "upstream" in live_check_schema.get("description", "")
             assert run_deployment.title == "Run Workflow Deployment"
             assert "deployment_id" in (run_deployment.description or "")
             assert "trace_range" in run_deployment.inputSchema["properties"]

tests/wf_mcp/workflow_surface/test_deployments.py

@@ -1,15 +1,41 @@
 from __future__ import annotations
 
 import asyncio
+from typing import cast
 
 import pytest
 
 from wf_artifacts import FileWorkflowArtifactStore, WorkflowDeployment
+from wf_mcp.models import AuthRecord, ConnectionConfig
+from wf_mcp.capabilities import DiscoveredTool
+from wf_mcp.sdk import BackendAdapter
 
-from ..test_support import local_temp_root
+from ..test_support import echo_tool, local_temp_root
 from .conftest import artifact, echo_artifact, handlers
 
 
+class FailingLivenessAdapter:
+    async def list_tools(
+        self,
+        connection: ConnectionConfig,
+        auth: AuthRecord | None,
+    ) -> list[DiscoveredTool]:
+        raise OSError("stdio process exited")
+
+
+class RecordingLivenessAdapter:
+    def __init__(self) -> None:
+        self.calls = 0
+
+    async def list_tools(
+        self,
+        connection: ConnectionConfig,
+        auth: AuthRecord | None,
+    ) -> list[DiscoveredTool]:
+        self.calls += 1
+        return []
+
+
 def test_workflow_surface_validates_deployment_dependencies() -> None:
     artifact_store = FileWorkflowArtifactStore(local_temp_root() / "surface_validate")
     artifact_store.save_artifact(artifact())
@@ -33,6 +59,99 @@ def test_workflow_surface_validates_deployment_dependencies() -> None:
     assert payload["diagnostics"][0]["code"] == "source_missing"
 
 
+def test_workflow_surface_validate_deployment_live_check_is_opt_in() -> None:
+    artifact_store = FileWorkflowArtifactStore(
+        local_temp_root() / "surface_live_opt_in"
+    )
+    artifact_store.save_artifact(echo_artifact())
+    artifact_store.save_deployment(
+        WorkflowDeployment(
+            id="echo.personal",
+            artifact_id="echo",
+            artifact_version=1,
+            bindings=[{"logical_source": "demo", "concrete_source": "demo.personal"}],
+        )
+    )
+    h = handlers(artifact_store)
+    adapter = RecordingLivenessAdapter()
+    h.service.register_connection(
+        ConnectionConfig(id="demo.personal", server="demo", account="personal")
+    )
+    h.service.register_specs("demo.personal", echo_tool)
+    h.service.register_adapter("demo", cast(BackendAdapter, adapter))
+
+    payload = asyncio.run(h.validate_deployment(deployment_id="echo.personal"))
+
+    assert payload["status"] == "runnable"
+    assert payload["diagnostics"] == []
+    assert adapter.calls == 0
+
+
+def test_workflow_surface_validate_deployment_live_check_reports_unreachable_source() -> (
+    None
+):
+    artifact_store = FileWorkflowArtifactStore(local_temp_root() / "surface_live_fail")
+    artifact_store.save_artifact(echo_artifact())
+    artifact_store.save_deployment(
+        WorkflowDeployment(
+            id="echo.personal",
+            artifact_id="echo",
+            artifact_version=1,
+            bindings=[{"logical_source": "demo", "concrete_source": "demo.personal"}],
+        )
+    )
+    h = handlers(artifact_store)
+    h.service.register_connection(
+        ConnectionConfig(id="demo.personal", server="demo", account="personal")
+    )
+    h.service.register_specs("demo.personal", echo_tool)
+    h.service.register_adapter(
+        "demo",
+        cast(BackendAdapter, FailingLivenessAdapter()),
+    )
+
+    payload = asyncio.run(
+        h.validate_deployment(deployment_id="echo.personal", live_check=True)
+    )
+
+    assert payload["status"] == "unrunnable"
+    assert payload["diagnostics"][0]["code"] == "source_unreachable"
+    assert payload["diagnostics"][0]["bound_source"] == "demo.personal"
+    assert "stdio process exited" in payload["diagnostics"][0]["message"]
+
+
+def test_workflow_surface_validate_deployment_live_check_reports_missing_connection() -> (
+    None
+):
+    artifact_store = FileWorkflowArtifactStore(
+        local_temp_root() / "surface_live_missing_connection"
+    )
+    artifact_store.save_artifact(echo_artifact())
+    artifact_store.save_deployment(
+        WorkflowDeployment(
+            id="echo.personal",
+            artifact_id="echo",
+            artifact_version=1,
+            bindings=[{"logical_source": "demo", "concrete_source": "demo.personal"}],
+        )
+    )
+    h = handlers(artifact_store)
+    h.service.register_connection(
+        ConnectionConfig(id="demo.personal", server="demo", account="personal")
+    )
+    h.service.register_specs("demo.personal", echo_tool)
+    del h.service.connections.connections["demo.personal"]
+
+    payload = asyncio.run(
+        h.validate_deployment(deployment_id="echo.personal", live_check=True)
+    )
+
+    assert payload["status"] == "unrunnable"
+    assert payload["diagnostics"][0]["code"] == "source_unreachable"
+    assert payload["diagnostics"][0]["bound_source"] == "demo.personal"
+    assert "KeyError" in payload["diagnostics"][0]["message"]
+
+
 def test_workflow_surface_records_artifact_and_deployment_save_events() -> None:
     artifact_store = FileWorkflowArtifactStore(local_temp_root() / "surface_events")
     h = handlers(artifact_store)