"""Content-addressed alignment audit for token-level training evidence.""" from __future__ import annotations from dataclasses import asdict, dataclass, is_dataclass import hashlib import json import math import re from typing import Any, Iterable MAX_EXAMPLES = 512 MAX_TOKENS_PER_EXAMPLE = 16_384 MAX_TOKEN_ID = 10_000_000 MIN_BINARY64_RESOLUTION = 1e-12 MAX_BINARY64_RESOLUTION = 1e-4 _IDENTIFIER = re.compile(r"^[A-Za-z0-9][A-Za-z0-9._:/@=+-]{0,159}$") _DIGEST = re.compile(r"^sha256:[0-9a-f]{64}$") def _identifier(name: str, value: object) -> str: if type(value) is not str or not _IDENTIFIER.fullmatch(value): raise ValueError(f"{name} must be a stable identifier") return value def _digest(name: str, value: object) -> str: if type(value) is not str or not _DIGEST.fullmatch(value): raise ValueError(f"{name} must be a sha256 digest") return value def _owner(name: str, value: object) -> str: checked = _identifier(name, value) if not checked.startswith("team:"): raise ValueError(f"{name} must identify an accountable team") return checked def _integer(name: str, value: object, minimum: int, maximum: int) -> int: if type(value) is not int or not minimum <= value <= maximum: raise ValueError(f"{name} must be an integer in [{minimum}, {maximum}]") return value def _finite(name: str, value: object) -> float: if isinstance(value, bool) or not isinstance(value, (int, float)): raise ValueError(f"{name} must be a real number") checked = float(value) if not math.isfinite(checked): raise ValueError(f"{name} must be finite") return checked def _canonical(value: Any) -> Any: if is_dataclass(value) and not isinstance(value, type): return _canonical(asdict(value)) if isinstance(value, dict): return {key: _canonical(item) for key, item in sorted(value.items())} if isinstance(value, tuple): return [_canonical(item) for item in value] return value def _content_id(prefix: str, payload: dict[str, Any]) -> str: encoded = json.dumps( _canonical(payload), sort_keys=True, separators=(",", ":"), allow_nan=False ).encode("utf-8") return f"{prefix}@sha256:{hashlib.sha256(encoded).hexdigest()}" @dataclass(frozen=True) class ObjectiveContract: contract_version: str objective_id: str objective_version: str objective_type: str tokenizer_id: str tokenizer_version: str label_policy: str mask_policy: str data_mixture_id: str data_mixture_version: str data_mixture_content_digest: str model_id: str model_version: str maximum_examples: int maximum_tokens_per_example: int minimum_active_tokens: int minimum_numeric_resolution: float maximum_token_loss: float numeric_convention: str data_owner: str objective_owner: str audit_owner: str def __post_init__(self) -> None: for field in ( "contract_version", "objective_id", "objective_version", "tokenizer_id", "tokenizer_version", "data_mixture_id", "data_mixture_version", "model_id", "model_version", ): _identifier(field, getattr(self, field)) _digest("data_mixture_content_digest", self.data_mixture_content_digest) for field in ("data_owner", "objective_owner", "audit_owner"): _owner(field, getattr(self, field)) if self.objective_type != "causal-next-token-cross-entropy": raise ValueError("unsupported objective_type") if self.label_policy != "next-token-shift-ignore=-100-v1": raise ValueError("unsupported label_policy") if self.mask_policy != "loss-mask-bool-active-v1": raise ValueError("unsupported mask_policy") if self.numeric_convention != "binary64-fsum-token-mean-v1": raise ValueError("unsupported numeric_convention") _integer("maximum_examples", self.maximum_examples, 1, MAX_EXAMPLES) _integer( "maximum_tokens_per_example", self.maximum_tokens_per_example, 2, MAX_TOKENS_PER_EXAMPLE, ) _integer( "minimum_active_tokens", self.minimum_active_tokens, 1, self.maximum_examples * self.maximum_tokens_per_example, ) resolution = _finite( "minimum_numeric_resolution", self.minimum_numeric_resolution ) if not MIN_BINARY64_RESOLUTION <= resolution <= MAX_BINARY64_RESOLUTION: raise ValueError("minimum_numeric_resolution is outside the safe range") maximum_loss = _finite("maximum_token_loss", self.maximum_token_loss) if not resolution <= maximum_loss <= 1.0 / resolution: raise ValueError("maximum_token_loss is outside the safe range") @property def content_id(self) -> str: return _content_id("objective-contract", asdict(self)) @dataclass(frozen=True) class ObjectiveExample: example_id: str document_id: str document_version: str mixture_partition: str token_ids: tuple[int, ...] label_ids: tuple[int, ...] loss_mask: tuple[bool, ...] token_losses: tuple[float, ...] example_content_id: str @classmethod def capture( cls, *, contract: ObjectiveContract, example_id: str, document_id: str, document_version: str, mixture_partition: str, token_ids: Iterable[int], label_ids: Iterable[int], loss_mask: Iterable[bool], token_losses: Iterable[float], ) -> "ObjectiveExample": if type(contract) is not ObjectiveContract: raise TypeError("contract must be a concrete ObjectiveContract") normalized_tokens = tuple(token_ids) normalized_labels = tuple(label_ids) normalized_mask = tuple(loss_mask) normalized_losses = tuple(token_losses) _validate_example_values( contract, normalized_tokens, normalized_labels, normalized_mask, normalized_losses, ) payload = { "example_id": example_id, "document_id": document_id, "document_version": document_version, "mixture_partition": mixture_partition, "token_ids": normalized_tokens, "label_ids": normalized_labels, "loss_mask": normalized_mask, "token_losses": normalized_losses, } return cls( **payload, example_content_id=_content_id("objective-example", payload), ) def __post_init__(self) -> None: for field in ( "example_id", "document_id", "document_version", "mixture_partition", "example_content_id", ): _identifier(field, getattr(self, field)) for field in ("token_ids", "label_ids", "loss_mask", "token_losses"): if type(getattr(self, field)) is not tuple: raise TypeError(f"{field} must be an immutable tuple") @dataclass(frozen=True) class ObjectiveEvidence: evidence_id: str observed_at: str source_id: str source_version: str contract_content_id: str objective_id: str objective_version: str tokenizer_id: str tokenizer_version: str data_mixture_id: str data_mixture_version: str data_mixture_content_digest: str model_id: str model_version: str examples: tuple[ObjectiveExample, ...] evidence_owner: str evidence_content_id: str @classmethod def capture( cls, *, contract: ObjectiveContract, evidence_id: str, observed_at: str, source_id: str, source_version: str, examples: Iterable[ObjectiveExample], evidence_owner: str, ) -> "ObjectiveEvidence": if type(contract) is not ObjectiveContract: raise TypeError("contract must be a concrete ObjectiveContract") frozen_examples = tuple(examples) if not 1 <= len(frozen_examples) <= contract.maximum_examples: raise ValueError("examples count is outside the contract") if not all(type(item) is ObjectiveExample for item in frozen_examples): raise TypeError("examples must contain concrete ObjectiveExample records") payload = { "evidence_id": evidence_id, "observed_at": observed_at, "source_id": source_id, "source_version": source_version, "contract_content_id": contract.content_id, "objective_id": contract.objective_id, "objective_version": contract.objective_version, "tokenizer_id": contract.tokenizer_id, "tokenizer_version": contract.tokenizer_version, "data_mixture_id": contract.data_mixture_id, "data_mixture_version": contract.data_mixture_version, "data_mixture_content_digest": contract.data_mixture_content_digest, "model_id": contract.model_id, "model_version": contract.model_version, "examples": frozen_examples, "evidence_owner": evidence_owner, } return cls( **payload, evidence_content_id=_content_id("objective-evidence", payload), ) def __post_init__(self) -> None: for field in ( "evidence_id", "observed_at", "source_id", "source_version", "contract_content_id", "objective_id", "objective_version", "tokenizer_id", "tokenizer_version", "data_mixture_id", "data_mixture_version", "model_id", "model_version", "evidence_content_id", ): _identifier(field, getattr(self, field)) _digest("data_mixture_content_digest", self.data_mixture_content_digest) _owner("evidence_owner", self.evidence_owner) if type(self.examples) is not tuple: raise TypeError("examples must be an immutable tuple") @dataclass(frozen=True) class ObjectiveAuditReport: contract_content_id: str evidence_content_id: str active_tokens: int mean_token_loss: float report_content_id: str status: str claim: str def _validate_example_values( contract: ObjectiveContract, token_ids: object, label_ids: object, loss_mask: object, token_losses: object, ) -> None: if not all(type(value) is tuple for value in (token_ids, label_ids, loss_mask, token_losses)): raise TypeError("token, label, mask, and loss values must be immutable tuples") length = len(token_ids) if not 2 <= length <= contract.maximum_tokens_per_example: raise ValueError("token count is outside the contract") if not len(label_ids) == len(loss_mask) == len(token_losses) == length: raise ValueError("token, label, mask, and loss lengths must match") for token in token_ids: _integer("token_id", token, 0, MAX_TOKEN_ID) for index, (label, active, loss) in enumerate( zip(label_ids, loss_mask, token_losses) ): if type(active) is not bool: raise TypeError("loss_mask entries must be bool") if active: _integer("label_id", label, 0, MAX_TOKEN_ID) if index >= length - 1 or label != token_ids[index + 1]: raise ValueError("active labels must equal the next token") elif label != -100: raise ValueError("inactive labels must use the declared -100 sentinel") checked_loss = _finite("token_loss", loss) if active: if checked_loss < contract.minimum_numeric_resolution: raise ValueError("active token loss is below numeric resolution") if checked_loss > contract.maximum_token_loss: raise ValueError("active token loss exceeds the contract") elif checked_loss != 0.0: raise ValueError("inactive token loss must be exactly zero") def audit_training_objective( contract: ObjectiveContract, evidence: ObjectiveEvidence ) -> ObjectiveAuditReport: """Validate objective alignment without attributing downstream capability.""" if type(contract) is not ObjectiveContract: raise TypeError("contract must be a concrete ObjectiveContract") if type(evidence) is not ObjectiveEvidence: raise TypeError("evidence must be concrete ObjectiveEvidence") contract.__post_init__() evidence.__post_init__() expected_scope = ( contract.content_id, contract.objective_id, contract.objective_version, contract.tokenizer_id, contract.tokenizer_version, contract.data_mixture_id, contract.data_mixture_version, contract.data_mixture_content_digest, contract.model_id, contract.model_version, ) actual_scope = ( evidence.contract_content_id, evidence.objective_id, evidence.objective_version, evidence.tokenizer_id, evidence.tokenizer_version, evidence.data_mixture_id, evidence.data_mixture_version, evidence.data_mixture_content_digest, evidence.model_id, evidence.model_version, ) if actual_scope != expected_scope: raise ValueError("evidence scope does not match the objective contract") if not 1 <= len(evidence.examples) <= contract.maximum_examples: raise ValueError("examples count is outside the contract") seen_examples: set[str] = set() seen_documents: set[tuple[str, str]] = set() active_losses: list[float] = [] for example in evidence.examples: if type(example) is not ObjectiveExample: raise TypeError("evidence contains a non-concrete example") example.__post_init__() if example.example_id in seen_examples: raise ValueError("duplicate example identity") document_identity = (example.document_id, example.document_version) if document_identity in seen_documents: raise ValueError("duplicate document identity") seen_examples.add(example.example_id) seen_documents.add(document_identity) _validate_example_values( contract, example.token_ids, example.label_ids, example.loss_mask, example.token_losses, ) payload = { field: getattr(example, field) for field in ( "example_id", "document_id", "document_version", "mixture_partition", "token_ids", "label_ids", "loss_mask", "token_losses", ) } if example.example_content_id != _content_id("objective-example", payload): raise ValueError("example content identity does not match contents") active_losses.extend( loss for loss, active in zip(example.token_losses, example.loss_mask) if active ) if len(active_losses) < contract.minimum_active_tokens: raise ValueError("evidence has insufficient active-token support") evidence_payload = { field: getattr(evidence, field) for field in ( "evidence_id", "observed_at", "source_id", "source_version", "contract_content_id", "objective_id", "objective_version", "tokenizer_id", "tokenizer_version", "data_mixture_id", "data_mixture_version", "data_mixture_content_digest", "model_id", "model_version", "examples", "evidence_owner", ) } if evidence.evidence_content_id != _content_id( "objective-evidence", evidence_payload ): raise ValueError("evidence content identity does not match contents") mean_loss = math.fsum(active_losses) / len(active_losses) if not math.isfinite(mean_loss) or mean_loss < contract.minimum_numeric_resolution: raise ArithmeticError("derived mean loss is numerically unsupported") report_payload = { "contract_content_id": contract.content_id, "evidence_content_id": evidence.evidence_content_id, "active_tokens": len(active_losses), "mean_token_loss": mean_loss, "status": "ALIGNED", "claim": "OBJECTIVE_ALIGNMENT_ONLY_NO_CAUSAL_CAPABILITY_CLAIM", } return ObjectiveAuditReport( **report_payload, report_content_id=_content_id("objective-report", report_payload), ) ILLUSTRATIVE_CONTRACT = ObjectiveContract( contract_version="training-objective-audit-v1", objective_id="causal-lm-objective", objective_version="objective-2026-08-25", objective_type="causal-next-token-cross-entropy", tokenizer_id="tokenizer:illustrative", tokenizer_version="tokenizer-v3", label_policy="next-token-shift-ignore=-100-v1", mask_policy="loss-mask-bool-active-v1", data_mixture_id="mixture:academy", data_mixture_version="mixture-v7", data_mixture_content_digest="sha256:" + "a" * 64, model_id="decoder:illustrative", model_version="model-v5", maximum_examples=16, maximum_tokens_per_example=128, minimum_active_tokens=3, minimum_numeric_resolution=1e-12, maximum_token_loss=100.0, numeric_convention="binary64-fsum-token-mean-v1", data_owner="team:data-curation", objective_owner="team:model-training", audit_owner="team:model-assurance", ) ILLUSTRATIVE_EXAMPLE = ObjectiveExample.capture( contract=ILLUSTRATIVE_CONTRACT, example_id="example-001", document_id="document-042", document_version="document-v2", mixture_partition="technical-text", token_ids=(101, 205, 309, 413), label_ids=(205, 309, 413, -100), loss_mask=(True, True, True, False), token_losses=(0.4, 0.8, 0.6, 0.0), ) ILLUSTRATIVE_EVIDENCE = ObjectiveEvidence.capture( contract=ILLUSTRATIVE_CONTRACT, evidence_id="objective-batch-001", observed_at="2026-08-25T00:00:00Z", source_id="trainer:batch-log", source_version="trainer-v9", examples=(ILLUSTRATIVE_EXAMPLE,), evidence_owner="team:model-assurance", ) def format_example() -> str: report = audit_training_objective( ILLUSTRATIVE_CONTRACT, ILLUSTRATIVE_EVIDENCE ) return "\n".join( ( "example=illustrative_only", f"contract_version={ILLUSTRATIVE_CONTRACT.contract_version}", f"objective={ILLUSTRATIVE_CONTRACT.objective_type}", f"evidence_content_id={report.evidence_content_id}", f"active_tokens={report.active_tokens}", f"mean_token_loss={report.mean_token_loss:.3f}", f"status={report.status}", f"claim={report.claim}", ) ) if __name__ == "__main__": print(format_example())