"""Audit activation and gradient signal paths under a bound evidence contract. The example is illustrative telemetry. A result classifies the observed path; it never attributes a training outcome to normalization or residual design. """ from __future__ import annotations from dataclasses import asdict, dataclass import hashlib import json import math import re from typing import Any, Iterable MAX_LAYERS = 512 MAX_SAMPLES_PER_LAYER = 10_000_000_000 MIN_BINARY64_RESOLUTION = 1e-15 MAX_BINARY64_RESOLUTION = 1e-3 _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 owner") return checked 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 _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 _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()}" @dataclass(frozen=True) class SignalObservation: observation_id: str layer_id: str layer_index: int activation_mean: float activation_variance: float gradient_mean: float gradient_variance: float sample_count: int architecture_id: str model_version: str dataset_version: str batch_id: str precision: str normalization_convention: str residual_convention: str window_id: str def __post_init__(self) -> None: for field in ( "observation_id", "layer_id", "architecture_id", "model_version", "dataset_version", "batch_id", "precision", "normalization_convention", "residual_convention", "window_id", ): _identifier(field, getattr(self, field)) _integer("layer_index", self.layer_index, minimum=0, maximum=MAX_LAYERS - 1) _integer( "sample_count", self.sample_count, minimum=1, maximum=MAX_SAMPLES_PER_LAYER, ) for field in ( "activation_mean", "activation_variance", "gradient_mean", "gradient_variance", ): value = _finite(field, getattr(self, field)) if "variance" in field and value < 0.0: raise ValueError(f"{field} must be non-negative") @classmethod def capture(cls, **values: object) -> "SignalObservation": return cls( observation_id=_identifier("observation_id", values["observation_id"]), layer_id=_identifier("layer_id", values["layer_id"]), layer_index=_integer( "layer_index", values["layer_index"], minimum=0, maximum=MAX_LAYERS - 1 ), activation_mean=_finite("activation_mean", values["activation_mean"]), activation_variance=_finite( "activation_variance", values["activation_variance"] ), gradient_mean=_finite("gradient_mean", values["gradient_mean"]), gradient_variance=_finite( "gradient_variance", values["gradient_variance"] ), sample_count=_integer( "sample_count", values["sample_count"], minimum=1, maximum=MAX_SAMPLES_PER_LAYER, ), architecture_id=_identifier("architecture_id", values["architecture_id"]), model_version=_identifier("model_version", values["model_version"]), dataset_version=_identifier("dataset_version", values["dataset_version"]), batch_id=_identifier("batch_id", values["batch_id"]), precision=_identifier("precision", values["precision"]), normalization_convention=_identifier( "normalization_convention", values["normalization_convention"] ), residual_convention=_identifier( "residual_convention", values["residual_convention"] ), window_id=_identifier("window_id", values["window_id"]), ) @dataclass(frozen=True) class SignalPathContract: contract_version: str architecture_id: str model_version: str dataset_version: str batch_id: str precision: str normalization_convention: str residual_convention: str window_id: str expected_layers: tuple[str, ...] minimum_sample_count: int minimum_variance_ratio: float maximum_variance_ratio: float maximum_absolute_mean: float maximum_absolute_statistic: float minimum_numeric_resolution: float maximum_layers: int numeric_convention: str data_owner: str model_owner: str training_owner: str def __post_init__(self) -> None: for field in ( "contract_version", "architecture_id", "model_version", "dataset_version", "batch_id", "precision", "normalization_convention", "residual_convention", "window_id", ): _identifier(field, getattr(self, field)) for field in ("data_owner", "model_owner", "training_owner"): _owner(field, getattr(self, field)) if type(self.expected_layers) is not tuple or len(self.expected_layers) < 2: raise TypeError("expected_layers must be an immutable tuple with two layers") maximum_layers = _integer( "maximum_layers", self.maximum_layers, minimum=2, maximum=MAX_LAYERS ) if len(self.expected_layers) > maximum_layers: raise ValueError("expected_layers exceed maximum_layers") checked_layers = tuple(_identifier("expected_layer", item) for item in self.expected_layers) if len(set(checked_layers)) != len(checked_layers): raise ValueError("expected_layers contain a duplicate identity") _integer( "minimum_sample_count", self.minimum_sample_count, minimum=2, maximum=MAX_SAMPLES_PER_LAYER, ) resolution = _finite( "minimum_numeric_resolution", self.minimum_numeric_resolution ) low = _finite("minimum_variance_ratio", self.minimum_variance_ratio) high = _finite("maximum_variance_ratio", self.maximum_variance_ratio) mean_limit = _finite("maximum_absolute_mean", self.maximum_absolute_mean) stat_limit = _finite( "maximum_absolute_statistic", self.maximum_absolute_statistic ) if not MIN_BINARY64_RESOLUTION <= resolution <= MAX_BINARY64_RESOLUTION: raise ValueError("minimum_numeric_resolution is outside the safe range") if low < resolution or low >= 1.0: raise ValueError("minimum_variance_ratio must be material and below one") if high <= 1.0 or high > 1.0 / resolution: raise ValueError("maximum_variance_ratio must be bounded above one") if mean_limit < resolution or mean_limit > 1.0 / resolution: raise ValueError("maximum_absolute_mean is outside the safe range") if stat_limit < mean_limit or stat_limit > 1.0 / resolution: raise ValueError("maximum_absolute_statistic is outside the safe range") if self.numeric_convention != "binary64-population-variance-ratio-v1": raise ValueError("unsupported numeric_convention") @property def content_id(self) -> str: return _content_id("signal-contract", asdict(self)) @dataclass(frozen=True) class SignalPathEvidence: evidence_id: str evidence_content_id: str contract_content_id: str observations: tuple[SignalObservation, ...] def __post_init__(self) -> None: _identifier("evidence_id", self.evidence_id) _identifier("evidence_content_id", self.evidence_content_id) _identifier("contract_content_id", self.contract_content_id) if type(self.observations) is not tuple: raise TypeError("observations must be an immutable tuple") if not self.observations or len(self.observations) > MAX_LAYERS: raise ValueError("observations are empty or exceed the global layer limit") if any(type(item) is not SignalObservation for item in self.observations): raise TypeError("observations must contain concrete SignalObservation values") @classmethod def capture( cls, *, evidence_id: str, contract: SignalPathContract, observations: Iterable[SignalObservation], ) -> "SignalPathEvidence": if type(contract) is not SignalPathContract: raise TypeError("contract must be a concrete SignalPathContract") contract.__post_init__() frozen = tuple(observations) if len(frozen) > contract.maximum_layers: raise ValueError("observations exceed contract maximum_layers") if any(type(item) is not SignalObservation for item in frozen): raise TypeError("observations must contain concrete SignalObservation values") selected_id = _identifier("evidence_id", evidence_id) return cls( evidence_id=selected_id, evidence_content_id=_evidence_content_id( contract.content_id, selected_id, frozen ), contract_content_id=contract.content_id, observations=frozen, ) @dataclass(frozen=True) class SignalPathReport: contract_content_id: str evidence_id: str evidence_content_id: str status: str first_flagged_layer: str | None activation_ratios: tuple[float | None, ...] gradient_ratios: tuple[float | None, ...] causal_claim: str def _evidence_content_id( contract_content_id: str, evidence_id: str, observations: tuple[SignalObservation, ...], ) -> str: return _content_id( "signal-evidence", { "contract_content_id": contract_content_id, "evidence_id": evidence_id, "observations": tuple(asdict(item) for item in observations), }, ) def _validate_stat(contract: SignalPathContract, name: str, value: object) -> float: checked = _finite(name, value) if abs(checked) > contract.maximum_absolute_statistic: raise ValueError(f"{name} exceeds maximum_absolute_statistic") if checked != 0.0 and abs(checked) < contract.minimum_numeric_resolution: raise ValueError(f"{name} is below the contracted numeric resolution") return checked def audit_signal_path( contract: SignalPathContract, evidence: SignalPathEvidence ) -> SignalPathReport: if type(contract) is not SignalPathContract: raise TypeError("contract must be a concrete SignalPathContract") if type(evidence) is not SignalPathEvidence: raise TypeError("evidence must be concrete SignalPathEvidence") contract.__post_init__() evidence.__post_init__() if evidence.contract_content_id != contract.content_id: raise ValueError("evidence belongs to a different contract") if len(evidence.observations) != len(contract.expected_layers): raise ValueError("evidence does not cover every expected layer") expected_content_id = _evidence_content_id( contract.content_id, evidence.evidence_id, evidence.observations ) if evidence.evidence_content_id != expected_content_id: raise ValueError("evidence content identity does not match its observations") observation_ids: set[str] = set() layer_ids: set[str] = set() scope = ( contract.architecture_id, contract.model_version, contract.dataset_version, contract.batch_id, contract.precision, contract.normalization_convention, contract.residual_convention, contract.window_id, ) for index, observation in enumerate(evidence.observations): observation.__post_init__() if observation.observation_id in observation_ids: raise ValueError("duplicate observation identity") if observation.layer_id in layer_ids: raise ValueError("duplicate layer identity") observation_ids.add(observation.observation_id) layer_ids.add(observation.layer_id) if observation.layer_index != index or observation.layer_id != contract.expected_layers[index]: raise ValueError("layer order or coverage violates the contract") observed_scope = ( observation.architecture_id, observation.model_version, observation.dataset_version, observation.batch_id, observation.precision, observation.normalization_convention, observation.residual_convention, observation.window_id, ) if observed_scope != scope: raise ValueError("observation scope does not match the contract") if observation.sample_count < contract.minimum_sample_count: raise ValueError("layer evidence is below minimum_sample_count") for name in ( "activation_mean", "activation_variance", "gradient_mean", "gradient_variance", ): value = _validate_stat(contract, name, getattr(observation, name)) if "variance" in name and value < 0.0: raise ValueError(f"{name} must be non-negative") activation_ratios: list[float | None] = [] gradient_ratios: list[float | None] = [] flags_by_layer: list[list[str]] = [ ["EXPLODING"] if abs(observation.activation_mean) > contract.maximum_absolute_mean or abs(observation.gradient_mean) > contract.maximum_absolute_mean else [] for observation in evidence.observations ] for layer_index, (previous, current) in enumerate( zip(evidence.observations, evidence.observations[1:]), start=1 ): for field, target in ( ("activation_variance", activation_ratios), ("gradient_variance", gradient_ratios), ): denominator = getattr(previous, field) numerator = getattr(current, field) if denominator < contract.minimum_numeric_resolution: target.append(None) flags_by_layer[layer_index].append("UNSUPPORTED") continue ratio = numerator / denominator if not math.isfinite(ratio): raise OverflowError("variance ratio overflowed") target.append(ratio) if ratio < contract.minimum_variance_ratio: flags_by_layer[layer_index].append("VANISHING") elif ratio > contract.maximum_variance_ratio: flags_by_layer[layer_index].append("EXPLODING") priority = {"UNSUPPORTED": 0, "EXPLODING": 1, "VANISHING": 2} status, first = "STABLE", None for observation, flags in zip(evidence.observations, flags_by_layer): if flags: status = min(flags, key=priority.__getitem__) first = observation.layer_id break return SignalPathReport( contract_content_id=contract.content_id, evidence_id=evidence.evidence_id, evidence_content_id=evidence.evidence_content_id, status=status, first_flagged_layer=first, activation_ratios=tuple(activation_ratios), gradient_ratios=tuple(gradient_ratios), causal_claim="TELEMETRY_ONLY_NO_CAUSAL_ATTRIBUTION", ) def _example() -> None: contract = SignalPathContract( contract_version="signal-path-v1", architecture_id="resnet-demo", model_version="model-v7", dataset_version="data-v3", batch_id="batch-0042", precision="binary32", normalization_convention="pre-norm-layernorm-v1", residual_convention="identity-add-v1", window_id="steps-1200-1299", expected_layers=("stem", "block-01", "block-02"), minimum_sample_count=32, minimum_variance_ratio=0.5, maximum_variance_ratio=2.0, maximum_absolute_mean=0.5, maximum_absolute_statistic=1000.0, minimum_numeric_resolution=1e-12, maximum_layers=16, numeric_convention="binary64-population-variance-ratio-v1", data_owner="team:data", model_owner="team:model", training_owner="team:training", ) observations = tuple( SignalObservation.capture( observation_id=f"signal-{index}", layer_id=layer_id, layer_index=index, activation_mean=0.01 * index, activation_variance=activation_variance, gradient_mean=-0.01 * index, gradient_variance=gradient_variance, sample_count=256, architecture_id=contract.architecture_id, model_version=contract.model_version, dataset_version=contract.dataset_version, batch_id=contract.batch_id, precision=contract.precision, normalization_convention=contract.normalization_convention, residual_convention=contract.residual_convention, window_id=contract.window_id, ) for index, (layer_id, activation_variance, gradient_variance) in enumerate( zip(contract.expected_layers, (1.0, 0.9, 0.81), (1.0, 1.1, 1.21)) ) ) evidence = SignalPathEvidence.capture( evidence_id="signal-window-001", contract=contract, observations=observations ) report = audit_signal_path(contract, evidence) print("example=illustrative_only") print(f"contract_version={contract.contract_version}") print(f"evidence={report.evidence_id}") print(f"evidence_content_id={report.evidence_content_id}") print(f"layers={len(evidence.observations)}") print(f"activation_ratios={','.join(f'{value:.3f}' for value in report.activation_ratios if value is not None)}") print(f"gradient_ratios={','.join(f'{value:.3f}' for value in report.gradient_ratios if value is not None)}") print(f"status={report.status}") print(f"claim={report.causal_claim}") if __name__ == "__main__": _example()