"""Fail-closed diagnostics for stochastic optimization windows. The module is intentionally dependency-light. It does not train a model or claim that one statistic proves convergence. It validates a versioned window, then reports complementary evidence about direction, noise, step scale, and observed objective movement. """ from __future__ import annotations from dataclasses import asdict, dataclass import hashlib import json import math import re from typing import Any _IDENTIFIER = re.compile(r"^[A-Za-z0-9][A-Za-z0-9._:/@-]{0,127}$") def _require_identifier(name: str, value: str) -> None: if not isinstance(value, str) or not _IDENTIFIER.fullmatch(value): raise ValueError(f"{name} must be a non-empty stable identifier") def _require_owner(name: str, value: str) -> None: _require_identifier(name, value) if not value.startswith("team:"): raise ValueError(f"{name} must name an accountable team: owner") def _require_positive_int(name: str, value: int) -> None: if isinstance(value, bool) or not isinstance(value, int) or value <= 0: raise ValueError(f"{name} must be a positive integer") def _require_finite(name: str, value: float) -> None: if isinstance(value, bool) or not isinstance(value, (int, float)): raise TypeError(f"{name} must be a real number") if not math.isfinite(value): raise ValueError(f"{name} must be finite") def _require_vector(name: str, value: tuple[float, ...], width: int) -> None: if not isinstance(value, tuple): raise TypeError(f"{name} must be an immutable tuple") if len(value) != width: raise ValueError(f"{name} shape must be ({width},), got ({len(value)},)") for index, item in enumerate(value): _require_finite(f"{name}[{index}]", item) def _canonical(value: Any) -> Any: if isinstance(value, dict): return {key: _canonical(item) for key, item in 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 OptimizationContract: contract_version: str parameter_names: tuple[str, ...] dtype: str learning_rate: float batch_size: int batches_per_window: int max_noise_scale: float min_alignment: float max_relative_step: float gradient_norm_floor: float parameter_norm_floor: float min_improving_fraction: float max_objective_excursion_ratio: float objective_scale_floor: float seed: int objective_id: str dataset_revision: str sampler_id: str parameter_version: str objective_owner: str gradient_owner: str data_owner: str gradient_source_id: str def __post_init__(self) -> None: _require_identifier("contract_version", self.contract_version) if not isinstance(self.parameter_names, tuple) or not self.parameter_names: raise ValueError("parameter_names must be a non-empty immutable tuple") for index, name in enumerate(self.parameter_names): _require_identifier(f"parameter_names[{index}]", name) if len(set(self.parameter_names)) != len(self.parameter_names): raise ValueError("parameter_names must be unique") if self.dtype != "binary64": raise ValueError("this implementation supports dtype=binary64 only") _require_positive_int("batch_size", self.batch_size) _require_positive_int("batches_per_window", self.batches_per_window) if isinstance(self.seed, bool) or not isinstance(self.seed, int) or self.seed < 0: raise ValueError("seed must be a non-negative integer") for name in ( "learning_rate", "max_noise_scale", "max_relative_step", "gradient_norm_floor", "parameter_norm_floor", "objective_scale_floor", ): value = getattr(self, name) _require_finite(name, value) if value <= 0: raise ValueError(f"{name} must be positive") _require_finite("min_alignment", self.min_alignment) if not -1.0 <= self.min_alignment <= 1.0: raise ValueError("min_alignment must be between -1 and 1") _require_finite("min_improving_fraction", self.min_improving_fraction) if not 0.0 <= self.min_improving_fraction <= 1.0: raise ValueError("min_improving_fraction must be between 0 and 1") _require_finite( "max_objective_excursion_ratio", self.max_objective_excursion_ratio ) if self.max_objective_excursion_ratio < 0.0: raise ValueError("max_objective_excursion_ratio must be non-negative") for name in ( "objective_id", "dataset_revision", "sampler_id", "parameter_version", ): _require_identifier(name, getattr(self, name)) for name in ("objective_owner", "gradient_owner", "data_owner"): _require_owner(name, getattr(self, name)) _require_identifier("gradient_source_id", self.gradient_source_id) @property def contract_id(self) -> str: return _content_id(self.contract_version, asdict(self)) @dataclass(frozen=True) class OptimizerWindow: contract_id: str window_id: str dtype: str gradient_source_id: str objective_id: str dataset_revision: str sampler_id: str parameter_version: str parameter_values: tuple[float, ...] reference_gradient: tuple[float, ...] batch_gradients: tuple[tuple[float, ...], ...] batch_sizes: tuple[int, ...] objective_values: tuple[float, ...] @dataclass(frozen=True) class OptimizationDiagnostic: contract_id: str window_id: str gradient_mean: tuple[float, ...] alignment: float noise_scale: float relative_step: float objective_progress: float improving_steps: int observed_steps: int improving_fraction: float max_objective_excursion_ratio: float decision: str def validate_window(contract: OptimizationContract, window: OptimizerWindow) -> None: if type(contract) is not OptimizationContract: raise TypeError("contract must be a concrete OptimizationContract") if type(window) is not OptimizerWindow: raise TypeError("window must be a concrete OptimizerWindow") if window.contract_id != contract.contract_id: raise ValueError("window contract_id does not match the full contract identity") _require_identifier("window_id", window.window_id) if window.dtype != contract.dtype: raise ValueError("window dtype does not match contract dtype") if window.gradient_source_id != contract.gradient_source_id: raise ValueError("window gradient_source_id does not match contract source") for name in ( "objective_id", "dataset_revision", "sampler_id", "parameter_version", ): if getattr(window, name) != getattr(contract, name): raise ValueError(f"window {name} does not match contract {name}") width = len(contract.parameter_names) _require_vector("parameter_values", window.parameter_values, width) _require_vector("reference_gradient", window.reference_gradient, width) if not isinstance(window.batch_gradients, tuple): raise TypeError("batch_gradients must be an immutable tuple") if len(window.batch_gradients) != contract.batches_per_window: raise ValueError("batch_gradients count does not match batches_per_window") for index, gradient in enumerate(window.batch_gradients): _require_vector(f"batch_gradients[{index}]", gradient, width) if not isinstance(window.batch_sizes, tuple): raise TypeError("batch_sizes must be an immutable tuple") if len(window.batch_sizes) != contract.batches_per_window: raise ValueError("batch_sizes count does not match batches_per_window") for index, size in enumerate(window.batch_sizes): _require_positive_int(f"batch_sizes[{index}]", size) if size != contract.batch_size: raise ValueError( f"batch_sizes[{index}] does not match contract batch_size" ) if not isinstance(window.objective_values, tuple): raise TypeError("objective_values must be an immutable tuple") expected_objectives = contract.batches_per_window + 1 if len(window.objective_values) != expected_objectives: raise ValueError( f"objective_values must contain {expected_objectives} boundary values" ) for index, objective in enumerate(window.objective_values): _require_finite(f"objective_values[{index}]", objective) def diagnose( contract: OptimizationContract, window: OptimizerWindow ) -> OptimizationDiagnostic: """Validate and summarize one fixed-source optimization window. ``reference_gradient`` is expected to come from the identity-bound probe source, not from one of the sampled mini-batches. The reported noise scale is the simple trace-variance divided by squared mean norm proxy; it is a diagnostic, not a universal convergence theorem or batch-size prescription. """ validate_window(contract, window) count = len(window.batch_gradients) width = len(contract.parameter_names) gradient_mean = tuple( math.fsum(gradient[index] for gradient in window.batch_gradients) / count for index in range(width) ) _require_vector("gradient_mean", gradient_mean, width) mean_norm_squared = math.fsum(value * value for value in gradient_mean) reference_norm_squared = math.fsum( value * value for value in window.reference_gradient ) parameter_norm_squared = math.fsum( value * value for value in window.parameter_values ) variance_trace = math.fsum( math.fsum( (gradient[index] - gradient_mean[index]) ** 2 for index in range(width) ) for gradient in window.batch_gradients ) / count for name, value in ( ("mean_norm_squared", mean_norm_squared), ("reference_norm_squared", reference_norm_squared), ("parameter_norm_squared", parameter_norm_squared), ("variance_trace", variance_trace), ): _require_finite(name, value) mean_norm = math.sqrt(mean_norm_squared) reference_norm = math.sqrt(reference_norm_squared) if reference_norm < contract.gradient_norm_floor: raise ValueError("reference gradient is below the diagnostic signal floor") if mean_norm < contract.gradient_norm_floor: alignment = 0.0 else: dot = math.fsum( mean * reference for mean, reference in zip(gradient_mean, window.reference_gradient) ) alignment = dot / (mean_norm * reference_norm) alignment = min(1.0, max(-1.0, alignment)) noise_denominator = max( mean_norm_squared, contract.gradient_norm_floor**2 ) noise_scale = variance_trace / noise_denominator step_components: list[float] = [] for index, gradient in enumerate(gradient_mean): step = contract.learning_rate * gradient if not math.isfinite(step): raise FloatingPointError(f"optimizer step overflow at parameter {index}") if gradient != 0.0 and step == 0.0: raise FloatingPointError(f"optimizer step underflow at parameter {index}") step_components.append(step) step_norm = math.sqrt(math.fsum(step * step for step in step_components)) if not math.isfinite(step_norm): raise FloatingPointError("optimizer step norm overflow") relative_step = step_norm / max( math.sqrt(parameter_norm_squared), contract.parameter_norm_floor ) objective_progress = window.objective_values[0] - window.objective_values[-1] improving_steps = sum( after < before for before, after in zip( window.objective_values, window.objective_values[1:] ) ) improving_fraction = improving_steps / count objective_scale = max( abs(window.objective_values[0]), contract.objective_scale_floor ) max_upward_excursion = max( 0.0, *( after - before for before, after in zip( window.objective_values, window.objective_values[1:] ) ), ) max_objective_excursion_ratio = max_upward_excursion / objective_scale for name, value in ( ("alignment", alignment), ("noise_scale", noise_scale), ("relative_step", relative_step), ("objective_progress", objective_progress), ("improving_fraction", improving_fraction), ("max_objective_excursion_ratio", max_objective_excursion_ratio), ): _require_finite(name, value) broken_signal = ( alignment < contract.min_alignment or objective_progress <= 0.0 or relative_step > contract.max_relative_step or improving_fraction < contract.min_improving_fraction or max_objective_excursion_ratio > contract.max_objective_excursion_ratio ) if broken_signal: decision = "STOP_AND_INVESTIGATE" elif noise_scale > contract.max_noise_scale: decision = "REDUCE_NOISE_OR_RATE" else: decision = "CONTINUE" return OptimizationDiagnostic( contract_id=contract.contract_id, window_id=window.window_id, gradient_mean=gradient_mean, alignment=alignment, noise_scale=noise_scale, relative_step=relative_step, objective_progress=objective_progress, improving_steps=improving_steps, observed_steps=count, improving_fraction=improving_fraction, max_objective_excursion_ratio=max_objective_excursion_ratio, decision=decision, ) ILLUSTRATIVE_CONTRACT = OptimizationContract( contract_version="noisy-optimizer-v1", parameter_names=("weight", "bias"), dtype="binary64", learning_rate=0.1, batch_size=32, batches_per_window=4, max_noise_scale=0.2, min_alignment=0.8, max_relative_step=0.2, gradient_norm_floor=1e-12, parameter_norm_floor=1e-12, min_improving_fraction=0.5, max_objective_excursion_ratio=0.25, objective_scale_floor=1e-12, seed=20260814, objective_id="illustrative-training-objective-v1", dataset_revision="illustrative-training-data-v1", sampler_id="illustrative-shuffled-minibatch-v1", parameter_version="illustrative-checkpoint-step-100-v1", objective_owner="team:model-quality", gradient_owner="team:training-platform", data_owner="team:training-data", gradient_source_id="illustrative-probe-set-v1", ) ILLUSTRATIVE_WINDOW = OptimizerWindow( contract_id=ILLUSTRATIVE_CONTRACT.contract_id, window_id="window-0001", dtype="binary64", gradient_source_id="illustrative-probe-set-v1", objective_id="illustrative-training-objective-v1", dataset_revision="illustrative-training-data-v1", sampler_id="illustrative-shuffled-minibatch-v1", parameter_version="illustrative-checkpoint-step-100-v1", parameter_values=(1.0, -1.0), reference_gradient=(1.0, -0.5), batch_gradients=( (1.2, -0.4), (0.8, -0.6), (1.1, -0.45), (0.9, -0.55), ), batch_sizes=(32, 32, 32, 32), objective_values=(2.0, 1.92, 1.95, 1.82, 1.76), ) def format_example() -> str: result = diagnose(ILLUSTRATIVE_CONTRACT, ILLUSTRATIVE_WINDOW) gradient = ",".join(f"{value:.6f}" for value in result.gradient_mean) return "\n".join( [ "example=illustrative_only", f"contract_version={ILLUSTRATIVE_CONTRACT.contract_version}", f"window={result.window_id}", f"gradient_mean=({gradient})", f"alignment={result.alignment:.3f}", f"noise_scale={result.noise_scale:.3f}", f"relative_step={result.relative_step:.3f}", f"objective_progress={result.objective_progress:.6f}", f"improving_steps={result.improving_steps}/{result.observed_steps}", "max_objective_excursion_ratio=" f"{result.max_objective_excursion_ratio:.3f}", f"decision={result.decision}", ] ) if __name__ == "__main__": print(format_example())