"""Bounded RoPE invariance audit with explicit context and extrapolation claims.""" from __future__ import annotations from dataclasses import asdict, dataclass from datetime import datetime import hashlib import json import math import re from typing import Any MAX_DIMENSION = 128 MAX_CONTEXT = 1_000_000 MAX_CASES = 128 MAX_OPERATIONS = 2_000_000 MIN_RESOLUTION = 1e-300 MAX_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 numeric and not bool") checked = float(value) if not math.isfinite(checked): raise ValueError(f"{name} must be finite") if abs(checked) > limit: raise ValueError(f"{name} exceeds its contracted magnitude") if checked != 0.0 and abs(checked) < resolution: raise ValueError(f"{name} is below the contracted numeric resolution") 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, dimension: int, resolution: float, limit: float ) -> tuple[float, ...]: if type(value) is not tuple or len(value) != dimension: raise TypeError(f"{name} must be an immutable tuple with declared dimension") return tuple(_finite(name, item, resolution, limit) for item in value) @dataclass(frozen=True) class PositionContract: contract_version: str model_id: str model_version: str layer_id: str mechanism: str invariant: str training_context_length: int declared_context_length: int position_offset: int model_dimension: int rotary_base: float maximum_shift: int maximum_cases: int maximum_operations: int minimum_numeric_resolution: float maximum_absolute_input: float absolute_tolerance: float extrapolation_claim_policy: str scope: str model_owner: str audit_owner: str def __post_init__(self) -> None: for field in ( "contract_version", "model_id", "model_version", "layer_id", "scope", "model_owner", "audit_owner", ): _identifier(field, getattr(self, field)) if self.mechanism != "rope-complex-pair-rotation-v1": raise ValueError("unsupported position mechanism") if self.invariant != "relative-dot-product-shift-invariance-v1": raise ValueError("unsupported invariant") training = _integer( "training_context_length", self.training_context_length, 1, MAX_CONTEXT ) declared = _integer( "declared_context_length", self.declared_context_length, 1, MAX_CONTEXT ) if declared < training: raise ValueError("declared context cannot be shorter than the training context") _integer("position_offset", self.position_offset, 0, MAX_CONTEXT) dimension = _integer("model_dimension", self.model_dimension, 2, MAX_DIMENSION) if dimension % 2: raise ValueError("RoPE model_dimension must be even") _integer("maximum_shift", self.maximum_shift, 1, MAX_CONTEXT) _integer("maximum_cases", self.maximum_cases, 1, MAX_CASES) _integer("maximum_operations", self.maximum_operations, 1, MAX_OPERATIONS) resolution = _finite( "minimum_numeric_resolution", self.minimum_numeric_resolution, MIN_RESOLUTION, MAX_RESOLUTION ) if not MIN_RESOLUTION <= resolution <= MAX_RESOLUTION: raise ValueError("minimum_numeric_resolution is outside supported bounds") input_limit = _finite( "maximum_absolute_input", self.maximum_absolute_input, resolution, 1 / resolution ) base = _finite("rotary_base", self.rotary_base, resolution, 1_000_000.0) tolerance = _finite( "absolute_tolerance", self.absolute_tolerance, resolution, 1e-6 ) if input_limit < resolution or base < 2.0: raise ValueError("input limit or rotary base is outside supported bounds") if not resolution <= tolerance <= 1e-6: raise ValueError("absolute_tolerance is outside supported bounds") if self.extrapolation_claim_policy != "invariant-only-quality-unproven-v1": raise ValueError("unsupported extrapolation_claim_policy") @property def last_declared_position(self) -> int: return self.position_offset + self.declared_context_length - 1 @property def last_training_position(self) -> int: return self.position_offset + self.training_context_length - 1 @property def content_id(self) -> str: return _content_id("position-contract", asdict(self)) @dataclass(frozen=True) class PositionCase: case_id: str query_position: int key_position: int translation: int query: tuple[float, ...] key: tuple[float, ...] def __post_init__(self) -> None: _identifier("case_id", self.case_id) if type(self.query_position) is not int or type(self.key_position) is not int: raise TypeError("positions must be concrete integers") if type(self.translation) is not int: raise TypeError("translation must be a concrete integer") if type(self.query) is not tuple or type(self.key) is not tuple: raise TypeError("query and key must be immutable tuples") @classmethod def capture( cls, *, contract: PositionContract, case_id: str, query_position: int, key_position: int, translation: int, query: object, key: object, ) -> "PositionCase": if type(contract) is not PositionContract: raise TypeError("contract must be a concrete PositionContract") contract.__post_init__() lower, upper = contract.position_offset, contract.last_declared_position q_position = _integer("query_position", query_position, lower, upper) k_position = _integer("key_position", key_position, lower, upper) shift = _integer( "translation", translation, -contract.maximum_shift, contract.maximum_shift ) if shift == 0: raise ValueError("translation must be non-zero") _integer("shifted_query_position", q_position + shift, lower, upper) _integer("shifted_key_position", k_position + shift, lower, upper) return cls( case_id=_identifier("case_id", case_id), query_position=q_position, key_position=k_position, translation=shift, query=_vector( "query", query, contract.model_dimension, contract.minimum_numeric_resolution, contract.maximum_absolute_input ), key=_vector( "key", key, contract.model_dimension, contract.minimum_numeric_resolution, contract.maximum_absolute_input ), ) @dataclass(frozen=True) class PositionEvidence: evidence_id: str evidence_content_id: str contract_content_id: str source_id: str source_version: str observed_at: str scope: str cases: tuple[PositionCase, ...] 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) if type(self.cases) is not tuple: raise TypeError("cases must be an immutable tuple") @classmethod def capture( cls, *, contract: PositionContract, evidence_id: str, source_id: str, source_version: str, observed_at: str, scope: str, cases: object, ) -> "PositionEvidence": if type(contract) is not PositionContract: raise TypeError("contract must be a concrete PositionContract") contract.__post_init__() checked_scope = _identifier("scope", scope) if checked_scope != contract.scope: raise ValueError("evidence scope does not match contract scope") if type(cases) is not tuple or not 1 <= len(cases) <= contract.maximum_cases: raise TypeError("cases must be a non-empty bounded immutable tuple") checked_cases: list[PositionCase] = [] for item in cases: if type(item) is not PositionCase: raise TypeError("cases must contain concrete PositionCase records") checked_cases.append( PositionCase.capture( contract=contract, case_id=item.case_id, query_position=item.query_position, key_position=item.key_position, translation=item.translation, query=item.query, key=item.key, ) ) case_ids = [item.case_id for item in checked_cases] if len(case_ids) != len(set(case_ids)): raise ValueError("case IDs must be unique") operations = len(checked_cases) * contract.model_dimension * 12 if operations > contract.maximum_operations: raise ValueError("position evidence exceeds maximum_operations") 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, "cases": tuple(asdict(item) for item in checked_cases), } return cls( evidence_content_id=_content_id("position-evidence", payload), cases=tuple(checked_cases), **{key: value for key, value in payload.items() if key != "cases"}, ) @dataclass(frozen=True) class PositionCaseResult: case_id: str relative_offset: int translated_relative_offset: int original_dot: float translated_dot: float absolute_error: float status: str touches_outside_training_range: bool @dataclass(frozen=True) class PositionAudit: decision: str contract_content_id: str evidence_content_id: str mechanism: str invariant: str maximum_observed_position: int training_range_status: str invariant_status: str extrapolation_conclusion: str cases: tuple[PositionCaseResult, ...] material_evidence_id: str def _rotate( vector: tuple[float, ...], position: int, contract: PositionContract ) -> tuple[float, ...]: relative_position = position - contract.position_offset result: list[float] = [] for pair_start in range(0, contract.model_dimension, 2): pair_index = pair_start // 2 frequency = contract.rotary_base ** (-2.0 * pair_index / contract.model_dimension) angle = relative_position * frequency cosine, sine = math.cos(angle), math.sin(angle) first, second = vector[pair_start], vector[pair_start + 1] result.extend((first * cosine - second * sine, first * sine + second * cosine)) if any(not math.isfinite(item) for item in result): raise ArithmeticError("position rotation produced a non-finite value") return tuple(result) def _revalidate( contract: PositionContract, evidence: PositionEvidence ) -> PositionEvidence: if type(contract) is not PositionContract or type(evidence) is not PositionEvidence: raise TypeError("audit requires concrete PositionContract and PositionEvidence") contract.__post_init__() evidence.__post_init__() rebuilt = PositionEvidence.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, cases=evidence.cases, ) if rebuilt.contract_content_id != evidence.contract_content_id: raise ValueError("evidence belongs to a different position contract") if rebuilt.evidence_content_id != evidence.evidence_content_id: raise ValueError("evidence content identity does not match its contents") return rebuilt def audit_position_extrapolation( contract: PositionContract, evidence: PositionEvidence ) -> PositionAudit: """Test the declared RoPE invariant without claiming long-context quality.""" checked = _revalidate(contract, evidence) results: list[PositionCaseResult] = [] maximum_position = contract.position_offset for case in checked.cases: shifted_query = case.query_position + case.translation shifted_key = case.key_position + case.translation maximum_position = max( maximum_position, case.query_position, case.key_position, shifted_query, shifted_key, ) original_dot = math.fsum( left * right for left, right in zip( _rotate(case.query, case.query_position, contract), _rotate(case.key, case.key_position, contract), ) ) translated_dot = math.fsum( left * right for left, right in zip( _rotate(case.query, shifted_query, contract), _rotate(case.key, shifted_key, contract), ) ) if not math.isfinite(original_dot) or not math.isfinite(translated_dot): raise ArithmeticError("position invariant dot product is non-finite") error = abs(original_dot - translated_dot) outside = max( case.query_position, case.key_position, shifted_query, shifted_key ) > contract.last_training_position results.append( PositionCaseResult( case_id=case.case_id, relative_offset=case.query_position - case.key_position, translated_relative_offset=shifted_query - shifted_key, original_dot=original_dot, translated_dot=translated_dot, absolute_error=error, status="PASS" if error <= contract.absolute_tolerance else "HOLD", touches_outside_training_range=outside, ) ) invariant_status = "PASS" if all(item.status == "PASS" for item in results) else "HOLD" outside_training = any(item.touches_outside_training_range for item in results) training_status = "OUTSIDE_TRAINING_RANGE" if outside_training else "WITHIN_TRAINING_RANGE" conclusion = ( "INVARIANT_PASSED_QUALITY_UNPROVEN" if invariant_status == "PASS" and outside_training else "INVARIANT_PASSED_WITHIN_TRAINING_RANGE" if invariant_status == "PASS" else "INVARIANT_FAILED_NO_EXTRAPOLATION_CLAIM" ) material = { "contract_content_id": contract.content_id, "evidence_content_id": checked.evidence_content_id, "mechanism": contract.mechanism, "invariant": contract.invariant, "training_range_status": training_status, "invariant_status": invariant_status, "extrapolation_conclusion": conclusion, "cases": tuple(asdict(item) for item in results), } return PositionAudit( decision="PASS" if invariant_status == "PASS" else "HOLD", contract_content_id=contract.content_id, evidence_content_id=checked.evidence_content_id, mechanism=contract.mechanism, invariant=contract.invariant, maximum_observed_position=maximum_position, training_range_status=training_status, invariant_status=invariant_status, extrapolation_conclusion=conclusion, cases=tuple(results), material_evidence_id=_content_id("position-material", material), ) ILLUSTRATIVE_CONTRACT = PositionContract( contract_version="position-extrapolation-v1", model_id="illustrative-transformer", model_version="model-v1", layer_id="layer-0-self-attention", mechanism="rope-complex-pair-rotation-v1", invariant="relative-dot-product-shift-invariance-v1", training_context_length=4, declared_context_length=8, position_offset=0, model_dimension=4, rotary_base=10000.0, maximum_shift=4, maximum_cases=8, maximum_operations=4096, minimum_numeric_resolution=1e-12, maximum_absolute_input=100.0, absolute_tolerance=1e-12, extrapolation_claim_policy="invariant-only-quality-unproven-v1", scope="academy:foundation-model-internals", model_owner="team:model", audit_owner="team:model-platform", ) ILLUSTRATIVE_EVIDENCE = PositionEvidence.capture( contract=ILLUSTRATIVE_CONTRACT, evidence_id="rope-shift-fixture", source_id="academy-fixture", source_version="fixture-v1", observed_at="2026-08-25T00:00:00+00:00", scope="academy:foundation-model-internals", cases=( PositionCase.capture( contract=ILLUSTRATIVE_CONTRACT, case_id="relative-offset-one", query_position=2, key_position=1, translation=3, query=(1.0, 2.0, 3.0, 4.0), key=(4.0, 3.0, 2.0, 1.0), ), ), ) def format_example() -> str: report = audit_position_extrapolation(ILLUSTRATIVE_CONTRACT, ILLUSTRATIVE_EVIDENCE) case = report.cases[0] return "\n".join( ( "example=illustrative_only", f"contract_id={report.contract_content_id}", f"evidence_id={report.evidence_content_id}", f"mechanism={report.mechanism}", f"relative_offsets={case.relative_offset},{case.translated_relative_offset}", f"maximum_position={report.maximum_observed_position}", f"training_range={report.training_range_status}", f"invariant={report.invariant_status}", f"extrapolation={report.extrapolation_conclusion}", f"material_id={report.material_evidence_id}", f"decision={report.decision}", ) ) if __name__ == "__main__": print(format_example())