diff --git a/scripts/benchmark-incremental-speed.py b/scripts/benchmark-incremental-speed.py index 82b7cb75d..f247d47fc 100755 --- a/scripts/benchmark-incremental-speed.py +++ b/scripts/benchmark-incremental-speed.py @@ -8,6 +8,7 @@ indexing only for that cache, and removes only paths it created. from __future__ import annotations import argparse +import hashlib import json import os import queue @@ -456,6 +457,23 @@ def parse_log_int_field(stderr: str, marker: str, field: str) -> int | None: return None +def parse_log_max_int_field(stderr: str, marker: str, field: str) -> int | None: + prefix = f"{field}=" + maximum: int | None = None + for line in stderr.splitlines(): + if marker not in line: + continue + for item in line.split(): + if not item.startswith(prefix): + continue + try: + value = int(item.split("=", 1)[1]) + except ValueError: + continue + maximum = value if maximum is None else max(maximum, value) + return maximum + + def parse_exact_reason(stderr: str) -> str | None: detail = parse_exact_route_detail(stderr) reason = detail.get("reason") @@ -584,6 +602,7 @@ def build_index_result( freshness_state = response_freshness_state(data) result: dict[str, Any] = { "elapsed_ms": elapsed_ms_int, + "peak_rss_mb": parse_log_max_int_field(stderr, "mem.phase", "peak_mb"), "indexed_work_elapsed_ms": indexed_ms, "unlogged_overhead_ms": (elapsed_ms_int - indexed_ms) if indexed_ms is not None else None, "response": data, @@ -1252,6 +1271,19 @@ def git_metadata(repo_root: Path, timeout: int) -> dict[str, Any]: } +def binary_metadata(binary: Path) -> dict[str, Any]: + digest = hashlib.sha256() + with binary.open("rb") as stream: + for chunk in iter(lambda: stream.read(1024 * 1024), b""): + digest.update(chunk) + stat = binary.stat() + return { + "path": str(binary.resolve()), + "size_bytes": stat.st_size, + "sha256": digest.hexdigest(), + } + + def clone_real_repo(url: str, target: Path, timeout: int) -> Path: target.parent.mkdir(parents=True, exist_ok=True) proc, _ = command_result( @@ -1618,6 +1650,7 @@ def run_matrix(args: argparse.Namespace, binary: Path) -> tuple[dict[str, Any], report: dict[str, Any] = { "generated_at_utc": datetime.now(timezone.utc).isoformat(), "binary": str(binary), + "binary_metadata": binary_metadata(binary), "work_root": str(work_root), "mode": "matrix", "parameters": { @@ -1776,6 +1809,7 @@ def run_self_dogfood(args: argparse.Namespace, binary: Path) -> tuple[dict[str, report: dict[str, Any] = { "generated_at_utc": datetime.now(timezone.utc).isoformat(), "binary": str(binary), + "binary_metadata": binary_metadata(binary), "work_root": str(work_root), "source_repo": str(source_repo), "source_git": git_metadata(source_repo, args.timeout), @@ -1940,6 +1974,7 @@ def main() -> int: report: dict[str, Any] = { "generated_at_utc": datetime.now(timezone.utc).isoformat(), "binary": str(binary), + "binary_metadata": binary_metadata(binary), "work_root": str(work_root), "parameters": { "files": args.files, diff --git a/scripts/summarize-benchmark-results.py b/scripts/summarize-benchmark-results.py new file mode 100644 index 000000000..e1882f73e --- /dev/null +++ b/scripts/summarize-benchmark-results.py @@ -0,0 +1,214 @@ +#!/usr/bin/env python3 +"""Aggregate existing CBM benchmark JSON into a quality-first Markdown table.""" + +from __future__ import annotations + +import argparse +import json +import math +import statistics +from collections import defaultdict +from pathlib import Path +from typing import Any + + +def percentile(values: list[float], quantile: float) -> float | None: + if not values: + return None + ordered = sorted(values) + index = max(0, math.ceil(quantile * len(ordered)) - 1) + return float(ordered[index]) + + +def ratio(passed: int, applicable: int) -> str: + return f"{passed}/{applicable}" if applicable else "n/a" + + +def cases_from_report(report: dict[str, Any]) -> list[dict[str, Any]]: + cases = report.get("cases") + if isinstance(cases, list): + return [case for case in cases if isinstance(case, dict)] + measurements = report.get("measurements") + derived = report.get("derived") + if isinstance(measurements, dict) and isinstance(derived, dict): + return [ + { + "passed": derived.get("passed"), + "incremental": measurements.get("incremental", {}), + "fresh_fast_full_after_change": measurements.get( + "fresh_fast_full_after_change", {} + ), + "speedup_full_rebuild_over_incremental": derived.get( + "speedup_full_rebuild_over_incremental" + ), + } + ] + return [] + + +def config_label(reports: list[dict[str, Any]]) -> str: + labels: set[str] = set() + for report in reports: + parameters = report.get("parameters", {}) + overrides = parameters.get("config_overrides", {}) if isinstance(parameters, dict) else {} + if isinstance(overrides, dict) and overrides: + labels.add(", ".join(f"{key}={overrides[key]}" for key in sorted(overrides))) + else: + labels.add("defaults") + return " / ".join(sorted(labels)) + + +def summarize_group(label: str, reports: list[dict[str, Any]]) -> dict[str, Any]: + cases = [case for report in reports for case in cases_from_report(report)] + canonical = [ + bool(case["canonical_graph"].get("equal")) + 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) + ] + 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] = [] + for case in cases: + incremental = case.get("incremental", {}) + full = case.get("fresh_fast_full_after_change", {}) + if isinstance(incremental, dict): + if isinstance(incremental.get("elapsed_ms"), (int, float)): + incremental_ms.append(float(incremental["elapsed_ms"])) + if isinstance(incremental.get("peak_rss_mb"), int): + peak_rss.append(incremental["peak_rss_mb"]) + if isinstance(full, dict): + if isinstance(full.get("elapsed_ms"), (int, float)): + full_ms.append(float(full["elapsed_ms"])) + if isinstance(full.get("peak_rss_mb"), int): + 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"])) + + quality_failed = any(not value for value in canonical) or any(not value for value in oracles) + if quality_failed: + decision = "REJECT: quality/correctness" + elif case_passes and not all(case_passes): + decision = "REJECT: benchmark gate" + elif not cases: + decision = "REJECT: no cases" + else: + decision = "PASS" + + cleanup_passes = sum( + bool(report.get("cleanup", {}).get("removed")) + for report in reports + if isinstance(report.get("cleanup"), dict) + ) + hashes = sorted( + { + str(report.get("binary_metadata", {}).get("sha256", "")) + for report in reports + if isinstance(report.get("binary_metadata"), dict) + and report.get("binary_metadata", {}).get("sha256") + } + ) + return { + "candidate": label, + "decision": decision, + "cases": ratio(sum(case_passes), len(case_passes)), + "canonical": ratio(sum(canonical), len(canonical)), + "oracles": ratio(sum(oracles), len(oracles)), + "capabilities": config_label(reports), + "incremental_p50_ms": percentile(incremental_ms, 0.50), + "incremental_p95_ms": percentile(incremental_ms, 0.95), + "full_p50_ms": percentile(full_ms, 0.50), + "speedup_p50": float(statistics.median(speedups)) if speedups else None, + "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", + } + + +def display(value: Any, digits: int = 1) -> str: + if value is None: + return "n/a" + if isinstance(value, float): + return f"{value:.{digits}f}" + return str(value).replace("|", "\\|") + + +def render_markdown(rows: list[dict[str, Any]]) -> str: + 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 |", + "|---|---|---:|---:|---:|---|---:|---:|---:|---:|---:|---:|---|", + ] + for row in rows: + lines.append( + "| " + + " | ".join( + ( + display(row["candidate"]), + display(row["decision"]), + display(row["cases"]), + display(row["canonical"]), + display(row["oracles"]), + display(row["capabilities"]), + display(row["incremental_p50_ms"]), + 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"]), + ) + ) + + " |" + ) + lines.extend( + ( + "", + "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.", + ) + ) + return "\n".join(lines) + "\n" + + +def parse_input(value: str) -> tuple[str, Path]: + label, separator, raw_path = value.partition("=") + if not separator or not label or not raw_path: + raise argparse.ArgumentTypeError("--input expects LABEL=PATH") + return label, Path(raw_path).expanduser() + + +def main() -> int: + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument("--input", action="append", required=True, type=parse_input) + parser.add_argument("--out", default="") + args = parser.parse_args() + grouped: dict[str, list[dict[str, Any]]] = defaultdict(list) + for label, path in args.input: + with path.open(encoding="utf-8") as stream: + document = json.load(stream) + 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()] + ) + if args.out: + output = Path(args.out).expanduser() + output.parent.mkdir(parents=True, exist_ok=True) + output.write_text(markdown, encoding="utf-8") + print(markdown, end="") + return 0 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/tests/test_benchmark_incremental_speed.py b/tests/test_benchmark_incremental_speed.py new file mode 100644 index 000000000..e4d0d69f2 --- /dev/null +++ b/tests/test_benchmark_incremental_speed.py @@ -0,0 +1,50 @@ +import importlib.util +import tempfile +import unittest +from pathlib import Path + + +SCRIPT = Path(__file__).resolve().parents[1] / "scripts" / "benchmark-incremental-speed.py" +SPEC = importlib.util.spec_from_file_location("benchmark_incremental_speed", SCRIPT) +assert SPEC and SPEC.loader +BENCHMARK = importlib.util.module_from_spec(SPEC) +SPEC.loader.exec_module(BENCHMARK) + + +class BenchmarkIncrementalSpeedTest(unittest.TestCase): + def test_binary_metadata_records_content_identity(self) -> None: + with tempfile.TemporaryDirectory() as tmpdir: + binary = Path(tmpdir) / "cbm" + binary.write_bytes(b"auditable-binary") + metadata = BENCHMARK.binary_metadata(binary) + self.assertEqual(metadata["size_bytes"], 16) + self.assertEqual( + metadata["sha256"], + "5d984f78de8a55923b5ab343710b12830af15f8415f350135e5346fc7753b4d5", + ) + self.assertTrue(metadata["path"].endswith("/cbm")) + + def test_build_index_result_reports_maximum_logged_peak_rss(self) -> None: + stderr = "\n".join( + ( + "level=info msg=mem.phase phase=registry_build rss_mb=120 peak_mb=144", + "level=info msg=mem.phase phase=parallel_resolve rss_mb=192 peak_mb=256", + "level=info msg=pipeline.done elapsed_ms=80", + ) + ) + result = BENCHMARK.build_index_result( + {"publish_kind": "full"}, stderr, stdout_bytes=10, elapsed_ms=100.0, + include_logs=False, + ) + self.assertEqual(result["peak_rss_mb"], 256) + + def test_build_index_result_uses_none_without_memory_markers(self) -> None: + result = BENCHMARK.build_index_result( + {"publish_kind": "full"}, "level=info msg=pipeline.done elapsed_ms=80", 10, + 100.0, False, + ) + self.assertIsNone(result["peak_rss_mb"]) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_summarize_benchmark_results.py b/tests/test_summarize_benchmark_results.py new file mode 100644 index 000000000..531089786 --- /dev/null +++ b/tests/test_summarize_benchmark_results.py @@ -0,0 +1,71 @@ +import importlib.util +import unittest +from pathlib import Path + + +SCRIPT = Path(__file__).resolve().parents[1] / "scripts" / "summarize-benchmark-results.py" +SPEC = importlib.util.spec_from_file_location("summarize_benchmark_results", SCRIPT) +assert SPEC and SPEC.loader +SUMMARY = importlib.util.module_from_spec(SPEC) +SPEC.loader.exec_module(SUMMARY) + + +def report(case: dict, sha: str = "a" * 64) -> dict: + return { + "binary_metadata": {"sha256": sha, "size_bytes": 123}, + "parameters": {"config_overrides": {"rank_enabled": "false"}}, + "cleanup": {"requested": True, "removed": True}, + "cases": [case], + } + + +class SummarizeBenchmarkResultsTest(unittest.TestCase): + def test_quality_failure_blocks_acceptance_even_with_high_speedup(self) -> None: + case = { + "passed": False, + "canonical_graph": {"equal": False}, + "oracles": {"passed": True}, + "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, + } + row = SUMMARY.summarize_group("latest-rank-off", [report(case)]) + self.assertEqual(row["decision"], "REJECT: quality/correctness") + self.assertEqual(row["canonical"], "0/1") + self.assertEqual(row["speedup_p50"], 10.0) + + def test_aggregate_reports_p50_p95_peak_rss_and_cleanup(self) -> None: + reports = [] + for elapsed, speedup, peak in ((10, 10.0, 90), (20, 5.0, 110), (30, 3.0, 100)): + case = { + "passed": True, + "canonical_graph": {"equal": True}, + "oracles": {"passed": True}, + "incremental": {"elapsed_ms": elapsed, "peak_rss_mb": peak}, + "fresh_fast_full_after_change": {"elapsed_ms": 100, "peak_rss_mb": 120}, + "speedup_full_rebuild_over_incremental": speedup, + } + reports.append(report(case)) + row = SUMMARY.summarize_group("latest-rank-off", reports) + self.assertEqual(row["incremental_p50_ms"], 20.0) + self.assertEqual(row["incremental_p95_ms"], 30.0) + self.assertEqual(row["peak_rss_mb"], 120) + self.assertEqual(row["cleanup"], "3/3") + self.assertEqual(row["decision"], "PASS") + + def test_markdown_places_quality_before_performance(self) -> None: + case = { + "passed": True, + "canonical_graph": {"equal": True}, + "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, + } + markdown = SUMMARY.render_markdown([SUMMARY.summarize_group("latest", [report(case)])]) + header = markdown.splitlines()[2] + self.assertLess(header.index("Decision"), header.index("Speedup p50")) + self.assertIn("Binary SHA-256", markdown) + + +if __name__ == "__main__": + unittest.main()