"""CPU mutation checks for the final release audit.""" import copy import json import os import sys import tempfile sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) sys.path.insert(0, os.path.dirname(os.path.abspath(__file__))) from compare_acceptance import compare_reports # noqa: E402 from hf_metadata import file_sha256 # noqa: E402 from hf_metadata import render_evaluation_section, render_model_card # noqa: E402 from release_audit import ( # noqa: E402 checkpoint_identity, load_json_snapshot, validate_ablation_arm_checkpoint, validate_acceptance_bundle, validate_acceptance_receipt_freshness, validate_e2_report_sampler, validate_export_model_card, validate_export_sourced_from_checkpoint, validate_external_verification_paths, validate_no_tampering, validate_rollout_report, validate_training_data_receipt_registration, validate_validation_report, ) from rollout_metrics import ( # noqa: E402 NEAR_TIE_MAX_ULPS, bf16_ulp, metric_value, paired_mean_difference_ci, select_policy, summarize_policy, ) from test_release_rollout_v3 import build_v3_rollout_fixture # noqa: E402 from validation_metrics import summarize_validation # noqa: E402 def acceptance_report( role, receipt_sha256, fim_value, shuf_value, verdict, checkpoint, ): metadata = [ { "document_id": f"target-{index}", "decoy_document_id": f"decoy-{index}", "token_offset": index * 10, } for index in range(4) ] metrics = [ { **identity, "acceptance": { "l2r": [0.4, 0.4], "fim": [fim_value, fim_value], "fim_shuf": [shuf_value, shuf_value], }, "mean_nll": { "l2r": 5.0, "fim": 4.8, "fim_shuf": 5.1, }, "mean_entropy": { "l2r": 4.0, "fim": 3.8, "fim_shuf": 4.1, }, } for identity in metadata ] ratio = fim_value / shuf_value ci95 = [ratio, ratio] step = 19073 if role == "trained" else 0 return { "schema_version": 1, "instrument_version": 4, "analysis_role": role, "publication_ready": True, "publication_contract": { "receipt_sha256": receipt_sha256, "role": role, }, "checkpoint": { "step": step, "meta_sha256": ( checkpoint["meta_sha256"] if role == "trained" else "d" * 64 ), "master_sha256": ( checkpoint["master_sha256"] if role == "trained" else "e" * 64 ), }, "tokenizer": {"sha256": "b" * 64}, "holdout": {"sha256": "a" * 64}, "seed": 0, "examples": 4, "depth": 2, "step": step, "dtype": "bfloat16", "temperature": 1.0, "top_k": 0, "bins": 4, "span_len": 64, "prefix_len": 256, "suffix_len": 128, "pair_metadata": metadata, "document_metrics": metrics, "mean_nll": { "l2r": 5.0, "fim": 4.8, "fim_shuf": 5.1, }, "ratios": { "fim_over_shuf": [ {"depth": 1, "ratio": ratio}, { "depth": 2, "ratio": ratio, "ci95": ci95, "documents": 4, }, ], }, "primary_endpoint": { "depth": 2, "ratio": ratio, "ci95": ci95, "comparison": "fim_over_shuf", "uniform_nll": 10.39720770839918, "l2r_nll_clears_floor": True, "verdict": verdict, }, } def rollout_rows(policy, manifest, accepted, target_forwards): return [ { "document_id": identity["document_id"], "decoy_document_id": identity["decoy_document_id"], "token_offset": identity["token_offset"], "policy": policy, "prompt_sha256": identity["prompt_sha256"], "target_sha256": identity["target_sha256"], "ar_output_sha256": "f" * 64, "output_sha256": "f" * 64, "output_matches_ar": True, "divergence": None, "accepted_drafts": accepted[index], "verification_forwards": 10, "tokens": 64, "target_forwards": target_forwards[index], "elapsed_seconds": 1.0 + index * 0.1, } for index, identity in enumerate(manifest) ] def main(): checkpoint = { "path": "/tmp/final-checkpoint", "step": 19073, "meta_sha256": "1" * 64, "master_sha256": "2" * 64, "optimizer_sha256": "3" * 64, } sampler_evidence = { "ablation_data_index_sha256": "a" * 64, "final_train_sampler": { "index_sha256": "a" * 64, "batches_drawn_since_reset": 300368, "rng_state_sha256": "b" * 64, }, "ablation_data_artifacts": [ { "kind": "shard", "split": "train", "path": "/tmp/train_0000.bin", "sha256": "c" * 64, }, ], } ablation_sampling_report = { "publication_contract": copy.deepcopy(sampler_evidence) } validate_e2_report_sampler( ablation_sampling_report, sampler_evidence, ) changed_sampling_report = copy.deepcopy(ablation_sampling_report) changed_sampling_report["publication_contract"][ "ablation_data_index_sha256" ] = "0" * 64 try: validate_e2_report_sampler( changed_sampling_report, sampler_evidence, ) except ValueError as exc: assert "different final sampler evidence" in str(exc) else: raise AssertionError("a mismatched final sampler passed release audit") changed_data_artifacts = copy.deepcopy(ablation_sampling_report) changed_data_artifacts["publication_contract"][ "ablation_data_artifacts" ][0]["sha256"] = "0" * 64 try: validate_e2_report_sampler( changed_data_artifacts, sampler_evidence, ) except ValueError as exc: assert "different final sampler evidence" in str(exc) else: raise AssertionError("altered no-FIM data evidence passed release audit") with tempfile.TemporaryDirectory() as temp_dir: malformed_json_path = os.path.join(temp_dir, "malformed.json") with open(malformed_json_path, "wb") as f: f.write(b"{not valid json") try: load_json_snapshot(malformed_json_path) except ValueError as exc: assert "invalid JSON" in str(exc) else: raise AssertionError("malformed JSON passed load_json_snapshot") non_object_json_path = os.path.join(temp_dir, "non_object.json") with open(non_object_json_path, "wb") as f: f.write(b"[1, 2, 3]") try: load_json_snapshot(non_object_json_path) except ValueError as exc: assert "top-level JSON value must be an object" in str(exc) else: raise AssertionError( "a top-level JSON array passed load_json_snapshot" ) artifacts = {} for name in ("config", "index", "validation", "tokenizer", "holdout"): path = os.path.join(temp_dir, name) with open(path, "wb") as f: f.write(name.encode("utf-8")) artifacts[name] = path settings = { "bootstrap_samples": 200, "bootstrap_seed": 3, } batch_hashes = [str(index) * 64 for index in range(4)] validation_receipt = { "instrument_version": 1, "settings": settings, "batch_manifest": {"target_tokens": 32768}, } validation_evidence = { "sha256": "4" * 64, "registered_at": "registered", "batch_manifest": {"batch_sha256": batch_hashes}, } validation_rows = [ { "batch": index, "sha256": batch_hashes[index], "total": 5.4 + index * 0.1, "main": 5.0 + index * 0.1, "mtp": [4.8 + index * 0.1, 4.9 + index * 0.1], } for index in range(4) ] validation_report = { "schema_version": 1, "instrument_version": 1, "publication_ready": True, "receipt": validation_evidence, "checkpoint": checkpoint, "config": {"sha256": file_sha256(artifacts["config"])}, "data_index": {"sha256": file_sha256(artifacts["index"])}, "validation_shard": { "sha256": file_sha256(artifacts["validation"]) }, "settings": settings, "elapsed_seconds": 10.0, "summary": summarize_validation( validation_rows, n_boot=200, seed=3 ), "batches": validation_rows, } validation_result = validate_validation_report( validation_report, validation_receipt, validation_evidence, checkpoint, artifacts["config"], artifacts["index"], artifacts["validation"], ) assert validation_result["target_tokens"] == 32768 mutated_validation = copy.deepcopy(validation_report) mutated_validation["summary"]["main_loss"]["mean"] += 1.0 try: validate_validation_report( mutated_validation, validation_receipt, validation_evidence, checkpoint, artifacts["config"], artifacts["index"], artifacts["validation"], ) except ValueError as exc: assert "does not recompute" in str(exc) else: raise AssertionError("a mutated validation summary passed release audit") def _expect_reject(mutated_report, expected_substring): try: validate_validation_report( mutated_report, validation_receipt, validation_evidence, checkpoint, artifacts["config"], artifacts["index"], artifacts["validation"], ) except ValueError as exc: assert expected_substring in str(exc), ( f"expected {expected_substring!r} in {exc}" ) else: raise AssertionError( f"a report that should have failed on " f"{expected_substring!r} passed release audit" ) bad = copy.deepcopy(validation_report) bad["schema_version"] = 2 _expect_reject(bad, "schema is not 1") bad = copy.deepcopy(validation_report) bad["instrument_version"] = 999 _expect_reject(bad, "instrument version differs") bad = copy.deepcopy(validation_report) bad["publication_ready"] = False _expect_reject(bad, "not publication-ready") bad = copy.deepcopy(validation_report) bad["receipt"]["sha256"] = "9" * 64 _expect_reject(bad, "receipt hash differs") bad = copy.deepcopy(validation_report) bad["receipt"]["registered_at"] = "a-different-time" _expect_reject(bad, "registration time differs") bad = copy.deepcopy(validation_report) bad["receipt"]["batch_manifest"] = { "batch_sha256": ["f" * 64] * len(batch_hashes) } _expect_reject(bad, "batch manifest differs") for key in ("step", "meta_sha256", "master_sha256", "optimizer_sha256"): bad = copy.deepcopy(validation_report) bad["checkpoint"] = copy.deepcopy(checkpoint) bad["checkpoint"][key] = "tampered" if key != "step" else -1 _expect_reject(bad, f"validation checkpoint {key} differs") for key in ("config", "data_index", "validation_shard"): bad = copy.deepcopy(validation_report) bad[key] = {"sha256": "0" * 64} _expect_reject(bad, f"validation {key} artifact hash differs") bad = copy.deepcopy(validation_report) bad["settings"] = dict(settings) bad["settings"]["bootstrap_seed"] = 999 _expect_reject(bad, "settings differ") bad = copy.deepcopy(validation_report) bad["batches"] = validation_rows[:-1] _expect_reject(bad, "row count differs") bad = copy.deepcopy(validation_report) bad["batches"] = copy.deepcopy(validation_rows) bad["batches"][1]["sha256"] = "f" * 64 _expect_reject(bad, "batch evidence differs at index 1") bad = copy.deepcopy(validation_report) bad["elapsed_seconds"] = -1.0 _expect_reject(bad, "elapsed time is not finite and positive") control_contract = { "metric": "trained_minus_untrained_fim_over_shuf_ratio", "depth": 2, "bootstrap_unit": "paired_target_document", "bootstrap_samples": 4000, "bootstrap_seed": 0, "clears_control_when": "ci95_lower_gt_0", } acceptance_receipt_sha256 = "5" * 64 acceptance_receipt = { "schema_version": 2, "clean_holdout": {"sha256": "a" * 64}, "pair_readiness": {"tokenizer_sha256": "b" * 64}, "analysis_contract": { "instrument_version": 4, "control_adjustment": control_contract, "trained_checkpoint_step": 19073, "untrained_control": {"step": 0}, "model_args": {"vocab_size": 32768}, "settings": { "seed": 0, "examples": 4, "depth": 2, "dtype": "bfloat16", "temperature": 1.0, "top_k": 0, "bins": 4, "span_len": 64, "prefix_len": 256, "suffix_len": 128, }, }, } trained = acceptance_report( "trained", acceptance_receipt_sha256, 0.65, 0.5, "POSITIVE", checkpoint, ) control = acceptance_report( "untrained-control", acceptance_receipt_sha256, 0.5, 0.5, "CONTROL: no hypothesis verdict", checkpoint, ) trained_sha256 = "6" * 64 control_sha256 = "7" * 64 comparison = compare_reports( trained, control, acceptance_receipt, acceptance_receipt_sha256, ) comparison["inputs"] = { "trained": {"sha256": trained_sha256}, "control": {"sha256": control_sha256}, "receipt": {"sha256": acceptance_receipt_sha256}, } acceptance_result = validate_acceptance_bundle( trained, trained_sha256, control, control_sha256, comparison, acceptance_receipt, acceptance_receipt_sha256, checkpoint, ) assert ( acceptance_result["combined_interpretation"] == "POSITIVE_AND_CLEARS_CONTROL" ) null_trained = acceptance_report( "trained", acceptance_receipt_sha256, 0.5, 0.5, "NULL: the interval includes 1.0", checkpoint, ) null_comparison = compare_reports( null_trained, control, acceptance_receipt, acceptance_receipt_sha256, ) null_comparison["inputs"] = { "trained": {"sha256": trained_sha256}, "control": {"sha256": control_sha256}, "receipt": {"sha256": acceptance_receipt_sha256}, } null_result = validate_acceptance_bundle( null_trained, trained_sha256, control, control_sha256, null_comparison, acceptance_receipt, acceptance_receipt_sha256, checkpoint, ) assert null_result["combined_interpretation"] == "PRIMARY_NULL" mutated_comparison = copy.deepcopy(comparison) mutated_comparison["trained_minus_untrained_ratio"]["difference"] = 9.0 try: validate_acceptance_bundle( trained, trained_sha256, control, control_sha256, mutated_comparison, acceptance_receipt, acceptance_receipt_sha256, checkpoint, ) except ValueError as exc: assert "does not recompute" in str(exc) else: raise AssertionError("a mutated acceptance comparison passed audit") # Negative control for the anti-contamination check: found missing by # actually disabling `validate_acceptance_bundle`'s "untrained control # uses the final trained weights" assertion in release_audit.py and # observing that this suite still passed. The fixture's control # report had always used a hardcoded dummy master_sha256 different # from the trained checkpoint by construction, so the check's # condition was trivially true whether the assertion existed or not; # no fixture ever exercised its failure path. This constructs that # exact scenario: a control report whose checkpoint is the trained # checkpoint's own weights. contaminated_control = copy.deepcopy(control) contaminated_control["checkpoint"]["master_sha256"] = checkpoint[ "master_sha256" ] try: validate_acceptance_bundle( trained, trained_sha256, contaminated_control, control_sha256, comparison, acceptance_receipt, acceptance_receipt_sha256, checkpoint, ) except ValueError as exc: assert "final trained weights" in str(exc) else: raise AssertionError( "an untrained control sharing the trained checkpoint's " "weights passed audit" ) for key in ("trained", "control", "receipt"): bad_comparison = copy.deepcopy(comparison) bad_comparison["inputs"][key]["sha256"] = "0" * 64 try: validate_acceptance_bundle( trained, trained_sha256, control, control_sha256, bad_comparison, acceptance_receipt, acceptance_receipt_sha256, checkpoint, ) except ValueError as exc: assert f"acceptance comparison {key} input hash differs" in str(exc) else: raise AssertionError( f"an acceptance comparison with a tampered {key} input " "hash passed audit" ) # "step" is deliberately excluded: compare_reports.validate_report # already rejects a trained report whose checkpoint step doesn't # match the acceptance receipt's registered step, so a step mutation # trips that earlier, unrelated check before reaching release_audit's # own checkpoint-identity comparison below. Only meta_sha256 and # master_sha256 are checked there merely for well-formedness (64 hex # chars), leaving room for a value that passes compare_reports but # still differs from the true final checkpoint. for key in ("meta_sha256", "master_sha256"): bad_trained = copy.deepcopy(trained) bad_trained["checkpoint"][key] = "f" * 64 try: validate_acceptance_bundle( bad_trained, trained_sha256, control, control_sha256, comparison, acceptance_receipt, acceptance_receipt_sha256, checkpoint, ) except ValueError as exc: assert ( f"trained acceptance checkpoint {key} differs" in str(exc) ) else: raise AssertionError( f"a trained acceptance report with a tampered {key} " "checkpoint field passed audit" ) manifest = [ { "index": index, "document_id": f"doc-{index}", "decoy_document_id": f"decoy-{index}", "token_offset": index * 10, "prompt_sha256": f"{index + 1:x}" * 64, "target_sha256": f"{index + 6:x}" * 64, } for index in range(5) ] manifest_sha256 = __import__("hashlib").sha256( json.dumps( manifest, sort_keys=True, separators=(",", ":") ).encode("utf-8") ).hexdigest() rollout_receipt = { "instrument_version": 2, "pair_settings": { "examples": 5, "pair_manifest_sha256": manifest_sha256, }, "split": { "calibration_documents": 2, "test_documents": 3, }, "policy": { "fixed_candidates": ["fixed_d1", "fixed_d2"], "adaptive_candidates": ["adaptive_h0.2", "adaptive_h0.3"], "selection_metric": "accepted_drafts_per_verification", }, "test_endpoint": { "comparison": "selected_adaptive_minus_selected_fixed", "metric": "accepted_drafts_per_verification", "bootstrap_samples": 200, "bootstrap_seed": 0, }, } rollout_evidence = { "sha256": "8" * 64, "registered_at": "registered", } calibration_manifest = manifest[:2] test_manifest = manifest[2:] calibration_rows = { "fixed_d1": rollout_rows( "fixed_d1", calibration_manifest, [5, 5], [20, 20] ), "fixed_d2": rollout_rows( "fixed_d2", calibration_manifest, [10, 10], [18, 18] ), "adaptive_h0.2": rollout_rows( "adaptive_h0.2", calibration_manifest, [8, 8], [17, 17] ), "adaptive_h0.3": rollout_rows( "adaptive_h0.3", calibration_manifest, [12, 12], [16, 16] ), } calibration_summaries = { name: summarize_policy(rows) for name, rows in calibration_rows.items() } selected_fixed = select_policy( calibration_summaries, rollout_receipt["policy"]["fixed_candidates"], "accepted_drafts_per_verification", ) selected_adaptive = select_policy( calibration_summaries, rollout_receipt["policy"]["adaptive_candidates"], "accepted_drafts_per_verification", ) test_rows = { "fixed_d2": rollout_rows( "fixed_d2", test_manifest, [8, 9, 10], [20, 20, 20] ), "adaptive_h0.3": rollout_rows( "adaptive_h0.3", test_manifest, [11, 12, 13], [16, 16, 16] ), } near_tie_max = 6.1875 near_tie_ulp = bf16_ulp(near_tie_max) certified_divergence = { "position": 10, "ar_token": 2776, "policy_token": 5187, "ar_token_logit": near_tie_max, "policy_token_logit": near_tie_max - near_tie_ulp, "row_max_logit": near_tie_max, "ulp_at_max": near_tie_ulp, "max_ulps": NEAR_TIE_MAX_ULPS, "ar_token_deficit_ulps": 0.0, "policy_token_deficit_ulps": 1.0, } test_rows["fixed_d2"][1]["output_matches_ar"] = False test_rows["fixed_d2"][1]["output_sha256"] = "e" * 64 test_rows["fixed_d2"][1]["divergence"] = certified_divergence test_summaries = { name: summarize_policy(rows) for name, rows in test_rows.items() } adaptive_values = [ metric_value(row, "accepted_drafts_per_verification") for row in test_rows["adaptive_h0.3"] ] fixed_values = [ metric_value(row, "accepted_drafts_per_verification") for row in test_rows["fixed_d2"] ] difference, lo, hi = paired_mean_difference_ci( adaptive_values, fixed_values, n_boot=200, seed=0, ) secondary_adaptive = [ metric_value(row, "output_tokens_per_target_forward") for row in test_rows["adaptive_h0.3"] ] secondary_fixed = [ metric_value(row, "output_tokens_per_target_forward") for row in test_rows["fixed_d2"] ] secondary_difference, secondary_lo, secondary_hi = ( paired_mean_difference_ci( secondary_adaptive, secondary_fixed, n_boot=200, seed=0, ) ) rollout_report = { "schema_version": 1, "instrument_version": 2, "publication_ready": True, "receipt": rollout_evidence, "checkpoint": checkpoint, "tokenizer": {"sha256": file_sha256(artifacts["tokenizer"])}, "holdout": { "sha256": file_sha256(artifacts["holdout"]), "pair_count": 5, "pair_manifest_sha256": manifest_sha256, "pair_manifest": manifest, "decoy_match": {"matched": 5}, }, "calibration": { "documents": 2, "candidate_order": [ "fixed_d1", "fixed_d2", "adaptive_h0.2", "adaptive_h0.3", ], "summaries": calibration_summaries, "selected_fixed": selected_fixed, "selected_adaptive": selected_adaptive, "rows": calibration_rows, }, "test": { "documents": 3, "summaries": test_summaries, "rows": test_rows, }, "quality_gate": { "reference": "greedy_ar", "rule": "exact_token_match_or_certified_near_tie", "near_tie_max_ulps": NEAR_TIE_MAX_ULPS, "exact_ar_matches": 13, "certified_divergences": 1, "passed": True, }, "primary_endpoint": { "comparison": "selected_adaptive_minus_selected_fixed", "metric": "accepted_drafts_per_verification", "adaptive_policy": "adaptive_h0.3", "fixed_policy": "fixed_d2", "difference": difference, "ci95": [lo, hi], "documents": 3, "verdict": "POSITIVE", }, "secondary_target_forward_endpoint": { "metric": "output_tokens_per_target_forward", "difference": secondary_difference, "ci95": [secondary_lo, secondary_hi], "documents": 3, }, } rollout_result = validate_rollout_report( rollout_report, rollout_receipt, rollout_evidence, checkpoint, artifacts["tokenizer"], artifacts["holdout"], ) assert rollout_result["primary_endpoint"]["verdict"] == "POSITIVE" assert rollout_result["quality_equivalent_to_greedy_ar"] is False assert rollout_result["branch_local_replay_valid"] is False assert rollout_result["historical_instrument_only"] is True def _expect_rollout_reject(mutated_report, expected_substring): try: validate_rollout_report( mutated_report, rollout_receipt, rollout_evidence, checkpoint, artifacts["tokenizer"], artifacts["holdout"], ) except ValueError as exc: assert expected_substring in str(exc), ( f"expected {expected_substring!r} in {exc}" ) else: raise AssertionError( f"a rollout report that should have failed on " f"{expected_substring!r} passed release audit" ) bad = copy.deepcopy(rollout_report) bad["schema_version"] = 2 _expect_rollout_reject(bad, "schema is not 1") bad = copy.deepcopy(rollout_report) bad["instrument_version"] = 999 _expect_rollout_reject(bad, "instrument version differs") bad = copy.deepcopy(rollout_report) bad["publication_ready"] = False _expect_rollout_reject(bad, "not publication-ready") bad = copy.deepcopy(rollout_report) bad["receipt"] = dict(rollout_evidence) bad["receipt"]["sha256"] = "9" * 64 _expect_rollout_reject(bad, "receipt evidence differs") for key in ("step", "meta_sha256", "master_sha256"): bad = copy.deepcopy(rollout_report) bad["checkpoint"] = copy.deepcopy(checkpoint) bad["checkpoint"][key] = -1 if key == "step" else "tampered" _expect_rollout_reject(bad, f"rollout checkpoint {key} differs") bad = copy.deepcopy(rollout_report) bad["tokenizer"] = {"sha256": "0" * 64} _expect_rollout_reject(bad, "rollout tokenizer hash differs") bad = copy.deepcopy(rollout_report) bad["holdout"] = copy.deepcopy(rollout_report["holdout"]) bad["holdout"]["sha256"] = "0" * 64 _expect_rollout_reject(bad, "rollout holdout hash differs") bad = copy.deepcopy(rollout_report) bad["holdout"]["pair_manifest"] = manifest[:-1] _expect_rollout_reject(bad, "pair manifest has the wrong size") bad = copy.deepcopy(rollout_report) mutated_manifest = copy.deepcopy(manifest) mutated_manifest[0]["document_id"] = "tampered-doc" bad["holdout"]["pair_manifest"] = mutated_manifest _expect_rollout_reject(bad, "pair manifest differs from registered identity") bad = copy.deepcopy(rollout_report) bad["holdout"]["decoy_match"] = {"matched": 4} _expect_rollout_reject(bad, "rollout report includes relaxed decoy matches") bad = copy.deepcopy(rollout_report) bad["calibration"] = copy.deepcopy(rollout_report["calibration"]) bad["calibration"]["documents"] = 3 _expect_rollout_reject(bad, "calibration split or candidate order differs") bad = copy.deepcopy(rollout_report) bad["calibration"] = copy.deepcopy(rollout_report["calibration"]) bad["calibration"]["rows"]["fixed_d1"] = bad["calibration"]["rows"][ "fixed_d1" ][:1] _expect_rollout_reject(bad, "has the wrong row count") bad = copy.deepcopy(rollout_report) bad["calibration"] = copy.deepcopy(rollout_report["calibration"]) bad["calibration"]["rows"]["fixed_d1"][0]["document_id"] = "tampered" _expect_rollout_reject(bad, "differs on document_id") bad = copy.deepcopy(rollout_report) bad["calibration"] = copy.deepcopy(rollout_report["calibration"]) bad["calibration"]["rows"]["fixed_d1"][0]["policy"] = "wrong_policy" _expect_rollout_reject(bad, "has the wrong policy label") bad = copy.deepcopy(rollout_report) bad["calibration"] = copy.deepcopy(rollout_report["calibration"]) bad["calibration"]["rows"]["fixed_d1"][0]["accepted_drafts"] += 5 _expect_rollout_reject(bad, "calibration summaries do not recompute") bad = copy.deepcopy(rollout_report) bad["calibration"] = copy.deepcopy(rollout_report["calibration"]) bad["calibration"]["selected_fixed"] = dict( bad["calibration"]["selected_fixed"] ) bad["calibration"]["selected_fixed"]["policy"] = "fixed_d1" _expect_rollout_reject( bad, "rollout stored policy selection differs from calibration" ) bad = copy.deepcopy(rollout_report) bad["test"] = copy.deepcopy(rollout_report["test"]) bad["test"]["documents"] = 2 _expect_rollout_reject( bad, "rollout test split contains the wrong policies or documents" ) bad = copy.deepcopy(rollout_report) bad["test"] = copy.deepcopy(rollout_report["test"]) bad["test"]["rows"]["fixed_d2"][0]["accepted_drafts"] += 5 _expect_rollout_reject(bad, "rollout test summaries do not recompute") bad = copy.deepcopy(rollout_report) bad["quality_gate"] = dict(rollout_report["quality_gate"]) bad["quality_gate"]["exact_ar_matches"] = 14 bad["quality_gate"]["certified_divergences"] = 0 _expect_rollout_reject( bad, "rollout quality equivalence gate did not pass exactly" ) bad = copy.deepcopy(rollout_report) bad["test"] = copy.deepcopy(rollout_report["test"]) bad["test"]["rows"]["fixed_d2"][1]["divergence"] = None _expect_rollout_reject(bad, "without a certified near-tie") bad = copy.deepcopy(rollout_report) bad["test"] = copy.deepcopy(rollout_report["test"]) wide = bad["test"]["rows"]["fixed_d2"][1]["divergence"] wide["policy_token_logit"] = wide["row_max_logit"] - ( (NEAR_TIE_MAX_ULPS + 1) * wide["ulp_at_max"] ) wide["policy_token_deficit_ulps"] = float(NEAR_TIE_MAX_ULPS + 1) _expect_rollout_reject(bad, "without a certified near-tie") bad = copy.deepcopy(rollout_report) bad["test"] = copy.deepcopy(rollout_report["test"]) bad["test"]["rows"]["fixed_d2"][1]["output_sha256"] = "f" * 64 _expect_rollout_reject(bad, "repeats its output hash") bad = copy.deepcopy(rollout_report) bad["test"] = copy.deepcopy(rollout_report["test"]) bad["test"]["rows"]["fixed_d2"][0]["divergence"] = copy.deepcopy( rollout_report["test"]["rows"]["fixed_d2"][1]["divergence"] ) _expect_rollout_reject( bad, "claims an exact greedy AR match it does not have" ) bad = copy.deepcopy(rollout_report) bad["primary_endpoint"] = dict(rollout_report["primary_endpoint"]) bad["primary_endpoint"]["difference"] = 999.0 _expect_rollout_reject( bad, "rollout primary endpoint does not recompute from test rows" ) bad = copy.deepcopy(rollout_report) bad["secondary_target_forward_endpoint"] = dict( rollout_report["secondary_target_forward_endpoint"] ) bad["secondary_target_forward_endpoint"]["difference"] = 999.0 _expect_rollout_reject( bad, "rollout secondary endpoint does not recompute from test rows", ) format_comparison = { "schema_version": 1, "publication_ready": True, "documents": 4, "baseline_revision_evidence": { "limitation": ( "Repository revisions were captured after the run 1 " "corpus build, so source drift cannot be ruled out." ), }, "runtime_code_evidence": { "limitation": ( "Run 1 source hashes were not recorded, so exact " "source-state equivalence cannot be proven." ), }, "primary_endpoint": { "difference_in_differences": 0.06, "ci95": [0.01, 0.11], "verdict": "POSITIVE", }, "secondary_endpoint": { "difference_in_differences": 0.03, "ci95": [-0.01, 0.07], "verdict": "NULL: the interval includes zero", }, } metadata_rollout_report = copy.deepcopy( build_v3_rollout_fixture(temp_dir)["report"] ) metadata_rollout_report["quality_gate"].update({ "scored_policy_documents": 600, "scored_tokens": 38400, "exact_argmax_tokens": 38400, "certified_near_tie_tokens": 0, "branch_replay_passes": 600, "cross_policy_trajectory_matches": 100, }) metadata_rollout_report["cached_ar_diagnostic"].update({ "scored_policy_documents": 600, "exact_output_matches": 600, "different_cached_ar_branches": 0, }) for endpoint_key in ( "primary_endpoint", "secondary_target_forward_endpoint", "secondary_draft_issued_proxy_endpoint", "secondary_draft_work_endpoint", ): metadata_rollout_report[endpoint_key]["documents"] = 60 template_path = os.path.join(temp_dir, "MODEL_CARD.md") with open(template_path, "w", encoding="utf-8") as f: f.write("# {{REPO_ID}}\n\n{{FINAL_EVALUATION}}\n") export_dir = os.path.join(temp_dir, "export") os.makedirs(export_dir) rendered_evaluation = render_evaluation_section( validation_report, comparison, format_comparison, metadata_rollout_report, ) rendered_card = render_model_card( template_path, "owner/model", rendered_evaluation, ) exported_readme = os.path.join(export_dir, "README.md") with open(exported_readme, "w", encoding="utf-8") as f: f.write(rendered_card) export_manifest = { "repo_id": "owner/model", "release_complete": True, "evaluation_sources": { "validation": {"sha256": "9" * 64}, "acceptance_comparison": {"sha256": "a" * 64}, "format_ablation": {"sha256": "c" * 64}, "rollout": {"sha256": "b" * 64}, }, "model_card_template_sha256": file_sha256(template_path), } validate_export_model_card( export_manifest, export_dir, template_path, validation_report, "9" * 64, comparison, "a" * 64, format_comparison, "c" * 64, metadata_rollout_report, "b" * 64, ) def _expect_export_reject(mutated_manifest, expected_substring): try: validate_export_model_card( mutated_manifest, export_dir, template_path, validation_report, "9" * 64, comparison, "a" * 64, format_comparison, "c" * 64, metadata_rollout_report, "b" * 64, ) except ValueError as exc: assert expected_substring in str(exc), ( f"expected {expected_substring!r} in {exc}" ) else: raise AssertionError( f"an export manifest that should have failed on " f"{expected_substring!r} passed release audit" ) bad_manifest = copy.deepcopy(export_manifest) bad_manifest["release_complete"] = False _expect_export_reject( bad_manifest, "export package is labelled as a development snapshot" ) bad_manifest = copy.deepcopy(export_manifest) bad_manifest["evaluation_sources"]["validation"]["sha256"] = "0" * 64 _expect_export_reject( bad_manifest, "export evaluation source hashes differ from audited reports" ) bad_manifest = copy.deepcopy(export_manifest) bad_manifest["model_card_template_sha256"] = "0" * 64 _expect_export_reject( bad_manifest, "export model card template differs from audited template" ) with open(exported_readme, "a", encoding="utf-8") as f: f.write("hand-edited claim\n") try: validate_export_model_card( export_manifest, export_dir, template_path, validation_report, "9" * 64, comparison, "a" * 64, format_comparison, "c" * 64, metadata_rollout_report, "b" * 64, ) except ValueError as exc: assert "does not render" in str(exc) else: raise AssertionError("a hand-edited exported model card passed audit") mutated_rollout = copy.deepcopy(rollout_report) mutated_rollout["test"]["rows"]["adaptive_h0.3"][0][ "output_sha256" ] = "0" * 64 try: validate_rollout_report( mutated_rollout, rollout_receipt, rollout_evidence, checkpoint, artifacts["tokenizer"], artifacts["holdout"], ) except ValueError as exc: assert "claims an exact greedy AR match it does not have" in str(exc) else: raise AssertionError("a rollout output mutation passed release audit") # validate_no_tampering had zero test coverage before this: it lives # directly in release_audit.py's main(), which this file never calls, # so these checks -- including the final tamper checks that are the # last thing before publication_ready:true -- had never executed # under any test. Extracted verbatim into its own function so it can # be tested here without needing the full ~15-file main() bundle. with tempfile.TemporaryDirectory() as tmp: def build_ckpt(root, step): meta = { "step": step, "config": {"run_name": "test-run", "max_steps": step}, "model_args": {"vocab_size": 32768}, "optimizer_state_included": True, } meta_path = os.path.join(root, "meta.json") os.makedirs(root, exist_ok=True) with open(meta_path, "w", encoding="utf-8") as f: json.dump(meta, f) with open(os.path.join(root, "master.safetensors"), "wb") as f: f.write(b"master") with open(os.path.join(root, "optimizer.safetensors"), "wb") as f: f.write(b"optimizer") return root run1_dir = build_ckpt(os.path.join(tmp, "run1"), 19073) run2_dir = build_ckpt(os.path.join(tmp, "run2"), 19073) _, run1_identity = checkpoint_identity(run1_dir) _, run2_identity = checkpoint_identity(run2_dir) artifact_path = os.path.join(tmp, "artifact.json") with open(artifact_path, "w", encoding="utf-8") as f: json.dump({"k": "v"}, f) tamper_artifacts = { "one": {"path": artifact_path, "sha256": file_sha256(artifact_path)} } # Happy path: nothing changed since it was read. validate_no_tampering( tamper_artifacts, run1_dir, run1_identity, run2_dir, run2_identity ) # An audited artifact file changes after being hashed. with open(artifact_path, "w", encoding="utf-8") as f: json.dump({"k": "mutated"}, f) try: validate_no_tampering( tamper_artifacts, run1_dir, run1_identity, run2_dir, run2_identity ) except ValueError as exc: assert "release artifact changed during audit" in str(exc) else: raise AssertionError( "a release artifact mutated after hashing passed audit" ) with open(artifact_path, "w", encoding="utf-8") as f: json.dump({"k": "v"}, f) # The final checkpoint's own weights change after being read. with open( os.path.join(run1_dir, "master.safetensors"), "wb" ) as f: f.write(b"mutated-master") try: validate_no_tampering( tamper_artifacts, run1_dir, run1_identity, run2_dir, run2_identity ) except ValueError as exc: assert "final checkpoint changed during release audit" in str(exc) else: raise AssertionError( "the final checkpoint mutated after reading passed audit" ) with open( os.path.join(run1_dir, "master.safetensors"), "wb" ) as f: f.write(b"master") # The run 2 ablation checkpoint changes after being read. with open( os.path.join(run2_dir, "master.safetensors"), "wb" ) as f: f.write(b"mutated-master") try: validate_no_tampering( tamper_artifacts, run1_dir, run1_identity, run2_dir, run2_identity ) except ValueError as exc: assert "run 2 checkpoint changed during release audit" in str(exc) else: raise AssertionError( "the run 2 checkpoint mutated after reading passed audit" ) # The remaining Class B guards from main()'s own body: extracted # the same way as validate_no_tampering above, each testable with # a tiny fixture instead of the full ~15-file main() bundle. holdout_path = os.path.join(tmp, "b_holdout.jsonl") tokenizer_path = os.path.join(tmp, "b_tokenizer.json") with open(holdout_path, "w", encoding="utf-8") as f: f.write("holdout") with open(tokenizer_path, "w", encoding="utf-8") as f: f.write("tokenizer") fresh_acceptance_receipt = { "clean_holdout": {"sha256": file_sha256(holdout_path)}, "pair_readiness": { "tokenizer_sha256": file_sha256(tokenizer_path) }, } validate_acceptance_receipt_freshness( fresh_acceptance_receipt, holdout_path, tokenizer_path ) stale_holdout_receipt = copy.deepcopy(fresh_acceptance_receipt) stale_holdout_receipt["clean_holdout"]["sha256"] = "0" * 64 try: validate_acceptance_receipt_freshness( stale_holdout_receipt, holdout_path, tokenizer_path ) except ValueError as exc: assert "does not match current clean holdout" in str(exc) else: raise AssertionError( "an acceptance receipt with a stale holdout hash passed audit" ) stale_tokenizer_receipt = copy.deepcopy(fresh_acceptance_receipt) stale_tokenizer_receipt["pair_readiness"]["tokenizer_sha256"] = ( "0" * 64 ) try: validate_acceptance_receipt_freshness( stale_tokenizer_receipt, holdout_path, tokenizer_path ) except ValueError as exc: assert "does not match current tokenizer" in str(exc) else: raise AssertionError( "an acceptance receipt with a stale tokenizer hash passed audit" ) training_data_receipt_path = "/tmp/training_data_receipt.json" training_data_receipt_sha256 = "1" * 64 registered_format_receipt = { "training_data_receipt": { "path": training_data_receipt_path, "sha256": training_data_receipt_sha256, } } validate_training_data_receipt_registration( registered_format_receipt, training_data_receipt_path, training_data_receipt_sha256, ) mismatched_format_receipt = copy.deepcopy(registered_format_receipt) mismatched_format_receipt["training_data_receipt"]["sha256"] = ( "2" * 64 ) try: validate_training_data_receipt_registration( mismatched_format_receipt, training_data_receipt_path, training_data_receipt_sha256, ) except ValueError as exc: assert "different training-data receipts" in str(exc) else: raise AssertionError( "an E2 report registered against a different training-data " "receipt passed audit" ) ablation_checkpoint = { "step": 19073, "meta_sha256": "3" * 64, "master_sha256": "4" * 64, } validate_ablation_arm_checkpoint( dict(ablation_checkpoint), ablation_checkpoint ) for key in ("step", "meta_sha256", "master_sha256"): bad_ablation_report_checkpoint = dict(ablation_checkpoint) bad_ablation_report_checkpoint[key] = ( -1 if key == "step" else "tampered" ) try: validate_ablation_arm_checkpoint( bad_ablation_report_checkpoint, ablation_checkpoint ) except ValueError as exc: assert ( f"format-ablation arm checkpoint {key} differs" in str(exc) ) else: raise AssertionError( f"a format-ablation report with a tampered {key} " "checkpoint field passed audit" ) export_manifest_source = { "source_checkpoint": { key: checkpoint[key] for key in ( "step", "meta_sha256", "master_sha256", "optimizer_sha256", ) } } validate_export_sourced_from_checkpoint( export_manifest_source, checkpoint ) wrong_source_manifest = copy.deepcopy(export_manifest_source) wrong_source_manifest["source_checkpoint"]["master_sha256"] = ( "5" * 64 ) try: validate_export_sourced_from_checkpoint( wrong_source_manifest, checkpoint ) except ValueError as exc: assert "not sourced from the audited final checkpoint" in str( exc ) else: raise AssertionError( "an export manifest sourced from a different checkpoint " "passed audit" ) export_dir = os.path.join(tmp, "b_export") ckpt_dir = os.path.join(tmp, "b_ckpt") os.makedirs(export_dir, exist_ok=True) os.makedirs(ckpt_dir, exist_ok=True) matching_external = { "package": {"export_dir": os.path.abspath(export_dir)}, "checkpoint": {"path": os.path.abspath(ckpt_dir)}, } validate_external_verification_paths( matching_external, export_dir, ckpt_dir ) wrong_export_dir_external = copy.deepcopy(matching_external) wrong_export_dir_external["package"]["export_dir"] = "/tmp/elsewhere" try: validate_external_verification_paths( wrong_export_dir_external, export_dir, ckpt_dir ) except ValueError as exc: assert "different export directory" in str(exc) else: raise AssertionError( "external verification pointing at a different export " "directory passed audit" ) wrong_ckpt_dir_external = copy.deepcopy(matching_external) wrong_ckpt_dir_external["checkpoint"]["path"] = "/tmp/elsewhere" try: validate_external_verification_paths( wrong_ckpt_dir_external, export_dir, ckpt_dir ) except ValueError as exc: assert "different checkpoint directory" in str(exc) else: raise AssertionError( "external verification pointing at a different " "checkpoint directory passed audit" ) print("release audit: PASS") if __name__ == "__main__": main()