fix: harden benchmark gate scripts with input validation

Raise BenchmarkDataError for malformed JSON or missing keys; skip zero
baselines safely; reject non-positive --slack values.

Author: clean6378-max-it

scripts/check_benchmark_regression.py

@@ -9,20 +9,62 @@
 THRESHOLD = 1.20
 
 
+class BenchmarkDataError(ValueError):
+    """Raised when benchmark JSON input is malformed or missing required fields."""
+
+
 def load_results(results_path: str | Path) -> dict[str, float]:
-    data = json.loads(Path(results_path).read_text(encoding="utf-8"))
-    return {entry["name"]: float(entry["stats"]["mean"]) for entry in data["benchmarks"]}
+    path = Path(results_path)
+    try:
+        data = json.loads(path.read_text(encoding="utf-8"))
+    except json.JSONDecodeError as exc:
+        raise BenchmarkDataError(f"invalid JSON in {path}: {exc}") from exc
+    try:
+        benchmarks = data["benchmarks"]
+    except (KeyError, TypeError) as exc:
+        raise BenchmarkDataError(f"{path} missing top-level 'benchmarks' array") from exc
+    if not isinstance(benchmarks, list):
+        raise BenchmarkDataError(f"{path} 'benchmarks' must be an array")
+
+    results: dict[str, float] = {}
+    for index, entry in enumerate(benchmarks):
+        if not isinstance(entry, dict):
+            raise BenchmarkDataError(f"{path} benchmarks[{index}] must be an object")
+        try:
+            name = entry["name"]
+            mean = entry["stats"]["mean"]
+        except (KeyError, TypeError) as exc:
+            raise BenchmarkDataError(
+                f"{path} benchmarks[{index}] missing 'name' or 'stats.mean'"
+            ) from exc
+        results[str(name)] = float(mean)
+    return results
 
 
 def load_baseline_means(baselines_path: str | Path) -> dict[str, float]:
-    data = json.loads(Path(baselines_path).read_text(encoding="utf-8"))
+    path = Path(baselines_path)
+    try:
+        data = json.loads(path.read_text(encoding="utf-8"))
+    except json.JSONDecodeError as exc:
+        raise BenchmarkDataError(f"invalid JSON in {path}: {exc}") from exc
+    if not isinstance(data, dict):
+        raise BenchmarkDataError(f"{path} root value must be an object")
+
     groups = data.get("groups", data)
+    if not isinstance(groups, dict):
+        raise BenchmarkDataError(f"{path} missing 'groups' object")
+
     means: dict[str, float] = {}
-    for key, value in groups.items():
+    for group_name, value in groups.items():
         if not isinstance(value, dict):
             continue
         for name, mean in value.items():
-            means[name] = float(mean)
+            try:
+                means[str(name)] = float(mean)
+            except (TypeError, ValueError) as exc:
+                raise BenchmarkDataError(
+                    f"{path} groups[{group_name!r}][{name!r}] is not a numeric mean"
+                ) from exc
     return means
 
 
@@ -42,6 +84,9 @@ def check_regression(
         if cur is None:
             print(f"WARN: no current result for baseline {name!r}; skipping")
             continue
+        if base == 0:
+            print(f"WARN: baseline for {name!r} is zero; skipping ratio check")
+            continue
         ratio = cur / base
         tag = "FAIL" if ratio > threshold else "ok"
         print(f"[{tag}] {name}: {cur:.6f}s vs {base:.6f}s ({ratio:.2f}x)")

scripts/reduce_baselines.py

@@ -11,6 +11,13 @@
 GATED_GROUPS = ("parse", "export", "search")
 
 
+def _positive_float(value: str) -> float:
+    parsed = float(value)
+    if parsed <= 0:
+        raise argparse.ArgumentTypeError("slack must be greater than zero")
+    return parsed
+
+
 def reduce_baselines(
     raw_path: str | Path,
     out_path: str | Path,
@@ -43,9 +50,9 @@ def main(argv: list[str] | None = None) -> int:
     parser.add_argument("out_path", help="destination baselines.json path")
     parser.add_argument(
         "--slack",
-        type=float,
+        type=_positive_float,
         default=1.0,
-        help="multiply means by this factor (e.g. 1.25 when capturing on a faster host)",
+        help="multiply means by this factor (must be > 0)",
     )
     args = parser.parse_args(argv)
     reduce_baselines(args.raw_path, args.out_path, slack=args.slack)

tests/test_check_benchmark_regression.py

@@ -6,7 +6,7 @@
 
 import pytest
 
-from scripts.check_benchmark_regression import check_regression
+from scripts.check_benchmark_regression import BenchmarkDataError, check_regression, load_results
 
 
 def _write_results(path, benchmarks: list[dict]) -> None:
@@ -75,3 +75,33 @@ def test_within_threshold_passes(tmp_path) -> None:
     )
 
     assert check_regression(results, baselines) == 0
+
+
+def test_load_results_rejects_malformed_json(tmp_path) -> None:
+    path = tmp_path / "bad.json"
+    path.write_text("{not json", encoding="utf-8")
+    with pytest.raises(BenchmarkDataError, match="invalid JSON"):
+        load_results(path)
+
+
+def test_load_results_requires_benchmarks_array(tmp_path) -> None:
+    path = tmp_path / "results.json"
+    path.write_text("{}", encoding="utf-8")
+    with pytest.raises(BenchmarkDataError, match="'benchmarks' array"):
+        load_results(path)
+
+
+def test_zero_baseline_skips_ratio_check(tmp_path, capsys: pytest.CaptureFixture[str]) -> None:
+    results = tmp_path / "results.json"
+    baselines = tmp_path / "baselines.json"
+    _write_results(
+        results,
+        [{"name": "test_parse_session_small", "stats": {"mean": 0.0002}}],
+    )
+    _write_baselines(
+        baselines,
+        {"parse": {"test_parse_session_small": 0.0}},
+    )
+
+    assert check_regression(results, baselines) == 0
+    assert "baseline for 'test_parse_session_small' is zero" in capsys.readouterr().out