feat: add source auth diagnostics

docs/current_roadmap.md

@@ -108,6 +108,8 @@ auth admin are implemented. The next work is polish, not new broad surfaces.
   Google OAuth client credentials). OAuth refresh-token support and provider
   profiles are now implemented. Production secret manager integration and
   encrypted-at-rest file format remain deferred.
+- Completed source auth diagnostics: `wf source diagnose <source_id>` now reports
+  transport/auth/catalog state without exposing secret payloads.
 - Active specs:
   - [`workflow config targets and sources`](superpowers/specs/2026-06-03-workflow-config-targets-and-sources.md)
   - [`store-backed source registry`](superpowers/specs/2026-06-03-store-backed-source-registry-design.md)

docs/historical/superpowers/plans/2026-06-13-source-auth-diagnostics.md

@@ -0,0 +1,875 @@
+# Source Auth Diagnostics Implementation Plan
+
+> **For agentic workers:** REQUIRED SUB-SKILL: Use superpowers:subagent-driven-development (recommended) or superpowers:executing-plans to implement this plan task-by-task. Steps use checkbox (`- [ ]`) syntax for tracking.
+
+**Goal:** Add source-level auth and health diagnostics so users can tell whether a source is configured, authenticated, catalogued, and safe to call before running `wf cap call`.
+
+**Architecture:** Keep `wf_api` neutral by adding an optional source diagnostics provider to `WorkflowSourceAdminApi`. Implement MCP-specific checks in `wf_mcp.broker.service.source_diagnostics`, wire that provider into MCP-backed `WorkflowServer`, and expose diagnostics over JSON-RPC and CLI via `wf source diagnose`. `source inspect` should include a compact `diagnostics` block when a provider exists, but local/static servers must keep working without MCP imports.
+
+**Tech Stack:** Python 3.14, dataclasses/protocols, Pydantic JSON-RPC params, Typer CLI, pytest-asyncio, existing `AuthRecord`/`StoredAuthRecord` and MCP broker services.
+
+---
+
+## File Structure
+
+- Modify `src/wf_api/source_admin.py`: add `WorkflowSourceDiagnosticsProvider` protocol, optional `diagnostics` provider, `diagnose_source()`, and optional diagnostics in `inspect_source()`.
+- Modify `src/wf_api/surface.py`: add `diagnose_source()` to `WorkflowSourceAdminSurface`.
+- Create `src/wf_mcp/broker/service/source_diagnostics.py`: MCP-specific diagnostics provider.
+- Modify `src/wf_mcp/broker/server.py`: wire diagnostics provider when building MCP-backed server.
+- Modify `src/wf_transport_rpc_http/models.py`: add `DiagnoseSourceParams`.
+- Modify `src/wf_transport_rpc_http/methods/sources.py`: register `workflow.sources.diagnose`.
+- Modify `src/wf_transport_rpc_http/client/sources.py`: add `diagnose_source()`.
+- Modify `src/wf_cli/commands/sources.py`: add `wf source diagnose <source_id>`.
+- Add/modify tests:
+  - `tests/wf_api/test_source_admin_api.py`
+  - `tests/wf_mcp/service/test_source_diagnostics.py`
+  - `tests/wf_mcp/test_mcp_workflow_server.py`
+  - `tests/wf_transport_rpc_http/test_app.py`
+  - `tests/wf_transport_rpc_http/test_client.py`
+  - `tests/wf_cli/test_remote_target.py`
+- Modify docs:
+  - `docs/wf_cli.md`
+  - `docs/current_roadmap.md`
+
+---
+
+### Task 1: Neutral Source Diagnostics API
+
+**Files:**
+- Modify: `src/wf_api/source_admin.py`
+- Modify: `src/wf_api/surface.py`
+- Test: `tests/wf_api/test_source_admin_api.py`
+
+- [ ] **Step 1: Write neutral API tests**
+
+Create `tests/wf_api/test_source_admin_api.py` if it does not exist. Add:
+
+```python
+from __future__ import annotations
+
+import pytest
+
+from wf_api.operation_context import WorkflowOperationContext
+from wf_api.source_admin import WorkflowSourceAdminApi
+from wf_api.stores import memory_workflow_stores
+from wf_platform import CapabilityBuckets, CapabilitySource
+
+
+class _Specs:
+    def __init__(self) -> None:
+        self.capability_sources = {
+            "demo.source": CapabilitySource(
+                id="demo.source",
+                kind="connection",
+                capabilities=CapabilityBuckets(),
+                description="Demo source",
+            )
+        }
+
+    def get_qualified_spec(self, qualified_name: str):
+        raise KeyError(qualified_name)
+
+
+class _Diagnostics:
+    def __init__(self) -> None:
+        self.calls: list[str] = []
+
+    def diagnose_source(self, source_id: str) -> dict[str, object]:
+        self.calls.append(source_id)
+        return {
+            "source_id": source_id,
+            "status": "ok",
+            "auth": {"record_present": True},
+            "diagnostics": [],
+        }
+
+
+def _api(*, diagnostics: object | None = None) -> WorkflowSourceAdminApi:
+    stores = memory_workflow_stores()
+    context = WorkflowOperationContext(
+        artifact_store=stores.artifact_store,
+        draft_workspace_store=stores.draft_workspace_store,
+        run_store=stores.run_store,
+        events=None,
+        specs=_Specs(),
+        runtime=None,
+        live_sources=None,
+    )
+    return WorkflowSourceAdminApi(context, diagnostics=diagnostics)
+
+
+@pytest.mark.asyncio
+async def test_inspect_source_includes_optional_diagnostics() -> None:
+    provider = _Diagnostics()
+    payload = await _api(diagnostics=provider).inspect_source(
+        source_id="demo.source"
+    )
+
+    assert payload["id"] == "demo.source"
+    assert payload["diagnostics"]["source_id"] == "demo.source"
+    assert provider.calls == ["demo.source"]
+
+
+@pytest.mark.asyncio
+async def test_inspect_source_omits_diagnostics_without_provider() -> None:
+    payload = await _api().inspect_source(source_id="demo.source")
+
+    assert payload["id"] == "demo.source"
+    assert "diagnostics" not in payload
+
+
+@pytest.mark.asyncio
+async def test_diagnose_source_uses_provider() -> None:
+    payload = await _api(diagnostics=_Diagnostics()).diagnose_source(
+        source_id="demo.source"
+    )
+
+    assert payload["status"] == "ok"
+    assert payload["auth"]["record_present"] is True
+
+
+@pytest.mark.asyncio
+async def test_diagnose_source_without_provider_returns_basic_status() -> None:
+    payload = await _api().diagnose_source(source_id="demo.source")
+
+    assert payload == {
+        "source_id": "demo.source",
+        "status": "unknown",
+        "diagnostics": [],
+        "message": "No source diagnostics provider is configured.",
+    }
+```
+
+- [ ] **Step 2: Run tests and confirm failure**
+
+Run:
+
+```bash
+uv run pytest tests/wf_api/test_source_admin_api.py -q
+```
+
+Expected: fails because `WorkflowSourceAdminApi.__init__()` has no `diagnostics` argument and `diagnose_source()` does not exist.
+
+- [ ] **Step 3: Implement neutral protocol and API**
+
+In `src/wf_api/source_admin.py`, add:
+
+```python
+from typing import Any, Protocol
+```
+
+Then add above `WorkflowSourceAdminApi`:
+
+```python
+class WorkflowSourceDiagnosticsProvider(Protocol):
+    """Optional source-specific diagnostics provider.
+
+    Implementations may know about transport/auth/catalog details. The neutral
+    API only forwards source ids and serializes returned dictionaries.
+    """
+
+    def diagnose_source(self, source_id: str) -> dict[str, Any]: ...
+```
+
+Change the constructor:
+
+```python
+def __init__(
+    self,
+    context: WorkflowOperationContext,
+    *,
+    diagnostics: WorkflowSourceDiagnosticsProvider | None = None,
+) -> None:
+    self.context = context
+    self.diagnostics = diagnostics
+```
+
+Change `inspect_source()` to:
+
+```python
+async def inspect_source(self, *, source_id: str) -> dict[str, Any]:
+    try:
+        source = self.context.specs.capability_sources[source_id]
+    except KeyError as exc:
+        raise KeyError(f"unknown source {source_id!r}") from exc
+    payload = source.as_inventory().model_dump(mode="json")
+    if self.diagnostics is not None:
+        payload["diagnostics"] = self.diagnostics.diagnose_source(source_id)
+    return payload
+```
+
+Add:
+
+```python
+async def diagnose_source(self, *, source_id: str) -> dict[str, Any]:
+    try:
+        self.context.specs.capability_sources[source_id]
+    except KeyError as exc:
+        raise KeyError(f"unknown source {source_id!r}") from exc
+    if self.diagnostics is None:
+        return {
+            "source_id": source_id,
+            "status": "unknown",
+            "diagnostics": [],
+            "message": "No source diagnostics provider is configured.",
+        }
+    return self.diagnostics.diagnose_source(source_id)
+```
+
+In `src/wf_api/surface.py`, add to `WorkflowSourceAdminSurface`:
+
+```python
+async def diagnose_source(self, *, source_id: str) -> dict[str, Any]: ...
+```
+
+- [ ] **Step 4: Run tests and commit**
+
+Run:
+
+```bash
+uv run pytest tests/wf_api/test_source_admin_api.py -q
+uv run basedpyright --level error src/wf_api/source_admin.py src/wf_api/surface.py tests/wf_api/test_source_admin_api.py
+```
+
+Expected: tests pass and typecheck has 0 errors.
+
+Commit:
+
+```bash
+git add src/wf_api/source_admin.py src/wf_api/surface.py tests/wf_api/test_source_admin_api.py
+git commit -m "feat: add neutral source diagnostics api"
+```
+
+---
+
+### Task 2: MCP Source Diagnostics Provider
+
+**Files:**
+- Create: `src/wf_mcp/broker/service/source_diagnostics.py`
+- Modify: `src/wf_mcp/broker/server.py`
+- Test: `tests/wf_mcp/service/test_source_diagnostics.py`
+- Test: `tests/wf_mcp/test_mcp_workflow_server.py`
+
+- [ ] **Step 1: Write provider tests**
+
+Create `tests/wf_mcp/service/test_source_diagnostics.py`:
+
+```python
+from __future__ import annotations
+
+from wf_api.auth import AuthRecord
+from wf_mcp.broker.service.source_diagnostics import SourceDiagnosticsProvider
+from wf_mcp.connections import ConnectionRegistry
+from wf_mcp.models import ConnectionConfig
+from wf_sources_mcp.catalog import CatalogSnapshot
+from wf_sources_mcp.storage import FileAuthStore, FileCatalogStore
+
+
+def _connection(**metadata: object) -> ConnectionConfig:
+    return ConnectionConfig(
+        id="demo.personal",
+        server="demo",
+        account="personal",
+        enabled=True,
+        metadata={"transport": "http", "url": "https://example.test/mcp", **metadata},
+    )
+
+
+def _provider(tmp_path, connection: ConnectionConfig) -> SourceDiagnosticsProvider:
+    registry = ConnectionRegistry()
+    registry.register(connection)
+    return SourceDiagnosticsProvider(
+        connection_lookup=registry.get,
+        auth_store=FileAuthStore(tmp_path / "auth"),
+        catalog_store=FileCatalogStore(tmp_path / "catalog"),
+    )
+
+
+def test_source_diagnostics_reports_present_auth(tmp_path) -> None:
+    connection = _connection(auth_ref="demo.creds")
+    provider = _provider(tmp_path, connection)
+    provider.auth_store.save_auth(
+        AuthRecord(
+            connection_id="demo.creds",
+            scheme="oauth_refresh_token",
+            payload={
+                "client_id": "client",
+                "client_secret": "secret",
+                "refresh_token": "refresh",
+                "token_url": "https://oauth2.example.test/token",
+            },
+        )
+    )
+
+    payload = provider.diagnose_source("demo.personal")
+
+    assert payload["source_id"] == "demo.personal"
+    assert payload["status"] == "ok"
+    assert payload["auth"]["auth_ref"] == "demo.creds"
+    assert payload["auth"]["record_present"] is True
+    assert payload["auth"]["scheme"] == "oauth_refresh_token"
+    assert payload["auth"]["transport_supported"] is True
+    assert payload["diagnostics"] == []
+
+
+def test_source_diagnostics_reports_missing_auth(tmp_path) -> None:
+    payload = _provider(
+        tmp_path,
+        _connection(auth_ref="missing.creds"),
+    ).diagnose_source("demo.personal")
+
+    assert payload["status"] == "error"
+    assert payload["auth"]["record_present"] is False
+    assert payload["diagnostics"][0]["code"] == "auth_not_found"
+
+
+def test_source_diagnostics_reports_unsupported_transport_auth(tmp_path) -> None:
+    connection = ConnectionConfig(
+        id="demo.personal",
+        server="demo",
+        account="personal",
+        metadata={"transport": "stdio", "command": "demo", "auth_ref": "demo.creds"},
+    )
+    provider = _provider(tmp_path, connection)
+    provider.auth_store.save_auth(
+        AuthRecord(
+            connection_id="demo.creds",
+            scheme="oauth_refresh_token",
+            payload={
+                "client_id": "client",
+                "client_secret": "secret",
+                "refresh_token": "refresh",
+                "token_url": "https://oauth2.example.test/token",
+            },
+        )
+    )
+
+    payload = provider.diagnose_source("demo.personal")
+
+    assert payload["status"] == "error"
+    assert payload["auth"]["transport_supported"] is False
+    assert payload["diagnostics"][0]["code"] == "auth_scheme_not_supported"
+
+
+def test_source_diagnostics_reports_catalog_snapshot(tmp_path) -> None:
+    provider = _provider(tmp_path, _connection())
+    provider.catalog_store.save_catalog(
+        CatalogSnapshot(
+            connection_id="demo.personal",
+            fetched_at_epoch_ms=123,
+            max_age_seconds=60,
+        )
+    )
+
+    payload = provider.diagnose_source("demo.personal")
+
+    assert payload["catalog"] == {
+        "has_snapshot": True,
+        "fetched_at_epoch_ms": 123,
+        "max_age_seconds": 60,
+        "node_count": 0,
+        "resource_count": 0,
+        "prompt_count": 0,
+    }
+```
+
+- [ ] **Step 2: Run tests and confirm failure**
+
+Run:
+
+```bash
+uv run pytest tests/wf_mcp/service/test_source_diagnostics.py -q
+```
+
+Expected: fails because `source_diagnostics.py` does not exist.
+
+- [ ] **Step 3: Implement provider**
+
+Create `src/wf_mcp/broker/service/source_diagnostics.py`:
+
+```python
+from __future__ import annotations
+
+from collections.abc import Callable
+from dataclasses import dataclass
+from typing import Any
+
+from wf_api.auth import AuthRecord
+from wf_core import DependencyDiagnostic, DiagnosticSeverity
+from wf_sources_mcp.auth import connection_auth_diagnostic
+from wf_sources_mcp.storage import AuthStore, CatalogStore
+
+from ...models import ConnectionConfig
+
+ConnectionLookup = Callable[[str], ConnectionConfig]
+
+
+def _auth_ref(connection: ConnectionConfig) -> str | None:
+    value = connection.metadata.get("auth_ref")
+    return value if isinstance(value, str) and value else None
+
+
+def _transport_kind(connection: ConnectionConfig) -> str | None:
+    value = connection.metadata.get("transport")
+    return value if isinstance(value, str) and value else None
+
+
+def _auth_scheme_supported(
+    *,
+    transport_kind: str | None,
+    auth: AuthRecord | None,
+) -> bool:
+    if auth is None:
+        return True
+    if transport_kind == "stdio":
+        return auth.scheme == "env"
+    if transport_kind == "http":
+        return auth.scheme in {"bearer", "headers", "oauth_refresh_token"}
+    return False
+
+
+def _unsupported_auth_diagnostic(
+    *,
+    source_id: str,
+    auth_ref: str,
+    scheme: str,
+    transport_kind: str | None,
+) -> dict[str, Any]:
+    return DependencyDiagnostic(
+        severity=DiagnosticSeverity.ERROR,
+        code="auth_scheme_not_supported",
+        logical_ref=auth_ref,
+        bound_source=source_id,
+        message=(
+            f"Source {source_id!r} uses {transport_kind or 'unknown'} transport, "
+            f"but auth record {auth_ref!r} has unsupported scheme {scheme!r}."
+        ),
+        repair_hint=(
+            "Use env auth for stdio MCP sources, or bearer/headers/"
+            "oauth_refresh_token auth for HTTP MCP sources."
+        ),
+    ).model_dump(mode="json")
+
+
+@dataclass(slots=True)
+class SourceDiagnosticsProvider:
+    """MCP broker diagnostics for source auth, transport, and catalog state."""
+
+    connection_lookup: ConnectionLookup
+    auth_store: AuthStore
+    catalog_store: CatalogStore
+
+    def diagnose_source(self, source_id: str) -> dict[str, Any]:
+        connection = self.connection_lookup(source_id)
+        auth_ref = _auth_ref(connection)
+        auth = self.auth_store.load_auth(auth_ref) if auth_ref else None
+        transport_kind = _transport_kind(connection)
+        snapshot = self.catalog_store.load_catalog(source_id)
+        diagnostics: list[dict[str, Any]] = []
+
+        missing_auth = connection_auth_diagnostic(
+            connection,
+            load_auth_ref=self.auth_store.load_auth,
+        )
+        if missing_auth is not None:
+            diagnostics.append(missing_auth.model_dump(mode="json"))
+
+        transport_supported = _auth_scheme_supported(
+            transport_kind=transport_kind,
+            auth=auth,
+        )
+        if auth_ref and auth is not None and not transport_supported:
+            diagnostics.append(
+                _unsupported_auth_diagnostic(
+                    source_id=source_id,
+                    auth_ref=auth_ref,
+                    scheme=auth.scheme,
+                    transport_kind=transport_kind,
+                )
+            )
+
+        return {
+            "source_id": source_id,
+            "status": "error" if diagnostics else "ok",
+            "enabled": connection.enabled,
+            "transport": {
+                "kind": transport_kind,
+                "configured": transport_kind is not None,
+            },
+            "auth": {
+                "auth_ref": auth_ref,
+                "record_present": auth is not None if auth_ref else None,
+                "scheme": None if auth is None else auth.scheme,
+                "transport_supported": transport_supported,
+            },
+            "catalog": {
+                "has_snapshot": snapshot is not None,
+                "fetched_at_epoch_ms": None
+                if snapshot is None
+                else snapshot.fetched_at_epoch_ms,
+                "max_age_seconds": None if snapshot is None else snapshot.max_age_seconds,
+                "node_count": 0 if snapshot is None else len(snapshot.nodes),
+                "resource_count": 0 if snapshot is None else len(snapshot.resources),
+                "prompt_count": 0 if snapshot is None else len(snapshot.prompts),
+            },
+            "diagnostics": diagnostics,
+        }
+```
+
+- [ ] **Step 4: Wire provider into MCP-backed server**
+
+In `src/wf_mcp/broker/server.py`, import:
+
+```python
+from .service.source_diagnostics import SourceDiagnosticsProvider
+```
+
+Inside `workflow_server_from_service()`, construct:
+
+```python
+source_diagnostics = SourceDiagnosticsProvider(
+    connection_lookup=service.connections.get,
+    auth_store=service.auth_store,
+    catalog_store=service.catalog_store,
+)
+```
+
+Change:
+
+```python
+source_admin = WorkflowSourceAdminApi(context)
+```
+
+to:
+
+```python
+source_admin = WorkflowSourceAdminApi(
+    context,
+    diagnostics=source_diagnostics,
+)
+```
+
+If the function does not currently name `source_admin`, apply the same change where `WorkflowSourceAdminApi` is constructed.
+
+- [ ] **Step 5: Add server wiring test**
+
+In `tests/wf_mcp/test_mcp_workflow_server.py`, add:
+
+```python
+async def test_workflow_server_source_admin_reports_mcp_diagnostics(tmp_path) -> None:
+    service = _service(tmp_path)
+    payload = await service.workflow_server.source_admin.diagnose_source(
+        source_id="demo.personal"
+    )
+
+    assert payload["source_id"] == "demo.personal"
+    assert "auth" in payload
+    assert "catalog" in payload
+```
+
+If the local helper is named differently, use the existing helper that builds a `WfMcpService`/`WorkflowServer` with `demo.personal`; keep the assertions above.
+
+- [ ] **Step 6: Run tests and commit**
+
+Run:
+
+```bash
+uv run pytest tests/wf_mcp/service/test_source_diagnostics.py tests/wf_mcp/test_mcp_workflow_server.py -q
+uv run basedpyright --level error src/wf_mcp/broker/service/source_diagnostics.py src/wf_mcp/broker/server.py tests/wf_mcp/service/test_source_diagnostics.py tests/wf_mcp/test_mcp_workflow_server.py
+```
+
+Expected: tests pass and typecheck has 0 errors.
+
+Commit:
+
+```bash
+git add src/wf_mcp/broker/service/source_diagnostics.py src/wf_mcp/broker/server.py tests/wf_mcp/service/test_source_diagnostics.py tests/wf_mcp/test_mcp_workflow_server.py
+git commit -m "feat: add mcp source diagnostics provider"
+```
+
+---
+
+### Task 3: JSON-RPC Source Diagnose Method
+
+**Files:**
+- Modify: `src/wf_transport_rpc_http/models.py`
+- Modify: `src/wf_transport_rpc_http/methods/sources.py`
+- Modify: `src/wf_transport_rpc_http/client/sources.py`
+- Test: `tests/wf_transport_rpc_http/test_app.py`
+- Test: `tests/wf_transport_rpc_http/test_client.py`
+
+- [ ] **Step 1: Add RPC tests**
+
+In `tests/wf_transport_rpc_http/test_app.py`, add:
+
+```python
+async def test_rpc_diagnoses_source(tmp_path) -> None:
+    server = build_local_static_workflow_server(tmp_path)
+    client = TestClient(create_app(server))
+
+    response = client.post(
+        "/rpc",
+        json={
+            "jsonrpc": "2.0",
+            "id": 1,
+            "method": "workflow.sources.diagnose",
+            "params": {"source_id": "wf.std"},
+        },
+    )
+
+    payload = response.json()
+    assert response.status_code == 200
+    assert payload["result"]["source_id"] == "wf.std"
+    assert payload["result"]["status"] == "unknown"
+```
+
+In `tests/wf_transport_rpc_http/test_client.py`, add:
+
+```python
+async def test_rpc_client_diagnoses_source(tmp_path) -> None:
+    calls: list[tuple[str, dict[str, object]]] = []
+
+    class Client(RpcSourceAdminClientMixin):
+        async def _call(self, method: str, params: dict[str, object]):
+            calls.append((method, params))
+            return {"source_id": params["source_id"], "status": "ok"}
+
+    payload = await Client().diagnose_source(source_id="demo.personal")
+
+    assert payload == {"source_id": "demo.personal", "status": "ok"}
+    assert calls == [
+        ("workflow.sources.diagnose", {"source_id": "demo.personal"})
+    ]
+```
+
+- [ ] **Step 2: Run tests and confirm failure**
+
+Run:
+
+```bash
+uv run pytest tests/wf_transport_rpc_http/test_app.py::test_rpc_diagnoses_source tests/wf_transport_rpc_http/test_client.py::test_rpc_client_diagnoses_source -q
+```
+
+Expected: fails because model/method/client are missing.
+
+- [ ] **Step 3: Implement RPC model/method/client**
+
+In `src/wf_transport_rpc_http/models.py`, add:
+
+```python
+class DiagnoseSourceParams(RpcParamsModel):
+    source_id: str = Field(min_length=1)
+```
+
+In `src/wf_transport_rpc_http/methods/sources.py`, import `DiagnoseSourceParams` and add:
+
+```python
+@entrypoint.method(name="workflow.sources.diagnose", errors=[WorkflowRpcError])
+async def workflow_sources_diagnose(
+    params: DiagnoseSourceParams = RpcParams(),
+) -> dict[str, Any]:
+    try:
+        return await server.source_admin.diagnose_source(
+            source_id=params.source_id
+        )
+    except (ValueError, KeyError, LookupError, FileNotFoundError) as exc:
+        raise_workflow_rpc_error(exc)
+```
+
+In `src/wf_transport_rpc_http/client/sources.py`, add:
+
+```python
+async def diagnose_source(self: RpcCaller, *, source_id: str) -> dict[str, Any]:
+    return await self._call(
+        "workflow.sources.diagnose",
+        {"source_id": source_id},
+    )
+```
+
+- [ ] **Step 4: Run tests and commit**
+
+Run:
+
+```bash
+uv run pytest tests/wf_transport_rpc_http/test_app.py::test_rpc_diagnoses_source tests/wf_transport_rpc_http/test_client.py::test_rpc_client_diagnoses_source -q
+uv run basedpyright --level error src/wf_transport_rpc_http/models.py src/wf_transport_rpc_http/methods/sources.py src/wf_transport_rpc_http/client/sources.py tests/wf_transport_rpc_http/test_app.py tests/wf_transport_rpc_http/test_client.py
+```
+
+Expected: tests pass and typecheck has 0 errors.
+
+Commit:
+
+```bash
+git add src/wf_transport_rpc_http/models.py src/wf_transport_rpc_http/methods/sources.py src/wf_transport_rpc_http/client/sources.py tests/wf_transport_rpc_http/test_app.py tests/wf_transport_rpc_http/test_client.py
+git commit -m "feat: expose source diagnostics over rpc"
+```
+
+---
+
+### Task 4: CLI Source Diagnose Command
+
+**Files:**
+- Modify: `src/wf_cli/commands/sources.py`
+- Test: `tests/wf_cli/test_remote_target.py`
+
+- [ ] **Step 1: Write CLI tests**
+
+In `tests/wf_cli/test_remote_target.py`, add:
+
+```python
+def test_wf_source_diagnose_uses_rpc_url_override(monkeypatch, tmp_path) -> None:
+    captured: dict[str, str] = {}
+
+    class FakeSourceAdmin:
+        async def diagnose_source(self, *, source_id: str) -> dict[str, object]:
+            captured["source_id"] = source_id
+            return {
+                "source_id": source_id,
+                "status": "ok",
+                "auth": {
+                    "auth_ref": "demo.creds",
+                    "record_present": True,
+                    "scheme": "bearer",
+                    "transport_supported": True,
+                },
+                "diagnostics": [],
+            }
+
+    class FakeContext:
+        source_admin = FakeSourceAdmin()
+
+    monkeypatch.setattr(
+        "wf_cli.commands.sources.load_cli_context_from_typer",
+        lambda ctx: FakeContext(),
+    )
+
+    result = runner.invoke(
+        app,
+        ["--url", "http://127.0.0.1:8765/rpc", "source", "diagnose", "demo.personal"],
+    )
+
+    assert result.exit_code == 0, result.output
+    payload = json.loads(result.output)
+    assert captured == {"source_id": "demo.personal"}
+    assert payload["status"] == "ok"
+    assert payload["auth"]["scheme"] == "bearer"
+```
+
+If the file already has a `runner` or `app` helper, use the existing helper names.
+
+- [ ] **Step 2: Run test and confirm failure**
+
+Run:
+
+```bash
+uv run pytest tests/wf_cli/test_remote_target.py::test_wf_source_diagnose_uses_rpc_url_override -q
+```
+
+Expected: fails because `wf source diagnose` is not registered.
+
+- [ ] **Step 3: Implement CLI command**
+
+In `src/wf_cli/commands/sources.py`, add:
+
+```python
+@app.command("diagnose")
+def diagnose_source(
+    ctx: typer.Context,
+    source_id: Annotated[str, typer.Argument(help="Workflow source id.")],
+) -> None:
+    """Diagnose source transport, auth, and catalog state."""
+    context = load_cli_context_from_typer(ctx)
+    payload = run_cli_operation(
+        context,
+        context.source_admin.diagnose_source(source_id=source_id),
+    )
+    emit_json(payload)
+```
+
+- [ ] **Step 4: Run tests and commit**
+
+Run:
+
+```bash
+uv run pytest tests/wf_cli/test_remote_target.py::test_wf_source_diagnose_uses_rpc_url_override -q
+uv run basedpyright --level error src/wf_cli/commands/sources.py tests/wf_cli/test_remote_target.py
+```
+
+Expected: tests pass and typecheck has 0 errors.
+
+Commit:
+
+```bash
+git add src/wf_cli/commands/sources.py tests/wf_cli/test_remote_target.py
+git commit -m "feat: add source diagnose cli"
+```
+
+---
+
+### Task 5: Docs And Final Verification
+
+**Files:**
+- Modify: `docs/wf_cli.md`
+- Modify: `docs/current_roadmap.md`
+
+- [ ] **Step 1: Update CLI docs**
+
+In `docs/wf_cli.md`, add a source diagnostics subsection near the source commands:
+
+```markdown
+### Diagnose A Source
+
+Use `wf source diagnose <source_id>` to inspect source health before calling
+capabilities:
+
+```bash
+wf --config wf.config.json source diagnose gdrive.personal
+```
+
+The output reports transport kind, auth reference, whether the auth record
+exists, whether the auth scheme is compatible with the transport, catalog
+snapshot counts, and non-secret diagnostics. Secret payload values are never
+printed.
+```
+```
+
+- [ ] **Step 2: Update roadmap**
+
+In `docs/current_roadmap.md`, add a completed item under the current source/auth roadmap section:
+
+```markdown
+- Completed source auth diagnostics: `wf source diagnose <source_id>` now reports
+  transport/auth/catalog state without exposing secret payloads.
+```
+
+- [ ] **Step 3: Run focused verification**
+
+Run:
+
+```bash
+uv run pytest tests/wf_api/test_source_admin_api.py tests/wf_mcp/service/test_source_diagnostics.py tests/wf_transport_rpc_http/test_app.py tests/wf_transport_rpc_http/test_client.py tests/wf_cli/test_remote_target.py -q
+uv run ruff check src/wf_api/source_admin.py src/wf_api/surface.py src/wf_mcp/broker/service/source_diagnostics.py src/wf_mcp/broker/server.py src/wf_transport_rpc_http/models.py src/wf_transport_rpc_http/methods/sources.py src/wf_transport_rpc_http/client/sources.py src/wf_cli/commands/sources.py tests/wf_api/test_source_admin_api.py tests/wf_mcp/service/test_source_diagnostics.py tests/wf_transport_rpc_http/test_app.py tests/wf_transport_rpc_http/test_client.py tests/wf_cli/test_remote_target.py
+uv run basedpyright --level error src/wf_api/source_admin.py src/wf_api/surface.py src/wf_mcp/broker/service/source_diagnostics.py src/wf_mcp/broker/server.py src/wf_transport_rpc_http/models.py src/wf_transport_rpc_http/methods/sources.py src/wf_transport_rpc_http/client/sources.py src/wf_cli/commands/sources.py tests/wf_api/test_source_admin_api.py tests/wf_mcp/service/test_source_diagnostics.py tests/wf_transport_rpc_http/test_app.py tests/wf_transport_rpc_http/test_client.py tests/wf_cli/test_remote_target.py
+```
+
+Expected:
+
+- Focused tests pass.
+- Ruff reports `All checks passed!`.
+- Basedpyright reports `0 errors`.
+
+- [ ] **Step 4: Commit docs**
+
+```bash
+git add docs/wf_cli.md docs/current_roadmap.md
+git commit -m "docs: document source diagnostics"
+```
+
+---
+
+## Self-Review
+
+- Spec coverage: plan adds neutral API, MCP diagnostics provider, RPC/client method, CLI command, docs, and tests. It explicitly avoids secret payload output.
+- Placeholder scan: no `TBD`, `TODO`, or "similar to" placeholders remain.
+- Type consistency: method name is `diagnose_source` across API, surface, RPC client, CLI, and tests. RPC method name is `workflow.sources.diagnose`. Payload uses `diagnostics` for the diagnostic list and `auth`/`catalog` for structured summaries.
+

docs/wf_cli.md

@@ -123,6 +123,20 @@ availability, durable run counts/latest run, admin counts, auth record count,
 and desired registry count when the target exposes those surfaces. It does not
 return auth payload values, trace entries, or checkpoint state.
 
+### Diagnose A Source
+
+Use `wf source diagnose <source_id>` to inspect source health before calling
+capabilities:
+
+```bash
+wf --config wf.config.json source diagnose gdrive.personal
+```
+
+The output reports transport kind, auth reference, whether the auth record
+exists, whether the auth scheme is compatible with the transport, catalog
+snapshot counts, and non-secret diagnostics. Secret payload values are never
+printed.
+
 ## Output Policy
 
 JSON is the default output format for every command.

src/wf_api/source_admin.py

@@ -1,12 +1,22 @@
 from __future__ import annotations
 
-from typing import Any
+from typing import Any, Protocol
 
 from wf_platform import page_items
 
 from .operation_context import WorkflowOperationContext
 
 
+class WorkflowSourceDiagnosticsProvider(Protocol):
+    """Optional source-specific diagnostics provider.
+
+    Implementations may know about transport/auth/catalog details. The neutral
+    API only forwards source ids and serializes returned dictionaries.
+    """
+
+    def diagnose_source(self, source_id: str) -> dict[str, Any]: ...
+
+
 class WorkflowSourceAdminApi:
     """Read-only protocol-neutral source inventory operations.
 
@@ -15,8 +25,14 @@ class WorkflowSourceAdminApi:
     lifecycle execution.
     """
 
-    def __init__(self, context: WorkflowOperationContext) -> None:
+    def __init__(
+        self,
+        context: WorkflowOperationContext,
+        *,
+        diagnostics: WorkflowSourceDiagnosticsProvider | None = None,
+    ) -> None:
         self.context = context
+        self.diagnostics = diagnostics
 
     async def list_sources(
         self,
@@ -43,4 +59,27 @@ class WorkflowSourceAdminApi:
             source = self.context.specs.capability_sources[source_id]
         except KeyError as exc:
             raise KeyError(f"unknown source {source_id!r}") from exc
-        return source.as_inventory().model_dump(mode="json")
+        payload = source.as_inventory().model_dump(mode="json")
+        if self.diagnostics is not None:
+            try:
+                payload["diagnostics"] = self.diagnostics.diagnose_source(source_id)
+            except Exception:
+                payload["diagnostics"] = {
+                    "status": "error",
+                    "message": "Diagnostics unavailable",
+                }
+        return payload
+
+    async def diagnose_source(self, *, source_id: str) -> dict[str, Any]:
+        try:
+            self.context.specs.capability_sources[source_id]
+        except KeyError as exc:
+            raise KeyError(f"unknown source {source_id!r}") from exc
+        if self.diagnostics is None:
+            return {
+                "source_id": source_id,
+                "status": "unknown",
+                "diagnostics": [],
+                "message": "No source diagnostics provider is configured.",
+            }
+        return self.diagnostics.diagnose_source(source_id)

src/wf_api/surface.py

@@ -244,6 +244,12 @@ class WorkflowSourceAdminSurface(Protocol):
         source_id: str,
     ) -> dict[str, Any]: ...
 
+    async def diagnose_source(
+        self,
+        *,
+        source_id: str,
+    ) -> dict[str, Any]: ...
+
 
 class WorkflowAdminSurface(Protocol):
     """Read-only connection/config admin methods exposed by platform frontends."""

src/wf_cli/commands/sources.py

@@ -56,3 +56,17 @@ def inspect_source(
         context.source_admin.inspect_source(source_id=source_id),
     )
     emit_json(payload)
+
+
+@app.command("diagnose")
+def diagnose_source(
+    ctx: typer.Context,
+    source_id: Annotated[str, typer.Argument(help="Workflow source id.")],
+) -> None:
+    """Diagnose source transport, auth, and catalog state."""
+    context = load_cli_context_from_typer(ctx)
+    payload = run_cli_operation(
+        context,
+        context.source_admin.diagnose_source(source_id=source_id),
+    )
+    emit_json(payload)

src/wf_mcp/broker/server.py

@@ -22,6 +22,7 @@ from .prompts import register_broker_prompts
 from .resources import register_broker_resources
 from .service import WfMcpService
 from .service.auth_admin import McpAuthAdminProvider
+from .service.source_diagnostics import SourceDiagnosticsProvider
 from .service.source_registry_admin import SourceRegistryAdminProvider
 from .service.workflow_operation_context import context_from_service
 from .tools import register_broker_tools
@@ -64,7 +65,15 @@ def workflow_server_from_service(
 
     context = context_from_service(service)
     api: WorkflowApi = durable_workflow_api(context)
-    source_admin = WorkflowSourceAdminApi(context)
+    source_diagnostics = SourceDiagnosticsProvider(
+        connection_lookup=service.connections.get,
+        auth_store=service.auth_store or service.store,
+        catalog_store=service.catalog_store or service.store,
+    )
+    source_admin = WorkflowSourceAdminApi(
+        context,
+        diagnostics=source_diagnostics,
+    )
     admin = WorkflowAdminApi(
         connections=service.connection_service,
         events=service.events,

src/wf_mcp/broker/service/source_diagnostics.py

@@ -0,0 +1,159 @@
+from __future__ import annotations
+
+from collections.abc import Callable
+from dataclasses import dataclass
+from typing import Any
+
+from wf_artifacts import DependencyDiagnostic, DiagnosticSeverity
+from wf_sources_mcp.storage import AuthStore, CatalogStore
+
+from ...models import ConnectionConfig
+
+ConnectionLookup = Callable[[str], ConnectionConfig]
+
+
+def _auth_ref(connection: ConnectionConfig) -> str | None:
+    value = connection.metadata.get("auth_ref")
+    return value if isinstance(value, str) and value else None
+
+
+def _transport_kind(connection: ConnectionConfig) -> str | None:
+    value = connection.metadata.get("transport")
+    return value if isinstance(value, str) and value else None
+
+
+def _auth_scheme_supported(
+    *,
+    transport_kind: str | None,
+    scheme: str | None,
+) -> bool:
+    if scheme is None:
+        return True
+    if transport_kind == "stdio":
+        return scheme == "env"
+    if transport_kind == "http":
+        return scheme in {"bearer", "headers", "oauth_refresh_token"}
+    return False
+
+
+def _unsupported_auth_diagnostic(
+    *,
+    source_id: str,
+    auth_ref: str,
+    scheme: str,
+    transport_kind: str | None,
+) -> dict[str, Any]:
+    return DependencyDiagnostic(
+        severity=DiagnosticSeverity.ERROR,
+        code="auth_scheme_not_supported",
+        logical_ref=auth_ref,
+        bound_source=source_id,
+        message=(
+            f"Source {source_id!r} uses {transport_kind or 'unknown'} transport, "
+            f"but auth record {auth_ref!r} has unsupported scheme {scheme!r}."
+        ),
+        repair_hint=(
+            "Use env auth for stdio MCP sources, or bearer/headers/"
+            "oauth_refresh_token auth for HTTP MCP sources."
+        ),
+    ).model_dump(mode="json")
+
+
+def _missing_auth_diagnostic(
+    *,
+    source_id: str,
+    auth_ref: str,
+) -> dict[str, Any]:
+    return DependencyDiagnostic(
+        severity=DiagnosticSeverity.ERROR,
+        code="auth_not_found",
+        logical_ref=auth_ref,
+        bound_source=source_id,
+        message=(
+            f"Source {source_id!r} references auth record {auth_ref!r}, "
+            "but no auth record was found."
+        ),
+        repair_hint=(
+            "Add an auth record for this auth_ref, update the source auth_ref, "
+            "or bind the deployment to a source that does not require it."
+        ),
+    ).model_dump(mode="json")
+
+
+def _missing_transport_diagnostic(*, source_id: str) -> dict[str, Any]:
+    return DependencyDiagnostic(
+        severity=DiagnosticSeverity.ERROR,
+        code="source_transport_missing",
+        logical_ref=source_id,
+        bound_source=source_id,
+        message=f"Source {source_id!r} has no MCP transport configured.",
+        repair_hint=(
+            "Configure the source with an MCP transport such as stdio or http, "
+            "then apply or restart the server."
+        ),
+    ).model_dump(mode="json")
+
+
+@dataclass(slots=True)
+class SourceDiagnosticsProvider:
+    """MCP broker diagnostics for source auth, transport, and catalog state."""
+
+    connection_lookup: ConnectionLookup
+    auth_store: AuthStore
+    catalog_store: CatalogStore
+
+    def diagnose_source(self, source_id: str) -> dict[str, Any]:
+        connection = self.connection_lookup(source_id)
+        auth_ref = _auth_ref(connection)
+        auth = self.auth_store.load_auth(auth_ref) if auth_ref else None
+        transport_kind = _transport_kind(connection)
+        snapshot = self.catalog_store.load_catalog(source_id)
+        diagnostics: list[dict[str, Any]] = []
+
+        if transport_kind is None:
+            diagnostics.append(_missing_transport_diagnostic(source_id=source_id))
+        if auth_ref is not None and auth is None:
+            diagnostics.append(
+                _missing_auth_diagnostic(source_id=source_id, auth_ref=auth_ref)
+            )
+
+        transport_supported = _auth_scheme_supported(
+            transport_kind=transport_kind,
+            scheme=None if auth is None else auth.scheme,
+        )
+        if auth_ref and auth is not None and not transport_supported:
+            diagnostics.append(
+                _unsupported_auth_diagnostic(
+                    source_id=source_id,
+                    auth_ref=auth_ref,
+                    scheme=auth.scheme,
+                    transport_kind=transport_kind,
+                )
+            )
+
+        return {
+            "source_id": source_id,
+            "status": "error" if diagnostics else "ok",
+            "enabled": connection.enabled,
+            "transport": {
+                "kind": transport_kind,
+                "configured": transport_kind is not None,
+            },
+            "auth": {
+                "auth_ref": auth_ref,
+                "record_present": auth is not None if auth_ref else None,
+                "scheme": None if auth is None else auth.scheme,
+                "transport_supported": transport_supported,
+            },
+            "catalog": {
+                "has_snapshot": snapshot is not None,
+                "fetched_at_epoch_ms": None
+                if snapshot is None
+                else snapshot.fetched_at_epoch_ms,
+                "max_age_seconds": None if snapshot is None else snapshot.max_age_seconds,
+                "node_count": 0 if snapshot is None else len(snapshot.nodes),
+                "resource_count": 0 if snapshot is None else len(snapshot.resources),
+                "prompt_count": 0 if snapshot is None else len(snapshot.prompts),
+            },
+            "diagnostics": diagnostics,
+        }

src/wf_transport_rpc_http/client/sources.py

@@ -27,3 +27,9 @@ class RpcSourceAdminClientMixin:
             "workflow.sources.inspect",
             {"source_id": source_id},
         )
+
+    async def diagnose_source(self: RpcCaller, *, source_id: str) -> dict[str, Any]:
+        return await self._call(
+            "workflow.sources.diagnose",
+            {"source_id": source_id},
+        )

src/wf_transport_rpc_http/methods/sources.py

@@ -7,7 +7,7 @@ import fastapi_jsonrpc as jsonrpc
 from wf_server import WorkflowServer
 
 from ..errors import WorkflowRpcError, raise_workflow_rpc_error
-from ..models import InspectSourceParams, ListSourcesParams
+from ..models import DiagnoseSourceParams, InspectSourceParams, ListSourcesParams
 from ..params import RpcParams
 
 
@@ -37,3 +37,14 @@ def register_methods(
             return await server.source_admin.inspect_source(source_id=params.source_id)
         except (ValueError, KeyError, LookupError, FileNotFoundError) as exc:
             raise_workflow_rpc_error(exc)
+
+    @entrypoint.method(name="workflow.sources.diagnose", errors=[WorkflowRpcError])
+    async def workflow_sources_diagnose(
+        params: DiagnoseSourceParams = RpcParams(),
+    ) -> dict[str, Any]:
+        try:
+            return await server.source_admin.diagnose_source(
+                source_id=params.source_id
+            )
+        except (ValueError, KeyError, LookupError, FileNotFoundError) as exc:
+            raise_workflow_rpc_error(exc)

src/wf_transport_rpc_http/models.py

@@ -50,6 +50,10 @@ class InspectSourceParams(RpcParamsModel):
     source_id: str = Field(min_length=1)
 
 
+class DiagnoseSourceParams(RpcParamsModel):
+    source_id: str = Field(min_length=1)
+
+
 class InspectCapabilityParams(RpcParamsModel):
     qualified_name: str = Field(min_length=1)
 

tests/wf_api/test_source_admin_api.py

@@ -9,6 +9,7 @@ from wf_api import WorkflowSourceAdminApi, WorkflowSourceAdminSurface
 from wf_api.models import RawWorkflowPlan
 from wf_api.operation_context import WorkflowOperationContext
 from wf_api.saved_subgraphs import SavedSubgraphTree
+from wf_api.source_admin import WorkflowSourceDiagnosticsProvider
 from wf_artifacts import WorkflowArtifact, WorkflowDeployment
 from wf_authoring import NodeSpec
 from wf_core import RunState
@@ -152,3 +153,115 @@ def test_source_admin_api_satisfies_surface_protocol() -> None:
     api: WorkflowSourceAdminSurface = _api(_source("demo.personal"))
 
     assert api is not None
+
+
+class _Diagnostics:
+    def __init__(self) -> None:
+        self.calls: list[str] = []
+
+    def diagnose_source(self, source_id: str) -> dict[str, object]:
+        self.calls.append(source_id)
+        return {
+            "source_id": source_id,
+            "status": "ok",
+            "auth": {"record_present": True},
+            "diagnostics": [],
+        }
+
+
+def _api_with_diagnostics(
+    *sources: CapabilitySource,
+    diagnostics: WorkflowSourceDiagnosticsProvider | None = None,
+) -> WorkflowSourceAdminApi:
+    provider = StaticSpecProvider({source.id: source for source in sources})
+    return WorkflowSourceAdminApi(
+        WorkflowOperationContext(
+            artifact_store=None,
+            draft_workspace_store=None,
+            run_store=None,
+            events=DummyEvents(),
+            specs=provider,
+            runtime=DummyRuntime(),
+            live_sources=None,
+        ),
+        diagnostics=diagnostics,
+    )
+
+
+def test_inspect_source_includes_optional_diagnostics() -> None:
+    provider = _Diagnostics()
+    payload = asyncio.run(
+        _api_with_diagnostics(
+            _source("demo.personal"),
+            diagnostics=provider,
+        ).inspect_source(source_id="demo.personal")
+    )
+
+    assert payload["id"] == "demo.personal"
+    assert payload["diagnostics"]["source_id"] == "demo.personal"
+    assert provider.calls == ["demo.personal"]
+
+
+def test_inspect_source_omits_diagnostics_without_provider() -> None:
+    payload = asyncio.run(
+        _api_with_diagnostics(_source("demo.personal")).inspect_source(
+            source_id="demo.personal"
+        )
+    )
+
+    assert payload["id"] == "demo.personal"
+    assert "diagnostics" not in payload
+
+
+class _BrokenDiagnostics:
+    def diagnose_source(self, source_id: str) -> dict[str, object]:
+        raise RuntimeError("diagnostics exploded")
+
+
+def test_inspect_source_tolerates_diagnostics_provider_failure() -> None:
+    payload = asyncio.run(
+        _api_with_diagnostics(
+            _source("demo.personal"),
+            diagnostics=_BrokenDiagnostics(),
+        ).inspect_source(source_id="demo.personal")
+    )
+
+    assert payload["id"] == "demo.personal"
+    assert payload["diagnostics"]["status"] == "error"
+    assert "Diagnostics unavailable" in payload["diagnostics"]["message"]
+
+
+def test_diagnose_source_uses_provider() -> None:
+    payload = asyncio.run(
+        _api_with_diagnostics(
+            _source("demo.personal"),
+            diagnostics=_Diagnostics(),
+        ).diagnose_source(source_id="demo.personal")
+    )
+
+    assert payload["status"] == "ok"
+    assert payload["auth"]["record_present"] is True
+
+
+def test_diagnose_source_without_provider_returns_basic_status() -> None:
+    payload = asyncio.run(
+        _api_with_diagnostics(_source("demo.personal")).diagnose_source(
+            source_id="demo.personal"
+        )
+    )
+
+    assert payload == {
+        "source_id": "demo.personal",
+        "status": "unknown",
+        "diagnostics": [],
+        "message": "No source diagnostics provider is configured.",
+    }
+
+
+def test_diagnose_source_unknown_raises_key_error() -> None:
+    with pytest.raises(KeyError, match="unknown source 'missing.source'"):
+        asyncio.run(
+            _api_with_diagnostics(_source("demo.personal")).diagnose_source(
+                source_id="missing.source"
+            )
+        )

tests/wf_cli/test_remote_target.py

@@ -30,6 +30,9 @@ class BrokenSourceAdmin:
     async def inspect_source(self, *, source_id: str) -> dict[str, Any]:
         raise RuntimeError(f"broken source admin for {source_id}")
 
+    async def diagnose_source(self, *, source_id: str) -> dict[str, Any]:
+        raise RuntimeError(f"broken source admin for {source_id}")
+
 
 def test_load_cli_context_uses_rpc_client_for_rpc_http_target(tmp_path) -> None:
     config_path = tmp_path / "wf.json"
@@ -815,3 +818,19 @@ def test_wf_draft_delete_succeeds_with_confirm(monkeypatch, tmp_path) -> None:
     payload = json.loads(result.output)
     assert payload["workspace_id"] == "delete-me"
     assert payload["deleted"] is True
+
+
+def test_wf_source_diagnose_uses_rpc_url_override(monkeypatch, tmp_path) -> None:
+    server = build_local_static_workflow_server(tmp_path / "store")
+    _patch_rpc_client_to_server(monkeypatch, server)
+    config_path = tmp_path / "wf.json"
+    config_path.write_text('{"version": 1}', encoding="utf-8")
+    runner = CliRunner()
+    base_args = ["--config", str(config_path), "--url", "http://test/rpc"]
+
+    result = runner.invoke(app, [*base_args, "source", "diagnose", "wf.std"])
+
+    assert result.exit_code == 0, result.output
+    payload = json.loads(result.output)
+    assert payload["source_id"] == "wf.std"
+    assert payload["status"] == "unknown"

tests/wf_mcp/service/test_source_diagnostics.py

@@ -0,0 +1,131 @@
+from __future__ import annotations
+
+from wf_mcp.broker.service.source_diagnostics import SourceDiagnosticsProvider
+from wf_mcp.connections import ConnectionRegistry
+from wf_mcp.models import ConnectionConfig
+from wf_sources_mcp.auth import AuthRecord
+from wf_sources_mcp.catalog import CatalogSnapshot
+from wf_sources_mcp.storage import FileAuthStore, FileCatalogStore
+
+
+def _connection(**metadata: object) -> ConnectionConfig:
+    return ConnectionConfig(
+        id="demo.personal",
+        server="demo",
+        account="personal",
+        enabled=True,
+        metadata={"transport": "http", "url": "https://example.test/mcp", **metadata},
+    )
+
+
+def _provider(tmp_path, connection: ConnectionConfig) -> SourceDiagnosticsProvider:
+    registry = ConnectionRegistry()
+    registry.register(connection)
+    return SourceDiagnosticsProvider(
+        connection_lookup=registry.get,
+        auth_store=FileAuthStore(tmp_path / "auth"),
+        catalog_store=FileCatalogStore(tmp_path / "catalog"),
+    )
+
+
+def test_source_diagnostics_reports_present_auth(tmp_path) -> None:
+    connection = _connection(auth_ref="demo.creds")
+    provider = _provider(tmp_path, connection)
+    provider.auth_store.save_auth(
+        AuthRecord(
+            connection_id="demo.creds",
+            scheme="oauth_refresh_token",
+            payload={
+                "client_id": "client",
+                "client_secret": "secret",
+                "refresh_token": "refresh",
+                "token_url": "https://oauth2.example.test/token",
+            },
+        )
+    )
+
+    payload = provider.diagnose_source("demo.personal")
+
+    assert payload["source_id"] == "demo.personal"
+    assert payload["status"] == "ok"
+    assert payload["auth"]["auth_ref"] == "demo.creds"
+    assert payload["auth"]["record_present"] is True
+    assert payload["auth"]["scheme"] == "oauth_refresh_token"
+    assert payload["auth"]["transport_supported"] is True
+    assert payload["diagnostics"] == []
+
+
+def test_source_diagnostics_reports_missing_auth(tmp_path) -> None:
+    payload = _provider(
+        tmp_path,
+        _connection(auth_ref="missing.creds"),
+    ).diagnose_source("demo.personal")
+
+    assert payload["status"] == "error"
+    assert payload["auth"]["record_present"] is False
+    assert payload["diagnostics"][0]["code"] == "auth_not_found"
+
+
+def test_source_diagnostics_reports_missing_transport(tmp_path) -> None:
+    connection = ConnectionConfig(
+        id="demo.personal",
+        server="demo",
+        account="personal",
+        metadata={},
+    )
+
+    payload = _provider(tmp_path, connection).diagnose_source("demo.personal")
+
+    assert payload["status"] == "error"
+    assert payload["transport"]["configured"] is False
+    assert payload["diagnostics"][0]["code"] == "source_transport_missing"
+
+
+def test_source_diagnostics_reports_unsupported_transport_auth(tmp_path) -> None:
+    connection = ConnectionConfig(
+        id="demo.personal",
+        server="demo",
+        account="personal",
+        metadata={"transport": "stdio", "command": "demo", "auth_ref": "demo.creds"},
+    )
+    provider = _provider(tmp_path, connection)
+    provider.auth_store.save_auth(
+        AuthRecord(
+            connection_id="demo.creds",
+            scheme="oauth_refresh_token",
+            payload={
+                "client_id": "client",
+                "client_secret": "secret",
+                "refresh_token": "refresh",
+                "token_url": "https://oauth2.example.test/token",
+            },
+        )
+    )
+
+    payload = provider.diagnose_source("demo.personal")
+
+    assert payload["status"] == "error"
+    assert payload["auth"]["transport_supported"] is False
+    assert payload["diagnostics"][0]["code"] == "auth_scheme_not_supported"
+
+
+def test_source_diagnostics_reports_catalog_snapshot(tmp_path) -> None:
+    provider = _provider(tmp_path, _connection())
+    provider.catalog_store.save_catalog(
+        CatalogSnapshot(
+            connection_id="demo.personal",
+            fetched_at_epoch_ms=123,
+            max_age_seconds=60,
+        )
+    )
+
+    payload = provider.diagnose_source("demo.personal")
+
+    assert payload["catalog"] == {
+        "has_snapshot": True,
+        "fetched_at_epoch_ms": 123,
+        "max_age_seconds": 60,
+        "node_count": 0,
+        "resource_count": 0,
+        "prompt_count": 0,
+    }

tests/wf_mcp/test_mcp_workflow_server.py

@@ -163,3 +163,24 @@ async def test_workflow_server_from_service_uses_focused_auth_store(tmp_path) ->
     result = await server.admin.inspect_auth_record("drive.work")
 
     assert result["id"] == "drive.work"
+
+
+async def test_workflow_server_source_admin_reports_mcp_diagnostics(tmp_path) -> None:
+    config = BrokerConfig(
+        store_root=tmp_path / "store",
+        connections=[
+            ConnectionConfig(id="demo.default", server="demo", account="default")
+        ],
+    )
+    service = build_service_from_config(config)
+    server = workflow_server_from_service(
+        service,
+        config=config,
+        source_registry_store=FileSourceRegistryStore(config.store_root),
+    )
+
+    payload = await server.source_admin.diagnose_source(source_id="demo.default")
+
+    assert payload["source_id"] == "demo.default"
+    assert "auth" in payload
+    assert "catalog" in payload

tests/wf_transport_rpc_http/test_app.py

@@ -616,3 +616,18 @@ async def test_rpc_runs_workflow_from_python_source_capability(tmp_path) -> None
     assert deployment["result"]["deployment_id"] == "python_echo.default"
     assert run["result"]["outcome"] == "ok"
     assert run["result"]["output"] == {"echoed": "hello workflow"}
+
+
+async def test_rpc_diagnoses_source(tmp_path) -> None:
+    server = build_local_static_workflow_server(tmp_path / "store")
+    app = create_rpc_app(server)
+    transport = httpx.ASGITransport(app=app)
+    async with httpx.AsyncClient(transport=transport, base_url="http://test") as client:
+        payload = await _rpc(
+            client,
+            "workflow.sources.diagnose",
+            {"source_id": "wf.std"},
+        )
+
+    assert payload["result"]["source_id"] == "wf.std"
+    assert payload["result"]["status"] == "unknown"

tests/wf_transport_rpc_http/test_client.py

@@ -7,6 +7,7 @@ from wf_api.surface import WorkflowDraftSurface
 from wf_core import END
 from wf_server import build_local_static_workflow_server
 from wf_transport_rpc_http import RpcWorkflowApiClient, create_rpc_app
+from wf_transport_rpc_http.client.sources import RpcSourceAdminClientMixin
 
 
 def _constant_plan() -> RawWorkflowPlan:
@@ -412,3 +413,19 @@ async def test_rpc_client_lists_runs(tmp_path) -> None:
 
     assert listed["total"] == 1
     assert listed["runs"][0]["run_id"] == started["run_id"]
+
+
+async def test_rpc_client_diagnoses_source(tmp_path) -> None:
+    calls: list[tuple[str, dict[str, object]]] = []
+
+    class Client(RpcSourceAdminClientMixin):
+        async def _call(self, method: str, params: dict[str, object]):
+            calls.append((method, params))
+            return {"source_id": params["source_id"], "status": "ok"}
+
+    payload = await Client().diagnose_source(source_id="demo.personal")
+
+    assert payload == {"source_id": "demo.personal", "status": "ok"}
+    assert calls == [
+        ("workflow.sources.diagnose", {"source_id": "demo.personal"})
+    ]