test(benchmark): retain resource and quality evidence

Parse the maximum peak_mb value from mem.phase records in scripts/benchmark-incremental-speed.py and store the tested binary path, size, and SHA-256 in every synthetic, matrix, and self-dogfood report.

Add scripts/summarize-benchmark-results.py to aggregate repeated JSON artifacts into a quality-first Markdown table. Canonical graph or task-oracle failures force a rejection even when measured speedup is high; accepted rows include p50/p95 latency, median speedup, maximum peak RSS, cleanup status, config overrides, and binary identity.

uv run --no-project python -m unittest tests/test_benchmark_incremental_speed.py tests/test_summarize_benchmark_results.py passed 6/6 contracts. The summarizer --help invocation passed.

Signed-off-by: Andrew Hundt <ATHundt@gmail.com>

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,

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())

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()

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()