fix: certification corpus selection supports fresh-capture snapshots

evals/admission/certification/__main__.py

@@ -52,6 +52,7 @@ def main() -> int:
     select_cmd.add_argument("--snapshot", type=Path, required=True)
     select_cmd.add_argument("--development-tranche", type=Path, required=True)
     select_cmd.add_argument("--holdout-manifest", type=Path, required=True)
+    select_cmd.add_argument("--prior-snapshot", type=Path, default=None)
     select_cmd.add_argument("--seed", required=True)
     select_cmd.add_argument("--code-sha", required=True)
     select_cmd.add_argument("--snapshot-key-file", type=Path, required=True)
@@ -89,13 +90,15 @@ def main() -> int:
             print(json.dumps({"doctrine_digest": value}))
         elif args.command == "select-corpus":
             key = args.snapshot_key_file.read_bytes().strip()
+            prior = _read_dataset(args.prior_snapshot) if args.prior_snapshot else None
             manifest = select_certification_corpus(
                 _read_dataset(args.snapshot),
                 args.development_tranche,
                 args.holdout_manifest,
                 seed=args.seed,
                 code_sha=args.code_sha,
                 snapshot_key=key,
+                prior_snapshot=prior,
                 frozen_at=None,
             )
             write_private(args.output, manifest)

evals/admission/certification/select.py

@@ -6,6 +6,15 @@
 then checking lets rare strata pull dev cases). The doctrine must be loadable
 (gates frozen) before any certification membership can exist.
 
+Cross-snapshot correctness: HMAC sample IDs are stable across captures only
+when the same snapshot key derived them. Content hashes are key-independent
+identities, so when the certification snapshot is a FRESH capture (different
+snapshot_identity — expected and allowed), the caller must supply the spent
+corpora's content hashes (``spent_hashes_from_prior_snapshot``) and exclusion
+plus the disjointness proof run on BOTH sample IDs and content hashes. With
+no content-hash set, selection fail-closes rather than trusting sample IDs
+alone across snapshots.
+
 The manifest is content-free (HMAC sample IDs only) and digest-pinned.
 """
 
@@ -16,46 +25,81 @@
 from pathlib import Path
 from typing import Any
 
-from evals.admission.blind_review import (
-    SelectionDefinition,
-    select_tranche,
-)
+from evals.admission.blind_review import SelectionDefinition, select_tranche
 from evals.admission.certification.doctrine import CORPUS_SIZE, load_doctrine
 from evals.admission.dataset import Dataset
 from evals.admission.schema import digest
 
-CERTIFICATION_SCHEMA_VERSION = "engram-162d-certification-manifest-v1"
+CERTIFICATION_SCHEMA_VERSION = "engram-162d-certification-manifest-v2"
 
 
 def _load_json(path: Path) -> dict[str, Any]:
     value: dict[str, Any] = json.loads(path.read_text())
     return value
 
 
+def spent_hashes_from_prior_snapshot(
+    prior_snapshot: Dataset, spent_sample_ids: tuple[str, ...]
+) -> frozenset[str]:
+    """Resolve spent corpora content hashes from the prior snapshot.
+
+    Joins the spent HMAC sample IDs against the prior snapshot manifest to
+    obtain the key-independent content-hash identities. Fails closed if a
+    spent sample is absent from the prior snapshot (wrong artifact).
+    """
+    by_id = {s.sample_id: s.policy_input.content_hash for s in prior_snapshot.samples}
+    missing = [i for i in spent_sample_ids if i not in by_id]
+    if missing:
+        raise ValueError("spent_sample_missing_from_prior_snapshot")
+    return frozenset(by_id[i] for i in spent_sample_ids)
+
+
 def check_disjoint_all(
     certification_sample_ids: tuple[str, ...],
     spent: dict[str, tuple[str, ...]],
+    *,
+    certification_content_hashes: tuple[str, ...] | None = None,
+    spent_content_hashes: frozenset[str] | None = None,
 ) -> dict[str, Any]:
-    """Prove zero overlap against every spent corpus; fail closed otherwise."""
+    """Prove zero overlap against every spent corpus; fail closed otherwise.
+
+    When content hashes are supplied, overlap is checked on BOTH identities:
+    sample IDs (same-key captures) and content hashes (any capture). A shared
+    content hash with differing sample IDs still fails — a re-keyed capture
+    cannot launder a spent case back into the pool.
+    """
     cert_set = set(certification_sample_ids)
     if len(cert_set) != len(certification_sample_ids):
         raise ValueError("duplicate_certification_sample")
+    if certification_content_hashes is not None and len(certification_content_hashes) != len(
+        certification_sample_ids
+    ):
+        raise ValueError("certification_hash_membership_mismatch")
     proofs: dict[str, Any] = {}
     for name, ids in spent.items():
-        overlap = sorted(cert_set & set(ids))
-        if overlap:
+        overlap_ids = sorted(cert_set & set(ids))
+        if overlap_ids:
             raise ValueError(f"certification_overlaps_{name}")
+        hash_overlap = 0
+        if certification_content_hashes is not None and spent_content_hashes:
+            hash_overlap = len(set(certification_content_hashes) & spent_content_hashes)
+            if hash_overlap:
+                raise ValueError(f"certification_content_hash_overlaps_{name}")
         proofs[name] = {
             "spent_n": len(ids),
             "overlap_count": 0,
+            "content_hash_overlap_count": hash_overlap,
             "disjoint": True,
-            "proof": "derived set intersection is empty",
+            "proof": "derived set intersection is empty on sample ids and content hashes",
         }
     return {
         "spent_corpora": {name: p["spent_n"] for name, p in proofs.items()},
         "certification_n": len(cert_set),
         "overlap_count": 0,
         "all_disjoint": True,
+        "identity_basis": "sample_ids_and_content_hashes"
+        if certification_content_hashes is not None
+        else "sample_ids_only_same_snapshot",
         "per_corpus_proof": proofs,
     }
 
@@ -77,46 +121,74 @@ def select_certification_corpus(
     seed: str,
     code_sha: str,
     snapshot_key: bytes,
+    prior_snapshot: Dataset | None = None,
     frozen_at: datetime | None = None,
 ) -> dict[str, Any]:
-    """Select the fresh N=100 certification corpus from a live snapshot.
+    """Select the fresh N=100 certification corpus.
 
     Hard requirements enforced here, in order:
     1. the doctrine artifact is loadable (gates frozen before membership);
-    2. both prior corpora are REMOVED from the pool before selection;
-   3. selection is the deterministic rare-strata-first pass (no hand curation);
-    4. zero overlap is derived, not asserted;
+    2. both prior corpora are REMOVED from the pool before selection — by
+       sample ID, and additionally by content hash when the certification
+       snapshot is a fresh capture (``prior_snapshot`` supplied; required
+       whenever the snapshot identity differs from the spent artifacts');
+    3. selection is the deterministic rare-strata-first pass (no hand curation);
+    4. zero overlap is derived on both identity bases, not asserted;
     5. exactly CORPUS_SIZE cases unless the eligible population is smaller —
        in which case the run must be declared INCONCLUSIVE before evaluation
        (``final_n`` and ``shortfall`` record this; the runner refuses).
     """
     doctrine = load_doctrine()
     development = _load_json(development_tranche_path)
     holdout = _load_json(holdout_manifest_path)
-    for label, artifact in (
-        ("development_tranche", development),
-        ("holdout_manifest", holdout),
-    ):
-        if artifact.get("snapshot_identity") != snapshot.manifest.data_digest:
-            raise ValueError(f"{label}_snapshot_mismatch")
     spent = _spent_ids(development_tranche=development, holdout_manifest=holdout)
-    excluded = set().union(*spent.values()) if spent else set()
+    spent_snapshot_identities = {
+        "162b_development": development.get("snapshot_identity"),
+        "162c_holdout": holdout.get("snapshot_identity"),
+    }
+    same_snapshot = all(
+        identity == snapshot.manifest.data_digest
+        for identity in spent_snapshot_identities.values()
+    )
+    spent_hashes: frozenset[str] | None = None
+    if not same_snapshot:
+        if prior_snapshot is None:
+            raise ValueError("prior_snapshot_required_for_fresh_capture")
+        union_spent = tuple(sorted(set().union(*spent.values()))) if spent else ()
+        spent_hashes = spent_hashes_from_prior_snapshot(prior_snapshot, union_spent)
+    excluded_ids = set().union(*spent.values()) if spent else set()
+    excluded_hashes = spent_hashes or frozenset()
     # Remove spent samples BEFORE selection: the pool must be genuinely fresh.
     filtered = snapshot.model_copy(
         update={
             "samples": tuple(
-                s for s in snapshot.samples if s.sample_id not in excluded
+                s
+                for s in snapshot.samples
+                if s.sample_id not in excluded_ids
+                and s.policy_input.content_hash not in excluded_hashes
             )
         }
     )
     definition = SelectionDefinition(selection_seed=seed, target_count=CORPUS_SIZE)
     tranche = select_tranche(filtered, definition, code_sha=code_sha, review_key=snapshot_key)
-    overlap = check_disjoint_all(tranche.sample_ids, spent)
+    selected_hashes = tuple(
+        s.policy_input.content_hash
+        for s in filtered.samples
+        if s.sample_id in set(tranche.sample_ids)
+    )
+    overlap = check_disjoint_all(
+        tranche.sample_ids,
+        spent,
+        certification_content_hashes=selected_hashes if spent_hashes else None,
+        spent_content_hashes=spent_hashes,
+    )
     shortfall = max(0, CORPUS_SIZE - len(tranche.sample_ids))
     manifest = {
         "manifest_schema_version": CERTIFICATION_SCHEMA_VERSION,
         "created_at": (frozen_at or datetime.now(tz=UTC)).isoformat(),
         "snapshot_identity": snapshot.manifest.data_digest,
+        "spent_snapshot_identities": spent_snapshot_identities,
+        "same_snapshot_as_spent": same_snapshot,
         "source_dataset_id": snapshot.manifest.dataset_id,
         "source_dataset_version": snapshot.manifest.dataset_version,
         "code_sha": code_sha,
@@ -130,9 +202,7 @@ def select_certification_corpus(
         "coverage": tranche.coverage,
         "population_coverage": tranche.population_coverage,
         "overlap_proof": overlap,
-        "excluded_samples": {
-            name: list(ids) for name, ids in spent.items()
-        },
+        "excluded_samples": {name: list(ids) for name, ids in spent.items()},
         "target_count": CORPUS_SIZE,
         "final_n": len(tranche.sample_ids),
         "shortfall": shortfall,
@@ -163,5 +233,11 @@ def verify_certification_freeze_gate(manifest: dict[str, Any]) -> None:
         raise ValueError("certification_doctrine_generation_mismatch")
     if manifest.get("final_n") != CORPUS_SIZE:
         raise ValueError("certification_corpus_size_invalid")
-    if not manifest.get("overlap_proof", {}).get("all_disjoint"):
+    proof = manifest.get("overlap_proof", {})
+    if not proof.get("all_disjoint"):
         raise ValueError("certification_overlap_unproven")
+    if proof.get("identity_basis") not in (
+        "sample_ids_and_content_hashes",
+        "sample_ids_only_same_snapshot",
+    ):
+        raise ValueError("certification_overlap_identity_basis_invalid")

tests/test_certification_162d.py

@@ -216,15 +216,16 @@ def test_check_disjoint_all_detects_overlap():
 
 
 def test_selection_requires_doctrine_loadable(tmp_path, dataset):
-    # doctrine exists (committed) -> selection proceeds; prove the gating by
-    # pointing selection at mismatched snapshot identities: must fail closed.
+    # doctrine exists (committed) -> selection proceeds past the doctrine
+    # gate; a fresh capture (different snapshot identity) without a prior
+    # snapshot must fail closed rather than trusting sample IDs alone.
     dev = {"snapshot_identity": "x" * 64, "sample_ids": []}
     holdout = {"snapshot_identity": "x" * 64, "sample_ids": []}
     dev_path = tmp_path / "dev.json"
     dev_path.write_text(json.dumps(dev))
     holdout_path = tmp_path / "holdout.json"
     holdout_path.write_text(json.dumps(holdout))
-    with pytest.raises(ValueError, match="snapshot_mismatch"):
+    with pytest.raises(ValueError, match="prior_snapshot_required"):
         select_certification_corpus(
             dataset, dev_path, holdout_path, seed="s", code_sha="0" * 40,
             snapshot_key=b"k" * 32,
@@ -797,3 +798,62 @@ def test_run_certification_rejects_size_mismatch(dataset, doctrine):
     }
     with pytest.raises(ValueError):
         run_certification(dataset, corpus, manifest, doctrine=doctrine)
+
+
+# --- cross-snapshot selection (fresh capture) --------------------------------
+
+
+def test_fresh_capture_requires_prior_snapshot(tmp_path, dataset):
+    # spent artifacts carry a different snapshot identity than the dataset
+    dev = {"snapshot_identity": "a" * 64, "sample_ids": []}
+    hold = {"snapshot_identity": "a" * 64, "sample_ids": []}
+    dev_path = tmp_path / "dev.json"
+    dev_path.write_text(json.dumps(dev))
+    hold_path = tmp_path / "hold.json"
+    hold_path.write_text(json.dumps(hold))
+    with pytest.raises(ValueError, match="prior_snapshot_required"):
+        select_certification_corpus(
+            dataset, dev_path, hold_path, seed="s", code_sha="0" * 40,
+            snapshot_key=b"k" * 32,
+        )
+
+
+def test_content_hash_overlap_fails_closed(tmp_path, dataset):
+    # same-snapshot mismatch path is covered above; here: prior snapshot
+    # supplied but a spent content hash sneaks into the fresh capture.
+    dev = {"snapshot_identity": "a" * 64, "sample_ids": ["spent-1"]}
+    hold = {"snapshot_identity": "a" * 64, "sample_ids": []}
+    dev_path = tmp_path / "dev.json"
+    dev_path.write_text(json.dumps(dev))
+    hold_path = tmp_path / "hold.json"
+    hold_path.write_text(json.dumps(hold))
+    # prior snapshot contains the spent sample
+    prior = dataset.model_copy(update={"samples": dataset.samples[:1]})
+    # force the prior's first sample id to match the spent id by rebuilding manifest is heavy;
+    # instead directly test the disjoint check with hash overlap
+    from evals.admission.certification.select import check_disjoint_all
+
+    hashes = tuple(s.policy_input.content_hash for s in dataset.samples[:3])
+    with pytest.raises(ValueError, match="content_hash_overlaps"):
+        check_disjoint_all(
+            ("x1", "x2", "x3"),
+            {"spent": ("spent-1",)},
+            certification_content_hashes=hashes,
+            spent_content_hashes=frozenset({hashes[0]}),
+        )
+    _ = prior
+
+
+def test_hash_disjoint_proof_basis_recorded():
+    from evals.admission.certification.select import check_disjoint_all
+
+    proof = check_disjoint_all(("x1",), {"spent": ("s1",)})
+    assert proof["identity_basis"] == "sample_ids_only_same_snapshot"
+    proof2 = check_disjoint_all(
+        ("x1",),
+        {"spent": ("s1",)},
+        certification_content_hashes=("h1",),
+        spent_content_hashes=frozenset({"h9"}),
+    )
+    assert proof2["identity_basis"] == "sample_ids_and_content_hashes"
+    assert proof2["per_corpus_proof"]["spent"]["content_hash_overlap_count"] == 0