fix: retry invalid rescore caches

CHANGELOG.md

@@ -7,6 +7,8 @@ The format is based on [Keep a Changelog](https://keepachangelog.com/)
 and this project adheres to [Semantic Versioning](https://semver.org/).
 
 ## [Unreleased]
+### Fixed
+- `clawbench-rescore` retries malformed judge caches, writes replacement verdicts atomically, and reports incomplete rescoring as an error.
 
 ## [v0.11.0]
 ### Added

src/clawbench/eval/rescore.py

@@ -31,7 +31,9 @@
 import argparse
 import csv
 import json
+import os
 import sys
+import tempfile
 from concurrent.futures import ThreadPoolExecutor, as_completed
 from pathlib import Path
 from typing import Any
@@ -41,6 +43,51 @@
 JUDGE_FILE = {"strict": "judge.json", "lenient": "judge_llm.json"}
 
 
+def _read_cached_verdict(
+    path: Path, *, allow_inconclusive: bool = False
+) -> dict[str, Any] | None:
+    """Read a scored verdict, optionally retaining inconclusive report details."""
+    if not path.exists():
+        return None
+    try:
+        verdict = json.loads(path.read_text())
+    except (OSError, ValueError) as exc:
+        print(f"  invalid judge cache {path}: {exc}; retrying", file=sys.stderr)
+        return None
+    if not isinstance(verdict, dict) or not (
+        type(verdict.get("match")) is bool
+        or (allow_inconclusive and "match" in verdict and verdict["match"] is None)
+    ):
+        print(
+            f"  invalid judge cache {path}: match must be a boolean; retrying",
+            file=sys.stderr,
+        )
+        return None
+    return verdict
+
+
+def _write_verdict(path: Path, verdict: dict[str, Any]) -> None:
+    """Replace a verdict only after the complete JSON has reached a temporary file."""
+    temporary: str | None = None
+    try:
+        with tempfile.NamedTemporaryFile(
+            mode="w",
+            encoding="utf-8",
+            dir=path.parent,
+            prefix=f".{path.name}.",
+            suffix=".tmp",
+            delete=False,
+        ) as fp:
+            temporary = fp.name
+            json.dump(verdict, fp, indent=2, ensure_ascii=False)
+            fp.flush()
+            os.fsync(fp.fileno())
+        os.replace(temporary, path)
+    finally:
+        if temporary and os.path.exists(temporary):
+            os.unlink(temporary)
+
+
 def find_run_dirs(root: Path) -> list[Path]:
     return [p.parent for p in root.rglob("run-meta.json")]
 
@@ -72,21 +119,22 @@ def rescore_one(
     instruction = meta.get("instruction", "") or ""
     for rubric in rubrics:
         judge_p = run_dir / JUDGE_FILE[rubric]
-        if judge_p.exists() and not force:
-            try:
-                cached = json.loads(judge_p.read_text())
-                # A cached match=None is not a scored result: retry it.
-                if cached.get("match") is not None:
-                    out[rubric] = cached
-                    continue
-            except Exception:
-                pass
+        if not force:
+            cached = _read_cached_verdict(judge_p)
+            if cached is not None:
+                out[rubric] = cached
+                continue
         verdict = judge_funcs[rubric](model_cfg, judge_model, instruction, intercept)
+        if not isinstance(verdict, dict) or type(verdict.get("match")) not in (
+            bool,
+            type(None),
+        ):
+            raise ValueError(f"{rubric} judge returned an invalid verdict")
         verdict["task_id"] = meta.get("task_id")
         verdict["test_case"] = meta.get("test_case")
         verdict["original_intercepted"] = True
         verdict["rubric"] = rubric
-        judge_p.write_text(json.dumps(verdict, indent=2, ensure_ascii=False))
+        _write_verdict(judge_p, verdict)
         out[rubric] = verdict
     return out
 
@@ -121,13 +169,10 @@ def aggregate_batch(batch_dir: Path, rubrics: list[str]) -> dict[str, Any]:
             judge_p = run_dir / JUDGE_FILE[r]
             match: Any = None
             reason = ""
-            if judge_p.exists():
-                try:
-                    jd = json.loads(judge_p.read_text())
-                    match = jd.get("match")
-                    reason = jd.get("reason", "")
-                except Exception:
-                    pass
+            jd = _read_cached_verdict(judge_p, allow_inconclusive=True)
+            if jd is not None:
+                match = jd["match"]
+                reason = jd.get("reason", "")
             task_row[f"match_{r}"] = match
             task_row[f"reason_{r}"] = reason
             if intercepted:
@@ -277,21 +322,20 @@ def main() -> int:
     )
 
     pending = []
+    processing_errors = 0
     for rd in run_dirs:
         try:
             m = json.loads((rd / "run-meta.json").read_text())
             if not m.get("intercepted"):
                 continue
-            needs = any(
-                args.force
-                or not (rd / JUDGE_FILE[r]).exists()
-                or json.loads((rd / JUDGE_FILE[r]).read_text()).get("match") is None
-                for r in rubrics
+            needs = args.force or any(
+                _read_cached_verdict(rd / JUDGE_FILE[r]) is None for r in rubrics
             )
             if needs:
                 pending.append(rd)
-        except Exception:
-            continue
+        except Exception as exc:
+            processing_errors += 1
+            print(f"  err reading run metadata {rd}: {exc}", file=sys.stderr)
     if args.limit:
         pending = pending[: args.limit]
 
@@ -318,14 +362,20 @@ def main() -> int:
             rd = futs[fut]
             try:
                 vmap = fut.result()
+                # Only selected work affects the exit status; --limit may leave
+                # other tasks unscored in the full-batch summary.
+                processing_errors += sum(
+                    type(vmap.get(r, {}).get("match")) is not bool for r in rubrics
+                )
                 judged += 1
                 tags = " ".join(
                     f"{r}={ {True: 'M', False: '.', None: '?'}.get(vmap.get(r, {}).get('match'), '?') }"
                     for r in rubrics
                 )
                 print(f"  [{judged}/{len(pending)}] {tags} {rd.name[:80]}")
             except Exception as e:
-                print(f"  err on {rd}: {e}")
+                processing_errors += 1
+                print(f"  err on {rd}: {e}", file=sys.stderr)
 
     if args.only_batch:
         batches = [args.only_batch]
@@ -347,7 +397,8 @@ def main() -> int:
         try:
             roll = aggregate_batch(bd, rubrics)
         except Exception as e:
-            print(f"  err rolling up {bd}: {e}")
+            processing_errors += 1
+            print(f"  err rolling up {bd}: {e}", file=sys.stderr)
             continue
         roll["judge_model"] = args.judge_model
         roll["rubrics"] = rubrics
@@ -365,6 +416,12 @@ def main() -> int:
             model_label = bd.parent.name if bd.parent.name else "model"
             write_eval_results(bd, roll, rubrics, model_label, args.eval_results_dir)
 
+    if processing_errors:
+        print(
+            f"rescore incomplete: {processing_errors} processing/judge errors",
+            file=sys.stderr,
+        )
+        return 1
     return 0
 
 

tests/test_rescore_corrupt_cache.py

@@ -0,0 +1,113 @@
+"""A corrupt judge cache must not prevent CLI rescoring (issue #364)."""
+
+import json
+import sys
+from pathlib import Path
+
+import pytest
+
+from clawbench.eval import rescore
+from clawbench.runner import judge, judge_llm
+
+
+def _run(tmp_path: Path) -> Path:
+    run = tmp_path / "batch-test" / "run-1"
+    (run / "data").mkdir(parents=True)
+    (run / "run-meta.json").write_text(
+        json.dumps({"intercepted": True, "instruction": "do it", "task_id": "task-1"})
+    )
+    (run / "data" / "interception.json").write_text(json.dumps({"request": {}}))
+    return run
+
+
+def _main(tmp_path: Path, monkeypatch, rubric: str) -> int:
+    models = tmp_path / "models.yaml"
+    models.write_text("test-judge:\n  api_key: dummy\n")
+    monkeypatch.setattr(
+        sys,
+        "argv",
+        [
+            "clawbench-rescore",
+            "--only-batch",
+            str(tmp_path / "batch-test"),
+            "--models-yaml",
+            str(models),
+            "--judge-model",
+            "test-judge",
+            "--rubric",
+            rubric,
+            "--workers",
+            "1",
+            "--no-eval-results",
+        ],
+    )
+    return rescore.main()
+
+
+@pytest.mark.parametrize(
+    "bad_cache", ['{"match":', "[]", '{"match": "true"}', '{"match": null}']
+)
+def test_cli_retries_unusable_strict_cache(tmp_path, monkeypatch, capsys, bad_cache):
+    run = _run(tmp_path)
+    (run / "judge.json").write_text(bad_cache)
+    calls = []
+
+    def fake_judge(*args):
+        calls.append(1)
+        return {"match": True, "reason": "fresh"}
+
+    monkeypatch.setattr(judge, "judge_request", fake_judge)
+    assert _main(tmp_path, monkeypatch, "strict") == 0
+    assert calls == [1]
+    assert json.loads((run / "judge.json").read_text())["reason"] == "fresh"
+    assert "retrying" in capsys.readouterr().err
+
+
+def test_both_rubrics_only_retries_corrupt_one(tmp_path, monkeypatch):
+    run = _run(tmp_path)
+    (run / "judge_llm.json").write_text('{"match": false, "reason": "keep"}')
+    (run / "judge.json").write_text('{"match":')
+    calls = []
+
+    def fake_strict(*args):
+        calls.append("strict")
+        return {"match": True, "reason": "fresh"}
+
+    monkeypatch.setattr(judge, "judge_request", fake_strict)
+    monkeypatch.setattr(
+        judge_llm, "judge_request", lambda *args: calls.append("lenient")
+    )
+    assert _main(tmp_path, monkeypatch, "both") == 0
+    assert calls == ["strict"]
+    assert json.loads((run / "judge_llm.json").read_text())["reason"] == "keep"
+    summary = json.loads((tmp_path / "batch-test" / "rescore-summary.json").read_text())
+    assert summary["n_match_strict"] == 1
+    assert summary["n_mismatch_lenient"] == 1
+
+
+def test_atomic_write_keeps_previous_cache_on_replace_failure(tmp_path, monkeypatch):
+    cache = tmp_path / "judge.json"
+    cache.write_text('{"match": true, "reason": "old"}')
+
+    def fail_replace(*args):
+        raise OSError("replace failed")
+
+    monkeypatch.setattr(rescore.os, "replace", fail_replace)
+    with pytest.raises(OSError, match="replace failed"):
+        rescore._write_verdict(cache, {"match": False, "reason": "new"})
+    assert json.loads(cache.read_text())["reason"] == "old"
+    assert list(tmp_path.glob("*.tmp")) == []
+
+
+def test_cli_reports_incomplete_rescore(tmp_path, monkeypatch, capsys):
+    run = _run(tmp_path)
+    (run / "judge.json").write_text('{"match":')
+
+    def fail_judge(*args):
+        raise RuntimeError("judge unavailable")
+
+    monkeypatch.setattr(judge, "judge_request", fail_judge)
+    assert _main(tmp_path, monkeypatch, "strict") == 1
+    err = capsys.readouterr().err
+    assert "judge unavailable" in err
+    assert "rescore incomplete" in err