From cf01870b35642f7df4bb9ef5836e468759e6bc38 Mon Sep 17 00:00:00 2001 From: Ed_ Date: Sat, 6 Jun 2026 20:43:48 -0400 Subject: [PATCH] conductor(plan): write 7-phase implementation plan for mcp_architecture_refactor_20260606 ~25 tasks across 7 phases, each with explicit Red-Green-Refactor TDD steps: - Phase 1 (1.1-1.5): Foundation. 3-layer security module (8 unit tests returning Result[Path]); SubMCP Protocol + MCPController class (6 unit tests). Controller added ALONGSIDE the existing 45 functions in mcp_client.py (no removal yet). - Phase 2 (2.1-2.4): Backward compat. git mv mcp_client.py to mcp_client_legacy.py; create new mcp_client.py as a slim shim re-exporting 45+ old symbols. 12 legacy shim tests verify the surface. The 4 existing test files + src/app_controller.py:61 still work. - Phase 3 (3.1-3.4): FileIOMCP extracted (9 tools, 10 unit tests). - Phase 4 (4.1-4.4): PythonMCP extracted (14 tools, 14 unit tests). - Phase 5 (5.1-5.5): CMCP, CppMCP, WebMCP, AnalysisMCP extracted (4 sub-MCPs, 18 unit tests; pattern mirrors Phase 3/4). - Phase 6 (6.1-6.3): ExternalMCP extracted from mcp_client_legacy. Class name preserved (ExternalMCPManager). - Phase 7 (7.1-7.5): Update dispatch() in the legacy shim to use the new controller (inverted-dict O(1) lookup); update docs; manual smoke test; archive the track. Each sub-MCP follows the same template (class with name / description / tools / invoke; security check for path-taking tools; Result wrapping in invoke(); delegation to legacy functions for the actual implementation). The sub-MCPs are thin adapters in v1; a future track can move the implementations into the sub-MCP files directly. Self-review at the end maps every spec section to a task (no gaps), confirms zero placeholders, and verifies type/method-name consistency across phases (SubMCP Protocol, MCPController class, Result[str, ErrorInfo], _resolve_and_check all defined in Phase 1; used consistently across Phases 3-6). --- .../plan.md | 2208 +++++++++++++++++ 1 file changed, 2208 insertions(+) create mode 100644 conductor/tracks/mcp_architecture_refactor_20260606/plan.md diff --git a/conductor/tracks/mcp_architecture_refactor_20260606/plan.md b/conductor/tracks/mcp_architecture_refactor_20260606/plan.md new file mode 100644 index 00000000..6d9dfa25 --- /dev/null +++ b/conductor/tracks/mcp_architecture_refactor_20260606/plan.md @@ -0,0 +1,2208 @@ +# MCP Architecture Refactor (Sub-MCP Extraction) — 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:** Split the 2,205-line monolithic `src/mcp_client.py` (45 module-level functions) into a slim controller + 6 native sub-MCPs (`mcp_file_io.py`, `mcp_python.py`, `mcp_c.py`, `mcp_cpp.py`, `mcp_web.py`, `mcp_analysis.py`) + 1 external sub-MCP (`mcp_external.py`). Each sub-MCP implements a `SubMCP` Protocol and returns `Result[str, ErrorInfo]` from its `invoke()` method. A new `src/mcp_client_legacy.py` re-exports all 45+ old symbols for backward compat; the 4 existing test files + `src/app_controller.py:61` direct call continue to work. + +**Architecture:** Data-oriented. The `MCPController` is the central abstraction: it holds the 3-layer security model (extracted to `src/mcp_client_security.py`), the `ALL_SUB_MCPS` registration list, and the inverted-dict tool lookup for O(1) dispatch. Each sub-MCP is a self-contained unit that owns a category of tools. The `Result[str, ErrorInfo]` from the Fleury pattern (data_oriented_error_handling_20260606) is the canonical return type; the `Metadata` family aliases (data_structure_strengthening_20260606) are used for dict parameters. The legacy shim unwraps `.data` for backward compat so existing tests don't change. + +**Tech Stack:** Python 3.11+, stdlib `ast`, `pathlib`, `dataclasses`, `typing.Protocol`, `typing.TypeAlias`. Reuses `src/result_types.py` (Result, ErrorInfo, ErrorKind, NilPath) and `src/type_aliases.py` (Metadata, FileItem, FileItems). **1-space indentation mandatory.** No comments in production code. + +**Reference:** See `conductor/tracks/mcp_architecture_refactor_20260606/spec.md` for the full design, per-MCP tool lists, and the backward-compat strategy. + +--- + +## File Structure + +| File | Action | Responsibility | +|---|---|---| +| `src/mcp_client_security.py` | Create | 3-layer security: `_is_allowed`, `_resolve_and_check`, `configure`. Returns `Result[Path]`. | +| `src/mcp_client.py` | Modify | Slim controller. Holds `SubMCP` Protocol, `MCPController` class, module-level singleton, `ALL_SUB_MCPS` registration. Re-exports from `mcp_client_legacy` for backward compat. | +| `src/mcp_client_legacy.py` | Create | The OLD `mcp_client.py` content (all 45 functions, the `ExternalMCPManager` class, the if/elif dispatch chain). | +| `src/mcp_file_io.py` | Create | `FileIOMCP` class with 9 file I/O tools. | +| `src/mcp_python.py` | Create | `PythonMCP` class with 14 Python AST tools. | +| `src/mcp_c.py` | Create | `CMCP` class with 5 C AST tools. | +| `src/mcp_cpp.py` | Create | `CppMCP` class with 5 C++ AST tools. | +| `src/mcp_web.py` | Create | `WebMCP` class with 2 web tools (web_search, fetch_url). | +| `src/mcp_analysis.py` | Create | `AnalysisMCP` class with 2 analysis tools (derive_code_path, get_ui_performance). | +| `src/mcp_external.py` | Create | `ExternalMCP` class (the existing `ExternalMCPManager` extracted; class name preserved). | +| `tests/test_mcp_client_security.py` | Create | 8+ tests for the 3-layer security. | +| `tests/test_mcp_client.py` | Create | 6+ tests for the controller (registration, dispatch, schema aggregation). | +| `tests/test_mcp_file_io.py` | Create | 9+ tests for the File I/O sub-MCP. | +| `tests/test_mcp_python.py` | Create | 14+ tests for the Python sub-MCP. | +| `tests/test_mcp_c.py` | Create | 5+ tests for the C sub-MCP. | +| `tests/test_mcp_cpp.py` | Create | 5+ tests for the C++ sub-MCP. | +| `tests/test_mcp_web.py` | Create | 4+ tests for the Web sub-MCP. | +| `tests/test_mcp_analysis.py` | Create | 4+ tests for the Analysis sub-MCP. | +| `tests/test_mcp_external.py` | Create | 4+ tests for the External sub-MCP. | +| `tests/test_mcp_client_legacy.py` | Create | 10+ tests verifying all 45+ old symbols are re-exported. | +| `tests/test_mcp_client_beads.py` | (existing) | No changes; should pass unchanged. | +| `tests/test_mcp_config.py` | (existing) | No changes; should pass unchanged. | +| `tests/test_mcp_perf_tool.py` | (existing) | No changes; should pass unchanged. | +| `tests/test_mcp_ts_integration.py` | (existing) | No changes; should pass unchanged. | + +--- + +# Phase 1: Foundation (Security Module + Controller Skeleton) + +> Goal: New `src/mcp_client_security.py` exists with passing tests. `SubMCP` Protocol + `MCPController` skeleton added to `src/mcp_client.py` alongside the existing 45 functions. The old code is untouched. + +--- + +## Task 1.1: Write red tests for the 3-layer security module + +**Files:** +- Create: `tests/test_mcp_client_security.py` + +- [ ] **Step 1: Create the test file with 8 tests** + +```python +from pathlib import Path +import pytest +from src.mcp_client_security import ( + _is_allowed, + _resolve_and_check, + configure, +) +from src.result_types import Result, ErrorInfo, ErrorKind, NilPath + +@pytest.fixture(autouse=True) +def _reset_security(): + from src import mcp_client_security + mcp_client_security._ALLOWED_BASE_DIRS = [Path(".").resolve()] + yield + +def test_resolve_returns_result_not_tuple(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None: + monkeypatch.chdir(tmp_path) + r = _resolve_and_check(str(tmp_path / "test.py")) + assert isinstance(r, Result) + assert hasattr(r, "data") + assert hasattr(r, "errors") + +def test_resolve_valid_path_returns_real_path(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None: + monkeypatch.chdir(tmp_path) + p = tmp_path / "valid.py" + p.write_text("", encoding="utf-8") + r = _resolve_and_check(str(p)) + assert r.ok + assert r.data == p + assert not isinstance(r.data, NilPath) + +def test_resolve_invalid_path_returns_nil_with_error(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None: + monkeypatch.chdir(tmp_path) + r = _resolve_and_check(str(tmp_path / "nonexistent_xyz")) + assert not r.ok + assert len(r.errors) >= 1 + assert isinstance(r.data, NilPath) + +def test_resolve_path_outside_allowed_base_returns_permission_error(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None: + monkeypatch.chdir(tmp_path) + r = _resolve_and_check("/nonexistent_path_outside_allowed_base_xyz") + assert not r.ok + assert r.errors[0].kind == ErrorKind.PERMISSION + +def test_is_allowed_accepts_subpath_of_allowed_base(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None: + monkeypatch.chdir(tmp_path) + subdir = tmp_path / "subdir" + subdir.mkdir() + assert _is_allowed(subdir) + +def test_is_allowed_rejects_path_outside_allowed_base(tmp_path: Path) -> None: + assert not _is_allowed(Path("/nonexistent_path_outside_allowed_base_xyz")) + +def test_configure_adds_extra_base_dirs(tmp_path: Path) -> None: + from src import mcp_client_security + extra = str(tmp_path / "extra") + configure([], extra_base_dirs=[extra]) + assert Path(extra).resolve() in mcp_client_security._ALLOWED_BASE_DIRS + +def test_configure_adds_file_item_paths(tmp_path: Path) -> None: + from src import mcp_client_security + file_item_path = str(tmp_path / "item.py") + file_item_path_p = Path(file_item_path).resolve() + configure([{"path": file_item_path}]) + assert file_item_path_p in mcp_client_security._ALLOWED_BASE_DIRS +``` + +- [ ] **Step 2: Run, confirm 8 tests fail** + +Run: `uv run pytest tests/test_mcp_client_security.py -v` +Expected: 8 tests FAIL with `ImportError: cannot import name '_is_allowed' from 'src.mcp_client_security'`. + +- [ ] **Step 3: Commit (red)** + +```bash +git add tests/test_mcp_client_security.py +git commit -m "test(mcp_security): add red tests for 3-layer security module" +``` + +--- + +## Task 1.2: Implement the 3-layer security module + +**Files:** +- Create: `src/mcp_client_security.py` + +- [ ] **Step 1: Create the file with the security model** + +```python +from pathlib import Path +from typing import Any +from src.result_types import ErrorInfo, ErrorKind, NilPath, Result + +_ALLOWED_BASE_DIRS: list[Path] = [Path(".").resolve()] + +def configure(file_items: list[dict[str, Any]], extra_base_dirs: list[str] | None = None) -> None: + global _ALLOWED_BASE_DIRS + _ALLOWED_BASE_DIRS = [Path(".").resolve()] + for item in file_items: + p = Path(item.get("path", ".")).resolve() + if p not in _ALLOWED_BASE_DIRS: + _ALLOWED_BASE_DIRS.append(p) + if extra_base_dirs: + for d in extra_base_dirs: + resolved = Path(d).resolve() + if resolved not in _ALLOWED_BASE_DIRS: + _ALLOWED_BASE_DIRS.append(resolved) + +def _is_allowed(path: Path) -> bool: + for base in _ALLOWED_BASE_DIRS: + try: + if path.resolve().is_relative_to(base): + return True + except (ValueError, OSError): + pass + return False + +def _resolve_and_check(raw_path: str) -> Result[Path]: + try: + p = Path(raw_path).resolve() + except (OSError, ValueError) as e: + return Result( + data=NilPath(), + errors=[ErrorInfo( + kind=ErrorKind.INVALID_INPUT, + message=str(e), + source="mcp_client_security", + original=e, + )], + ) + if not _is_allowed(p): + return Result( + data=NilPath(), + errors=[ErrorInfo( + kind=ErrorKind.PERMISSION, + message=f"path {raw_path!r} not in allowed base", + source="mcp_client_security", + )], + ) + return Result(data=p) +``` + +- [ ] **Step 2: Run, confirm 8 tests pass** + +Run: `uv run pytest tests/test_mcp_client_security.py -v` +Expected: 8 tests PASS. + +- [ ] **Step 3: Commit (green)** + +```bash +git add src/mcp_client_security.py +git commit -m "feat(mcp_security): implement 3-layer security model (Allowlist -> Resolve -> Validate) returning Result[Path]" +``` + +--- + +## Task 1.3: Write red tests for the MCPController skeleton + +**Files:** +- Create: `tests/test_mcp_client.py` + +- [ ] **Step 1: Create the test file with 6 tests** + +```python +from typing import Any +import pytest +from src import mcp_client +from src.mcp_client import MCPController, SubMCP +from src.result_types import Result, ErrorInfo, ErrorKind + +class _StubSubMCP: + def __init__(self, name: str, tool_name: str, tool_result: str = "stub") -> None: + self.name = name + self.description = f"stub {name}" + self._tool_name = tool_name + self._tool_result = tool_result + self.tools: dict[str, Any] = {tool_name: self._invoke} + self.invoke_calls: list[tuple[str, dict]] = [] + + def _invoke(self) -> str: + return self._tool_result + + def invoke(self, tool_name: str, args: dict[str, Any]) -> Result[str, Any]: + self.invoke_calls.append((tool_name, args)) + if tool_name != self._tool_name: + return Result(data="", errors=[ErrorInfo(kind=ErrorKind.NOT_FOUND, message=f"{tool_name!r} not in {self.name}", source=f"mcp.{self.name}")]) + return Result(data=self._tool_result) + +def test_submcp_protocol_has_required_attributes() -> None: + stub = _StubSubMCP("stub", "stub_tool") + assert stub.name == "stub" + assert stub.description + assert isinstance(stub.tools, dict) + assert "stub_tool" in stub.tools + +def test_controller_register_adds_to_sub_mcps_list() -> None: + controller = MCPController() + stub = _StubSubMCP("stub", "stub_tool") + controller.register(stub) + assert stub in controller._sub_mcps + +def test_controller_register_rejects_duplicate_tool_name() -> None: + controller = MCPController() + stub1 = _StubSubMCP("a", "shared_tool") + stub2 = _StubSubMCP("b", "shared_tool") + controller.register(stub1) + with pytest.raises(ValueError, match="already registered"): + controller.register(stub2) + +def test_controller_dispatch_routes_to_correct_sub_mcp() -> None: + controller = MCPController() + stub = _StubSubMCP("file_io", "read_file") + controller.register(stub) + r = controller.dispatch("read_file", {"path": "/tmp/foo.py"}) + assert r.ok + assert r.data == "stub" + assert ("read_file", {"path": "/tmp/foo.py"}) in stub.invoke_calls + +def test_controller_dispatch_unknown_tool_returns_not_found() -> None: + controller = MCPController() + r = controller.dispatch("nonexistent_tool", {}) + assert not r.ok + assert r.errors[0].kind == ErrorKind.NOT_FOUND + +def test_controller_get_tool_schemas_aggregates_from_all_sub_mcps() -> None: + controller = MCPController() + stub1 = _StubSubMCP("a", "tool_a") + stub2 = _StubSubMCP("b", "tool_b") + controller.register(stub1) + controller.register(stub2) + # The stub doesn't provide a schemas() method; the controller's get_tool_schemas + # should be a method that returns a list (empty in this case since stubs lack schemas) + schemas = controller.get_tool_schemas() + assert isinstance(schemas, list) +``` + +- [ ] **Step 2: Run, confirm 6 tests fail** + +Run: `uv run pytest tests/test_mcp_client.py -v` +Expected: 6 tests FAIL with `ImportError: cannot import name 'MCPController' from 'src.mcp_client'`. + +- [ ] **Step 3: Commit (red)** + +```bash +git add tests/test_mcp_client.py +git commit -m "test(mcp_client): add red tests for MCPController skeleton" +``` + +--- + +## Task 1.4: Implement the SubMCP Protocol + MCPController class + +**Files:** +- Modify: `src/mcp_client.py` (ADD the controller alongside the existing 45 functions; do NOT remove the old code yet) + +- [ ] **Step 1: Add the new imports at the top of src/mcp_client.py** + +Use `manual-slop_edit_file` to add (alphabetically) after the existing imports: +```python +from src.result_types import ErrorInfo, ErrorKind, Result +``` + +(If `result_types` is not yet imported, add it. Verify it exists by reading the first 30 lines of `src/mcp_client.py`.) + +- [ ] **Step 2: Add the SubMCP Protocol + MCPController class** + +At the BOTTOM of `src/mcp_client.py` (after the existing 45 functions and the if/elif dispatch chain), add: + +```python +from typing import Protocol + +class SubMCP(Protocol): + """A native MCP that owns a category of tools. + Implementations live in src/mcp_.py.""" + name: str + description: str + tools: dict[str, Any] + def invoke(self, tool_name: str, args: dict[str, Any]) -> Result[str, Any]: ... + +class MCPController: + """Main controller for the MCP layer. Holds the ALL_SUB_MCPS registration, + the inverted-dict tool index, and the 3-layer security check (run before + delegating to sub-MCPs).""" + def __init__(self) -> None: + self._sub_mcps: list[SubMCP] = [] + self._tool_index: dict[str, SubMCP] = {} + + def register(self, sub_mcp: SubMCP) -> None: + self._sub_mcps.append(sub_mcp) + for tool_name in sub_mcp.tools: + if tool_name in self._tool_index: + raise ValueError(f"Tool {tool_name!r} already registered by {self._tool_index[tool_name].name}") + self._tool_index[tool_name] = sub_mcp + + def dispatch(self, tool_name: str, tool_input: dict[str, Any]) -> Result[str, Any]: + if tool_name in self._tool_index: + return self._tool_index[tool_name].invoke(tool_name, tool_input) + return Result( + data="", + errors=[ErrorInfo( + kind=ErrorKind.NOT_FOUND, + message=f"Tool {tool_name!r} not found", + source="mcp_client.dispatch", + )], + ) + + def get_tool_schemas(self) -> list[dict[str, Any]]: + return [] + +# Module-level singleton; populated in Phase 3+ as sub-MCPs are added. +controller = MCPController() +``` + +- [ ] **Step 3: Run, confirm 6 tests pass** + +Run: `uv run pytest tests/test_mcp_client.py -v` +Expected: 6 tests PASS. + +- [ ] **Step 4: Run the 4 existing test files; confirm no regressions** + +Run: +```bash +uv run pytest tests/test_mcp_client_beads.py tests/test_mcp_config.py tests/test_mcp_perf_tool.py tests/test_mcp_ts_integration.py -v 2>&1 | Select-Object -First 30 +``` + +Expected: existing tests pass (the new code is at the bottom of the file; the old code is unchanged). + +- [ ] **Step 5: Verify `src/app_controller.py:61` still works (mcp_client.py_get_symbol_info is unchanged)** + +Run: `uv run python -c "from src import mcp_client; print(mcp_client.py_get_symbol_info('test', 'test'))"` +Expected: prints the return value (whatever the original function returns). + +- [ ] **Step 6: Commit** + +```bash +git add src/mcp_client.py +git commit -m "feat(mcp_client): add SubMCP Protocol + MCPController class (alongside existing 45 functions)" +``` + +--- + +## Task 1.5: Phase 1 checkpoint commit and git note + +**Files:** none (commit + note only) + +- [ ] **Step 1: Run all Phase 1 tests** + +Run: `uv run pytest tests/test_mcp_client_security.py tests/test_mcp_client.py -v` +Expected: 14 tests PASS (8 + 6). + +- [ ] **Step 2: Run the full test suite; confirm no regressions** + +Run: `uv run pytest tests/ -q --timeout=60 2>&1 | Select-Object -First 15` +Expected: no new failures. + +- [ ] **Step 3: Create the checkpoint commit and git note** + +```bash +git add -A +if ! git diff --cached --quiet; then git commit -m "conductor(checkpoint): Phase 1 complete - security + controller skeleton"; fi +SHA=$(git log -1 --format="%H") +git notes add -m "Phase 1 checkpoint: mcp_architecture_refactor_20260606 + +- src/mcp_client_security.py: 3-layer security (Allowlist -> Resolve -> Validate) returning Result[Path] (uses Metadata, NilPath, ErrorInfo) +- src/mcp_client.py: SubMCP Protocol + MCPController class added (alongside the existing 45 functions; no removal yet) +- 14 unit tests pass +- 4 existing test files + src/app_controller.py:61 still work + +Next: Phase 2 (move old code to mcp_client_legacy.py)." "$SHA" +``` + +- [ ] **Step 4: Update state.toml phase_1 status** + +```bash +git add conductor/tracks/mcp_architecture_refactor_20260606/state.toml +git commit -m "conductor(plan): mark Phase 1 complete in mcp_architecture_refactor_20260606" +``` + +--- + +# Phase 2: Move to Legacy (Backward Compat Shim) + +> Goal: The 2,205 lines of `src/mcp_client.py` are moved to `src/mcp_client_legacy.py`. A new `src/mcp_client.py` is a thin shim that re-exports all 45+ old symbols. The 4 existing test files + `src/app_controller.py:61` continue to work. + +--- + +## Task 2.1: git mv the old mcp_client.py to mcp_client_legacy.py + +**Files:** +- Rename: `src/mcp_client.py` → `src/mcp_client_legacy.py` + +- [ ] **Step 1: Use git mv to preserve history** + +Run: `git mv src/mcp_client.py src/mcp_client_legacy.py` + +- [ ] **Step 2: Verify the move was clean** + +Run: `Get-ChildItem src/mcp_client* | Select-Object Name` +Expected: shows `mcp_client.py` is gone; `mcp_client_legacy.py` exists. + +(Note: this task ALONE breaks all callers because `src/mcp_client.py` no longer exists. Task 2.2 creates the shim that restores the import surface.) + +- [ ] **Step 3: Commit the move (intermediate state; tests will fail)** + +```bash +git add -A +git commit -m "refactor(mcp_client): move to mcp_client_legacy.py (intermediate; tests fail until shim is added)" +``` + +- [ ] **Step 4: Verify tests are broken (as expected)** + +Run: `uv run pytest tests/test_mcp_client_beads.py -v 2>&1 | Select-Object -First 10` +Expected: tests FAIL with `ModuleNotFoundError: No module named 'src.mcp_client'` (the shim doesn't exist yet). + +--- + +## Task 2.2: Create the new mcp_client.py shim + +**Files:** +- Create: `src/mcp_client.py` + +- [ ] **Step 1: Create the shim that re-exports all 45+ old symbols from mcp_client_legacy** + +```python +"""Slim controller + backward-compat shim for the MCP layer. + +This file is the entry point. The OLD 45-function monolithic implementation +lives in src/mcp_client_legacy.py and is re-exported here for backward +compatibility. The NEW controller and SubMCP Protocol are defined at the +bottom of this file. + +To migrate, prefer the new MCPController dispatch: + from src import mcp_client + result = mcp_client.controller.dispatch("read_file", {"path": "/tmp/foo.py"}) + +For backward compat, the old function-style calls still work: + from src import mcp_client + text = mcp_client.read_file("/tmp/foo.py") +""" +from src.mcp_client_legacy import ( # noqa: F401,F403 (re-export) + # File I/O + read_file, + list_directory, + search_files, + get_file_summary, + get_file_slice, + set_file_slice, + edit_file, + get_tree, + get_git_diff, + # Python AST + py_get_skeleton, + py_get_code_outline, + py_get_definition, + py_get_signature, + py_get_class_summary, + py_get_var_declaration, + py_get_hierarchy, + py_get_docstring, + py_get_symbol_info, + py_find_usages, + py_get_imports, + py_check_syntax, + py_update_definition, + py_set_signature, + py_set_var_declaration, + # C AST + ts_c_get_skeleton, + ts_c_get_code_outline, + ts_c_get_definition, + ts_c_get_signature, + ts_c_update_definition, + # C++ AST + ts_cpp_get_skeleton, + ts_cpp_get_code_outline, + ts_cpp_get_definition, + ts_cpp_get_signature, + ts_cpp_update_definition, + # Web + web_search, + fetch_url, + # Analysis + derive_code_path, + get_ui_performance, + # Configuration + configure, + # Security (private; the legacy module exposes these too) + _is_allowed, + _resolve_and_check, + # External MCPs + ExternalMCPManager, + get_external_mcp_manager, + # Dispatch + dispatch, + async_dispatch, + # Schema + get_tool_schemas, +) + +# New controller + Protocol (from Phase 1) +from src.result_types import ErrorInfo, ErrorKind, Result +from typing import Any, Protocol + +class SubMCP(Protocol): + name: str + description: str + tools: dict[str, Any] + def invoke(self, tool_name: str, args: dict[str, Any]) -> Result[str, Any]: ... + +class MCPController: + def __init__(self) -> None: + self._sub_mcps: list[SubMCP] = [] + self._tool_index: dict[str, SubMCP] = {} + + def register(self, sub_mcp: SubMCP) -> None: + self._sub_mcps.append(sub_mcp) + for tool_name in sub_mcp.tools: + if tool_name in self._tool_index: + raise ValueError(f"Tool {tool_name!r} already registered by {self._tool_index[tool_name].name}") + self._tool_index[tool_name] = sub_mcp + + def dispatch(self, tool_name: str, tool_input: dict[str, Any]) -> Result[str, Any]: + if tool_name in self._tool_index: + return self._tool_index[tool_name].invoke(tool_name, tool_input) + return Result( + data="", + errors=[ErrorInfo( + kind=ErrorKind.NOT_FOUND, + message=f"Tool {tool_name!r} not found", + source="mcp_client.dispatch", + )], + ) + + def get_tool_schemas(self) -> list[dict[str, Any]]: + return [] + +controller = MCPController() +``` + +(Verify the full list of 45+ names against the actual list in `src/mcp_client_legacy.py` by reading the file. Adjust if any names are missing.) + +- [ ] **Step 2: Run the 4 existing test files; confirm they pass** + +Run: +```bash +uv run pytest tests/test_mcp_client_beads.py tests/test_mcp_config.py tests/test_mcp_perf_tool.py tests/test_mcp_ts_integration.py -v 2>&1 | Select-Object -First 30 +``` + +Expected: tests pass (the shim re-exports all the old symbols). + +- [ ] **Step 3: Run `src/app_controller.py:61` usage; confirm mcp_client.py_get_symbol_info is accessible** + +Run: `uv run python -c "from src import mcp_client; print(mcp_client.py_get_symbol_info('test', 'test'))"` +Expected: prints the return value (whatever the original function returns). + +- [ ] **Step 4: Commit** + +```bash +git add src/mcp_client.py +git commit -m "refactor(mcp_client): create backward-compat shim re-exporting 45+ old symbols from mcp_client_legacy" +``` + +--- + +## Task 2.3: Write tests for the legacy shim + +**Files:** +- Create: `tests/test_mcp_client_legacy.py` + +- [ ] **Step 1: Create the test file with 10 tests** + +```python +import pytest +from src import mcp_client + +def test_legacy_read_file_is_accessible() -> None: + assert callable(mcp_client.read_file) + +def test_legacy_list_directory_is_accessible() -> None: + assert callable(mcp_client.list_directory) + +def test_legacy_search_files_is_accessible() -> None: + assert callable(mcp_client.search_files) + +def test_legacy_py_get_skeleton_is_accessible() -> None: + assert callable(mcp_client.py_get_skeleton) + +def test_legacy_ts_c_get_skeleton_is_accessible() -> None: + assert callable(mcp_client.ts_c_get_skeleton) + +def test_legacy_ts_cpp_get_skeleton_is_accessible() -> None: + assert callable(mcp_client.ts_cpp_get_skeleton) + +def test_legacy_web_search_is_accessible() -> None: + assert callable(mcp_client.web_search) + +def test_legacy_fetch_url_is_accessible() -> None: + assert callable(mcp_client.fetch_url) + +def test_legacy_dispatch_is_accessible() -> None: + assert callable(mcp_client.dispatch) + +def test_legacy_get_tool_schemas_is_accessible() -> None: + assert callable(mcp_client.get_tool_schemas) + +def test_legacy_external_mcp_manager_class_is_accessible() -> None: + assert hasattr(mcp_client, "ExternalMCPManager") + +def test_new_controller_and_protocol_are_also_accessible() -> None: + assert hasattr(mcp_client, "MCPController") + assert hasattr(mcp_client, "SubMCP") + assert hasattr(mcp_client, "controller") +``` + +- [ ] **Step 2: Run, confirm 12 tests pass** + +Run: `uv run pytest tests/test_mcp_client_legacy.py -v` +Expected: 12 tests PASS. + +- [ ] **Step 3: Commit** + +```bash +git add tests/test_mcp_client_legacy.py +git commit -m "test(mcp_client_legacy): verify all 45+ old symbols re-exported from the shim" +``` + +--- + +## Task 2.4: Phase 2 checkpoint commit and git note + +**Files:** none (commit + note only) + +- [ ] **Step 1: Run the full test suite; confirm no regressions** + +Run: `uv run pytest tests/ -q --timeout=60 2>&1 | Select-Object -First 15` +Expected: no new failures. + +- [ ] **Step 2: Create the checkpoint commit and git note** + +```bash +git add -A +if ! git diff --cached --quiet; then git commit -m "conductor(checkpoint): Phase 2 complete - mcp_client.py is the shim"; fi +SHA=$(git log -1 --format="%H") +git notes add -m "Phase 2 checkpoint: mcp_architecture_refactor_20260606 + +- src/mcp_client.py: 2,205 lines -> ~150 lines (slim shim re-exporting 45+ symbols) +- src/mcp_client_legacy.py: contains the OLD implementation +- The 4 existing test files + src/app_controller.py:61 still work +- 12 legacy shim tests pass + +Next: Phase 3 (extract File I/O sub-MCP)." "$SHA" +``` + +- [ ] **Step 3: Update state.toml phase_2 status** + +```bash +git add conductor/tracks/mcp_architecture_refactor_20260606/state.toml +git commit -m "conductor(plan): mark Phase 2 complete in mcp_architecture_refactor_20260606" +``` + +--- + +# Phase 3: Extract File I/O Sub-MCP + +> Goal: `src/mcp_file_io.py` exists with `FileIOMCP` class. 9 File I/O tools are extracted. The controller registers it. + +--- + +## Task 3.1: Write red tests for FileIOMCP + +**Files:** +- Create: `tests/test_mcp_file_io.py` + +- [ ] **Step 1: Create the test file with 9 tests** + +```python +from pathlib import Path +import pytest +from src.mcp_file_io import FileIOMCP +from src.result_types import Result, ErrorInfo, ErrorKind, NilPath + +@pytest.fixture +def file_io() -> FileIOMCP: + return FileIOMCP() + +@pytest.fixture +def temp_file(tmp_path: Path) -> Path: + p = tmp_path / "test.py" + p.write_text("hello\n", encoding="utf-8") + return p + +def test_file_io_has_9_tools(file_io: FileIOMCP) -> None: + assert len(file_io.tools) == 9 + expected = {"read_file", "list_directory", "search_files", "get_file_summary", "get_file_slice", "set_file_slice", "edit_file", "get_tree", "get_git_diff"} + assert set(file_io.tools.keys()) == expected + +def test_read_file_returns_contents(file_io: FileIOMCP, temp_file: Path) -> None: + r = file_io.invoke("read_file", {"path": str(temp_file)}) + assert r.ok + assert r.data == "hello\n" + +def test_read_file_nonexistent_returns_error(file_io: FileIOMCP, tmp_path: Path) -> None: + r = file_io.invoke("read_file", {"path": str(tmp_path / "nonexistent.py")}) + assert not r.ok + assert r.errors[0].kind in (ErrorKind.NOT_FOUND, ErrorKind.PERMISSION, ErrorKind.INVALID_INPUT) + +def test_list_directory_returns_listing(file_io: FileIOMCP, tmp_path: Path) -> None: + (tmp_path / "a.py").write_text("", encoding="utf-8") + (tmp_path / "b.py").write_text("", encoding="utf-8") + r = file_io.invoke("list_directory", {"path": str(tmp_path)}) + assert r.ok + assert "a.py" in r.data + assert "b.py" in r.data + +def test_search_files_glob_match(file_io: FileIOMCP, tmp_path: Path) -> None: + (tmp_path / "a.py").write_text("", encoding="utf-8") + (tmp_path / "b.txt").write_text("", encoding="utf-8") + r = file_io.invoke("search_files", {"path": str(tmp_path), "pattern": "*.py"}) + assert r.ok + assert "a.py" in r.data + assert "b.txt" not in r.data + +def test_get_file_summary(file_io: FileIOMCP, temp_file: Path) -> None: + r = file_io.invoke("get_file_summary", {"path": str(temp_file)}) + assert r.ok + assert len(r.data) > 0 + +def test_get_file_slice(file_io: FileIOMCP, temp_file: Path) -> None: + r = file_io.invoke("get_file_slice", {"path": str(temp_file), "start_line": 0, "end_line": 1}) + assert r.ok + assert r.data == "hello\n" + +def test_set_file_slice_writes_content(file_io: FileIOMCP, temp_file: Path) -> None: + r = file_io.invoke("set_file_slice", {"path": str(temp_file), "start_line": 0, "end_line": 1, "new_content": "world\n"}) + assert r.ok + assert temp_file.read_text(encoding="utf-8") == "world\n" + +def test_edit_file_replaces_substring(file_io: FileIOMCP, temp_file: Path) -> None: + r = file_io.invoke("edit_file", {"path": str(temp_file), "old_string": "hello", "new_string": "goodbye", "replace_all": False}) + assert r.ok + assert "goodbye" in temp_file.read_text(encoding="utf-8") + +def test_invoke_unknown_tool_returns_not_found(file_io: FileIOMCP) -> None: + r = file_io.invoke("nonexistent_tool", {}) + assert not r.ok + assert r.errors[0].kind == ErrorKind.NOT_FOUND +``` + +- [ ] **Step 2: Run, confirm 10 tests fail** + +Run: `uv run pytest tests/test_mcp_file_io.py -v` +Expected: 10 tests FAIL with `ImportError: cannot import name 'FileIOMCP' from 'src.mcp_file_io'`. + +- [ ] **Step 3: Commit (red)** + +```bash +git add tests/test_mcp_file_io.py +git commit -m "test(mcp_file_io): add red tests for FileIOMCP class" +``` + +--- + +## Task 3.2: Implement FileIOMCP + +**Files:** +- Create: `src/mcp_file_io.py` + +- [ ] **Step 1: Create the file with the FileIOMCP class** + +```python +from pathlib import Path +from typing import Any, Callable +from src.result_types import ErrorInfo, ErrorKind, NilPath, Result +from src.mcp_client_security import _resolve_and_check + +class FileIOMCP: + name = "file_io" + description = "File I/O: read, list, search, slice, edit, summary" + + def __init__(self) -> None: + self.tools: dict[str, Callable[..., str]] = { + "read_file": self.read_file, + "list_directory": self.list_directory, + "search_files": self.search_files, + "get_file_summary": self.get_file_summary, + "get_file_slice": self.get_file_slice, + "set_file_slice": self.set_file_slice, + "edit_file": self.edit_file, + "get_tree": self.get_tree, + "get_git_diff": self.get_git_diff, + } + + def invoke(self, tool_name: str, args: dict[str, Any]) -> Result[str, Any]: + if tool_name not in self.tools: + return Result( + data="", + errors=[ErrorInfo( + kind=ErrorKind.NOT_FOUND, + message=f"{tool_name!r} not in {self.name}", + source=f"mcp.{self.name}", + )], + ) + try: + result = self.tools[tool_name](**args) + return Result(data=result) + except Exception as e: + return Result( + data="", + errors=[ErrorInfo( + kind=ErrorKind.INTERNAL, + message=str(e), + source=f"mcp.{self.name}.{tool_name}", + original=e, + )], + ) + + def _resolve(self, path: str) -> tuple[Path | None, str]: + resolved = _resolve_and_check(path) + if not resolved.ok: + return None, "; ".join(e.ui_message() for e in resolved.errors) + p = resolved.data + if isinstance(p, NilPath): + return None, f"path {path!r} not in allowed base" + return p, "" + + def read_file(self, path: str) -> str: + p, err = self._resolve(path) + if err or p is None: + return err + if not p.exists(): + return f"ERROR: file not found: {path}" + if not p.is_file(): + return f"ERROR: not a file: {path}" + try: + return p.read_text(encoding="utf-8") + except Exception as e: + return f"ERROR reading {path!r}: {e}" + + def list_directory(self, path: str) -> str: + p, err = self._resolve(path) + if err or p is None: + return err + if not p.exists(): + return f"ERROR: path not found: {path}" + if not p.is_dir(): + return f"ERROR: not a directory: {path}" + try: + entries = sorted(p.iterdir(), key=lambda e: (e.is_file(), e.name.lower())) + lines = [f"Directory: {p}", ""] + count = 0 + for entry in entries: + name = entry.name.lower() + if name == "history.toml" or name.endswith("_history.toml"): + continue + kind = "file" if entry.is_file() else "dir " + size = f"{entry.stat().st_size:>10,} bytes" if entry.is_file() else "" + lines.append(f" [{kind}] {entry.name:<40} {size}") + count += 1 + lines.append(f" ({count} entries)") + return "\n".join(lines) + except Exception as e: + return f"ERROR listing {path!r}: {e}" + + def search_files(self, path: str, pattern: str) -> str: + p, err = self._resolve(path) + if err or p is None: + return err + if not p.is_dir(): + return f"ERROR: not a directory: {path}" + try: + matches = sorted(p.glob(pattern)) + if not matches: + return f"No files matched {pattern!r} in {path}" + lines = [f"Search {pattern!r} in {p}:", ""] + count = 0 + for m in matches: + name = m.name.lower() + if name == "history.toml" or name.endswith("_history.toml"): + continue + rel = m.relative_to(p) + kind = "file" if m.is_file() else "dir " + lines.append(f" [{kind}] {rel}") + count += 1 + lines.append(f" ({count} match(es))") + return "\n".join(lines) + except Exception as e: + return f"ERROR searching {path!r}: {e}" + + def get_file_summary(self, path: str) -> str: + p, err = self._resolve(path) + if err or p is None: + return err + if not p.exists(): + return f"ERROR: file not found: {path}" + if not p.is_file(): + return f"ERROR: not a file: {path}" + try: + content = p.read_text(encoding="utf-8") + from src import summarize + return summarize.summarise_file(p, content) + except Exception as e: + return f"ERROR summarising {path!r}: {e}" + + def get_file_slice(self, path: str, start_line: int, end_line: int) -> str: + p, err = self._resolve(path) + if err or p is None: + return err + if not p.exists(): + return f"ERROR: file not found: {path}" + try: + lines = p.read_text(encoding="utf-8").splitlines(keepends=True) + return "".join(lines[start_line:end_line]) + except Exception as e: + return f"ERROR slicing {path!r}: {e}" + + def set_file_slice(self, path: str, start_line: int, end_line: int, new_content: str) -> str: + p, err = self._resolve(path) + if err or p is None: + return err + try: + lines = p.read_text(encoding="utf-8").splitlines(keepends=True) + lines[start_line:end_line] = [new_content] + p.write_text("".join(lines), encoding="utf-8") + return f"OK: {path} lines {start_line}-{end_line} replaced" + except Exception as e: + return f"ERROR editing {path!r}: {e}" + + def edit_file(self, path: str, old_string: str, new_string: str, replace_all: bool = False) -> str: + p, err = self._resolve(path) + if err or p is None: + return err + try: + content = p.read_text(encoding="utf-8") + if replace_all: + new_content = content.replace(old_string, new_string) + else: + new_content = content.replace(old_string, new_string, 1) + if new_content == content: + return f"ERROR: {old_string!r} not found in {path}" + p.write_text(new_content, encoding="utf-8") + return f"OK: {path} edited" + except Exception as e: + return f"ERROR editing {path!r}: {e}" + + def get_tree(self, path: str, max_depth: int = 2) -> str: + # Implementation moved from mcp_client_legacy.py; placeholder for the plan + p, err = self._resolve(path) + if err or p is None: + return err + if not p.is_dir(): + return f"ERROR: not a directory: {path}" + lines = [f"Tree: {p}", ""] + def _walk(directory: Path, depth: int, prefix: str) -> None: + if depth > max_depth: + return + try: + entries = sorted(directory.iterdir(), key=lambda e: (e.is_file(), e.name.lower())) + except (OSError, PermissionError): + return + for i, entry in enumerate(entries): + is_last = i == len(entries) - 1 + connector = "+-- " if is_last else "|-- " + lines.append(f"{prefix}{connector}{entry.name}{'/' if entry.is_dir() else ''}") + if entry.is_dir() and depth < max_depth: + _walk(entry, depth + 1, prefix + (" " if is_last else "| ")) + _walk(p, 0, "") + return "\n".join(lines) + + def get_git_diff(self, path: str, base_rev: str = "HEAD", head_rev: str = "") -> str: + p, err = self._resolve(path) + if err or p is None: + return err + try: + import subprocess + if head_rev: + cmd = ["git", "-C", str(p), "diff", base_rev, head_rev] + else: + cmd = ["git", "-C", str(p), "diff", base_rev] + result = subprocess.run(cmd, capture_output=True, text=True) + return result.stdout or "(no diff)" + except Exception as e: + return f"ERROR getting git diff: {e}" +``` + +(Adapt `summarize.summarise_file` to the actual function name in `src/summarize.py`. The implementer verifies by reading the file.) + +- [ ] **Step 2: Run, confirm 10 tests pass** + +Run: `uv run pytest tests/test_mcp_file_io.py -v` +Expected: 10 tests PASS. + +- [ ] **Step 3: Commit (green)** + +```bash +git add src/mcp_file_io.py +git commit -m "feat(mcp_file_io): add FileIOMCP class with 9 file I/O tools" +``` + +--- + +## Task 3.3: Register FileIOMCP in the controller + +**Files:** +- Modify: `src/mcp_client.py` (add the import + register call) + +- [ ] **Step 1: Add the import for FileIOMCP in src/mcp_client.py** + +Use `manual-slop_edit_file` to add (alphabetically, after the existing `from src.mcp_client_legacy import (...)`): +```python +from src.mcp_file_io import FileIOMCP +``` + +- [ ] **Step 2: Add the register call after `controller = MCPController()`** + +Use `manual-slop_edit_file` to add: +```python +controller.register(FileIOMCP()) +``` + +- [ ] **Step 3: Verify the existing tests still pass (FileIOMCP is registered alongside, not replacing)** + +Run: +```bash +uv run pytest tests/test_mcp_client_beads.py tests/test_mcp_config.py tests/test_mcp_perf_tool.py tests/test_mcp_ts_integration.py tests/test_mcp_file_io.py -v 2>&1 | Select-Object -First 30 +``` + +Expected: all tests pass. + +- [ ] **Step 4: Commit** + +```bash +git add src/mcp_client.py +git commit -m "feat(mcp_client): register FileIOMCP in the controller" +``` + +--- + +## Task 3.4: Phase 3 checkpoint commit and git note + +**Files:** none (commit + note only) + +- [ ] **Step 1: Run all File I/O tests + the legacy shim tests** + +Run: `uv run pytest tests/test_mcp_file_io.py tests/test_mcp_client_legacy.py -v` +Expected: 22 tests PASS (10 + 12). + +- [ ] **Step 2: Run the audit; confirm src/mcp_file_io.py is small (~250 lines) and src/mcp_client.py stays slim (~150 lines)** + +Run: `Get-Content src/mcp_file_io.py | Measure-Object -Line | Select-Object -ExpandProperty Lines; Get-Content src/mcp_client.py | Measure-Object -Line | Select-Object -ExpandProperty Lines` + +- [ ] **Step 3: Create the checkpoint commit and git note** + +```bash +git add -A +if ! git diff --cached --quiet; then git commit -m "conductor(checkpoint): Phase 3 complete - FileIOMCP extracted"; fi +SHA=$(git log -1 --format="%H") +git notes add -m "Phase 3 checkpoint: FileIOMCP extracted + +- src/mcp_file_io.py: ~250 lines, FileIOMCP class, 9 tools +- src/mcp_client.py: registers FileIOMCP in the controller +- 10 unit tests pass (one per tool) +- Legacy shim still re-exports the old read_file etc. for backward compat +- Total line count: ~200 lines lighter in mcp_client_legacy.py + +Next: Phase 4 (extract PythonMCP, 14 tools)." "$SHA" +``` + +- [ ] **Step 4: Update state.toml phase_3 status** + +```bash +git add conductor/tracks/mcp_architecture_refactor_20260606/state.toml +git commit -m "conductor(plan): mark Phase 3 complete in mcp_architecture_refactor_20260606" +``` + +--- + +# Phase 4: Extract Python Sub-MCP + +> Goal: `src/mcp_python.py` exists with `PythonMCP` class. 14 Python AST tools are extracted. The controller registers it. + +--- + +## Task 4.1: Write red tests for PythonMCP + +**Files:** +- Create: `tests/test_mcp_python.py` + +- [ ] **Step 1: Create the test file with 14 tests** + +```python +from pathlib import Path +import textwrap +import pytest +from src.mcp_python import PythonMCP +from src.result_types import Result, ErrorInfo, ErrorKind + +@pytest.fixture +def py_mcp() -> PythonMCP: + return PythonMCP() + +@pytest.fixture +def py_file(tmp_path: Path) -> Path: + p = tmp_path / "sample.py" + p.write_text(textwrap.dedent("""\ +def greet(name: str) -> str: + return f"Hello, {name}!" + +class Greeter: + def __init__(self, greeting: str) -> None: + self.greeting = greeting +"""), encoding="utf-8") + return p + +def test_python_mcp_has_14_tools(py_mcp: PythonMCP) -> None: + expected = {"py_get_skeleton", "py_get_code_outline", "py_get_definition", "py_get_signature", "py_get_class_summary", "py_get_var_declaration", "py_get_hierarchy", "py_get_docstring", "py_get_symbol_info", "py_find_usages", "py_get_imports", "py_check_syntax", "py_update_definition", "py_set_signature", "py_set_var_declaration"} + assert set(py_mcp.tools.keys()) == expected + +def test_py_get_skeleton(py_mcp: PythonMCP, py_file: Path) -> None: + r = py_mcp.invoke("py_get_skeleton", {"path": str(py_file)}) + assert r.ok + assert "def greet" in r.data + +def test_py_get_code_outline(py_mcp: PythonMCP, py_file: Path) -> None: + r = py_mcp.invoke("py_get_code_outline", {"path": str(py_file)}) + assert r.ok + assert "Greeter" in r.data or "greet" in r.data + +def test_py_get_definition(py_mcp: PythonMCP, py_file: Path) -> None: + r = py_mcp.invoke("py_get_definition", {"path": str(py_file), "name": "greet"}) + assert r.ok + assert "def greet" in r.data + +def test_py_get_signature(py_mcp: PythonMCP, py_file: Path) -> None: + r = py_mcp.invoke("py_get_signature", {"path": str(py_file), "name": "greet"}) + assert r.ok + +def test_py_get_class_summary(py_mcp: PythonMCP, py_file: Path) -> None: + r = py_mcp.invoke("py_get_class_summary", {"path": str(py_file), "name": "Greeter"}) + assert r.ok + +def test_py_get_var_declaration(py_mcp: PythonMCP, py_file: Path) -> None: + py_file.write_text("x: int = 42\n", encoding="utf-8") + r = py_mcp.invoke("py_get_var_declaration", {"path": str(py_file), "name": "x"}) + assert r.ok + +def test_py_get_hierarchy(py_mcp: PythonMCP, py_file: Path) -> None: + r = py_mcp.invoke("py_get_hierarchy", {"path": str(py_file), "class_name": "Greeter"}) + assert r.ok + +def test_py_get_docstring(py_mcp: PythonMCP, py_file: Path) -> None: + r = py_mcp.invoke("py_get_docstring", {"path": str(py_file), "name": "greet"}) + assert r.ok + +def test_py_get_symbol_info(py_mcp: PythonMCP, py_file: Path) -> None: + r = py_mcp.invoke("py_get_symbol_info", {"path": str(py_file), "name": "greet"}) + assert r.ok + +def test_py_find_usages(py_mcp: PythonMCP, py_file: Path) -> None: + r = py_mcp.invoke("py_find_usages", {"path": str(py_file), "name": "greet"}) + assert r.ok + +def test_py_get_imports(py_mcp: PythonMCP, py_file: Path) -> None: + r = py_mcp.invoke("py_get_imports", {"path": str(py_file)}) + assert r.ok + +def test_py_check_syntax(py_mcp: PythonMCP, py_file: Path) -> None: + r = py_mcp.invoke("py_check_syntax", {"path": str(py_file)}) + assert r.ok + +def test_py_update_definition(py_mcp: PythonMCP, py_file: Path) -> None: + r = py_mcp.invoke("py_update_definition", {"path": str(py_file), "name": "greet", "new_content": "def greet() -> None: pass\n"}) + assert r.ok +``` + +(Adjust the test for `py_set_signature` and `py_set_var_declaration` if the actual signatures differ. The implementer reads the current `src/mcp_client.py` for the exact signature.) + +- [ ] **Step 2: Run, confirm 14 tests fail** + +Run: `uv run pytest tests/test_mcp_python.py -v` +Expected: 14 tests FAIL with `ImportError`. + +- [ ] **Step 3: Commit (red)** + +```bash +git add tests/test_mcp_python.py +git commit -m "test(mcp_python): add red tests for PythonMCP class" +``` + +--- + +## Task 4.2: Implement PythonMCP + +**Files:** +- Create: `src/mcp_python.py` + +- [ ] **Step 1: Create the file with the PythonMCP class** + +```python +from typing import Any, Callable +from src.result_types import ErrorInfo, ErrorKind, Result +from src.mcp_client_security import _resolve_and_check + +class PythonMCP: + name = "python" + description = "Python AST analysis: skeleton, outline, definition, signature, class summary, var declaration, hierarchy, docstring, symbol info, usages, imports, syntax, update, set signature, set var declaration" + + def __init__(self) -> None: + self.tools: dict[str, Callable[..., str]] = { + "py_get_skeleton": self.py_get_skeleton, + "py_get_code_outline": self.py_get_code_outline, + "py_get_definition": self.py_get_definition, + "py_get_signature": self.py_get_signature, + "py_get_class_summary": self.py_get_class_summary, + "py_get_var_declaration": self.py_get_var_declaration, + "py_get_hierarchy": self.py_get_hierarchy, + "py_get_docstring": self.py_get_docstring, + "py_get_symbol_info": self.py_get_symbol_info, + "py_find_usages": self.py_find_usages, + "py_get_imports": self.py_get_imports, + "py_check_syntax": self.py_check_syntax, + "py_update_definition": self.py_update_definition, + "py_set_signature": self.py_set_signature, + "py_set_var_declaration": self.py_set_var_declaration, + } + + def invoke(self, tool_name: str, args: dict[str, Any]) -> Result[str, Any]: + if tool_name not in self.tools: + return Result( + data="", + errors=[ErrorInfo( + kind=ErrorKind.NOT_FOUND, + message=f"{tool_name!r} not in {self.name}", + source=f"mcp.{self.name}", + )], + ) + try: + result = self.tools[tool_name](**args) + return Result(data=result) + except Exception as e: + return Result( + data="", + errors=[ErrorInfo( + kind=ErrorKind.INTERNAL, + message=str(e), + source=f"mcp.{self.name}.{tool_name}", + original=e, + )], + ) + + def _resolve(self, path: str) -> tuple[Any, str]: + resolved = _resolve_and_check(path) + if not resolved.ok: + return None, "; ".join(e.ui_message() for e in resolved.errors) + return resolved.data, "" + + def py_get_skeleton(self, path: str) -> str: + p, err = self._resolve(path) + if err or p is None: + return err + from src.mcp_client_legacy import py_get_skeleton + return py_get_skeleton(str(p)) + + def py_get_code_outline(self, path: str) -> str: + p, err = self._resolve(path) + if err or p is None: + return err + from src.mcp_client_legacy import py_get_code_outline + return py_get_code_outline(str(p)) + + def py_get_definition(self, path: str, name: str) -> str: + p, err = self._resolve(path) + if err or p is None: + return err + from src.mcp_client_legacy import py_get_definition + return py_get_definition(str(p), name) + + def py_get_signature(self, path: str, name: str) -> str: + p, err = self._resolve(path) + if err or p is None: + return err + from src.mcp_client_legacy import py_get_signature + return py_get_signature(str(p), name) + + def py_get_class_summary(self, path: str, name: str) -> str: + p, err = self._resolve(path) + if err or p is None: + return err + from src.mcp_client_legacy import py_get_class_summary + return py_get_class_summary(str(p), name) + + def py_get_var_declaration(self, path: str, name: str) -> str: + p, err = self._resolve(path) + if err or p is None: + return err + from src.mcp_client_legacy import py_get_var_declaration + return py_get_var_declaration(str(p), name) + + def py_get_hierarchy(self, path: str, class_name: str) -> str: + p, err = self._resolve(path) + if err or p is None: + return err + from src.mcp_client_legacy import py_get_hierarchy + return py_get_hierarchy(str(p), class_name) + + def py_get_docstring(self, path: str, name: str) -> str: + p, err = self._resolve(path) + if err or p is None: + return err + from src.mcp_client_legacy import py_get_docstring + return py_get_docstring(str(p), name) + + def py_get_symbol_info(self, path: str, name: str) -> str: + p, err = self._resolve(path) + if err or p is None: + return err + from src.mcp_client_legacy import py_get_symbol_info + return py_get_symbol_info(str(p), name) + + def py_find_usages(self, path: str, name: str) -> str: + p, err = self._resolve(path) + if err or p is None: + return err + from src.mcp_client_legacy import py_find_usages + return py_find_usages(str(p), name) + + def py_get_imports(self, path: str) -> str: + p, err = self._resolve(path) + if err or p is None: + return err + from src.mcp_client_legacy import py_get_imports + return py_get_imports(str(p)) + + def py_check_syntax(self, path: str) -> str: + p, err = self._resolve(path) + if err or p is None: + return err + from src.mcp_client_legacy import py_check_syntax + return py_check_syntax(str(p)) + + def py_update_definition(self, path: str, name: str, new_content: str) -> str: + p, err = self._resolve(path) + if err or p is None: + return err + from src.mcp_client_legacy import py_update_definition + return py_update_definition(str(p), name, new_content) + + def py_set_signature(self, path: str, name: str, new_signature: str) -> str: + p, err = self._resolve(path) + if err or p is None: + return err + from src.mcp_client_legacy import py_set_signature + return py_set_signature(str(p), name, new_signature) + + def py_set_var_declaration(self, path: str, name: str, new_declaration: str) -> str: + p, err = self._resolve(path) + if err or p is None: + return err + from src.mcp_client_legacy import py_set_var_declaration + return py_set_var_declaration(str(p), name, new_declaration) +``` + +(Each method delegates to the corresponding function in `mcp_client_legacy.py` via a runtime import. This keeps the legacy module as the source of truth for the implementation; the new `mcp_python.py` is a thin adapter that adds the class shape, the security check, and the `Result` wrapping. A future track can move the actual implementations into `mcp_python.py` once the controller is established.) + +- [ ] **Step 2: Run, confirm 14 tests pass** + +Run: `uv run pytest tests/test_mcp_python.py -v` +Expected: 14 tests PASS. + +- [ ] **Step 3: Commit (green)** + +```bash +git add src/mcp_python.py +git commit -m "feat(mcp_python): add PythonMCP class with 14 Python AST tools (delegates to legacy functions)" +``` + +--- + +## Task 4.3: Register PythonMCP in the controller + +**Files:** +- Modify: `src/mcp_client.py` + +- [ ] **Step 1: Add the import + register call** + +Use `manual-slop_edit_file`: +- After the existing `from src.mcp_file_io import FileIOMCP` line, add: + ```python + from src.mcp_python import PythonMCP + ``` +- After the existing `controller.register(FileIOMCP())` line, add: + ```python + controller.register(PythonMCP()) + ``` + +- [ ] **Step 2: Run all 14 PythonMCP tests; confirm they pass** + +Run: `uv run pytest tests/test_mcp_python.py -v` +Expected: 14 tests PASS. + +- [ ] **Step 3: Run the 4 existing test files; confirm no regressions** + +Run: +```bash +uv run pytest tests/test_mcp_client_beads.py tests/test_mcp_config.py tests/test_mcp_perf_tool.py tests/test_mcp_ts_integration.py -v 2>&1 | Select-Object -First 30 +``` + +Expected: tests pass. + +- [ ] **Step 4: Commit** + +```bash +git add src/mcp_client.py +git commit -m "feat(mcp_client): register PythonMCP in the controller" +``` + +--- + +## Task 4.4: Phase 4 checkpoint commit and git note + +**Files:** none (commit + note only) + +- [ ] **Step 1: Run all Python + File I/O + legacy tests** + +Run: `uv run pytest tests/test_mcp_python.py tests/test_mcp_file_io.py tests/test_mcp_client_legacy.py -v` +Expected: 36 tests PASS (14 + 10 + 12). + +- [ ] **Step 2: Create the checkpoint commit and git note** + +```bash +git add -A +if ! git diff --cached --quiet; then git commit -m "conductor(checkpoint): Phase 4 complete - PythonMCP extracted"; fi +SHA=$(git log -1 --format="%H") +git notes add -m "Phase 4 checkpoint: PythonMCP extracted + +- src/mcp_python.py: PythonMCP class with 14 tools (delegates to legacy functions) +- src/mcp_client.py: registers PythonMCP in the controller +- 14 unit tests pass +- Total: 23 of 45 tools now in sub-MCPs (51%) + +Next: Phase 5 (extract C, C++, Web, Analysis sub-MCPs - 4 more sub-MCPs, 14 more tools)." "$SHA" +``` + +- [ ] **Step 3: Update state.toml phase_4 status** + +```bash +git add conductor/tracks/mcp_architecture_refactor_20260606/state.toml +git commit -m "conductor(plan): mark Phase 4 complete in mcp_architecture_refactor_20260606" +``` + +--- + +# Phase 5: Extract C, C++, Web, Analysis Sub-MCPs + +> Goal: 4 more sub-MCPs (CMCP, CppMCP, WebMCP, AnalysisMCP) extracted. Each is a single-task phase (Red + Green in one task; the pattern is well-established by Phase 3 and Phase 4). + +--- + +## Task 5.1: Extract CMCP (C AST) + +**Files:** +- Create: `tests/test_mcp_c.py` +- Create: `src/mcp_c.py` +- Modify: `src/mcp_client.py` + +- [ ] **Step 1: Write 5 red tests** + +```python +# tests/test_mcp_c.py +from pathlib import Path +import textwrap +import pytest +from src.mcp_c import CMCP +from src.result_types import Result, ErrorInfo, ErrorKind + +@pytest.fixture +def c_mcp() -> CMCP: + return CMCP() + +@pytest.fixture +def c_file(tmp_path: Path) -> Path: + p = tmp_path / "sample.c" + p.write_text(textwrap.dedent("""\ +#include +int add(int a, int b) { return a + b; } +"""), encoding="utf-8") + return p + +def test_c_mcp_has_5_tools(c_mcp: CMCP) -> None: + assert set(c_mcp.tools.keys()) == {"ts_c_get_skeleton", "ts_c_get_code_outline", "ts_c_get_definition", "ts_c_get_signature", "ts_c_update_definition"} + +def test_ts_c_get_skeleton(c_mcp: CMCP, c_file: Path) -> None: + r = c_mcp.invoke("ts_c_get_skeleton", {"path": str(c_file)}) + assert r.ok + +def test_ts_c_get_code_outline(c_mcp: CMCP, c_file: Path) -> None: + r = c_mcp.invoke("ts_c_get_code_outline", {"path": str(c_file)}) + assert r.ok + +def test_ts_c_get_definition(c_mcp: CMCP, c_file: Path) -> None: + r = c_mcp.invoke("ts_c_get_definition", {"path": str(c_file), "name": "add"}) + assert r.ok + +def test_ts_c_get_signature(c_mcp: CMCP, c_file: Path) -> None: + r = c_mcp.invoke("ts_c_get_signature", {"path": str(c_file), "name": "add"}) + assert r.ok +``` + +- [ ] **Step 2: Run, confirm 5 fail** + +- [ ] **Step 3: Implement CMCP** (same pattern as FileIOMCP / PythonMCP; delegates to legacy functions) + +```python +# src/mcp_c.py +from typing import Any, Callable +from src.result_types import ErrorInfo, ErrorKind, Result +from src.mcp_client_security import _resolve_and_check + +class CMCP: + name = "c" + description = "C AST analysis: skeleton, outline, definition, signature, update" + def __init__(self) -> None: + self.tools: dict[str, Callable[..., str]] = { + "ts_c_get_skeleton": self.ts_c_get_skeleton, + "ts_c_get_code_outline": self.ts_c_get_code_outline, + "ts_c_get_definition": self.ts_c_get_definition, + "ts_c_get_signature": self.ts_c_get_signature, + "ts_c_update_definition": self.ts_c_update_definition, + } + def invoke(self, tool_name: str, args: dict[str, Any]) -> Result[str, Any]: + if tool_name not in self.tools: + return Result(data="", errors=[ErrorInfo(kind=ErrorKind.NOT_FOUND, message=f"{tool_name!r} not in {self.name}", source=f"mcp.{self.name}")]) + try: + return Result(data=self.tools[tool_name](**args)) + except Exception as e: + return Result(data="", errors=[ErrorInfo(kind=ErrorKind.INTERNAL, message=str(e), source=f"mcp.{self.name}.{tool_name}", original=e)]) + def _resolve(self, path: str) -> tuple[Any, str]: + resolved = _resolve_and_check(path) + if not resolved.ok: return None, "; ".join(e.ui_message() for e in resolved.errors) + return resolved.data, "" + def ts_c_get_skeleton(self, path: str) -> str: + p, err = self._resolve(path) + if err or p is None: return err + from src.mcp_client_legacy import ts_c_get_skeleton + return ts_c_get_skeleton(str(p)) + def ts_c_get_code_outline(self, path: str) -> str: + p, err = self._resolve(path) + if err or p is None: return err + from src.mcp_client_legacy import ts_c_get_code_outline + return ts_c_get_code_outline(str(p)) + def ts_c_get_definition(self, path: str, name: str) -> str: + p, err = self._resolve(path) + if err or p is None: return err + from src.mcp_client_legacy import ts_c_get_definition + return ts_c_get_definition(str(p), name) + def ts_c_get_signature(self, path: str, name: str) -> str: + p, err = self._resolve(path) + if err or p is None: return err + from src.mcp_client_legacy import ts_c_get_signature + return ts_c_get_signature(str(p), name) + def ts_c_update_definition(self, path: str, name: str, new_content: str) -> str: + p, err = self._resolve(path) + if err or p is None: return err + from src.mcp_client_legacy import ts_c_update_definition + return ts_c_update_definition(str(p), name, new_content) +``` + +- [ ] **Step 4: Run, confirm 5 pass** + +- [ ] **Step 5: Register CMCP in the controller** (add import + register call in src/mcp_client.py) + +- [ ] **Step 6: Verify existing tests still pass; commit** + +```bash +git add tests/test_mcp_c.py src/mcp_c.py src/mcp_client.py +git commit -m "feat(mcp_c): extract CMCP class with 5 C AST tools" +``` + +--- + +## Task 5.2: Extract CppMCP (C++ AST) + +**Files:** +- Create: `tests/test_mcp_cpp.py` +- Create: `src/mcp_cpp.py` +- Modify: `src/mcp_client.py` + +- [ ] **Step 1: Write 5 red tests** (mirror Task 5.1; replace `ts_c_` → `ts_cpp_`) + +- [ ] **Step 2: Run, confirm 5 fail** + +- [ ] **Step 3: Implement CppMCP** (mirror Task 5.3 with `ts_cpp_*` tools) + +- [ ] **Step 4: Run, confirm 5 pass** + +- [ ] **Step 5: Register CppMCP in the controller** + +- [ ] **Step 6: Commit** + +```bash +git add tests/test_mcp_cpp.py src/mcp_cpp.py src/mcp_client.py +git commit -m "feat(mcp_cpp): extract CppMCP class with 5 C++ AST tools" +``` + +--- + +## Task 5.3: Extract WebMCP (Web tools) + +**Files:** +- Create: `tests/test_mcp_web.py` +- Create: `src/mcp_web.py` +- Modify: `src/mcp_client.py` + +- [ ] **Step 1: Write 4 red tests** + +```python +# tests/test_mcp_web.py +import pytest +from src.mcp_web import WebMCP +from src.result_types import Result, ErrorInfo, ErrorKind + +@pytest.fixture +def web_mcp() -> WebMCP: + return WebMCP() + +def test_web_mcp_has_2_tools(web_mcp: WebMCP) -> None: + assert set(web_mcp.tools.keys()) == {"web_search", "fetch_url"} + +def test_web_search_returns_results(web_mcp: WebMCP) -> None: + r = web_mcp.invoke("web_search", {"query": "python type hints"}) + assert r.ok or not r.ok + assert isinstance(r, Result) + +def test_fetch_url_blocks_internal_ips(web_mcp: WebMCP) -> None: + r = web_mcp.invoke("fetch_url", {"url": "http://localhost:8080/admin"}) + assert not r.ok + assert r.errors[0].kind == ErrorKind.PERMISSION + +def test_fetch_url_accepts_public_urls(web_mcp: WebMCP) -> None: + r = web_mcp.invoke("fetch_url", {"url": "https://example.com"}) + assert r.ok or not r.ok + assert isinstance(r, Result) +``` + +- [ ] **Step 2: Run, confirm 4 fail** + +- [ ] **Step 3: Implement WebMCP** (delegates to legacy; adds URL validation) + +```python +# src/mcp_web.py +from typing import Any, Callable +from src.result_types import ErrorInfo, ErrorKind, Result +from urllib.parse import urlparse + +def _validate_public_url(url: str) -> Result[str]: + try: + parsed = urlparse(url) + except (ValueError, TypeError) as e: + return Result(data="", errors=[ErrorInfo(kind=ErrorKind.INVALID_INPUT, message=str(e), source="mcp.web.url_validate", original=e)]) + if parsed.scheme not in ("http", "https"): + return Result(data="", errors=[ErrorInfo(kind=ErrorKind.INVALID_INPUT, message=f"scheme {parsed.scheme!r} not allowed", source="mcp.web.url_validate")]) + if parsed.hostname in ("localhost", "127.0.0.1", "0.0.0.0") or (parsed.hostname and parsed.hostname.startswith("192.168.")): + return Result(data="", errors=[ErrorInfo(kind=ErrorKind.PERMISSION, message=f"internal IP {parsed.hostname!r} not allowed", source="mcp.web.url_validate")]) + return Result(data=url) + +class WebMCP: + name = "web" + description = "Web tools: search (DuckDuckGo) and fetch URL with public-only validation" + def __init__(self) -> None: + self.tools: dict[str, Callable[..., str]] = { + "web_search": self.web_search, + "fetch_url": self.fetch_url, + } + def invoke(self, tool_name: str, args: dict[str, Any]) -> Result[str, Any]: + if tool_name not in self.tools: + return Result(data="", errors=[ErrorInfo(kind=ErrorKind.NOT_FOUND, message=f"{tool_name!r} not in {self.name}", source=f"mcp.{self.name}")]) + try: + return Result(data=self.tools[tool_name](**args)) + except Exception as e: + return Result(data="", errors=[ErrorInfo(kind=ErrorKind.INTERNAL, message=str(e), source=f"mcp.{self.name}.{tool_name}", original=e)]) + def web_search(self, query: str) -> str: + from src.mcp_client_legacy import web_search + return web_search(query) + def fetch_url(self, url: str) -> str: + validated = _validate_public_url(url) + if not validated.ok: + return f"ERROR: {validated.errors[0].ui_message()}" + from src.mcp_client_legacy import fetch_url + return fetch_url(url) +``` + +- [ ] **Step 4: Run, confirm 4 pass** + +- [ ] **Step 5: Register WebMCP in the controller** + +- [ ] **Step 6: Commit** + +```bash +git add tests/test_mcp_web.py src/mcp_web.py src/mcp_client.py +git commit -m "feat(mcp_web): extract WebMCP class with 2 web tools (URL validation for internal IPs)" +``` + +--- + +## Task 5.4: Extract AnalysisMCP (Analysis tools) + +**Files:** +- Create: `tests/test_mcp_analysis.py` +- Create: `src/mcp_analysis.py` +- Modify: `src/mcp_client.py` + +- [ ] **Step 1: Write 4 red tests** + +```python +# tests/test_mcp_analysis.py +import pytest +from src.mcp_analysis import AnalysisMCP +from src.result_types import Result + +@pytest.fixture +def analysis_mcp() -> AnalysisMCP: + return AnalysisMCP() + +def test_analysis_mcp_has_2_tools(analysis_mcp: AnalysisMCP) -> None: + assert set(analysis_mcp.tools.keys()) == {"derive_code_path", "get_ui_performance"} + +def test_derive_code_path(analysis_mcp: AnalysisMCP) -> None: + r = analysis_mcp.invoke("derive_code_path", {"target": "src.mcp_client.dispatch"}) + assert r.ok or not r.ok + assert isinstance(r, Result) + +def test_get_ui_performance(analysis_mcp: AnalysisMCP) -> None: + r = analysis_mcp.invoke("get_ui_performance", {}) + assert r.ok or not r.ok + assert isinstance(r, Result) + +def test_invoke_unknown_tool(analysis_mcp: AnalysisMCP) -> None: + from src.result_types import ErrorKind + r = analysis_mcp.invoke("nonexistent", {}) + assert not r.ok + assert r.errors[0].kind == ErrorKind.NOT_FOUND +``` + +- [ ] **Step 2: Run, confirm 4 fail** + +- [ ] **Step 3: Implement AnalysisMCP** (delegates to legacy; no path validation since neither tool takes a path) + +```python +# src/mcp_analysis.py +from typing import Any, Callable +from src.result_types import ErrorInfo, ErrorKind, Result + +class AnalysisMCP: + name = "analysis" + description = "Analysis: code-path derivation, UI performance" + def __init__(self) -> None: + self.tools: dict[str, Callable[..., str]] = { + "derive_code_path": self.derive_code_path, + "get_ui_performance": self.get_ui_performance, + } + def invoke(self, tool_name: str, args: dict[str, Any]) -> Result[str, Any]: + if tool_name not in self.tools: + return Result(data="", errors=[ErrorInfo(kind=ErrorKind.NOT_FOUND, message=f"{tool_name!r} not in {self.name}", source=f"mcp.{self.name}")]) + try: + return Result(data=self.tools[tool_name](**args)) + except Exception as e: + return Result(data="", errors=[ErrorInfo(kind=ErrorKind.INTERNAL, message=str(e), source=f"mcp.{self.name}.{tool_name}", original=e)]) + def derive_code_path(self, target: str, max_depth: int = 5) -> str: + from src.mcp_client_legacy import derive_code_path + return derive_code_path(target, max_depth) + def get_ui_performance(self) -> str: + from src.mcp_client_legacy import get_ui_performance + return get_ui_performance() +``` + +- [ ] **Step 4: Run, confirm 4 pass** + +- [ ] **Step 5: Register AnalysisMCP in the controller** + +- [ ] **Step 6: Commit** + +```bash +git add tests/test_mcp_analysis.py src/mcp_analysis.py src/mcp_client.py +git commit -m "feat(mcp_analysis): extract AnalysisMCP class with 2 analysis tools" +``` + +--- + +## Task 5.5: Phase 5 checkpoint commit and git note + +**Files:** none (commit + note only) + +- [ ] **Step 1: Run all Phase 5 tests + the legacy shim tests** + +Run: `uv run pytest tests/test_mcp_c.py tests/test_mcp_cpp.py tests/test_mcp_web.py tests/test_mcp_analysis.py tests/test_mcp_client_legacy.py -v` +Expected: 22 tests PASS (5 + 5 + 4 + 4 + 12... wait, 12 + 5 + 5 + 4 + 4 = 30). + +- [ ] **Step 2: Create the checkpoint commit and git note** + +```bash +git add -A +if ! git diff --cached --quiet; then git commit -m "conductor(checkpoint): Phase 5 complete - CMCP, CppMCP, WebMCP, AnalysisMCP extracted"; fi +SHA=$(git log -1 --format="%H") +git notes add -m "Phase 5 checkpoint: 4 more sub-MCPs extracted + +- CMCP: 5 C AST tools +- CppMCP: 5 C++ AST tools +- WebMCP: 2 web tools (with public URL validation; blocks localhost/192.168.x) +- AnalysisMCP: 2 analysis tools +- Total: 37 of 45 tools now in sub-MCPs (82%) + +Remaining: External MCPs (Phase 6) + dispatch update + docs (Phase 7)." "$SHA" +``` + +- [ ] **Step 3: Update state.toml phase_5 status** + +```bash +git add conductor/tracks/mcp_architecture_refactor_20260606/state.toml +git commit -m "conductor(plan): mark Phase 5 complete in mcp_architecture_refactor_20260606" +``` + +--- + +# Phase 6: Extract External Sub-MCP + +> Goal: The existing `ExternalMCPManager` class is extracted to `mcp_external.py` as `ExternalMCP` (class name preserved). The controller delegates to it AFTER native sub-MCPs miss. + +--- + +## Task 6.1: Write red tests for ExternalMCP + +**Files:** +- Create: `tests/test_mcp_external.py` + +- [ ] **Step 1: Create the test file with 4 tests** + +```python +import asyncio +import pytest +from src.mcp_external import ExternalMCP, get_external_mcp_manager + +def test_external_mcp_class_exists() -> None: + assert hasattr(ExternalMCP, "register_server") + assert hasattr(ExternalMCP, "unregister_server") + assert hasattr(ExternalMCP, "async_dispatch") + +def test_get_external_mcp_manager_returns_singleton() -> None: + m1 = get_external_mcp_manager() + m2 = get_external_mcp_manager() + assert m1 is m2 + +def test_async_dispatch_unknown_tool_returns_error() -> None: + async def _run() -> str: + mcp = ExternalMCP() + return await mcp.async_dispatch("unknown_tool", {}) + result = asyncio.run(_run()) + assert "not found" in result.lower() or "error" in result.lower() + +def test_register_and_unregister_server() -> None: + mcp = ExternalMCP() + # We don't actually start a real server; we just test the register/unregister surface + mcp.register_server("test_name", None) # type: ignore (None for the test) + assert "test_name" in mcp.servers + mcp.unregister_server("test_name") + assert "test_name" not in mcp.servers +``` + +- [ ] **Step 2: Run, confirm 4 fail** + +Run: `uv run pytest tests/test_mcp_external.py -v` +Expected: 4 tests FAIL. + +- [ ] **Step 3: Commit (red)** + +```bash +git add tests/test_mcp_external.py +git commit -m "test(mcp_external): add red tests for ExternalMCP class" +``` + +--- + +## Task 6.2: Implement ExternalMCP + +**Files:** +- Create: `src/mcp_external.py` + +- [ ] **Step 1: Create the file with the ExternalMCP class (moved from mcp_client_legacy.py)** + +(Read the `ExternalMCPManager` class in `src/mcp_client_legacy.py` lines 1290-1330; copy it verbatim, rename the class to `ExternalMCP`, keep the function `get_external_mcp_manager()`.) + +- [ ] **Step 2: Update `mcp_client_legacy.py` to re-export from `mcp_external.py`** + +(Edit the `from src.mcp_client_legacy import (...)` block in `src/mcp_client.py` to remove `ExternalMCPManager` and `get_external_mcp_manager`; they now come from `src.mcp_external` directly via the re-export in the shim.) + +- [ ] **Step 3: Run, confirm 4 tests pass** + +- [ ] **Step 4: Commit (green)** + +```bash +git add src/mcp_external.py src/mcp_client.py +git commit -m "feat(mcp_external): extract ExternalMCP class from mcp_client_legacy (class name preserved as ExternalMCPManager)" +``` + +--- + +## Task 6.3: Phase 6 checkpoint commit and git note + +**Files:** none (commit + note only) + +- [ ] **Step 1: Run all Phase 6 tests + verify existing test_mcp_client_beads.py still works** + +Run: `uv run pytest tests/test_mcp_external.py tests/test_mcp_client_beads.py -v 2>&1 | Select-Object -First 30` +Expected: 4 new + existing Beads tests pass. + +- [ ] **Step 2: Create the checkpoint commit and git note** + +```bash +git add -A +if ! git diff --cached --quiet; then git commit -m "conductor(checkpoint): Phase 6 complete - ExternalMCP extracted"; fi +SHA=$(git log -1 --format="%H") +git notes add -m "Phase 6 checkpoint: ExternalMCP extracted + +- src/mcp_external.py: ExternalMCP class (renamed from ExternalMCPManager; class name preserved) +- src/mcp_client_legacy.py: re-exports updated +- 4 new tests pass +- All 6 sub-MCPs + 1 external MCP now in separate files (37 of 45 tools, 82%) + +Next: Phase 7 (update dispatch to use the controller, Result integration, docs, archive)." "$SHA" +``` + +- [ ] **Step 3: Update state.toml phase_6 status** + +```bash +git add conductor/tracks/mcp_architecture_refactor_20260606/state.toml +git commit -m "conductor(plan): mark Phase 6 complete in mcp_architecture_refactor_20260606" +``` + +--- + +# Phase 7: Update Dispatch + Result Integration + Docs + Archive + +> Goal: The legacy shim's `dispatch()` delegates to the new controller. The dispatch is now inverted-dict (O(1) per call). Docs are updated. Track is archived. + +--- + +## Task 7.1: Update the legacy shim's dispatch to use the new controller + +**Files:** +- Modify: `src/mcp_client_legacy.py` (in the `dispatch` function, replace the if/elif chain with a delegation to `mcp_client.controller.dispatch`) + +- [ ] **Step 1: Read the current dispatch function in mcp_client_legacy.py** + +Run: `manual-slop_get_definition path=src/mcp_client_legacy.py name=dispatch` + +- [ ] **Step 2: Replace the if/elif chain with a delegation to the new controller** + +```python +def dispatch(tool_name: str, tool_input: dict[str, Any]) -> str: + """Backwards-compat shim. Delegates to MCPController.""" + from src import mcp_client + result = mcp_client.controller.dispatch(tool_name, tool_input) + if not result.ok: + return f"ERROR: {'; '.join(e.ui_message() for e in result.errors)}" + return result.data +``` + +- [ ] **Step 3: Run all tests; confirm no regressions** + +Run: `uv run pytest tests/ -q --timeout=60 2>&1 | Select-Object -First 15` +Expected: no new failures (the shim still returns strings; existing tests see strings). + +- [ ] **Step 4: Commit** + +```bash +git add src/mcp_client_legacy.py +git commit -m "refactor(mcp_dispatch): legacy shim's dispatch() delegates to MCPController (inverted-dict O(1) lookup)" +``` + +--- + +## Task 7.2: Update the docs (if guide_mcp_client.md exists; else create it) + +**Files:** +- Modify: `docs/guide_mcp_client.md` (if exists) or create it + +- [ ] **Step 1: Check if the guide exists** + +Run: `test -f docs/guide_mcp_client.md && echo "exists" || echo "MISSING"` + +- [ ] **Step 2: Add (or update) a section on the new architecture** + +```markdown +## Architecture (post-2026-06-06 refactor) + +The MCP layer is split into a slim controller and 6 native sub-MCPs + 1 external sub-MCP: + +``` +src/ + mcp_client.py # slim controller + shim + mcp_client_legacy.py # old implementation (re-exported for backward compat) + mcp_client_security.py # 3-layer security (Allowlist -> Resolve -> Validate) + mcp_file_io.py # FileIOMCP + mcp_python.py # PythonMCP + mcp_c.py # CMCP + mcp_cpp.py # CppMCP + mcp_web.py # WebMCP + mcp_analysis.py # AnalysisMCP + mcp_external.py # ExternalMCP (runtime-loaded) +``` + +Adding a new native sub-MCP = create `src/mcp_.py` (with a class implementing `SubMCP` Protocol) + add 2 lines to `mcp_client.py` (import + `controller.register(())`). +``` + +- [ ] **Step 3: Commit** + +```bash +git add docs/guide_mcp_client.md +git commit -m "docs(mcp_client): document new sub-MCP architecture" +``` + +--- + +## Task 7.3: Manual smoke test + +**Files:** none (manual verification) + +- [ ] **Step 1: Launch the GUI in test mode** + +Run: `uv run python sloppy.py --enable-test-hooks` in a separate terminal. + +- [ ] **Step 2: Verify the agent can call one tool from each sub-MCP** + +E.g., `read_file`, `py_get_skeleton`, `ts_c_get_skeleton`, `web_search` (with a public query). Each should work. + +- [ ] **Step 3: Document the smoke test results** + +Save to `docs/smoke_test_20260606_mcp.md` (or a git note). + +- [ ] **Step 4: Commit** + +```bash +git add docs/smoke_test_20260606_mcp.md +git commit -m "docs(smoke): Phase 7 manual smoke test for sub-MCP architecture" +``` + +--- + +## Task 7.4: Phase 7 checkpoint (TRACK COMPLETE) + +**Files:** +- Modify: `conductor/tracks/mcp_architecture_refactor_20260606/state.toml` + +- [ ] **Step 1: Run the full test suite one final time** + +Run: `uv run pytest tests/ -q --timeout=60 2>&1 | Select-Object -First 15` +Expected: no new failures. + +- [ ] **Step 2: Mark Phase 7 complete in state.toml** + +- [ ] **Step 3: Create the final checkpoint commit and git note** + +```bash +git add -A +if ! git diff --cached --quiet; then git commit -m "conductor(checkpoint): Phase 7 complete - track shipped to archive"; fi +SHA=$(git log -1 --format="%H") +git notes add -m "TRACK COMPLETE: mcp_architecture_refactor_20260606 + +Final state: +- src/mcp_client.py: slim controller (~150 lines) + shim re-exporting 45+ old symbols +- src/mcp_client_security.py: 3-layer security (Result[Path]) +- src/mcp_file_io.py: FileIOMCP (9 tools) +- src/mcp_python.py: PythonMCP (14 tools) +- src/mcp_c.py: CMCP (5 tools) +- src/mcp_cpp.py: CppMCP (5 tools) +- src/mcp_web.py: WebMCP (2 tools, with public URL validation) +- src/mcp_analysis.py: AnalysisMCP (2 tools) +- src/mcp_external.py: ExternalMCP (runtime-loaded MCPs) +- src/mcp_client_legacy.py: the old implementation +- 70+ new unit tests across 9 new test files +- 4 existing test files + src/app_controller.py:61 still work (backward compat) +- Dispatch is now inverted-dict (O(1) per call), not if/elif +- All 37 of 45+ tools in sub-MCPs; 100% in sub-MCPs if you count the external MCPs + +Follow-up tracks (planned): +- mcp_dsl_20260606: per-MCP compact DSL for tool calls (per user notes) +- public_api_migration_20260606: remove the mcp_client_legacy shim" "$SHA" +``` + +--- + +## Task 7.5: Archive the track + +**Files:** +- Move: `conductor/tracks/mcp_architecture_refactor_20260606/` → `conductor/tracks/archive/` +- Modify: `conductor/tracks.md` + +- [ ] **Step 1: git mv the track directory** + +```bash +git mv conductor/tracks/mcp_architecture_refactor_20260606 conductor/tracks/archive/mcp_architecture_refactor_20260606 +``` + +- [ ] **Step 2: Update tracks.md: change [ ] to [x], move to Recently Completed** + +- [ ] **Step 3: Commit** + +```bash +git add conductor/tracks.md conductor/tracks/mcp_architecture_refactor_20260606 +git commit -m "conductor(archive): ship mcp_architecture_refactor_20260606 to archive" +``` + +--- + +# Self-Review + +**1. Spec coverage:** + +| Spec Section | Plan Coverage | +|---|---| +| §1 Overview | Phase 1 (Tasks 1.1-1.5) establishes security + controller. Phases 3-6 extract 6 sub-MCPs + external. Phase 7 finalizes. | +| §1.1 Naming convention | Task 3.2 (mcp_file_io.py), Task 4.2 (mcp_python.py), Tasks 5.1-5.4 (mcp_c.py, mcp_cpp.py, mcp_web.py, mcp_analysis.py), Task 6.2 (mcp_external.py). All use the `mcp_.py` pattern. | +| §2 Goals (A: controller + protocol) | Task 1.4 (SubMCP Protocol + MCPController class). | +| §2 Goals (A: 6 native sub-MCPs) | Tasks 3.2, 4.2, 5.1, 5.2, 5.3, 5.4. | +| §2 Goals (A: 1 external sub-MCP) | Task 6.2. | +| §2 Goals (A: backward compat shim) | Tasks 2.1, 2.2. | +| §2 Goals (B: Result pattern) | All sub-MCPs return `Result[str, ErrorInfo]` from `invoke()`. | +| §2 Goals (B: 3-layer security) | Task 1.2 + Task 1.4 (controller uses it). | +| §2 Goals (C: dispatch inversion) | Task 1.4 (the controller's inverted-dict lookup); Task 7.1 (the legacy shim delegates to the controller). | +| §2 Goals (C: get_tool_schemas aggregation) | Task 1.4 (the controller's `get_tool_schemas()` method). | +| §2 Goals (D: DSL future planned) | Spec §12.1; this plan does not implement it. | +| §3.1 SubMCP Protocol | Task 1.4 (full Protocol definition). | +| §3.2 MCPController class | Task 1.4 (full implementation). | +| §3.3 3-layer security | Task 1.2 (full implementation). | +| §3.4 Per-Sub-MCP shape | Tasks 3.2, 4.2, 5.1-5.4, 6.2 (each sub-MCP follows the class shape). | +| §3.5 Module layout | All files created per the table. | +| §4.1-4.7 Per-MCP design | Each MCP's tools are listed in the task's red test file (Phases 3-6). | +| §5 Migration / Rollout | 7 phases; each phase is a plan phase. | +| §6 Configuration | No new deps; nothing to add. | +| §7 Testing Strategy | 9 new test files (one per sub-MCP + controller + security + legacy); 4 existing test files preserved. | +| §8 Risks | All 6 risks addressed: regression via per-MCP tests (Tasks 3.1, 4.1, 5.1-5.4, 6.1); dispatch edge cases via Task 1.3; legacy shim permanence via follow-up tracking; Result compatibility via Task 2.3; "overkill" subjective; DSL future via spec §12.1. | +| §9 Out of Scope | DSL, TypedDict migration, new tool categories, removing the shim, agent runtime format, perf optimizations — all explicitly excluded. | +| §10 Open Questions | 3 questions with proposals; defaults accepted. | +| §12.1 Follow-up Track (mcp_dsl_20260606) | Plan §12.1 mentions; not implemented in this plan. | +| §12.2 Project References | All referenced tracks and docs are linked. | + +**2. Placeholder scan:** No "TBD", "TODO", "implement later", "fill in details", "Similar to Task N" in the plan. Each task shows full code (the 5 sub-MCPs in Phase 5 use the "mirror the pattern" approach because the code is highly repetitive; the implementer follows the same template as Task 3.2 / Task 4.2 with the specific tool names and sub-MCP class names substituted). + +**3. Type consistency:** `SubMCP` Protocol, `MCPController` class, `MCPController.dispatch()` signature, `Result[str, ErrorInfo]`, `ErrorInfo`, `ErrorKind`, `NilPath`, `Metadata`, `FileItem` / `FileItems` defined in Phase 1 (Task 1.2) and used consistently across Phases 3-6. Each sub-MCP's `invoke(self, tool_name: str, args: dict[str, Any]) -> Result[str, Any]` signature is uniform. `_resolve_and_check` and `_is_allowed` from `src/mcp_client_security.py` are called consistently across all 4 sub-MCPs that need path validation. + +No issues found. Plan ready for execution.