d292b9f3f2
Agents asking "is this function used?" get resolved, unknown, or unreferenced per function instead of a bare zero in-degree. Unknown lists why callers may exist: same-name unresolved call sites (excluding foreign receivers and macros), name references (callbacks, address-taken, macro bodies), virtual/trait-impl/external-base/outside-class members, framework attributes, implicit protocol invocations, entry points, test code, incomplete parses, and, unless assume_closed_world is set, exported or external-linkage functions. Unreferenced is a review candidate with an explicit heuristic/compiler evidence basis, never a dead-code proof. The queries are read-only over validated schema-4 models and return complete canonical nodes and edges plus metrics.resolution coverage. Co-Authored-By: Claude Opus 5.5 (1M context) <noreply@anthropic.com>
371 lines
15 KiB
Python
371 lines
15 KiB
Python
#!/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())
|