#!/usr/bin/env python3 """Artifact-first benchmark for Brick's read-only MCP query surface.""" from __future__ import annotations import argparse import datetime as dt import json import os import pathlib import platform import shutil import statistics import subprocess import sys import time import runner ROOT = pathlib.Path(__file__).resolve().parents[2] PLAN = ROOT / "tests" / "benchmarks" / "plans" / "mcp_queries.json" PROTOCOL_VERSION = "2025-06-18" TOOL_NAMES = [ "brick_impact", "brick_cycles", "brick_bridges", "brick_callers", "brick_uncalled", "brick_evidence", ] def load_plan() -> dict: plan = json.loads(PLAN.read_text(encoding="utf-8")) if plan.get("schema_version") != 1 or plan.get("name") != "mcp_queries": raise ValueError("invalid MCP benchmark plan identity") if int(plan.get("repeats", 0)) < 3: raise ValueError("MCP benchmark requires at least three repeats") cases = plan.get("cases", []) if not cases: raise ValueError("MCP benchmark has no cases") names: set[str] = set() for case in cases: name = case.get("name") if ( not isinstance(name, str) or not name or any(not (character.isascii() and (character.isalnum() or character in "-_")) for character in name) or name in names ): raise ValueError(f"invalid MCP benchmark case name: {name!r}") names.add(name) project_value = case.get("project") if not isinstance(project_value, str): raise ValueError(f"invalid MCP benchmark project: {name!r}") project = (ROOT / project_value).resolve() if not project.is_relative_to(ROOT) or not project.is_dir(): raise ValueError(f"MCP benchmark project is missing or out of root: {name!r}") for field in ["expected_impacted_nodes", "expected_cycles", "expected_bridges"]: if not isinstance(case.get(field), int) or case[field] < 0: raise ValueError(f"invalid {field}: {case.get('name')!r}") if case.get("evidence_kind") not in {"node", "edge"}: raise ValueError(f"invalid evidence_kind: {case.get('name')!r}") return plan def run_logged(command: list[str], stdout_path: pathlib.Path, stderr_path: pathlib.Path) -> None: completed = subprocess.run(command, cwd=ROOT, text=True, capture_output=True, check=False) stdout_path.write_text(completed.stdout, encoding="utf-8") stderr_path.write_text(completed.stderr, encoding="utf-8") if completed.returncode: raise RuntimeError(completed.stderr or completed.stdout) def run_measured( command: list[str], stdin_path: pathlib.Path, stdout_path: pathlib.Path, stderr_path: pathlib.Path, ) -> tuple[int, float, int]: if not hasattr(os, "wait4"): raise RuntimeError("RSS measurement requires os.wait4") started = time.perf_counter() with ( stdin_path.open("rb") as stdin, stdout_path.open("wb") as stdout, stderr_path.open("wb") as stderr, ): process = subprocess.Popen(command, cwd=ROOT, stdin=stdin, stdout=stdout, stderr=stderr) _, status, usage = os.wait4(process.pid, 0) process.returncode = os.waitstatus_to_exitcode(status) elapsed_ms = round((time.perf_counter() - started) * 1000, 3) max_rss_kib = int(usage.ru_maxrss) if sys.platform == "darwin": max_rss_kib //= 1024 return process.returncode, elapsed_ms, max_rss_kib def named_node(graph: dict, name: str) -> dict: matches = [node for node in graph["nodes"] if node.get("name") == name] if len(matches) != 1: raise AssertionError(f"expected one node named {name!r}, found {len(matches)}") return matches[0] def evidence_entity_id(graph: dict, case: dict) -> str: if case["evidence_kind"] == "node": return named_node(graph, case["evidence_name"])["id"] source = named_node(graph, case["evidence_source"])["id"] target = named_node(graph, case["evidence_target"])["id"] matches = [ edge for edge in graph["edges"] if edge["source"] == source and edge["target"] == target and edge["kind"] == "calls" ] if len(matches) != 1: raise AssertionError(f"expected one evidence edge, found {len(matches)}") return matches[0]["id"] def request_messages(start_id: str, entity_id: str) -> list[dict]: def call(request_id: int, name: str, arguments: dict) -> dict: return { "jsonrpc": "2.0", "id": request_id, "method": "tools/call", "params": {"name": name, "arguments": arguments}, } return [ { "jsonrpc": "2.0", "id": 1, "method": "initialize", "params": { "protocolVersion": PROTOCOL_VERSION, "capabilities": {}, "clientInfo": {"name": "brick-benchmark", "version": "1"}, }, }, {"jsonrpc": "2.0", "method": "notifications/initialized"}, {"jsonrpc": "2.0", "id": 2, "method": "tools/list", "params": {}}, call(3, "brick_impact", {"node_id": start_id, "direction": "dependents", "max_depth": 2}), call(4, "brick_cycles", {}), call(5, "brick_bridges", {}), call(6, "brick_evidence", {"entity_id": entity_id}), ] def complete_evidence(edges: list[dict], case_name: str) -> None: required_edge = {"id", "source", "target", "kind", "evidence"} required_evidence = {"lane", "backend", "confidence", "provenance"} for edge in edges: if not required_edge.issubset(edge): raise AssertionError(f"{case_name} emitted an incomplete edge") if not edge["evidence"]: raise AssertionError(f"{case_name} emitted an edge without evidence") for evidence in edge["evidence"]: if set(evidence) != required_evidence: raise AssertionError(f"{case_name} emitted incomplete evidence: {set(evidence)}") def validate_responses(stdout_path: pathlib.Path, graph: dict, case: dict) -> dict: messages = [ json.loads(line) for line in stdout_path.read_text(encoding="utf-8").splitlines() if line ] if len(messages) != 6: raise AssertionError(f"{case['name']} emitted {len(messages)} responses, expected 6") if [message.get("id") for message in messages] != [1, 2, 3, 4, 5, 6]: raise AssertionError(f"{case['name']} response IDs differ") if messages[0]["result"].get("protocolVersion") != PROTOCOL_VERSION: raise AssertionError(f"{case['name']} protocol version differs") tools = messages[1]["result"].get("tools", []) if [tool.get("name") for tool in tools] != TOOL_NAMES: raise AssertionError(f"{case['name']} tool inventory differs") results = [message["result"]["structuredContent"] for message in messages[2:]] for result, query in zip( results, ["impact", "cycles", "bridges", "evidence"], strict=True ): if result.get("query_schema_version") != 1: raise AssertionError(f"{case['name']} query schema differs") if result.get("building_schema_version") != 4: raise AssertionError(f"{case['name']} building schema differs") if result.get("query") != query: raise AssertionError(f"{case['name']} query identity differs") if result.get("scope") != graph["scope"] or result.get("root") != graph["root"]: raise AssertionError(f"{case['name']} query scope differs") for message, result in zip(messages[2:], results, strict=True): if message["result"].get("isError") is not False: raise AssertionError(f"{case['name']} query unexpectedly failed") text = json.loads(message["result"]["content"][0]["text"]) if text != result: raise AssertionError(f"{case['name']} text and structured results differ") impact, cycles, bridges, evidence = results observed = { "impacted_nodes": len(impact["impacted_nodes"]), "cycles": len(cycles["cycles"]), "bridges": len(bridges["bridges"]), } expected = { "impacted_nodes": case["expected_impacted_nodes"], "cycles": case["expected_cycles"], "bridges": case["expected_bridges"], } if observed != expected: raise AssertionError(f"{case['name']} observations {observed}, expected {expected}") if bridges.get("graph_semantics") != "weak-undirected-dependency": raise AssertionError(f"{case['name']} bridge semantics differ") if evidence.get("entity_kind") != case["evidence_kind"]: raise AssertionError(f"{case['name']} evidence entity kind differs") returned_edges = list(impact["edges"]) for cycle in cycles["cycles"]: returned_edges.extend(cycle["edges"]) returned_edges.extend(entry["edge"] for entry in bridges["bridges"]) if evidence["entity_kind"] == "edge": returned_edges.append(evidence["edge"]) else: returned_edges.extend(evidence["incident_edges"]) complete_evidence(returned_edges, case["name"]) return {"results": results, **observed} def run(out_dir: pathlib.Path | None = None) -> pathlib.Path: plan = load_plan() stamp = dt.datetime.now(dt.UTC).strftime("%Y%m%dT%H%M%SZ") artifact = (out_dir or ROOT / "build" / "benchmarks" / "mcp_queries" / stamp).resolve() artifact.mkdir(parents=True, exist_ok=False) binary = ROOT / "target" / "debug" / "brick" prepare = ["cargo", "build", "--quiet"] run_logged(prepare, artifact / "prepare.stdout", artifact / "prepare.stderr") manifest = { "schema_version": 1, "started_at": stamp, "plan": "mcp_queries", "root": str(ROOT), "protocol_version": PROTOCOL_VERSION, "provenance": runner.git_provenance(ROOT), "host": {"platform": platform.platform(), "python": platform.python_version()}, "prepare_command": prepare, } (artifact / "run_manifest.json").write_text( json.dumps(manifest, indent=2) + "\n", encoding="utf-8" ) (artifact / "plan.resolved.json").write_text( json.dumps(plan, indent=2) + "\n", encoding="utf-8" ) samples: list[dict] = [] signatures: dict[str, str] = {} repositories = artifact / "repositories" for case in plan["cases"]: shutil.copytree(ROOT / case["project"], repositories / case["name"]) for repeat in range(plan["repeats"]): for case in plan["cases"]: name = case["name"] graph_path = artifact / f"{name}-graph-{repeat}.json" run_logged( [ str(binary), "build", str(repositories / name), "--json", str(graph_path), "--html", str(artifact / f"{name}-building-{repeat}.html"), ], artifact / f"{name}-build-{repeat}.stdout", artifact / f"{name}-build-{repeat}.stderr", ) graph = json.loads(graph_path.read_text(encoding="utf-8")) start_id = named_node(graph, case["start_name"])["id"] entity_id = evidence_entity_id(graph, case) stdin_path = artifact / f"{name}-requests-{repeat}.jsonl" stdin_path.write_text( "".join(json.dumps(message, separators=(",", ":")) + "\n" for message in request_messages(start_id, entity_id)), encoding="utf-8", ) stdout_path = artifact / f"{name}-responses-{repeat}.jsonl" stderr_path = artifact / f"{name}-mcp-{repeat}.stderr" command = [str(binary), "mcp", str(graph_path)] returncode, elapsed_ms, max_rss_kib = run_measured( command, stdin_path, stdout_path, stderr_path ) if returncode: raise AssertionError(f"{name} MCP exited {returncode}: {stderr_path.read_text(encoding='utf-8')}") observations = validate_responses(stdout_path, graph, case) signature = json.dumps(observations["results"], sort_keys=True, separators=(",", ":")) if name in signatures and signatures[name] != signature: raise AssertionError(f"unstable MCP results for {name} repeat {repeat}") signatures[name] = signature samples.append( { "repeat": repeat, "case": name, "command": command, "returncode": returncode, "elapsed_ms": elapsed_ms, "max_rss_kib": max_rss_kib, "impacted_nodes": observations["impacted_nodes"], "cycles": observations["cycles"], "bridges": observations["bridges"], } ) cases = {} for case in plan["cases"]: case_samples = [sample for sample in samples if sample["case"] == case["name"]] elapsed = [sample["elapsed_ms"] for sample in case_samples] rss = [sample["max_rss_kib"] for sample in case_samples] cases[case["name"]] = { "identity_stable": True, "impacted_nodes": case["expected_impacted_nodes"], "cycles": case["expected_cycles"], "bridges": case["expected_bridges"], "elapsed_ms_median": round(statistics.median(elapsed), 3), "elapsed_ms_min": min(elapsed), "elapsed_ms_max": max(elapsed), "max_rss_kib_median": statistics.median(rss), "max_rss_kib_min": min(rss), "max_rss_kib_max": max(rss), } summary = { "schema_version": 1, "plan": "mcp_queries", "protocol_version": PROTOCOL_VERSION, "repeats": plan["repeats"], "cases": cases, "samples": samples, } (artifact / "summary.json").write_text( json.dumps(summary, indent=2) + "\n", encoding="utf-8" ) report = ["# Brick benchmark: MCP queries", "", f"- Protocol: {PROTOCOL_VERSION}", ""] for name, result in cases.items(): report.extend( [ f"## {name}", "", "- Stable results: yes", f"- Impacted nodes / cycles / bridges: {result['impacted_nodes']} / {result['cycles']} / {result['bridges']}", f"- Median: {result['elapsed_ms_median']} ms", f"- Range: {result['elapsed_ms_min']}–{result['elapsed_ms_max']} ms", f"- Max RSS median: {result['max_rss_kib_median']} KiB", "", ] ) (artifact / "REPORT.md").write_text("\n".join(report), encoding="utf-8") return artifact def main() -> int: parser = argparse.ArgumentParser() parser.add_argument("action", choices=["run", "self-test"]) parser.add_argument("--out", type=pathlib.Path) args = parser.parse_args() plan = load_plan() if args.action == "self-test": print(f"MCP benchmark self-test passed ({len(plan['cases'])} cases).") return 0 artifact = run(args.out) print(f"ARTIFACT={artifact}") return 0 if __name__ == "__main__": raise SystemExit(main())