trending/codebase-memory-mcp
1
0
Fork 0
mirror of https://github.com/DeusData/codebase-memory-mcp.git synced 2026-11-05 08:23:53 +08:00
Code Issues Packages Projects Releases Wiki Activity

test(benchmark): score query quality against response cost

Browse Source 
Record canonical JSON payload bytes separately from CLI/MCP framing in scripts/benchmark-incremental-speed.py and attach the deterministic utf8_bytes_div_4_ceil token estimate to every query oracle.

Require every applicable marker, changed-path, architecture, and route oracle to pass. Render quality checks, query latency, response size, indexing latency, and peak RSS in scripts/summarize-benchmark-results.py; mark Pareto candidates only after correctness passes and all comparison axes are present.

Validated by 10 unittest cases across tests/test_benchmark_incremental_speed.py and tests/test_summarize_benchmark_results.py plus py_compile and git diff --check.

Signed-off-by: Andrew Hundt <ATHundt@gmail.com>
This commit is contained in:
Andrew Hundt
2026-07-14 11:41:49 -04:00
parent 6be5547f56
commit 8bd307e8df
4 changed files with 288 additions and 21 deletions
Download Patch File Download Diff File Expand all files Collapse all files
scripts/benchmark-incremental-speed.py
+72 -4
View File
@@ -269,6 +269,19 @@ def unwrap_mcp_result(response: dict[str, Any]) -> dict[str, Any]:
return result
TOKEN_ESTIMATOR = "utf8_bytes_div_4_ceil"
def canonical_response_bytes(data: dict[str, Any]) -> bytes:
"""Serialize the tool payload independently of CLI/MCP envelopes."""
return json.dumps(data, separators=(",", ":"), sort_keys=True).encode("utf-8")
def estimate_response_tokens(payload: bytes) -> int:
"""Return a deterministic, dependency-free byte/4 token estimate."""
return (len(payload) + 3) // 4
class McpClient:
def __init__(self, binary: Path, env: dict[str, str], timeout: int) -> None:
self.binary = binary
@@ -713,9 +726,16 @@ def build_tool_call_result(
elapsed_ms: float,
include_logs: bool,
) -> dict[str, Any]:
payload = canonical_response_bytes(data)
result: dict[str, Any] = {
"elapsed_ms": round(elapsed_ms, 3),
# Preserve the historical field while separating transport framing from
# the canonical payload used for cross-transport comparisons.
"stdout_bytes": stdout_bytes,
"transport_response_bytes": stdout_bytes,
"response_bytes": len(payload),
"response_token_estimate": estimate_response_tokens(payload),
"token_estimator": TOKEN_ESTIMATOR,
"response": data,
"freshness_state": response_freshness_state(data) or None,
"freshness": response_freshness(data),
@@ -1452,6 +1472,38 @@ def oracle_passed(tool_result: dict[str, Any], marker: str | None) -> bool:
return marker in json.dumps(response, sort_keys=True)
def score_quality_oracles(
oracles: dict[str, Any],
expectations: dict[str, tuple[str | None, str]],
) -> dict[str, Any]:
"""Attach auditable per-oracle verdicts and summarize applicable checks."""
applicable_count = 0
passed_count = 0
for name, result in oracles.items():
if not isinstance(result, dict):
continue
expected, criterion = expectations.get(name, (None, "no quality criterion"))
applicable = expected is not None
passed = False
if applicable:
applicable_count += 1
response = result.get("response")
passed = expected in json.dumps(response, separators=(",", ":"), sort_keys=True)
passed_count += int(passed)
result["quality"] = {
"applicable": applicable,
"passed": passed if applicable else None,
"criterion": criterion,
"expected_substring": expected,
}
return {
"passed": passed_count == applicable_count,
"passed_count": passed_count,
"applicable_count": applicable_count,
"score": round(passed_count / applicable_count, 6) if applicable_count else None,
}
def run_self_dogfood_oracles(
transport: str,
binary: Path,
@@ -1465,6 +1517,7 @@ def run_self_dogfood_oracles(
changed_paths = list(mutation.get("changed_paths") or [])
first_changed = changed_paths[0] if changed_paths else ""
oracles: dict[str, Any] = {}
expectations: dict[str, tuple[str | None, str]] = {}
if marker:
search_code_args: dict[str, Any] = {"project": project, "pattern": marker, "limit": 5}
if first_changed:
@@ -1480,6 +1533,7 @@ def run_self_dogfood_oracles(
args.include_logs,
client,
)
expectations["marker_search_graph"] = (marker, "mutated symbol appears in graph search")
oracles["marker_search_code"] = run_tool_call_for_transport(
transport,
binary,
@@ -1490,6 +1544,7 @@ def run_self_dogfood_oracles(
args.include_logs,
client,
)
expectations["marker_search_code"] = (marker, "mutated symbol appears in source search")
if first_changed:
oracles["changed_file_query_graph"] = run_tool_call_for_transport(
transport,
@@ -1507,6 +1562,10 @@ def run_self_dogfood_oracles(
args.include_logs,
client,
)
expectations["changed_file_query_graph"] = (
first_changed,
"changed file path appears in graph query",
)
oracles["scoped_architecture"] = run_tool_call_for_transport(
transport,
binary,
@@ -1517,6 +1576,10 @@ def run_self_dogfood_oracles(
args.include_logs,
client,
)
expectations["scoped_architecture"] = (
first_changed,
"changed file path appears in scoped architecture",
)
oracles["route_freshness_probe"] = run_tool_call_for_transport(
transport,
binary,
@@ -1527,11 +1590,16 @@ def run_self_dogfood_oracles(
args.include_logs,
client,
)
oracles["passed"] = all(
oracle_passed(result, marker)
for key, result in oracles.items()
if key in {"marker_search_graph", "marker_search_code"}
route_expected = "/api/pan4-oracle" if mutation.get("description", "").startswith(
"HTTP UI handler"
) else None
expectations["route_freshness_probe"] = (
route_expected,
"new route literal appears in route search" if route_expected else "route mutation not applicable",
)
quality = score_quality_oracles(oracles, expectations)
oracles["quality"] = quality
oracles["passed"] = quality["passed"]
return oracles
scripts/summarize-benchmark-results.py
+120 -15
View File
@@ -65,16 +65,27 @@ def summarize_group(label: str, reports: list[dict[str, Any]]) -> dict[str, Any]
for case in cases
if isinstance(case.get("canonical_graph"), dict)
]
oracles = [
bool(case["oracles"].get("passed"))
for case in cases
if isinstance(case.get("oracles"), dict)
]
oracles: list[bool] = []
for case in cases:
case_oracles = case.get("oracles")
if not isinstance(case_oracles, dict):
continue
verdict = case_oracles.get("passed")
if not isinstance(verdict, bool):
quality = case_oracles.get("quality")
verdict = quality.get("passed") if isinstance(quality, dict) else None
if isinstance(verdict, bool):
oracles.append(verdict)
case_passes = [bool(case.get("passed")) for case in cases]
incremental_ms: list[float] = []
full_ms: list[float] = []
speedups: list[float] = []
peak_rss: list[int] = []
query_latency_ms: list[float] = []
query_response_bytes: list[float] = []
query_response_tokens: list[float] = []
quality_passed = 0
quality_applicable = 0
for case in cases:
incremental = case.get("incremental", {})
full = case.get("fresh_fast_full_after_change", {})
@@ -90,6 +101,21 @@ def summarize_group(label: str, reports: list[dict[str, Any]]) -> dict[str, Any]
peak_rss.append(full["peak_rss_mb"])
if isinstance(case.get("speedup_full_rebuild_over_incremental"), (int, float)):
speedups.append(float(case["speedup_full_rebuild_over_incremental"]))
case_oracles = case.get("oracles", {})
if isinstance(case_oracles, dict):
quality = case_oracles.get("quality", {})
if isinstance(quality, dict):
quality_passed += int(quality.get("passed_count") or 0)
quality_applicable += int(quality.get("applicable_count") or 0)
for oracle in case_oracles.values():
if not isinstance(oracle, dict):
continue
if isinstance(oracle.get("elapsed_ms"), (int, float)):
query_latency_ms.append(float(oracle["elapsed_ms"]))
if isinstance(oracle.get("response_bytes"), (int, float)):
query_response_bytes.append(float(oracle["response_bytes"]))
if isinstance(oracle.get("response_token_estimate"), (int, float)):
query_response_tokens.append(float(oracle["response_token_estimate"]))
quality_failed = any(not value for value in canonical) or any(not value for value in oracles)
if quality_failed:
@@ -120,6 +146,13 @@ def summarize_group(label: str, reports: list[dict[str, Any]]) -> dict[str, Any]
"cases": ratio(sum(case_passes), len(case_passes)),
"canonical": ratio(sum(canonical), len(canonical)),
"oracles": ratio(sum(oracles), len(oracles)),
"quality_score": (
quality_passed / quality_applicable if quality_applicable else None
),
"quality_checks": ratio(quality_passed, quality_applicable),
"query_response_p50_bytes": percentile(query_response_bytes, 0.50),
"query_response_p50_tokens": percentile(query_response_tokens, 0.50),
"query_latency_p50_ms": percentile(query_latency_ms, 0.50),
"capabilities": config_label(reports),
"incremental_p50_ms": percentile(incremental_ms, 0.50),
"incremental_p95_ms": percentile(incremental_ms, 0.95),
@@ -128,9 +161,58 @@ def summarize_group(label: str, reports: list[dict[str, Any]]) -> dict[str, Any]
"peak_rss_mb": max(peak_rss) if peak_rss else None,
"cleanup": ratio(cleanup_passes, len(reports)),
"binary_sha256": ", ".join(value[:12] for value in hashes) or "n/a",
"pareto": "unclassified",
}
PARETO_MINIMIZE = (
"incremental_p50_ms",
"query_latency_p50_ms",
"query_response_p50_tokens",
"peak_rss_mb",
)
def dominates(left: dict[str, Any], right: dict[str, Any]) -> bool:
left_quality = left.get("quality_score")
right_quality = right.get("quality_score")
if not isinstance(left_quality, (int, float)) or not isinstance(
right_quality, (int, float)
):
return False
left_values = [left.get(key) for key in PARETO_MINIMIZE]
right_values = [right.get(key) for key in PARETO_MINIMIZE]
if not all(isinstance(value, (int, float)) for value in left_values + right_values):
return False
no_worse = left_quality >= right_quality and all(
left_value <= right_value
for left_value, right_value in zip(left_values, right_values, strict=True)
)
strictly_better = left_quality > right_quality or any(
left_value < right_value
for left_value, right_value in zip(left_values, right_values, strict=True)
)
return no_worse and strictly_better
def mark_pareto_frontier(rows: list[dict[str, Any]]) -> None:
"""Mark correctness-admissible, fully measured non-dominated candidates."""
eligible = [
row
for row in rows
if row.get("decision") == "PASS"
and isinstance(row.get("quality_score"), (int, float))
and all(isinstance(row.get(key), (int, float)) for key in PARETO_MINIMIZE)
]
for row in rows:
row["pareto"] = "ineligible"
for row in eligible:
dominators = [other for other in eligible if other is not row and dominates(other, row)]
row["pareto"] = (
f"dominated by {dominators[0]['candidate']}" if dominators else "frontier"
)
def display(value: Any, digits: int = 1) -> str:
if value is None:
return "n/a"
@@ -140,13 +222,14 @@ def display(value: Any, digits: int = 1) -> str:
def render_markdown(rows: list[dict[str, Any]]) -> str:
mark_pareto_frontier(rows)
lines = [
"# Codebase Memory performance and quality summary",
"",
"| Candidate | Decision | Cases | Canonical | Task oracles | Capabilities | "
"Incremental p50 ms | Incremental p95 ms | Full p50 ms | Speedup p50 | "
"Peak RSS MB | Cleanup | Binary SHA-256 |",
"|---|---|---:|---:|---:|---|---:|---:|---:|---:|---:|---:|---|",
"| Candidate | Decision | Quality | Checks | Canonical | Task oracles | "
"Response p50 bytes | Response p50 tokens* | Query p50 ms | Incremental p50 ms | "
"Peak RSS MB | Pareto |",
"|---|---|---:|---:|---:|---:|---:|---:|---:|---:|---:|---|",
]
for row in rows:
lines.append(
@@ -155,15 +238,32 @@ def render_markdown(rows: list[dict[str, Any]]) -> str:
(
display(row["candidate"]),
display(row["decision"]),
display(row["cases"]),
display(row["quality_score"], 3),
display(row["quality_checks"]),
display(row["canonical"]),
display(row["oracles"]),
display(row["capabilities"]),
display(row["query_response_p50_bytes"]),
display(row["query_response_p50_tokens"]),
display(row["query_latency_p50_ms"]),
display(row["incremental_p50_ms"]),
display(row["peak_rss_mb"]),
display(row["pareto"]),
)
)
+ " |"
)
lines.extend(("", "## Performance and provenance", "", "| Candidate | Cases | Capabilities | Incremental p95 ms | Full p50 ms | Speedup p50 | Cleanup | Binary SHA-256 |", "|---|---:|---|---:|---:|---:|---:|---|"))
for row in rows:
lines.append(
"| "
+ " | ".join(
(
display(row["candidate"]),
display(row["cases"]),
display(row["capabilities"]),
display(row["incremental_p95_ms"]),
display(row["full_p50_ms"]),
display(row["speedup_p50"], 2),
display(row["peak_rss_mb"]),
display(row["cleanup"]),
display(row["binary_sha256"]),
)
@@ -172,6 +272,13 @@ def render_markdown(rows: list[dict[str, Any]]) -> str:
)
lines.extend(
(
"",
"* Response tokens use the recorded `utf8_bytes_div_4_ceil` deterministic estimate; "
"bytes remain the exact canonical JSON payload measurement.",
"",
"Pareto status considers only candidates that pass correctness/quality and have every "
"axis measured. It maximizes quality while minimizing incremental and query latency, "
"response-token estimate, and peak RSS.",
"",
"A speedup is accepted only when the case gate and every applicable canonical-graph "
"and task-oracle check pass. `n/a` means the input artifact did not measure that axis.",
@@ -199,9 +306,7 @@ def main() -> int:
if not isinstance(document, dict):
raise SystemExit(f"error: expected JSON object in {path}")
grouped[label].append(document)
markdown = render_markdown(
[summarize_group(label, reports) for label, reports in grouped.items()]
)
markdown = render_markdown([summarize_group(label, reports) for label, reports in grouped.items()])
if args.out:
output = Path(args.out).expanduser()
output.parent.mkdir(parents=True, exist_ok=True)
tests/test_benchmark_incremental_speed.py
+35
View File
@@ -12,6 +12,41 @@ SPEC.loader.exec_module(BENCHMARK)
class BenchmarkIncrementalSpeedTest(unittest.TestCase):
def test_tool_result_measures_canonical_payload_not_transport_envelope(self) -> None:
result = BENCHMARK.build_tool_call_result(
{"name": "alpha", "items": [1, 2]}, "", 999, 12.5, False
)
canonical = b'{"items":[1,2],"name":"alpha"}'
self.assertEqual(result["transport_response_bytes"], 999)
self.assertEqual(result["response_bytes"], len(canonical))
self.assertEqual(
result["response_token_estimate"], BENCHMARK.estimate_response_tokens(canonical)
)
self.assertEqual(result["token_estimator"], "utf8_bytes_div_4_ceil")
def test_quality_summary_requires_every_applicable_oracle(self) -> None:
oracles = {
"marker_search_graph": {
"response": {"results": [{"name": "wanted_marker"}]}
},
"changed_file_query_graph": {
"response": {"results": [{"file_path": "wrong.c"}]}
},
"route_freshness_probe": {"response": {"routes": []}},
}
expectations = {
"marker_search_graph": ("wanted_marker", "marker returned"),
"changed_file_query_graph": ("src/wanted.c", "changed path returned"),
"route_freshness_probe": (None, "route check not applicable"),
}
summary = BENCHMARK.score_quality_oracles(oracles, expectations)
self.assertFalse(summary["passed"])
self.assertEqual(summary["passed_count"], 1)
self.assertEqual(summary["applicable_count"], 2)
self.assertEqual(summary["score"], 0.5)
self.assertFalse(oracles["changed_file_query_graph"]["quality"]["passed"])
self.assertFalse(oracles["route_freshness_probe"]["quality"]["applicable"])
def test_binary_metadata_records_content_identity(self) -> None:
with tempfile.TemporaryDirectory() as tmpdir:
binary = Path(tmpdir) / "cbm"
tests/test_summarize_benchmark_results.py
+61 -2
View File
@@ -62,10 +62,69 @@ class SummarizeBenchmarkResultsTest(unittest.TestCase):
"speedup_full_rebuild_over_incremental": 10.0,
}
markdown = SUMMARY.render_markdown([SUMMARY.summarize_group("latest", [report(case)])])
header = markdown.splitlines()[2]
self.assertLess(header.index("Decision"), header.index("Speedup p50"))
self.assertLess(markdown.index("Decision"), markdown.index("Speedup p50"))
self.assertIn("Binary SHA-256", markdown)
def test_query_quality_size_latency_and_pareto_frontier(self) -> None:
compact_case = {
"passed": True,
"canonical_graph": {"equal": True},
"oracles": {
"quality": {"passed": True, "passed_count": 2, "applicable_count": 2},
"marker": {
"elapsed_ms": 5,
"response_bytes": 80,
"response_token_estimate": 20,
},
},
"incremental": {"elapsed_ms": 10, "peak_rss_mb": 80},
"fresh_fast_full_after_change": {"elapsed_ms": 100, "peak_rss_mb": 90},
"speedup_full_rebuild_over_incremental": 10.0,
}
slower_case = {
**compact_case,
"oracles": {
"quality": {"passed": True, "passed_count": 2, "applicable_count": 2},
"marker": {
"elapsed_ms": 8,
"response_bytes": 120,
"response_token_estimate": 30,
},
},
"incremental": {"elapsed_ms": 20, "peak_rss_mb": 100},
}
rows = [
SUMMARY.summarize_group("compact", [report(compact_case)]),
SUMMARY.summarize_group("slower", [report(slower_case)]),
]
SUMMARY.mark_pareto_frontier(rows)
self.assertEqual(rows[0]["quality_score"], 1.0)
self.assertEqual(rows[0]["query_response_p50_bytes"], 80.0)
self.assertEqual(rows[0]["query_response_p50_tokens"], 20.0)
self.assertEqual(rows[0]["query_latency_p50_ms"], 5.0)
self.assertEqual(rows[0]["pareto"], "frontier")
self.assertEqual(rows[1]["pareto"], "dominated by compact")
def test_failed_quality_is_not_pareto_eligible(self) -> None:
case = {
"passed": False,
"canonical_graph": {"equal": True},
"oracles": {
"quality": {"passed": False, "passed_count": 1, "applicable_count": 2},
"marker": {
"elapsed_ms": 1,
"response_bytes": 4,
"response_token_estimate": 1,
},
},
"incremental": {"elapsed_ms": 1, "peak_rss_mb": 1},
"fresh_fast_full_after_change": {"elapsed_ms": 2, "peak_rss_mb": 2},
}
row = SUMMARY.summarize_group("bad-quality", [report(case)])
SUMMARY.mark_pareto_frontier([row])
self.assertEqual(row["decision"], "REJECT: quality/correctness")
self.assertEqual(row["pareto"], "ineligible")
if __name__ == "__main__":
unittest.main()