diff --git a/docs/content/docs/mcp.mdx b/docs/content/docs/mcp.mdx index af6302825..7fe121ff8 100644 --- a/docs/content/docs/mcp.mdx +++ b/docs/content/docs/mcp.mdx @@ -187,6 +187,8 @@ Install Headroom so it's globally on PATH — `uv tool install "headroom-ai[mcp] ## Architecture +For user-managed Serena drift, run `headroom mcp reconcile` to inspect the current recommendation. Add `--adopt` only when you want Headroom to replace the Serena entry. + ### MCP only (no proxy) The LLM calls `headroom_compress` on demand. Compression happens locally in the MCP process. Originals are stored in a local `CompressionStore` with 1-hour TTL. diff --git a/headroom/cli/mcp.py b/headroom/cli/mcp.py index 7563822bb..40e8c25ca 100644 --- a/headroom/cli/mcp.py +++ b/headroom/cli/mcp.py @@ -216,6 +216,52 @@ def mcp_uninstall() -> None: click.echo("Headroom MCP is not configured. Nothing to uninstall.") +@mcp.command("reconcile") +@click.option("--adopt", is_flag=True, help="Replace only the Serena entry with Headroom's spec.") +def mcp_reconcile(adopt: bool) -> None: + """Inspect or explicitly reconcile a user-managed Serena MCP entry.""" + from headroom.mcp_registry import ( + CLAUDE_SERENA_CONTEXT, + ClaudeConfigMutationError, + ClaudeRegistrar, + RegisterStatus, + build_serena_spec, + ) + from headroom.mcp_registry.ledger import ( + LedgerMutationError, + record_install, + validate_ledger_for_mutation, + ) + + registrar = ClaudeRegistrar() + if not registrar.detect(): + raise click.ClickException("claude is not detected") + recommended = build_serena_spec(CLAUDE_SERENA_CONTEXT) + observed = registrar.get_server("serena") + + if adopt: + try: + registrar.validate_configs_for_mutation() + validate_ledger_for_mutation() + except (ClaudeConfigMutationError, LedgerMutationError) as exc: + raise click.ClickException(str(exc)) from exc + if adopt: + result = registrar.register_server(recommended, force=True) + if result.status not in (RegisterStatus.REGISTERED, RegisterStatus.ALREADY): + raise click.ClickException(result.detail or "could not adopt Serena configuration") + record_install("claude", recommended) + click.echo( + "Adopted Headroom's Serena configuration for Claude; unrelated config preserved." + ) + return + + click.echo("Serena reconciliation for Claude") + click.echo(f" observed: {'absent' if observed is None else 'present'}") + click.echo(f" recommendation: {recommended.command} {' '.join(recommended.args)}") + if observed is not None and observed != recommended: + click.echo(" action: use --adopt to replace it") + + @mcp.command("status") def mcp_status() -> None: """Check Headroom MCP configuration status. diff --git a/headroom/cli/wrap.py b/headroom/cli/wrap.py index 638d9e56a..1c7176e2c 100644 --- a/headroom/cli/wrap.py +++ b/headroom/cli/wrap.py @@ -2018,17 +2018,18 @@ def _setup_serena_mcp( spec = build_serena_spec(context) result = registrar.register_server(spec, force=force) + owned_drift = ( + result.status == RegisterStatus.MISMATCH + and not force + and headroom_installed_matching(registrar.name, registrar.get_server("serena")) + ) # Migrate a stale Headroom-installed entry. register_server won't overwrite # a differing spec without force, so an older Headroom Serena entry would # otherwise persist across re-wraps. Force-update it only when the ledger # proves Headroom installed the entry that's currently on disk — never a # user-managed Serena. - if ( - result.status == RegisterStatus.MISMATCH - and not force - and headroom_installed_matching(registrar.name, registrar.get_server("serena")) - ): + if result.status == RegisterStatus.MISMATCH and not force and owned_drift: result = registrar.register_server(spec, force=True) if result.status == RegisterStatus.REGISTERED: click.echo(" Serena MCP: migrated previously-installed entry to current spec") @@ -2041,7 +2042,13 @@ def _setup_serena_mcp( result, label="Serena MCP", verbose=verbose, - overwrite_hint="update or remove the existing serena MCP entry, then rerun headroom wrap", + overwrite_hint=( + "run headroom wrap again" + if owned_drift + else "run headroom mcp reconcile --adopt" + if registrar.name == "claude" + else "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: @@ -4932,11 +4939,11 @@ def claude( click.echo(" Skipping MCP retrieve tool (--no-mcp)") # Coding-task compressor: Serena (retires any legacy tokensave entry). - from headroom.mcp_registry import ClaudeRegistrar + from headroom.mcp_registry import CLAUDE_SERENA_CONTEXT, ClaudeRegistrar _setup_coding_compressor( ClaudeRegistrar(), - serena_context="claude-code", + serena_context=CLAUDE_SERENA_CONTEXT, serena=serena, no_serena=no_serena, no_tokensave=no_tokensave, diff --git a/headroom/mcp_registry/__init__.py b/headroom/mcp_registry/__init__.py index 5f1ff4f8f..f19ad1ba7 100644 --- a/headroom/mcp_registry/__init__.py +++ b/headroom/mcp_registry/__init__.py @@ -14,11 +14,12 @@ without changing the calling code. from __future__ import annotations from .base import MCPRegistrar, RegisterResult, RegisterStatus, ServerSpec -from .claude import ClaudeRegistrar +from .claude import ClaudeConfigMutationError, ClaudeRegistrar from .codex import CodexRegistrar from .display import any_succeeded, format_result, format_results from .grok import GrokRegistrar from .install import ( + CLAUDE_SERENA_CONTEXT, DEFAULT_PROXY_URL, build_headroom_spec, build_serena_spec, @@ -30,6 +31,8 @@ from .server_json import build_server_json, render_server_json __all__ = [ "DEFAULT_PROXY_URL", + "CLAUDE_SERENA_CONTEXT", + "ClaudeConfigMutationError", "ClaudeRegistrar", "CodexRegistrar", "GrokRegistrar", diff --git a/headroom/mcp_registry/claude.py b/headroom/mcp_registry/claude.py index 7acbc02fb..de7ee308c 100644 --- a/headroom/mcp_registry/claude.py +++ b/headroom/mcp_registry/claude.py @@ -26,6 +26,10 @@ from .base import MCPRegistrar, RegisterResult, RegisterStatus, ServerSpec logger = logging.getLogger(__name__) +class ClaudeConfigMutationError(ValueError): + """Raised when a Claude config cannot be safely changed.""" + + class ClaudeRegistrar(MCPRegistrar): """Register MCP servers with Claude Code.""" @@ -84,6 +88,34 @@ class ClaudeRegistrar(MCPRegistrar): return entry return None + def validate_configs_for_mutation(self) -> None: + """Validate every Claude config root before an explicit mutation.""" + seen: set[Path] = set() + for config_path in (self._modern_config, self._legacy_config): + if config_path in seen or not config_path.exists(): + continue + seen.add(config_path) + try: + raw = config_path.read_text(encoding="utf-8") + except OSError as exc: + raise ClaudeConfigMutationError( + f"could not read Claude config {config_path}: {exc}" + ) from exc + try: + config = json.loads(raw) + except json.JSONDecodeError as exc: + raise ClaudeConfigMutationError( + f"Claude config {config_path} is not valid JSON; refusing to mutate" + ) from exc + if not isinstance(config, dict): + raise ClaudeConfigMutationError( + f"Claude config {config_path} must contain a JSON object" + ) + if "mcpServers" in config and not isinstance(config["mcpServers"], dict): + raise ClaudeConfigMutationError( + f"Claude config {config_path} has a non-object mcpServers; refusing to mutate" + ) + def register_server(self, spec: ServerSpec, *, force: bool = False) -> RegisterResult: existing = self.get_server(spec.name) if existing is not None: diff --git a/headroom/mcp_registry/install.py b/headroom/mcp_registry/install.py index e550f6ded..4d1fbeb72 100644 --- a/headroom/mcp_registry/install.py +++ b/headroom/mcp_registry/install.py @@ -14,6 +14,7 @@ from .opencode import OpencodeRegistrar #: Default proxy URL used when none is given. DEFAULT_PROXY_URL = "http://127.0.0.1:8787" +CLAUDE_SERENA_CONTEXT = "claude-code" def get_all_registrars() -> list[MCPRegistrar]: diff --git a/headroom/mcp_registry/ledger.py b/headroom/mcp_registry/ledger.py index 6bb0160e4..e802227e7 100644 --- a/headroom/mcp_registry/ledger.py +++ b/headroom/mcp_registry/ledger.py @@ -21,6 +21,10 @@ from .base import ServerSpec _LEDGER_FILE = "mcp_installs.json" +class LedgerMutationError(ValueError): + """Raised when a ledger cannot be safely updated.""" + + def ledger_path() -> Path: """Return the Headroom MCP install ledger path.""" return paths.workspace_dir() / _LEDGER_FILE @@ -41,9 +45,17 @@ def spec_fingerprint(spec: ServerSpec) -> str: 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() + # Automatic installs must recover from a stale or damaged ledger. The + # explicit reconcile route performs strict validation before config writes. data = _read_ledger(ledger_file) - agents = data.setdefault("agents", {}) - agent_entry = agents.setdefault(agent, {}) + agents = data.get("agents") + if not isinstance(agents, dict): + agents = {} + data["agents"] = agents + agent_entry = agents.get(agent) + if not isinstance(agent_entry, dict): + agent_entry = {} + agents[agent] = agent_entry agent_entry[spec.name] = { "fingerprint": spec_fingerprint(spec), "installed_at": datetime.now(timezone.utc).isoformat(), @@ -89,16 +101,45 @@ def headroom_installed_matching( return entry.get("fingerprint") == spec_fingerprint(current_spec) -def _read_ledger(path: Path) -> dict[str, Any]: +def validate_ledger_for_mutation(path: Path | None = None) -> None: + """Reject malformed ledger structure before a config mutation.""" + _read_ledger(path or ledger_path(), for_mutation=True) + + +def _read_ledger(path: Path, *, for_mutation: bool = False) -> dict[str, Any]: try: raw = path.read_text(encoding="utf-8") - except OSError: + except FileNotFoundError: + return {} + except OSError as exc: + if for_mutation: + raise LedgerMutationError(f"MCP install ledger is unreadable: {path}") from exc return {} try: data = json.loads(raw) - except json.JSONDecodeError: + except json.JSONDecodeError as exc: + if for_mutation: + raise LedgerMutationError(f"MCP install ledger is invalid JSON: {path}") from exc return {} - return data if isinstance(data, dict) else {} + if not isinstance(data, dict): + if for_mutation: + raise LedgerMutationError("MCP install ledger must contain a JSON object") + return {} + if for_mutation: + for section in ("agents",): + section_data = data.get(section) + if not isinstance(section_data, dict) or any( + not isinstance(agent_entry, dict) + or any( + not isinstance(server_entry, dict) + or not isinstance(server_entry.get("fingerprint"), str) + or not isinstance(server_entry.get("installed_at"), str) + for server_entry in agent_entry.values() + ) + for agent_entry in section_data.values() + ): + raise LedgerMutationError(f"MCP install ledger section {section!r} is malformed") + return data def _write_ledger(path: Path, data: dict[str, Any]) -> None: diff --git a/tests/fixtures/headroom-issue-3054.json b/tests/fixtures/headroom-issue-3054.json new file mode 100644 index 000000000..e5eb4f12b --- /dev/null +++ b/tests/fixtures/headroom-issue-3054.json @@ -0,0 +1,26 @@ +{ + "issue": 3054, + "url": "https://github.com/headroomlabs-ai/headroom/issues/3054", + "old_serena_args": [ + "--from", + "git+https://github.com/oraios/serena", + "serena", + "start-mcp-server", + "--project-from-cwd", + "--context", + "claude-code", + "--open-web-dashboard", + "False" + ], + "recommended_serena_args": [ + "--from", + "serena-agent", + "serena", + "start-mcp-server", + "--project-from-cwd", + "--context", + "claude-code", + "--open-web-dashboard", + "False" + ] +} diff --git a/tests/test_cli/test_mcp_reconcile.py b/tests/test_cli/test_mcp_reconcile.py new file mode 100644 index 000000000..519aa0591 --- /dev/null +++ b/tests/test_cli/test_mcp_reconcile.py @@ -0,0 +1,291 @@ +from __future__ import annotations + +import json +import os +import sys +from pathlib import Path + +import pytest +from click.testing import CliRunner + +from headroom.cli.main import main +from headroom.mcp_registry import ClaudeRegistrar, build_serena_spec +from headroom.mcp_registry.ledger import headroom_installed_matching + +FIXTURE = Path(__file__).parents[1] / "fixtures" / "headroom-issue-3054.json" + + +def _setup(monkeypatch, tmp_path: Path): + config = tmp_path / ".claude.json" + config.write_text( + json.dumps( + { + "oauthAccount": {"email": "user@example.com"}, + "mcpServers": { + "serena": { + "command": "uvx", + "args": json.loads(FIXTURE.read_text())["old_serena_args"], + }, + "other": {"command": "other", "args": []}, + }, + "projects": {"/repo": {"trust": True}}, + } + ) + ) + registrar = ClaudeRegistrar(claude_cli=None, home_dir=tmp_path) + monkeypatch.setattr("headroom.mcp_registry.ClaudeRegistrar", lambda: registrar) + ledger = tmp_path / "ledger.json" + monkeypatch.setattr("headroom.mcp_registry.ledger.ledger_path", lambda: ledger) + return config, ledger + + +def test_issue_fixture_reconcile_is_base_fail_head_pass(monkeypatch, tmp_path: Path): + config, _ = _setup(monkeypatch, tmp_path) + fixture = json.loads(FIXTURE.read_text()) + recommended = build_serena_spec("claude-code") + assert list(recommended.args) == fixture["recommended_serena_args"] + assert CliRunner().invoke(main, ["mcp", "reconcile"]).exit_code == 0 + adopted = CliRunner().invoke(main, ["mcp", "reconcile", "--adopt"]) + assert adopted.exit_code == 0, adopted.output + assert json.loads(config.read_text())["mcpServers"]["serena"]["args"] == list(recommended.args) + + +def test_read_only_preserves_config_and_ledger_bytes_and_mtimes(monkeypatch, tmp_path: Path): + config, ledger = _setup(monkeypatch, tmp_path) + ledger.write_text("not json") + before = ( + config.read_bytes(), + ledger.read_bytes(), + os.stat(config).st_mtime_ns, + os.stat(ledger).st_mtime_ns, + ) + result = CliRunner().invoke(main, ["mcp", "reconcile"]) + assert result.exit_code == 0, result.output + after = ( + config.read_bytes(), + ledger.read_bytes(), + os.stat(config).st_mtime_ns, + os.stat(ledger).st_mtime_ns, + ) + assert after == before + assert "--adopt" in result.output + + +def test_adopt_preserves_unrelated_config_and_records_ownership(monkeypatch, tmp_path: Path): + config, ledger = _setup(monkeypatch, tmp_path) + result = CliRunner().invoke(main, ["mcp", "reconcile", "--adopt"]) + assert result.exit_code == 0, result.output + data = json.loads(config.read_text()) + assert data["oauthAccount"] == {"email": "user@example.com"} + assert data["projects"] == {"/repo": {"trust": True}} + assert data["mcpServers"]["other"] == {"command": "other", "args": []} + assert data["mcpServers"]["serena"]["args"] == list(build_serena_spec("claude-code").args) + assert json.loads(ledger.read_text())["agents"]["claude"]["serena"]["fingerprint"] + + +@pytest.mark.parametrize( + "contents", + [ + "not json", + "[]", + '{"agents": null}', + '{"agents": []}', + '{"agents": {"claude": null}}', + '{"agents": {"claude": []}}', + '{"agents": {"claude": {"serena": null}}}', + ], +) +def test_malformed_ledger_blocks_adopt_before_config_write( + monkeypatch, tmp_path: Path, contents: str +): + config, ledger = _setup(monkeypatch, tmp_path) + before = config.read_bytes() + ledger.write_text(contents) + result = CliRunner().invoke(main, ["mcp", "reconcile", "--adopt"]) + assert result.exit_code != 0 + assert "ledger" in result.output.lower() + assert config.read_bytes() == before + + +def test_corrupt_ledger_is_tolerated_by_read_only(monkeypatch, tmp_path: Path): + _, ledger = _setup(monkeypatch, tmp_path) + ledger.write_text('{"agents": []}') + result = CliRunner().invoke(main, ["mcp", "reconcile"]) + assert result.exit_code == 0, result.output + + +def test_reconcile_rejects_absent_claude(monkeypatch, tmp_path: Path): + _, _ = _setup(monkeypatch, tmp_path) + registrar = ClaudeRegistrar(claude_cli=None, home_dir=tmp_path) + monkeypatch.setattr(registrar, "detect", lambda: False) + monkeypatch.setattr("headroom.mcp_registry.ClaudeRegistrar", lambda: registrar) + + result = CliRunner().invoke(main, ["mcp", "reconcile", "--adopt"]) + + assert result.exit_code != 0 + assert "claude is not detected" in result.output + + +def test_reconcile_adopt_preserves_malformed_config(monkeypatch, tmp_path: Path): + config, _ = _setup(monkeypatch, tmp_path) + config.write_text("not json") + before = config.read_bytes() + + result = CliRunner().invoke(main, ["mcp", "reconcile", "--adopt"]) + + assert result.exit_code != 0 + assert config.read_bytes() == before + + +def test_adopt_rejects_malformed_modern_before_touching_valid_legacy(monkeypatch, tmp_path: Path): + modern = tmp_path / ".claude.json" + legacy = tmp_path / ".claude" / "mcp.json" + legacy.parent.mkdir() + modern.write_text("not json") + legacy.write_text( + json.dumps( + { + "mcpServers": { + "serena": {"command": "uvx", "args": ["--from", "user"]}, + "other": {"command": "other"}, + } + } + ) + ) + registrar = ClaudeRegistrar(claude_cli=None, home_dir=tmp_path) + monkeypatch.setattr("headroom.mcp_registry.ClaudeRegistrar", lambda: registrar) + ledger = tmp_path / "ledger.json" + monkeypatch.setattr("headroom.mcp_registry.ledger.ledger_path", lambda: ledger) + before = (modern.read_bytes(), legacy.read_bytes()) + + result = CliRunner().invoke(main, ["mcp", "reconcile", "--adopt"]) + + assert result.exit_code != 0 + assert "not valid JSON" in result.output + assert (modern.read_bytes(), legacy.read_bytes()) == before + + +def test_adopt_rejects_non_dict_mcp_servers_in_legacy_root(monkeypatch, tmp_path: Path): + modern, _ = _setup(monkeypatch, tmp_path) + legacy = tmp_path / ".claude" / "mcp.json" + legacy.parent.mkdir() + legacy.write_text(json.dumps({"mcpServers": []})) + before = (modern.read_bytes(), legacy.read_bytes()) + + result = CliRunner().invoke(main, ["mcp", "reconcile", "--adopt"]) + + assert result.exit_code != 0 + assert "non-object mcpServers" in result.output + assert (modern.read_bytes(), legacy.read_bytes()) == before + + +def test_unreadable_ledger_blocks_adopt_without_partial_mutation(monkeypatch, tmp_path: Path): + config, ledger = _setup(monkeypatch, tmp_path) + ledger.write_text(json.dumps({"agents": {}})) + before = (config.read_bytes(), ledger.read_bytes()) + original_read_text = Path.read_text + + def unreadable(path: Path, *args, **kwargs): + if path == ledger: + raise PermissionError("test unreadable ledger") + return original_read_text(path, *args, **kwargs) + + monkeypatch.setattr(Path, "read_text", unreadable) + + result = CliRunner().invoke(main, ["mcp", "reconcile", "--adopt"]) + + assert result.exit_code != 0 + assert "unreadable" in result.output + assert (config.read_bytes(), ledger.read_bytes()) == before + + +@pytest.mark.parametrize("state", ["absent", "matching", "user-drift", "headroom-drift"]) +@pytest.mark.parametrize("adopt", [False, True]) +def test_reconcile_state_matrix(monkeypatch, tmp_path: Path, state: str, adopt: bool): + config, ledger = _setup(monkeypatch, tmp_path) + data = json.loads(config.read_text()) + recommended = build_serena_spec("claude-code") + owned_spec = None + if state == "absent": + del data["mcpServers"]["serena"] + elif state == "matching": + data["mcpServers"]["serena"] = { + "command": recommended.command, + "args": list(recommended.args), + } + elif state == "user-drift": + data["mcpServers"]["serena"]["args"] = ["--from", "user-managed"] + elif state == "headroom-drift": + from headroom.mcp_registry.ledger import record_install + + stale = build_serena_spec("claude-code") + stale.args = ("--from", "headroom-installed-old") + owned_spec = stale + data["mcpServers"]["serena"] = { + "command": stale.command, + "args": list(stale.args), + } + record_install("claude", stale, path=ledger) + config.write_text(json.dumps(data)) + if owned_spec is not None: + assert headroom_installed_matching("claude", owned_spec, path=ledger) + result = CliRunner().invoke(main, ["mcp", "reconcile"] + (["--adopt"] if adopt else [])) + assert result.exit_code == 0, result.output + observed = json.loads(config.read_text())["mcpServers"].get("serena") + ownership = observed is not None and headroom_installed_matching( + "claude", + build_serena_spec("claude-code") if observed["args"] == list(recommended.args) else None, + path=ledger, + ) + if adopt: + assert observed == { + "command": recommended.command, + "args": list(recommended.args), + } + assert ownership + assert "Adopted Headroom" in result.output + elif state == "headroom-drift": + assert observed["args"] == ["--from", "headroom-installed-old"] + assert headroom_installed_matching("claude", owned_spec, path=ledger) + assert ownership is False + assert "observed: present" in result.output + else: + assert not ownership + assert "Serena reconciliation for Claude" in result.output + + +def test_only_adopt_is_a_reconcile_mutation(monkeypatch, tmp_path: Path): + _setup(monkeypatch, tmp_path) + result = CliRunner().invoke(main, ["mcp", "reconcile", "--help"]) + assert result.exit_code == 0 + assert "--adopt" in result.output + for option in ("--acknowledge", "--clear", "--agent", "--server"): + assert option not in result.output + + +def test_ordinary_install_does_not_adopt_serena(monkeypatch, tmp_path: Path): + config, _ = _setup(monkeypatch, tmp_path) + before = config.read_bytes() + monkeypatch.setitem(sys.modules, "mcp", object()) + registrar = ClaudeRegistrar(claude_cli=None, home_dir=tmp_path) + monkeypatch.setattr("headroom.mcp_registry.install.get_all_registrars", lambda: [registrar]) + result = CliRunner().invoke(main, ["mcp", "install", "--agent", "claude"]) + assert result.exit_code == 0, result.output + after = json.loads(config.read_text()) + before_data = json.loads(before) + assert after["mcpServers"]["serena"] == before_data["mcpServers"]["serena"] + assert after["mcpServers"]["headroom"]["args"] == ["mcp", "serve"] + assert "mcp reconcile --adopt" not in result.output + + +def test_mcp_install_force_preserves_user_managed_serena(monkeypatch, tmp_path: Path): + config, _ = _setup(monkeypatch, tmp_path) + before = json.loads(config.read_text())["mcpServers"]["serena"] + monkeypatch.setitem(sys.modules, "mcp", object()) + registrar = ClaudeRegistrar(claude_cli=None, home_dir=tmp_path) + monkeypatch.setattr("headroom.mcp_registry.install.get_all_registrars", lambda: [registrar]) + + result = CliRunner().invoke(main, ["mcp", "install", "--agent", "claude", "--force"]) + + assert result.exit_code == 0, result.output + assert json.loads(config.read_text())["mcpServers"]["serena"] == before diff --git a/tests/test_cli/test_serena_reconcile.py b/tests/test_cli/test_serena_reconcile.py new file mode 100644 index 000000000..f65cbea1a --- /dev/null +++ b/tests/test_cli/test_serena_reconcile.py @@ -0,0 +1,124 @@ +from __future__ import annotations + +from pathlib import Path + +from headroom.cli import wrap as wrap_cli +from headroom.mcp_registry import build_serena_spec +from headroom.mcp_registry.base import RegisterResult, RegisterStatus, ServerSpec +from headroom.mcp_registry.ledger import headroom_installed_matching, record_install + + +class _Registrar: + display_name = "Claude Code" + + def __init__(self, current: ServerSpec | None, *, name: str = "claude"): + self.name = name + self.current = current + self.force_calls: list[bool] = [] + + def detect(self) -> bool: + return True + + def get_server(self, name: str) -> ServerSpec | None: + return self.current if name == "serena" else None + + def register_server(self, spec: ServerSpec, *, force: bool = False) -> RegisterResult: + self.force_calls.append(force) + if self.current == spec: + return RegisterResult(RegisterStatus.ALREADY, "matches") + if self.current is not None and not force: + return RegisterResult(RegisterStatus.MISMATCH, "different") + self.current = spec + return RegisterResult(RegisterStatus.REGISTERED, "updated") + + +def _quiet(monkeypatch): + monkeypatch.setattr(wrap_cli, "_ensure_serena_dashboard_disabled", lambda **kwargs: None) + monkeypatch.setattr(wrap_cli, "_inject_serena_instructions", lambda *args, **kwargs: None) + monkeypatch.setattr(wrap_cli, "_serena_project_skip_reason", lambda root: "test") + monkeypatch.setattr(wrap_cli, "_index_serena_project", lambda **kwargs: None) + monkeypatch.setattr(wrap_cli.shutil, "which", lambda name: "uvx" if name == "uvx" else None) + + +def test_automatic_wrap_migrates_owned_drift_and_recurs_to_noop( + monkeypatch, tmp_path: Path, capsys +): + _quiet(monkeypatch) + monkeypatch.setattr( + "headroom.mcp_registry.ledger.ledger_path", lambda: tmp_path / "ledger.json" + ) + stale = ServerSpec("serena", "uvx", ("--from", "old")) + record_install("claude", stale) + registrar = _Registrar(stale) + wrap_cli._setup_serena_mcp(registrar, context="claude-code", verbose=True) + assert registrar.current == build_serena_spec("claude-code") + assert registrar.force_calls == [False, True] + assert headroom_installed_matching("claude", registrar.current) + capsys.readouterr() + wrap_cli._setup_serena_mcp(registrar, context="claude-code", verbose=True) + assert registrar.force_calls == [False, True, False] + + +def test_automatic_wrap_owned_drift_suggests_rerun_wrap(monkeypatch, tmp_path: Path, capsys): + _quiet(monkeypatch) + monkeypatch.setattr( + "headroom.mcp_registry.ledger.ledger_path", lambda: tmp_path / "ledger.json" + ) + stale = ServerSpec("serena", "uvx", ("--from", "old")) + record_install("claude", stale) + + class _FailedMigrationRegistrar(_Registrar): + def register_server(self, spec, *, force=False): + if force: + self.force_calls.append(force) + return RegisterResult(RegisterStatus.MISMATCH, "still different") + return super().register_server(spec, force=force) + + wrap_cli._setup_serena_mcp( + _FailedMigrationRegistrar(stale), context="claude-code", verbose=True + ) + + output = capsys.readouterr().out + assert "run headroom wrap again" in output + assert "mcp reconcile --adopt" not in output + + +def test_automatic_wrap_preserves_user_managed_warning(monkeypatch, tmp_path: Path, capsys): + _quiet(monkeypatch) + monkeypatch.setattr( + "headroom.mcp_registry.ledger.ledger_path", lambda: tmp_path / "ledger.json" + ) + user = ServerSpec("serena", "uvx", ("--from", "user")) + registrar = _Registrar(user) + wrap_cli._setup_serena_mcp(registrar, context="claude-code", verbose=True) + assert registrar.current == user + assert registrar.force_calls == [False] + assert "existing config differs" in capsys.readouterr().out + + +def test_automatic_wrap_recovers_from_malformed_ledger(monkeypatch, tmp_path: Path): + _quiet(monkeypatch) + ledger = tmp_path / "ledger.json" + ledger.write_text("not json") + monkeypatch.setattr("headroom.mcp_registry.ledger.ledger_path", lambda: ledger) + registrar = _Registrar(None) + + wrap_cli._setup_serena_mcp(registrar, context="claude-code", verbose=True) + + current = registrar.get_server("serena") + assert current == build_serena_spec("claude-code") + assert headroom_installed_matching("claude", current) + + +def test_non_claude_wrap_keeps_usable_remediation_hint(monkeypatch, tmp_path: Path, capsys): + _quiet(monkeypatch) + monkeypatch.setattr( + "headroom.mcp_registry.ledger.ledger_path", lambda: tmp_path / "ledger.json" + ) + registrar = _Registrar(ServerSpec("serena", "uvx", ("--from", "user")), name="codex") + + wrap_cli._setup_serena_mcp(registrar, context="codex", verbose=True) + + output = capsys.readouterr().out + assert "update or remove the existing serena MCP entry" in output + assert "mcp reconcile --adopt" not in output diff --git a/tests/test_mcp_registry/test_ledger.py b/tests/test_mcp_registry/test_ledger.py index 73100a32b..aa7ff08dc 100644 --- a/tests/test_mcp_registry/test_ledger.py +++ b/tests/test_mcp_registry/test_ledger.py @@ -1,53 +1,97 @@ from __future__ import annotations +import json + +import pytest + +import headroom.mcp_registry.ledger as ledger_module from headroom.mcp_registry.base import ServerSpec from headroom.mcp_registry.ledger import ( + LedgerMutationError, clear_install, headroom_installed_matching, record_install, spec_fingerprint, + validate_ledger_for_mutation, ) def _spec(command: str = "uvx") -> ServerSpec: - return ServerSpec( - name="serena", - command=command, - args=("--from", "git+https://github.com/oraios/serena", "serena"), - ) + return ServerSpec("serena", command, ("--from", "serena-agent", "serena")) -def test_ledger_records_matching_install(tmp_path): +def test_ledger_records_and_clears_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) + clear_install("claude", "serena", path=ledger) + assert not headroom_installed_matching("claude", spec, path=ledger) + + +def test_spec_fingerprint_is_stable_for_env_order(): + a = ServerSpec("serena", "uvx", env={"B": "2", "A": "1"}) + b = ServerSpec("serena", "uvx", env={"A": "1", "B": "2"}) + assert spec_fingerprint(a) == spec_fingerprint(b) + + +@pytest.mark.parametrize( + "value", + [ + "not json", + [], + {"agents": None}, + {"agents": []}, + {"agents": {"claude": None}}, + {"agents": {"claude": []}}, + {"agents": {"claude": {"serena": None}}}, + {"agents": {"claude": {"serena": {"fingerprint": "only"}}}}, + ], +) +def test_mutation_preflight_rejects_unsafe_shapes(tmp_path, value): + ledger = tmp_path / "mcp_installs.json" + ledger.write_text(value if isinstance(value, str) else json.dumps(value)) + with pytest.raises(LedgerMutationError): + validate_ledger_for_mutation(ledger) + + +def test_mutation_preflight_rejects_unreadable_ledger(monkeypatch, tmp_path): + ledger = tmp_path / "mcp_installs.json" + ledger.write_text('{"agents": {}}') + original_read_text = ledger_module.Path.read_text + + def unreadable(path, *args, **kwargs): + if path == ledger: + raise PermissionError("test unreadable ledger") + return original_read_text(path, *args, **kwargs) + + monkeypatch.setattr(ledger_module.Path, "read_text", unreadable) + + with pytest.raises(LedgerMutationError, match="unreadable"): + validate_ledger_for_mutation(ledger) + + +def test_read_matching_tolerates_corrupt_ledger(tmp_path): + ledger = tmp_path / "mcp_installs.json" + ledger.write_text("not json") + assert not headroom_installed_matching("claude", _spec(), path=ledger) + + +def test_record_install_recovers_from_corrupt_ledger(tmp_path): + ledger = tmp_path / "mcp_installs.json" + ledger.write_text("not json") + spec = _spec() record_install("claude", spec, path=ledger) - assert headroom_installed_matching("claude", spec, path=ledger) is True + assert headroom_installed_matching("claude", spec, path=ledger) -def test_ledger_rejects_changed_spec(tmp_path): +@pytest.mark.parametrize("contents", ['{"agents": null}', '{"agents": {"claude": null}}']) +def test_record_install_recovers_from_unsafe_ledger_shape(tmp_path, contents): ledger = tmp_path / "mcp_installs.json" + ledger.write_text(contents) 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) + assert headroom_installed_matching("claude", _spec(), path=ledger)