test org 3

tests/wf_mcp/service/conftest.py

@@ -0,0 +1,113 @@
+from __future__ import annotations
+
+import asyncio
+import shutil
+from typing import Any, cast
+
+from wf_artifacts import FileDraftWorkspaceStore, WorkflowDeployment
+from wf_authoring import NodeSpec, build_async_registry, node
+from wf_core import END, NodeUse, RunStatus, RuntimeContext
+from wf_mcp.broker import WfMcpService
+from wf_mcp.capabilities import DiscoveredTool
+from wf_mcp.models import AuthRecord, ConnectionConfig, RawWorkflowPlan
+from wf_mcp.runtime import ToolExecutor
+from wf_mcp.sdk import ToolCallResult
+from wf_mcp.shared.errors import error_payload
+from wf_mcp.storage import FileStore
+from wf_platform import (
+    CapabilityBuckets,
+    CapabilitySource,
+    SourceVisibility,
+)
+
+from ..test_support import (
+    EchoInput,
+    EchoOutput,
+    FailingDiscoveryAdapter,
+    FakeAdapter,
+    echo_tool,
+    finalize_tool,
+    input_binding,
+    local_temp_root,
+    output_binding,
+)
+
+
+@node(name="foo.bar")
+def pro_dotted_echo_tool(payload: EchoInput) -> EchoOutput:
+    return EchoOutput(echoed=f"pro:{payload.text}")
+
+
+class ContentOnlyOutputAdapter(FakeAdapter):
+    """Adapter fixture for MCP tools that expose the raw content envelope."""
+
+    async def list_tools(
+        self,
+        connection: ConnectionConfig,
+        auth: AuthRecord | None,
+    ) -> list[DiscoveredTool]:
+        return [
+            DiscoveredTool(
+                name="echo_tool",
+                title="Echo Tool",
+                description="Echo text back",
+                input_schema={
+                    "type": "object",
+                    "properties": {"message": {"type": "string"}},
+                    "required": ["message"],
+                },
+                output_schema={
+                    "type": "object",
+                    "properties": {"content": {"type": "array"}},
+                    "required": ["content"],
+                },
+            )
+        ]
+
+    async def call_tool(
+        self,
+        connection: ConnectionConfig,
+        auth: AuthRecord | None,
+        tool_name: str,
+        payload: dict[str, Any],
+    ) -> ToolCallResult:
+        message = payload.get("message", "")
+        return ToolCallResult(
+            outcome="ok",
+            output={"content": [{"type": "text", "text": f"Echo: {message}"}]},
+        )
+
+
+def single_echo_plan(plan_name: str, node_name: str) -> RawWorkflowPlan:
+    return raw_plan(
+        name=plan_name,
+        input_schema={
+            "type": "object",
+            "properties": {"text": {"type": "string"}},
+            "required": ["text"],
+        },
+        state_schema={"fields": {"echoed": {"type": "string"}}},
+        output_schema={
+            "type": "object",
+            "properties": {"echoed": {"type": "string"}},
+            "required": ["echoed"],
+        },
+        start="echo",
+        nodes=[
+            {
+                "id": "echo",
+                "type": "node",
+                "node": node_name,
+                "input": [input_binding("input.text", "text")],
+                "output": [output_binding("echoed", "state.echoed")],
+            }
+        ],
+        edges=[
+            {"from": "echo", "outcome": "ok", "to": END},
+        ],
+    )
+
+
+def raw_plan(**payload: object) -> RawWorkflowPlan:
+    """Parse JSON-shaped workflow input through the public typed boundary."""
+    return RawWorkflowPlan.model_validate(payload)

tests/wf_mcp/service/test_adapters.py

@@ -0,0 +1,59 @@
+from __future__ import annotations
+
+import asyncio
+
+from wf_mcp.broker import WfMcpService
+from wf_mcp.models import ConnectionConfig
+from wf_mcp.storage import FileStore
+
+from ..test_support import (
+    FakeAdapter,
+    local_temp_root,
+)
+from .conftest import ContentOnlyOutputAdapter
+
+
+def test_service_catalog_preserves_json_schema_description_metadata() -> None:
+    service = WfMcpService(store=FileStore(local_temp_root() / "schema_doc_store"))
+    service.register_connection(
+        ConnectionConfig(id="demo.personal", server="demo", account="personal")
+    )
+    service.register_adapter("demo", FakeAdapter())
+
+    asyncio.run(service.refresh_connection_catalog("demo.personal"))
+
+    payload = service.get_catalog().as_payload()
+    node = payload["nodes"][0]
+    assert node["input_schema"]["properties"]["text"]["description"] == "Text to echo"
+
+
+def test_service_preserves_content_only_tool_output_schema_for_workflows() -> None:
+    service = WfMcpService(store=FileStore(local_temp_root() / "content_only_store"))
+    service.register_connection(
+        ConnectionConfig(
+            id="everything.default", server="everything", account="default"
+        )
+    )
+    service.register_adapter("everything", ContentOnlyOutputAdapter())
+
+    asyncio.run(service.refresh_connection_catalog("everything.default"))
+
+    payload = service.get_catalog().as_payload()
+    node = payload["nodes"][0]
+    assert "content" in node["output_schema"]["properties"]
+    assert node["output_schema"]["required"] == ["content"]
+
+
+def test_service_wrapped_tool_adapter_model_validates_simple_schema_types() -> None:
+    service = WfMcpService(store=FileStore(local_temp_root() / "adapter_model_store"))
+    service.register_connection(
+        ConnectionConfig(id="demo.personal", server="demo", account="personal")
+    )
+    service.register_adapter("demo", FakeAdapter())
+
+    asyncio.run(service.refresh_connection_catalog("demo.personal"))
+
+    source = service.capability_sources["demo.personal"]
+    spec = source.capabilities.node_specs["demo.personal.echo_tool"]
+    assert spec.input_model.model_fields["text"].annotation is str
+    assert spec.output_model.model_fields["echoed"].annotation is str

tests/wf_mcp/service/test_catalog.py

@@ -0,0 +1,298 @@
+from __future__ import annotations
+
+import asyncio
+import shutil
+
+from wf_artifacts import FileDraftWorkspaceStore
+from wf_authoring import NodeSpec, node
+from wf_core import RunStatus
+from wf_mcp.broker import WfMcpService
+from wf_mcp.models import ConnectionConfig
+from wf_mcp.storage import FileStore
+from wf_platform import (
+    CapabilityBuckets,
+    CapabilitySource,
+    SourceVisibility,
+)
+
+from ..test_support import (
+    FakeAdapter,
+    echo_tool,
+    finalize_tool,
+    local_temp_root,
+)
+from .conftest import single_echo_plan
+
+
+def test_service_builds_namespaced_catalog() -> None:
+    service = WfMcpService(store=FileStore(local_temp_root() / "catalog_store"))
+    service.register_connection(
+        ConnectionConfig(id="demo.personal", server="demo", account="personal")
+    )
+    service.register_specs("demo.personal", echo_tool, finalize_tool)
+
+    payload = service.get_catalog().as_payload()
+    names = [node["qualified_name"] for node in payload["nodes"]]
+
+    assert names == [
+        "demo.personal.echo_tool",
+        "demo.personal.finalize_tool",
+    ]
+
+
+def test_service_rejects_reserved_connection_ids() -> None:
+    service = WfMcpService(store=FileStore(local_temp_root() / "reserved_ids_store"))
+
+    for connection_id in ("wf.admin", "wf.mcp"):
+        try:
+            service.register_connection(
+                ConnectionConfig(id=connection_id, server="wf", account="reserved")
+            )
+        except ValueError as exc:
+            assert connection_id in str(exc)
+            assert "reserved by wf-mcp" in str(exc)
+        else:
+            raise AssertionError(f"expected {connection_id!r} to be rejected")
+
+
+def test_service_installs_builtin_stdlib_specs_by_default() -> None:
+    service = WfMcpService(store=FileStore(local_temp_root() / "builtin_store"))
+
+    assert (
+        "wf.std.runtime_error"
+        in service.capability_sources["wf.std"].capabilities.node_specs
+    )
+    assert "wf.mcp" not in service.capability_sources
+
+
+def test_service_installs_default_draft_workspace_store() -> None:
+    root = local_temp_root() / "service_default_draft_workspace_store"
+    service = WfMcpService(store=FileStore(root))
+
+    assert isinstance(service.draft_workspace_store, FileDraftWorkspaceStore)
+    assert service.draft_workspace_store.root == root
+
+
+def test_service_registers_empty_source_for_connection_without_catalog() -> None:
+    service = WfMcpService(store=FileStore(local_temp_root() / "empty_source"))
+
+    service.register_connection(
+        ConnectionConfig(id="demo.personal", server="demo", account="personal")
+    )
+
+    source = service.capability_sources["demo.personal"]
+    assert source.enabled is True
+    assert source.capabilities.node_specs == {}
+    assert source.description == "No catalog loaded for demo.personal."
+
+
+def test_service_lists_all_capability_sources_with_owned_capability_names() -> None:
+    service = WfMcpService(store=FileStore(local_temp_root() / "source_inventory"))
+
+    sources = service.list_sources()
+    sources_by_id = {source["id"]: source for source in sources}
+
+    std_source = sources_by_id["wf.std"]
+    assert "wf.std.runtime_error" in std_source["capabilities"]["node_specs"]
+    assert set(std_source["capabilities"]["reducers"]) == {
+        "wf.std.append",
+        "wf.std.max",
+        "wf.std.merge_object",
+        "wf.std.replace",
+        "wf.std.set_union",
+        "wf.std.add",
+    }
+    assert std_source["capabilities"]["tools"] == []
+    assert std_source["reducer_count"] == 6
+
+    admin_source = sources_by_id["wf.admin"]
+    assert admin_source["visibility"]["planner"] is False
+    assert "wf.admin.list_sources" in admin_source["capabilities"]["tools"]
+
+
+def test_service_lists_compact_source_summaries() -> None:
+    service = WfMcpService(store=FileStore(local_temp_root() / "source_summaries"))
+
+    payload = service.list_source_summaries(limit=1)
+
+    assert len(payload["sources"]) == 1
+    assert payload["total"] >= 2
+    assert payload["next_cursor"] == "1"
+    assert "capabilities" not in payload["sources"][0]
+
+    full_page = service.list_source_summaries(limit=100)
+    sources_by_id = {source["id"]: source for source in full_page["sources"]}
+    std_source = sources_by_id["wf.std"]
+    assert "wf.std.coalesce" in std_source["preview"]["node_specs"]
+    assert std_source["has_more"]["node_specs"] is True
+
+
+def test_wf_std_source_contains_authoring_ops() -> None:
+    service = WfMcpService(store=FileStore(local_temp_root() / "stdlib_source_store"))
+    specs = service.capability_sources["wf.std"].capabilities.node_specs
+
+    expected = {
+        "wf.std.coalesce",
+        "wf.std.default_if_none",
+        "wf.std.constant",
+        "wf.std.pick_key",
+        "wf.std.pick_path",
+        "wf.std.project_fields",
+        "wf.std.rename_fields",
+        "wf.std.truthy",
+        "wf.std.runtime_error",
+        "wf.std.first_item",
+        "wf.std.first_item_or_none",
+        "wf.std.first_item_maybe",
+        "wf.std.last_item",
+        "wf.std.last_item_or_none",
+        "wf.std.length",
+        "wf.std.is_empty",
+        "wf.std.filter_items",
+        "wf.std.filter_items_present",
+        "wf.std.extract_field",
+        "wf.std.concat",
+    }
+    assert set(specs) == expected
+
+
+def test_wf_std_source_contains_builtin_reducers() -> None:
+    service = WfMcpService(store=FileStore(local_temp_root() / "stdlib_reducer_store"))
+    reducers = service.capability_sources["wf.std"].capabilities.reducers
+
+    assert set(reducers) == {
+        "wf.std.replace",
+        "wf.std.append",
+        "wf.std.max",
+        "wf.std.merge_object",
+        "wf.std.set_union",
+        "wf.std.add",
+    }
+
+
+def test_service_sources_have_visibility_and_capability_buckets() -> None:
+    service = WfMcpService(store=FileStore(local_temp_root() / "source_shape_store"))
+
+    std_source = service.capability_sources["wf.std"]
+
+    assert std_source.id == "wf.std"
+    assert std_source.kind == "system"
+    assert std_source.visibility.planner is True
+    assert std_source.visibility.mcp_client is True
+    assert std_source.visibility.admin_dashboard is True
+    assert "wf.std.runtime_error" in std_source.capabilities.node_specs
+    assert not std_source.capabilities.tools
+
+
+def test_wf_recipes_source_contains_composed_capabilities() -> None:
+    service = WfMcpService(store=FileStore(local_temp_root() / "recipes_source_store"))
+    specs = service.capability_sources["wf.recipes"].capabilities.node_specs
+
+    assert set(specs) == {"wf.recipes.extract_text_content"}
+    assert (
+        service.capability_sources["wf.recipes"].permissions.safe_for_workflow is True
+    )
+
+
+def test_wf_admin_source_exists_but_is_not_planner_visible() -> None:
+    service = WfMcpService(store=FileStore(local_temp_root() / "admin_source_store"))
+    source = service.capability_sources["wf.admin"]
+
+    assert source.kind == "system"
+    assert source.visibility.planner is False
+    assert source.visibility.mcp_client is False
+    assert source.visibility.admin_dashboard is True
+    assert source.permissions.safe_for_workflow is False
+    assert source.permissions.calls_upstream is False
+    assert source.permissions.mutates_config is True
+    assert source.permissions.mutates_auth is True
+    assert "wf.admin.list_sources" in source.capabilities.tools
+    assert "wf.admin.disable_source" in source.capabilities.tools
+    assert "wf.admin.enable_source" in source.capabilities.tools
+    assert "wf.admin" not in service.get_planner_catalog().snapshots
+
+
+def test_service_can_disable_builtin_stdlib_specs() -> None:
+    service = WfMcpService(
+        store=FileStore(local_temp_root() / "no_builtin_store"),
+        include_builtin_specs=False,
+    )
+
+    assert "wf.std" not in service.capability_sources
+    assert "wf.recipes" not in service.capability_sources
+
+
+def test_service_planner_catalog_excludes_hidden_sources() -> None:
+    service = WfMcpService(store=FileStore(local_temp_root() / "hidden_list_store"))
+    hidden_echo_tool = NodeSpec(
+        name="hidden.source.echo_tool",
+        input_model=echo_tool.input_model,
+        output_model=echo_tool.output_model,
+        outcomes=echo_tool.outcomes,
+        fn=echo_tool.fn,
+        description=echo_tool.description,
+        is_async=echo_tool.is_async,
+        accepts_context=echo_tool.accepts_context,
+        input_schema_contract=echo_tool.input_schema_contract,
+        output_schema_contract=echo_tool.output_schema_contract,
+    )
+    service.register_capability_source(
+        CapabilitySource(
+            id="hidden.source",
+            kind="system",
+            capabilities=CapabilityBuckets(
+                node_specs={"hidden.source.echo_tool": hidden_echo_tool}
+            ),
+            visibility=SourceVisibility(planner=False, admin_dashboard=False),
+        )
+    )
+
+    planner_names = {
+        entry.qualified_name for entry in service.get_planner_catalog().entries()
+    }
+    assert "hidden.source.echo_tool" not in planner_names
+
+
+def test_service_catalog_split_keeps_system_specs_out_of_backend_catalog() -> None:
+    service = WfMcpService(store=FileStore(local_temp_root() / "planner_store"))
+
+    backend_payload = service.get_catalog().as_payload()
+    planner_payload = service.get_planner_catalog().as_payload()
+
+    assert backend_payload["nodes"] == []
+    planner_node_names = {node["qualified_name"] for node in planner_payload["nodes"]}
+    assert "wf.std.runtime_error" in planner_node_names
+    available_names = {entry.qualified_name for entry in service.list_available_specs()}
+    assert "wf.std.runtime_error" in available_names
+
+
+def test_service_hydrates_planner_specs_from_stored_catalog() -> None:
+    store = local_temp_root() / "restart_planner_store"
+    shutil.rmtree(store, ignore_errors=True)
+    first_service = WfMcpService(store=FileStore(store))
+    first_service.register_connection(
+        ConnectionConfig(id="demo.personal", server="demo", account="personal")
+    )
+    first_service.register_adapter("demo", FakeAdapter())
+    asyncio.run(first_service.refresh_connection_catalog("demo.personal"))
+
+    second_service = WfMcpService(store=FileStore(store))
+    second_service.register_connection(
+        ConnectionConfig(id="demo.personal", server="demo", account="personal")
+    )
+    second_service.register_adapter("demo", FakeAdapter())
+
+    planner_names = {
+        node["qualified_name"]
+        for node in second_service.get_planner_catalog().as_payload()["nodes"]
+    }
+    run = asyncio.run(
+        second_service.run_workflow_from_plan(
+            single_echo_plan("hydrated_plan", "demo.personal.echo_tool"),
+            {"text": "hello"},
+        )
+    )
+
+    assert "demo.personal.echo_tool" in planner_names
+    assert run.status == RunStatus.COMPLETED
+    assert run.output["echoed"] == "hello"

tests/wf_mcp/service/test_events.py

@@ -0,0 +1,328 @@
+from __future__ import annotations
+
+import asyncio
+from typing import Any, cast
+
+from wf_authoring import build_async_registry
+from wf_core import RuntimeContext
+from wf_mcp.broker import WfMcpService
+from wf_mcp.models import AuthRecord, ConnectionConfig
+from wf_mcp.runtime import ToolExecutor
+from wf_mcp.sdk import ToolCallResult
+from wf_mcp.shared.errors import error_payload
+from wf_mcp.storage import FileStore
+
+from ..test_support import (
+    FailingDiscoveryAdapter,
+    FakeAdapter,
+    local_temp_root,
+)
+
+
+def test_service_records_tool_call_events() -> None:
+    from wf_authoring import node
+    from wf_core import END, RunStatus
+
+    from ..test_support import echo_tool, input_binding, output_binding
+    from .conftest import raw_plan
+
+    service = WfMcpService(store=FileStore(local_temp_root() / "tool_event_store"))
+    service.register_connection(
+        ConnectionConfig(id="demo.personal", server="demo", account="personal")
+    )
+    service.register_adapter("demo", FakeAdapter())
+
+    asyncio.run(service.refresh_connection_catalog("demo.personal"))
+
+    plan = raw_plan(
+        name="tool_event_plan",
+        input_schema={
+            "type": "object",
+            "properties": {"text": {"type": "string"}},
+            "required": ["text"],
+        },
+        state_schema={"fields": {"echoed": {"type": "string"}}},
+        output_schema={
+            "type": "object",
+            "properties": {"echoed": {"type": "string"}},
+            "required": ["echoed"],
+        },
+        start="echo",
+        nodes=[
+            {
+                "id": "echo",
+                "type": "node",
+                "node": "demo.personal.echo_tool",
+                "input": [input_binding("input.text", "text")],
+                "output": [output_binding("echoed", "state.echoed")],
+            }
+        ],
+        edges=[{"from": "echo", "outcome": "ok", "to": END}],
+    )
+
+    run = asyncio.run(service.run_workflow_from_plan(plan, {"text": "hello"}))
+
+    assert run.status == RunStatus.COMPLETED
+    tool_events = [
+        event for event in service.list_events() if "tool_call" in event.kind
+    ]
+    assert [event.kind for event in tool_events] == [
+        "tool_call_started",
+        "tool_call_completed",
+    ]
+    assert tool_events[0].capability_id == "demo.personal.echo_tool"
+    assert tool_events[1].payload["outcome"] == "ok"
+
+
+def test_service_rejects_text_binding_for_raw_mcp_content_contract() -> None:
+    from wf_authoring import node
+    from wf_core import END
+
+    from ..test_support import input_binding, output_binding
+    from .conftest import ContentOnlyOutputAdapter, raw_plan
+
+    service = WfMcpService(store=FileStore(local_temp_root() / "raw_content_contract"))
+    service.register_connection(
+        ConnectionConfig(id="demo.personal", server="demo", account="personal")
+    )
+    service.register_adapter("demo", ContentOnlyOutputAdapter())
+    asyncio.run(service.refresh_connection_catalog("demo.personal"))
+    plan = raw_plan(
+        name="raw_content_contract",
+        input_schema={
+            "type": "object",
+            "properties": {"text": {"type": "string"}},
+            "required": ["text"],
+        },
+        state_schema={"properties": {"outline": {"type": "string"}}},
+        output_schema={
+            "type": "object",
+            "properties": {"outline": {"type": "string"}},
+            "required": ["outline"],
+        },
+        output=[
+            {
+                "target": {"root": "local", "parts": ["outline"]},
+                "path": {"root": "state", "parts": ["outline"]},
+            }
+        ],
+        start="echo",
+        nodes=[
+            {
+                "id": "echo",
+                "type": "node",
+                "node": "demo.personal.echo_tool",
+                "input": [input_binding("input.text", "message")],
+                "output": [output_binding("text", "state.outline")],
+            }
+        ],
+        edges=[{"from": "echo", "outcome": "ok", "to": END}],
+    )
+
+    workflow = service.compile_plan(plan)
+    report = workflow.validate_structure()
+
+    assert not report.ok
+    assert any(
+        "source field 'text' is not declared in node output schema" in issue.message
+        for issue in report.errors
+    )
+
+
+def test_service_can_inspect_resources_and_prompts() -> None:
+    service = WfMcpService(store=FileStore(local_temp_root() / "inspect_store"))
+    service.register_connection(
+        ConnectionConfig(id="demo.personal", server="demo", account="personal")
+    )
+    service.register_adapter("demo", FakeAdapter())
+
+    asyncio.run(service.refresh_connection_catalog("demo.personal"))
+
+    resources = service.list_resources(connection_id="demo.personal")
+    prompts = service.list_prompts(connection_id="demo.personal")
+
+    assert [resource.qualified_name for resource in resources] == [
+        "demo.personal.resource.welcome"
+    ]
+    assert [prompt.qualified_name for prompt in prompts] == [
+        "demo.personal.prompt.summarize"
+    ]
+
+    resource = service.get_resource("demo.personal.resource.welcome")
+    prompt = service.get_prompt("demo.personal.prompt.summarize")
+
+    assert resource.uri == "demo://docs/welcome"
+    assert prompt.arguments[0]["name"] == "text"
+
+
+def test_service_reports_connection_statuses() -> None:
+    import shutil
+
+    store = local_temp_root() / "status_store"
+    shutil.rmtree(store, ignore_errors=True)
+    service = WfMcpService(store=FileStore(store))
+    service.register_connection(
+        ConnectionConfig(id="demo.personal", server="demo", account="personal")
+    )
+    service.register_adapter("demo", FakeAdapter())
+
+    before = service.connection_statuses()
+    assert before == [
+        {
+            "connection_id": "demo.personal",
+            "server": "demo",
+            "account": "personal",
+            "enabled": True,
+            "has_snapshot": False,
+            "fetched_at_epoch_ms": None,
+            "max_age_seconds": None,
+            "node_count": 0,
+            "resource_count": 0,
+            "prompt_count": 0,
+        }
+    ]
+
+    asyncio.run(service.refresh_connection_catalog("demo.personal"))
+    after = service.connection_statuses()
+    assert after[0]["has_snapshot"] is True
+    assert after[0]["node_count"] == 1
+    assert after[0]["resource_count"] == 1
+    assert after[0]["prompt_count"] == 1
+
+
+def test_service_can_proxy_resource_reads_and_prompt_gets() -> None:
+    import shutil
+
+    store = local_temp_root() / "proxy_store"
+    shutil.rmtree(store, ignore_errors=True)
+    service = WfMcpService(store=FileStore(store))
+    service.register_connection(
+        ConnectionConfig(id="demo.personal", server="demo", account="personal")
+    )
+    service.register_adapter("demo", FakeAdapter())
+
+    asyncio.run(service.refresh_connection_catalog("demo.personal"))
+
+    resource_result = asyncio.run(
+        service.read_resource("demo.personal.resource.welcome")
+    )
+    prompt_result = asyncio.run(
+        service.render_prompt(
+            "demo.personal.prompt.summarize",
+            arguments={"text": "hello world"},
+        )
+    )
+
+    assert (
+        resource_result["contents"][0]["text"]
+        == "Welcome from the fake adapter resource."
+    )
+    assert (
+        prompt_result["messages"][0]["content"]["text"]
+        == "Summarize this text:\n\nhello world"
+    )
+
+    event_kinds = [event.kind for event in service.list_events()]
+    assert "resource_read_started" in event_kinds
+    assert "resource_read_completed" in event_kinds
+    assert "prompt_get_started" in event_kinds
+    assert "prompt_get_completed" in event_kinds
+
+
+def test_service_can_invoke_raw_method_and_notification() -> None:
+    service = WfMcpService(store=FileStore(local_temp_root() / "raw_store"))
+    service.register_connection(
+        ConnectionConfig(id="demo.personal", server="demo", account="personal")
+    )
+    service.register_adapter("demo", FakeAdapter())
+
+    result = asyncio.run(
+        service.invoke_method("demo.personal", "demo.echo", params={"text": "hello"})
+    )
+    asyncio.run(
+        service.send_notification(
+            "demo.personal",
+            "notifications/progress",
+            params={"progress": 1},
+        )
+    )
+
+    assert result == {"echoed": "hello"}
+    event_kinds = [event.kind for event in service.list_events()]
+    assert "raw_method_started" in event_kinds
+    assert "raw_method_completed" in event_kinds
+    assert "raw_notification_started" in event_kinds
+    assert "raw_notification_completed" in event_kinds
+
+
+def test_generated_specs_use_injected_tool_executor() -> None:
+    class RecordingExecutor:
+        def __init__(self) -> None:
+            self.payloads: list[dict[str, Any]] = []
+
+        async def call_tool(
+            self,
+            connection: ConnectionConfig,
+            auth: AuthRecord | None,
+            tool_name: str,
+            payload: dict[str, Any],
+        ) -> ToolCallResult:
+            self.payloads.append(payload)
+            return ToolCallResult(outcome="ok", output={"echoed": payload["text"]})
+
+    executor = RecordingExecutor()
+    service = WfMcpService(
+        store=FileStore(local_temp_root() / "injected_executor_store"),
+        tool_executor=cast(ToolExecutor, executor),
+    )
+    service.register_connection(
+        ConnectionConfig(id="demo.personal", server="demo", account="personal")
+    )
+    service.register_adapter("demo", FakeAdapter())
+
+    asyncio.run(service.refresh_connection_catalog("demo.personal"))
+    spec = service._get_qualified_spec("demo.personal.echo_tool")
+    handler = build_async_registry(spec)[spec.name]
+
+    async def run_node() -> dict[str, Any]:
+        return await handler({"text": "hello"}, RuntimeContext(current_node_id="echo"))
+
+    result = asyncio.run(run_node())
+
+    assert result["outcome"] == "ok"
+    assert result["output"]["echoed"] == "hello"
+    assert executor.payloads == [{"text": "hello"}]
+
+
+def test_service_records_catalog_refresh_failures() -> None:
+    service = WfMcpService(store=FileStore(local_temp_root() / "refresh_fail_store"))
+    service.register_connection(
+        ConnectionConfig(id="demo.personal", server="demo", account="personal")
+    )
+    service.register_adapter("demo", FailingDiscoveryAdapter())
+
+    try:
+        asyncio.run(service.refresh_connection_catalog("demo.personal"))
+    except PermissionError as exc:
+        assert str(exc) == "Access is denied"
+    else:
+        raise AssertionError("expected refresh to fail")
+
+    failure_events = [
+        event
+        for event in service.list_events()
+        if event.kind == "catalog_refresh_failed"
+    ]
+    assert len(failure_events) == 1
+    assert failure_events[0].payload == {
+        "error_type": "PermissionError",
+        "error": "Access is denied",
+    }
+
+
+def test_error_payload_unwraps_exception_group() -> None:
+    exc = ExceptionGroup("outer", [PermissionError("Access is denied")])
+    assert error_payload(exc) == {
+        "error_type": "PermissionError",
+        "error": "Access is denied",
+    }

tests/wf_mcp/service/test_sources.py

@@ -0,0 +1,287 @@
+from __future__ import annotations
+
+import asyncio
+
+from wf_authoring import NodeSpec, node
+from wf_mcp.broker import WfMcpService
+from wf_mcp.models import ConnectionConfig
+from wf_mcp.storage import FileStore
+from wf_platform import (
+    CapabilityBuckets,
+    CapabilitySource,
+    SourceVisibility,
+)
+
+from ..test_support import (
+    FakeAdapter,
+    echo_tool,
+    finalize_tool,
+    local_temp_root,
+)
+from .conftest import raw_plan, single_echo_plan
+
+
+def test_service_compiles_and_runs_raw_plan() -> None:
+    service = WfMcpService(store=FileStore(local_temp_root() / "run_store"))
+    service.register_connection(
+        ConnectionConfig(id="demo.personal", server="demo", account="personal")
+    )
+    service.register_specs("demo.personal", echo_tool)
+
+    plan = raw_plan(
+        name="demo_plan",
+        input_schema={
+            "type": "object",
+            "properties": {"text": {"type": "string"}},
+            "required": ["text"],
+        },
+        state_schema={"fields": {"echoed": {"type": "string"}}},
+        output_schema={
+            "type": "object",
+            "properties": {"echoed": {"type": "string"}},
+            "required": ["echoed"],
+        },
+        start="echo",
+        nodes=[
+            {
+                "id": "echo",
+                "type": "node",
+                "node": "demo.personal.echo_tool",
+                "input": [{"target": {"root": "local", "parts": ["text"]}, "path": {"root": "input", "parts": ["text"]}}],
+                "output": [{"source": {"root": "local", "parts": ["echoed"]}, "target": {"root": "state", "parts": ["echoed"]}}],
+            }
+        ],
+        edges=[{"from": "echo", "outcome": "ok", "to": "__end__"}],
+    )
+
+    run = asyncio.run(service.run_workflow_from_plan(plan, {"text": "hello"}))
+
+    assert run.status == "completed"
+    assert run.output["echoed"] == "hello"
+
+
+def test_service_preserves_raw_plan_root_output_bindings() -> None:
+    service = WfMcpService(store=FileStore(local_temp_root() / "root_output_store"))
+    service.register_connection(
+        ConnectionConfig(id="demo.personal", server="demo", account="personal")
+    )
+    service.register_specs("demo.personal", echo_tool)
+
+    plan = raw_plan(
+        name="root_output_plan",
+        input_schema={"type": "object", "properties": {"text": {"type": "string"}}, "required": ["text"]},
+        state_schema={"fields": {"echoed": {"type": "string"}}},
+        output_schema={"type": "object", "properties": {"echoed": {"type": "string"}}, "required": ["echoed"]},
+        start="echo",
+        nodes=[
+            {
+                "id": "echo",
+                "type": "node",
+                "node": "demo.personal.echo_tool",
+                "input": [{"target": {"root": "local", "parts": ["text"]}, "path": {"root": "input", "parts": ["text"]}}],
+                "output": [{"source": {"root": "local", "parts": ["echoed"]}, "target": {"root": "state", "parts": ["echoed"]}}],
+            }
+        ],
+        edges=[{"from": "echo", "outcome": "ok", "to": "__end__"}],
+    )
+
+    run = asyncio.run(service.run_workflow_from_plan(plan, {"text": "hello"}))
+
+    assert run.output["echoed"] == "hello"
+
+
+def test_service_resolves_registered_spec_with_dotted_local_name() -> None:
+    service = WfMcpService(store=FileStore(local_temp_root() / "dotted_store"))
+    service.register_connection(
+        ConnectionConfig(id="demo.personal", server="demo", account="personal")
+    )
+    service.register_specs("demo.personal", echo_tool)
+
+    plan = single_echo_plan("dotted_plan", "demo.personal.echo_tool")
+
+    run = asyncio.run(service.run_workflow_from_plan(plan, {"text": "hello"}))
+
+    assert run.status == "completed"
+    assert run.output["echoed"] == "hello"
+
+
+def test_service_runs_logical_source_plan_with_dotted_local_name() -> None:
+    import shutil
+
+    from wf_artifacts import WorkflowDeployment
+    from wf_authoring import NodeSpec
+
+    from ..test_support import echo_tool
+
+    store = local_temp_root() / "logical_source_store"
+    shutil.rmtree(store, ignore_errors=True)
+    service = WfMcpService(store=FileStore(store))
+    service.register_connection(
+        ConnectionConfig(id="demo.personal", server="demo", account="personal")
+    )
+    dotted_echo_tool = NodeSpec(
+        name="foo.bar",
+        input_model=echo_tool.input_model,
+        output_model=echo_tool.output_model,
+        outcomes=echo_tool.outcomes,
+        fn=echo_tool.fn,
+        description=echo_tool.description,
+        is_async=echo_tool.is_async,
+        accepts_context=echo_tool.accepts_context,
+        input_schema_contract=echo_tool.input_schema_contract,
+        output_schema_contract=echo_tool.output_schema_contract,
+    )
+    service.register_specs("demo.personal", dotted_echo_tool)
+
+    plan = single_echo_plan("logical_plan", "demo.foo.bar")
+
+    run = asyncio.run(
+        service.run_workflow_from_plan(
+            plan,
+            {"text": "hello"},
+            deployment=WorkflowDeployment(
+                id="logical_dotted.personal",
+                artifact_id="logical_dotted",
+                artifact_version=1,
+                bindings=[
+                    {"logical_source": "demo", "concrete_source": "demo.personal"}
+                ],
+            ),
+        )
+    )
+
+    assert run.status == "completed"
+    assert run.output["echoed"] == "hello"
+
+
+def test_service_binds_longest_logical_source_prefix_first() -> None:
+    service = WfMcpService(store=FileStore(local_temp_root() / "prefix_store"))
+    service.register_connection(
+        ConnectionConfig(id="demo.personal", server="demo", account="personal")
+    )
+    service.register_connection(
+        ConnectionConfig(id="demo.work", server="demo", account="work")
+    )
+    service.register_specs("demo.personal", echo_tool)
+
+    plan = single_echo_plan("prefix_plan", "demo.personal.echo_tool")
+
+    run = asyncio.run(service.run_workflow_from_plan(plan, {"text": "hello"}))
+
+    assert run.status == "completed"
+
+
+def test_service_does_not_resolve_specs_hidden_from_planner() -> None:
+    service = WfMcpService(store=FileStore(local_temp_root() / "hidden_resolve_store"))
+    hidden_echo_tool = NodeSpec(
+        name="hidden.source.echo_tool",
+        input_model=echo_tool.input_model,
+        output_model=echo_tool.output_model,
+        outcomes=echo_tool.outcomes,
+        fn=echo_tool.fn,
+        description=echo_tool.description,
+        is_async=echo_tool.is_async,
+        accepts_context=echo_tool.accepts_context,
+        input_schema_contract=echo_tool.input_schema_contract,
+        output_schema_contract=echo_tool.output_schema_contract,
+    )
+    service.register_capability_source(
+        CapabilitySource(
+            id="hidden.source",
+            kind="system",
+            capabilities=CapabilityBuckets(
+                node_specs={"hidden.source.echo_tool": hidden_echo_tool}
+            ),
+            visibility=SourceVisibility(planner=False, admin_dashboard=False),
+        )
+    )
+
+    try:
+        service._get_qualified_spec("hidden.source.echo_tool")
+    except KeyError:
+        pass
+    else:
+        raise AssertionError("expected hidden spec to be unresolvable")
+
+
+def test_service_excludes_disabled_connection_specs_from_planner_catalog() -> None:
+    service = WfMcpService(store=FileStore(local_temp_root() / "disabled_store"))
+    service.register_connection(
+        ConnectionConfig(id="demo.personal", server="demo", account="personal")
+    )
+    service.register_specs("demo.personal", echo_tool)
+    service.capability_sources["demo.personal"].enabled = False
+
+    planner_names = {
+        entry.qualified_name for entry in service.get_planner_catalog().entries()
+    }
+    assert "demo.personal.echo_tool" not in planner_names
+
+
+def test_service_preserves_disabled_connection_source_on_reregistration() -> None:
+    service = WfMcpService(store=FileStore(local_temp_root() / "disabled_reregister"))
+    service.register_connection(
+        ConnectionConfig(id="demo.personal", server="demo", account="personal")
+    )
+    service.register_specs("demo.personal", echo_tool)
+    service.capability_sources["demo.personal"].enabled = False
+    service.register_specs("demo.personal", finalize_tool)
+
+    source = service.capability_sources["demo.personal"]
+    assert source.enabled is False
+    assert "demo.personal.finalize_tool" in source.capabilities.node_specs
+    assert "demo.personal.echo_tool" not in source.capabilities.node_specs
+
+
+def test_service_excludes_planner_hidden_connection_specs_from_planner_catalog() -> (
+    None
+):
+    service = WfMcpService(store=FileStore(local_temp_root() / "planner_hidden_store"))
+    service.register_connection(
+        ConnectionConfig(id="demo.personal", server="demo", account="personal")
+    )
+    service.register_specs("demo.personal", echo_tool)
+    service.capability_sources["demo.personal"].visibility = SourceVisibility(
+        planner=False,
+        mcp_client=True,
+        admin_dashboard=True,
+    )
+
+    planner_names = {
+        entry.qualified_name for entry in service.get_planner_catalog().entries()
+    }
+    assert "demo.personal.echo_tool" not in planner_names
+
+
+def test_service_preserves_planner_hidden_connection_source_on_reregistration() -> None:
+    service = WfMcpService(store=FileStore(local_temp_root() / "planner_hidden_rereg"))
+    service.register_connection(
+        ConnectionConfig(id="demo.personal", server="demo", account="personal")
+    )
+    service.register_specs("demo.personal", echo_tool)
+    service.capability_sources["demo.personal"].visibility = SourceVisibility(
+        planner=False,
+        mcp_client=True,
+        admin_dashboard=True,
+    )
+    service.register_specs("demo.personal", finalize_tool)
+
+    source = service.capability_sources["demo.personal"]
+    assert source.visibility.planner is False
+    assert source.visibility.mcp_client is True
+    assert source.visibility.admin_dashboard is True
+    assert "demo.personal.finalize_tool" in source.capabilities.node_specs
+    assert "demo.personal.echo_tool" not in source.capabilities.node_specs
+
+
+def test_service_refreshes_catalog_from_adapter() -> None:
+    service = WfMcpService(store=FileStore(local_temp_root() / "refresh_store"))
+    service.register_connection(
+        ConnectionConfig(id="demo.personal", server="demo", account="personal")
+    )
+    service.register_adapter("demo", FakeAdapter())
+
+    asyncio.run(service.refresh_connection_catalog("demo.personal"))
+
+    source = service.capability_sources["demo.personal"]
+    assert "demo.personal.echo_tool" in source.capabilities.node_specs