From e210645206098e194f8ccbd4ce2c1be2f3248538 Mon Sep 17 00:00:00 2001 From: Eric Sutphen Date: Sat, 5 Sep 2026 15:17:10 -0400 Subject: [PATCH] fix: certification corpus selection supports fresh-capture snapshots The #162B/#162C spent corpora derive from one snapshot (239 proposals, digest 1d1de9f7...). A fresh certification capture has a different snapshot identity and, with a different snapshot key, non-comparable HMAC sample IDs. Selection now: allows the fresh capture, requires the prior snapshot to resolve spent content hashes, excludes and proves disjointness on BOTH sample IDs and content hashes (a re-keyed capture cannot launder a spent case back in), and records the identity basis in the manifest. The freeze gate validates the basis. --- evals/admission/certification/__main__.py | 3 + evals/admission/certification/select.py | 128 +++++++++++++++++----- tests/test_certification_162d.py | 66 ++++++++++- 3 files changed, 168 insertions(+), 29 deletions(-) diff --git a/evals/admission/certification/__main__.py b/evals/admission/certification/__main__.py index 69aaaeb..29ac67b 100644 --- a/evals/admission/certification/__main__.py +++ b/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,6 +90,7 @@ 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, @@ -96,6 +98,7 @@ def main() -> int: seed=args.seed, code_sha=args.code_sha, snapshot_key=key, + prior_snapshot=prior, frozen_at=None, ) write_private(args.output, manifest) diff --git a/evals/admission/certification/select.py b/evals/admission/certification/select.py index 49bf037..d015193 100644 --- a/evals/admission/certification/select.py +++ b/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,15 +25,12 @@ 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]: @@ -32,30 +38,68 @@ def _load_json(path: Path) -> dict[str, Any]: 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,15 +121,19 @@ 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). @@ -93,30 +141,54 @@ def select_certification_corpus( 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") diff --git a/tests/test_certification_162d.py b/tests/test_certification_162d.py index bd9c6cd..12f5a29 100644 --- a/tests/test_certification_162d.py +++ b/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