Something went wrong. Try again.
small gleam coding and persistent agent daemon with a detachable cli and a canvas ui forked from okami.mom/albedo
Something went wrong. Try again.
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256"""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 jsonfrom pathlib import Pathimport secretsimport subprocessimport sysimport unittestimport 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@exclusiveclass 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()