"""Validate outgoing MCP handshake revisions and refused-overlay cleanup. The strict server records initialize parameters and rejects non-revision values. Both configured servers and editor overlays use the native MCP client. """ import json from pathlib import Path import secrets import subprocess import sys import unittest import urllib.error from harness import ( Albedo, Provider, Reply, exclusive, operation_id, text, wait_until, ) SERVER = Path(__file__).with_name("fake_mcp_server.py") LEGACY = {"2024-11-05", "2025-03-26", "2025-06-18", "2025-11-25"} def records(path): """The initialize params the server received, oldest first.""" if not path.exists(): return [] return [json.loads(line) for line in path.read_text().splitlines()] def offered_versions(test, path): """The protocolVersion of every initialize, each a concrete legacy revision.""" offered = [params.get("protocolVersion") for params in records(path)] test.assertTrue(offered, "the server never received an initialize") for version in offered: test.assertIn( version, LEGACY, f"initialize offered {version!r}, not a legacy revision" ) return offered def exited(pid): """True once the process is gone, or a zombie that nothing will reap.""" state = subprocess.run( ["ps", "-o", "stat=", "-p", str(pid)], capture_output=True, text=True, check=False, ).stdout.strip() return state == "" or state.startswith("Z") def refused(test, call, status, code): with test.assertRaises(urllib.error.HTTPError) as failure: call() test.assertEqual(failure.exception.code, status) test.assertEqual(json.load(failure.exception)["code"], code) class ToolModel: """A model that calls the first MCP tool once, then answers. It keeps the names it was offered and each tool output it was shown.""" def __init__(self): self.offered = [] self.outputs = [] self.provider = Provider(self.answer) def answer(self, request): outputs = [ item for item in request["input"] if item.get("type") == "function_call_output" ] if outputs: self.outputs.append(outputs[-1]["output"]) return text("done") self.offered = sorted( tool["name"] for tool in request.get("tools", []) if tool["name"].startswith("mcp_") ) if not self.offered: return text("no mcp tools") return Reply( "python", tool_name=self.offered[0], tool_arguments={"message": "wire check"}, ) # exclusive: configured servers live in extensions.json, and the strict flags # come from the daemon's environment, both global to the daemon @exclusive class McpWireConfiguredTests(unittest.TestCase): def setUp(self): self.model = ToolModel() self.addCleanup(self.model.provider.close) self.record = None def prepare(app): self.record = app.workspace / "initialize.jsonl" app.env.update( ALBEDO_MCP_STRICT="1", ALBEDO_MCP_RECORD=str(self.record), ) self.app = Albedo(self.model.provider, protocol="responses", prepare=prepare) self.app.__enter__() self.addCleanup(self.app.__exit__, None, None, None) self.session = self.app.session() def configure(self): """One strict stdio server named fake, with no protocolVersion.""" path = self.app.home / "extensions.json" settings = json.loads(path.read_text()) if path.exists() else {} settings["mcp"] = { "servers": { "fake": { "type": "stdio", "command": sys.executable, "args": [str(SERVER)], "env": { "FAKE_MCP_STRICT": {"env": "ALBEDO_MCP_STRICT"}, "FAKE_MCP_RECORD": {"env": "ALBEDO_MCP_RECORD"}, }, "startupTimeoutMs": 20000, } } } path.write_text(json.dumps(settings)) def select_mcp(self, enabled): route = f"/sessions/{self.session}?view=configuration" with self.app.api(route) as response: json.load(response) etag = response.headers["ETag"] self.app.api( route, {"selection": {"extensions": {"mcp": enabled}}}, method="PATCH", headers={"If-Match": etag}, ).close() with self.app.api( f"/sessions/{self.session}/reload", {"target": "session"} ) as response: return json.load(response) def turn(self): self.app.prompt(self.session, "use the server").close() wait_until( lambda: self.model.provider.requests, 60, "the model never saw the turn", ) self.app.idle(self.session, timeout=60) def test_a_server_with_no_pinned_version_is_offered_a_real_revision(self): self.configure() self.assertEqual(self.select_mcp(True)["session"]["state"], "applied") self.turn() offered_versions(self, self.record) self.assertEqual(len(self.model.offered), 1, self.model.offered) self.assertTrue(self.model.offered[0].startswith("mcp_fake_echo_")) self.assertTrue(self.model.outputs, "the tool never ran") self.assertIn("wire check", self.model.outputs[-1]) class EditorClient: """The ACP creation and read routes, as one editor sends them.""" def __init__(self, app): self.app = app self.token = secrets.token_urlsafe(32) def server(self, name, **env): """An editor stdio server; the strict flag is always on.""" values = {"FAKE_MCP_STRICT": "1", **env} return { "name": name, "command": sys.executable, "args": [str(SERVER)], "env": [{"name": key, "value": value} for key, value in values.items()], } def headers(self, extra=None): return {"Albedo-Attachment-Token": self.token, **(extra or {})} def create(self, session_id, servers): with self.app.api( f"/sessions/{session_id}", { "kind": "acp", "workspace": str(self.app.workspace), "mcp_servers": list(servers), }, method="PUT", headers=self.headers({"If-None-Match": "*"}), ) as response: return json.load(response) def read(self, session_id): with self.app.api(f"/sessions/{session_id}?tail=0") as response: return json.load(response) # Refusal ends before any turn, so the overlay's session can share the daemon. class McpWireOverlayCleanupTests(unittest.TestCase): def setUp(self): self.app = Albedo() self.app.__enter__() self.addCleanup(self.app.__exit__, None, None, None) self.editor = EditorClient(self.app) def test_a_strict_overlay_server_that_refuses_initialize_leaves_nothing_running( self, ): pidfile = self.app.workspace / "refused.pid" session_id = operation_id() refused( self, lambda: self.editor.create( session_id, [ self.editor.server( "strict", FAKE_MCP_REFUSE_INITIALIZE="1", FAKE_MCP_PIDFILE=str(pidfile), ) ], ), 409, "mcp_overlay_failed", ) refused( self, lambda: self.editor.read(session_id), 404, "session_not_found", ) pid = int(pidfile.read_text()) wait_until( lambda: exited(pid), 15, f"the refused overlay server (pid {pid}) is still running", ) if __name__ == "__main__": unittest.main()