"""Pure Python metadata helpers for the Wisp Hugging Face package.""" import hashlib import json import math import os import tempfile EOS_TOKEN = "<|endoftext|>" PAD_TOKEN = "<|pad|>" ADDITIONAL_SPECIAL_TOKENS = [ "<|fim_prefix|>", "<|fim_middle|>", "<|fim_suffix|>", "<|repo_name|>", "<|file_sep|>", ] REPO_ID_PLACEHOLDER = "{{REPO_ID}}" EVALUATION_PLACEHOLDER = "{{FINAL_EVALUATION}}" EVALUATION_SOURCE_KEYS = ( "validation", "acceptance_comparison", "format_ablation", "rollout", ) E3_ROLLOUT_SCHEMA_VERSION = 2 E3_ROLLOUT_INSTRUMENT_VERSION = 3 E3_REPLAY_REFERENCE = "branch_local_full_sequence_replay" E3_REPLAY_RULE = ( "every_emitted_token_equals_reference_argmax_or_is_a_certified_" "bf16_near_tie_on_the_same_realized_prefix" ) E3_NEAR_TIE_MAX_ULPS = 8 E3_SCORED_POLICY_DOCUMENTS = 600 E3_SCORED_TOKENS = 38400 E3_CROSS_POLICY_DOCUMENTS = 100 E3_TEST_DOCUMENTS = 60 E3_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." ) E3_ATTESTATION_PROVENANCE_SCOPE = ( "Unsigned local attestation bound to a pushed pre-execution receipt commit " "and the registered source, inputs, checkpoint, and argv; it is not a " "signed external or trusted-execution witness." ) RELEASE_FILES = [ "LICENSE", "README.md", "config.json", "generation_config.json", "model.safetensors", "mtp.safetensors", "mtp_config.json", "special_tokens_map.json", "tokenizer.json", "tokenizer_config.json", ] def llama_config(args): return { "architectures": ["LlamaForCausalLM"], "model_type": "llama", "hidden_size": args.dim, "intermediate_size": args.ffn_hidden, "num_hidden_layers": args.n_layers, "num_attention_heads": args.n_heads, "num_key_value_heads": args.n_kv_heads, "head_dim": args.head_dim, "max_position_embeddings": args.max_seq_len, "rms_norm_eps": args.norm_eps, "rope_theta": args.rope_theta, "vocab_size": args.vocab_size, "tie_word_embeddings": bool(args.tie_embeddings), "hidden_act": "silu", "attention_bias": False, "mlp_bias": False, "torch_dtype": "bfloat16", "bos_token_id": None, "eos_token_id": 0, "pad_token_id": 1, } def tokenizer_config(args): return { "tokenizer_class": "PreTrainedTokenizerFast", "model_max_length": args.max_seq_len, "clean_up_tokenization_spaces": False, "bos_token": None, "eos_token": EOS_TOKEN, "pad_token": PAD_TOKEN, "unk_token": None, "additional_special_tokens": ADDITIONAL_SPECIAL_TOKENS, } def special_tokens_map(): return { "eos_token": EOS_TOKEN, "pad_token": PAD_TOKEN, "additional_special_tokens": ADDITIONAL_SPECIAL_TOKENS, } def generation_config(): return { "_from_model_config": True, "bos_token_id": None, "eos_token_id": 0, "pad_token_id": 1, } def mtp_config(args, meta): return { "mtp_layers": args.mtp_layers, "mtp_depth_trained": args.mtp_depth, "shared_lm_head": True, "recursive": True, "note": ( "One shared MTP module applied recursively, Qwen3-Next style. It " "consumes the trunk hidden state at position i and the embedding of " "the token at i+k, and predicts the token at i+k+1. The LM head is " "shared with the trunk, which ties both computations to one output " "projection but does not guarantee close distributions. The module " "contains a transformer block whose attention was trained under a " "causal mask over the whole window: at inference it must be given " "the sequence, not a single position." ), "trained_steps": meta.get("step"), } def write_json(path, value): with open(path, "w", encoding="utf-8") as f: json.dump(value, f, indent=2, sort_keys=True) f.write("\n") def write_json_atomic(path, value): path = os.path.abspath(path) if os.path.lexists(path): raise FileExistsError(f"refusing to replace JSON artifact: {path}") parent = os.path.dirname(path) os.makedirs(parent, exist_ok=True) rendered = json.dumps(value, indent=2, sort_keys=True, allow_nan=False) temporary = None try: with tempfile.NamedTemporaryFile( "w", encoding="utf-8", prefix=f".{os.path.basename(path)}.", suffix=".tmp", dir=parent, delete=False, ) as f: temporary = f.name f.write(rendered) f.write("\n") f.flush() os.fsync(f.fileno()) if os.path.lexists(path): raise FileExistsError( f"JSON target appeared during staging: {path}" ) os.rename(temporary, path) temporary = None finally: if temporary is not None and os.path.isfile(temporary): os.unlink(temporary) return path def validate_repo_id(repo_id): if ( not isinstance(repo_id, str) or repo_id != repo_id.strip() or repo_id.count("/") != 1 or any(not part for part in repo_id.split("/")) or any(character.isspace() for character in repo_id) ): raise ValueError("--repo-id must have the form namespace/model") return repo_id def render_model_card(path, repo_id, evaluation_markdown): validate_repo_id(repo_id) with open(path, encoding="utf-8") as f: card = f.read() if REPO_ID_PLACEHOLDER not in card: raise ValueError( f"model card must contain the placeholder {REPO_ID_PLACEHOLDER}" ) if EVALUATION_PLACEHOLDER not in card: raise ValueError( f"model card must contain the placeholder {EVALUATION_PLACEHOLDER}" ) if ( not isinstance(evaluation_markdown, str) or not evaluation_markdown.strip() ): raise ValueError("model card evaluation text is empty") rendered = card.replace(REPO_ID_PLACEHOLDER, repo_id).replace( EVALUATION_PLACEHOLDER, evaluation_markdown.strip(), ) if ( REPO_ID_PLACEHOLDER in rendered or EVALUATION_PLACEHOLDER in rendered ): raise ValueError("unresolved placeholder in rendered model card") return rendered def _metric(value, label): if ( not isinstance(value, (int, float)) or isinstance(value, bool) or not math.isfinite(value) ): raise ValueError(f"{label} is not finite") return float(value) def _interval(value, label): if not isinstance(value, list) or len(value) != 2: raise ValueError(f"{label} is not a two-element interval") lo = _metric(value[0], f"{label} lower") hi = _metric(value[1], f"{label} upper") if lo > hi: raise ValueError(f"{label} is reversed") return lo, hi def _verdict(value, label): if ( not isinstance(value, str) or not value.strip() or "\n" in value or "|" in value ): raise ValueError(f"{label} is not a safe single-line verdict") return value def render_evaluation_section( validation, acceptance, format_ablation, rollout, ): """Render final model-card numbers from structured registered reports.""" if validation.get("publication_ready") is not True: raise ValueError("validation report is not publication-ready") if rollout.get("publication_ready") is not True: raise ValueError("rollout report is not publication-ready") if ( rollout.get("schema_version") != E3_ROLLOUT_SCHEMA_VERSION or rollout.get("instrument_version") != E3_ROLLOUT_INSTRUMENT_VERSION ): raise ValueError( "final metadata requires rollout schema 2 instrument 3; " "historical rollout instruments are not publication evidence" ) if acceptance.get("schema_version") != 1: raise ValueError("acceptance comparison schema is not 1") if format_ablation.get("schema_version") != 1: raise ValueError("format-ablation comparison schema is not 1") if format_ablation.get("publication_ready") is not True: raise ValueError("format-ablation comparison is not publication-ready") summary = validation.get("summary", {}) main = summary.get("main_loss", {}) main_mean = _metric(main.get("mean"), "validation main loss") main_lo, main_hi = _interval( main.get("ci95"), "validation main loss interval" ) perplexity = _metric( summary.get("main_perplexity"), "validation perplexity" ) mtp = summary.get("mtp_loss") if not isinstance(mtp, list) or len(mtp) != 2: raise ValueError("validation report does not have two MTP losses") mtp_values = [] for index, row in enumerate(mtp, 1): mean = _metric(row.get("mean"), f"MTP depth {index} loss") lo, hi = _interval( row.get("ci95"), f"MTP depth {index} loss interval" ) mtp_values.append((mean, lo, hi)) primary = acceptance.get("trained_primary_endpoint", {}) primary_ratio = _metric(primary.get("ratio"), "acceptance primary ratio") primary_lo, primary_hi = _interval( primary.get("ci95"), "acceptance primary interval" ) primary_verdict = _verdict( primary.get("verdict"), "acceptance primary verdict" ) adjusted = acceptance.get("trained_minus_untrained_ratio", {}) adjusted_difference = _metric( adjusted.get("difference"), "acceptance control-adjusted difference" ) adjusted_lo, adjusted_hi = _interval( adjusted.get("ci95"), "acceptance control-adjusted interval" ) adjusted_verdict = _verdict( adjusted.get("verdict"), "acceptance control verdict" ) combined = _verdict( acceptance.get("combined_interpretation"), "acceptance combined interpretation", ) acceptance_documents = acceptance.get("documents") if ( not isinstance(acceptance_documents, int) or isinstance(acceptance_documents, bool) or acceptance_documents < 2 ): raise ValueError("acceptance document count is invalid") format_documents = format_ablation.get("documents") if ( not isinstance(format_documents, int) or isinstance(format_documents, bool) or format_documents < 2 ): raise ValueError("format-ablation document count is invalid") format_primary = format_ablation.get("primary_endpoint", {}) format_primary_difference = _metric( format_primary.get("difference_in_differences"), "format-ablation primary difference", ) format_primary_lo, format_primary_hi = _interval( format_primary.get("ci95"), "format-ablation primary interval", ) format_primary_verdict = _verdict( format_primary.get("verdict"), "format-ablation primary verdict", ) format_secondary = format_ablation.get("secondary_endpoint", {}) format_secondary_difference = _metric( format_secondary.get("difference_in_differences"), "format-ablation secondary difference", ) format_secondary_lo, format_secondary_hi = _interval( format_secondary.get("ci95"), "format-ablation secondary interval", ) format_secondary_verdict = _verdict( format_secondary.get("verdict"), "format-ablation secondary verdict", ) format_limitation = _verdict( format_ablation.get("baseline_revision_evidence", {}).get( "limitation" ), "format-ablation source limitation", ) format_runtime_limitation = _verdict( format_ablation.get("runtime_code_evidence", {}).get("limitation"), "format-ablation runtime limitation", ) quality = rollout.get("quality_gate", {}) def quality_count(name): value = quality.get(name) if ( not isinstance(value, int) or isinstance(value, bool) or value < 0 ): raise ValueError( "rollout branch-local replay has an invalid " f"{name.replace('_', ' ')} count" ) return value scored_policy_documents = quality_count("scored_policy_documents") scored_tokens = quality_count("scored_tokens") exact_argmax_tokens = quality_count("exact_argmax_tokens") certified_near_tie_tokens = quality_count( "certified_near_tie_tokens" ) failed_tokens = quality_count("failed_tokens") branch_replay_passes = quality_count("branch_replay_passes") cross_policy_matches = quality_count( "cross_policy_trajectory_matches" ) if ( quality.get("reference") != E3_REPLAY_REFERENCE or quality.get("rule") != E3_REPLAY_RULE or quality.get("near_tie_max_ulps") != E3_NEAR_TIE_MAX_ULPS or quality.get("passed") is not True or scored_policy_documents != E3_SCORED_POLICY_DOCUMENTS or scored_tokens != E3_SCORED_TOKENS or exact_argmax_tokens + certified_near_tie_tokens + failed_tokens != scored_tokens or failed_tokens != 0 or branch_replay_passes != scored_policy_documents ): raise ValueError("rollout branch-local replay did not pass") if cross_policy_matches != E3_CROSS_POLICY_DOCUMENTS: raise ValueError("rollout cross-policy output identity did not pass") if certified_near_tie_tokens == 0: replay_cell = ( "| Branch-local greedy replay | " f"Exact argmax for all {scored_tokens} emitted tokens across " f"{scored_policy_documents} scored policy-document rollouts |" ) else: near_tie_claim = ( "1 token certified as a bfloat16 near-tie" if certified_near_tie_tokens == 1 else ( f"{certified_near_tie_tokens} tokens certified as " "bfloat16 near-ties" ) ) replay_cell = ( "| Branch-local greedy replay | " f"Exact argmax on {exact_argmax_tokens} of {scored_tokens} " f"emitted tokens; {near_tie_claim} within " f"{E3_NEAR_TIE_MAX_ULPS} ulps on " "their realized branches |" ) identity_cell = ( "| Cross-policy output identity | " f"Identical realized output branches for all {cross_policy_matches} " "calibration/test documents across compared policies |" ) rollout_primary = rollout.get("primary_endpoint", {}) if ( rollout_primary.get("comparison") != "selected_adaptive_minus_selected_fixed" or rollout_primary.get("metric") != "accepted_drafts_per_verification" ): raise ValueError("rollout primary endpoint differs from registration") rollout_difference = _metric( rollout_primary.get("difference"), "rollout primary difference" ) rollout_lo, rollout_hi = _interval( rollout_primary.get("ci95"), "rollout primary interval" ) rollout_verdict = _verdict( rollout_primary.get("verdict"), "rollout primary verdict" ) fixed_policy = _verdict( rollout_primary.get("fixed_policy"), "rollout fixed policy" ) adaptive_policy = _verdict( rollout_primary.get("adaptive_policy"), "rollout adaptive policy" ) rollout_documents = rollout_primary.get("documents") if ( not isinstance(rollout_documents, int) or isinstance(rollout_documents, bool) or rollout_documents != E3_TEST_DOCUMENTS ): raise ValueError("rollout test document count is invalid") def companion_endpoint(key, expected_metric, label): endpoint = rollout.get(key, {}) if ( endpoint.get("comparison") != "selected_adaptive_minus_selected_fixed" or endpoint.get("metric") != expected_metric or endpoint.get("adaptive_policy") != adaptive_policy or endpoint.get("fixed_policy") != fixed_policy or endpoint.get("documents") != rollout_documents ): raise ValueError(f"{label} differs from registration") difference = _metric( endpoint.get("difference"), f"{label} difference" ) lo, hi = _interval(endpoint.get("ci95"), f"{label} interval") verdict = _verdict(endpoint.get("verdict"), f"{label} verdict") return difference, lo, hi, verdict target_forward = companion_endpoint( "secondary_target_forward_endpoint", "output_tokens_per_target_forward", "rollout target-forward companion endpoint", ) draft_issued_proxy = companion_endpoint( "secondary_draft_issued_proxy_endpoint", "drafts_issued_per_output_token", "rollout draft-issuance proxy endpoint", ) draft_work = companion_endpoint( "secondary_draft_work_endpoint", "draft_recursions_per_output_token", "rollout draft-work companion endpoint", ) if rollout.get("endpoint_scope_note") != E3_ENDPOINT_SCOPE_NOTE: raise ValueError("rollout endpoint scope disclosure differs from code") return "\n".join([ "### Registered final results", "", "| Measurement | Result |", "|---|---|", ( "| Final validation main NLL | " f"{main_mean:.4f} [{main_lo:.4f}, {main_hi:.4f}], " f"perplexity {perplexity:.2f} |" ), ( "| Validation MTP depth 1 NLL | " f"{mtp_values[0][0]:.4f} " f"[{mtp_values[0][1]:.4f}, {mtp_values[0][2]:.4f}] |" ), ( "| Validation MTP depth 2 NLL | " f"{mtp_values[1][0]:.4f} " f"[{mtp_values[1][1]:.4f}, {mtp_values[1][2]:.4f}] |" ), ( "| FIM / shuffled-suffix acceptance, depth 2 | " f"{primary_ratio:.4f} [{primary_lo:.4f}, {primary_hi:.4f}], " f"{primary_verdict}, {acceptance_documents} documents |" ), ( "| Trained minus initialized acceptance-ratio lift | " f"{adjusted_difference:+.4f} " f"[{adjusted_lo:+.4f}, {adjusted_hi:+.4f}], " f"{adjusted_verdict} |" ), f"| Acceptance interpretation | {combined} |", ( "| FIM-training effect on shuffled-FIM minus L2R acceptance | " f"{format_primary_difference:+.4f} " f"[{format_primary_lo:+.4f}, {format_primary_hi:+.4f}], " f"{format_primary_verdict}, {format_documents} documents |" ), ( "| FIM-training effect on true-suffix minus shuffled-suffix " "acceptance | " f"{format_secondary_difference:+.4f} " f"[{format_secondary_lo:+.4f}, {format_secondary_hi:+.4f}], " f"{format_secondary_verdict} |" ), ( "| Adaptive minus fixed accepted drafts per verification | " f"{rollout_difference:+.4f} " f"[{rollout_lo:+.4f}, {rollout_hi:+.4f}], " f"{rollout_verdict}, {rollout_documents} test documents |" ), ( "| Adaptive minus fixed output tokens per target forward | " f"{target_forward[0]:+.4f} " f"[{target_forward[1]:+.4f}, {target_forward[2]:+.4f}], " f"{target_forward[3]}, {rollout_documents} test documents |" ), ( "| Adaptive minus fixed drafts issued per output token | " f"{draft_issued_proxy[0]:+.4f} " f"[{draft_issued_proxy[1]:+.4f}, " f"{draft_issued_proxy[2]:+.4f}], " f"{draft_issued_proxy[3]}, {rollout_documents} test documents " "(issuance proxy) |" ), ( "| Adaptive minus fixed draft recursions per output token | " f"{draft_work[0]:+.4f} " f"[{draft_work[1]:+.4f}, {draft_work[2]:+.4f}], " f"{draft_work[3]}, {rollout_documents} test documents |" ), f"| Rollout policies selected on calibration | {adaptive_policy} versus {fixed_policy} |", replay_cell, identity_cell, "", ( "Validation intervals measure Monte Carlo uncertainty from the frozen " "random-window sampler. Acceptance and rollout intervals resample " "paired target documents. They do not measure training-run or model " "uncertainty. Null and negative outcomes are retained rather than " "filtered from the release." ), "", f"Rollout endpoint scope: {E3_ENDPOINT_SCOPE_NOTE}", "", ( "Independent replay provenance scope: " f"{E3_ATTESTATION_PROVENANCE_SCOPE}" ), "", f"Format-ablation provenance limitation: {format_limitation}", "", f"Format-ablation runtime limitation: {format_runtime_limitation}", ]) def development_evaluation_section(step, max_steps): return ( "### Development snapshot\n\n" f"This package is an incomplete checkpoint at step {step} of " f"{max_steps}. It has no final registered evaluation claims and must " "not be published as the Wisp release." ) def validate_export_checkpoint(meta, allow_incomplete=False): """Reject controls, malformed metadata, and accidental snapshot releases.""" problems = [] step = meta.get("step") config = meta.get("config") model_args = meta.get("model_args") if not isinstance(step, int) or isinstance(step, bool) or step < 1: problems.append(f"checkpoint step must be a positive integer, got {step!r}") if not isinstance(config, dict): problems.append("checkpoint config is missing") config = {} if not isinstance(model_args, dict) or not model_args: problems.append("checkpoint model_args are missing") max_steps = config.get("max_steps") if not isinstance(max_steps, int) or isinstance(max_steps, bool) or max_steps < 1: problems.append( f"checkpoint config max_steps must be a positive integer, got {max_steps!r}" ) elif isinstance(step, int) and step > max_steps: problems.append(f"checkpoint step {step} exceeds max_steps {max_steps}") elif isinstance(step, int) and step != max_steps and not allow_incomplete: problems.append( f"checkpoint step {step} is not final step {max_steps}; " "use --allow-incomplete only for a development export" ) if config.get("initialization_only") is True: problems.append("an initialization-only control cannot be exported") if not config.get("run_name"): problems.append("checkpoint config run_name is missing") if step == max_steps and meta.get("optimizer_state_included") is not True: problems.append( "final checkpoint does not attest that optimizer state was saved" ) if problems: raise ValueError("checkpoint is not releasable:\n- " + "\n- ".join(problems)) return { "step": step, "max_steps": max_steps, "run_name": config["run_name"], "complete": step == max_steps, } def file_sha256(path): digest = hashlib.sha256() with open(path, "rb") as f: for block in iter(lambda: f.read(1024 * 1024), b""): digest.update(block) return digest.hexdigest() def require_unchanged_files(artifacts, activity): """Fail if any named file differs from its start-of-activity snapshot.""" for label, evidence in artifacts.items(): path = evidence.get("path") expected = evidence.get("sha256") if ( not isinstance(path, str) or not isinstance(expected, str) or file_sha256(path) != expected ): raise RuntimeError(f"{label} changed during {activity}") def write_export_manifest( out_dir, meta, n_trunk, n_mtp, repo_id, checkpoint_hashes, release_complete, evaluation_sources, model_card_template_sha256, ): files = {} for name in RELEASE_FILES: path = os.path.join(out_dir, name) if not os.path.isfile(path): raise FileNotFoundError(f"release artifact is missing {name}") files[name] = { "bytes": os.path.getsize(path), "sha256": file_sha256(path), } manifest = { "schema_version": 3, "repo_id": repo_id, "release_complete": release_complete, "evaluation_sources": evaluation_sources, "model_card_template_sha256": model_card_template_sha256, "source_checkpoint": { "step": meta.get("step"), "meta_sha256": checkpoint_hashes["meta_sha256"], "master_sha256": checkpoint_hashes["master_sha256"], "optimizer_sha256": checkpoint_hashes["optimizer_sha256"], }, "trunk_parameters": n_trunk, "mtp_parameters_excluding_shared_embedding_and_head": n_mtp, "files": files, } write_json(os.path.join(out_dir, "export_manifest.json"), manifest) def verify_export_manifest(out_dir): """Verify an export is exactly the payload recorded by its manifest.""" manifest_path = os.path.join(out_dir, "export_manifest.json") with open(manifest_path, encoding="utf-8") as f: manifest = json.load(f) if manifest.get("schema_version") != 3: raise ValueError("export manifest must use schema_version 3") release_complete = manifest.get("release_complete") if not isinstance(release_complete, bool): raise ValueError("export manifest release_complete is not boolean") template_digest = manifest.get("model_card_template_sha256") if not _is_sha256(template_digest): raise ValueError("model card template hash is not a SHA-256") evaluation_sources = manifest.get("evaluation_sources") if release_complete: if ( not isinstance(evaluation_sources, dict) or sorted(evaluation_sources) != sorted(EVALUATION_SOURCE_KEYS) ): raise ValueError( "final export does not declare exact evaluation sources" ) for key in EVALUATION_SOURCE_KEYS: if not _is_sha256(evaluation_sources.get(key, {}).get("sha256")): raise ValueError( f"evaluation source {key} hash is not a SHA-256" ) elif evaluation_sources is not None: raise ValueError( "development export must not declare final evaluation sources" ) declared = manifest.get("files") if not isinstance(declared, dict): raise ValueError("export manifest has no files object") if sorted(declared) != RELEASE_FILES: raise ValueError( f"export manifest payload differs from required files: {sorted(declared)}" ) actual_names = sorted(os.listdir(out_dir)) expected_names = sorted([*RELEASE_FILES, "export_manifest.json"]) if actual_names != expected_names: raise ValueError( f"export directory contains unexpected or missing files: {actual_names}" ) for name in RELEASE_FILES: path = os.path.join(out_dir, name) actual = { "bytes": os.path.getsize(path), "sha256": file_sha256(path), } if declared[name] != actual: raise ValueError( f"release artifact {name} does not match export manifest" ) checkpoint = manifest.get("source_checkpoint") if not isinstance(checkpoint, dict): raise ValueError("export manifest has no source_checkpoint") step = checkpoint.get("step") if not isinstance(step, int) or isinstance(step, bool) or step < 1: raise ValueError("source_checkpoint.step is not a positive integer") for key in ("meta_sha256", "master_sha256", "optimizer_sha256"): digest = checkpoint.get(key) valid_digest = ( isinstance(digest, str) and len(digest) == 64 and all( character in "0123456789abcdef" for character in digest.lower() ) ) if not valid_digest: raise ValueError(f"source_checkpoint.{key} is not a SHA-256") repo_id = validate_repo_id(manifest.get("repo_id")) with open(os.path.join(out_dir, "README.md"), encoding="utf-8") as f: card = f.read() if ( REPO_ID_PLACEHOLDER in card or EVALUATION_PLACEHOLDER in card or repo_id not in card ): raise ValueError( "rendered model card does not match export manifest repo_id" ) return manifest def validate_external_verification_receipt(receipt, manifest, manifest_sha256): """Verify a structured Transformers comparison receipt fails closed.""" problems = [] if receipt.get("schema_version") != 1: problems.append("schema_version is not 1") if receipt.get("status") != "verified" or receipt.get("passed") is not True: problems.append("receipt is not a verified pass") package = receipt.get("package", {}) if package.get("export_manifest_sha256") != manifest_sha256: problems.append("export manifest hash does not match") if package.get("repo_id") != manifest.get("repo_id"): problems.append("repository ID does not match export manifest") if package.get("source_checkpoint") != manifest.get("source_checkpoint"): problems.append("source checkpoint does not match export manifest") export_dir = package.get("export_dir") if not isinstance(export_dir, str) or not os.path.isabs(export_dir): problems.append("export directory is not an absolute path") checkpoint = receipt.get("checkpoint", {}) source_checkpoint = manifest.get("source_checkpoint", {}) checkpoint_hashes = { key: checkpoint.get(key) for key in ("meta_sha256", "master_sha256", "optimizer_sha256") } expected_hashes = { key: source_checkpoint.get(key) for key in ("meta_sha256", "master_sha256", "optimizer_sha256") } if checkpoint_hashes != expected_hashes: problems.append("verified checkpoint hashes do not match package source") checkpoint_path = checkpoint.get("path") if not isinstance(checkpoint_path, str) or not os.path.isabs(checkpoint_path): problems.append("checkpoint path is not absolute") probe = receipt.get("probe", {}) probe_length = probe.get("length") probe_digest = probe.get("token_ids_sha256") if ( probe.get("seed") != 0 or not isinstance(probe_length, int) or isinstance(probe_length, bool) or probe_length < 1 or not _is_sha256(probe_digest) ): problems.append("deterministic token probe is malformed") tokenizer = receipt.get("tokenizer", {}) if ( tokenizer.get("special_token_ids") != list(range(7)) or tokenizer.get("eos_token_id") != 0 or tokenizer.get("pad_token_id") != 1 or tokenizer.get("byte_roundtrip_exact") is not True or not _is_sha256(tokenizer.get("probe_sha256")) ): problems.append("tokenizer verification is incomplete") logits = receipt.get("logits", {}) shape = logits.get("shape") max_delta = logits.get("max_abs_delta") scale = logits.get("logit_scale") relative = logits.get("relative_max_abs_delta") threshold = logits.get("relative_delta_threshold") agreement = logits.get("argmax_agreement") if ( not isinstance(shape, list) or len(shape) != 2 or any( not isinstance(item, int) or isinstance(item, bool) or item < 1 for item in shape ) or shape[0] != probe_length ): problems.append("logit shape does not match deterministic probe") if ( not _is_nonnegative_finite(max_delta) or not _is_nonnegative_finite(scale) or not _is_nonnegative_finite(relative) or threshold != 1e-2 or relative >= threshold ): problems.append("relative logit delta did not pass its threshold") expected_relative = ( max_delta / max(scale, 1e-9) if _is_nonnegative_finite(max_delta) and _is_nonnegative_finite(scale) else None ) if ( expected_relative is None or not _is_nonnegative_finite(relative) or not math.isclose(relative, expected_relative, rel_tol=1e-12, abs_tol=0.0) ): problems.append("relative logit delta is inconsistent with raw metrics") if agreement != 1.0: problems.append("argmax agreement is not exact") toolchain = receipt.get("toolchain", {}) for key in ("python", "platform", "mlx", "torch", "transformers"): if not isinstance(toolchain.get(key), str) or not toolchain[key].strip(): problems.append(f"toolchain {key} is missing") if problems: raise ValueError( "external verification receipt is invalid:\n- " + "\n- ".join(problems) ) return receipt def _is_sha256(value): return ( isinstance(value, str) and len(value) == 64 and all(character in "0123456789abcdef" for character in value.lower()) ) def _is_nonnegative_finite(value): return ( isinstance(value, (int, float)) and not isinstance(value, bool) and math.isfinite(value) and value >= 0 )