mirror of
https://github.com/headroomlabs-ai/headroom.git
synced 2026-08-27 14:17:10 -04:00
Merge pull request #447 from chopratejas/serena-mcp-wrap-default
fix: register serena mcp during wrap
This commit is contained in:
Tejas Chopra
GitHub
commit
44231f68cd
11 changed files with 486 additions and 12 deletions
|
|
@ -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...")
|
||||
|
|
|
|||
|
|
@ -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",
|
||||
|
|
|
|||
|
|
@ -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.<name>] 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)
|
||||
|
||||
|
||||
|
|
|
|||
|
|
@ -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,
|
||||
*,
|
||||
|
|
|
|||
106
headroom/mcp_registry/ledger.py
Normal file
106
headroom/mcp_registry/ledger.py
Normal file
|
|
@ -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")
|
||||
|
|
@ -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(
|
||||
|
|
|
|||
|
|
@ -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:
|
||||
|
|
|
|||
|
|
@ -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:
|
||||
|
|
|
|||
|
|
@ -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()
|
||||
|
|
|
|||
|
|
@ -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
|
||||
# ----------------------------------------------------------------------
|
||||
|
|
|
|||
53
tests/test_mcp_registry/test_ledger.py
Normal file
53
tests/test_mcp_registry/test_ledger.py
Normal file
|
|
@ -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)
|
||||
Loading…
Add table
Add a link
Reference in a new issue