docs: plan stateful runtime listing

2026-06-07 20:51:12 +07:00
parent 8bb2f32d9a
commit 9fea19c550
1 changed files with 638 additions and 0 deletions
@@ -0,0 +1,638 @@
 # Stateful Runtime List Resources Prompts 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:** Complete the stateful MCP runtime's safe client surface by adding persistent `list_resources()` and `list_prompts()`, and split runtime protocols into tool/resource/prompt slices.
 **Architecture:** `McpSourceClient` already wraps initialized MCP client operations for tools, resources, and prompts. `McpRuntimePool` should be able to route the same safe resource/prompt listing operations through its owner-task queue so session-scoped listings can observe runtime state. Protocols should describe the slices explicitly: tool execution, resource operations, prompt operations, and their combined stateful runtime.
 **Tech Stack:** Python 3.14, MCP Python SDK, pytest/pytest-asyncio, ruff, basedpyright.
 ---
 ## File Structure
 - Modify `src/wf_sources_mcp/sdk/protocols.py`
  - Add `ToolRuntime`, `ResourceRuntime`, `PromptRuntime`.
  - Make `StatefulMcpRuntime(ToolRuntime, ResourceRuntime, PromptRuntime, Protocol)`.
  - Keep `ToolExecutor` as a compatibility alias/subprotocol for workflow node execution.
 - Modify `src/wf_sources_mcp/sdk/__init__.py`
  - Export new protocol slices.
 - Modify compatibility shims:
  - `src/wf_mcp/sdk/base.py`
  - `src/wf_mcp/sdk/__init__.py`
 - Modify runtime:
  - `src/wf_sources_mcp/runtime/session.py`
  - `src/wf_sources_mcp/runtime/factory.py`
  - `src/wf_sources_mcp/runtime/pool.py`
 - Modify tests:
  - `tests/wf_sources_mcp/test_sdk_protocols.py`
  - `tests/wf_sources_mcp/test_runtime.py`
  - `tests/wf_mcp/test_compat_imports.py`
 - Update docs:
  - `docs/current_roadmap.md`
  - `docs/superpowers/specs/2026-06-03-long-lived-workflow-api-boundary.md`
 ## Hard Boundaries
 - Do not add raw `invoke_method` or `send_notification`.
 - Do not add runtime `list_tools` in this slice.
 - Do not use runtime `isinstance` checks or `@runtime_checkable` protocols.
 - Do not route catalog refresh/discovery through stateful runtime in this slice.
 - Do not assume resource/prompt listings are stateless; these new methods exist because they can be session-scoped.
 ---
 ### Task 1: Split Protocol Slices
 **Files:**
 - Modify: `src/wf_sources_mcp/sdk/protocols.py`
 - Modify: `src/wf_sources_mcp/sdk/__init__.py`
 - Modify: `tests/wf_sources_mcp/test_sdk_protocols.py`
 - [ ] **Step 1: Add protocol shape tests**
 Append to `tests/wf_sources_mcp/test_sdk_protocols.py`:
 ```python
 def test_stateful_runtime_protocol_slices_export() -> None:
    from wf_sources_mcp.sdk import (
        PromptRuntime,
        ResourceRuntime,
        StatefulMcpRuntime,
        ToolExecutor,
        ToolRuntime,
    )
    assert ToolRuntime.__name__ == "ToolRuntime"
    assert ResourceRuntime.__name__ == "ResourceRuntime"
    assert PromptRuntime.__name__ == "PromptRuntime"
    assert ToolExecutor.__name__ == "ToolExecutor"
    assert StatefulMcpRuntime.__name__ == "StatefulMcpRuntime"
 ```
 - [ ] **Step 2: Add protocol slices**
 In `src/wf_sources_mcp/sdk/protocols.py`, replace the current `ToolExecutor` / `StatefulMcpRuntime` section with:
 ```python
 class ToolRuntime(Protocol):
    """Runtime boundary for executing MCP tools from workflow nodes."""
    async def call_tool(
        self,
        connection: McpSourceConnection,
        auth: AuthRecord | None,
        tool_name: str,
        payload: dict[str, Any],
    ) -> ToolCallResult: ...
 class ToolExecutor(ToolRuntime, Protocol):
    """Compatibility name for workflow-node tool execution."""
 class ResourceRuntime(Protocol):
    """Stateful resource operations for configured MCP sources."""
    async def list_resources(
        self,
        connection: McpSourceConnection,
        auth: AuthRecord | None,
    ) -> list[DiscoveredResource]: ...
    async def read_resource(
        self,
        connection: McpSourceConnection,
        auth: AuthRecord | None,
        uri: str,
    ) -> dict[str, Any]: ...
 class PromptRuntime(Protocol):
    """Stateful prompt operations for configured MCP sources."""
    async def list_prompts(
        self,
        connection: McpSourceConnection,
        auth: AuthRecord | None,
    ) -> list[DiscoveredPrompt]: ...
    async def get_prompt(
        self,
        connection: McpSourceConnection,
        auth: AuthRecord | None,
        prompt_name: str,
        arguments: dict[str, str] | None = None,
    ) -> dict[str, Any]: ...
 class StatefulMcpRuntime(ToolRuntime, ResourceRuntime, PromptRuntime, Protocol):
    """Stateful execution/read/list boundary for configured MCP sources.
    Implementations keep source session state across calls. Catalog refresh may
    still use one-shot adapters by policy.
    """
 ```
 Update `__all__`:
 ```python
 __all__ = [
    "BackendAdapter",
    "PromptRuntime",
    "ResourceRuntime",
    "StatefulMcpRuntime",
    "ToolCallResult",
    "ToolExecutor",
    "ToolRuntime",
 ]
 ```
 - [ ] **Step 3: Export slices from package root**
 In `src/wf_sources_mcp/sdk/__init__.py`, import/export the new symbols:
 ```python
 from .protocols import (
    BackendAdapter,
    PromptRuntime,
    ResourceRuntime,
    StatefulMcpRuntime,
    ToolCallResult,
    ToolExecutor,
    ToolRuntime,
 )
 ```
 Include `"PromptRuntime"`, `"ResourceRuntime"`, and `"ToolRuntime"` in `__all__`.
 - [ ] **Step 4: Run protocol tests**
 Run:
 ```bash
 uv run pytest tests/wf_sources_mcp/test_sdk_protocols.py -q
 ```
 Expected: pass.
 ---
 ### Task 2: Preserve Compatibility Exports
 **Files:**
 - Modify: `src/wf_mcp/sdk/base.py`
 - Modify: `src/wf_mcp/sdk/__init__.py`
 - Modify: `tests/wf_mcp/test_compat_imports.py`
 - [ ] **Step 1: Update compatibility shims**
 In `src/wf_mcp/sdk/base.py`, re-export all protocol slices:
 ```python
 from wf_sources_mcp.sdk import (
    BackendAdapter,
    PromptRuntime,
    ResourceRuntime,
    StatefulMcpRuntime,
    ToolCallResult,
    ToolRuntime,
 )
 __all__ = [
    "BackendAdapter",
    "PromptRuntime",
    "ResourceRuntime",
    "StatefulMcpRuntime",
    "ToolCallResult",
    "ToolRuntime",
 ]
 ```
 In `src/wf_mcp/sdk/__init__.py`, update imports/exports similarly while keeping `McpSdkAdapter`:
 ```python
 from wf_sources_mcp.sdk import (
    BackendAdapter,
    McpSdkAdapter,
    PromptRuntime,
    ResourceRuntime,
    StatefulMcpRuntime,
    ToolCallResult,
    ToolRuntime,
 )
 __all__ = [
    "BackendAdapter",
    "McpSdkAdapter",
    "PromptRuntime",
    "ResourceRuntime",
    "StatefulMcpRuntime",
    "ToolCallResult",
    "ToolRuntime",
 ]
 ```
 - [ ] **Step 2: Extend compatibility test**
 In `tests/wf_mcp/test_compat_imports.py`, extend `test_wf_mcp_sdk_protocol_shims_reexport_wf_sources_mcp_sdk`:
 ```python
    from wf_mcp.sdk import PromptRuntime as CompatPromptRuntime
    from wf_mcp.sdk import ResourceRuntime as CompatResourceRuntime
    from wf_mcp.sdk import ToolRuntime as CompatToolRuntime
    from wf_mcp.sdk.base import PromptRuntime as CompatBasePromptRuntime
    from wf_mcp.sdk.base import ResourceRuntime as CompatBaseResourceRuntime
    from wf_mcp.sdk.base import ToolRuntime as CompatBaseToolRuntime
    from wf_sources_mcp.sdk import PromptRuntime, ResourceRuntime, ToolRuntime
    assert CompatPromptRuntime is PromptRuntime
    assert CompatResourceRuntime is ResourceRuntime
    assert CompatToolRuntime is ToolRuntime
    assert CompatBasePromptRuntime is PromptRuntime
    assert CompatBaseResourceRuntime is ResourceRuntime
    assert CompatBaseToolRuntime is ToolRuntime
 ```
 - [ ] **Step 3: Run compatibility test**
 Run:
 ```bash
 uv run pytest tests/wf_mcp/test_compat_imports.py -q
 ```
 Expected: pass.
 ---
 ### Task 3: Add Persistent Resource/Prompt Listing Methods
 **Files:**
 - Modify: `src/wf_sources_mcp/runtime/session.py`
 - Modify: `src/wf_sources_mcp/runtime/factory.py`
 - Modify: `src/wf_sources_mcp/runtime/pool.py`
 - Modify: `tests/wf_sources_mcp/test_runtime.py`
 - [ ] **Step 1: Extend fake client with list methods**
 In `tests/wf_sources_mcp/test_runtime.py`, add imports:
 ```python
 from mcp.types import ListPromptsResult, ListResourcesResult, Prompt, Resource
 ```
 Update `_FakeClient` inside `_FakeFactory._create_with_stack()`:
 ```python
            async def list_resources(self) -> ListResourcesResult:
                return ListResourcesResult(
                    resources=[
                        Resource(
                            uri=AnyUrl("fixture://docs/runtime"),
                            name="resource.runtime",
                            title="Runtime Resource",
                            description="Runtime-scoped resource.",
                            mimeType="text/plain",
                        )
                    ]
                )
            async def list_prompts(self) -> ListPromptsResult:
                return ListPromptsResult(
                    prompts=[
                        Prompt(
                            name="prompt.runtime",
                            title="Runtime Prompt",
                            description="Runtime-scoped prompt.",
                            arguments=[],
                        )
                    ]
                )
 ```
 - [ ] **Step 2: Add runtime tests**
 Append to `tests/wf_sources_mcp/test_runtime.py`:
 ```python
@pytest.mark.asyncio
 async def test_persistent_session_factory_routes_resource_and_prompt_lists() -> None:
    factory = _FakeFactory()
    session = await factory.create(_connection(), None)
    resources = await session.list_resources()
    prompts = await session.list_prompts()
    await session.close()
    assert resources[0].name == "resource.runtime"
    assert resources[0].uri == "fixture://docs/runtime"
    assert prompts[0].name == "prompt.runtime"
@pytest.mark.asyncio
 async def test_runtime_pool_reuses_session_for_resource_and_prompt_lists() -> None:
    factory = _FakeFactory()
    pool = McpRuntimePool(factory.create)
    connection = _connection()
    resources = await pool.list_resources(connection, None)
    prompts = await pool.list_prompts(connection, None)
    await pool.close_all()
    assert resources[0].name == "resource.runtime"
    assert prompts[0].name == "prompt.runtime"
    assert factory.created_connections == [connection]
 ```
 Update public surface test:
 ```python
    assert "list_resources" in public_operations
    assert "list_prompts" in public_operations
    assert "invoke_method" not in public_operations
    assert "send_notification" not in public_operations
 ```
 - [ ] **Step 3: Add session callback types and methods**
 In `src/wf_sources_mcp/runtime/session.py`, import DTOs:
 ```python
 from wf_sources_mcp.catalog import DiscoveredPrompt, DiscoveredResource
 ```
 Add callback types:
 ```python
 RawResourceLister = Callable[[], Awaitable[list[DiscoveredResource]]]
 RawPromptLister = Callable[[], Awaitable[list[DiscoveredPrompt]]]
 ```
 Add fields:
 ```python
    list_resources_callback: RawResourceLister | None = None
    list_prompts_callback: RawPromptLister | None = None
 ```
 Add methods:
 ```python
    async def list_resources(self) -> list[DiscoveredResource]:
        """List MCP resources through the owner task or injected session."""
        if self.list_resources_callback is not None:
            return await self.list_resources_callback()
        if self.client is not None:
            from wf_sources_mcp.sdk.converters import resource_to_discovered
            result = await self.client.list_resources()
            return [resource_to_discovered(resource) for resource in result.resources]
        raise RuntimeError("persistent MCP session has no resource list transport")
    async def list_prompts(self) -> list[DiscoveredPrompt]:
        """List MCP prompts through the owner task or injected session."""
        if self.list_prompts_callback is not None:
            return await self.list_prompts_callback()
        if self.client is not None:
            from wf_sources_mcp.sdk.converters import prompt_to_discovered
            result = await self.client.list_prompts()
            return [prompt_to_discovered(prompt) for prompt in result.prompts]
        raise RuntimeError("persistent MCP session has no prompt list transport")
 ```
 - [ ] **Step 4: Wire owner methods**
 In `src/wf_sources_mcp/runtime/factory.py`, add callbacks in `PersistentSessionFactory.create()`:
 ```python
            list_resources_callback=owner.list_resources,
            list_prompts_callback=owner.list_prompts,
 ```
 Add owner methods:
 ```python
    async def list_resources(self) -> list[DiscoveredResource]:
        """Submit resource listing through the generic owner-task operation queue."""
        return await self.submit(
            operation="list_resources",
            run=lambda client: client.list_resources(),
        )
    async def list_prompts(self) -> list[DiscoveredPrompt]:
        """Submit prompt listing through the generic owner-task operation queue."""
        return await self.submit(
            operation="list_prompts",
            run=lambda client: client.list_prompts(),
        )
 ```
 Import DTOs:
 ```python
 from wf_sources_mcp.catalog import DiscoveredPrompt, DiscoveredResource
 ```
 - [ ] **Step 5: Add pool methods**
 In `src/wf_sources_mcp/runtime/pool.py`, import DTOs:
 ```python
 from wf_sources_mcp.catalog import DiscoveredPrompt, DiscoveredResource
 ```
 Add below `get_prompt()`:
 ```python
    async def list_resources(
        self,
        connection: McpSourceConnection,
        auth: AuthRecord | None,
    ) -> list[DiscoveredResource]:
        session = await self.get_session(connection, auth)
        return await session.list_resources()
    async def list_prompts(
        self,
        connection: McpSourceConnection,
        auth: AuthRecord | None,
    ) -> list[DiscoveredPrompt]:
        session = await self.get_session(connection, auth)
        return await session.list_prompts()
 ```
 - [ ] **Step 6: Run runtime tests**
 Run:
 ```bash
 uv run pytest tests/wf_sources_mcp/test_runtime.py -q
 ```
 Expected: pass.
 ---
 ### Task 4: Static Protocol Conformance Tests
 **Files:**
 - Modify: `tests/wf_sources_mcp/test_runtime.py`
 - [ ] **Step 1: Add assignment-based type conformance test**
 Append:
 ```python
 def test_runtime_pool_satisfies_stateful_protocol_static_shape() -> None:
    from wf_sources_mcp.sdk import (
        PromptRuntime,
        ResourceRuntime,
        StatefulMcpRuntime,
        ToolRuntime,
    )
    factory = _FakeFactory()
    pool = McpRuntimePool(factory.create)
    tool_runtime: ToolRuntime = pool
    resource_runtime: ResourceRuntime = pool
    prompt_runtime: PromptRuntime = pool
    stateful_runtime: StatefulMcpRuntime = pool
    assert tool_runtime is pool
    assert resource_runtime is pool
    assert prompt_runtime is pool
    assert stateful_runtime is pool
 ```
 This is a static typecheck target. Do not add `isinstance` runtime checks.
 - [ ] **Step 2: Run focused typecheck**
 Run:
 ```bash
 uv run basedpyright --level error tests/wf_sources_mcp/test_runtime.py
 ```
 Expected: `0 errors, 0 warnings, 0 notes`.
 ---
 ### Task 5: 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 after completion: `docs/superpowers/plans/2026-06-07-stateful-runtime-list-resources-prompts.md` to `docs/historical/superpowers/plans/2026-06-07-stateful-runtime-list-resources-prompts.md`
 - [ ] **Step 1: Update roadmap**
 In `docs/current_roadmap.md`, under the MCP upstream source runtime cleanup bullets, add:
 ```markdown
   - Completed: stateful MCP runtime now has protocol slices for tools,
     resources, and prompts, and can route session-scoped `list_resources` and
     `list_prompts` through the owner task. Catalog refresh still uses one-shot
     adapter policy.
 ```
 - [ ] **Step 2: Update long-lived API boundary spec**
 In `docs/superpowers/specs/2026-06-03-long-lived-workflow-api-boundary.md`, add after the content-access routing item:
 ```markdown
 14. Complete: stateful MCP runtime protocols split into tool/resource/prompt
    slices. Runtime can route `list_resources` and `list_prompts` through the
    owner task for session-scoped listings; catalog refresh remains one-shot.
 ```
 Renumber following item if needed.
 - [ ] **Step 3: Archive completed plan after implementation**
 Run only after all code/tests pass:
 ```bash
 git mv docs/superpowers/plans/2026-06-07-stateful-runtime-list-resources-prompts.md docs/historical/superpowers/plans/2026-06-07-stateful-runtime-list-resources-prompts.md
 ```
 ---
 ### Task 6: Final Verification
 **Files:**
 - All changed files.
 - [ ] **Step 1: Run focused tests**
 Run:
 ```bash
 uv run pytest tests/wf_sources_mcp tests/wf_mcp/test_compat_imports.py -q
 ```
 Expected: pass.
 - [ ] **Step 2: Run source typecheck**
 Run:
 ```bash
 uv run basedpyright --level error src
 ```
 Expected: `0 errors, 0 warnings, 0 notes`.
 - [ ] **Step 3: Run focused test typecheck**
 Run:
 ```bash
 uv run basedpyright --level error tests/wf_sources_mcp/test_runtime.py tests/wf_sources_mcp/test_sdk_protocols.py tests/wf_mcp/test_compat_imports.py
 ```
 Expected: `0 errors, 0 warnings, 0 notes`.
 - [ ] **Step 4: Run focused lint**
 Run:
 ```bash
 uv run ruff check src/wf_sources_mcp/sdk src/wf_sources_mcp/runtime src/wf_mcp/sdk tests/wf_sources_mcp/test_runtime.py tests/wf_sources_mcp/test_sdk_protocols.py tests/wf_mcp/test_compat_imports.py
 ```
 Expected: `All checks passed!`
 - [ ] **Step 5: Check raw runtime methods remain absent**
 Run:
 ```bash
 rg -n "def (invoke_method|send_notification)" src/wf_sources_mcp/runtime
 ```
 Expected: no matches.
 - [ ] **Step 6: Check whitespace**
 Run:
 ```bash
 git diff --check
 ```
 Expected: no whitespace errors. CRLF warnings are acceptable on Windows.
 ---
 ## Self-Review
 - Spec coverage: The plan adds session-scoped resource/prompt listings and splits protocols by capability without expanding raw MCP method support.
 - Placeholder scan: No placeholder steps remain.
 - Type consistency: resource/prompt list methods return `list[DiscoveredResource]` and `list[DiscoveredPrompt]`.
 - Policy boundary: catalog refresh/discovery policy remains separate and one-shot.