"""Content-addressed scaled dot-product attention routing audit.""" from __future__ import annotations from dataclasses import asdict, dataclass from datetime import datetime import hashlib import json import math import re import sys from typing import Any MAX_SEQUENCE = 128 MAX_DIMENSION = 128 MAX_OPERATIONS = 2_000_000 MIN_BINARY64_RESOLUTION = 1e-300 MAX_BINARY64_RESOLUTION = 1e-3 IDENTIFIER = re.compile(r"^[A-Za-z0-9][A-Za-z0-9._:@/-]{0,127}$") SHA256_ID = re.compile(r"^[a-z][a-z0-9-]*@sha256:[0-9a-f]{64}$") def _identifier(name: str, value: object) -> str: if type(value) is not str or IDENTIFIER.fullmatch(value) is None: raise ValueError(f"{name} must be a bounded identifier") return value def _timestamp(name: str, value: object) -> str: if type(value) is not str or len(value) > 64: raise ValueError(f"{name} must be a bounded ISO-8601 timestamp") try: parsed = datetime.fromisoformat(value.replace("Z", "+00:00")) except ValueError as exc: raise ValueError(f"{name} must be an ISO-8601 timestamp") from exc if parsed.tzinfo is None: raise ValueError(f"{name} must include an offset") return value 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, *, resolution: float, limit: float) -> float: if type(value) not in (int, float) or type(value) is bool: raise TypeError(f"{name} must be an int or float, not bool") result = float(value) if not math.isfinite(result): raise ValueError(f"{name} must be finite") if abs(result) > limit: raise ValueError(f"{name} exceeds the contracted magnitude") if result != 0.0 and abs(result) < resolution: raise ValueError(f"{name} is below the contracted numeric resolution") return result 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 _matrix( name: str, value: object, *, rows: int, columns: int, resolution: float, limit: float, ) -> tuple[tuple[float, ...], ...]: if type(value) is not tuple or len(value) != rows: raise TypeError(f"{name} must be an immutable tuple with declared row count") checked_rows: list[tuple[float, ...]] = [] for row in value: if type(row) is not tuple or len(row) != columns: raise TypeError(f"{name} rows must be immutable tuples with declared width") checked_rows.append( tuple( _finite(name, item, resolution=resolution, limit=limit) for item in row ) ) return tuple(checked_rows) def _positions(name: str, value: object, *, count: int, maximum: int) -> tuple[int, ...]: if type(value) is not tuple or len(value) != count: raise TypeError(f"{name} must be an immutable tuple with declared length") checked = tuple(_integer(name, item, minimum=0, maximum=maximum) for item in value) if len(checked) != len(set(checked)) or tuple(sorted(checked)) != checked: raise ValueError(f"{name} must be unique and strictly increasing") return checked @dataclass(frozen=True) class AttentionContract: contract_version: str model_id: str model_version: str attention_role: str query_semantics: str key_semantics: str value_semantics: str query_dimension: int key_dimension: int value_dimension: int maximum_queries: int maximum_keys: int maximum_operations: int mask_semantics: str scaling_semantics: str softmax_semantics: str minimum_numeric_resolution: float maximum_absolute_input: float maximum_absolute_logit: float scope: str model_owner: str audit_owner: str def __post_init__(self) -> None: for field in ( "contract_version", "model_id", "model_version", "attention_role", "query_semantics", "key_semantics", "value_semantics", "scope", "model_owner", "audit_owner", ): _identifier(field, getattr(self, field)) for field in ("query_dimension", "key_dimension", "value_dimension"): _integer(field, getattr(self, field), minimum=1, maximum=MAX_DIMENSION) if self.query_dimension != self.key_dimension: raise ValueError("query and key dimensions must match for dot-product routing") for field in ("maximum_queries", "maximum_keys"): _integer(field, getattr(self, field), minimum=1, maximum=MAX_SEQUENCE) _integer( "maximum_operations", self.maximum_operations, minimum=1, maximum=MAX_OPERATIONS, ) if self.mask_semantics != "causal-key-position-lte-query-position-v1": raise ValueError("unsupported mask_semantics") if self.scaling_semantics != "divide-dot-by-sqrt-key-dimension-v1": raise ValueError("unsupported scaling_semantics") if self.softmax_semantics != "rowwise-max-shift-binary64-v1": raise ValueError("unsupported softmax_semantics") resolution = _finite( "minimum_numeric_resolution", self.minimum_numeric_resolution, resolution=MIN_BINARY64_RESOLUTION, limit=MAX_BINARY64_RESOLUTION, ) if not MIN_BINARY64_RESOLUTION <= resolution <= MAX_BINARY64_RESOLUTION: raise ValueError("minimum_numeric_resolution is outside supported binary64 bounds") input_limit = _finite( "maximum_absolute_input", self.maximum_absolute_input, resolution=resolution, limit=1.0 / resolution, ) logit_limit = _finite( "maximum_absolute_logit", self.maximum_absolute_logit, resolution=resolution, limit=100.0, ) if input_limit < resolution or logit_limit < resolution: raise ValueError("numeric bounds must be at least the declared resolution") @property def content_id(self) -> str: return _content_id("attention-contract", asdict(self)) @dataclass(frozen=True) class AttentionEvidence: evidence_id: str evidence_content_id: str contract_content_id: str source_id: str source_version: str observed_at: str scope: str query_positions: tuple[int, ...] key_positions: tuple[int, ...] queries: tuple[tuple[float, ...], ...] keys: tuple[tuple[float, ...], ...] values: tuple[tuple[float, ...], ...] def __post_init__(self) -> None: for field in ("evidence_id", "source_id", "source_version", "scope"): _identifier(field, getattr(self, field)) for field in ("evidence_content_id", "contract_content_id"): if type(getattr(self, field)) is not str or SHA256_ID.fullmatch(getattr(self, field)) is None: raise ValueError(f"{field} must be a content identity") _timestamp("observed_at", self.observed_at) for field in ("query_positions", "key_positions", "queries", "keys", "values"): if type(getattr(self, field)) is not tuple: raise TypeError(f"{field} must be an immutable tuple") @classmethod def capture( cls, *, contract: AttentionContract, evidence_id: str, source_id: str, source_version: str, observed_at: str, scope: str, query_positions: object, key_positions: object, queries: object, keys: object, values: object, ) -> "AttentionEvidence": if type(contract) is not AttentionContract: raise TypeError("contract must be a concrete AttentionContract") contract.__post_init__() checked_scope = _identifier("scope", scope) if checked_scope != contract.scope: raise ValueError("evidence scope does not match contract scope") if type(queries) is not tuple or type(keys) is not tuple or type(values) is not tuple: raise TypeError("Q, K, and V must be immutable tuples") query_count, key_count = len(queries), len(keys) if not 1 <= query_count <= contract.maximum_queries: raise ValueError("query count is outside the contract") if not 1 <= key_count <= contract.maximum_keys or len(values) != key_count: raise ValueError("key/value count is outside the contract or misaligned") operations = query_count * key_count * ( contract.key_dimension + contract.value_dimension ) if operations > contract.maximum_operations: raise ValueError("attention evidence exceeds maximum_operations") resolution = contract.minimum_numeric_resolution limit = contract.maximum_absolute_input checked = { "query_positions": _positions( "query_positions", query_positions, count=query_count, maximum=MAX_SEQUENCE * 4 ), "key_positions": _positions( "key_positions", key_positions, count=key_count, maximum=MAX_SEQUENCE * 4 ), "queries": _matrix( "queries", queries, rows=query_count, columns=contract.query_dimension, resolution=resolution, limit=limit ), "keys": _matrix( "keys", keys, rows=key_count, columns=contract.key_dimension, resolution=resolution, limit=limit ), "values": _matrix( "values", values, rows=key_count, columns=contract.value_dimension, resolution=resolution, limit=limit ), } payload = { "contract_content_id": contract.content_id, "evidence_id": _identifier("evidence_id", evidence_id), "source_id": _identifier("source_id", source_id), "source_version": _identifier("source_version", source_version), "observed_at": _timestamp("observed_at", observed_at), "scope": checked_scope, **checked, } return cls( evidence_content_id=_content_id("attention-evidence", payload), **payload, ) @dataclass(frozen=True) class AttentionAudit: decision: str contract_content_id: str evidence_content_id: str mask_semantics: str scaling_semantics: str weights: tuple[tuple[float, ...], ...] contexts: tuple[tuple[float, ...], ...] row_sums: tuple[float, ...] material_evidence_id: str def _checked_evidence( contract: AttentionContract, evidence: AttentionEvidence ) -> AttentionEvidence: if type(contract) is not AttentionContract or type(evidence) is not AttentionEvidence: raise TypeError("audit requires concrete AttentionContract and AttentionEvidence") contract.__post_init__() evidence.__post_init__() rebuilt = AttentionEvidence.capture( contract=contract, evidence_id=evidence.evidence_id, source_id=evidence.source_id, source_version=evidence.source_version, observed_at=evidence.observed_at, scope=evidence.scope, query_positions=evidence.query_positions, key_positions=evidence.key_positions, queries=evidence.queries, keys=evidence.keys, values=evidence.values, ) if rebuilt.contract_content_id != evidence.contract_content_id: raise ValueError("evidence belongs to a different attention contract") if rebuilt.evidence_content_id != evidence.evidence_content_id: raise ValueError("evidence content identity does not match its contents") return rebuilt def _stable_softmax(logits: tuple[float, ...], *, resolution: float) -> tuple[float, ...]: if not logits: raise ValueError("a query must have at least one unmasked key") peak = max(logits) exponentials = tuple(math.exp(value - peak) for value in logits) if any(value != 0.0 and value < resolution for value in exponentials): raise ValueError("softmax term is below the contracted numeric resolution") total = math.fsum(exponentials) if not math.isfinite(total) or total <= 0.0: raise ArithmeticError("softmax normalization is not finite and positive") weights = tuple(value / total for value in exponentials) if any(not math.isfinite(value) for value in weights): raise ArithmeticError("softmax produced a non-finite weight") return weights def audit_attention( contract: AttentionContract, evidence: AttentionEvidence ) -> AttentionAudit: """Compute masked, scaled routing with stable row-wise softmax.""" checked = _checked_evidence(contract, evidence) scale = math.sqrt(contract.key_dimension) all_weights: list[tuple[float, ...]] = [] contexts: list[tuple[float, ...]] = [] for query_position, query in zip(checked.query_positions, checked.queries): allowed_indices = tuple( index for index, key_position in enumerate(checked.key_positions) if key_position <= query_position ) logits: list[float] = [] for index in allowed_indices: dot = math.fsum(left * right for left, right in zip(query, checked.keys[index])) scaled = dot / scale if not math.isfinite(scaled) or abs(scaled) > contract.maximum_absolute_logit: raise ValueError("scaled attention logit exceeds the contracted range") logits.append(scaled) allowed_weights = _stable_softmax( tuple(logits), resolution=contract.minimum_numeric_resolution ) row = [0.0] * len(checked.keys) for index, weight in zip(allowed_indices, allowed_weights): row[index] = weight all_weights.append(tuple(row)) context = tuple( math.fsum(row[index] * checked.values[index][column] for index in range(len(row))) for column in range(contract.value_dimension) ) if any(not math.isfinite(value) for value in context): raise ArithmeticError("attention context is non-finite") contexts.append(context) rows = tuple(all_weights) row_sums = tuple(math.fsum(row) for row in rows) if any(not math.isclose(total, 1.0, rel_tol=0.0, abs_tol=sys.float_info.epsilon * 8) for total in row_sums): raise AssertionError("attention weights do not sum to one") material = { "contract_content_id": contract.content_id, "evidence_content_id": checked.evidence_content_id, "mask_semantics": contract.mask_semantics, "scaling_semantics": contract.scaling_semantics, "weights": rows, "contexts": tuple(contexts), } return AttentionAudit( decision="PASS", contract_content_id=contract.content_id, evidence_content_id=checked.evidence_content_id, mask_semantics=contract.mask_semantics, scaling_semantics=contract.scaling_semantics, weights=rows, contexts=tuple(contexts), row_sums=row_sums, material_evidence_id=_content_id("attention-material", material), ) ILLUSTRATIVE_CONTRACT = AttentionContract( contract_version="attention-routing-v1", model_id="illustrative-decoder", model_version="model-v1", attention_role="decoder-self-attention", query_semantics="request-for-content-by-token-position", key_semantics="address-of-content-by-token-position", value_semantics="payload-to-route-by-token-position", query_dimension=2, key_dimension=2, value_dimension=2, maximum_queries=8, maximum_keys=8, maximum_operations=1024, mask_semantics="causal-key-position-lte-query-position-v1", scaling_semantics="divide-dot-by-sqrt-key-dimension-v1", softmax_semantics="rowwise-max-shift-binary64-v1", minimum_numeric_resolution=1e-12, maximum_absolute_input=100.0, maximum_absolute_logit=40.0, scope="academy:foundation-model-internals", model_owner="team:model", audit_owner="team:model-platform", ) ILLUSTRATIVE_EVIDENCE = AttentionEvidence.capture( contract=ILLUSTRATIVE_CONTRACT, evidence_id="routing-fixture-1", source_id="academy-fixture", source_version="fixture-v1", observed_at="2026-08-25T00:00:00+00:00", scope="academy:foundation-model-internals", query_positions=(0, 1), key_positions=(0, 1), queries=((1.0, 0.0), (0.0, 1.0)), keys=((1.0, 0.0), (0.0, 1.0)), values=((10.0, 0.0), (0.0, 20.0)), ) def format_example() -> str: report = audit_attention(ILLUSTRATIVE_CONTRACT, ILLUSTRATIVE_EVIDENCE) return "\n".join( ( "example=illustrative_only", f"contract_id={report.contract_content_id}", f"evidence_id={report.evidence_content_id}", f"mask={report.mask_semantics}", f"query0_weights={','.join(f'{value:.3f}' for value in report.weights[0])}", f"query1_weights={','.join(f'{value:.3f}' for value in report.weights[1])}", f"query1_context={','.join(f'{value:.3f}' for value in report.contexts[1])}", f"material_id={report.material_evidence_id}", f"decision={report.decision}", ) ) if __name__ == "__main__": print(format_example())