feat: add immutable evaluation manifests

Author: ReverseZoom2151

src/utils/evaluation_manifest.py

@@ -0,0 +1,51 @@
+"""Immutable episode manifests for trustworthy held-out evaluation."""
+
+from __future__ import annotations
+
+import hashlib
+import json
+from pathlib import Path
+
+from src.utils.data_recorder import load_npz_shard, shard_checksum
+
+
+def create_evaluation_manifest(data_dir: str | Path, destination: str | Path, limit: int | None = None) -> Path:
+    """Select complete recorded episodes and bind them to exact source checksums."""
+    data_dir, destination = Path(data_dir), Path(destination)
+    metadata_path = data_dir / "metadata.json"
+    metadata = json.loads(metadata_path.read_text(encoding="utf-8")) if metadata_path.exists() else {}
+    selected, sources = [], {}
+    for shard in sorted(data_dir.glob("shard_*.npz")):
+        checksum = shard_checksum(shard)
+        episodes = load_npz_shard(shard, metadata.get("shard_checksums", {}).get(shard.name))
+        sources[shard.name] = checksum
+        for episode in episodes:
+            selected.append({
+                "shard": shard.name,
+                "episode_id": episode["episode_id"],
+                "env_id": episode["env_id"],
+                "length": episode["length"],
+                "scenario": episode.get("metadata", {}).get("scenario", "unknown"),
+            })
+            if limit is not None and len(selected) >= limit:
+                break
+        if limit is not None and len(selected) >= limit:
+            break
+    if not selected:
+        raise ValueError("no validated recording episodes are available for evaluation")
+    payload = {"format_version": 1, "episodes": selected, "source_shards": sources}
+    payload["manifest_sha256"] = hashlib.sha256(json.dumps(payload, sort_keys=True).encode()).hexdigest()
+    destination.parent.mkdir(parents=True, exist_ok=True)
+    destination.write_text(json.dumps(payload, indent=2, sort_keys=True), encoding="utf-8")
+    return destination
+
+
+def validate_evaluation_manifest(path: str | Path, data_dir: str | Path) -> dict:
+    """Reject a manifest when any referenced recording shard has changed."""
+    payload = json.loads(Path(path).read_text(encoding="utf-8"))
+    if payload.get("format_version") != 1 or not payload.get("episodes"):
+        raise ValueError("invalid evaluation manifest")
+    for name, expected in payload.get("source_shards", {}).items():
+        if shard_checksum(Path(data_dir) / name) != expected:
+            raise ValueError(f"evaluation source checksum mismatch: {name}")
+    return payload

tests/test_evaluation_manifest.py

@@ -0,0 +1,20 @@
+import numpy as np
+import pytest
+
+from src.utils.data_recorder import EpisodeRecorder
+from src.utils.evaluation_manifest import create_evaluation_manifest, validate_evaluation_manifest
+
+
+def test_evaluation_manifest_is_episode_based_and_checksum_bound(tmp_path):
+    data = tmp_path / "recordings"
+    recorder = EpisodeRecorder(str(data), save_frequency=1)
+    frame = np.zeros((2, 2, 3), dtype=np.uint8)
+    recorder.add_transition(frame, 0, 0.0, frame, terminated=True, metadata={"scenario": "level-a"})
+    recorder.finalize()
+    manifest = create_evaluation_manifest(data, tmp_path / "evaluation.json")
+    payload = validate_evaluation_manifest(manifest, data)
+    assert payload["episodes"][0]["scenario"] == "level-a"
+    shard = next(data.glob("shard_*.npz"))
+    shard.write_bytes(shard.read_bytes() + b"changed")
+    with pytest.raises(ValueError, match="checksum"):
+        validate_evaluation_manifest(manifest, data)