fix: avoid opening mcp session for local metadata

docs/superpowers/plans/2026-06-08-mcp-runtime-full-operation-surface.md

@@ -0,0 +1,452 @@
+# MCP Runtime Full Operation Surface 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:** Make the persistent MCP runtime pool implement the same MCP operation surface as the one-shot backend adapter, then route broker operations through the persistent runtime whenever it is configured.
+
+**Architecture:** `BackendAdapter` and `StatefulMcpRuntime` both represent "execute MCP operations for a configured source"; they should differ by session strategy, not by operation set. `McpSdkAdapter` remains the one-shot implementation. `McpRuntimePool` becomes the persistent implementation and should be preferred by `UpstreamTransportService` for tool calls, resource/prompt operations, catalog listings, raw methods, notifications, and liveness probes when available.
+
+**Tech Stack:** Python 3.14, MCP SDK `ClientSession`, `McpSourceClient`, persistent owner-task queue, pytest-asyncio, ruff, basedpyright.
+
+---
+
+## Hard Boundaries
+
+- Do not remove `McpSdkAdapter`; it remains the one-shot fallback.
+- Do not remove `BackendAdapter` or `StatefulMcpRuntime` names in this slice.
+- Do not change public JSON-RPC or CLI payload shapes.
+- Do not change source registry or config file formats.
+- Do not add e2e JSON-RPC session-reuse tests in this slice; that is the next QA slice.
+- Do not expose raw `ClientSession` outside `wf_sources_mcp.runtime` internals.
+- Do not dispatch owner-task operations via `getattr(client, operation)`; continue passing explicit callables.
+- Do not commit unless the caller explicitly asks for a commit.
+
+## File Map
+
+- Modify `src/wf_sources_mcp/sdk/protocols.py`: expand `StatefulMcpRuntime` to full operation surface or introduce a shared full-surface protocol.
+- Modify `src/wf_sources_mcp/runtime/session.py`: add persistent session methods for `list_tools`, `get_connection_metadata`, `invoke_method`, and `send_notification`.
+- Modify `src/wf_sources_mcp/runtime/factory.py`: route those new methods through `_SessionOwner.submit()`.
+- Modify `src/wf_sources_mcp/runtime/pool.py`: expose full operation methods.
+- Modify `src/wf_sources_mcp/client/source_client.py`: ensure all operations needed by runtime owner already exist and stay reusable.
+- Modify `src/wf_mcp/broker/service/upstream_transport.py`: prefer `stateful_runtime` for all MCP operations that can use an existing session.
+- Modify tests:
+  - `tests/wf_sources_mcp/test_sdk_protocols.py`
+  - `tests/wf_sources_mcp/test_runtime.py`
+  - `tests/wf_mcp/service/test_upstream_transport.py`
+- Modify docs:
+  - `docs/current_roadmap.md`
+  - `docs/superpowers/specs/2026-06-03-long-lived-workflow-api-boundary.md`
+- Move this plan to `docs/historical/superpowers/plans/` after implementation is verified.
+
+---
+
+### Task 1: Expand the Runtime Protocol Surface
+
+**Files:**
+- Modify: `src/wf_sources_mcp/sdk/protocols.py`
+- Modify: `tests/wf_sources_mcp/test_sdk_protocols.py`
+
+- [ ] **Step 1: Add protocol tests first**
+
+In `tests/wf_sources_mcp/test_sdk_protocols.py`, add or extend a static-shape test that assigns an object implementing all operations to both `BackendAdapter` and `StatefulMcpRuntime`.
+
+The object should include:
+
+- `get_connection_metadata`
+- `list_tools`
+- `list_resources`
+- `list_prompts`
+- `call_tool`
+- `read_resource`
+- `get_prompt`
+- `invoke_method`
+- `send_notification`
+
+Use existing test fakes where possible.
+
+- [ ] **Step 2: Update protocol definitions**
+
+In `src/wf_sources_mcp/sdk/protocols.py`, prefer this shape:
+
+```python
+class McpSourceOperations(Protocol):
+    async def list_tools(...): ...
+    async def list_resources(...): ...
+    async def list_prompts(...): ...
+    async def get_connection_metadata(...): ...
+    async def read_resource(...): ...
+    async def get_prompt(...): ...
+    async def invoke_method(...): ...
+    async def send_notification(...): ...
+    async def call_tool(...): ...
+
+
+class BackendAdapter(McpSourceOperations, Protocol):
+    """One-shot or adapter-style MCP operation executor."""
+
+
+class StatefulMcpRuntime(McpSourceOperations, Protocol):
+    """Persistent MCP operation executor for configured sources."""
+```
+
+Keep existing narrow aliases (`ToolRuntime`, `ResourceRuntime`, `PromptRuntime`, `ToolExecutor`) for compatibility.
+
+Add `McpSourceOperations` to `__all__` and package exports in `src/wf_sources_mcp/sdk/__init__.py`.
+
+- [ ] **Step 3: Run protocol tests**
+
+Run:
+
+```bash
+uv run pytest tests/wf_sources_mcp/test_sdk_protocols.py -q
+```
+
+Expected: pass.
+
+---
+
+### Task 2: Add Full Operations to `PersistentMcpSession`
+
+**Files:**
+- Modify: `src/wf_sources_mcp/runtime/session.py`
+- Modify: `tests/wf_sources_mcp/test_runtime.py`
+
+- [ ] **Step 1: Add tests for session-level operations**
+
+Extend fake client/session tests in `tests/wf_sources_mcp/test_runtime.py` to cover:
+
+- `list_tools()`
+- `get_connection_metadata()`
+- `invoke_method()`
+- `send_notification()`
+
+For injected/fake `client` path, verify:
+
+- list tools are converted to `DiscoveredTool`;
+- metadata is local `{server, transport}`;
+- invoke returns dumped payload;
+- notification records that it was sent.
+
+- [ ] **Step 2: Add callback types and methods**
+
+In `src/wf_sources_mcp/runtime/session.py`, add callback types:
+
+```python
+RawToolLister = Callable[[], Awaitable[list[DiscoveredTool]]]
+RawMetadataGetter = Callable[[], Awaitable[dict[str, Any]]]
+RawMethodInvoker = Callable[[str, dict[str, Any] | None], Awaitable[dict[str, Any]]]
+RawNotificationSender = Callable[[str, dict[str, Any] | None], Awaitable[None]]
+```
+
+Add corresponding dataclass fields.
+
+Implement methods:
+
+- `list_tools()`
+- `get_connection_metadata()`
+- `invoke_method(method, params=None)`
+- `send_notification(method, params=None)`
+
+Use callback first, then injected `client` fallback where applicable. Metadata can be computed locally from `self.connection` without client.
+
+- [ ] **Step 3: Run runtime tests**
+
+Run:
+
+```bash
+uv run pytest tests/wf_sources_mcp/test_runtime.py -q
+```
+
+Expected: pass.
+
+---
+
+### Task 3: Route Full Operations Through the Owner Task
+
+**Files:**
+- Modify: `src/wf_sources_mcp/runtime/factory.py`
+- Modify: `tests/wf_sources_mcp/test_runtime.py`
+
+- [ ] **Step 1: Add owner-routing tests**
+
+In `tests/wf_sources_mcp/test_runtime.py`, add or extend a test proving `PersistentSessionFactory` routes these operations through `_SessionOwner.submit()`:
+
+- `list_tools`
+- `invoke_method`
+- `send_notification`
+
+Existing tests already cover resources/prompts/tool calls. Keep all operations explicit; do not test implementation internals like queue item class unless existing patterns do.
+
+- [ ] **Step 2: Wire callbacks in `PersistentSessionFactory.create()`**
+
+In `create()`, pass:
+
+```python
+list_tools_callback=owner.list_tools
+get_connection_metadata_callback=owner.get_connection_metadata
+invoke_method_callback=owner.invoke_method
+send_notification_callback=owner.send_notification
+```
+
+- [ ] **Step 3: Add `_SessionOwner` methods**
+
+In `_SessionOwner`, add:
+
+```python
+async def list_tools(self) -> list[DiscoveredTool]:
+    return await self.submit("list_tools", lambda client: client.list_tools())
+
+async def get_connection_metadata(self) -> dict[str, Any]:
+    return await self.submit(
+        "get_connection_metadata",
+        lambda client: client.get_connection_metadata(),
+    )
+
+async def invoke_method(self, method: str, params: dict[str, Any] | None = None) -> dict[str, Any]:
+    return await self.submit(
+        "invoke_method",
+        lambda client: client.invoke_method(method, params),
+    )
+
+async def send_notification(self, method: str, params: dict[str, Any] | None = None) -> None:
+    await self.submit(
+        "send_notification",
+        lambda client: client.send_notification(method, params),
+    )
+```
+
+- [ ] **Step 4: Run runtime tests**
+
+Run:
+
+```bash
+uv run pytest tests/wf_sources_mcp/test_runtime.py -q
+```
+
+Expected: pass.
+
+---
+
+### Task 4: Add Full Operations to `McpRuntimePool`
+
+**Files:**
+- Modify: `src/wf_sources_mcp/runtime/pool.py`
+- Modify: `tests/wf_sources_mcp/test_runtime.py`
+
+- [ ] **Step 1: Add pool-level tests**
+
+Add tests proving the pool reuses one session for:
+
+- `list_tools`
+- `invoke_method`
+- `send_notification`
+
+Also extend the public protocol/static-shape test so `McpRuntimePool` satisfies `StatefulMcpRuntime`.
+
+- [ ] **Step 2: Implement pool methods**
+
+Add methods to `McpRuntimePool`:
+
+```python
+async def list_tools(...)
+async def get_connection_metadata(...)
+async def invoke_method(...)
+async def send_notification(...)
+```
+
+Each method should:
+
+1. call `get_session(connection, auth)`;
+2. delegate to the matching `PersistentMcpSession` method.
+
+- [ ] **Step 3: Run runtime tests**
+
+Run:
+
+```bash
+uv run pytest tests/wf_sources_mcp/test_runtime.py tests/wf_sources_mcp/test_sdk_protocols.py -q
+```
+
+Expected: pass.
+
+---
+
+### Task 5: Route Broker Upstream Operations Through Stateful Runtime
+
+**Files:**
+- Modify: `src/wf_mcp/broker/service/upstream_transport.py`
+- Modify: `tests/wf_mcp/service/test_upstream_transport.py`
+
+- [ ] **Step 1: Add routing tests**
+
+Extend `_StatefulRuntime` fake in `tests/wf_mcp/service/test_upstream_transport.py` to implement the full operation surface.
+
+Add tests proving that when `stateful_runtime` is configured:
+
+- `invoke_method()` uses runtime, not adapter;
+- `send_notification()` uses runtime, not adapter;
+- `refresh_connection_catalog()` uses runtime for discovery/listing operations;
+- `deployment_diagnostics()` uses runtime for live `list_tools()` checks.
+
+Use an adapter fake that raises if called for the operation under test.
+
+- [ ] **Step 2: Add a local operation selector helper**
+
+In `UpstreamTransportService`, add a private helper:
+
+```python
+def _operations_for(self, connection: ConnectionConfig) -> StatefulMcpRuntime | BackendAdapter:
+    if self.stateful_runtime is not None:
+        return self.stateful_runtime
+    return require_adapter(connection, self.adapters)
+```
+
+If basedpyright cannot narrow the union cleanly, keep explicit branches in each method instead of introducing a helper. Correctness is more important than DRY.
+
+- [ ] **Step 3: Update `invoke_method()` and `send_notification()`**
+
+Prefer `stateful_runtime` when present. Fall back to adapter when absent. Preserve event names and payloads.
+
+- [ ] **Step 4: Update `refresh_connection_catalog()`**
+
+For discovery/listing, prefer `stateful_runtime` when present:
+
+```python
+operations = self.stateful_runtime or require_adapter(connection, self.adapters)
+capabilities = await discover_connection_capabilities(
+    connection=source_connection,
+    auth=auth,
+    adapter=operations,
+)
+```
+
+This requires `StatefulMcpRuntime` to satisfy the same protocol as `BackendAdapter`.
+
+- [ ] **Step 5: Update `deployment_diagnostics()`**
+
+Use `stateful_runtime.list_tools()` when present; otherwise adapter fallback.
+
+- [ ] **Step 6: Run upstream transport tests**
+
+Run:
+
+```bash
+uv run pytest tests/wf_mcp/service/test_upstream_transport.py tests/wf_mcp/service/test_content_access.py tests/wf_mcp/service/test_catalog.py -q
+```
+
+Expected: pass.
+
+---
+
+### Task 6: Update Docs and Archive Plan
+
+**Files:**
+- Modify: `docs/current_roadmap.md`
+- Modify: `docs/superpowers/specs/2026-06-03-long-lived-workflow-api-boundary.md`
+- Move: `docs/superpowers/plans/2026-06-08-mcp-runtime-full-operation-surface.md` to `docs/historical/superpowers/plans/2026-06-08-mcp-runtime-full-operation-surface.md`
+
+- [ ] **Step 1: Update roadmap**
+
+Under the MCP runtime/source cleanup section, add:
+
+```markdown
+      Persistent MCP runtime now implements the full upstream MCP operation
+      surface used by the one-shot adapter. Broker upstream operations prefer
+      the shared runtime pool when configured, with one-shot adapters retained
+      as fallback.
+```
+
+- [ ] **Step 2: Update long-lived boundary spec**
+
+Add a completed numbered item:
+
+```markdown
+24. Complete: `McpRuntimePool` implements the full MCP operation surface
+    (`list_tools`, resources, prompts, tools, raw methods, notifications, and
+    local metadata). Broker upstream operations prefer the persistent runtime
+    when configured and fall back to one-shot adapters.
+```
+
+Renumber pending items if needed.
+
+- [ ] **Step 3: Archive the plan**
+
+Run:
+
+```bash
+git mv docs/superpowers/plans/2026-06-08-mcp-runtime-full-operation-surface.md docs/historical/superpowers/plans/2026-06-08-mcp-runtime-full-operation-surface.md
+```
+
+Expected: plan moved under `docs/historical/...`.
+
+---
+
+### Task 7: Final Verification
+
+**Files:**
+- No code edits unless verification finds a real issue.
+
+- [ ] **Step 1: Run focused tests**
+
+Run:
+
+```bash
+uv run pytest tests/wf_sources_mcp/test_runtime.py tests/wf_sources_mcp/test_sdk_protocols.py tests/wf_sources_mcp/test_sdk_adapter.py tests/wf_mcp/service/test_upstream_transport.py tests/wf_mcp/service/test_content_access.py tests/wf_mcp/service/test_catalog.py -q
+```
+
+Expected: pass.
+
+- [ ] **Step 2: Run env-gated everything-server test if `.env` has settings**
+
+If `.env` contains `MCP_EVERYTHING_COMMAND` or `MCP_EVERYTHING_URL`, run:
+
+```bash
+uv run --env-file .env pytest tests/wf_mcp/test_sdk_adapter.py::test_mcp_sdk_adapter_can_probe_everything_server -q
+```
+
+Expected: pass or skip only for environment permission issues. Do not hide real failures.
+
+- [ ] **Step 3: Run lint**
+
+Run:
+
+```bash
+uv run ruff check src/wf_sources_mcp/runtime src/wf_sources_mcp/sdk src/wf_mcp/broker/service/upstream_transport.py tests/wf_sources_mcp/test_runtime.py tests/wf_sources_mcp/test_sdk_protocols.py tests/wf_mcp/service/test_upstream_transport.py
+```
+
+Expected: `All checks passed!`
+
+- [ ] **Step 4: Run typecheck**
+
+Run:
+
+```bash
+uv run basedpyright --level error src/wf_sources_mcp/runtime src/wf_sources_mcp/sdk src/wf_mcp/broker/service/upstream_transport.py tests/wf_sources_mcp/test_runtime.py tests/wf_sources_mcp/test_sdk_protocols.py tests/wf_mcp/service/test_upstream_transport.py
+```
+
+Expected: `0 errors, 0 warnings, 0 notes`.
+
+- [ ] **Step 5: Check whitespace**
+
+Run:
+
+```bash
+git diff --check
+```
+
+Expected: no whitespace errors. CRLF warnings on Windows are acceptable.
+
+---
+
+## Expected Final Report
+
+The implementer should report:
+
+- Files created, modified, and moved.
+- Exact verification commands and pass/fail output.
+- Confirmation that `McpRuntimePool` satisfies the full MCP operation protocol.
+- Confirmation that `UpstreamTransportService` prefers stateful runtime for all MCP operations when configured.
+- Confirmation that `McpSdkAdapter` remains the one-shot fallback.
+- Any deviations from this plan.
+
+Do not claim "full suite passed" unless the full suite was actually run.

src/wf_sources_mcp/sdk/adapter.py

@@ -58,8 +58,13 @@ class McpSdkAdapter(BackendAdapter):
         connection: McpSourceConnection,
         auth: AuthRecord | None,
     ) -> dict[str, Any]:
-        async with self._client(connection, auth) as client:
+        # Metadata is derived from local connection config. Opening an MCP
-            return await client.get_connection_metadata()
+        # transport here would create an unnecessary session during discovery.
+        transport = connection.transport
+        return {
+            "server": connection.provider,
+            "transport": transport.kind if transport is not None else None,
+        }
 
     async def read_resource(
         self,

tests/wf_sources_mcp/test_sdk_adapter.py

@@ -164,6 +164,11 @@ class _FakeAdapter(McpSdkAdapter):
         )
 
 
+class _ExplodingClientAdapter(McpSdkAdapter):
+    def _client(self, connection: McpSourceConnection, auth: object | None):
+        raise AssertionError("metadata lookup must not open an MCP session")
+
+
 def test_mcp_sdk_adapter_implements_backend_protocol() -> None:
     adapter: BackendAdapter = McpSdkAdapter()
 
@@ -219,3 +224,13 @@ async def test_mcp_sdk_adapter_uses_session_for_all_backend_methods() -> None:
         "tool": "echo",
         "payload": {"text": "hello"},
     }
+
+
+@pytest.mark.asyncio
+async def test_mcp_sdk_adapter_metadata_does_not_open_session() -> None:
+    metadata = await _ExplodingClientAdapter().get_connection_metadata(
+        _connection(),
+        None,
+    )
+
+    assert metadata == {"server": "demo", "transport": "stdio"}