Harden resumable benchmark execution and evidence integrity

configs/experiments/benchmark-v01.yaml

@@ -6,6 +6,10 @@ corpus:
 dataset:
   path: datasets/regrag-br/
   split: test  # dev split is reserved for router few-shot examples (ADR-0003)
+  # Cost-controlled comparison: deterministic, query-class-stratified sample.
+  # Remove max_questions only for the final full-test publication run.
+  max_questions: 60
+  sampling_seed: regrag-br-benchmark-sample-v1
 strategies:
   - dense
   - sparse_bm25

docs/adr/0017-auditable-evidence-lineage.md

@@ -217,7 +217,17 @@ Artifacts SHALL use:
 4. atomic rename or transaction;
 5. checksum generation after all files close.
 
-Manifest status changes to `completed` only after checksum verification.
+`artifact_root_hash` covers every final artifact except `manifest.json` and
+`checksums.sha256`; excluding the manifest avoids a circular hash because the
+root is stored inside that manifest. `checksums.sha256` SHALL still cover the
+exact final `manifest.json` bytes. The checksum inventory is published before
+the final atomic manifest rename, so manifest status changes to `completed`
+only when the final inventory is ready.
+
+New and resumed auditable runs SHALL require a clean Git worktree, excluding
+only untracked generated output under `artifacts/`, `experiments/`, and
+`.ragforge/`. A resume SHALL preserve the original `started_at` and Git SHA
+and fail closed when the current commit or another manifest identity differs.
 
 A failed run remains inspectable.
 

scripts/verify_run.py

@@ -21,12 +21,14 @@
 import sys
 from pathlib import Path
 
+from ragforge.evaluation.canonical_hash import canonical_json_hash
 from ragforge.evaluation.event_log import compute_event_hash
 from ragforge.evaluation.lineage_ports import EventEnvelope
 from ragforge.ingestion.snapshot import snapshot_hash
 
 ROOT = Path(__file__).resolve().parents[1]
 ARTIFACTS_DIR = ROOT / "artifacts" / "runs"
+_MANIFEST_FILENAME = "manifest.json"
 
 
 def parse_args() -> argparse.Namespace:
@@ -36,6 +38,20 @@ def parse_args() -> argparse.Namespace:
     return parser.parse_args()
 
 
+def load_recorded_checksums(artifacts_dir: Path) -> dict[str, str]:
+    """Load the standard checksum inventory into a relative-path mapping."""
+    recorded: dict[str, str] = {}
+    checksums_path = artifacts_dir / "checksums.sha256"
+    for line in checksums_path.read_text(encoding="utf-8").splitlines():
+        if not line.strip():
+            continue
+        digest, separator, relative = line.partition("  ")
+        if not separator or not digest or not relative:
+            raise ValueError(f"invalid checksum line: {line!r}")
+        recorded[relative] = digest
+    return recorded
+
+
 def verify_checksums(artifacts_dir: Path) -> list[str]:
     """Return every problem found comparing checksums.sha256 against the files on disk.
 
@@ -49,12 +65,10 @@ def verify_checksums(artifacts_dir: Path) -> list[str]:
     if not checksums_path.exists():
         return [f"checksums.sha256 not found at {checksums_path}"]
 
-    recorded: dict[str, str] = {}
-    for line in checksums_path.read_text(encoding="utf-8").splitlines():
-        if not line.strip():
-            continue
-        digest, _, relative = line.partition("  ")
-        recorded[relative] = digest
+    try:
+        recorded = load_recorded_checksums(artifacts_dir)
+    except ValueError as exc:
+        return [str(exc)]
 
     for relative, digest in recorded.items():
         path = artifacts_dir / relative
@@ -152,6 +166,24 @@ def verify_manifest(artifacts_dir: Path) -> list[str]:
         problems.append(
             f"manifest.json declares strategies with no question artifacts: {sorted(missing)}"
         )
+    checksums_path = artifacts_dir / "checksums.sha256"
+    if checksums_path.exists():
+        try:
+            recorded = load_recorded_checksums(artifacts_dir)
+        except ValueError as exc:
+            problems.append(str(exc))
+        else:
+            root_inputs = {
+                relative: digest
+                for relative, digest in recorded.items()
+                if relative != _MANIFEST_FILENAME
+            }
+            expected_root_hash = canonical_json_hash(root_inputs)
+            if manifest.get("artifact_root_hash") != expected_root_hash:
+                problems.append(
+                    "manifest.json artifact_root_hash does not match the final artifact "
+                    "checksum inventory"
+                )
     return problems
 
 

src/ragforge/evaluation/answer_harness.py

@@ -12,6 +12,7 @@
 from concurrent.futures import CancelledError
 from dataclasses import dataclass
 from statistics import mean
+from typing import Protocol, runtime_checkable
 
 from ragforge.domain.models import Answer, Judgment, RetrievalResult
 from ragforge.domain.protocols import RetrievalStrategy
@@ -25,6 +26,13 @@
 _DEFAULT_MAX_WORKERS = 5
 
 
+@runtime_checkable
+class _ClosableJudge(Protocol):
+    """Optional lifecycle exposed by judges that own network clients."""
+
+    def close(self) -> None: ...
+
+
 @dataclass(frozen=True, slots=True)
 class AnswerEvaluationResult:
     """Aggregate answer-quality metrics plus one AnswerRecord per attempted judgment (ADR-0012).
@@ -64,16 +72,12 @@ def evaluate_answer_quality(
     then run concurrently across up to ``max_workers`` questions at once,
     since each question's Gemini calls are independent HTTP requests.
 
-    ``judge_factory`` builds one judge, not many: ``generator`` (a plain
-    synchronous HTTP client) is shared safely across worker threads, but
-    RagasJudge.evaluate() calls ragas's sync wrapper, which internally does
-    ``asyncio.run(self.ascore(...))`` - a fresh event loop per call, reusing
-    the same async client underneath. Calling that concurrently from
-    multiple threads against one shared judge corrupts its connection pool
-    (observed for real: thousands of leaked CLOSE_WAIT sockets and a stalled
-    run). Each worker thread instead lazily builds and caches its own judge
-    via ``judge_factory``, so no judge instance is ever touched from more
-    than one thread.
+    ``generator`` (a plain synchronous HTTP client) is shared safely across
+    worker threads. Each worker thread instead lazily builds and caches its
+    own judge via ``judge_factory`` so one RagasJudge and its persistent
+    asyncio event loop are never touched concurrently from multiple threads.
+    Judges exposing ``close()`` are closed after the worker pool finishes,
+    including when scoring raises.
 
     A retrieval, generation, or judge-scoring failure for one question is
     counted in "answer_errors" and excluded from the averages rather than
@@ -143,12 +147,16 @@ def evaluate_answer_quality(
         retrieved.append((judgment, results))
 
     thread_local = threading.local()
+    judges: list[AnswerQualityJudge] = []
+    judges_lock = threading.Lock()
 
     def _judge_for_this_thread() -> AnswerQualityJudge:
         judge = getattr(thread_local, "judge", None)
         if judge is None:
             judge = judge_factory()
             thread_local.judge = judge
+            with judges_lock:
+                judges.append(judge)
         return judge
 
     def _score_one(
@@ -197,12 +205,17 @@ def _on_scoring_result(
             consecutive_scoring_errors += 1
             return consecutive_scoring_errors >= _MAX_CONSECUTIVE_ERRORS
 
-        outcomes = run_bounded(
-            retrieved,
-            lambda pair: _score_one(pair[0], pair[1]),
-            max_workers=max_workers,
-            on_result=_on_scoring_result,
-        )
+        try:
+            outcomes = run_bounded(
+                retrieved,
+                lambda pair: _score_one(pair[0], pair[1]),
+                max_workers=max_workers,
+                on_result=_on_scoring_result,
+            )
+        finally:
+            for judge in judges:
+                if isinstance(judge, _ClosableJudge):
+                    judge.close()
         for (judgment, _results), outcome in zip(retrieved, outcomes, strict=True):
             if isinstance(outcome, BaseException):
                 is_cancelled = isinstance(outcome, CancelledError)

src/ragforge/evaluation/artifact_writer.py

@@ -8,6 +8,7 @@
 covers a partially-written file.
 """
 
+from collections.abc import Mapping
 from pathlib import Path
 
 from ragforge.ingestion.snapshot import snapshot_hash
@@ -26,7 +27,11 @@ def write_atomic(path: Path, content: str) -> None:
     tmp_path.replace(path)
 
 
-def compute_checksums(root: Path) -> dict[str, str]:
+def compute_checksums(
+    root: Path,
+    *,
+    excluded_paths: frozenset[str] = frozenset(),
+) -> dict[str, str]:
     """Return ``{relative_posix_path: sha256_hex}`` for every regular file under ``root``.
 
     The checksums file itself (if already present from a prior partial
@@ -37,19 +42,23 @@ def compute_checksums(root: Path) -> dict[str, str]:
         if not path.is_file():
             continue
         relative = path.relative_to(root).as_posix()
-        if relative == _CHECKSUMS_FILENAME:
+        if relative == _CHECKSUMS_FILENAME or relative in excluded_paths:
             continue
         checksums[relative] = snapshot_hash(path)
     return checksums
 
 
-def write_checksums_file(root: Path) -> None:
+def write_checksums_file(
+    root: Path,
+    checksums: Mapping[str, str] | None = None,
+) -> None:
     """Write ``root/checksums.sha256`` in standard ``sha256sum`` format (verifiable externally).
 
-    Every existing file under ``root`` at call time is included - callers
-    should call this only after every other artifact for the run has been
-    written.
+    With no explicit mapping, every existing file under ``root`` at call
+    time is included. A precomputed mapping supports evidence finalization,
+    where the inventory for the intended final manifest is published before
+    that manifest's last atomic rename.
     """
-    checksums = compute_checksums(root)
-    lines = [f"{digest}  {relative}" for relative, digest in sorted(checksums.items())]
+    resolved_checksums = compute_checksums(root) if checksums is None else checksums
+    lines = [f"{digest}  {relative}" for relative, digest in sorted(resolved_checksums.items())]
     write_atomic(root / _CHECKSUMS_FILENAME, "\n".join(lines) + "\n" if lines else "")

src/ragforge/evaluation/event_log.py

@@ -60,6 +60,31 @@ def __init__(self, run_id: str, path: Path) -> None:
         self._lock = threading.Lock()
         self._sequence = 0
         self._previous_event_hash: str | None = None
+        self._restore_tail()
+
+    def _restore_tail(self) -> None:
+        """Continue an existing run's sequence and hash chain when resuming."""
+        if not self._path.exists():
+            return
+        lines = self._path.read_text(encoding="utf-8").splitlines()
+        if not lines:
+            return
+        payload: object = json.loads(lines[-1])
+        if not isinstance(payload, dict):
+            raise ValueError(f"last event in {self._path} must be a JSON object")
+        stored_run_id = payload.get("run_id")
+        sequence = payload.get("sequence")
+        event_hash = payload.get("event_hash")
+        if stored_run_id != self._run_id:
+            raise ValueError(
+                f"event log run_id mismatch: expected {self._run_id!r}, found {stored_run_id!r}"
+            )
+        if isinstance(sequence, bool) or not isinstance(sequence, int) or sequence < 1:
+            raise ValueError(f"last event in {self._path} has an invalid sequence")
+        if not isinstance(event_hash, str) or not event_hash:
+            raise ValueError(f"last event in {self._path} has an invalid event_hash")
+        self._sequence = sequence
+        self._previous_event_hash = event_hash
 
     def emit(
         self,