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