evidence/source/e3_rollout_v3/scripts/release_audit.py

"""Fail-closed audit over every artifact required for a Wisp release.

Research outcomes are evidence, not publication gates. A null or negative
registered result remains releasable when the result is internally consistent,
bound to the final checkpoint, and reported honestly.

Usage:
    .venv/bin/python scripts/release_audit.py \
      --ckpt out/run1/ckpt_latest \
      --validation out/run1/final_validation.json \
      --trained-acceptance out/run1/acceptance.trained.json \
      --control-acceptance \
        out/run1-untrained/acceptance.untrained-control.json \
      --acceptance-comparison \
        out/run1/acceptance.control-comparison.json \
      --ablation-ckpt out/run2-no-fim/ckpt_latest \
      --ablation-acceptance \
        out/run2-no-fim/acceptance.no-fim-ablation.json \
      --format-ablation \
        out/run2-no-fim/acceptance.format-ablation.json \
      --rollout out/run1/rollout.registered.v3.json \
      --rollout-verification \
        out/run1/rollout.replay-verification.v3.json \
      --export export/wisp-coder-110m \
      --external-verification \
        out/run1/external_export_verification.json \
      --out out/run1/release_audit.json
"""

import argparse
from datetime import datetime, timezone
import hashlib
import json
import math
import os
import sys

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 compare_format_ablation import (  # noqa: E402
    validate_comparison_report as validate_format_comparison,
)
from e2_contract import validate_e2_evaluation_inputs  # noqa: E402
from hf_metadata import (  # noqa: E402
    file_sha256,
    render_evaluation_section,
    render_model_card,
    validate_export_checkpoint,
    validate_external_verification_receipt,
    verify_export_manifest,
    write_json_atomic,
)
from rollout_metrics import (  # noqa: E402
    NEAR_TIE_MAX_ULPS,
    V3_ATTESTATION_PROVENANCE_SCOPE,
    V3_INSTRUMENT_VERSION,
    V3_REPLAY_VERIFICATION_ARGV,
    V3_REPLAY_VERIFIER_METHOD,
    V3_REPLAY_REFERENCE,
    V3_REPLAY_RULE,
    V3_REPORT_SCHEMA_VERSION,
    bf16_ulp,
    canonical_json_sha256,
    metric_value,
    paired_mean_difference_ci,
    select_policy,
    summarize_policy,
    summarize_policy_v3,
    token_ids_sha256,
    validate_cross_policy_trajectories,
    validate_divergence_evidence,
    validate_pair_payload_manifest,
    validate_rollout_checkpoint,
    validate_rollout_receipt,
)
from training_data_contract import (  # noqa: E402
    validate_publication_text,
    validate_training_data_receipt,
)
from validation_metrics import (  # noqa: E402
    summarize_validation,
    validate_validation_checkpoint,
    validate_validation_receipt,
)


SCHEMA_VERSION = 2


def load_json_snapshot(path):
    with open(path, "rb") as f:
        content = f.read()
    digest = hashlib.sha256(content).hexdigest()
    try:
        value = json.loads(content)
    except (UnicodeDecodeError, json.JSONDecodeError) as exc:
        raise ValueError(f"{path}: invalid JSON: {exc}") from exc
    if not isinstance(value, dict):
        raise ValueError(f"{path}: top-level JSON value must be an object")
    return value, digest


def checkpoint_identity(ckpt_dir):
    paths = {
        "meta": os.path.join(ckpt_dir, "meta.json"),
        "master": os.path.join(ckpt_dir, "master.safetensors"),
        "optimizer": os.path.join(ckpt_dir, "optimizer.safetensors"),
    }
    meta, meta_sha256 = load_json_snapshot(paths["meta"])
    validate_export_checkpoint(meta)
    return meta, {
        "path": os.path.abspath(ckpt_dir),
        "step": meta["step"],
        "meta_sha256": meta_sha256,
        "master_sha256": file_sha256(paths["master"]),
        "optimizer_sha256": file_sha256(paths["optimizer"]),
    }


def _require(condition, message):
    if not condition:
        raise ValueError(message)


def _artifact_matches(value, path):
    return (
        isinstance(value, dict)
        and value.get("sha256") == file_sha256(path)
    )


def validate_e2_report_sampler(report, e2_evidence):
    contract = report.get("publication_contract", {})
    _require(
        isinstance(contract, dict)
        and contract.get("ablation_data_index_sha256")
        == e2_evidence.get("ablation_data_index_sha256")
        and contract.get("final_train_sampler")
        == e2_evidence.get("final_train_sampler")
        and contract.get("ablation_data_artifacts")
        == e2_evidence.get("ablation_data_artifacts"),
        "format-ablation report uses different final sampler evidence",
    )
    return contract["final_train_sampler"]


def validate_validation_report(
    report,
    receipt,
    receipt_evidence,
    checkpoint,
    config_path,
    data_index_path,
    validation_shard_path,
):
    _require(report.get("schema_version") == 1, "validation schema is not 1")
    _require(
        report.get("instrument_version") == receipt.get("instrument_version"),
        "validation instrument version differs from its receipt",
    )
    _require(
        report.get("publication_ready") is True,
        "validation is not publication-ready",
    )
    evidence = report.get("receipt", {})
    _require(
        evidence.get("sha256") == receipt_evidence.get("sha256"),
        "validation receipt hash differs from registered input",
    )
    _require(
        evidence.get("registered_at") == receipt_evidence.get("registered_at"),
        "validation registration time differs from registered input",
    )
    _require(
        evidence.get("batch_manifest")
        == receipt_evidence.get("batch_manifest"),
        "validation batch manifest differs from registered input",
    )
    reported_checkpoint = report.get("checkpoint", {})
    for key in ("step", "meta_sha256", "master_sha256", "optimizer_sha256"):
        _require(
            reported_checkpoint.get(key) == checkpoint.get(key),
            f"validation checkpoint {key} differs from final checkpoint",
        )
    for key, path in (
        ("config", config_path),
        ("data_index", data_index_path),
        ("validation_shard", validation_shard_path),
    ):
        _require(
            _artifact_matches(report.get(key), path),
            f"validation {key} artifact hash differs",
        )
    settings = receipt["settings"]
    _require(
        report.get("settings") == settings,
        "validation settings differ from registered settings",
    )
    rows = report.get("batches")
    expected_hashes = receipt_evidence["batch_manifest"]["batch_sha256"]
    _require(
        isinstance(rows, list) and len(rows) == len(expected_hashes),
        "validation row count differs from frozen batches",
    )
    for index, (row, expected_hash) in enumerate(zip(rows, expected_hashes)):
        _require(
            row.get("batch") == index and row.get("sha256") == expected_hash,
            f"validation batch evidence differs at index {index}",
        )
    expected_summary = summarize_validation(
        rows,
        n_boot=settings["bootstrap_samples"],
        seed=settings["bootstrap_seed"],
    )
    _require(
        report.get("summary") == expected_summary,
        "validation summary does not recompute from batch rows",
    )
    elapsed = report.get("elapsed_seconds")
    _require(
        isinstance(elapsed, (int, float))
        and not isinstance(elapsed, bool)
        and math.isfinite(elapsed)
        and elapsed > 0,
        "validation elapsed time is not finite and positive",
    )
    return {
        "main_loss": expected_summary["main_loss"],
        "main_perplexity": expected_summary["main_perplexity"],
        "mtp_loss": expected_summary["mtp_loss"],
        "target_tokens": receipt["batch_manifest"]["target_tokens"],
    }


def validate_acceptance_bundle(
    trained,
    trained_sha256,
    control,
    control_sha256,
    comparison,
    receipt,
    receipt_sha256,
    checkpoint,
):
    expected = compare_reports(
        trained,
        control,
        receipt,
        receipt_sha256,
    )
    stored_core = {
        key: value
        for key, value in comparison.items()
        if key != "inputs"
    }
    _require(
        stored_core == expected,
        "acceptance comparison does not recompute from source reports",
    )
    inputs = comparison.get("inputs", {})
    expected_inputs = {
        "trained": trained_sha256,
        "control": control_sha256,
        "receipt": receipt_sha256,
    }
    for key, digest in expected_inputs.items():
        _require(
            inputs.get(key, {}).get("sha256") == digest,
            f"acceptance comparison {key} input hash differs",
        )
    trained_checkpoint = trained.get("checkpoint", {})
    for key in ("step", "meta_sha256", "master_sha256"):
        _require(
            trained_checkpoint.get(key) == checkpoint.get(key),
            f"trained acceptance checkpoint {key} differs from final checkpoint",
        )
    control_checkpoint = control.get("checkpoint", {})
    _require(
        control_checkpoint.get("master_sha256") != checkpoint["master_sha256"],
        "untrained control uses the final trained weights",
    )
    return {
        "trained_primary_endpoint": expected["trained_primary_endpoint"],
        "control_adjustment": expected["trained_minus_untrained_ratio"],
        "combined_interpretation": expected["combined_interpretation"],
        "documents": expected["documents"],
    }


def _manifest_sha256(rows):
    encoded = json.dumps(
        rows, sort_keys=True, separators=(",", ":")
    ).encode("utf-8")
    return hashlib.sha256(encoded).hexdigest()


def _validate_policy_rows_v2(policy_name, rows, manifest):
    _require(
        isinstance(rows, list) and len(rows) == len(manifest),
        f"rollout policy {policy_name} has the wrong row count",
    )
    for index, (row, identity) in enumerate(zip(rows, manifest)):
        expected = {
            "document_id": identity["document_id"],
            "decoy_document_id": identity["decoy_document_id"],
            "token_offset": identity["token_offset"],
            "prompt_sha256": identity["prompt_sha256"],
            "target_sha256": identity["target_sha256"],
        }
        for key, value in expected.items():
            _require(
                row.get(key) == value,
                f"rollout policy {policy_name} row {index} differs on {key}",
            )
        _require(
            row.get("policy") == policy_name,
            f"rollout row {index} has the wrong policy label",
        )
        if row.get("output_matches_ar") is True:
            _require(
                row.get("output_sha256") == row.get("ar_output_sha256")
                and row.get("divergence") is None,
                f"rollout policy {policy_name} row {index} claims an exact "
                "greedy AR match it does not have",
            )
        else:
            _require(
                row.get("output_sha256") != row.get("ar_output_sha256"),
                f"rollout policy {policy_name} row {index} diverges from "
                "greedy AR yet repeats its output hash",
            )
            try:
                validate_divergence_evidence(row.get("divergence"))
            except ValueError as error:
                _require(
                    False,
                    f"rollout policy {policy_name} row {index} differs from "
                    f"greedy AR without a certified near-tie: {error}",
                )
    return summarize_policy(rows)


def _validate_rollout_report_v2(
    report,
    receipt,
    receipt_evidence,
    checkpoint,
    tokenizer_path,
    holdout_path,
):
    _require(report.get("schema_version") == 1, "rollout schema is not 1")
    _require(
        report.get("instrument_version") == receipt.get("instrument_version"),
        "rollout instrument version differs from its receipt",
    )
    _require(
        report.get("publication_ready") is True,
        "rollout report is not publication-ready",
    )
    evidence = report.get("receipt", {})
    _require(
        evidence.get("sha256") == receipt_evidence.get("sha256")
        and evidence.get("registered_at") == receipt_evidence.get("registered_at"),
        "rollout receipt evidence differs from registered input",
    )
    reported_checkpoint = report.get("checkpoint", {})
    for key in ("step", "meta_sha256", "master_sha256"):
        _require(
            reported_checkpoint.get(key) == checkpoint.get(key),
            f"rollout checkpoint {key} differs from final checkpoint",
        )
    _require(
        _artifact_matches(report.get("tokenizer"), tokenizer_path),
        "rollout tokenizer hash differs",
    )
    _require(
        _artifact_matches(report.get("holdout"), holdout_path),
        "rollout holdout hash differs",
    )

    pair_settings = receipt["pair_settings"]
    holdout = report["holdout"]
    manifest = holdout.get("pair_manifest")
    _require(
        isinstance(manifest, list)
        and len(manifest) == pair_settings["examples"],
        "rollout pair manifest has the wrong size",
    )
    _require(
        holdout.get("pair_count") == pair_settings["examples"]
        and holdout.get("pair_manifest_sha256")
        == pair_settings["pair_manifest_sha256"]
        and _manifest_sha256(manifest)
        == pair_settings["pair_manifest_sha256"],
        "rollout pair manifest differs from registered identity",
    )
    _require(
        holdout.get("decoy_match") == {"matched": pair_settings["examples"]},
        "rollout report includes relaxed decoy matches",
    )

    split = receipt["split"]
    calibration_count = split["calibration_documents"]
    calibration_manifest = manifest[:calibration_count]
    test_manifest = manifest[calibration_count:]
    policy = receipt["policy"]
    candidate_order = [
        *policy["fixed_candidates"],
        *policy["adaptive_candidates"],
    ]
    calibration = report.get("calibration", {})
    calibration_rows = calibration.get("rows", {})
    _require(
        calibration.get("documents") == len(calibration_manifest)
        and calibration.get("candidate_order") == candidate_order
        and set(calibration_rows) == set(candidate_order),
        "rollout calibration split or candidate order differs",
    )
    calibration_summaries = {
            name: _validate_policy_rows_v2(
            name, calibration_rows[name], calibration_manifest
        )
        for name in candidate_order
    }
    _require(
        calibration.get("summaries") == calibration_summaries,
        "rollout calibration summaries do not recompute",
    )
    selected_fixed = select_policy(
        calibration_summaries,
        policy["fixed_candidates"],
        policy["selection_metric"],
    )
    selected_adaptive = select_policy(
        calibration_summaries,
        policy["adaptive_candidates"],
        policy["selection_metric"],
    )
    _require(
        calibration.get("selected_fixed") == selected_fixed
        and calibration.get("selected_adaptive") == selected_adaptive,
        "rollout stored policy selection differs from calibration",
    )

    selected_names = [
        selected_fixed["policy"],
        selected_adaptive["policy"],
    ]
    test = report.get("test", {})
    test_rows = test.get("rows", {})
    _require(
        test.get("documents") == len(test_manifest)
        and set(test_rows) == set(selected_names),
        "rollout test split contains the wrong policies or documents",
    )
    test_summaries = {
        name: _validate_policy_rows_v2(name, test_rows[name], test_manifest)
        for name in selected_names
    }
    _require(
        test.get("summaries") == test_summaries,
        "rollout test summaries do not recompute",
    )
    audited_rows = [
        row
        for rows in (calibration_rows, test_rows)
        for name in rows
        for row in rows[name]
    ]
    exact_matches = sum(
        1 for row in audited_rows if row.get("output_matches_ar") is True
    )
    _require(
        report.get("quality_gate")
        == {
            "reference": "greedy_ar",
            "rule": "exact_token_match_or_certified_near_tie",
            "near_tie_max_ulps": NEAR_TIE_MAX_ULPS,
            "exact_ar_matches": exact_matches,
            "certified_divergences": len(audited_rows) - exact_matches,
            "passed": True,
        },
        "rollout quality equivalence gate did not pass exactly",
    )

    endpoint = receipt["test_endpoint"]
    metric = endpoint["metric"]
    adaptive_rows = test_rows[selected_adaptive["policy"]]
    fixed_rows = test_rows[selected_fixed["policy"]]
    adaptive_values = [metric_value(row, metric) for row in adaptive_rows]
    fixed_values = [metric_value(row, metric) for row in fixed_rows]
    difference, lo, hi = paired_mean_difference_ci(
        adaptive_values,
        fixed_values,
        n_boot=endpoint["bootstrap_samples"],
        seed=endpoint["bootstrap_seed"],
    )
    if lo > 0:
        verdict = "POSITIVE"
    elif hi < 0:
        verdict = "NEGATIVE"
    else:
        verdict = "NULL: the interval includes 0"
    expected_primary = {
        "comparison": endpoint["comparison"],
        "metric": metric,
        "adaptive_policy": selected_adaptive["policy"],
        "fixed_policy": selected_fixed["policy"],
        "difference": difference,
        "ci95": [lo, hi],
        "documents": len(test_manifest),
        "verdict": verdict,
    }
    _require(
        report.get("primary_endpoint") == expected_primary,
        "rollout primary endpoint does not recompute from test rows",
    )
    secondary_metric = "output_tokens_per_target_forward"
    secondary_adaptive = [
        metric_value(row, secondary_metric) for row in adaptive_rows
    ]
    secondary_fixed = [
        metric_value(row, secondary_metric) for row in fixed_rows
    ]
    secondary_difference, secondary_lo, secondary_hi = (
        paired_mean_difference_ci(
            secondary_adaptive,
            secondary_fixed,
            n_boot=endpoint["bootstrap_samples"],
            seed=endpoint["bootstrap_seed"],
        )
    )
    expected_secondary = {
        "metric": secondary_metric,
        "difference": secondary_difference,
        "ci95": [secondary_lo, secondary_hi],
        "documents": len(test_manifest),
    }
    _require(
        report.get("secondary_target_forward_endpoint")
        == expected_secondary,
        "rollout secondary endpoint does not recompute from test rows",
    )
    return {
        "selected_fixed": selected_fixed,
        "selected_adaptive": selected_adaptive,
        "primary_endpoint": expected_primary,
        "secondary_target_forward_endpoint": expected_secondary,
        "quality_equivalent_to_greedy_ar": False,
        "branch_local_replay_valid": False,
        "cross_policy_trajectory_identical": False,
        "historical_instrument_only": True,
    }


_V3_ROLLOUT_ROW_KEYS = {
    "pair_index",
    "seed",
    "document_id",
    "decoy_document_id",
    "token_offset",
    "policy",
    "prompt_sha256",
    "target_sha256",
    "ar_output_sha256",
    "ar_output_token_ids",
    "output_sha256",
    "output_token_ids",
    "output_matches_ar",
    "cached_ar_diagnostic",
    "trace_sha256",
    "generation_trace",
    "branch_replay",
    "tokens",
    "accepted_drafts",
    "verification_forwards",
    "target_forwards",
    "drafts_issued",
    "draft_recursions",
    "corrections",
    "elapsed_seconds",
    "ar_tok_per_sec",
    "rollout",
    "target_position_accuracy",
    "target_common_prefix_tokens",
    "target_exact_match",
}

_V3_REPORT_KEYS = {
    "schema_version",
    "instrument_version",
    "publication_ready",
    "execution",
    "receipt",
    "checkpoint",
    "tokenizer",
    "holdout",
    "calibration",
    "test",
    "quality_gate",
    "cached_ar_diagnostic",
    "primary_endpoint",
    "secondary_target_forward_endpoint",
    "secondary_draft_issued_proxy_endpoint",
    "secondary_draft_work_endpoint",
    "wall_clock_note",
    "endpoint_scope_note",
}

_V3_WALL_CLOCK_NOTE = (
    "This reference recomputes full prefixes and has no rollback-capable "
    "KV cache. Wall time is recorded for audit, not claimed as deployment "
    "latency."
)

_V3_ENDPOINT_SCOPE_NOTE = (
    "The primary endpoint measures accepted drafts per verification. "
    "It does not establish verification-width cost or deployment latency. "
    "Draft recursions per output token is the registered drafter-work "
    "companion; issued drafts per output token is retained only as an issuance "
    "proxy. Target forwards exclude the added post-hoc branch-replay forward "
    "and independent verification pass."
)


def _validate_v3_execution(execution, receipt, receipt_evidence):
    _require(
        isinstance(execution, dict)
        and set(execution) == {"started_at", "completed_at", "argv", "runtime"},
        "rollout v3 execution evidence has the wrong fields",
    )
    timestamps = []
    for key in ("started_at", "completed_at"):
        value = execution[key]
        _require(
            isinstance(value, str) and value.endswith("Z"),
            f"rollout v3 execution {key} is not a UTC timestamp",
        )
        try:
            parsed = datetime.fromisoformat(value.replace("Z", "+00:00"))
        except ValueError as error:
            raise ValueError(
                f"rollout v3 execution {key} is not an ISO timestamp"
            ) from error
        _require(
            parsed.utcoffset() == timezone.utc.utcoffset(parsed),
            f"rollout v3 execution {key} is not UTC",
        )
        timestamps.append(parsed)
    _require(
        timestamps[0] <= timestamps[1],
        "rollout v3 execution completes before it starts",
    )
    registered_at = receipt.get("registered_at")
    _require(
        registered_at == receipt_evidence.get("registered_at")
        and isinstance(registered_at, str)
        and registered_at.endswith("Z"),
        "rollout v3 registration timestamp differs from its receipt evidence",
    )
    try:
        registered_timestamp = datetime.fromisoformat(
            registered_at.replace("Z", "+00:00")
        )
    except ValueError as error:
        raise ValueError(
            "rollout v3 registration timestamp is not ISO-8601"
        ) from error
    _require(
        registered_timestamp.utcoffset()
        == timezone.utc.utcoffset(registered_timestamp)
        and registered_timestamp <= timestamps[0],
        "rollout v3 execution started before registration",
    )
    registration_git = receipt_evidence.get("registration_git")
    _require(
        isinstance(registration_git, dict)
        and set(registration_git)
        == {"commit", "committed_at", "path", "blob_sha256", "origin_ref"}
        and isinstance(registration_git["commit"], str)
        and len(registration_git["commit"]) == 40
        and registration_git["path"] == "config/eval_rollout_receipt_v3.json"
        and registration_git["blob_sha256"] == receipt_evidence["sha256"]
        and registration_git["origin_ref"] == "origin/main",
        "rollout v3 pushed registration-commit evidence differs",
    )
    try:
        committed_timestamp = datetime.fromisoformat(
            registration_git["committed_at"].replace("Z", "+00:00")
        )
    except (AttributeError, ValueError) as error:
        raise ValueError(
            "rollout v3 registration commit time is not ISO-8601"
        ) from error
    _require(
        committed_timestamp.tzinfo is not None
        and registered_timestamp <= committed_timestamp <= timestamps[0],
        "rollout v3 receipt was not committed before execution",
    )
    argv = execution["argv"]
    _require(
        isinstance(argv, list)
        and bool(argv)
        and all(isinstance(value, str) for value in argv),
        "rollout v3 execution argv is invalid",
    )
    _require(
        argv == receipt.get("execution_argv")
        and receipt_evidence.get("execution_argv") == argv,
        "rollout v3 execution argv differs from registration",
    )
    runtime = execution["runtime"]
    expected_runtime = {"python", "mlx", "numpy", "tokenizers", "platform"}
    _require(
        isinstance(runtime, dict)
        and set(runtime) == expected_runtime
        and all(
            isinstance(runtime[key], str) and bool(runtime[key])
            for key in expected_runtime
        ),
        "rollout v3 runtime evidence is invalid",
    )
    registered_runtime = receipt.get("runtime_requirements")
    _require(
        isinstance(registered_runtime, dict)
        and receipt_evidence.get("runtime_requirements") == registered_runtime
        and {
            key: runtime[key]
            for key in ("python", "mlx", "numpy", "tokenizers")
        }
        == registered_runtime,
        "rollout v3 runtime differs from its registered environment",
    )


def _validate_v3_artifact(value, path, label):
    _require(
        isinstance(value, dict)
        and set(value) == {"path", "sha256"}
        and isinstance(value["path"], str)
        and os.path.abspath(value["path"]) == os.path.abspath(path)
        and value["sha256"] == file_sha256(path),
        f"rollout v3 {label} identity differs",
    )


def _validate_v3_policy_rows(
    policy_name,
    rows,
    pair_payloads,
    *,
    decoding_seed,
    max_depth,
    max_tokens,
    vocab_size,
):
    _require(
        isinstance(rows, list) and len(rows) == len(pair_payloads),
        f"rollout v3 policy {policy_name} has the wrong row count",
    )
    if policy_name.startswith("fixed_d"):
        expected_policy = "fixed"
        expected_threshold = None
        try:
            expected_depth = int(policy_name.removeprefix("fixed_d"))
        except ValueError as error:
            raise ValueError(
                f"rollout v3 policy {policy_name} is malformed"
            ) from error
    elif policy_name.startswith("adaptive_h"):
        expected_policy = "adaptive"
        try:
            expected_threshold = float(
                policy_name.removeprefix("adaptive_h")
            )
        except ValueError as error:
            raise ValueError(
                f"rollout v3 policy {policy_name} is malformed"
            ) from error
        expected_depth = max_depth
    else:
        raise ValueError(f"rollout v3 policy {policy_name} is unknown")
    _require(
        1 <= expected_depth <= max_depth,
        f"rollout v3 policy {policy_name} depth differs from registration",
    )

    for local_index, (row, payload) in enumerate(zip(rows, pair_payloads)):
        _require(
            isinstance(row, dict) and set(row) == _V3_ROLLOUT_ROW_KEYS,
            f"rollout v3 policy {policy_name} row {local_index} "
            "has the wrong fields",
        )
        expected_identity = {
            "pair_index": payload["index"],
            "seed": decoding_seed + local_index,
            "document_id": payload["document_id"],
            "decoy_document_id": payload["decoy_document_id"],
            "token_offset": payload["token_offset"],
            "policy": policy_name,
            "prompt_sha256": payload["prompt_sha256"],
            "target_sha256": payload["target_sha256"],
        }
        for key, expected in expected_identity.items():
            _require(
                row.get(key) == expected,
                f"rollout v3 policy {policy_name} row {local_index} "
                f"differs on {key}",
            )
        rollout = row.get("rollout")
        _require(
            isinstance(rollout, dict)
            and rollout.get("policy") == expected_policy
            and rollout.get("entropy_threshold") == expected_threshold,
            f"rollout v3 policy {policy_name} row {local_index} "
            "runtime policy differs",
        )
        accepted_by_depth = rollout.get("rollout_accepted_per_depth")
        trials_by_depth = rollout.get("rollout_trials_per_depth")
        _require(
            isinstance(accepted_by_depth, list)
            and isinstance(trials_by_depth, list)
            and len(accepted_by_depth) == expected_depth
            and len(trials_by_depth) == expected_depth,
            f"rollout v3 policy {policy_name} row {local_index} "
            "runtime depth differs",
        )
        ar_tok_per_sec = row.get("ar_tok_per_sec")
        _require(
            isinstance(ar_tok_per_sec, (int, float))
            and not isinstance(ar_tok_per_sec, bool)
            and math.isfinite(ar_tok_per_sec)
            and ar_tok_per_sec > 0,
            f"rollout v3 policy {policy_name} row {local_index} "
            "cached AR timing is invalid",
        )
    try:
        return summarize_policy_v3(
            rows,
            pair_payloads,
            max_tokens=max_tokens,
            vocab_size=vocab_size,
        )
    except (KeyError, TypeError, ValueError) as error:
        raise ValueError(
            f"rollout v3 policy {policy_name} evidence is invalid: {error}"
        ) from error


def _rollout_endpoint_result(
    specification,
    adaptive_rows,
    fixed_rows,
    *,
    adaptive_policy,
    fixed_policy,
    documents,
):
    metric = specification["metric"]
    adaptive_values = [metric_value(row, metric) for row in adaptive_rows]
    fixed_values = [metric_value(row, metric) for row in fixed_rows]
    difference, lo, hi = paired_mean_difference_ci(
        adaptive_values,
        fixed_values,
        n_boot=specification["bootstrap_samples"],
        seed=specification["bootstrap_seed"],
    )
    if lo > 0:
        verdict = "POSITIVE"
    elif hi < 0:
        verdict = "NEGATIVE"
    else:
        verdict = "NULL: the interval includes 0"
    return {
        "comparison": specification["comparison"],
        "metric": metric,
        "adaptive_policy": adaptive_policy,
        "fixed_policy": fixed_policy,
        "difference": difference,
        "ci95": [lo, hi],
        "documents": documents,
        "verdict": verdict,
    }


def _validate_rollout_report_v3(
    report,
    receipt,
    receipt_evidence,
    checkpoint,
    tokenizer_path,
    holdout_path,
):
    _require(
        isinstance(report, dict) and set(report) == _V3_REPORT_KEYS,
        "rollout v3 report has the wrong fields",
    )
    _require(
        report.get("schema_version") == V3_REPORT_SCHEMA_VERSION,
        f"rollout v3 schema is not {V3_REPORT_SCHEMA_VERSION}",
    )
    _require(
        report.get("instrument_version") == V3_INSTRUMENT_VERSION
        and receipt.get("instrument_version") == V3_INSTRUMENT_VERSION,
        "rollout v3 instrument differs from its receipt",
    )
    _require(
        report.get("publication_ready") is True,
        "rollout v3 report is not publication-ready",
    )
    _validate_v3_execution(report.get("execution"), receipt, receipt_evidence)
    _require(
        report.get("receipt") == receipt_evidence,
        "rollout v3 receipt evidence differs from registered input",
    )

    reported_checkpoint = report.get("checkpoint")
    _require(
        isinstance(reported_checkpoint, dict)
        and set(reported_checkpoint)
        == {"path", "step", "meta_sha256", "master_sha256"}
        and isinstance(reported_checkpoint["path"], str),
        "rollout v3 checkpoint evidence has the wrong fields",
    )
    for key in ("step", "meta_sha256", "master_sha256"):
        _require(
            reported_checkpoint.get(key) == checkpoint.get(key),
            f"rollout v3 checkpoint {key} differs from final checkpoint",
        )
    registered_checkpoint = receipt.get("checkpoint")
    _require(
        isinstance(registered_checkpoint, dict)
        and receipt_evidence.get("checkpoint") == registered_checkpoint
        and reported_checkpoint["meta_sha256"]
        == registered_checkpoint.get("meta_sha256")
        and reported_checkpoint["master_sha256"]
        == registered_checkpoint.get("master_sha256")
        and os.path.abspath(reported_checkpoint["path"])
        == os.path.abspath(registered_checkpoint.get("path", "")),
        "rollout v3 checkpoint differs from registration",
    )
    _validate_v3_artifact(report.get("tokenizer"), tokenizer_path, "tokenizer")

    holdout = report.get("holdout")
    expected_holdout_keys = {
        "path",
        "sha256",
        "pair_count",
        "pair_manifest_sha256",
        "pair_manifest",
        "pair_payload_manifest_sha256",
        "pair_payload_manifest",
        "decoy_match",
    }
    _require(
        isinstance(holdout, dict) and set(holdout) == expected_holdout_keys,
        "rollout v3 holdout evidence has the wrong fields",
    )
    _require(
        isinstance(holdout["path"], str)
        and os.path.abspath(holdout["path"]) == os.path.abspath(holdout_path)
        and holdout["sha256"] == file_sha256(holdout_path),
        "rollout v3 holdout identity differs",
    )
    pair_settings = receipt["pair_settings"]
    manifest = holdout["pair_manifest"]
    _require(
        isinstance(manifest, list)
        and len(manifest) == pair_settings["examples"],
        "rollout v3 pair manifest has the wrong size",
    )
    expected_identity_keys = {
        "index",
        "document_id",
        "decoy_document_id",
        "token_offset",
        "prompt_sha256",
        "target_sha256",
    }
    for index, identity in enumerate(manifest):
        _require(
            isinstance(identity, dict)
            and set(identity) == expected_identity_keys
            and identity.get("index") == index,
            f"rollout v3 pair identity {index} is malformed",
        )
    _require(
        holdout["pair_count"] == pair_settings["examples"]
        and holdout["pair_manifest_sha256"]
        == pair_settings["pair_manifest_sha256"]
        and _manifest_sha256(manifest)
        == pair_settings["pair_manifest_sha256"],
        "rollout v3 pair manifest differs from registered identity",
    )
    payload_manifest = holdout["pair_payload_manifest"]
    payload_digest = validate_pair_payload_manifest(
        payload_manifest,
        manifest,
        max_tokens=receipt["decoding"]["max_tokens"],
        vocab_size=receipt["model_vocab_size"],
    )
    _require(
        payload_digest == holdout["pair_payload_manifest_sha256"]
        and payload_digest
        == pair_settings["pair_payload_manifest_sha256"],
        "rollout v3 pair payload manifest differs from registration",
    )
    _require(
        holdout["decoy_match"] == {"matched": pair_settings["examples"]},
        "rollout v3 report includes relaxed decoy matches",
    )

    split = receipt["split"]
    calibration_count = split["calibration_documents"]
    _require(
        calibration_count + split["test_documents"] == len(manifest),
        "rollout v3 frozen split does not cover the pair manifest",
    )
    calibration_payloads = payload_manifest[:calibration_count]
    test_payloads = payload_manifest[calibration_count:]
    policy = receipt["policy"]
    candidate_order = [
        *policy["fixed_candidates"],
        *policy["adaptive_candidates"],
    ]
    decoding = receipt["decoding"]

    calibration = report.get("calibration")
    _require(
        isinstance(calibration, dict)
        and set(calibration)
        == {
            "documents",
            "candidate_order",
            "trajectory_identity",
            "summaries",
            "selected_fixed",
            "selected_adaptive",
            "rows",
        },
        "rollout v3 calibration evidence has the wrong fields",
    )
    calibration_rows = calibration["rows"]
    _require(
        calibration["documents"] == len(calibration_payloads)
        and calibration["candidate_order"] == candidate_order
        and isinstance(calibration_rows, dict)
        and set(calibration_rows) == set(candidate_order),
        "rollout v3 calibration split or candidate order differs",
    )
    calibration_summaries = {
        name: _validate_v3_policy_rows(
            name,
            calibration_rows[name],
            calibration_payloads,
            decoding_seed=decoding["seed"],
            max_depth=policy["max_depth"],
            max_tokens=decoding["max_tokens"],
            vocab_size=receipt["model_vocab_size"],
        )
        for name in candidate_order
    }
    _require(
        calibration["summaries"] == calibration_summaries,
        "rollout v3 calibration summaries do not recompute",
    )
    calibration_trajectory = validate_cross_policy_trajectories(
        calibration_rows, candidate_order
    )
    _require(
        calibration["trajectory_identity"] == calibration_trajectory,
        "rollout v3 calibration trajectory identity does not recompute",
    )
    selected_fixed = select_policy(
        calibration_summaries,
        policy["fixed_candidates"],
        policy["selection_metric"],
    )
    selected_adaptive = select_policy(
        calibration_summaries,
        policy["adaptive_candidates"],
        policy["selection_metric"],
    )
    _require(
        calibration["selected_fixed"] == selected_fixed
        and calibration["selected_adaptive"] == selected_adaptive,
        "rollout v3 stored policy selection differs from calibration",
    )

    selected_names = [
        selected_fixed["policy"],
        selected_adaptive["policy"],
    ]
    test = report.get("test")
    _require(
        isinstance(test, dict)
        and set(test)
        == {"documents", "trajectory_identity", "summaries", "rows"},
        "rollout v3 test evidence has the wrong fields",
    )
    test_rows = test["rows"]
    _require(
        test["documents"] == len(test_payloads)
        and isinstance(test_rows, dict)
        and set(test_rows) == set(selected_names),
        "rollout v3 test split contains the wrong policies or documents",
    )
    test_summaries = {
        name: _validate_v3_policy_rows(
            name,
            test_rows[name],
            test_payloads,
            decoding_seed=decoding["seed"],
            max_depth=policy["max_depth"],
            max_tokens=decoding["max_tokens"],
            vocab_size=receipt["model_vocab_size"],
        )
        for name in selected_names
    }
    _require(
        test["summaries"] == test_summaries,
        "rollout v3 test summaries do not recompute",
    )
    test_trajectory = validate_cross_policy_trajectories(
        test_rows, selected_names
    )
    _require(
        test["trajectory_identity"] == test_trajectory,
        "rollout v3 test trajectory identity does not recompute",
    )

    all_summaries = [
        *calibration_summaries.values(),
        *test_summaries.values(),
    ]
    scored_policy_documents = sum(
        summary["documents"] for summary in all_summaries
    )
    scored_tokens = sum(summary["total_tokens"] for summary in all_summaries)
    exact_argmax_tokens = sum(
        summary["exact_argmax_tokens"] for summary in all_summaries
    )
    certified_near_tie_tokens = sum(
        summary["certified_near_tie_tokens"] for summary in all_summaries
    )
    branch_replay_passes = sum(
        summary["branch_replay_passes"] for summary in all_summaries
    )
    cross_policy_matches = (
        calibration_trajectory["matching_documents"]
        + test_trajectory["matching_documents"]
    )
    expected_quality = {
        "reference": V3_REPLAY_REFERENCE,
        "rule": V3_REPLAY_RULE,
        "near_tie_max_ulps": NEAR_TIE_MAX_ULPS,
        "scored_policy_documents": scored_policy_documents,
        "scored_tokens": scored_tokens,
        "exact_argmax_tokens": exact_argmax_tokens,
        "certified_near_tie_tokens": certified_near_tie_tokens,
        "failed_tokens": 0,
        "branch_replay_passes": branch_replay_passes,
        "cross_policy_trajectory_matches": cross_policy_matches,
        "passed": True,
    }
    _require(
        exact_argmax_tokens + certified_near_tie_tokens == scored_tokens
        and branch_replay_passes == scored_policy_documents,
        "rollout v3 branch-local replay totals do not close",
    )
    _require(
        report.get("quality_gate") == expected_quality,
        "rollout v3 quality gate does not recompute",
    )
    cached_ar_exact = sum(
        summary["cached_ar_exact_documents"] for summary in all_summaries
    )
    expected_cached_diagnostic = {
        "scored_policy_documents": scored_policy_documents,
        "exact_output_matches": cached_ar_exact,
        "different_cached_ar_branches": (
            scored_policy_documents - cached_ar_exact
        ),
        "claim_scope": (
            "diagnostic_only; release quality is established by "
            "branch-local replay, not cached-AR byte identity"
        ),
    }
    _require(
        report.get("cached_ar_diagnostic") == expected_cached_diagnostic,
        "rollout v3 cached-AR diagnostic does not recompute",
    )

    adaptive_rows = test_rows[selected_adaptive["policy"]]
    fixed_rows = test_rows[selected_fixed["policy"]]
    expected_primary = _rollout_endpoint_result(
        receipt["test_endpoint"],
        adaptive_rows,
        fixed_rows,
        adaptive_policy=selected_adaptive["policy"],
        fixed_policy=selected_fixed["policy"],
        documents=len(test_payloads),
    )
    _require(
        report.get("primary_endpoint") == expected_primary,
        "rollout v3 primary endpoint does not recompute from test rows",
    )
    companions = receipt.get("companion_endpoints")
    _require(
        isinstance(companions, list)
        and [item.get("metric") for item in companions]
        == [
            "output_tokens_per_target_forward",
            "drafts_issued_per_output_token",
            "draft_recursions_per_output_token",
        ],
        "rollout v3 companion endpoint registration differs",
    )
    expected_companions = {
        item["metric"]: _rollout_endpoint_result(
            item,
            adaptive_rows,
            fixed_rows,
            adaptive_policy=selected_adaptive["policy"],
            fixed_policy=selected_fixed["policy"],
            documents=len(test_payloads),
        )
        for item in companions
    }
    expected_target_forward = expected_companions[
        "output_tokens_per_target_forward"
    ]
    expected_draft_issued_proxy = expected_companions[
        "drafts_issued_per_output_token"
    ]
    expected_draft_work = expected_companions[
        "draft_recursions_per_output_token"
    ]
    _require(
        report.get("secondary_target_forward_endpoint")
        == expected_target_forward,
        "rollout v3 target-forward endpoint does not recompute from test rows",
    )
    _require(
        report.get("secondary_draft_issued_proxy_endpoint")
        == expected_draft_issued_proxy,
        "rollout v3 draft-issuance proxy does not recompute from test rows",
    )
    _require(
        report.get("secondary_draft_work_endpoint") == expected_draft_work,
        "rollout v3 draft-work endpoint does not recompute from test rows",
    )
    _require(
        report.get("wall_clock_note") == _V3_WALL_CLOCK_NOTE,
        "rollout v3 wall-clock claim scope differs",
    )
    _require(
        report.get("endpoint_scope_note") == _V3_ENDPOINT_SCOPE_NOTE,
        "rollout v3 endpoint claim scope differs",
    )
    return {
        "selected_fixed": selected_fixed,
        "selected_adaptive": selected_adaptive,
        "primary_endpoint": expected_primary,
        "secondary_target_forward_endpoint": expected_target_forward,
        "secondary_draft_issued_proxy_endpoint": expected_draft_issued_proxy,
        "secondary_draft_work_endpoint": expected_draft_work,
        "quality_equivalent_to_greedy_ar": False,
        "branch_local_replay_valid": True,
        "cross_policy_trajectory_identical": True,
        "historical_instrument_only": False,
    }


def validate_rollout_report(
    report,
    receipt,
    receipt_evidence,
    checkpoint,
    tokenizer_path,
    holdout_path,
):
    instrument_version = receipt.get("instrument_version")
    _require(
        report.get("instrument_version") == instrument_version,
        "rollout instrument version differs from its receipt",
    )
    if instrument_version == V3_INSTRUMENT_VERSION:
        return _validate_rollout_report_v3(
            report,
            receipt,
            receipt_evidence,
            checkpoint,
            tokenizer_path,
            holdout_path,
        )
    if instrument_version in (1, 2):
        return _validate_rollout_report_v2(
            report,
            receipt,
            receipt_evidence,
            checkpoint,
            tokenizer_path,
            holdout_path,
        )
    raise ValueError(f"unsupported rollout instrument {instrument_version!r}")


_V3_REPLAY_ATTESTATION_ROW_DOMAIN = (
    b"WISP_E3_V3_INDEPENDENT_ROW_MANIFEST\0"
)


def _is_lower_sha256(value):
    return (
        isinstance(value, str)
        and len(value) == 64
        and all(character in "0123456789abcdef" for character in value)
    )


def _parse_attestation_time(value, label):
    _require(
        isinstance(value, str) and value.endswith("Z"),
        f"{label} is not a UTC timestamp",
    )
    try:
        parsed = datetime.fromisoformat(value.replace("Z", "+00:00"))
    except ValueError as error:
        raise ValueError(f"{label} is not ISO-8601") from error
    _require(
        parsed.utcoffset() == timezone.utc.utcoffset(parsed),
        f"{label} is not UTC",
    )
    return parsed


def _expected_attestation_rows(report):
    expected = []
    calibration = report["calibration"]
    for local_index in range(calibration["documents"]):
        for policy_name in calibration["candidate_order"]:
            expected.append(
                (
                    "calibration",
                    policy_name,
                    calibration["rows"][policy_name][local_index],
                )
            )
    selected_names = [
        calibration["selected_fixed"]["policy"],
        calibration["selected_adaptive"]["policy"],
    ]
    test = report["test"]
    for local_index in range(test["documents"]):
        for policy_name in selected_names:
            expected.append(
                ("test", policy_name, test["rows"][policy_name][local_index])
            )
    return expected


def _validate_attested_branch_quality(quality, prompt, output, row_index):
    expected_keys = {
        "input_sha256",
        "output_sha256",
        "reference_argmax_sha256",
        "aligned_logits_float32_le_sha256",
        "exact_argmax_tokens",
        "certified_near_tie_tokens",
        "failed_tokens",
        "certified_near_ties",
        "failures",
        "passed",
    }
    _require(
        isinstance(quality, dict) and set(quality) == expected_keys,
        f"independent replay row {row_index} branch-quality fields differ",
    )
    _require(
        quality["input_sha256"] == token_ids_sha256(prompt + output[:-1])
        and quality["output_sha256"] == token_ids_sha256(output)
        and _is_lower_sha256(quality["reference_argmax_sha256"])
        and _is_lower_sha256(quality["aligned_logits_float32_le_sha256"]),
        f"independent replay row {row_index} branch hashes differ",
    )
    exact = quality["exact_argmax_tokens"]
    certified_count = quality["certified_near_tie_tokens"]
    failed = quality["failed_tokens"]
    certified = quality["certified_near_ties"]
    failures = quality["failures"]
    _require(
        isinstance(exact, int)
        and not isinstance(exact, bool)
        and exact >= 0
        and isinstance(certified_count, int)
        and not isinstance(certified_count, bool)
        and certified_count >= 0
        and failed == 0
        and isinstance(certified, list)
        and len(certified) == certified_count
        and failures == []
        and quality["passed"] is True
        and exact + certified_count == len(output),
        f"independent replay row {row_index} branch counts do not close",
    )
    seen_positions = set()
    near_tie_keys = {
        "position",
        "prefix_sha256",
        "row_index",
        "emitted_token",
        "reference_argmax_token",
        "emitted_token_logit",
        "row_max_logit",
        "reference_logits_float32_le_sha256",
        "ulp_at_max",
        "max_ulps",
        "emitted_token_deficit_ulps",
    }
    for evidence in certified:
        _require(
            isinstance(evidence, dict) and set(evidence) == near_tie_keys,
            f"independent replay row {row_index} near-tie fields differ",
        )
        position = evidence["position"]
        _require(
            isinstance(position, int)
            and not isinstance(position, bool)
            and 0 <= position < len(output)
            and position not in seen_positions
            and evidence["prefix_sha256"]
            == token_ids_sha256(prompt + output[:position])
            and evidence["row_index"] == len(prompt) - 1 + position
            and evidence["emitted_token"] == output[position]
            and isinstance(evidence["reference_argmax_token"], int)
            and not isinstance(evidence["reference_argmax_token"], bool)
            and evidence["reference_argmax_token"] != output[position]
            and _is_lower_sha256(
                evidence["reference_logits_float32_le_sha256"]
            ),
            f"independent replay row {row_index} near-tie binding differs",
        )
        seen_positions.add(position)
        emitted_logit = evidence["emitted_token_logit"]
        row_max = evidence["row_max_logit"]
        _require(
            isinstance(emitted_logit, float)
            and math.isfinite(emitted_logit)
            and isinstance(row_max, float)
            and math.isfinite(row_max)
            and emitted_logit <= row_max
            and evidence["max_ulps"] == NEAR_TIE_MAX_ULPS,
            f"independent replay row {row_index} near-tie numerics differ",
        )
        ulp = bf16_ulp(row_max)
        deficit = (row_max - emitted_logit) / ulp
        _require(
            evidence["ulp_at_max"] == ulp
            and evidence["emitted_token_deficit_ulps"] == deficit
            and 0.0 <= deficit <= NEAR_TIE_MAX_ULPS,
            f"independent replay row {row_index} near-tie exceeds its gate",
        )
    return {
        "exact_argmax_tokens": exact,
        "certified_near_tie_tokens": certified_count,
        "failed_tokens": failed,
    }


def validate_rollout_replay_attestation(
    attestation,
    attestation_path,
    rollout_path,
    rollout_sha256,
    receipt_path,
    receipt_sha256,
    receipt,
    report,
    checkpoint,
    source_root,
):
    """Require a registered, independent replay of quality and execution."""
    expected_keys = {
        "schema_version",
        "instrument_version",
        "publication_ready",
        "branch_quality_independently_verified",
        "rollout_execution_reproduced",
        "passed",
        "provenance_scope",
        "verification",
        "time_order",
        "report",
        "receipt",
        "checkpoint",
        "verifier",
        "frozen_inputs",
        "row_manifest_sha256",
        "rows",
        "aggregate",
    }
    _require(
        isinstance(attestation, dict) and set(attestation) == expected_keys,
        "independent rollout replay attestation has the wrong fields",
    )
    _require(
        attestation["schema_version"] == 1
        and attestation["instrument_version"] == V3_INSTRUMENT_VERSION
        and attestation["publication_ready"] is True
        and attestation["branch_quality_independently_verified"] is True
        and attestation["rollout_execution_reproduced"] is True
        and attestation["passed"] is True,
        "independent rollout replay attestation did not pass",
    )
    _require(
        attestation["provenance_scope"] == V3_ATTESTATION_PROVENANCE_SCOPE,
        "independent rollout replay provenance scope is overstated",
    )
    registration = receipt["independent_replay_verification"]
    verification = attestation["verification"]
    _require(
        isinstance(verification, dict)
        and set(verification)
        == {
            "method",
            "started_at",
            "completed_at",
            "argv",
            "runtime",
            "timing_scope",
        }
        and verification["method"] == V3_REPLAY_VERIFIER_METHOD
        and verification["argv"] == list(V3_REPLAY_VERIFICATION_ARGV)
        and verification["argv"] == registration["execution_argv"]
        and verification["timing_scope"]
        == (
            "producer elapsed_seconds, tok_per_sec, and ar_tok_per_sec "
            "are excluded from deterministic reproduction"
        ),
        "independent rollout replay execution differs from registration",
    )
    runtime = verification["runtime"]
    _require(
        isinstance(runtime, dict)
        and set(runtime) == {"python", "mlx", "numpy", "tokenizers", "platform"}
        and {
            key: runtime[key]
            for key in ("python", "mlx", "numpy", "tokenizers")
        }
        == receipt["runtime_requirements"]
        and isinstance(runtime["platform"], str)
        and bool(runtime["platform"]),
        "independent rollout replay runtime differs from registration",
    )

    registered_at = _parse_attestation_time(
        receipt["registered_at"], "rollout receipt registration"
    )
    registration_committed = _parse_attestation_time(
        report["receipt"]["registration_git"]["committed_at"],
        "rollout receipt registration commit",
    )
    report_started = _parse_attestation_time(
        report["execution"]["started_at"], "rollout execution start"
    )
    report_completed = _parse_attestation_time(
        report["execution"]["completed_at"], "rollout execution completion"
    )
    verifier_started = _parse_attestation_time(
        verification["started_at"], "independent replay start"
    )
    verifier_completed = _parse_attestation_time(
        verification["completed_at"], "independent replay completion"
    )
    _require(
        registered_at
        <= registration_committed
        <= report_started
        <= report_completed
        <= verifier_started
        <= verifier_completed,
        "independent replay timestamps violate registration order",
    )
    expected_time_order = {
        "registered_at": receipt["registered_at"],
        "registration_committed_at": report["receipt"]["registration_git"][
            "committed_at"
        ],
        "report_started_at": report["execution"]["started_at"],
        "report_completed_at": report["execution"]["completed_at"],
        "verification_started_at": verification["started_at"],
        "verification_completed_at": verification["completed_at"],
    }
    _require(
        attestation["time_order"] == expected_time_order,
        "independent replay time-order evidence differs",
    )
    _require(
        attestation["report"]
        == {"path": rollout_path, "sha256": rollout_sha256},
        "independent replay is bound to a different rollout report",
    )
    _require(
        attestation["receipt"]
        == {
            "path": receipt_path,
            "sha256": receipt_sha256,
            "registered_at": receipt["registered_at"],
            "registration_git": report["receipt"]["registration_git"],
        },
        "independent replay is bound to a different rollout receipt",
    )
    registration_git = report["receipt"]["registration_git"]
    _require(
        isinstance(registration_git, dict)
        and set(registration_git)
        == {"commit", "committed_at", "path", "blob_sha256", "origin_ref"}
        and registration_git["blob_sha256"] == receipt_sha256
        and registration_git["path"] == "config/eval_rollout_receipt_v3.json"
        and registration_git["origin_ref"] == "origin/main",
        "independent replay registration-commit binding differs",
    )

    attested_checkpoint = attestation["checkpoint"]
    _require(
        isinstance(attested_checkpoint, dict)
        and set(attested_checkpoint)
        == {"path", "step", "meta_sha256", "master_sha256"}
        and attested_checkpoint["path"] == receipt["checkpoint"]["path"]
        and attested_checkpoint["meta_sha256"]
        == receipt["checkpoint"]["meta_sha256"]
        and attested_checkpoint["master_sha256"]
        == receipt["checkpoint"]["master_sha256"]
        and all(
            attested_checkpoint[key] == checkpoint[key]
            for key in ("step", "meta_sha256", "master_sha256")
        ),
        "independent replay checkpoint differs from the final weights",
    )

    verifier = attestation["verifier"]
    source = registration["source"]
    _require(
        isinstance(verifier, dict)
        and set(verifier)
        == {
            "method",
            "execution_argv",
            "registered_source",
            "live_source",
        }
        and verifier["method"] == registration["method"]
        and verifier["execution_argv"] == registration["execution_argv"]
        and verifier["registered_source"] == source
        and verifier["live_source"] == source,
        "independent replay verifier source differs from registration",
    )
    live_source_path = os.path.join(source_root, source["path"])
    _require(
        os.path.getsize(live_source_path) == source["bytes"]
        and file_sha256(live_source_path) == source["sha256"],
        "independent replay verifier source changed after attestation",
    )
    expected_frozen_inputs = {
        "acceptance_receipt": dict(receipt["acceptance_receipt"]),
        "holdout": dict(receipt["holdout"]),
        "tokenizer": dict(receipt["tokenizer"]),
        "pair_count": receipt["pair_settings"]["examples"],
        "pair_manifest_sha256": receipt["pair_settings"][
            "pair_manifest_sha256"
        ],
        "pair_payload_manifest_sha256": receipt["pair_settings"][
            "pair_payload_manifest_sha256"
        ],
    }
    _require(
        attestation["frozen_inputs"] == expected_frozen_inputs,
        "independent replay frozen-input evidence differs from registration",
    )

    rows = attestation["rows"]
    expected_rows = _expected_attestation_rows(report)
    _require(
        isinstance(rows, list) and len(rows) == len(expected_rows),
        "independent replay row manifest has the wrong size",
    )
    aggregate = {
        "policy_documents": 0,
        "output_tokens": 0,
        "exact_argmax_tokens": 0,
        "certified_near_tie_tokens": 0,
        "failed_tokens": 0,
        "reproduced_policy_documents": 0,
        "failed_reproductions": 0,
    }
    row_keys = {
        "verification_index",
        "split",
        "policy",
        "pair_index",
        "document_id",
        "seed",
        "prompt_sha256",
        "output_sha256",
        "reproduction",
        "branch_quality",
    }
    reproduction_keys = {
        "output_sha256",
        "trace_sha256",
        "deterministic_stats_sha256",
        "output_tokens_match",
        "generation_trace_matches",
        "deterministic_stats_match",
    }
    for index, (attested, expected) in enumerate(zip(rows, expected_rows)):
        split_name, policy_name, producer_row = expected
        prompt = report["holdout"]["pair_payload_manifest"][
            producer_row["pair_index"]
        ]["prompt_token_ids"]
        output = producer_row["output_token_ids"]
        _require(
            isinstance(attested, dict)
            and set(attested) == row_keys
            and attested["verification_index"] == index
            and attested["split"] == split_name
            and attested["policy"] == policy_name
            and attested["pair_index"] == producer_row["pair_index"]
            and attested["document_id"] == producer_row["document_id"]
            and attested["seed"] == producer_row["seed"]
            and attested["prompt_sha256"] == producer_row["prompt_sha256"]
            and attested["output_sha256"] == producer_row["output_sha256"],
            f"independent replay row {index} differs from the producer row",
        )
        reproduction = attested["reproduction"]
        _require(
            isinstance(reproduction, dict)
            and set(reproduction) == reproduction_keys
            and reproduction["output_sha256"] == producer_row["output_sha256"]
            and reproduction["trace_sha256"] == producer_row["trace_sha256"]
            and _is_lower_sha256(
                reproduction["deterministic_stats_sha256"]
            )
            and reproduction["output_tokens_match"] is True
            and reproduction["generation_trace_matches"] is True
            and reproduction["deterministic_stats_match"] is True,
            f"independent replay row {index} did not reproduce",
        )
        quality_counts = _validate_attested_branch_quality(
            attested["branch_quality"], prompt, output, index
        )
        aggregate["policy_documents"] += 1
        aggregate["output_tokens"] += len(output)
        aggregate["exact_argmax_tokens"] += quality_counts[
            "exact_argmax_tokens"
        ]
        aggregate["certified_near_tie_tokens"] += quality_counts[
            "certified_near_tie_tokens"
        ]
        aggregate["failed_tokens"] += quality_counts["failed_tokens"]
        aggregate["reproduced_policy_documents"] += 1

    _require(
        attestation["row_manifest_sha256"]
        == canonical_json_sha256(rows, _V3_REPLAY_ATTESTATION_ROW_DOMAIN),
        "independent replay row-manifest hash does not recompute",
    )
    _require(
        attestation["aggregate"] == aggregate
        and aggregate["policy_documents"]
        == report["quality_gate"]["scored_policy_documents"]
        and aggregate["output_tokens"]
        == report["quality_gate"]["scored_tokens"]
        and aggregate["reproduced_policy_documents"] == len(expected_rows)
        and aggregate["failed_reproductions"] == 0
        and aggregate["failed_tokens"] == 0
        and aggregate["exact_argmax_tokens"]
        + aggregate["certified_near_tie_tokens"]
        == aggregate["output_tokens"],
        "independent replay aggregate counts do not close",
    )
    return {
        "path": os.path.abspath(attestation_path),
        "sha256": file_sha256(attestation_path),
        "branch_quality_independently_verified": True,
        "rollout_execution_reproduced": True,
        "policy_documents": aggregate["policy_documents"],
        "output_tokens": aggregate["output_tokens"],
        "exact_argmax_tokens": aggregate["exact_argmax_tokens"],
        "certified_near_tie_tokens": aggregate[
            "certified_near_tie_tokens"
        ],
        "provenance_scope": V3_ATTESTATION_PROVENANCE_SCOPE,
    }


def _snapshot_evidence(path, digest):
    return {"path": os.path.abspath(path), "sha256": digest}


def validate_no_tampering(
    artifacts, ckpt_dir, checkpoint, ablation_ckpt_dir, ablation_checkpoint
):
    """
    The last checks before the audit is written: nothing this audit read
    changed while it was reading everything else.

    Extracted verbatim from `main()`'s own body, no behavior change, so it can
    be tested directly. `test_release_audit.py` never called `main()` at all,
    so these three checks -- including the final tamper checks that are the
    last thing standing between "audit ran" and `publication_ready: true` --
    had zero test coverage, not merely an under-varied fixture like most of
    this file's other unfalsified guards.
    """
    for item in artifacts.values():
        _require(
            file_sha256(item["path"]) == item["sha256"],
            f"release artifact changed during audit: {item['path']}",
        )
    _, final_checkpoint = checkpoint_identity(ckpt_dir)
    _require(
        final_checkpoint == checkpoint,
        "final checkpoint changed during release audit",
    )
    _, final_ablation_checkpoint = checkpoint_identity(ablation_ckpt_dir)
    _require(
        final_ablation_checkpoint == ablation_checkpoint,
        "run 2 checkpoint changed during release audit",
    )


def validate_export_model_card(
    manifest,
    export_dir,
    template_path,
    validation,
    validation_sha256,
    acceptance_comparison,
    comparison_sha256,
    format_ablation,
    format_ablation_sha256,
    rollout,
    rollout_sha256,
):
    _require(
        manifest.get("release_complete") is True,
        "export package is labelled as a development snapshot",
    )
    expected_evaluation_sources = {
        "validation": {"sha256": validation_sha256},
        "acceptance_comparison": {"sha256": comparison_sha256},
        "format_ablation": {"sha256": format_ablation_sha256},
        "rollout": {"sha256": rollout_sha256},
    }
    _require(
        manifest.get("evaluation_sources") == expected_evaluation_sources,
        "export evaluation source hashes differ from audited reports",
    )
    template_sha256 = file_sha256(template_path)
    _require(
        manifest.get("model_card_template_sha256") == template_sha256,
        "export model card template differs from audited template",
    )
    evaluation_markdown = render_evaluation_section(
        validation,
        acceptance_comparison,
        format_ablation,
        rollout,
    )
    expected_card = render_model_card(
        template_path,
        manifest["repo_id"],
        evaluation_markdown,
    )
    with open(
        os.path.join(export_dir, "README.md"), encoding="utf-8"
    ) as f:
        exported_card = f.read()
    _require(
        exported_card == expected_card,
        "exported model card does not render from audited reports",
    )
    return template_sha256


def validate_acceptance_receipt_freshness(
    acceptance_receipt, holdout_path, tokenizer_path
):
    """The acceptance receipt must still describe the holdout/tokenizer on
    disk right now, not whatever they were when the receipt was registered."""
    _require(
        acceptance_receipt.get("clean_holdout", {}).get("sha256")
        == file_sha256(holdout_path),
        "acceptance receipt does not match current clean holdout",
    )
    _require(
        acceptance_receipt.get("pair_readiness", {}).get("tokenizer_sha256")
        == file_sha256(tokenizer_path),
        "acceptance receipt does not match current tokenizer",
    )


def validate_training_data_receipt_registration(
    format_receipt, training_data_receipt_path, training_data_receipt_sha256
):
    """The training-data receipt E2 registered must be the exact same file
    this audit is reading, not a different receipt with the same shape."""
    registered_training_data = format_receipt.get(
        "training_data_receipt", {}
    )
    _require(
        os.path.abspath(registered_training_data.get("path", ""))
        == os.path.abspath(training_data_receipt_path)
        and registered_training_data.get("sha256")
        == training_data_receipt_sha256,
        "E2 and release audit use different training-data receipts",
    )


def validate_ablation_arm_checkpoint(ablation_report_checkpoint, ablation_checkpoint):
    for key in ("step", "meta_sha256", "master_sha256"):
        _require(
            ablation_report_checkpoint.get(key)
            == ablation_checkpoint.get(key),
            f"format-ablation arm checkpoint {key} differs from run 2",
        )


def validate_export_sourced_from_checkpoint(manifest, checkpoint):
    expected_source = {
        key: checkpoint[key]
        for key in (
            "step",
            "meta_sha256",
            "master_sha256",
            "optimizer_sha256",
        )
    }
    _require(
        manifest.get("source_checkpoint") == expected_source,
        "export package is not sourced from the audited final checkpoint",
    )


def validate_external_verification_paths(external, export_dir, ckpt_dir):
    _require(
        external["package"]["export_dir"] == os.path.abspath(export_dir),
        "external verification points to a different export directory",
    )
    _require(
        external["checkpoint"]["path"] == os.path.abspath(ckpt_dir),
        "external verification points to a different checkpoint directory",
    )


def main():
    parser = argparse.ArgumentParser()
    parser.add_argument("--ckpt", required=True)
    parser.add_argument("--validation", required=True)
    parser.add_argument("--trained-acceptance", required=True)
    parser.add_argument("--control-acceptance", required=True)
    parser.add_argument("--acceptance-comparison", required=True)
    parser.add_argument("--ablation-ckpt", required=True)
    parser.add_argument("--ablation-acceptance", required=True)
    parser.add_argument("--format-ablation", required=True)
    parser.add_argument("--rollout", required=True)
    parser.add_argument(
        "--rollout-verification",
        default="out/run1/rollout.replay-verification.v3.json",
    )
    parser.add_argument("--export", required=True)
    parser.add_argument("--external-verification", required=True)
    parser.add_argument(
        "--validation-receipt",
        default="config/final_validation_receipt.json",
    )
    parser.add_argument(
        "--acceptance-receipt",
        default="config/eval_holdout_receipt.json",
    )
    parser.add_argument(
        "--rollout-receipt",
        default="config/eval_rollout_receipt_v3.json",
    )
    parser.add_argument(
        "--format-ablation-receipt",
        default="config/eval_format_ablation_receipt.json",
    )
    parser.add_argument(
        "--training-data-receipt",
        default="config/training_data_receipt.json",
    )
    parser.add_argument("--config", default="config/run1.json")
    parser.add_argument("--data-index", default="data/shards/index.json")
    parser.add_argument(
        "--validation-shard", default="data/shards/val_0000.bin"
    )
    parser.add_argument("--holdout", default="data/eval/holdout.clean.jsonl")
    parser.add_argument("--tokenizer", default="tokenizer/code32k.json")
    parser.add_argument(
        "--model-card-template", default="MODEL_CARD.md"
    )
    parser.add_argument("--out", required=True)
    cli = parser.parse_args()

    _, checkpoint = checkpoint_identity(cli.ckpt)

    validation_receipt, validation_receipt_sha256 = load_json_snapshot(
        cli.validation_receipt
    )
    validation_evidence = validate_validation_receipt(
        validation_receipt,
        cli.validation_receipt,
        cli.config,
        cli.data_index,
        cli.validation_shard,
    )
    validate_validation_checkpoint(
        load_json_snapshot(os.path.join(cli.ckpt, "meta.json"))[0],
        validation_receipt,
    )
    validation, validation_sha256 = load_json_snapshot(cli.validation)
    validation_result = validate_validation_report(
        validation,
        validation_receipt,
        validation_evidence,
        checkpoint,
        cli.config,
        cli.data_index,
        cli.validation_shard,
    )

    acceptance_receipt, acceptance_receipt_sha256 = load_json_snapshot(
        cli.acceptance_receipt
    )
    validate_acceptance_receipt_freshness(
        acceptance_receipt, cli.holdout, cli.tokenizer
    )
    trained, trained_sha256 = load_json_snapshot(cli.trained_acceptance)
    control, control_sha256 = load_json_snapshot(cli.control_acceptance)
    comparison, comparison_sha256 = load_json_snapshot(
        cli.acceptance_comparison
    )
    acceptance_result = validate_acceptance_bundle(
        trained,
        trained_sha256,
        control,
        control_sha256,
        comparison,
        acceptance_receipt,
        acceptance_receipt_sha256,
        checkpoint,
    )

    format_receipt, format_receipt_sha256 = load_json_snapshot(
        cli.format_ablation_receipt
    )
    training_data_receipt, training_data_receipt_sha256 = load_json_snapshot(
        cli.training_data_receipt
    )
    training_data_evidence = validate_training_data_receipt(
        training_data_receipt,
        cli.training_data_receipt,
    )
    validate_training_data_receipt_registration(
        format_receipt, cli.training_data_receipt, training_data_receipt_sha256
    )
    ablation_meta, ablation_checkpoint = checkpoint_identity(
        cli.ablation_ckpt
    )
    e2_evidence, _ = validate_e2_evaluation_inputs(
        cli.format_ablation_receipt,
        cli.holdout,
        cli.tokenizer,
        format_receipt["evaluation_settings"],
        ablation_meta,
    )
    ablation, ablation_sha256 = load_json_snapshot(
        cli.ablation_acceptance
    )
    validate_e2_report_sampler(ablation, e2_evidence)
    validate_ablation_arm_checkpoint(
        ablation.get("checkpoint", {}), ablation_checkpoint
    )
    format_comparison, format_comparison_sha256 = load_json_snapshot(
        cli.format_ablation
    )
    format_result = validate_format_comparison(
        format_comparison,
        trained,
        trained_sha256,
        ablation,
        ablation_sha256,
        acceptance_receipt,
        acceptance_receipt_sha256,
        format_receipt,
        format_receipt_sha256,
    )

    rollout_receipt, rollout_receipt_sha256 = load_json_snapshot(
        cli.rollout_receipt
    )
    rollout_evidence = validate_rollout_receipt(
        rollout_receipt,
        cli.rollout_receipt,
        cli.acceptance_receipt,
        cli.holdout,
        cli.tokenizer,
        instrument_version=V3_INSTRUMENT_VERSION,
        source_root=os.path.dirname(
            os.path.dirname(os.path.abspath(__file__))
        ),
    )
    validate_rollout_checkpoint(
        load_json_snapshot(os.path.join(cli.ckpt, "meta.json"))[0],
        rollout_receipt,
    )
    rollout, rollout_sha256 = load_json_snapshot(cli.rollout)
    rollout_result = validate_rollout_report(
        rollout,
        rollout_receipt,
        rollout_evidence,
        checkpoint,
        cli.tokenizer,
        cli.holdout,
    )
    _require(
        rollout_result.get("branch_local_replay_valid") is True
        and rollout_result.get("cross_policy_trajectory_identical") is True
        and rollout_result.get("historical_instrument_only") is False,
        "release publication requires a valid E3 v3 rollout report",
    )
    rollout_verification, rollout_verification_sha256 = load_json_snapshot(
        cli.rollout_verification
    )
    rollout_verification_result = validate_rollout_replay_attestation(
        rollout_verification,
        cli.rollout_verification,
        cli.rollout,
        rollout_sha256,
        cli.rollout_receipt,
        rollout_receipt_sha256,
        rollout_receipt,
        rollout,
        checkpoint,
        os.path.dirname(os.path.dirname(os.path.abspath(__file__))),
    )
    _require(
        rollout_verification_result[
            "branch_quality_independently_verified"
        ]
        is True
        and rollout_verification_result["rollout_execution_reproduced"] is True,
        "release publication requires independent E3 v3 replay verification",
    )

    manifest = verify_export_manifest(cli.export)
    manifest_path = os.path.join(cli.export, "export_manifest.json")
    manifest_sha256 = file_sha256(manifest_path)
    validate_export_sourced_from_checkpoint(manifest, checkpoint)
    template_sha256 = validate_export_model_card(
        manifest,
        cli.export,
        cli.model_card_template,
        validation,
        validation_sha256,
        comparison,
        comparison_sha256,
        format_comparison,
        format_comparison_sha256,
        rollout,
        rollout_sha256,
    )
    data_document = training_data_receipt["publication_documents"][
        "data_document"
    ]
    with open(data_document["path"], encoding="utf-8") as f:
        data_document_text = f.read()
    with open(
        os.path.join(cli.export, "README.md"), encoding="utf-8"
    ) as f:
        exported_model_card_text = f.read()
    validate_publication_text(
        data_document_text,
        exported_model_card_text,
    )
    external, external_sha256 = load_json_snapshot(
        cli.external_verification
    )
    validate_external_verification_receipt(
        external,
        manifest,
        manifest_sha256,
    )
    validate_external_verification_paths(external, cli.export, cli.ckpt)

    artifacts = {
        "validation": _snapshot_evidence(cli.validation, validation_sha256),
        "trained_acceptance": _snapshot_evidence(
            cli.trained_acceptance, trained_sha256
        ),
        "control_acceptance": _snapshot_evidence(
            cli.control_acceptance, control_sha256
        ),
        "acceptance_comparison": _snapshot_evidence(
            cli.acceptance_comparison, comparison_sha256
        ),
        "ablation_acceptance": _snapshot_evidence(
            cli.ablation_acceptance, ablation_sha256
        ),
        "format_ablation": _snapshot_evidence(
            cli.format_ablation, format_comparison_sha256
        ),
        "rollout": _snapshot_evidence(cli.rollout, rollout_sha256),
        "rollout_verification": _snapshot_evidence(
            cli.rollout_verification, rollout_verification_sha256
        ),
        "external_verification": _snapshot_evidence(
            cli.external_verification, external_sha256
        ),
        "validation_receipt": _snapshot_evidence(
            cli.validation_receipt, validation_receipt_sha256
        ),
        "acceptance_receipt": _snapshot_evidence(
            cli.acceptance_receipt, acceptance_receipt_sha256
        ),
        "rollout_receipt": _snapshot_evidence(
            cli.rollout_receipt, rollout_receipt_sha256
        ),
        "format_ablation_receipt": _snapshot_evidence(
            cli.format_ablation_receipt, format_receipt_sha256
        ),
        "training_data_receipt": _snapshot_evidence(
            cli.training_data_receipt, training_data_receipt_sha256
        ),
        "ablation_data_index": _snapshot_evidence(
            e2_evidence["ablation_data_index_path"],
            e2_evidence["ablation_data_index_sha256"],
        ),
        "export_manifest": _snapshot_evidence(
            manifest_path, manifest_sha256
        ),
        "model_card_template": _snapshot_evidence(
            cli.model_card_template, template_sha256
        ),
    }
    for key, artifact in training_data_receipt[
        "publication_documents"
    ].items():
        artifacts[f"training_data_{key}"] = _snapshot_evidence(
            artifact["path"],
            training_data_evidence["publication_documents"][key],
        )
    for index, artifact in enumerate(
        e2_evidence["ablation_data_artifacts"]
    ):
        artifacts[
            f"ablation_{artifact['kind']}_{index:03d}"
        ] = _snapshot_evidence(
            artifact["path"],
            artifact["sha256"],
        )
    for key in (
        "baseline_config",
        "ablation_config",
        "training_data_receipt",
        "training_data_contract",
        "run1_shard_integrity_receipt",
        "derivation_script",
        "audit_corpus_script",
        "preparation_script",
        "corpus_script",
        "training_script",
        "checkpoint_script",
        "model_script",
        "data_script",
        "holdout",
        "tokenizer",
    ):
        registered_artifact = format_receipt[key]
        artifacts[f"format_{key}"] = _snapshot_evidence(
            registered_artifact["path"],
            registered_artifact["sha256"],
        )
    validate_no_tampering(
        artifacts, cli.ckpt, checkpoint, cli.ablation_ckpt, ablation_checkpoint
    )
    verify_export_manifest(cli.export)

    report = {
        "schema_version": SCHEMA_VERSION,
        "created_at": datetime.now(timezone.utc).isoformat(),
        "publication_ready": True,
        "research_outcomes_are_not_release_gates": True,
        "checkpoint": checkpoint,
        "ablation_checkpoint": ablation_checkpoint,
        "artifacts": artifacts,
        "validation": validation_result,
        "acceptance": acceptance_result,
        "format_ablation": format_result,
        "rollout": rollout_result,
        "rollout_verification": rollout_verification_result,
        "export": {
            "path": os.path.abspath(cli.export),
            "repo_id": manifest["repo_id"],
            "manifest_sha256": manifest_sha256,
            "payload_files": len(manifest["files"]),
        },
        "external_verification": {
            "relative_max_abs_delta": external["logits"][
                "relative_max_abs_delta"
            ],
            "relative_delta_threshold": external["logits"][
                "relative_delta_threshold"
            ],
            "argmax_agreement": external["logits"]["argmax_agreement"],
            "toolchain": external["toolchain"],
        },
        "audit_source_sha256": file_sha256(os.path.abspath(__file__)),
    }
    written = write_json_atomic(cli.out, report)
    print("release audit: PASS")
    print(f"wrote {written}")


if __name__ == "__main__":
    main()