evidence/source/e3_rollout_v3/scripts/test_release_audit.py

"""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()