"""Release-level adversarial checks for the E3 v3 rollout report.""" import copy from datetime import datetime, timezone import hashlib import json import os import sys import tempfile sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) from scripts.release_audit import ( # noqa: E402 validate_rollout_replay_attestation, validate_rollout_report, ) from scripts.rollout_metrics import ( # noqa: E402 NEAR_TIE_MAX_ULPS, V3_ATTESTATION_PROVENANCE_SCOPE, V3_INSTRUMENT_VERSION, V3_PAIR_PAYLOAD_DOMAIN, V3_REPLAY_REFERENCE, V3_REPLAY_RULE, V3_REPLAY_VERIFICATION_ARGV, V3_REPLAY_VERIFIER_METHOD, V3_REPORT_SCHEMA_VERSION, canonical_json_sha256, metric_value, paired_mean_difference_ci, select_policy, summarize_policy_v3, token_ids_sha256, validate_cross_policy_trajectories, ) from scripts.test_rollout_replay import ( # noqa: E402 PROMPT, VOCAB_SIZE, make_row, resign_trace, ) RUNTIME = { "python": "3.14.6", "mlx": "0.32.0", "numpy": "2.5.1", "tokenizers": "0.22.2", } REPO_ROOT = os.path.dirname(os.path.dirname(os.path.abspath(__file__))) EXECUTION_ARGV = [ "scripts/eval_rollout.py", "--ckpt", "fixture-checkpoint", "--receipt", "fixture-receipt.json", "--out", "fixture-report.json", ] def _file_sha256(path): digest = hashlib.sha256() with open(path, "rb") as f: for chunk in iter(lambda: f.read(1024 * 1024), b""): digest.update(chunk) return digest.hexdigest() def _identity_manifest_sha256(rows): encoded = json.dumps( rows, sort_keys=True, separators=(",", ":") ).encode("utf-8") return hashlib.sha256(encoded).hexdigest() def _endpoint( specification, adaptive_rows, fixed_rows, adaptive_policy, fixed_policy, ): adaptive = [ metric_value(row, specification["metric"]) for row in adaptive_rows ] fixed = [ metric_value(row, specification["metric"]) for row in fixed_rows ] difference, lo, hi = paired_mean_difference_ci( adaptive, fixed, n_boot=specification["bootstrap_samples"], seed=specification["bootstrap_seed"], ) if lo > 0: verdict = "POSITIVE" elif hi < 0: verdict = "NEGATIVE" else: verdict = "NULL: the interval includes 0" return { "comparison": specification["comparison"], "metric": specification["metric"], "adaptive_policy": adaptive_policy, "fixed_policy": fixed_policy, "difference": difference, "ci95": [lo, hi], "documents": len(adaptive_rows), "verdict": verdict, } def _rows(policy_name, payloads, seed): rows = [] for local_index, payload in enumerate(payloads): row = make_row( [3, 4, 5], [3, 4, 5], pair_index=payload["index"], policy=policy_name, ) row["seed"] = seed + local_index if policy_name.startswith("adaptive_h"): row["rollout"]["policy"] = "adaptive" row["rollout"]["entropy_threshold"] = float( policy_name.removeprefix("adaptive_h") ) rows.append(row) return rows def build_v3_rollout_fixture(root): """Build a small, fully valid report without invoking MLX.""" fixture_root = os.path.join(root, "release-rollout-v3") os.makedirs(fixture_root, exist_ok=True) tokenizer_path = os.path.join(fixture_root, "tokenizer.json") holdout_path = os.path.join(fixture_root, "holdout.jsonl") with open(tokenizer_path, "w", encoding="utf-8") as f: json.dump({"fixture": "tokenizer"}, f) with open(holdout_path, "w", encoding="utf-8") as f: f.write('{"fixture":"holdout"}\n') outputs = [[3, 4, 5] for _ in range(4)] payloads = [] identities = [] for index, target in enumerate(outputs): identity = { "index": index, "document_id": f"doc-{index}", "decoy_document_id": f"decoy-{index}", "token_offset": index, "prompt_sha256": token_ids_sha256(PROMPT), "target_sha256": token_ids_sha256(target), } identities.append(identity) payloads.append({ **identity, "prompt_token_ids": list(PROMPT), "target_token_ids": list(target), }) identity_sha256 = _identity_manifest_sha256(identities) payload_sha256 = canonical_json_sha256( payloads, V3_PAIR_PAYLOAD_DOMAIN ) primary_spec = { "comparison": "selected_adaptive_minus_selected_fixed", "metric": "accepted_drafts_per_verification", "bootstrap_unit": "paired_target_document", "bootstrap_samples": 200, "bootstrap_seed": 0, "positive_when": "ci95_lower_gt_0", } companion_specs = [ { "comparison": "selected_adaptive_minus_selected_fixed", "metric": "output_tokens_per_target_forward", "bootstrap_unit": "paired_target_document", "bootstrap_samples": 200, "bootstrap_seed": 0, }, { "comparison": "selected_adaptive_minus_selected_fixed", "metric": "drafts_issued_per_output_token", "bootstrap_unit": "paired_target_document", "bootstrap_samples": 200, "bootstrap_seed": 0, }, { "comparison": "selected_adaptive_minus_selected_fixed", "metric": "draft_recursions_per_output_token", "bootstrap_unit": "paired_target_document", "bootstrap_samples": 200, "bootstrap_seed": 0, }, ] checkpoint = { "path": os.path.join(fixture_root, "fixture-checkpoint"), "step": 19073, "meta_sha256": "a" * 64, "master_sha256": "b" * 64, } registered_checkpoint = { "path": checkpoint["path"], "meta_sha256": checkpoint["meta_sha256"], "master_sha256": checkpoint["master_sha256"], } receipt = { "schema_version": 2, "instrument_version": V3_INSTRUMENT_VERSION, "registered_at": "2026-07-30T12:00:00Z", "checkpoint": registered_checkpoint, "execution_argv": list(EXECUTION_ARGV), "runtime_requirements": dict(RUNTIME), "model_vocab_size": VOCAB_SIZE, "acceptance_receipt": { "path": os.path.join(fixture_root, "acceptance-receipt.json"), "sha256": "f" * 64, }, "holdout": { "path": holdout_path, "sha256": _file_sha256(holdout_path), }, "tokenizer": { "path": tokenizer_path, "sha256": _file_sha256(tokenizer_path), }, "pair_settings": { "examples": len(payloads), "pair_manifest_sha256": identity_sha256, "pair_payload_manifest_sha256": payload_sha256, }, "split": { "calibration_documents": 2, "test_documents": 2, }, "policy": { "fixed_candidates": ["fixed_d2"], "adaptive_candidates": ["adaptive_h0.2"], "max_depth": 2, "selection_metric": "accepted_drafts_per_verification", }, "decoding": { "max_tokens": 3, "seed": 2718, }, "test_endpoint": primary_spec, "companion_endpoints": companion_specs, } receipt_evidence = { "path": os.path.join(fixture_root, "fixture-receipt.json"), "sha256": "c" * 64, "registered_at": "2026-07-30T12:00:00Z", "registration_git": { "commit": "e" * 40, "committed_at": "2026-07-30T12:00:00Z", "path": "config/eval_rollout_receipt_v3.json", "blob_sha256": "c" * 64, "origin_ref": "origin/main", }, "implementation": { "git_commit": "d" * 40, "source_files": {}, }, "checkpoint": registered_checkpoint, "execution_argv": list(EXECUTION_ARGV), "runtime_requirements": dict(RUNTIME), "prior_instruments": [], } calibration_payloads = payloads[:2] test_payloads = payloads[2:] candidate_order = ["fixed_d2", "adaptive_h0.2"] calibration_rows = { name: _rows(name, calibration_payloads, receipt["decoding"]["seed"]) for name in candidate_order } calibration_summaries = { name: summarize_policy_v3( rows, calibration_payloads, max_tokens=3, vocab_size=VOCAB_SIZE, ) for name, rows in calibration_rows.items() } selected_fixed = select_policy( calibration_summaries, receipt["policy"]["fixed_candidates"], receipt["policy"]["selection_metric"], ) selected_adaptive = select_policy( calibration_summaries, receipt["policy"]["adaptive_candidates"], receipt["policy"]["selection_metric"], ) selected_names = [ selected_fixed["policy"], selected_adaptive["policy"], ] test_rows = { name: _rows(name, test_payloads, receipt["decoding"]["seed"]) for name in selected_names } test_summaries = { name: summarize_policy_v3( rows, test_payloads, max_tokens=3, vocab_size=VOCAB_SIZE, ) for name, rows in test_rows.items() } calibration_trajectory = validate_cross_policy_trajectories( calibration_rows, candidate_order ) test_trajectory = validate_cross_policy_trajectories( test_rows, selected_names ) all_summaries = [ *calibration_summaries.values(), *test_summaries.values(), ] scored_documents = sum(item["documents"] for item in all_summaries) scored_tokens = sum(item["total_tokens"] for item in all_summaries) exact_tokens = sum( item["exact_argmax_tokens"] for item in all_summaries ) near_tie_tokens = sum( item["certified_near_tie_tokens"] for item in all_summaries ) branch_passes = sum( item["branch_replay_passes"] for item in all_summaries ) cached_exact = sum( item["cached_ar_exact_documents"] for item in all_summaries ) adaptive_test = test_rows[selected_adaptive["policy"]] fixed_test = test_rows[selected_fixed["policy"]] primary = _endpoint( primary_spec, adaptive_test, fixed_test, selected_adaptive["policy"], selected_fixed["policy"], ) companions = { item["metric"]: _endpoint( item, adaptive_test, fixed_test, selected_adaptive["policy"], selected_fixed["policy"], ) for item in companion_specs } now = datetime.now(timezone.utc).isoformat().replace("+00:00", "Z") report = { "schema_version": V3_REPORT_SCHEMA_VERSION, "instrument_version": V3_INSTRUMENT_VERSION, "publication_ready": True, "execution": { "started_at": now, "completed_at": now, "argv": list(EXECUTION_ARGV), "runtime": {**RUNTIME, "platform": "fixture-platform"}, }, "receipt": receipt_evidence, "checkpoint": checkpoint, "tokenizer": { "path": tokenizer_path, "sha256": _file_sha256(tokenizer_path), }, "holdout": { "path": holdout_path, "sha256": _file_sha256(holdout_path), "pair_count": len(payloads), "pair_manifest_sha256": identity_sha256, "pair_manifest": identities, "pair_payload_manifest_sha256": payload_sha256, "pair_payload_manifest": payloads, "decoy_match": {"matched": len(payloads)}, }, "calibration": { "documents": len(calibration_payloads), "candidate_order": candidate_order, "trajectory_identity": calibration_trajectory, "summaries": calibration_summaries, "selected_fixed": selected_fixed, "selected_adaptive": selected_adaptive, "rows": calibration_rows, }, "test": { "documents": len(test_payloads), "trajectory_identity": test_trajectory, "summaries": test_summaries, "rows": test_rows, }, "quality_gate": { "reference": V3_REPLAY_REFERENCE, "rule": V3_REPLAY_RULE, "near_tie_max_ulps": NEAR_TIE_MAX_ULPS, "scored_policy_documents": scored_documents, "scored_tokens": scored_tokens, "exact_argmax_tokens": exact_tokens, "certified_near_tie_tokens": near_tie_tokens, "failed_tokens": 0, "branch_replay_passes": branch_passes, "cross_policy_trajectory_matches": ( calibration_trajectory["matching_documents"] + test_trajectory["matching_documents"] ), "passed": True, }, "cached_ar_diagnostic": { "scored_policy_documents": scored_documents, "exact_output_matches": cached_exact, "different_cached_ar_branches": scored_documents - cached_exact, "claim_scope": ( "diagnostic_only; release quality is established by " "branch-local replay, not cached-AR byte identity" ), }, "primary_endpoint": primary, "secondary_target_forward_endpoint": companions[ "output_tokens_per_target_forward" ], "secondary_draft_issued_proxy_endpoint": companions[ "drafts_issued_per_output_token" ], "secondary_draft_work_endpoint": companions[ "draft_recursions_per_output_token" ], "wall_clock_note": ( "This reference recomputes full prefixes and has no rollback-capable " "KV cache. Wall time is recorded for audit, not claimed as deployment " "latency." ), "endpoint_scope_note": ( "The primary endpoint measures accepted drafts per verification. " "It does not establish verification-width cost or deployment " "latency. Draft recursions per output token is the registered " "drafter-work companion; issued drafts per output token is retained " "only as an issuance proxy. Target forwards exclude the added " "post-hoc branch-replay forward and independent verification pass." ), } return { "report": report, "receipt": receipt, "receipt_evidence": receipt_evidence, "checkpoint": checkpoint, "tokenizer_path": tokenizer_path, "holdout_path": holdout_path, } def build_replay_attestation_fixture(root, fixture): source_path = os.path.join(REPO_ROOT, "scripts", "verify_rollout_replay.py") source = { "path": "scripts/verify_rollout_replay.py", "bytes": os.path.getsize(source_path), "sha256": _file_sha256(source_path), } registration = { "method": V3_REPLAY_VERIFIER_METHOD, "execution_argv": list(V3_REPLAY_VERIFICATION_ARGV), "source": source, } fixture["receipt"]["independent_replay_verification"] = registration report = fixture["report"] rows = [] expected = [] calibration = report["calibration"] for local_index in range(calibration["documents"]): for policy in calibration["candidate_order"]: expected.append( ( "calibration", policy, calibration["rows"][policy][local_index], ) ) selected = [ calibration["selected_fixed"]["policy"], calibration["selected_adaptive"]["policy"], ] for local_index in range(report["test"]["documents"]): for policy in selected: expected.append( ( "test", policy, report["test"]["rows"][policy][local_index], ) ) aggregate = { "policy_documents": 0, "output_tokens": 0, "exact_argmax_tokens": 0, "certified_near_tie_tokens": 0, "failed_tokens": 0, "reproduced_policy_documents": 0, "failed_reproductions": 0, } for index, (split_name, policy, producer) in enumerate(expected): payload = report["holdout"]["pair_payload_manifest"][ producer["pair_index"] ] output = producer["output_token_ids"] rows.append({ "verification_index": index, "split": split_name, "policy": policy, "pair_index": producer["pair_index"], "document_id": producer["document_id"], "seed": producer["seed"], "prompt_sha256": producer["prompt_sha256"], "output_sha256": producer["output_sha256"], "reproduction": { "output_sha256": producer["output_sha256"], "trace_sha256": producer["trace_sha256"], "deterministic_stats_sha256": "d" * 64, "output_tokens_match": True, "generation_trace_matches": True, "deterministic_stats_match": True, }, "branch_quality": { "input_sha256": token_ids_sha256( payload["prompt_token_ids"] + output[:-1] ), "output_sha256": producer["output_sha256"], "reference_argmax_sha256": producer["output_sha256"], "aligned_logits_float32_le_sha256": "e" * 64, "exact_argmax_tokens": len(output), "certified_near_tie_tokens": 0, "failed_tokens": 0, "certified_near_ties": [], "failures": [], "passed": True, }, }) aggregate["policy_documents"] += 1 aggregate["output_tokens"] += len(output) aggregate["exact_argmax_tokens"] += len(output) aggregate["reproduced_policy_documents"] += 1 completed = report["execution"]["completed_at"] rollout_path = os.path.join(root, "rollout.registered.v3.json") receipt_path = os.path.join(root, "eval_rollout_receipt_v3.json") rollout_sha256 = "9" * 64 receipt_sha256 = fixture["report"]["receipt"]["sha256"] attestation = { "schema_version": 1, "instrument_version": V3_INSTRUMENT_VERSION, "publication_ready": True, "branch_quality_independently_verified": True, "rollout_execution_reproduced": True, "passed": True, "provenance_scope": V3_ATTESTATION_PROVENANCE_SCOPE, "verification": { "method": V3_REPLAY_VERIFIER_METHOD, "started_at": completed, "completed_at": completed, "argv": list(V3_REPLAY_VERIFICATION_ARGV), "runtime": {**RUNTIME, "platform": "fixture-platform"}, "timing_scope": ( "producer elapsed_seconds, tok_per_sec, and ar_tok_per_sec " "are excluded from deterministic reproduction" ), }, "time_order": { "registered_at": fixture["receipt"]["registered_at"], "registration_committed_at": fixture["report"]["receipt"][ "registration_git" ]["committed_at"], "report_started_at": report["execution"]["started_at"], "report_completed_at": completed, "verification_started_at": completed, "verification_completed_at": completed, }, "report": {"path": rollout_path, "sha256": rollout_sha256}, "receipt": { "path": receipt_path, "sha256": receipt_sha256, "registered_at": fixture["receipt"]["registered_at"], "registration_git": fixture["report"]["receipt"][ "registration_git" ], }, "checkpoint": fixture["checkpoint"], "verifier": { "method": V3_REPLAY_VERIFIER_METHOD, "execution_argv": list(V3_REPLAY_VERIFICATION_ARGV), "registered_source": source, "live_source": source, }, "frozen_inputs": { "acceptance_receipt": dict( fixture["receipt"]["acceptance_receipt"] ), "holdout": dict(fixture["receipt"]["holdout"]), "tokenizer": dict(fixture["receipt"]["tokenizer"]), "pair_count": fixture["receipt"]["pair_settings"]["examples"], "pair_manifest_sha256": fixture["receipt"]["pair_settings"][ "pair_manifest_sha256" ], "pair_payload_manifest_sha256": fixture["receipt"][ "pair_settings" ]["pair_payload_manifest_sha256"], }, "row_manifest_sha256": canonical_json_sha256( rows, b"WISP_E3_V3_INDEPENDENT_ROW_MANIFEST\0" ), "rows": rows, "aggregate": aggregate, } attestation_path = os.path.join(root, "rollout.replay-verification.v3.json") with open(attestation_path, "w", encoding="utf-8") as handle: json.dump(attestation, handle, sort_keys=True) return { "attestation": attestation, "path": attestation_path, "rollout_path": rollout_path, "rollout_sha256": rollout_sha256, "receipt_path": receipt_path, "receipt_sha256": receipt_sha256, } def _validate(fixture, report): return validate_rollout_report( report, fixture["receipt"], fixture["receipt_evidence"], fixture["checkpoint"], fixture["tokenizer_path"], fixture["holdout_path"], ) def _expect_reject(fixture, report, substring): try: _validate(fixture, report) except ValueError as error: assert substring in str(error), ( f"expected {substring!r} in {error!r}" ) else: raise AssertionError(f"mutation unexpectedly passed: {substring}") def main(): with tempfile.TemporaryDirectory() as root: fixture = build_v3_rollout_fixture(root) report = fixture["report"] result = _validate(fixture, report) assert result["branch_local_replay_valid"] is True assert result["cross_policy_trajectory_identical"] is True assert result["quality_equivalent_to_greedy_ar"] is False verification = build_replay_attestation_fixture(root, fixture) def validate_attestation(value): return validate_rollout_replay_attestation( value, verification["path"], verification["rollout_path"], verification["rollout_sha256"], verification["receipt_path"], verification["receipt_sha256"], fixture["receipt"], report, fixture["checkpoint"], REPO_ROOT, ) attestation_result = validate_attestation( verification["attestation"] ) assert ( attestation_result["branch_quality_independently_verified"] is True ) assert attestation_result["rollout_execution_reproduced"] is True for mutate, substring in ( ( lambda value: value.__setitem__( "rollout_execution_reproduced", False ), "did not pass", ), ( lambda value: value["report"].__setitem__( "sha256", "0" * 64 ), "different rollout report", ), ( lambda value: value.__setitem__( "provenance_scope", "externally trusted" ), "provenance scope", ), ( lambda value: value["rows"][0]["reproduction"].__setitem__( "generation_trace_matches", False ), "did not reproduce", ), ( lambda value: value["rows"][0]["branch_quality"].__setitem__( "failed_tokens", 1 ), "branch counts", ), ( lambda value: value.__setitem__( "row_manifest_sha256", "0" * 64 ), "row-manifest hash", ), ): bad_attestation = copy.deepcopy(verification["attestation"]) mutate(bad_attestation) try: validate_attestation(bad_attestation) except ValueError as error: assert substring in str(error), ( f"expected {substring!r} in {error!r}" ) else: raise AssertionError( f"mutated independent attestation passed: {substring}" ) copied = copy.deepcopy(report) copied["calibration"]["rows"]["fixed_d2"][1] = copy.deepcopy( copied["calibration"]["rows"]["fixed_d2"][0] ) _expect_reject(fixture, copied, "differs on pair_index") bad = copy.deepcopy(report) bad["holdout"]["pair_payload_manifest"][0]["target_sha256"] = "0" * 64 _expect_reject(fixture, bad, "differs on target_sha256") bad = copy.deepcopy(report) bad["calibration"]["rows"]["fixed_d2"][0][ "output_sha256" ] = "0" * 64 _expect_reject(fixture, bad, "output hash") bad = copy.deepcopy(report) row = bad["calibration"]["rows"]["fixed_d2"][0] row["generation_trace"]["cycles"][0]["verification"][ "base_row_index" ] += 1 resign_trace(row) _expect_reject(fixture, bad, "verification binding") bad = copy.deepcopy(report) row = bad["calibration"]["rows"]["fixed_d2"][0] row["generation_trace"]["cycles"][0]["verification"]["outcomes"][1][ "target_row_index" ] += 1 resign_trace(row) _expect_reject(fixture, bad, "off by one") bad = copy.deepcopy(report) row = bad["calibration"]["rows"]["fixed_d2"][0] row["generation_trace"]["cycles"][0][ "state_source" ] = "verification_reuse" resign_trace(row) _expect_reject(fixture, bad, "stale state") bad = copy.deepcopy(report) replay = bad["calibration"]["rows"]["fixed_d2"][0]["branch_replay"] replay["reference_argmax_token_ids"][-1] = 6 replay["reference_argmax_sha256"] = token_ids_sha256( replay["reference_argmax_token_ids"] ) _expect_reject(fixture, bad, "complete ordered mismatch set") bad = copy.deepcopy(report) payload = bad["holdout"]["pair_payload_manifest"][2] alternate = make_row( [3, 4, 6], [3, 4, 6], pair_index=payload["index"], policy="adaptive_h0.2", ) alternate["seed"] = 2718 alternate["target_sha256"] = payload["target_sha256"] alternate["target_position_accuracy"] = 2 / 3 alternate["target_common_prefix_tokens"] = 2 alternate["target_exact_match"] = False alternate["rollout"]["policy"] = "adaptive" alternate["rollout"]["entropy_threshold"] = 0.2 bad["test"]["rows"]["adaptive_h0.2"][0] = alternate _expect_reject(fixture, bad, "cross-policy trajectory differs") print("release rollout v3 audit: PASS") if __name__ == "__main__": main()