Module 1. The Modern LLM Landscape

import importlib.util
import subprocess
import sys

missing = [name for name in ("pandas", "matplotlib") if importlib.util.find_spec(name) is None]
if missing:
    subprocess.check_call([sys.executable, "-m", "pip", "install", "-q", *missing])

import copy
import hashlib
import os
import json
import math
import re
from dataclasses import asdict, dataclass
from numbers import Integral, Real
from pathlib import Path
from typing import Any, Callable, Iterable, Mapping, Sequence

import matplotlib.pyplot as plt
import pandas as pd

pd.set_option("display.width", 180)
pd.set_option("display.max_colwidth", 48)
pd.set_option("display.float_format", lambda value: f"{value:,.3f}")
TEXT_COLOR, MUTED, BORDER = "#2E3440", "#4C566A", "#D8DEE9"
PRIMARY, ORANGE = "#5E81AC", "#D08770"

DOCUMENTS = {'A': 'Keep an experiment log for 30 days after the experiment ends.', 'B': 'Delete a temporary export 7 days after it is created.'}
QUESTION = 'How many days must an experiment log be kept after the experiment ends?'
print(QUESTION)
for name, passage in DOCUMENTS.items():
    print(name, passage)

def check_prepared_answer(answer, documents, reference):
    """Check a structured day count and an exact reference passage for one fixed question."""
    if type(answer.get("days")) is not int or answer["days"] < 0:
        raise ValueError("days must be a nonnegative integer")
    doc_id = answer.get("document_id")
    quote = answer.get("quote")
    if not isinstance(doc_id, str) or not isinstance(quote, str):
        raise ValueError("document_id and quote must be strings")
    answer_correct = answer["days"] == reference["days"]
    quote_exists = bool(quote) and doc_id in documents and quote in documents[doc_id]
    # The required passage is manually labelled for this question; it is not inferred.
    supported = (quote_exists and answer_correct
                 and doc_id == reference["document_id"]
                 and reference["required_quote"] in quote)
    return dict(answer_correct=answer_correct, quote_exists=quote_exists,
                supported=supported, accepted=answer_correct and supported)

REFERENCE_ANSWER = {'days': 30, 'document_id': 'A', 'required_quote': 'Keep an experiment log for 30 days after the experiment ends.'}
prepared = [{'days': 30, 'document_id': 'A'}, {'days': 7, 'document_id': 'A'}, {'days': 30, 'document_id': 'B'}]
answers = [dict(row, quote=DOCUMENTS[row["document_id"]]) for row in prepared]
checks = [check_prepared_answer(row, DOCUMENTS, REFERENCE_ANSWER) for row in answers]
labels = ['Correct answer and source', 'Wrong retention period', 'Source about another object']
for label, result in zip(labels, checks):
    print(label, result)
assert checks == [{'answer_correct': True, 'quote_exists': True, 'supported': True, 'accepted': True}, {'answer_correct': False, 'quote_exists': True, 'supported': False, 'accepted': False}, {'answer_correct': True, 'quote_exists': True, 'supported': False, 'accepted': False}]

total = 8_000_000_000
active = 2_000_000_000
assert type(total) is int and type(active) is int and 0 < active <= total
bf16_bytes = total * 2
raw_4bit_bytes = (total + 1) // 2
weight_flop_per_token = 2 * active
print("BF16, GiB:", bf16_bytes / 2**30)
print("raw 4-bit, GiB:", raw_4bit_bytes / 2**30)
print("GFLOP/token:", weight_flop_per_token / 10**9)
# These numerical checks belong to the stated reference dimensions.
if total == 8_000_000_000:
    assert bf16_bytes == 16_000_000_000
    assert raw_4bit_bytes == 4_000_000_000
if active == 2_000_000_000:
    assert weight_flop_per_token == 4_000_000_000

from decimal import Decimal

def task_cost(row, tasks):
    """Charge every attempt; count each accepted task once. None means no successes."""
    for key in ("input_tokens", "output_tokens", "successes"):
        if type(row[key]) is not int or row[key] < 0:
            raise ValueError(f"{key} must be a nonnegative integer")
    if type(tasks) is not int or tasks <= 0 or row["successes"] > tasks:
        raise ValueError("invalid task count or success count")
    prices = [Decimal(str(row[k])) for k in
              ("input_usd_per_million", "output_usd_per_million", "tool_usd")]
    if any(not p.is_finite() or p < 0 for p in prices):
        raise ValueError("costs must be finite and nonnegative")
    input_price, output_price, tool_cost = prices
    total = (row["input_tokens"] * input_price + row["output_tokens"] * output_price) / Decimal(1_000_000) + tool_cost
    accepted = row["successes"]
    return {
        "name": row["name"],
        "total_usd": total,
        "success_rate": Decimal(accepted) / tasks,
        "usd_per_success": total / accepted if accepted else None,
    }

SCENARIO = {'schema_version': 1, 'kind': 'synthetic_teaching_example', 'date': '2026-10-10', 'tasks': 100, 'minimum_successes': 80, 'scope': 'All billed attempts on the same held-out tasks; no cache, batch discount, infrastructure or human-review cost. Prices and outcomes are fictional.', 'candidates': [{'name': 'A', 'input_tokens': 1000000, 'output_tokens': 1000000, 'input_usd_per_million': '1', 'output_usd_per_million': '2', 'tool_usd': '0.60', 'successes': 60}, {'name': 'B', 'input_tokens': 200000, 'output_tokens': 200000, 'input_usd_per_million': '2', 'output_usd_per_million': '10', 'tool_usd': '0.40', 'successes': 80}]}
results = [task_cost(row, SCENARIO["tasks"]) for row in SCENARIO["candidates"]]
display(pd.DataFrame(results).set_index("name"))
assert [row["total_usd"] for row in results] == [Decimal("3.60"), Decimal("2.80")]
assert [row["usd_per_success"] for row in results] == [Decimal("0.060"), Decimal("0.035")]
eligible = [row["name"] for row in SCENARIO["candidates"]
            if row["successes"] >= SCENARIO["minimum_successes"]]
assert eligible == ["B"]
print("eligible:", eligible)
no_success = task_cost(dict(SCENARIO["candidates"][0], successes=0), SCENARIO["tasks"])
assert no_success["usd_per_success"] is None
# Exercise E5: a cheaper token is not enough, and a lower ratio is not a quality gate.
worse_b = task_cost(dict(SCENARIO["candidates"][1], successes=40), SCENARIO["tasks"])
assert worse_b["usd_per_success"] == Decimal("0.070")

class ConfigSchemaError(ValueError):
    """Raised when a model configuration violates the supported teaching schema."""


class ConfigIntegrityError(ValueError):
    """Raised when a config digest differs from the explicitly expected digest."""


def _positive_int(name: str, value: object) -> int:
    """Return a strictly positive integer and reject booleans explicitly."""

    if isinstance(value, bool) or not isinstance(value, Integral):
        raise ConfigSchemaError(f"{name} must be an integer")
    result = int(value)
    if result <= 0:
        raise ConfigSchemaError(f"{name} must be positive")
    return result


def _non_negative_int(name: str, value: object) -> int:
    """Return a non-negative integer and reject booleans explicitly."""

    if isinstance(value, bool) or not isinstance(value, Integral):
        raise ConfigSchemaError(f"{name} must be an integer")
    result = int(value)
    if result < 0:
        raise ConfigSchemaError(f"{name} must be non-negative")
    return result


def _finite_positive_real(name: str, value: object) -> float:
    """Return a positive finite real number."""

    if isinstance(value, bool) or not isinstance(value, Real):
        raise ConfigSchemaError(f"{name} must be a real scalar")
    result = float(value)
    if not math.isfinite(result) or result <= 0.0:
        raise ConfigSchemaError(f"{name} must be finite and positive")
    return result


def _string(name: str, value: object) -> str:
    """Return a non-empty string stripped of surrounding whitespace."""

    if not isinstance(value, str):
        raise ConfigSchemaError(f"{name} must be a string")
    result = value.strip()
    if not result:
        raise ConfigSchemaError(f"{name} must be non-empty")
    return result


def _mapping(name: str, value: object) -> dict[str, Any]:
    """Return a shallow dictionary copy of a string-keyed mapping."""

    if not isinstance(value, Mapping):
        raise ConfigSchemaError(f"{name} must be an object")
    result = dict(value)
    if any(not isinstance(key, str) for key in result):
        raise ConfigSchemaError(f"{name} keys must be strings")
    return result


def _synonym_value(
    cfg: Mapping[str, Any],
    keys: Sequence[str],
    *,
    default: object = None,
) -> object:
    """Return one synonym value and reject conflicting aliases."""

    present = [(key, cfg[key]) for key in keys if key in cfg and cfg[key] is not None]
    if not present:
        return default
    first_key, first_value = present[0]
    conflicts = [key for key, value in present[1:] if value != first_value]
    if conflicts:
        names = ", ".join([first_key, *conflicts])
        raise ConfigSchemaError(f"conflicting synonym fields: {names}")
    return first_value


def strict_json_loads(text: object) -> dict[str, Any]:
    """Parse a JSON object while rejecting duplicate keys and non-finite constants."""

    if not isinstance(text, str) or not text.strip():
        raise ConfigSchemaError("config JSON must be a non-empty string")

    def object_pairs_hook(pairs: list[tuple[str, Any]]) -> dict[str, Any]:
        """Build one object while rejecting a repeated key at its own nesting level."""

        result: dict[str, Any] = {}
        for key, value in pairs:
            if key in result:
                raise ConfigSchemaError(f"duplicate JSON key: {key}")
            result[key] = value
        return result

    def parse_constant(value: str) -> object:
        """Reject JavaScript-style non-finite constants accepted by Python's parser."""

        raise ConfigSchemaError(f"non-standard JSON constant is forbidden: {value}")

    try:
        payload = json.loads(
            text,
            object_pairs_hook=object_pairs_hook,
            parse_constant=parse_constant,
        )
    except ConfigSchemaError:
        raise
    except json.JSONDecodeError as exc:
        raise ConfigSchemaError(f"invalid JSON: {exc.msg}") from exc
    if not isinstance(payload, dict):
        raise ConfigSchemaError("config JSON must contain a top-level object")
    return payload


def canonical_json(value: object) -> str:
    """Serialize JSON-compatible data in one deterministic UTF-8 representation."""

    return json.dumps(value, ensure_ascii=False, sort_keys=True, separators=(",", ":"))


def sha256_json(value: object) -> str:
    """Return SHA-256 of canonical JSON bytes."""

    return hashlib.sha256(canonical_json(value).encode("utf-8")).hexdigest()

REFERENCE_JSON = '{"context_milestones":[{"model":"GPT-2","tokens":1024,"year":2019.1},{"model":"GPT-3","tokens":2048,"year":2020.4},{"model":"GPT-3.5","tokens":4096,"year":2022.9},{"model":"GPT-4","tokens":32768,"year":2023.2},{"model":"Claude","tokens":100000,"year":2023.4},{"model":"GPT-4 Turbo","tokens":128000,"year":2023.9},{"model":"Gemini 1.5 Pro","tokens":1000000,"year":2024.15},{"model":"Claude 3","tokens":200000,"year":2024.2},{"model":"Llama 4 Scout","tokens":10000000,"year":2025.27},{"model":"Kimi K3","tokens":1048576,"year":2026.57},{"model":"DeepSeek V4.1 Flash","tokens":1000000,"year":2026.69},{"model":"GPT-6.1 Sol","tokens":1050000,"year":2026.74}],"cost_assumptions":{"bf16_bits":16.0,"card_capacity_gb_decimal":80.0,"four_bit_effective_bits":4.25},"models":[{"active_convention":"core_plus_unembedding","active_tolerance_pct":3.0,"claimed_active_b":3.6,"claimed_total_b":21.0,"config":{"architectures":["GptOssForCausalLM"],"head_dim":64,"hidden_size":2880,"intermediate_size":2880,"max_position_embeddings":131072,"num_attention_heads":64,"num_experts_per_tok":4,"num_hidden_layers":24,"num_key_value_heads":8,"num_local_experts":32,"rope_scaling":{"beta_fast":32.0,"beta_slow":1.0,"factor":32.0,"original_max_position_embeddings":4096,"rope_type":"yarn","truncate":false},"rope_theta":150000,"sliding_window":128,"tie_word_embeddings":false,"vocab_size":201088},"config_sha256":"11104a535b8a87bb1bb5d6cd35fb299491f48b9cc347ebce744dcc33fd3da5bc","display_name":"gpt-oss-20b","expected":{"active_params":3608307264,"attention":"GQA 64/8 + sliding 128","context":"131,072 (yarn ×32.0 from 4,096)","delta_active_pct":0.23075733333333126,"delta_total_pct":-0.4059181714285759,"ffn":"MoE 32 × top-4","total_params":20914757184},"provenance":{"config_url":"https://huggingface.co/openai/gpt-oss-20b/resolve/6cee5e81ee83917806bbde320786a8fb61efebee/config.json","count_scope":"Main autoregressive model; DeepSeek-V3 MTP weights are excluded; counts are logical parameters, not packed tensor elements.","model_card_url":"https://huggingface.co/openai/gpt-oss-20b/blob/6cee5e81ee83917806bbde320786a8fb61efebee/README.md","projection":"Retain only fields used by the teaching schema; nested rope_scaling is retained intact. Missing head_dim is derived from hidden_size / num_attention_heads.","raw_config_file":"configs/openai--gpt-oss-20b.json","raw_config_sha256":"3a2a26ded679375b7928ddeca59764df7cea83220c1961035f6d6e232659e9ce","revision":"6cee5e81ee83917806bbde320786a8fb61efebee","verified_on":"2026-10-11"},"repo_id":"openai/gpt-oss-20b","total_tolerance_pct":1.5},{"active_convention":"core_plus_unembedding","active_tolerance_pct":3.0,"claimed_active_b":5.1,"claimed_total_b":117.0,"config":{"architectures":["GptOssForCausalLM"],"head_dim":64,"hidden_size":2880,"intermediate_size":2880,"max_position_embeddings":131072,"num_attention_heads":64,"num_experts_per_tok":4,"num_hidden_layers":36,"num_key_value_heads":8,"num_local_experts":128,"rope_scaling":{"beta_fast":32.0,"beta_slow":1.0,"factor":32.0,"original_max_position_embeddings":4096,"rope_type":"yarn","truncate":false},"rope_theta":150000,"sliding_window":128,"tie_word_embeddings":false,"vocab_size":201088},"config_sha256":"74dfcb4d69ea006218ba25fe0096d2e9dc03c7a2ea2b5ce8f3feba4501d5cfb4","display_name":"gpt-oss-120b","expected":{"active_params":5132849472,"attention":"GQA 64/8 + sliding 128","context":"131,072 (yarn ×32.0 from 4,096)","delta_active_pct":0.6441072941176578,"delta_total_pct":-0.1460199384615416,"ffn":"MoE 128 × top-4","total_params":116829156672},"provenance":{"config_url":"https://huggingface.co/openai/gpt-oss-120b/resolve/b5c939de8f754692c1647ca79fbf85e8c1e70f8a/config.json","count_scope":"Main autoregressive model; DeepSeek-V3 MTP weights are excluded; counts are logical parameters, not packed tensor elements.","model_card_url":"https://huggingface.co/openai/gpt-oss-120b/blob/b5c939de8f754692c1647ca79fbf85e8c1e70f8a/README.md","projection":"Retain only fields used by the teaching schema; nested rope_scaling is retained intact. Missing head_dim is derived from hidden_size / num_attention_heads.","raw_config_file":"configs/openai--gpt-oss-120b.json","raw_config_sha256":"933aeb666a3fd851133ddd7686414f369bc564c4185fb5704416550879f10566","revision":"b5c939de8f754692c1647ca79fbf85e8c1e70f8a","verified_on":"2026-10-11"},"repo_id":"openai/gpt-oss-120b","total_tolerance_pct":1.5},{"active_convention":"total","active_tolerance_pct":3.0,"claimed_active_b":8.2,"claimed_total_b":8.2,"config":{"architectures":["Qwen3ForCausalLM"],"head_dim":128,"hidden_size":4096,"intermediate_size":12288,"max_position_embeddings":40960,"num_attention_heads":32,"num_hidden_layers":36,"num_key_value_heads":8,"rope_theta":1000000,"tie_word_embeddings":false,"vocab_size":151936},"config_sha256":"2f6b8ee2713f204b479ecfccb1ea067b75c1f992e592f80050856eca832676a8","display_name":"Qwen3-8B","expected":{"active_params":8190735360,"attention":"GQA 32/8","context":"40,960","delta_active_pct":-0.1129834146341415,"delta_total_pct":-0.1129834146341415,"ffn":"dense","total_params":8190735360},"provenance":{"config_url":"https://huggingface.co/Qwen/Qwen3-8B/resolve/b968826d9c46dd6066d109eabc6255188de91218/config.json","count_scope":"Main autoregressive model; DeepSeek-V3 MTP weights are excluded; counts are logical parameters, not packed tensor elements.","model_card_url":"https://huggingface.co/Qwen/Qwen3-8B/blob/b968826d9c46dd6066d109eabc6255188de91218/README.md","projection":"Retain only fields used by the teaching schema; nested rope_scaling is retained intact. Missing head_dim is derived from hidden_size / num_attention_heads.","raw_config_file":"configs/Qwen--Qwen3-8B.json","raw_config_sha256":"f7c4eadfbbf522470667b797a3c89be2524832d2d599797248dc304fff447c30","revision":"b968826d9c46dd6066d109eabc6255188de91218","verified_on":"2026-10-11"},"repo_id":"Qwen/Qwen3-8B","total_tolerance_pct":1.5},{"active_convention":"total","active_tolerance_pct":3.0,"claimed_active_b":14.8,"claimed_total_b":14.8,"config":{"architectures":["Qwen3ForCausalLM"],"head_dim":128,"hidden_size":5120,"intermediate_size":17408,"max_position_embeddings":40960,"num_attention_heads":40,"num_hidden_layers":40,"num_key_value_heads":8,"rope_theta":1000000,"tie_word_embeddings":false,"vocab_size":151936},"config_sha256":"7860c198333491c74a58ffb86dc2777c19fa052a87b15c724f2a8255d85654ed","display_name":"Qwen3-14B","expected":{"active_params":14768307200,"attention":"GQA 40/8","context":"40,960","delta_active_pct":-0.21414054054054105,"delta_total_pct":-0.21414054054054105,"ffn":"dense","total_params":14768307200},"provenance":{"config_url":"https://huggingface.co/Qwen/Qwen3-14B/resolve/40c069824f4251a91eefaf281ebe4c544efd3e18/config.json","count_scope":"Main autoregressive model; DeepSeek-V3 MTP weights are excluded; counts are logical parameters, not packed tensor elements.","model_card_url":"https://huggingface.co/Qwen/Qwen3-14B/blob/40c069824f4251a91eefaf281ebe4c544efd3e18/README.md","projection":"Retain only fields used by the teaching schema; nested rope_scaling is retained intact. Missing head_dim is derived from hidden_size / num_attention_heads.","raw_config_file":"configs/Qwen--Qwen3-14B.json","raw_config_sha256":"e73c3664ca09b10a673fef0c22e8a6b456201d49bd4713c9691f775720e8857a","revision":"40c069824f4251a91eefaf281ebe4c544efd3e18","verified_on":"2026-10-11"},"repo_id":"Qwen/Qwen3-14B","total_tolerance_pct":1.5},{"active_convention":"total","active_tolerance_pct":3.0,"claimed_active_b":32.8,"claimed_total_b":32.8,"config":{"architectures":["Qwen3ForCausalLM"],"head_dim":128,"hidden_size":5120,"intermediate_size":25600,"max_position_embeddings":40960,"num_attention_heads":64,"num_hidden_layers":64,"num_key_value_heads":8,"rope_theta":1000000,"tie_word_embeddings":false,"vocab_size":151936},"config_sha256":"95c5ebf076a44ba2847a43dd5a4cc12bebc54de9f36bee4f8b57e59669242647","display_name":"Qwen3-32B","expected":{"active_params":32762123264,"attention":"GQA 64/8","context":"40,960","delta_active_pct":-0.11547785365851855,"delta_total_pct":-0.11547785365851855,"ffn":"dense","total_params":32762123264},"provenance":{"config_url":"https://huggingface.co/Qwen/Qwen3-32B/resolve/9216db5781bf21249d130ec9da846c4624c16137/config.json","count_scope":"Main autoregressive model; DeepSeek-V3 MTP weights are excluded; counts are logical parameters, not packed tensor elements.","model_card_url":"https://huggingface.co/Qwen/Qwen3-32B/blob/9216db5781bf21249d130ec9da846c4624c16137/README.md","projection":"Retain only fields used by the teaching schema; nested rope_scaling is retained intact. Missing head_dim is derived from hidden_size / num_attention_heads.","raw_config_file":"configs/Qwen--Qwen3-32B.json","raw_config_sha256":"97e295b63283935788fac5e4f8860862a56d4089538cafc93f0431f2ebe483bb","revision":"9216db5781bf21249d130ec9da846c4624c16137","verified_on":"2026-10-11"},"repo_id":"Qwen/Qwen3-32B","total_tolerance_pct":1.5},{"active_convention":"core_plus_embeddings","active_tolerance_pct":3.0,"claimed_active_b":3.3,"claimed_total_b":30.5,"config":{"architectures":["Qwen3MoeForCausalLM"],"head_dim":128,"hidden_size":2048,"intermediate_size":6144,"max_position_embeddings":40960,"moe_intermediate_size":768,"num_attention_heads":32,"num_experts":128,"num_experts_per_tok":8,"num_hidden_layers":48,"num_key_value_heads":4,"rope_theta":1000000.0,"tie_word_embeddings":false,"vocab_size":151936},"config_sha256":"d095ec5feb2ceff207abb531d1b88a900c2e63d4c0f615ae9251886a3173d2df","display_name":"Qwen3-30B-A3B","expected":{"active_params":3353032704,"attention":"GQA 32/4","context":"40,960","delta_active_pct":1.6070516363636378,"delta_total_pct":0.10532007868852274,"ffn":"MoE 128 × top-8","total_params":30532122624},"provenance":{"config_url":"https://huggingface.co/Qwen/Qwen3-30B-A3B/resolve/ad44e777bcd18fa416d9da3bd8f70d33ebb85d39/config.json","count_scope":"Main autoregressive model; DeepSeek-V3 MTP weights are excluded; counts are logical parameters, not packed tensor elements.","model_card_url":"https://huggingface.co/Qwen/Qwen3-30B-A3B/blob/ad44e777bcd18fa416d9da3bd8f70d33ebb85d39/README.md","projection":"Retain only fields used by the teaching schema; nested rope_scaling is retained intact. Missing head_dim is derived from hidden_size / num_attention_heads.","raw_config_file":"configs/Qwen--Qwen3-30B-A3B.json","raw_config_sha256":"2850ddb3bf7aecad20b611e2d44f3077fc8193f4827c93beddd4c02ad63c2297","revision":"ad44e777bcd18fa416d9da3bd8f70d33ebb85d39","verified_on":"2026-10-11"},"repo_id":"Qwen/Qwen3-30B-A3B","total_tolerance_pct":1.5},{"active_convention":"core_plus_embeddings","active_tolerance_pct":3.0,"claimed_active_b":12.9,"claimed_total_b":46.7,"config":{"architectures":["MixtralForCausalLM"],"hidden_size":4096,"intermediate_size":14336,"max_position_embeddings":32768,"num_attention_heads":32,"num_experts_per_tok":2,"num_hidden_layers":32,"num_key_value_heads":8,"num_local_experts":8,"rope_theta":1000000.0,"tie_word_embeddings":false,"vocab_size":32000},"config_sha256":"7580709b5dae3e23ef3135c2badbed683d27e08228296a00543dd3e8a040732e","display_name":"Mixtral-8x7B-v0.1","expected":{"active_params":12879925248,"attention":"GQA 32/8","context":"32,768","delta_active_pct":-0.15561823255814794,"delta_total_pct":0.0059800942184020595,"ffn":"MoE 8 × top-2","total_params":46702792704},"provenance":{"config_url":"https://huggingface.co/mistralai/Mixtral-8x7B-v0.1/resolve/fc7ac94680e38d7348cfa806e51218e6273104b0/config.json","count_scope":"Main autoregressive model; DeepSeek-V3 MTP weights are excluded; counts are logical parameters, not packed tensor elements.","model_card_url":"https://huggingface.co/mistralai/Mixtral-8x7B-v0.1/blob/fc7ac94680e38d7348cfa806e51218e6273104b0/README.md","projection":"Retain only fields used by the teaching schema; nested rope_scaling is retained intact. Missing head_dim is derived from hidden_size / num_attention_heads.","raw_config_file":"configs/mistralai--Mixtral-8x7B-v0.1.json","raw_config_sha256":"9d56d04b36d0fd12ff54ae4c5bac769cc176e254e64ff71144614b6318b40793","revision":"fc7ac94680e38d7348cfa806e51218e6273104b0","verified_on":"2026-10-11"},"repo_id":"mistralai/Mixtral-8x7B-v0.1","total_tolerance_pct":1.5},{"active_convention":"core_plus_unembedding","active_tolerance_pct":3.0,"claimed_active_b":37.0,"claimed_total_b":671.0,"config":{"architectures":["DeepseekV3ForCausalLM"],"first_k_dense_replace":3,"hidden_size":7168,"intermediate_size":18432,"kv_lora_rank":512,"max_position_embeddings":163840,"moe_intermediate_size":2048,"n_routed_experts":256,"n_shared_experts":1,"num_attention_heads":128,"num_experts_per_tok":8,"num_hidden_layers":61,"num_key_value_heads":128,"q_lora_rank":1536,"qk_nope_head_dim":128,"qk_rope_head_dim":64,"rope_scaling":{"beta_fast":32,"beta_slow":1,"factor":40,"mscale":1.0,"mscale_all_dim":1.0,"original_max_position_embeddings":4096,"type":"yarn"},"tie_word_embeddings":false,"v_head_dim":128,"vocab_size":129280},"config_sha256":"b1f72e9c8ca7a502ddc73a365e4a7898475ba399ee536a39eed75bc501bacfb1","display_name":"DeepSeek-V3","expected":{"active_params":36625618432,"attention":"MLA (kv_rank=512, q_rank=1536)","context":"163,840 (yarn ×40 from 4,096)","delta_active_pct":-1.0118420756756672,"delta_total_pct":0.003937287630397004,"ffn":"MoE 256 × top-8 + 1 shared","total_params":671026419200},"provenance":{"config_url":"https://huggingface.co/deepseek-ai/DeepSeek-V3/resolve/e815299b0bcbac849fa540c768ef21845365c9eb/config.json","count_scope":"Main autoregressive model; DeepSeek-V3 MTP weights are excluded; counts are logical parameters, not packed tensor elements.","model_card_url":"https://huggingface.co/deepseek-ai/DeepSeek-V3/blob/e815299b0bcbac849fa540c768ef21845365c9eb/README.md","projection":"Retain only fields used by the teaching schema; nested rope_scaling is retained intact. Missing head_dim is derived from hidden_size / num_attention_heads.","raw_config_file":"configs/deepseek-ai--DeepSeek-V3.json","raw_config_sha256":"cbf0b95dc614de208a109bb5fd4e7eed11385e9c68411d2c17db5319443035d9","revision":"e815299b0bcbac849fa540c768ef21845365c9eb","verified_on":"2026-10-11"},"repo_id":"deepseek-ai/DeepSeek-V3","total_tolerance_pct":1.5}],"schema_version":1,"snapshot_date":"2026-10-11"}'
EXPECTED_PAYLOAD_SHA256 = '77583e08e678b30369f15f353832173e53c054c6b03d0cf99b3b4b5797f872e4'

REFERENCE = strict_json_loads(REFERENCE_JSON)
assert hashlib.sha256(REFERENCE_JSON.encode("utf-8")).hexdigest() == EXPECTED_PAYLOAD_SHA256
assert sha256_json(REFERENCE) == EXPECTED_PAYLOAD_SHA256
print(
    f"schema={REFERENCE['schema_version']} | snapshot={REFERENCE['snapshot_date']} | "
    f"models={len(REFERENCE['models'])}"
)
print("payload sha256:", EXPECTED_PAYLOAD_SHA256)

print('The Module 1 snapshot and SHA-256 are verified')

SOURCE_CONFIG_TEXTS = {'openai/gpt-oss-20b': '{\n  "architectures": [\n    "GptOssForCausalLM"\n  ],\n  "attention_bias": true,\n  "attention_dropout": 0.0,\n  "eos_token_id": 200002,\n  "experts_per_token": 4,\n  "head_dim": 64,\n  "hidden_act": "silu",\n  "hidden_size": 2880,\n  "initial_context_length": 4096,\n  "initializer_range": 0.02,\n  "intermediate_size": 2880,\n  "layer_types": [\n    "sliding_attention",\n    "full_attention",\n    "sliding_attention",\n    "full_attention",\n    "sliding_attention",\n    "full_attention",\n    "sliding_attention",\n    "full_attention",\n    "sliding_attention",\n    "full_attention",\n    "sliding_attention",\n    "full_attention",\n    "sliding_attention",\n    "full_attention",\n    "sliding_attention",\n    "full_attention",\n    "sliding_attention",\n    "full_attention",\n    "sliding_attention",\n    "full_attention",\n    "sliding_attention",\n    "full_attention",\n    "sliding_attention",\n    "full_attention"\n  ],\n  "max_position_embeddings": 131072,\n  "model_type": "gpt_oss",\n  "num_attention_heads": 64,\n  "num_experts_per_tok": 4,\n  "num_hidden_layers": 24,\n  "num_key_value_heads": 8,\n  "num_local_experts": 32,\n  "output_router_logits": false,\n  "pad_token_id": 199999,\n  "quantization_config": {\n    "modules_to_not_convert": [\n      "model.layers.*.self_attn",\n      "model.layers.*.mlp.router",\n      "model.embed_tokens",\n      "lm_head"\n    ],\n    "quant_method": "mxfp4"\n  },\n  "rms_norm_eps": 1e-05,\n  "rope_scaling": {\n    "beta_fast": 32.0,\n    "beta_slow": 1.0,\n    "factor": 32.0,\n    "original_max_position_embeddings": 4096,\n    "rope_type": "yarn",\n    "truncate": false\n  },\n  "rope_theta": 150000,\n  "router_aux_loss_coef": 0.9,\n  "sliding_window": 128,\n  "swiglu_limit": 7.0,\n  "tie_word_embeddings": false,\n  "transformers_version": "4.55.0.dev0",\n  "use_cache": true,\n  "vocab_size": 201088\n}\n', 'openai/gpt-oss-120b': '{\n  "architectures": [\n    "GptOssForCausalLM"\n  ],\n  "attention_bias": true,\n  "attention_dropout": 0.0,\n  "eos_token_id": 200002,\n  "experts_per_token": 4,\n  "head_dim": 64,\n  "hidden_act": "silu",\n  "hidden_size": 2880,\n  "initial_context_length": 4096,\n  "initializer_range": 0.02,\n  "intermediate_size": 2880,\n  "layer_types": [\n    "sliding_attention",\n    "full_attention",\n    "sliding_attention",\n    "full_attention",\n    "sliding_attention",\n    "full_attention",\n    "sliding_attention",\n    "full_attention",\n    "sliding_attention",\n    "full_attention",\n    "sliding_attention",\n    "full_attention",\n    "sliding_attention",\n    "full_attention",\n    "sliding_attention",\n    "full_attention",\n    "sliding_attention",\n    "full_attention",\n    "sliding_attention",\n    "full_attention",\n    "sliding_attention",\n    "full_attention",\n    "sliding_attention",\n    "full_attention",\n    "sliding_attention",\n    "full_attention",\n    "sliding_attention",\n    "full_attention",\n    "sliding_attention",\n    "full_attention",\n    "sliding_attention",\n    "full_attention",\n    "sliding_attention",\n    "full_attention",\n    "sliding_attention",\n    "full_attention"\n  ],\n  "max_position_embeddings": 131072,\n  "model_type": "gpt_oss",\n  "num_attention_heads": 64,\n  "num_experts_per_tok": 4,\n  "num_hidden_layers": 36,\n  "num_key_value_heads": 8,\n  "num_local_experts": 128,\n  "output_router_logits": false,\n  "pad_token_id": 199999,\n  "quantization_config": {\n    "modules_to_not_convert": [\n      "model.layers.*.self_attn",\n      "model.layers.*.mlp.router",\n      "model.embed_tokens",\n      "lm_head"\n    ],\n    "quant_method": "mxfp4"\n  },\n  "rms_norm_eps": 1e-05,\n  "rope_scaling": {\n    "beta_fast": 32.0,\n    "beta_slow": 1.0,\n    "factor": 32.0,\n    "original_max_position_embeddings": 4096,\n    "rope_type": "yarn",\n    "truncate": false\n  },\n  "rope_theta": 150000,\n  "router_aux_loss_coef": 0.9,\n  "sliding_window": 128,\n  "swiglu_limit": 7.0,\n  "tie_word_embeddings": false,\n  "transformers_version": "4.55.0.dev0",\n  "use_cache": true,\n  "vocab_size": 201088\n}\n', 'Qwen/Qwen3-8B': '{\n  "architectures": [\n    "Qwen3ForCausalLM"\n  ],\n  "attention_bias": false,\n  "attention_dropout": 0.0,\n  "bos_token_id": 151643,\n  "eos_token_id": 151645,\n  "head_dim": 128,\n  "hidden_act": "silu",\n  "hidden_size": 4096,\n  "initializer_range": 0.02,\n  "intermediate_size": 12288,\n  "max_position_embeddings": 40960,\n  "max_window_layers": 36,\n  "model_type": "qwen3",\n  "num_attention_heads": 32,\n  "num_hidden_layers": 36,\n  "num_key_value_heads": 8,\n  "rms_norm_eps": 1e-06,\n  "rope_scaling": null,\n  "rope_theta": 1000000,\n  "sliding_window": null,\n  "tie_word_embeddings": false,\n  "torch_dtype": "bfloat16",\n  "transformers_version": "4.51.0",\n  "use_cache": true,\n  "use_sliding_window": false,\n  "vocab_size": 151936\n}', 'Qwen/Qwen3-14B': '{\n  "architectures": [\n    "Qwen3ForCausalLM"\n  ],\n  "attention_bias": false,\n  "attention_dropout": 0.0,\n  "bos_token_id": 151643,\n  "eos_token_id": 151645,\n  "head_dim": 128,\n  "hidden_act": "silu",\n  "hidden_size": 5120,\n  "initializer_range": 0.02,\n  "intermediate_size": 17408,\n  "max_position_embeddings": 40960,\n  "max_window_layers": 40,\n  "model_type": "qwen3",\n  "num_attention_heads": 40,\n  "num_hidden_layers": 40,\n  "num_key_value_heads": 8,\n  "rms_norm_eps": 1e-06,\n  "rope_scaling": null,\n  "rope_theta": 1000000,\n  "sliding_window": null,\n  "tie_word_embeddings": false,\n  "torch_dtype": "bfloat16",\n  "transformers_version": "4.51.0",\n  "use_cache": true,\n  "use_sliding_window": false,\n  "vocab_size": 151936\n}', 'Qwen/Qwen3-32B': '{\n  "architectures": [\n    "Qwen3ForCausalLM"\n  ],\n  "attention_bias": false,\n  "attention_dropout": 0.0,\n  "bos_token_id": 151643,\n  "eos_token_id": 151645,\n  "head_dim": 128,\n  "hidden_act": "silu",\n  "hidden_size": 5120,\n  "initializer_range": 0.02,\n  "intermediate_size": 25600,\n  "max_position_embeddings": 40960,\n  "max_window_layers": 64,\n  "model_type": "qwen3",\n  "num_attention_heads": 64,\n  "num_hidden_layers": 64,\n  "num_key_value_heads": 8,\n  "rms_norm_eps": 1e-06,\n  "rope_scaling": null,\n  "rope_theta": 1000000,\n  "sliding_window": null,\n  "tie_word_embeddings": false,\n  "torch_dtype": "bfloat16",\n  "transformers_version": "4.51.0",\n  "use_cache": true,\n  "use_sliding_window": false,\n  "vocab_size": 151936\n}', 'Qwen/Qwen3-30B-A3B': '{\n  "architectures": [\n    "Qwen3MoeForCausalLM"\n  ],\n  "attention_bias": false,\n  "attention_dropout": 0.0,\n  "bos_token_id": 151643,\n  "decoder_sparse_step": 1,\n  "eos_token_id": 151645,\n  "head_dim": 128,\n  "hidden_act": "silu",\n  "hidden_size": 2048,\n  "initializer_range": 0.02,\n  "intermediate_size": 6144,\n  "max_position_embeddings": 40960,\n  "max_window_layers": 48,\n  "mlp_only_layers": [],\n  "model_type": "qwen3_moe",\n  "moe_intermediate_size": 768,\n  "norm_topk_prob": true,\n  "num_attention_heads": 32,\n  "num_experts": 128,\n  "num_experts_per_tok": 8,\n  "num_hidden_layers": 48,\n  "num_key_value_heads": 4,\n  "output_router_logits": false,\n  "rms_norm_eps": 1e-06,\n  "rope_scaling": null,\n  "rope_theta": 1000000.0,\n  "router_aux_loss_coef": 0.001,\n  "sliding_window": null,\n  "tie_word_embeddings": false,\n  "torch_dtype": "bfloat16",\n  "transformers_version": "4.51.0",\n  "use_cache": true,\n  "use_sliding_window": false,\n  "vocab_size": 151936\n}', 'mistralai/Mixtral-8x7B-v0.1': '{\n  "architectures": [\n    "MixtralForCausalLM"\n  ],\n  "attention_dropout": 0.0,\n  "bos_token_id": 1,\n  "eos_token_id": 2,\n  "hidden_act": "silu",\n  "hidden_size": 4096,\n  "initializer_range": 0.02,\n  "intermediate_size": 14336,\n  "max_position_embeddings": 32768,\n  "model_type": "mixtral",\n  "num_attention_heads": 32,\n  "num_experts_per_tok": 2,\n  "num_hidden_layers": 32,\n  "num_key_value_heads": 8,\n  "num_local_experts": 8,\n  "output_router_logits": false,\n  "rms_norm_eps": 1e-05,\n  "rope_theta": 1000000.0,\n  "router_aux_loss_coef": 0.02,\n  "sliding_window": null,\n  "tie_word_embeddings": false,\n  "torch_dtype": "bfloat16",\n  "transformers_version": "4.36.0.dev0",\n  "use_cache": true,\n  "vocab_size": 32000\n}\n', 'deepseek-ai/DeepSeek-V3': '{\n  "architectures": [\n    "DeepseekV3ForCausalLM"\n  ],\n  "attention_bias": false,\n  "attention_dropout": 0.0,\n  "auto_map": {\n    "AutoConfig": "configuration_deepseek.DeepseekV3Config",\n    "AutoModel": "modeling_deepseek.DeepseekV3Model",\n    "AutoModelForCausalLM": "modeling_deepseek.DeepseekV3ForCausalLM"\n  },\n  "bos_token_id": 0,\n  "eos_token_id": 1,\n  "ep_size": 1,\n  "first_k_dense_replace": 3,\n  "hidden_act": "silu",\n  "hidden_size": 7168,\n  "initializer_range": 0.02,\n  "intermediate_size": 18432,\n  "kv_lora_rank": 512,\n  "max_position_embeddings": 163840,\n  "model_type": "deepseek_v3",\n  "moe_intermediate_size": 2048,\n  "moe_layer_freq": 1,\n  "n_group": 8,\n  "n_routed_experts": 256,\n  "n_shared_experts": 1,\n  "norm_topk_prob": true,\n  "num_attention_heads": 128,\n  "num_experts_per_tok": 8,\n  "num_hidden_layers": 61,\n  "num_key_value_heads": 128,\n  "num_nextn_predict_layers": 1,\n  "q_lora_rank": 1536,\n  "qk_nope_head_dim": 128,\n  "qk_rope_head_dim": 64,\n  "quantization_config": {\n    "activation_scheme": "dynamic",\n    "fmt": "e4m3",\n    "quant_method": "fp8",\n    "weight_block_size": [\n      128,\n      128\n    ]\n  },\n  "rms_norm_eps": 1e-06,\n  "rope_scaling": {\n    "beta_fast": 32,\n    "beta_slow": 1,\n    "factor": 40,\n    "mscale": 1.0,\n    "mscale_all_dim": 1.0,\n    "original_max_position_embeddings": 4096,\n    "type": "yarn"\n  },\n  "rope_theta": 10000,\n  "routed_scaling_factor": 2.5,\n  "scoring_func": "sigmoid",\n  "tie_word_embeddings": false,\n  "topk_group": 4,\n  "topk_method": "noaux_tc",\n  "torch_dtype": "bfloat16",\n  "transformers_version": "4.33.1",\n  "use_cache": true,\n  "v_head_dim": 128,\n  "vocab_size": 129280\n}\n'}

source_configs = {}
for item in REFERENCE["models"]:
    source_text = SOURCE_CONFIG_TEXTS[item["repo_id"]]
    assert hashlib.sha256(source_text.encode("utf-8")).hexdigest() == item["provenance"]["raw_config_sha256"]
    full_config = strict_json_loads(source_text)
    assert all(full_config[key] == value for key, value in item["config"].items())
    source_configs[item["repo_id"]] = full_config

display(pd.DataFrame([
    {"model": m["display_name"], "revision": m["provenance"]["revision"],
     "verified_on": m["provenance"]["verified_on"],
     "source_sha256": m["provenance"]["raw_config_sha256"]}
    for m in REFERENCE["models"]
]))

EXPERT_KEYS = ("num_local_experts", "n_routed_experts", "num_experts")
TOP_K_KEYS = ("num_experts_per_tok", "num_experts_per_token")
SHARED_EXPERT_KEYS = ("n_shared_experts", "num_shared_experts")
ROPE_TYPE_KEYS = ("rope_type", "type")
MLA_FIELDS = (
    "q_lora_rank",
    "kv_lora_rank",
    "qk_nope_head_dim",
    "qk_rope_head_dim",
    "v_head_dim",
)


def validate_repo_id(repo_id: object) -> str:
    """Validate a two-component Hugging Face repository identifier."""

    value = _string("repo_id", repo_id)
    if not re.fullmatch(r"[A-Za-z0-9][A-Za-z0-9_.-]*/[A-Za-z0-9][A-Za-z0-9_.-]*", value):
        raise ConfigSchemaError("repo_id must have the form 'namespace/name'")
    return value


def validate_commit_revision(revision: object) -> str:
    """Require an immutable full 40-hex Git commit revision."""

    value = _string("revision", revision)
    if not re.fullmatch(r"[0-9a-fA-F]{40}", value):
        raise ConfigSchemaError("revision must be an exact 40-hex commit, not 'main' or a tag")
    return value.lower()


def validate_source_revision(revision: object) -> str:
    """Validate a branch, tag, PR ref, or full commit used for an exploratory Hub lookup.

    This intentionally accepts only the common Hub forms used in the lesson while rejecting
    whitespace, control characters, repeated path separators, and ambiguous Git-ref constructs.
    """

    value = _string("revision", revision)
    if len(value) > 200 or not re.fullmatch(r"[A-Za-z0-9][A-Za-z0-9._/-]*", value):
        raise ConfigSchemaError("source revision contains unsupported characters")
    if ".." in value or "//" in value or value.endswith(".lock"):
        raise ConfigSchemaError("source revision contains an unsafe or ambiguous construct")
    return value


def expert_count(cfg: Mapping[str, Any]) -> int | None:
    """Return the routed-expert count after validating synonym consistency."""

    value = _synonym_value(cfg, EXPERT_KEYS)
    return None if value is None else _positive_int("expert count", value)


def experts_per_token(cfg: Mapping[str, Any]) -> int | None:
    """Return top-k routed experts after validating synonym consistency."""

    value = _synonym_value(cfg, TOP_K_KEYS)
    return None if value is None else _positive_int("experts per token", value)


def shared_expert_count(cfg: Mapping[str, Any]) -> int:
    """Return the non-negative number of always-on shared experts."""

    value = _synonym_value(cfg, SHARED_EXPERT_KEYS, default=0)
    return _non_negative_int("shared expert count", value)


def resolved_head_dim(cfg: Mapping[str, Any]) -> int:
    """Return explicit head_dim or infer it only when hidden/heads divide exactly."""

    if cfg.get("head_dim") is not None:
        return _positive_int("head_dim", cfg["head_dim"])
    hidden = _positive_int("hidden_size", cfg.get("hidden_size"))
    heads = _positive_int("num_attention_heads", cfg.get("num_attention_heads"))
    if hidden % heads:
        raise ConfigSchemaError("head_dim is missing and hidden_size is not divisible by heads")
    return hidden // heads


def validate_config(config: object) -> dict[str, Any]:
    """Validate fields used by the Module 1 passport and return a defensive copy.

    Unknown fields are retained because real configs evolve. The validator is strict only for
    fields consumed by the teaching formulas; unsupported architectures should fail loudly
    instead of being silently forced into the nearest familiar family.
    """

    cfg = _mapping("config", config)
    hidden = _positive_int("hidden_size", cfg.get("hidden_size"))
    layers = _positive_int("num_hidden_layers", cfg.get("num_hidden_layers"))
    heads = _positive_int("num_attention_heads", cfg.get("num_attention_heads"))
    _positive_int("vocab_size", cfg.get("vocab_size"))
    _positive_int("max_position_embeddings", cfg.get("max_position_embeddings"))
    resolved_head_dim(cfg)

    kv_heads = cfg.get("num_key_value_heads", heads)
    kv_heads = _positive_int("num_key_value_heads", kv_heads)
    if kv_heads > heads or heads % kv_heads:
        raise ConfigSchemaError("num_key_value_heads must divide num_attention_heads")

    if "tie_word_embeddings" in cfg and not isinstance(cfg["tie_word_embeddings"], bool):
        raise ConfigSchemaError("tie_word_embeddings must be boolean")

    architectures = cfg.get("architectures")
    if architectures is not None:
        if not isinstance(architectures, list) or not architectures:
            raise ConfigSchemaError("architectures must be a non-empty list")
        for index, architecture in enumerate(architectures):
            _string(f"architectures[{index}]", architecture)

    rope_scaling = cfg.get("rope_scaling")
    if rope_scaling is not None:
        rs = _mapping("rope_scaling", rope_scaling)
        rope_type = _synonym_value(rs, ROPE_TYPE_KEYS)
        if rope_type is not None:
            _string("rope_scaling type", rope_type)
        if rs.get("factor") is not None:
            _finite_positive_real("rope_scaling.factor", rs["factor"])
        if rs.get("original_max_position_embeddings") is not None:
            _positive_int(
                "rope_scaling.original_max_position_embeddings",
                rs["original_max_position_embeddings"],
            )

    experts = expert_count(cfg)
    top_k = experts_per_token(cfg)
    shared = shared_expert_count(cfg)
    if experts is None:
        if top_k is not None or shared:
            raise ConfigSchemaError("top-k/shared experts require a routed-expert count")
        _positive_int("intermediate_size", cfg.get("intermediate_size"))
    else:
        if top_k is None:
            raise ConfigSchemaError("MoE config requires num_experts_per_tok")
        if top_k > experts:
            raise ConfigSchemaError("experts per token cannot exceed routed experts")
        inter_value = cfg.get("moe_intermediate_size", cfg.get("intermediate_size"))
        _positive_int("MoE intermediate size", inter_value)
        dense_prefix = _non_negative_int(
            "first_k_dense_replace",
            cfg.get("first_k_dense_replace", 0),
        )
        if dense_prefix > layers:
            raise ConfigSchemaError("first_k_dense_replace cannot exceed layer count")
        if dense_prefix and cfg.get("intermediate_size") is None:
            raise ConfigSchemaError("dense prefix requires intermediate_size")

    has_any_mla = any(cfg.get(field) is not None for field in MLA_FIELDS)
    if has_any_mla:
        missing = [field for field in MLA_FIELDS if cfg.get(field) is None]
        if missing:
            raise ConfigSchemaError(f"incomplete MLA signature; missing {', '.join(missing)}")
        for field in MLA_FIELDS:
            _positive_int(field, cfg[field])

    if hidden <= 0:  # retained as an explicit local invariant for readability
        raise ConfigSchemaError("hidden_size must be positive")
    return copy.deepcopy(cfg)


@dataclass(frozen=True)
class SnapshotModel:
    """One validated model fixture and the convention used for its active count."""

    repo_id: str
    display_name: str
    config: dict[str, Any]
    config_sha256: str
    claimed_total_b: float
    claimed_active_b: float
    active_convention: str
    total_tolerance_pct: float
    active_tolerance_pct: float


@dataclass(frozen=True)
class ReferenceBundle:
    """Validated Module 1 snapshot, assumptions, and context milestones."""

    snapshot_date: str
    models: tuple[SnapshotModel, ...]
    context_milestones: tuple[dict[str, Any], ...]
    cost_assumptions: dict[str, float]


def load_reference_bundle(payload: object) -> ReferenceBundle:
    """Validate the frozen module payload and return typed immutable records."""

    root = _mapping("reference payload", payload)
    if root.get("schema_version") != 1:
        raise ConfigSchemaError("unsupported reference schema_version")
    snapshot_date = _string("snapshot_date", root.get("snapshot_date"))
    raw_models = root.get("models")
    if not isinstance(raw_models, list) or not raw_models:
        raise ConfigSchemaError("models must be a non-empty list")

    models: list[SnapshotModel] = []
    seen_repo_ids: set[str] = set()
    seen_display_names: set[str] = set()
    for index, raw in enumerate(raw_models):
        item = _mapping(f"models[{index}]", raw)
        repo_id = validate_repo_id(item.get("repo_id"))
        display_name = _string("display_name", item.get("display_name"))
        if repo_id in seen_repo_ids:
            raise ConfigSchemaError(f"duplicate repo_id in snapshot: {repo_id}")
        if display_name in seen_display_names:
            raise ConfigSchemaError(f"duplicate display_name in snapshot: {display_name}")
        seen_repo_ids.add(repo_id)
        seen_display_names.add(display_name)
        config = validate_config(item.get("config"))
        expected_hash = _string("config_sha256", item.get("config_sha256"))
        if not re.fullmatch(r"[0-9a-f]{64}", expected_hash):
            raise ConfigSchemaError("config_sha256 must be lowercase 64-hex")
        if sha256_json(config) != expected_hash:
            raise ConfigSchemaError(f"config checksum mismatch for {repo_id}")
        convention = _string("active_convention", item.get("active_convention"))
        if convention not in {
            "total",
            "core_plus_unembedding",
            "core_plus_embeddings",
        }:
            raise ConfigSchemaError(f"unsupported active convention: {convention}")
        models.append(
            SnapshotModel(
                repo_id=repo_id,
                display_name=display_name,
                config=config,
                config_sha256=expected_hash,
                claimed_total_b=_finite_positive_real(
                    "claimed_total_b", item.get("claimed_total_b")
                ),
                claimed_active_b=_finite_positive_real(
                    "claimed_active_b", item.get("claimed_active_b")
                ),
                active_convention=convention,
                total_tolerance_pct=_finite_positive_real(
                    "total_tolerance_pct", item.get("total_tolerance_pct")
                ),
                active_tolerance_pct=_finite_positive_real(
                    "active_tolerance_pct", item.get("active_tolerance_pct")
                ),
            )
        )

    raw_milestones = root.get("context_milestones")
    if not isinstance(raw_milestones, list) or not raw_milestones:
        raise ConfigSchemaError("context_milestones must be a non-empty list")
    milestones: list[dict[str, Any]] = []
    for index, raw in enumerate(raw_milestones):
        item = _mapping(f"context_milestones[{index}]", raw)
        year = _finite_positive_real("milestone year", item.get("year"))
        tokens = _positive_int("milestone tokens", item.get("tokens"))
        model = _string("milestone model", item.get("model"))
        milestones.append({"year": year, "tokens": tokens, "model": model})
    if milestones != sorted(milestones, key=lambda row: row["year"]):
        raise ConfigSchemaError("context_milestones must be sorted by year")

    assumptions_raw = _mapping("cost_assumptions", root.get("cost_assumptions"))
    assumptions = {
        "bf16_bits": _finite_positive_real("bf16_bits", assumptions_raw.get("bf16_bits")),
        "four_bit_effective_bits": _finite_positive_real(
            "four_bit_effective_bits", assumptions_raw.get("four_bit_effective_bits")
        ),
        "card_capacity_gb_decimal": _finite_positive_real(
            "card_capacity_gb_decimal", assumptions_raw.get("card_capacity_gb_decimal")
        ),
    }
    return ReferenceBundle(
        snapshot_date=snapshot_date,
        models=tuple(models),
        context_milestones=tuple(milestones),
        cost_assumptions=assumptions,
    )

bundle = load_reference_bundle(REFERENCE)
by_repo = {model.repo_id: model for model in bundle.models}
rows = [
    {
        "repo_id": model.repo_id,
        "display_name": model.display_name,
        "snapshot": bundle.snapshot_date,
        "config_sha256": model.config_sha256[:12],
        "active_convention": model.active_convention,
    }
    for model in bundle.models
]
display(pd.DataFrame(rows))
assert len(by_repo) == len(bundle.models) == 8
assert all(sha256_json(model.config) == model.config_sha256 for model in bundle.models)

print('Schema validation passed for every fixture')

def attention_kind(config: object) -> str:
    """Classify standard MHA/GQA/MQA or MLA after schema validation."""

    cfg = validate_config(config)
    if "kv_lora_rank" in cfg:
        return f"MLA (kv_rank={cfg['kv_lora_rank']}, q_rank={cfg['q_lora_rank']})"
    heads = int(cfg["num_attention_heads"])
    kv_heads = int(cfg.get("num_key_value_heads", heads))
    if kv_heads == heads:
        label = "MHA"
    elif kv_heads == 1:
        label = "MQA"
    else:
        label = f"GQA {heads}/{kv_heads}"
    window = cfg.get("sliding_window")
    if window is not None:
        label += f" + sliding {_positive_int('sliding_window', window)}"
    return label


def moe_kind(config: object) -> str:
    """Return a readable dense/MoE classifier from validated synonym fields."""

    cfg = validate_config(config)
    experts = expert_count(cfg)
    if experts is None:
        return "dense"
    label = f"MoE {experts} × top-{experts_per_token(cfg)}"
    shared = shared_expert_count(cfg)
    if shared:
        label += f" + {shared} shared"
    return label


def context_info(config: object) -> str:
    """Summarize maximum positions and optional RoPE scaling."""

    cfg = validate_config(config)
    maximum = int(cfg["max_position_embeddings"])
    scaling = cfg.get("rope_scaling")
    if not scaling:
        return f"{maximum:,}"
    rope_type = _synonym_value(scaling, ROPE_TYPE_KEYS, default="unspecified")
    factor = scaling.get("factor", "?")
    original = scaling.get("original_max_position_embeddings")
    suffix = f"{rope_type} ×{factor}"
    if original is not None:
        suffix += f" from {int(original):,}"
    return f"{maximum:,} ({suffix})"

passport_rows = [
    {
        "model": model.display_name,
        "attention": attention_kind(model.config),
        "ffn": moe_kind(model.config),
        "context": context_info(model.config),
        "hidden": model.config["hidden_size"],
        "layers": model.config["num_hidden_layers"],
        "vocab": model.config["vocab_size"],
    }
    for model in bundle.models
]
passport_df = pd.DataFrame(passport_rows).set_index("model")
display(passport_df)
assert attention_kind(by_repo["deepseek-ai/DeepSeek-V3"].config).startswith("MLA")
assert attention_kind(by_repo["openai/gpt-oss-120b"].config) == "GQA 64/8 + sliding 128"
assert moe_kind(by_repo["mistralai/Mixtral-8x7B-v0.1"].config) == "MoE 8 × top-2"
assert moe_kind(by_repo["Qwen/Qwen3-32B"].config) == "dense"

print('Architecture classifiers are verified')

@dataclass(frozen=True)
class ParameterBreakdown:
    """Approximate parameter accounting for one supported Transformer family."""

    total: int
    active_core: int
    active_plus_unembedding: int
    active_plus_embeddings: int
    attention: int
    ffn_total: int
    embedding: int
    unembedding: int
    norms: int
    extras: int

    def active(self, convention: str) -> int:
        """Return the active count under one explicitly declared convention."""

        if convention == "total":
            return self.total
        if convention == "core_plus_unembedding":
            return self.active_plus_unembedding
        if convention == "core_plus_embeddings":
            return self.active_plus_embeddings
        raise ConfigSchemaError(f"unsupported active convention: {convention}")


def _architecture_name(cfg: Mapping[str, Any]) -> str:
    """Return the first declared architecture name or an empty string."""

    architectures = cfg.get("architectures")
    if isinstance(architectures, list) and architectures and isinstance(architectures[0], str):
        return architectures[0]
    return ""


def attention_parameter_count(config: object) -> int:
    """Count attention parameters per layer for the supported teaching families.

    For gpt-oss the public implementation uses fused QKV and output biases. For
    DeepSeek-V3 the public configuration/implementation also exposes Q- and latent-KV
    normalization scales. Those small terms matter because this module promises an
    explicit reconstruction rather than a matrix-only headline.
    """

    cfg = validate_config(config)
    architecture = _architecture_name(cfg)
    heads = int(cfg["num_attention_heads"])
    hidden = int(cfg["hidden_size"])
    if "kv_lora_rank" in cfg:
        q_rank = int(cfg["q_lora_rank"])
        kv_rank = int(cfg["kv_lora_rank"])
        d_nope = int(cfg["qk_nope_head_dim"])
        d_rope = int(cfg["qk_rope_head_dim"])
        d_value = int(cfg["v_head_dim"])
        query = hidden * q_rank + q_rank * heads * (d_nope + d_rope)
        key_value = hidden * (kv_rank + d_rope) + kv_rank * heads * (d_nope + d_value)
        output = heads * d_value * hidden
        q_kv_norm_scales = q_rank + kv_rank if architecture == "DeepseekV3ForCausalLM" else 0
        return query + key_value + output + q_kv_norm_scales
    head_dim = resolved_head_dim(cfg)
    kv_heads = int(cfg.get("num_key_value_heads", heads))
    query = hidden * heads * head_dim
    key_value = 2 * hidden * kv_heads * head_dim
    output = heads * head_dim * hidden
    biases = 0
    if architecture == "GptOssForCausalLM":
        q_dim = heads * head_dim
        kv_dim = kv_heads * head_dim
        biases = q_dim + 2 * kv_dim + hidden
    # Qwen3 applies a learned RMSNorm to Q and K, each with head_dim weights.
    qk_norms = 2 * head_dim if architecture in {"Qwen3ForCausalLM", "Qwen3MoeForCausalLM"} else 0
    return query + key_value + output + biases + qk_norms


def ffn_parameter_count(config: object, *, dense_layer: bool) -> tuple[int, int]:
    """Return total and per-token-active FFN/router parameters for one layer."""

    cfg = validate_config(config)
    hidden = int(cfg["hidden_size"])
    experts = expert_count(cfg)
    if experts is None or dense_layer:
        dense = 3 * hidden * int(cfg["intermediate_size"])
        return dense, dense
    intermediate_value = cfg.get("moe_intermediate_size")
    if intermediate_value is None:
        intermediate_value = cfg.get("intermediate_size")
    intermediate = int(intermediate_value)
    architecture = _architecture_name(cfg)
    expert = 3 * hidden * intermediate
    if architecture == "GptOssForCausalLM":
        expert += 2 * intermediate + hidden
    router = hidden * experts
    if architecture in {"GptOssForCausalLM", "DeepseekV3ForCausalLM"}:
        router += experts
    shared = shared_expert_count(cfg) * expert
    total = experts * expert + shared + router
    active = int(experts_per_token(cfg)) * expert + shared + router
    return total, active


def count_parameters(config: object) -> ParameterBreakdown:
    """Count approximate total and convention-dependent active parameters."""

    cfg = validate_config(config)
    supported = {"GptOssForCausalLM", "LlamaForCausalLM", "Qwen3ForCausalLM",
                 "Qwen3MoeForCausalLM", "MixtralForCausalLM", "DeepseekV3ForCausalLM"}
    if _architecture_name(cfg) not in supported:
        raise ConfigSchemaError("parameter accounting requires a supported architecture contract")
    hidden = int(cfg["hidden_size"])
    layers = int(cfg["num_hidden_layers"])
    vocab = int(cfg["vocab_size"])
    embedding = vocab * hidden
    unembedding = 0 if cfg.get("tie_word_embeddings", False) else vocab * hidden
    dense_prefix = int(cfg.get("first_k_dense_replace", 0)) if expert_count(cfg) else layers
    attention = attention_parameter_count(cfg) * layers
    ffn_total = 0
    ffn_active = 0
    for layer in range(layers):
        total_layer, active_layer = ffn_parameter_count(
            cfg,
            dense_layer=layer < dense_prefix,
        )
        ffn_total += total_layer
        ffn_active += active_layer
    norms = (2 * layers + 1) * hidden
    architecture = _architecture_name(cfg)
    extras = layers * int(cfg["num_attention_heads"]) if architecture == "GptOssForCausalLM" else 0
    active_core = attention + ffn_active + norms + extras
    total = embedding + unembedding + attention + ffn_total + norms + extras
    return ParameterBreakdown(
        total=total,
        active_core=active_core,
        active_plus_unembedding=active_core + unembedding,
        active_plus_embeddings=active_core + unembedding + embedding,
        attention=attention,
        ffn_total=ffn_total,
        embedding=embedding,
        unembedding=unembedding,
        norms=norms,
        extras=extras,
    )

breakdowns = {model.repo_id: count_parameters(model.config) for model in bundle.models}
parameter_rows = []
for model in bundle.models:
    breakdown = breakdowns[model.repo_id]
    parameter_rows.append(
        {
            "model": model.display_name,
            "total_B": breakdown.total / 1e9,
            "active_core_B": breakdown.active_core / 1e9,
            "declared_active_B": breakdown.active(model.active_convention) / 1e9,
            "active_convention": model.active_convention,
            "attention_B": breakdown.attention / 1e9,
            "ffn_total_B": breakdown.ffn_total / 1e9,
            "implementation_extras_B": breakdown.extras / 1e9,
        }
    )
parameter_df = pd.DataFrame(parameter_rows).set_index("model")
display(parameter_df)
assert breakdowns["Qwen/Qwen3-8B"].total == 8_190_735_360
assert breakdowns["openai/gpt-oss-120b"].total == 116_829_156_672
assert breakdowns["openai/gpt-oss-120b"].active("core_plus_unembedding") == 5_132_849_472

gpt_model = by_repo["openai/gpt-oss-120b"]
gpt_breakdown = breakdowns[gpt_model.repo_id]
explicit = gpt_breakdown.active(gpt_model.active_convention)
alternative = gpt_breakdown.active("core_plus_embeddings")
assert explicit != alternative
print("explicit convention:", gpt_model.active_convention, f"{explicit / 1e9:.6f}B")
print("different convention:", "core_plus_embeddings", f"{alternative / 1e9:.6f}B")

print('The parameter breakdown is built')

@dataclass(frozen=True)
class ReconciliationRow:
    """One model's computed passport and declared-card reconciliation."""

    repo_id: str
    model: str
    attention: str
    ffn: str
    context: str
    total_params: int
    active_params: int
    claimed_total_b: float
    claimed_active_b: float
    delta_total_pct: float
    delta_active_pct: float
    active_convention: str


def reconcile_model(model: SnapshotModel) -> ReconciliationRow:
    """Compute one snapshot passport without choosing a convention post hoc."""

    breakdown = count_parameters(model.config)
    active = breakdown.active(model.active_convention)
    total_b = breakdown.total / 1e9
    active_b = active / 1e9
    delta_total = 100.0 * (total_b - model.claimed_total_b) / model.claimed_total_b
    delta_active = 100.0 * (active_b - model.claimed_active_b) / model.claimed_active_b
    if abs(delta_total) > model.total_tolerance_pct:
        raise AssertionError(
            f"{model.repo_id}: total differs by {delta_total:.3f}% "
            f"(limit {model.total_tolerance_pct:.3f}%)"
        )
    if abs(delta_active) > model.active_tolerance_pct:
        raise AssertionError(
            f"{model.repo_id}: active differs by {delta_active:.3f}% "
            f"(limit {model.active_tolerance_pct:.3f}%)"
        )
    return ReconciliationRow(
        repo_id=model.repo_id,
        model=model.display_name,
        attention=attention_kind(model.config),
        ffn=moe_kind(model.config),
        context=context_info(model.config),
        total_params=breakdown.total,
        active_params=active,
        claimed_total_b=model.claimed_total_b,
        claimed_active_b=model.claimed_active_b,
        delta_total_pct=delta_total,
        delta_active_pct=delta_active,
        active_convention=model.active_convention,
    )


@dataclass(frozen=True)
class WeightCost:
    """Weight-payload floor and dense-matmul FLOP approximation for one model."""

    payload_bytes: float
    payload_gb_decimal: float
    payload_gib_binary: float
    gflops_per_token: float
    payload_only_cards: int


def weight_cost(
    *,
    total_params: int,
    active_params: int,
    bits_per_parameter: Real,
    card_capacity_gb_decimal: Real,
) -> WeightCost:
    """Return payload-only memory and a ``2 × active`` matmul FLOP approximation."""

    total = _positive_int("total_params", total_params)
    active = _positive_int("active_params", active_params)
    bits = _finite_positive_real("bits_per_parameter", bits_per_parameter)
    card_gb = _finite_positive_real("card_capacity_gb_decimal", card_capacity_gb_decimal)
    payload_bytes = total * bits / 8.0
    return WeightCost(
        payload_bytes=payload_bytes,
        payload_gb_decimal=payload_bytes / 1e9,
        payload_gib_binary=payload_bytes / (2**30),
        gflops_per_token=(2.0 * active) / 1e9,
        payload_only_cards=math.ceil(payload_bytes / (card_gb * 1e9)),
    )


def flatten_config(value: object, prefix: str = "") -> dict[str, object]:
    """Flatten nested config objects into dot-separated paths for drift reports."""

    if isinstance(value, Mapping):
        result: dict[str, object] = {}
        for key in sorted(value):
            path = f"{prefix}.{key}" if prefix else str(key)
            result.update(flatten_config(value[key], path))
        return result
    if isinstance(value, list):
        result = {}
        for index, item in enumerate(value):
            path = f"{prefix}[{index}]"
            result.update(flatten_config(item, path))
        return result
    return {prefix: value}


def config_diff(snapshot: object, live: object) -> list[dict[str, object]]:
    """Return deterministic path-level changes between two validated configs."""

    left = flatten_config(validate_config(snapshot))
    right = flatten_config(validate_config(live))
    changes: list[dict[str, object]] = []
    for path in sorted(set(left) | set(right)):
        if left.get(path) != right.get(path):
            changes.append({"path": path, "snapshot": left.get(path), "live": right.get(path)})
    return changes

reconciled = [reconcile_model(model) for model in bundle.models]
claim_df = pd.DataFrame([asdict(row) for row in reconciled]).set_index("model")
claim_df["total_B"] = claim_df["total_params"] / 1e9
claim_df["active_B"] = claim_df["active_params"] / 1e9
display(
    claim_df[
        [
            "attention",
            "ffn",
            "total_B",
            "claimed_total_b",
            "delta_total_pct",
            "active_B",
            "claimed_active_b",
            "delta_active_pct",
            "active_convention",
        ]
    ]
)

cost_rows = []
for model, row in zip(bundle.models, reconciled):
    bf16 = weight_cost(
        total_params=row.total_params,
        active_params=row.active_params,
        bits_per_parameter=bundle.cost_assumptions["bf16_bits"],
        card_capacity_gb_decimal=bundle.cost_assumptions["card_capacity_gb_decimal"],
    )
    q4 = weight_cost(
        total_params=row.total_params,
        active_params=row.active_params,
        bits_per_parameter=bundle.cost_assumptions["four_bit_effective_bits"],
        card_capacity_gb_decimal=bundle.cost_assumptions["card_capacity_gb_decimal"],
    )
    cost_rows.append(
        {
            "model": model.display_name,
            "bf16_GB": bf16.payload_gb_decimal,
            "bf16_GiB": bf16.payload_gib_binary,
            "4.25bit_GB": q4.payload_gb_decimal,
            "4.25bit_GiB": q4.payload_gib_binary,
            "GFLOP_token_floor": q4.gflops_per_token,
            "4.25bit_payload_only_cards_at_80GB_decimal": q4.payload_only_cards,
        }
    )
cost_df = pd.DataFrame(cost_rows).set_index("model")
display(cost_df)
example = cost_df.loc["gpt-oss-120b"]
assert 61.0 < example["4.25bit_GB"] < 63.0
assert 57.0 < example["4.25bit_GiB"] < 59.0
assert example["4.25bit_payload_only_cards_at_80GB_decimal"] == 1
print("payload-only fit is not a deployment-memory guarantee")

print('Claims are reconciled under explicit active conventions')
print('GB/GiB and the FLOP floor are computed')

ConfigTransport = Callable[[str, str, str | None], str]
RevisionResolver = Callable[[str, str, str | None], str]


@dataclass(frozen=True)
class FetchedConfig:
    """A schema-validated config tied to an immutable repository revision."""

    repo_id: str
    revision: str
    config: dict[str, Any]
    config_sha256: str


@dataclass(frozen=True)
class SourceFetchedConfig:
    """A live-source config plus the mutable name and immutable commit that produced it."""

    repo_id: str
    requested_revision: str
    resolved_commit: str
    config: dict[str, Any]
    config_sha256: str


def fetch_config_with_transport(
    repo_id: object,
    revision: object,
    transport: ConfigTransport,
    *,
    token: str | None = None,
    expected_config_sha256: str | None = None,
) -> FetchedConfig:
    """Fetch and validate one config through an injected exact-commit text transport.

    There is deliberately no implicit fallback. Snapshot mode and live mode are separate
    operations so a network, authentication, or revision error cannot silently change the data
    source used by a published calculation.
    """

    repository = validate_repo_id(repo_id)
    commit = validate_commit_revision(revision)
    if not callable(transport):
        raise TypeError("transport must be callable")
    text = transport(repository, commit, token)
    config = validate_config(strict_json_loads(text))
    digest = sha256_json(config)
    if expected_config_sha256 is not None:
        expected = _string("expected_config_sha256", expected_config_sha256).lower()
        if not re.fullmatch(r"[0-9a-f]{64}", expected):
            raise ConfigSchemaError("expected_config_sha256 must be lowercase 64-hex")
        if digest != expected:
            raise ConfigIntegrityError(
                f"config checksum mismatch: expected {expected}, got {digest}"
            )
    return FetchedConfig(
        repo_id=repository,
        revision=commit,
        config=config,
        config_sha256=digest,
    )


def fetch_source_config_with_transports(
    repo_id: object,
    requested_revision: object,
    resolver: RevisionResolver,
    transport: ConfigTransport,
    *,
    token: str | None = None,
    expected_config_sha256: str | None = None,
) -> SourceFetchedConfig:
    """Resolve a live branch/tag to a commit, then fetch and validate that exact commit.

    ``requested_revision='main'`` is useful for a live illustration but is not reproducible by
    itself. The returned ``resolved_commit`` is the provenance record that makes the observed
    config reproducible later. Resolver, download, access, schema, and checksum failures remain
    visible; this function never falls back to the frozen course snapshot.
    """

    repository = validate_repo_id(repo_id)
    requested = validate_source_revision(requested_revision)
    if not callable(resolver):
        raise TypeError("resolver must be callable")
    if not callable(transport):
        raise TypeError("transport must be callable")
    resolved = validate_commit_revision(resolver(repository, requested, token))
    fetched = fetch_config_with_transport(
        repository,
        resolved,
        transport,
        token=token,
        expected_config_sha256=expected_config_sha256,
    )
    return SourceFetchedConfig(
        repo_id=repository,
        requested_revision=requested,
        resolved_commit=resolved,
        config=fetched.config,
        config_sha256=fetched.config_sha256,
    )


def huggingface_hub_transport(repo_id: str, revision: str, token: str | None) -> str:
    """Download only ``config.json`` at an exact commit and return its UTF-8 text."""

    try:
        from huggingface_hub import hf_hub_download
    except ImportError as exc:
        raise RuntimeError(
            "live mode requires 'huggingface_hub'; install it explicitly before use"
        ) from exc
    path = hf_hub_download(
        repo_id=repo_id,
        filename="config.json",
        revision=revision,
        token=token,
    )
    return Path(path).read_text(encoding="utf-8")


def huggingface_hub_revision_resolver(
    repo_id: str,
    revision: str,
    token: str | None,
) -> str:
    """Resolve a Hub branch, tag, PR ref, or commit to the repository's full commit SHA."""

    try:
        from huggingface_hub import HfApi
    except ImportError as exc:
        raise RuntimeError(
            "live mode requires 'huggingface_hub'; install it explicitly before use"
        ) from exc
    info = HfApi(token=token).model_info(repo_id=repo_id, revision=revision)
    if not info.sha:
        raise RuntimeError("the Hub response did not contain a resolved commit SHA")
    return str(info.sha)


def fetch_hf_config(
    repo_id: object,
    revision: object,
    *,
    token: str | None = None,
    expected_config_sha256: str | None = None,
) -> FetchedConfig:
    """Fetch one Hugging Face config with exact revision pinning and no silent fallback."""

    return fetch_config_with_transport(
        repo_id,
        revision,
        huggingface_hub_transport,
        token=token,
        expected_config_sha256=expected_config_sha256,
    )


def fetch_hf_config_from_source(
    repo_id: object,
    revision: object = "main",
    *,
    token: str | None = None,
    expected_config_sha256: str | None = None,
) -> SourceFetchedConfig:
    """Fetch a current/source config, recording the immutable commit actually observed.

    Use ``revision='main'`` or a tag only for an opt-in live illustration. Save the returned
    ``resolved_commit`` and pass it to :func:`fetch_hf_config` when reproducibility matters.
    """

    return fetch_source_config_with_transports(
        repo_id,
        revision,
        huggingface_hub_revision_resolver,
        huggingface_hub_transport,
        token=token,
        expected_config_sha256=expected_config_sha256,
    )

fake_revision = "a" * 40
calls = []
snapshot_model = by_repo["openai/gpt-oss-20b"]

def fake_transport(repo_id, revision, token):
    """Return one local config while recording the pinned request."""

    calls.append((repo_id, revision, token))
    return canonical_json(snapshot_model.config)

fetched = fetch_config_with_transport(
    snapshot_model.repo_id,
    fake_revision,
    fake_transport,
    token="not-a-real-token",
    expected_config_sha256=snapshot_model.config_sha256,
)
assert fetched.revision == fake_revision
assert fetched.config_sha256 == snapshot_model.config_sha256
assert len(calls) == 1

source_snapshot = source_configs[snapshot_model.repo_id]
changed = copy.deepcopy(source_snapshot)
changed["max_position_embeddings"] += 1
changes = config_diff(source_snapshot, changed)
assert changes == [
    {
        "path": "max_position_embeddings",
        "snapshot": snapshot_model.config["max_position_embeddings"],
        "live": snapshot_model.config["max_position_embeddings"] + 1,
    }
]
display(pd.DataFrame(changes))

# Exact-commit optional call — intentionally not executed:
# pinned = fetch_hf_config(
#     "org/model", "<full 40-hex commit>", token=None,
#     expected_config_sha256="<optional reviewed config digest>",
# )
# display(pd.DataFrame(config_diff(source_snapshot, pinned.config)))

print('The pinned live-mode contract is verified without network')

RUN_LIVE_SOURCE_EXAMPLE = False
LIVE_REPO_ID = "openai/gpt-oss-20b"
LIVE_REVISION = "main"
LIVE_TOKEN = os.environ.get("HF_TOKEN")

resolver_calls = []


def fake_resolver(repo_id, revision, token):
    """Resolve a mutable source name to the reviewed fake commit used in this test."""

    resolver_calls.append((repo_id, revision, token))
    return fake_revision


resolved_fake = fetch_source_config_with_transports(
    snapshot_model.repo_id,
    "main",
    fake_resolver,
    fake_transport,
    token=None,
    expected_config_sha256=snapshot_model.config_sha256,
)
assert resolved_fake.requested_revision == "main"
assert resolved_fake.resolved_commit == fake_revision
assert resolved_fake.config_sha256 == snapshot_model.config_sha256
assert resolver_calls == [(snapshot_model.repo_id, "main", None)]

if RUN_LIVE_SOURCE_EXAMPLE:
    if importlib.util.find_spec("huggingface_hub") is None:
        subprocess.check_call(
            [sys.executable, "-m", "pip", "install", "-q", "huggingface_hub"]
        )
    live_source = fetch_hf_config_from_source(
        LIVE_REPO_ID,
        LIVE_REVISION,
        token=LIVE_TOKEN,
    )
    live_breakdown = count_parameters(live_source.config)
    display(
        pd.DataFrame(
            [
                {
                    "repo_id": live_source.repo_id,
                    "requested_revision": live_source.requested_revision,
                    "resolved_commit": live_source.resolved_commit,
                    "config_sha256": live_source.config_sha256,
                    "attention": attention_kind(live_source.config),
                    "ffn": moe_kind(live_source.config),
                    "context": context_info(live_source.config),
                    "shape_derived_total_B": live_breakdown.total / 1e9,
                }
            ]
        )
    )
    if LIVE_REPO_ID in source_configs:
        drift = config_diff(source_configs[LIVE_REPO_ID], live_source.config)
        display(pd.DataFrame(drift))
else:
    print(
        "Optional live source cell skipped. Set RUN_LIVE_SOURCE_EXAMPLE=True to resolve "
        "LIVE_REVISION and read config.json from the model repository."
    )
print('The live-source resolution contract is verified without network')

INK = '#2E3440'
MUTED = '#4C566A'
BORDER = '#D8DEE9'
BLUE = '#5E81AC'
ORANGE = '#D08770'
def plot_style():
    import matplotlib.pyplot as plt
    plt.rcParams.update({
        "font.family": "DejaVu Sans", "font.size": 12,
        "text.color": INK, "axes.labelcolor": INK, "axes.titlecolor": INK,
        "axes.titlesize": 17, "axes.titleweight": "bold", "axes.titlelocation": "left",
        "axes.labelsize": 13, "xtick.labelsize": 11, "ytick.labelsize": 11,
        "xtick.color": MUTED, "ytick.color": MUTED,
        "axes.edgecolor": BORDER, "axes.linewidth": .7,
        "axes.facecolor": "white", "figure.facecolor": "white", "savefig.facecolor": "white",
        "grid.color": BORDER, "grid.linewidth": .7, "axes.axisbelow": True,
        "legend.fontsize": 11, "legend.frameon": False, "savefig.dpi": 150,
        "xtick.major.size": 0, "ytick.major.size": 0,
    })

def draw_context(milestones, language="en"):
    """A logarithmic dot plot avoids overlapping labels and implied quality scores."""
    ru = language == "ru"
    rows = sorted(milestones, key=lambda r: r["year"])
    fig, ax = plt.subplots(figsize=(11, max(6.5, len(rows)*.42+1.6)), layout="constrained")
    fig.get_layout_engine().set(w_pad=.16, h_pad=.16)
    labels = [f"{int(r['year'])} · {r['model']}" for r in rows]
    ax.scatter([r["tokens"] for r in rows], range(len(rows)), color=BLUE, s=55)
    for i, row in enumerate(rows):
        ax.annotate(f"{row['tokens']:,}", (row["tokens"], i), xytext=(9, 0),
                    textcoords="offset points", va="center", fontsize=11, color=INK)
    ax.set_yticks(range(len(rows)), labels)
    ax.invert_yaxis()
    ax.set_xscale("log")
    ax.set_xlim(650, 80_000_000)
    ax.set_xticks([1e3, 1e4, 1e5, 1e6, 1e7], ["1K", "10K", "100K", "1M", "10M"])
    ax.grid(axis="x")
    ax.set_xlabel("заявленный контекст, токенов · логарифмическая ось" if ru else "declared context, tokens · logarithmic axis")
    ax.set_title("Выбранные контекстные окна · 2019–2026" if ru else "Selected context windows · 2019–2026", pad=18)
    return fig

def draw_parameters(rows, language="en"):
    """Pair total and active counts on one axis; do not imply latency from their ratio."""
    ru = language == "ru"
    rows = sorted(rows, key=lambda r: r["total_B"])
    fig, ax = plt.subplots(figsize=(11, max(5.5, len(rows)*.44+1.1)), layout="constrained")
    fig.get_layout_engine().set(w_pad=.16, h_pad=.16)
    for i, row in enumerate(rows):
        ax.plot([row["active_B"], row["total_B"]], [i, i], color=BORDER, linewidth=2)
    for field, color, marker, label in [("total_B", BLUE, "o", "total"), ("active_B", ORANGE, "s", "active")]:
        values = [r[field] for r in rows]
        ax.scatter(values, range(len(rows)), color=color, marker=marker, s=52, label=label, zorder=3)
        for i, value in enumerate(values):
            if field == "active_B" and value == rows[i]["total_B"]:
                continue
            ax.annotate(f"{value:g}", (value, i), xytext=(8, 0), textcoords="offset points", va="center", fontsize=11, color=INK)
    ax.set_yticks(range(len(rows)), [r["model"] for r in rows])
    ax.invert_yaxis()
    ax.set_xscale("log")
    ax.set_xlim(min(r["active_B"] for r in rows)*.7, max(r["total_B"] for r in rows)*2.4)
    ax.grid(axis="x")
    ax.set_xlabel("миллиарды параметров · логарифмическая ось" if ru else "billions of parameters · logarithmic axis")
    ax.set_title("Полные и активные параметры" if ru else "Total and active parameters", pad=36)
    ax.text(0, 1.012, "11 октября 2026 · соглашения active различаются; это не оценка скорости" if ru else
            "11 October 2026 · active conventions differ; not a speed estimate", transform=ax.transAxes,
            fontsize=11, color=MUTED)
    ax.legend(loc="upper right")
    return fig

plot_style()
parameter_rows = [dict(model=name, total_B=row["total_B"], active_B=row["active_B"]) for name,row in claim_df.iterrows()]
fig = draw_parameters(parameter_rows, 'en')
plt.show()
plt.close(fig)
fig = draw_context(bundle.context_milestones, 'en')
plt.show()
plt.close(fig)

"""KV examples for M1 §§7–8; weight payload and runtime overhead are excluded."""
from decimal import Decimal


def kv_payload_bytes(retained_lengths, kv_heads, head_dim, bytes_per_value):
    """Count K and V for one request, with an explicit retained length per layer."""
    lengths = list(retained_lengths)
    for name, value in (("kv_heads", kv_heads), ("head_dim", head_dim), ("bytes_per_value", bytes_per_value)):
        if type(value) is not int or value <= 0:
            raise ValueError(f"{name} must be a positive integer")
    if any(type(length) is not int or length < 0 for length in lengths):
        raise ValueError("retained lengths must be non-negative integers")
    return 2 * sum(lengths) * kv_heads * head_dim * bytes_per_value


def run_context_examples():
    lengths = [131072 if layer % 2 == 0 else 128 for layer in range(36)]
    hybrid = kv_payload_bytes(lengths, 8, 64, 2)
    full = kv_payload_bytes([131072]*36, 8, 64, 2)
    million = kv_payload_bytes([1_000_000]*80, 8, 128, 2)
    # Independent arithmetic: enumerate the K and V payload of each layer.
    enumerated = sum(length*8*64*2 for length in lengths for _ in ("K", "V"))
    assert hybrid == enumerated == 4_836_556_800
    assert Decimal(hybrid)/Decimal(2**30) == Decimal("4.50439453125")
    assert full == 9*2**30
    assert million == 327_680_000_000
    assert round(Decimal(million)/Decimal(2**30), 2) == Decimal("305.18")
    assert kv_payload_bytes([0, 0], 8, 64, 2) == 0
    for bad in ([True], [-1], [1.5]):
        try:
            kv_payload_bytes(bad, 8, 64, 2)
        except ValueError:
            pass
        else:
            raise AssertionError("Invalid cache length accepted")
    return {"hybrid_GiB": hybrid/2**30, "all_full_GiB": full/2**30,
            "million_token_GiB": million/2**30}



print(run_context_examples())

def expect_error(expected, fn, *args, **kwargs):
    """Assert that one explicit failure mode raises the declared exception."""

    try:
        fn(*args, **kwargs)
    except expected:
        return
    raise AssertionError(f"expected {expected.__name__} from {fn.__name__}")

expect_error(ConfigSchemaError, validate_commit_revision, "main")
expect_error(ConfigSchemaError, validate_source_revision, "")
expect_error(ConfigSchemaError, validate_source_revision, "../main")
expect_error(ConfigSchemaError, strict_json_loads, '{"x": 1, "x": 2}')
expect_error(ConfigSchemaError, strict_json_loads, '{"x": NaN}')

conflicting = copy.deepcopy(snapshot_model.config)
conflicting["num_experts"] = conflicting["num_local_experts"] + 1
expect_error(ConfigSchemaError, validate_config, conflicting)

too_many = copy.deepcopy(snapshot_model.config)
too_many["num_experts_per_tok"] = too_many["num_local_experts"] + 1
expect_error(ConfigSchemaError, validate_config, too_many)

bad_heads = copy.deepcopy(by_repo["Qwen/Qwen3-8B"].config)
bad_heads["num_key_value_heads"] = 7
expect_error(ConfigSchemaError, validate_config, bad_heads)

unsupported = dict(by_repo["Qwen/Qwen3-8B"].config,
                   architectures=["UnreviewedHybridForCausalLM"])
expect_error(ConfigSchemaError, count_parameters, unsupported)

expect_error(ConfigSchemaError, gpt_breakdown.active, "best_fit")
expect_error(ConfigSchemaError, weight_cost,
             total_params=10, active_params=5, bits_per_parameter=0,
             card_capacity_gb_decimal=80)

expect_error(
    ConfigIntegrityError,
    fetch_config_with_transport,
    snapshot_model.repo_id,
    fake_revision,
    fake_transport,
    expected_config_sha256="0" * 64,
)

def failing_transport(repo_id, revision, token):
    """Simulate a visible network/authentication failure without fallback."""

    raise ConnectionError("simulated live-mode failure")

expect_error(
    ConnectionError,
    fetch_config_with_transport,
    snapshot_model.repo_id,
    fake_revision,
    failing_transport,
)


def failing_resolver(repo_id, revision, token):
    """Simulate a visible Hub-resolution failure without fallback."""

    raise ConnectionError("simulated revision-resolution failure")


expect_error(
    ConnectionError,
    fetch_source_config_with_transports,
    snapshot_model.repo_id,
    "main",
    failing_resolver,
    fake_transport,
)

print('Negative tests passed')

import json

"""A minimal model passport from a validated config.json."""
import json

EXAMPLE_CONFIG_JSON = r"""
{
  "hidden_size": 2880,
  "num_hidden_layers": 24,
  "num_attention_heads": 64,
  "num_key_value_heads": 8,
  "num_local_experts": 32,
  "num_experts_per_tok": 4,
  "max_position_embeddings": 131072,
  "vocab_size": 201088,
  "rope_scaling": {"rope_type": "yarn", "factor": 32}
}
"""


def positive_int(config: dict, key: str) -> int:
    """Read one required positive integer field."""

    value = config.get(key)
    if isinstance(value, bool) or not isinstance(value, int) or value <= 0:
        raise ValueError(f"{key} must be a positive integer")
    return value


def one_of(config: dict, *keys: str):
    """Read one alias field without hiding conflicting values."""

    found = [(key, config[key]) for key in keys if config.get(key) is not None]
    if not found:
        return None
    if any(type(value) is not type(found[0][1]) or value != found[0][1]
           for _, value in found[1:]):
        raise ValueError(f"conflicting aliases: {[key for key, _ in found]}")
    return found[0][1]


def attention_kind(config: dict) -> str:
    """Classify MHA (Multi-Head Attention), MQA (Multi-Query Attention), GQA, or MLA from validated fields."""

    if "kv_lora_rank" in config:  # check the MLA signature first
        return f"MLA (kv_rank={positive_int(config, 'kv_lora_rank')})"
    heads = positive_int(config, "num_attention_heads")
    kv_heads = config.get("num_key_value_heads", heads)
    if isinstance(kv_heads, bool) or not isinstance(kv_heads, int) or kv_heads <= 0:
        raise ValueError("num_key_value_heads must be a positive integer")
    if kv_heads > heads or heads % kv_heads:
        raise ValueError("KV heads must divide Q heads")
    if kv_heads == heads:
        return f"MHA ({heads} heads)"
    if kv_heads == 1:
        return f"MQA ({heads} Q heads / 1 KV)"
    return f"GQA ({heads} Q heads / {kv_heads} KV)"


def moe_kind(config: dict) -> str:
    """Classify dense/MoE across common field aliases."""

    experts = one_of(config, "num_local_experts", "n_routed_experts", "num_experts")
    if experts is None:
        if any(config.get(k) is not None for k in ("num_experts_per_tok", "num_experts_per_token", "n_shared_experts", "num_shared_experts")):
            raise ValueError("expert settings require an expert count")
        return "dense"
    if isinstance(experts, bool) or not isinstance(experts, int) or experts <= 0:
        raise ValueError("expert count must be a positive integer")
    top_k = one_of(config, "num_experts_per_tok", "num_experts_per_token")
    if isinstance(top_k, bool) or not isinstance(top_k, int) or not 1 <= top_k <= experts:
        raise ValueError("top-k must be in [1, expert count]")
    shared = one_of(config, "n_shared_experts", "num_shared_experts")
    shared = 0 if shared is None else shared
    if isinstance(shared, bool) or not isinstance(shared, int) or shared < 0:
        raise ValueError("n_shared_experts must be a non-negative integer")
    return f"MoE: {experts} experts, top-{top_k}" + (f" + {shared} shared" if shared else "")


def passport(config: dict) -> dict:
    """Return a minimal structural passport without a network call."""

    return {
        "layers_x_hidden": (
            positive_int(config, "num_hidden_layers"),
            positive_int(config, "hidden_size"),
        ),
        "attention": attention_kind(config),
        "ffn": moe_kind(config),
        "context": positive_int(config, "max_position_embeddings"),
        "vocabulary": positive_int(config, "vocab_size"),
    }

base = dict(hidden_size=64, num_hidden_layers=2, num_attention_heads=8,
            num_key_value_heads=2, max_position_embeddings=256, vocab_size=128)
configs = {
    "dense_gqa": dict(base),
    "toy_local_moe": dict(base, num_local_experts=4, num_experts_per_tok=2),
    "toy_mla_moe": dict(base, kv_lora_rank=16, n_routed_experts=8,
                        num_experts_per_tok=2, n_shared_experts=1),
}
for name, cfg in configs.items():
    print(name, passport(cfg))

assert "MoE" in passport(configs["toy_local_moe"])["ffn"]
assert passport(configs["toy_mla_moe"])["attention"].startswith("MLA")
for change in ({"num_key_value_heads": 3}, {"num_attention_heads": True},
               {"num_local_experts": 4, "num_experts_per_tok": 5},
               {"num_local_experts": 4, "num_experts": 8, "num_experts_per_tok": 2}):
    try:
        passport(dict(base, **change))
    except ValueError:
        pass
    else:
        raise AssertionError("invalid configuration accepted")

system_passport = {'task': 'Answer the log-retention question with supporting evidence', 'documents': ['A', 'B'], 'model': 'Teaching substitute: prepared responses', 'permissions': 'Read-only; no network or writes', 'memory': 'Per question', 'acceptance': 'Correct period and manually labelled supporting quote', 'latency': 'Not measured for an LLM', 'cost': 'Not measured for an LLM'}
print(json.dumps(system_passport, ensure_ascii=False, indent=2))

DOCUMENTS = {'A': 'Keep an experiment log for 30 days after the experiment ends.', 'B': 'Delete a temporary export 7 days after it is created.'}
QUESTION = 'How many days must an experiment log be kept after the experiment ends?'
print(QUESTION)
for name, passage in DOCUMENTS.items():
    print(name, passage)

def check_prepared_answer(answer, documents, reference):
    """Check a structured day count and an exact reference passage for one fixed question."""
    if type(answer.get("days")) is not int or answer["days"] < 0:
        raise ValueError("days must be a nonnegative integer")
    doc_id = answer.get("document_id")
    quote = answer.get("quote")
    if not isinstance(doc_id, str) or not isinstance(quote, str):
        raise ValueError("document_id and quote must be strings")
    answer_correct = answer["days"] == reference["days"]
    quote_exists = bool(quote) and doc_id in documents and quote in documents[doc_id]
    # The required passage is manually labelled for this question; it is not inferred.
    supported = (quote_exists and answer_correct
                 and doc_id == reference["document_id"]
                 and reference["required_quote"] in quote)
    return dict(answer_correct=answer_correct, quote_exists=quote_exists,
                supported=supported, accepted=answer_correct and supported)

REFERENCE_ANSWER = {'days': 30, 'document_id': 'A', 'required_quote': 'Keep an experiment log for 30 days after the experiment ends.'}
prepared = [{'days': 30, 'document_id': 'A'}, {'days': 7, 'document_id': 'A'}, {'days': 30, 'document_id': 'B'}]
answers = [dict(row, quote=DOCUMENTS[row["document_id"]]) for row in prepared]
checks = [check_prepared_answer(row, DOCUMENTS, REFERENCE_ANSWER) for row in answers]
labels = ['Correct answer and source', 'Wrong retention period', 'Source about another object']
for label, result in zip(labels, checks):
    print(label, result)
assert checks == [{'answer_correct': True, 'quote_exists': True, 'supported': True, 'accepted': True}, {'answer_correct': False, 'quote_exists': True, 'supported': False, 'accepted': False}, {'answer_correct': True, 'quote_exists': True, 'supported': False, 'accepted': False}]

previous = {
    "model-a": {"access": "API", "context": 128_000, "max_output": 16_000, "date": "2026-08-06"},
    "model-b": {"access": "promised_weights", "context": 64_000, "max_output": 8_000, "date": "2026-08-06"},
    "model-d": {"access": "API", "context": 64_000, "max_output": 8_000, "date": "2026-08-06"},
}
current = {
    "model-a": {"access": "API", "context": 128_000, "max_output": 32_000, "date": "2026-10-11"},
    "model-b": {"access": "published_weights", "context": 64_000, "max_output": 8_000, "date": "2026-10-11"},
    "model-c": {"access": "restricted", "context": None, "max_output": 1_000_000, "date": "2026-10-11"},
    "model-d": {"access": "API", "context": 64_000, "max_output": 8_000, "date": "2026-10-11"},
}

def registry_changes(previous, current):
    changes = []
    for name in sorted(previous.keys() | current.keys()):
        if name not in previous:
            changes.append((name, "added", current[name]))
        elif name not in current:
            changes.append((name, "removed", previous[name]))
        else:
            delta = {key: (previous[name].get(key), current[name].get(key))
                     for key in sorted(previous[name].keys() | current[name].keys())
                     if key != "date" and previous[name].get(key) != current[name].get(key)}
            if delta:
                changes.append((name, "changed", delta))
    return changes

changes = registry_changes(previous, current)
for change in changes:
    print(change)
assert {name for name, *_ in changes} == {"model-a", "model-b", "model-c"}
assert current["model-c"]["context"] is None
assert registry_changes({"x":{"date":"old"}}, {"x":{"date":"new"}}) == []
assert registry_changes({"x":{}}, {}) == [("x", "removed", {})]