"""A bounded, dependency-free transformer-block contract. This is an inspectable teaching implementation. It validates one pre-norm, single-head, causal block; it is not a production attention kernel. """ from __future__ import annotations from dataclasses import asdict, dataclass import hashlib import json import math import re from typing import Any, Iterable MAX_HIDDEN = 32 MAX_SEQUENCE = 128 MAX_PARAMETERS = 100_000 MIN_BINARY64_RESOLUTION = 1e-12 MAX_BINARY64_RESOLUTION = 1e-4 _IDENTIFIER = re.compile(r"^[A-Za-z0-9][A-Za-z0-9._:/@-]{0,127}$") 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 _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 _bounded_number( name: str, value: object, *, resolution: float, maximum_absolute_value: float, ) -> float: checked = _finite(name, value) if checked != 0.0 and abs(checked) < resolution: raise ValueError(f"{name} contains a sub-resolution value") if abs(checked) > maximum_absolute_value: raise ValueError(f"{name} exceeds the contracted magnitude") return checked def _canonical(value: Any) -> Any: 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()}" def _vector( name: str, value: object, *, size: int, resolution: float, maximum_absolute_value: float, ) -> tuple[float, ...]: if type(value) is not tuple or len(value) != size: raise TypeError(f"{name} must be an immutable tuple of length {size}") return tuple( _bounded_number( name, item, resolution=resolution, maximum_absolute_value=maximum_absolute_value, ) for item in value ) def _matrix( name: str, value: object, *, rows: int, columns: int, resolution: float, maximum_absolute_value: float, ) -> tuple[tuple[float, ...], ...]: if type(value) is not tuple or len(value) != rows: raise TypeError(f"{name} must be an immutable tuple with {rows} rows") return tuple( _vector( f"{name}[{index}]", row, size=columns, resolution=resolution, maximum_absolute_value=maximum_absolute_value, ) for index, row in enumerate(value) ) @dataclass(frozen=True) class TransformerBlockContract: contract_version: str model_id: str model_version: str architecture_version: str hidden_size: int sequence_length: int mask_policy: str norm_placement: str norm_epsilon: float activation: str numeric_convention: str minimum_numeric_resolution: float maximum_absolute_input: float maximum_absolute_parameter: float maximum_absolute_output: float maximum_parameters: int model_owner: str contract_owner: str def __post_init__(self) -> None: for field in ( "contract_version", "model_id", "model_version", "architecture_version", ): _identifier(field, getattr(self, field)) _owner("model_owner", self.model_owner) _owner("contract_owner", self.contract_owner) _integer("hidden_size", self.hidden_size, 1, MAX_HIDDEN) _integer("sequence_length", self.sequence_length, 1, MAX_SEQUENCE) _integer("maximum_parameters", self.maximum_parameters, 1, MAX_PARAMETERS) if self.mask_policy != "causal-plus-padding-v1": raise ValueError("unsupported mask_policy") if self.norm_placement != "pre-layer-norm": raise ValueError("only the declared pre-layer-norm block is supported") if self.activation != "relu": raise ValueError("unsupported activation") if self.numeric_convention != "binary64-fsum-stable-softmax-v1": raise ValueError("unsupported numeric_convention") 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") epsilon = _finite("norm_epsilon", self.norm_epsilon) if epsilon < resolution or epsilon > 1e-2: raise ValueError("norm_epsilon is outside the safe range") input_limit = _finite("maximum_absolute_input", self.maximum_absolute_input) parameter_limit = _finite( "maximum_absolute_parameter", self.maximum_absolute_parameter ) output_limit = _finite( "maximum_absolute_output", self.maximum_absolute_output ) if not resolution <= input_limit <= 1.0 / resolution: raise ValueError("maximum_absolute_input is outside the safe range") if not resolution <= parameter_limit <= 1.0 / resolution: raise ValueError("maximum_absolute_parameter is outside the safe range") if not max(input_limit, parameter_limit) <= output_limit <= 1.0 / resolution: raise ValueError("maximum_absolute_output is outside the safe range") @property def content_id(self) -> str: return _content_id("transformer-contract", asdict(self)) @dataclass(frozen=True) class TransformerBlockWeights: weights_id: str weights_version: str contract_content_id: str q: tuple[tuple[float, ...], ...] k: tuple[tuple[float, ...], ...] v: tuple[tuple[float, ...], ...] o: tuple[tuple[float, ...], ...] norm1_scale: tuple[float, ...] norm1_bias: tuple[float, ...] mlp_in: tuple[tuple[float, ...], ...] mlp_out: tuple[tuple[float, ...], ...] norm2_scale: tuple[float, ...] norm2_bias: tuple[float, ...] weights_content_id: str @classmethod def capture( cls, *, contract: TransformerBlockContract, weights_id: str, weights_version: str, q: Iterable[Iterable[float]], k: Iterable[Iterable[float]], v: Iterable[Iterable[float]], o: Iterable[Iterable[float]], norm1_scale: Iterable[float], norm1_bias: Iterable[float], mlp_in: Iterable[Iterable[float]], mlp_out: Iterable[Iterable[float]], norm2_scale: Iterable[float], norm2_bias: Iterable[float], ) -> "TransformerBlockWeights": if type(contract) is not TransformerBlockContract: raise TypeError("contract must be a concrete TransformerBlockContract") size = contract.hidden_size normalized = { "q": tuple(tuple(row) for row in q), "k": tuple(tuple(row) for row in k), "v": tuple(tuple(row) for row in v), "o": tuple(tuple(row) for row in o), "norm1_scale": tuple(norm1_scale), "norm1_bias": tuple(norm1_bias), "mlp_in": tuple(tuple(row) for row in mlp_in), "mlp_out": tuple(tuple(row) for row in mlp_out), "norm2_scale": tuple(norm2_scale), "norm2_bias": tuple(norm2_bias), } for field in ("q", "k", "v", "o", "mlp_in", "mlp_out"): _matrix( field, normalized[field], rows=size, columns=size, resolution=contract.minimum_numeric_resolution, maximum_absolute_value=contract.maximum_absolute_parameter, ) for field in ("norm1_scale", "norm1_bias", "norm2_scale", "norm2_bias"): _vector( field, normalized[field], size=size, resolution=contract.minimum_numeric_resolution, maximum_absolute_value=contract.maximum_absolute_parameter, ) payload = { "weights_id": weights_id, "weights_version": weights_version, "contract_content_id": contract.content_id, **normalized, } return cls( **payload, weights_content_id=_content_id("transformer-weights", payload), ) def __post_init__(self) -> None: _identifier("weights_id", self.weights_id) _identifier("weights_version", self.weights_version) _identifier("contract_content_id", self.contract_content_id) _identifier("weights_content_id", self.weights_content_id) @dataclass(frozen=True) class TransformerBlockInput: evidence_id: str source_id: str source_version: str observed_at: str model_id: str model_version: str contract_content_id: str tokens: tuple[tuple[float, ...], ...] padding_mask: tuple[bool, ...] evidence_owner: str evidence_content_id: str @classmethod def capture( cls, *, contract: TransformerBlockContract, evidence_id: str, source_id: str, source_version: str, observed_at: str, tokens: Iterable[Iterable[float]], padding_mask: Iterable[bool], evidence_owner: str, ) -> "TransformerBlockInput": if type(contract) is not TransformerBlockContract: raise TypeError("contract must be a concrete TransformerBlockContract") normalized_tokens = tuple(tuple(row) for row in tokens) normalized_mask = tuple(padding_mask) _matrix( "tokens", normalized_tokens, rows=contract.sequence_length, columns=contract.hidden_size, resolution=contract.minimum_numeric_resolution, maximum_absolute_value=contract.maximum_absolute_input, ) if len(normalized_mask) != contract.sequence_length or not all( type(item) is bool for item in normalized_mask ): raise TypeError("padding_mask must contain one bool per token") payload = { "evidence_id": evidence_id, "source_id": source_id, "source_version": source_version, "observed_at": observed_at, "model_id": contract.model_id, "model_version": contract.model_version, "contract_content_id": contract.content_id, "tokens": normalized_tokens, "padding_mask": normalized_mask, "evidence_owner": evidence_owner, } return cls( **payload, evidence_content_id=_content_id("transformer-input", payload), ) def __post_init__(self) -> None: for field in ( "evidence_id", "source_id", "source_version", "observed_at", "model_id", "model_version", "contract_content_id", "evidence_content_id", ): _identifier(field, getattr(self, field)) _owner("evidence_owner", self.evidence_owner) if type(self.tokens) is not tuple: raise TypeError("tokens must be an immutable tuple") if type(self.padding_mask) is not tuple or not all( type(item) is bool for item in self.padding_mask ): raise TypeError("padding_mask must be an immutable tuple of bools") @dataclass(frozen=True) class TransformerBlockReport: contract_content_id: str weights_content_id: str evidence_content_id: str output: tuple[tuple[float, ...], ...] output_content_id: str status: str mask_policy: str norm_placement: str claim: str def _validate_weights( contract: TransformerBlockContract, weights: TransformerBlockWeights ) -> None: if type(weights) is not TransformerBlockWeights: raise TypeError("weights must be concrete TransformerBlockWeights") weights.__post_init__() if weights.contract_content_id != contract.content_id: raise ValueError("weights were captured for a different contract") size = contract.hidden_size parameter_count = 6 * size * size + 4 * size if parameter_count > contract.maximum_parameters: raise ValueError("weights exceed maximum_parameters") for field in ("q", "k", "v", "o", "mlp_in", "mlp_out"): checked = _matrix( field, getattr(weights, field), rows=size, columns=size, resolution=contract.minimum_numeric_resolution, maximum_absolute_value=contract.maximum_absolute_parameter, ) if checked != getattr(weights, field): raise ValueError(f"{field} is not canonically normalized") for field in ("norm1_scale", "norm1_bias", "norm2_scale", "norm2_bias"): checked = _vector( field, getattr(weights, field), size=size, resolution=contract.minimum_numeric_resolution, maximum_absolute_value=contract.maximum_absolute_parameter, ) if checked != getattr(weights, field): raise ValueError(f"{field} is not canonically normalized") payload = { field: getattr(weights, field) for field in ( "weights_id", "weights_version", "contract_content_id", "q", "k", "v", "o", "norm1_scale", "norm1_bias", "mlp_in", "mlp_out", "norm2_scale", "norm2_bias", ) } if weights.weights_content_id != _content_id("transformer-weights", payload): raise ValueError("weights content identity does not match contents") def _validate_input( contract: TransformerBlockContract, evidence: TransformerBlockInput ) -> None: if type(evidence) is not TransformerBlockInput: raise TypeError("evidence must be concrete TransformerBlockInput") evidence.__post_init__() if ( evidence.contract_content_id != contract.content_id or evidence.model_id != contract.model_id or evidence.model_version != contract.model_version ): raise ValueError("evidence scope does not match the contract") if len(evidence.tokens) != contract.sequence_length: raise ValueError("token sequence length does not match the contract") checked_tokens = _matrix( "tokens", evidence.tokens, rows=contract.sequence_length, columns=contract.hidden_size, resolution=contract.minimum_numeric_resolution, maximum_absolute_value=contract.maximum_absolute_input, ) if checked_tokens != evidence.tokens: raise ValueError("tokens are not canonically normalized") if len(evidence.padding_mask) != contract.sequence_length: raise ValueError("padding_mask length does not match the contract") seen_padding = False for padded in evidence.padding_mask: seen_padding = seen_padding or padded if seen_padding and not padded: raise ValueError("padding_mask must be a contiguous suffix") if all(evidence.padding_mask): raise ValueError("at least one token must be unmasked") payload = { field: getattr(evidence, field) for field in ( "evidence_id", "source_id", "source_version", "observed_at", "model_id", "model_version", "contract_content_id", "tokens", "padding_mask", "evidence_owner", ) } if evidence.evidence_content_id != _content_id("transformer-input", payload): raise ValueError("evidence content identity does not match contents") def _matvec(matrix: tuple[tuple[float, ...], ...], vector: tuple[float, ...]) -> tuple[float, ...]: return tuple(math.fsum(weight * item for weight, item in zip(row, vector)) for row in matrix) def _checked_output(contract: TransformerBlockContract, value: float) -> float: if not math.isfinite(value) or abs(value) > contract.maximum_absolute_output: raise OverflowError("derived transformer output breached the numeric contract") if value != 0.0 and abs(value) < contract.minimum_numeric_resolution: raise ArithmeticError("derived transformer output is below numeric resolution") return value def _add(contract: TransformerBlockContract, left: tuple[float, ...], right: tuple[float, ...]) -> tuple[float, ...]: return tuple(_checked_output(contract, math.fsum((a, b))) for a, b in zip(left, right)) def _layer_norm( contract: TransformerBlockContract, vector: tuple[float, ...], scale: tuple[float, ...], bias: tuple[float, ...], ) -> tuple[float, ...]: mean = math.fsum(vector) / len(vector) variance = math.fsum((item - mean) ** 2 for item in vector) / len(vector) denominator = math.sqrt(variance + contract.norm_epsilon) if not math.isfinite(denominator) or denominator <= 0.0: raise ArithmeticError("layer normalization denominator is invalid") return tuple( _checked_output(contract, ((item - mean) / denominator) * gain + offset) for item, gain, offset in zip(vector, scale, bias) ) def _softmax(values: tuple[float, ...]) -> tuple[float, ...]: pivot = max(values) exponentials = tuple(math.exp(value - pivot) for value in values) total = math.fsum(exponentials) if not math.isfinite(total) or total <= 0.0: raise ArithmeticError("stable softmax normalization failed") probabilities = tuple(value / total for value in exponentials) if not all(math.isfinite(value) for value in probabilities): raise ArithmeticError("stable softmax produced a non-finite probability") return probabilities def audit_transformer_block( contract: TransformerBlockContract, weights: TransformerBlockWeights, evidence: TransformerBlockInput, ) -> TransformerBlockReport: """Validate scope and run one bounded pre-norm causal block.""" if type(contract) is not TransformerBlockContract: raise TypeError("contract must be concrete TransformerBlockContract") contract.__post_init__() _validate_weights(contract, weights) _validate_input(contract, evidence) normalized = tuple( _layer_norm(contract, row, weights.norm1_scale, weights.norm1_bias) for row in evidence.tokens ) queries = tuple(_matvec(weights.q, row) for row in normalized) keys = tuple(_matvec(weights.k, row) for row in normalized) values = tuple(_matvec(weights.v, row) for row in normalized) scale = math.sqrt(contract.hidden_size) attention_outputs: list[tuple[float, ...]] = [] for query_index, query in enumerate(queries): if evidence.padding_mask[query_index]: attention_outputs.append((0.0,) * contract.hidden_size) continue visible = tuple( key_index for key_index in range(query_index + 1) if not evidence.padding_mask[key_index] ) if not visible: raise ValueError("an unmasked query has no visible keys") scores = tuple( math.fsum(a * b for a, b in zip(query, keys[index])) / scale for index in visible ) probabilities = _softmax(scores) attended = tuple( _checked_output( contract, math.fsum( probability * values[index][dimension] for probability, index in zip(probabilities, visible) ), ) for dimension in range(contract.hidden_size) ) attention_outputs.append(_matvec(weights.o, attended)) after_attention = tuple( row if padded else _add(contract, row, attention) for row, attention, padded in zip( evidence.tokens, attention_outputs, evidence.padding_mask ) ) output: list[tuple[float, ...]] = [] for row, padded in zip(after_attention, evidence.padding_mask): if padded: output.append(row) continue normalized_row = _layer_norm( contract, row, weights.norm2_scale, weights.norm2_bias ) hidden = tuple( max(0.0, _checked_output(contract, value)) for value in _matvec(weights.mlp_in, normalized_row) ) output.append(_add(contract, row, _matvec(weights.mlp_out, hidden))) frozen_output = tuple(output) output_content_id = _content_id( "transformer-output", { "contract_content_id": contract.content_id, "weights_content_id": weights.weights_content_id, "evidence_content_id": evidence.evidence_content_id, "output": frozen_output, }, ) return TransformerBlockReport( contract_content_id=contract.content_id, weights_content_id=weights.weights_content_id, evidence_content_id=evidence.evidence_content_id, output=frozen_output, output_content_id=output_content_id, status="PASS", mask_policy=contract.mask_policy, norm_placement=contract.norm_placement, claim="TEACHING_BLOCK_NOT_PRODUCTION_KERNEL", ) ILLUSTRATIVE_CONTRACT = TransformerBlockContract( contract_version="transformer-block-v1", model_id="tiny-decoder", model_version="model-2026-08-25", architecture_version="pre-ln-single-head-v1", hidden_size=2, sequence_length=3, mask_policy="causal-plus-padding-v1", norm_placement="pre-layer-norm", norm_epsilon=1e-5, activation="relu", numeric_convention="binary64-fsum-stable-softmax-v1", minimum_numeric_resolution=1e-12, maximum_absolute_input=100.0, maximum_absolute_parameter=100.0, maximum_absolute_output=1_000_000.0, maximum_parameters=1_000, model_owner="team:foundation-models", contract_owner="team:model-assurance", ) _IDENTITY = ((1.0, 0.0), (0.0, 1.0)) ILLUSTRATIVE_WEIGHTS = TransformerBlockWeights.capture( contract=ILLUSTRATIVE_CONTRACT, weights_id="tiny-block-weights", weights_version="weights-v1", q=_IDENTITY, k=_IDENTITY, v=_IDENTITY, o=_IDENTITY, norm1_scale=(1.0, 1.0), norm1_bias=(0.0, 0.0), mlp_in=((0.5, -0.25), (0.25, 0.5)), mlp_out=((0.25, 0.0), (0.0, 0.25)), norm2_scale=(1.0, 1.0), norm2_bias=(0.0, 0.0), ) ILLUSTRATIVE_INPUT = TransformerBlockInput.capture( contract=ILLUSTRATIVE_CONTRACT, evidence_id="block-input-001", source_id="fixture:transformer-block", source_version="fixture-v1", observed_at="2026-08-25T00:00:00Z", tokens=((1.0, -1.0), (0.5, 0.25), (-0.25, 0.75)), padding_mask=(False, False, False), evidence_owner="team:model-assurance", ) def format_example() -> str: report = audit_transformer_block( ILLUSTRATIVE_CONTRACT, ILLUSTRATIVE_WEIGHTS, ILLUSTRATIVE_INPUT ) return "\n".join( ( "example=illustrative_only", f"contract_version={ILLUSTRATIVE_CONTRACT.contract_version}", f"contract_content_id={report.contract_content_id}", f"weights_content_id={report.weights_content_id}", f"input_content_id={report.evidence_content_id}", f"shape={len(report.output)}x{len(report.output[0])}", "mask=causal-plus-padding", f"norm={report.norm_placement}", f"status={report.status}", f"claim={report.claim}", ) ) if __name__ == "__main__": print(format_example())