From a7160f7eab87813f065ab398692231398c221b87 Mon Sep 17 00:00:00 2001 From: Tejas Chopra Date: Sat, 9 May 2026 22:57:35 -0700 Subject: [PATCH] fix: register serena mcp during wrap --- headroom/cli/wrap.py | 78 +++++++++++++ headroom/mcp_registry/__init__.py | 2 + headroom/mcp_registry/codex.py | 36 ++++-- headroom/mcp_registry/install.py | 17 +++ headroom/mcp_registry/ledger.py | 106 ++++++++++++++++++ headroom/proxy/handlers/anthropic.py | 19 +++- tests/test_cli/test_unwrap_claude.py | 82 ++++++++++++++ tests/test_cli/test_wrap_codex.py | 42 +++++++ .../test_mcp_registry/test_codex_registrar.py | 46 ++++++++ tests/test_mcp_registry/test_install.py | 17 +++ tests/test_mcp_registry/test_ledger.py | 53 +++++++++ 11 files changed, 486 insertions(+), 12 deletions(-) create mode 100644 headroom/mcp_registry/ledger.py create mode 100644 tests/test_mcp_registry/test_ledger.py diff --git a/headroom/cli/wrap.py b/headroom/cli/wrap.py index 93fec93a7..4be5c2976 100644 --- a/headroom/cli/wrap.py +++ b/headroom/cli/wrap.py @@ -347,6 +347,53 @@ def _setup_headroom_mcp( click.echo(line) +def _setup_serena_mcp( + registrar: Any, *, context: str, verbose: bool = False, force: bool = False +) -> None: + """Register Serena MCP with the given agent (idempotent).""" + from headroom.mcp_registry import build_serena_spec, format_result + from headroom.mcp_registry.base import RegisterStatus + from headroom.mcp_registry.ledger import record_install + + if not registrar.detect(): + if verbose: + click.echo(f" Serena MCP: {registrar.display_name} not detected — skipping") + return + + if shutil.which("uvx") is None: + click.echo(" Serena MCP: uvx not found — install uv/uvx to enable Serena; skipping") + return + + spec = build_serena_spec(context) + result = registrar.register_server(spec, force=force) + if result.status == RegisterStatus.REGISTERED: + record_install(registrar.name, spec) + + line = format_result( + registrar.name, + result, + label="Serena MCP", + verbose=verbose, + overwrite_hint="update or remove the existing serena MCP entry, then rerun headroom wrap", + restart_hint=f"restart {registrar.display_name} if it was already running", + ) + if line is not None: + click.echo(line) + + +def _remove_headroom_installed_serena_mcp(registrar: Any) -> str: + """Remove Serena MCP only if the ledger proves Headroom installed it.""" + from headroom.mcp_registry.ledger import clear_install, headroom_installed_matching + + current = registrar.get_server("serena") + if not headroom_installed_matching(registrar.name, current): + return "not_headroom_owned" + if registrar.unregister_server("serena"): + clear_install(registrar.name, "serena") + return "removed" + return "failed" + + _CBM_MCP_SERVER_NAME = "codebase-memory-mcp" @@ -568,6 +615,12 @@ def _strip_codex_headroom_blocks(content: str, *, remove_mcp: bool = False) -> s if remove_mcp: # Remove Headroom-managed MCP blocks written by `wrap codex`. content = _remove_marker_span(content, _CODEX_MCP_MARKER, _CODEX_MCP_END) + content = re.sub( + r"(?ms)^# --- Headroom MCP server: [^\n]+ ---\n.*?" + r"^# --- end Headroom MCP server: [^\n]+ ---\n?", + "", + content, + ) content = _remove_marker_span(content, _MEMORY_MCP_MARKER, _MEMORY_MCP_END) # Strip any leftover top-level keys that older (or crashed) versions of @@ -1663,6 +1716,7 @@ def unwrap() -> None: is_flag=True, help="Skip headroom MCP server registration (compression markers will be unactionable)", ) +@click.option("--no-serena", is_flag=True, help="Skip Serena MCP server registration") @click.option( "--code-graph", is_flag=True, @@ -1680,6 +1734,7 @@ def claude( port: int, no_rtk: bool, no_mcp: bool, + no_serena: bool, code_graph: bool, no_proxy: bool, learn: bool, @@ -1703,6 +1758,7 @@ def claude( headroom wrap claude --code-graph # With code graph intelligence headroom wrap claude --no-rtk # Skip rtk (proxy only) headroom wrap claude --no-mcp # Skip MCP retrieve tool registration + headroom wrap claude --no-serena # Skip Serena MCP registration """ if prepare_only: if not no_rtk: @@ -1787,6 +1843,13 @@ def claude( elif verbose: click.echo(" Skipping MCP retrieve tool (--no-mcp)") + if not no_serena: + from headroom.mcp_registry import ClaudeRegistrar + + _setup_serena_mcp(ClaudeRegistrar(), context="claude-code", verbose=verbose) + elif verbose: + click.echo(" Skipping Serena MCP (--no-serena)") + if code_graph: _setup_code_graph(verbose=verbose) @@ -1843,12 +1906,17 @@ def unwrap_claude( if registrar.detect(): removed_headroom = registrar.unregister_server("headroom") removed_code_graph = registrar.unregister_server(_CBM_MCP_SERVER_NAME) + serena_status = _remove_headroom_installed_serena_mcp(registrar) if removed_headroom: click.echo(" Removed Headroom MCP retrieve tool from Claude.") else: click.echo(" Headroom MCP retrieve tool was not registered in Claude.") if removed_code_graph: click.echo(" Removed code graph MCP server from Claude.") + if serena_status == "removed": + click.echo(" Removed Headroom-installed Serena MCP server from Claude.") + elif serena_status == "failed": + click.echo(" Serena MCP server matched Headroom ledger but could not be removed.") else: click.echo(" Claude Code not detected; skipped MCP cleanup.") else: @@ -2056,6 +2124,7 @@ def copilot( is_flag=True, help="Skip headroom MCP server registration (compression markers will be unactionable)", ) +@click.option("--no-serena", is_flag=True, help="Skip Serena MCP server registration") @click.option( "--code-graph", is_flag=True, @@ -2086,6 +2155,7 @@ def codex( port: int, no_rtk: bool, no_mcp: bool, + no_serena: bool, code_graph: bool, no_proxy: bool, learn: bool, @@ -2112,6 +2182,7 @@ def codex( headroom wrap codex -- "fix the bug" # Pass prompt to codex headroom wrap codex --no-rtk # Skip rtk setup headroom wrap codex --no-mcp # Skip MCP retrieve tool registration + headroom wrap codex --no-serena # Skip Serena MCP registration headroom wrap codex --port 9999 # Custom proxy port headroom wrap codex --backend anyllm --anyllm-provider groq """ @@ -2149,6 +2220,13 @@ def codex( elif verbose: click.echo(" Skipping MCP retrieve tool (--no-mcp)") + if not no_serena: + from headroom.mcp_registry import CodexRegistrar + + _setup_serena_mcp(CodexRegistrar(), context="codex", verbose=verbose, force=True) + elif verbose: + click.echo(" Skipping Serena MCP (--no-serena)") + # Setup memory MCP server for Codex (native tool integration) if memory: click.echo(" Setting up memory for Codex...") diff --git a/headroom/mcp_registry/__init__.py b/headroom/mcp_registry/__init__.py index e65508777..61c9eb163 100644 --- a/headroom/mcp_registry/__init__.py +++ b/headroom/mcp_registry/__init__.py @@ -20,6 +20,7 @@ from .display import any_succeeded, format_result, format_results from .install import ( DEFAULT_PROXY_URL, build_headroom_spec, + build_serena_spec, get_all_registrars, install_everywhere, ) @@ -34,6 +35,7 @@ __all__ = [ "ServerSpec", "any_succeeded", "build_headroom_spec", + "build_serena_spec", "format_result", "format_results", "get_all_registrars", diff --git a/headroom/mcp_registry/codex.py b/headroom/mcp_registry/codex.py index 5dcd4771e..8b6d3722e 100644 --- a/headroom/mcp_registry/codex.py +++ b/headroom/mcp_registry/codex.py @@ -28,6 +28,18 @@ _MARKER_START = "# --- Headroom MCP server ---" _MARKER_END = "# --- end Headroom MCP server ---" +def _marker_start(server_name: str) -> str: + if server_name == "headroom": + return _MARKER_START + return f"# --- Headroom MCP server: {server_name} ---" + + +def _marker_end(server_name: str) -> str: + if server_name == "headroom": + return _MARKER_END + return f"# --- end Headroom MCP server: {server_name} ---" + + class CodexRegistrar(MCPRegistrar): """Register MCP servers with the OpenAI Codex CLI.""" @@ -64,7 +76,7 @@ class CodexRegistrar(MCPRegistrar): if existing is not None and not force: content = self._read_text() - if _MARKER_START not in content: + if _marker_start(spec.name) not in content: # Entry exists but wasn't written by us — refuse to clobber. return RegisterResult( RegisterStatus.MISMATCH, @@ -76,7 +88,7 @@ class CodexRegistrar(MCPRegistrar): if existing is not None and force: content = self._read_text() - if _MARKER_START not in content: + if _marker_start(spec.name) not in content: # Even force=True is only allowed to replace blocks that # Headroom owns. Otherwise appending our table would create a # duplicate [mcp_servers.] TOML section and may clobber a @@ -98,11 +110,13 @@ class CodexRegistrar(MCPRegistrar): if not self._config_file.exists(): return False content = self._read_text() - if _MARKER_START not in content or _MARKER_END not in content: + marker_start = _marker_start(server_name) + marker_end = _marker_end(server_name) + if marker_start not in content or marker_end not in content: return False try: - start = content.index(_MARKER_START) - end = content.index(_MARKER_END) + len(_MARKER_END) + start = content.index(marker_start) + end = content.index(marker_end) + len(marker_end) except ValueError: return False before = content[:start].rstrip("\n") @@ -142,9 +156,11 @@ class CodexRegistrar(MCPRegistrar): try: self._codex_dir.mkdir(parents=True, exist_ok=True) content = self._read_text() - if _MARKER_START in content and _MARKER_END in content: - start = content.index(_MARKER_START) - end = content.index(_MARKER_END) + len(_MARKER_END) + marker_start = _marker_start(spec.name) + marker_end = _marker_end(spec.name) + if marker_start in content and marker_end in content: + start = content.index(marker_start) + end = content.index(marker_end) + len(marker_end) content = ( content[:start].rstrip("\n") + ("\n\n" if content[:start].rstrip("\n") else "") @@ -172,7 +188,7 @@ class CodexRegistrar(MCPRegistrar): def _render_block(spec: ServerSpec) -> str: """Render a Headroom-marked TOML block for ``spec``.""" lines: list[str] = [ - _MARKER_START, + _marker_start(spec.name), f"[mcp_servers.{spec.name}]", f"command = {_toml_str(spec.command)}", ] @@ -184,7 +200,7 @@ def _render_block(spec: ServerSpec) -> str: lines.append(f"[mcp_servers.{spec.name}.env]") for k, v in spec.env.items(): lines.append(f"{k} = {_toml_str(v)}") - lines.append(_MARKER_END) + lines.append(_marker_end(spec.name)) return "\n".join(lines) diff --git a/headroom/mcp_registry/install.py b/headroom/mcp_registry/install.py index f63c0e365..dbdb6628c 100644 --- a/headroom/mcp_registry/install.py +++ b/headroom/mcp_registry/install.py @@ -37,6 +37,23 @@ def build_headroom_spec(proxy_url: str = DEFAULT_PROXY_URL) -> ServerSpec: ) +def build_serena_spec(context: str) -> ServerSpec: + """Construct the canonical Serena MCP server spec for an agent context.""" + return ServerSpec( + name="serena", + command="uvx", + args=( + "--from", + "git+https://github.com/oraios/serena", + "serena", + "start-mcp-server", + "--project-from-cwd", + "--context", + context, + ), + ) + + def install_everywhere( proxy_url: str = DEFAULT_PROXY_URL, *, diff --git a/headroom/mcp_registry/ledger.py b/headroom/mcp_registry/ledger.py new file mode 100644 index 000000000..6bb0160e4 --- /dev/null +++ b/headroom/mcp_registry/ledger.py @@ -0,0 +1,106 @@ +"""Headroom-owned MCP install ledger. + +The ledger tracks MCP servers that Headroom registered on the user's behalf +when the target agent config cannot carry Headroom-specific ownership markers. +It lets unwrap remove only entries still matching the spec Headroom installed, +preserving user-managed MCP servers with the same name. +""" + +from __future__ import annotations + +import hashlib +import json +from datetime import datetime, timezone +from pathlib import Path +from typing import Any + +from headroom import paths + +from .base import ServerSpec + +_LEDGER_FILE = "mcp_installs.json" + + +def ledger_path() -> Path: + """Return the Headroom MCP install ledger path.""" + return paths.workspace_dir() / _LEDGER_FILE + + +def spec_fingerprint(spec: ServerSpec) -> str: + """Stable fingerprint for a registered MCP server spec.""" + payload = { + "name": spec.name, + "command": spec.command, + "args": list(spec.args), + "env": dict(sorted(spec.env.items())), + } + raw = json.dumps(payload, sort_keys=True, separators=(",", ":")).encode("utf-8") + return hashlib.sha256(raw).hexdigest() + + +def record_install(agent: str, spec: ServerSpec, *, path: Path | None = None) -> None: + """Record that Headroom installed ``spec`` for ``agent``.""" + ledger_file = path or ledger_path() + data = _read_ledger(ledger_file) + agents = data.setdefault("agents", {}) + agent_entry = agents.setdefault(agent, {}) + agent_entry[spec.name] = { + "fingerprint": spec_fingerprint(spec), + "installed_at": datetime.now(timezone.utc).isoformat(), + } + _write_ledger(ledger_file, data) + + +def clear_install(agent: str, server_name: str, *, path: Path | None = None) -> None: + """Remove one ledger entry if present.""" + ledger_file = path or ledger_path() + data = _read_ledger(ledger_file) + agents = data.get("agents") + if not isinstance(agents, dict): + return + agent_entry = agents.get(agent) + if not isinstance(agent_entry, dict) or server_name not in agent_entry: + return + del agent_entry[server_name] + if not agent_entry: + del agents[agent] + if not agents: + data.pop("agents", None) + _write_ledger(ledger_file, data) + + +def headroom_installed_matching( + agent: str, + current_spec: ServerSpec | None, + *, + path: Path | None = None, +) -> bool: + """Return True when the ledger says Headroom installed ``current_spec``.""" + if current_spec is None: + return False + ledger_file = path or ledger_path() + data = _read_ledger(ledger_file) + try: + entry = data["agents"][agent][current_spec.name] + except (KeyError, TypeError): + return False + if not isinstance(entry, dict): + return False + return entry.get("fingerprint") == spec_fingerprint(current_spec) + + +def _read_ledger(path: Path) -> dict[str, Any]: + try: + raw = path.read_text(encoding="utf-8") + except OSError: + return {} + try: + data = json.loads(raw) + except json.JSONDecodeError: + return {} + return data if isinstance(data, dict) else {} + + +def _write_ledger(path: Path, data: dict[str, Any]) -> None: + path.parent.mkdir(parents=True, exist_ok=True) + path.write_text(json.dumps(data, indent=2, sort_keys=True) + "\n", encoding="utf-8") diff --git a/headroom/proxy/handlers/anthropic.py b/headroom/proxy/handlers/anthropic.py index b1c2b5570..3e06aae72 100644 --- a/headroom/proxy/handlers/anthropic.py +++ b/headroom/proxy/handlers/anthropic.py @@ -46,8 +46,23 @@ class AnthropicHandlerMixin: canonical = str(tool) return (name, canonical) - # _extract_anthropic_cache_ttl_metrics is defined in StreamingMixin - # (which takes precedence via MRO). Do not duplicate here. + @staticmethod + def _extract_anthropic_cache_ttl_metrics(usage: dict[str, Any] | None) -> tuple[int, int]: + """Extract observed Anthropic cache-write TTL bucket usage. + + HeadroomProxy also inherits StreamingMixin, which exposes the same + helper for SSE usage parsing. Keep this local copy so the Anthropic + handler remains safe when tested or embedded without StreamingMixin. + """ + if not isinstance(usage, dict): + return (0, 0) + cache_creation = usage.get("cache_creation") + if not isinstance(cache_creation, dict): + return (0, 0) + return ( + int(cache_creation.get("ephemeral_5m_input_tokens", 0) or 0), + int(cache_creation.get("ephemeral_1h_input_tokens", 0) or 0), + ) @classmethod def _sort_tools_deterministically( diff --git a/tests/test_cli/test_unwrap_claude.py b/tests/test_cli/test_unwrap_claude.py index db6b9c7ba..90cc6fcb4 100644 --- a/tests/test_cli/test_unwrap_claude.py +++ b/tests/test_cli/test_unwrap_claude.py @@ -86,6 +86,8 @@ def test_unwrap_claude_removes_mcp_rtk_and_stops_proxy( unregistered: list[str] = [] class Registrar: + name = "claude" + def detect(self) -> bool: return True @@ -93,6 +95,9 @@ def test_unwrap_claude_removes_mcp_rtk_and_stops_proxy( unregistered.append(server_name) return True + def get_server(self, server_name: str): + return None + with ( patch("headroom.mcp_registry.ClaudeRegistrar", return_value=Registrar()), patch( @@ -109,6 +114,83 @@ def test_unwrap_claude_removes_mcp_rtk_and_stops_proxy( assert "hooks" not in json.loads(settings.read_text(encoding="utf-8")) +def test_unwrap_claude_preserves_user_managed_serena( + runner: CliRunner, + monkeypatch: pytest.MonkeyPatch, + tmp_path: Path, +) -> None: + monkeypatch.setenv("HEADROOM_WORKSPACE_DIR", str(tmp_path / ".headroom")) + unregistered: list[str] = [] + + class Registrar: + name = "claude" + + def detect(self) -> bool: + return True + + def unregister_server(self, server_name: str) -> bool: + unregistered.append(server_name) + return True + + def get_server(self, server_name: str): + if server_name == "serena": + from headroom.mcp_registry.base import ServerSpec + + return ServerSpec(name="serena", command="/usr/local/bin/custom-serena") + return None + + with ( + patch("headroom.mcp_registry.ClaudeRegistrar", return_value=Registrar()), + patch("headroom.cli.wrap._remove_claude_rtk_hooks", return_value=False), + patch("headroom.cli.wrap._stop_local_proxy_for_unwrap"), + ): + result = runner.invoke(main, ["unwrap", "claude"]) + + assert result.exit_code == 0, result.output + assert unregistered == ["headroom", "codebase-memory-mcp"] + + +def test_unwrap_claude_removes_headroom_installed_serena( + runner: CliRunner, + monkeypatch: pytest.MonkeyPatch, + tmp_path: Path, +) -> None: + monkeypatch.setenv("HEADROOM_WORKSPACE_DIR", str(tmp_path / ".headroom")) + + from headroom.mcp_registry import build_serena_spec + from headroom.mcp_registry.ledger import record_install + + serena_spec = build_serena_spec("claude-code") + record_install("claude", serena_spec) + unregistered: list[str] = [] + + class Registrar: + name = "claude" + + def detect(self) -> bool: + return True + + def unregister_server(self, server_name: str) -> bool: + unregistered.append(server_name) + return True + + def get_server(self, server_name: str): + if server_name == "serena": + return serena_spec + return None + + with ( + patch("headroom.mcp_registry.ClaudeRegistrar", return_value=Registrar()), + patch("headroom.cli.wrap._remove_claude_rtk_hooks", return_value=False), + patch("headroom.cli.wrap._stop_local_proxy_for_unwrap"), + ): + result = runner.invoke(main, ["unwrap", "claude"]) + + assert result.exit_code == 0, result.output + assert unregistered == ["headroom", "codebase-memory-mcp", "serena"] + assert "Removed Headroom-installed Serena MCP server" in result.output + + def test_unwrap_claude_keep_flags_skip_cleanup( runner: CliRunner, ) -> None: diff --git a/tests/test_cli/test_wrap_codex.py b/tests/test_cli/test_wrap_codex.py index a9aa75875..00cfedf0d 100644 --- a/tests/test_cli/test_wrap_codex.py +++ b/tests/test_cli/test_wrap_codex.py @@ -92,6 +92,10 @@ class TestStripCodexHeadroomBlocks: "[mcp_servers.headroom]\n" 'command = "headroom"\n' f"{wrap_mod._CODEX_MCP_END}\n\n" + "# --- Headroom MCP server: serena ---\n" + "[mcp_servers.serena]\n" + 'command = "uvx"\n' + "# --- end Headroom MCP server: serena ---\n\n" f"{wrap_mod._MEMORY_MCP_MARKER}\n" "[mcp_servers.headroom_memory]\n" 'command = "python"\n' @@ -101,6 +105,7 @@ class TestStripCodexHeadroomBlocks: cleaned = wrap_mod._strip_codex_headroom_blocks(content, remove_mcp=True) assert "[mcp_servers.headroom]" not in cleaned + assert "[mcp_servers.serena]" not in cleaned assert "[mcp_servers.headroom_memory]" not in cleaned assert 'model = "gpt-4o"' in cleaned @@ -490,6 +495,43 @@ def test_wrap_codex_prepare_only_updates_stale_mcp_proxy_url( assert "http://127.0.0.1:9000" not in content +def test_wrap_codex_prepare_only_registers_serena_when_uvx_exists( + runner: CliRunner, monkeypatch: pytest.MonkeyPatch, tmp_path: Path +) -> None: + _set_test_home(monkeypatch, tmp_path) + config_file = tmp_path / ".codex" / "config.toml" + config_file.parent.mkdir(parents=True) + + def fake_which(cmd: str) -> str | None: + if cmd == "uvx": + return "/usr/local/bin/uvx" + return None + + with patch("headroom.cli.wrap._ensure_rtk_binary", return_value=None): + with patch("headroom.cli.wrap.shutil.which", side_effect=fake_which): + result = runner.invoke(main, ["wrap", "codex", "--prepare-only"]) + + assert result.exit_code == 0, result.output + content = config_file.read_text() + assert "[mcp_servers.serena]" in content + assert 'command = "uvx"' in content + assert '"--context", "codex"' in content + + +def test_wrap_codex_prepare_only_no_serena_skips_serena( + runner: CliRunner, monkeypatch: pytest.MonkeyPatch, tmp_path: Path +) -> None: + _set_test_home(monkeypatch, tmp_path) + config_file = tmp_path / ".codex" / "config.toml" + config_file.parent.mkdir(parents=True) + + with patch("headroom.cli.wrap._ensure_rtk_binary", return_value=None): + result = runner.invoke(main, ["wrap", "codex", "--prepare-only", "--no-serena"]) + + assert result.exit_code == 0, result.output + assert "[mcp_servers.serena]" not in config_file.read_text() + + def test_unwrap_codex_restores_prior_config_end_to_end( runner: CliRunner, monkeypatch: pytest.MonkeyPatch, tmp_path: Path ) -> None: diff --git a/tests/test_mcp_registry/test_codex_registrar.py b/tests/test_mcp_registry/test_codex_registrar.py index aca53f89e..36a29f2d6 100644 --- a/tests/test_mcp_registry/test_codex_registrar.py +++ b/tests/test_mcp_registry/test_codex_registrar.py @@ -29,6 +29,22 @@ def _spec(env: dict[str, str] | None = None) -> ServerSpec: ) +def _serena_spec() -> ServerSpec: + return ServerSpec( + name="serena", + command="uvx", + args=( + "--from", + "git+https://github.com/oraios/serena", + "serena", + "start-mcp-server", + "--project-from-cwd", + "--context", + "codex", + ), + ) + + def _config_path(tmp_path: Path) -> Path: return tmp_path / ".codex" / "config.toml" @@ -137,6 +153,23 @@ def test_register_includes_env_subtable(tmp_path: Path) -> None: } +def test_register_headroom_and_serena_coexist(tmp_path: Path) -> None: + reg = _make_registrar(tmp_path) + + assert reg.register_server(_spec()).status == RegisterStatus.REGISTERED + assert reg.register_server(_serena_spec()).status == RegisterStatus.REGISTERED + + text = _config_path(tmp_path).read_text() + assert "[mcp_servers.headroom]" in text + assert "[mcp_servers.serena]" in text + assert "# --- Headroom MCP server ---" in text + assert "# --- Headroom MCP server: serena ---" in text + + parsed = tomllib.loads(text) + assert parsed["mcp_servers"]["headroom"]["command"] == "headroom" + assert parsed["mcp_servers"]["serena"]["command"] == "uvx" + + def test_register_omits_env_subtable_when_env_empty(tmp_path: Path) -> None: _make_registrar(tmp_path).register_server(_spec()) text = _config_path(tmp_path).read_text() @@ -228,6 +261,19 @@ def test_unregister_removes_marker_block(tmp_path: Path) -> None: assert "[other_section]" in text +def test_unregister_serena_preserves_headroom_block(tmp_path: Path) -> None: + reg = _make_registrar(tmp_path) + reg.register_server(_spec()) + reg.register_server(_serena_spec()) + + assert reg.unregister_server("serena") is True + text = _config_path(tmp_path).read_text() + assert "[mcp_servers.headroom]" in text + assert "[mcp_servers.serena]" not in text + assert "# --- Headroom MCP server ---" in text + assert "# --- Headroom MCP server: serena ---" not in text + + def test_unregister_returns_false_when_no_block(tmp_path: Path) -> None: cfg = _config_path(tmp_path) cfg.parent.mkdir() diff --git a/tests/test_mcp_registry/test_install.py b/tests/test_mcp_registry/test_install.py index 5ffb38628..3bd97dacc 100644 --- a/tests/test_mcp_registry/test_install.py +++ b/tests/test_mcp_registry/test_install.py @@ -11,6 +11,7 @@ from headroom.mcp_registry.base import ( from headroom.mcp_registry.install import ( DEFAULT_PROXY_URL, build_headroom_spec, + build_serena_spec, install_everywhere, ) @@ -68,6 +69,22 @@ def test_build_spec_default_url_omits_env() -> None: assert spec.env == {} +def test_build_serena_spec_uses_agent_context() -> None: + spec = build_serena_spec("codex") + assert spec.name == "serena" + assert spec.command == "uvx" + assert spec.args == ( + "--from", + "git+https://github.com/oraios/serena", + "serena", + "start-mcp-server", + "--project-from-cwd", + "--context", + "codex", + ) + assert spec.env == {} + + # ---------------------------------------------------------------------- # install_everywhere # ---------------------------------------------------------------------- diff --git a/tests/test_mcp_registry/test_ledger.py b/tests/test_mcp_registry/test_ledger.py new file mode 100644 index 000000000..73100a32b --- /dev/null +++ b/tests/test_mcp_registry/test_ledger.py @@ -0,0 +1,53 @@ +from __future__ import annotations + +from headroom.mcp_registry.base import ServerSpec +from headroom.mcp_registry.ledger import ( + clear_install, + headroom_installed_matching, + record_install, + spec_fingerprint, +) + + +def _spec(command: str = "uvx") -> ServerSpec: + return ServerSpec( + name="serena", + command=command, + args=("--from", "git+https://github.com/oraios/serena", "serena"), + ) + + +def test_ledger_records_matching_install(tmp_path): + ledger = tmp_path / "mcp_installs.json" + spec = _spec() + + record_install("claude", spec, path=ledger) + + assert headroom_installed_matching("claude", spec, path=ledger) is True + + +def test_ledger_rejects_changed_spec(tmp_path): + ledger = tmp_path / "mcp_installs.json" + + record_install("claude", _spec(), path=ledger) + + assert ( + headroom_installed_matching("claude", _spec(command="/custom/serena"), path=ledger) is False + ) + + +def test_clear_install_removes_entry(tmp_path): + ledger = tmp_path / "mcp_installs.json" + spec = _spec() + record_install("claude", spec, path=ledger) + + clear_install("claude", "serena", path=ledger) + + assert headroom_installed_matching("claude", spec, path=ledger) is False + + +def test_spec_fingerprint_stable_for_env_order(): + a = ServerSpec(name="serena", command="uvx", env={"B": "2", "A": "1"}) + b = ServerSpec(name="serena", command="uvx", env={"A": "1", "B": "2"}) + + assert spec_fingerprint(a) == spec_fingerprint(b)