"""Versioned gates for an illustrative optimization-window diagnostic.""" from __future__ import annotations from dataclasses import dataclass from datetime import datetime import hashlib import json import math import sys from typing import NoReturn MAX_STEPS = 512 MAX_TEXT_LENGTH = 240 def _fail(message: str) -> NoReturn: raise ValueError(message) def _text(value: object, field: str) -> str: if type(value) is not str or not value.strip() or len(value) > MAX_TEXT_LENGTH: _fail(f"{field} must be non-empty bounded text") return value def _number(value: object, field: str, *, allow_zero: bool = True) -> float: if type(value) not in (int, float): _fail(f"{field} must be a concrete number") result = float(value) if not math.isfinite(result): _fail(f"{field} must be finite") if result != 0.0 and abs(result) < sys.float_info.min: _fail(f"{field} is below the supported numeric resolution") if result < 0.0 or (not allow_zero and result == 0.0): _fail(f"{field} must be {'positive' if not allow_zero else 'non-negative'}") return result def _utc_timestamp(value: object, field: str) -> datetime: text = _text(value, field) try: parsed = datetime.strptime(text, "%Y-%m-%dT%H:%M:%SZ") except ValueError: _fail(f"{field} must be a valid UTC timestamp in YYYY-MM-DDTHH:MM:SSZ form") if parsed.strftime("%Y-%m-%dT%H:%M:%SZ") != text: _fail(f"{field} must use canonical UTC timestamp form") return parsed def _update_ratio(update_norm: float, parameter_norm: float) -> float: ratio = update_norm / parameter_norm if not math.isfinite(ratio): _fail("update ratio overflowed the supported numeric range") if update_norm != 0.0 and (ratio == 0.0 or abs(ratio) < sys.float_info.min): _fail("update ratio is below the supported numeric resolution") return ratio def _digest(payload: object) -> str: data = json.dumps(payload, sort_keys=True, separators=(",", ":"), ensure_ascii=True) return "sha256:" + hashlib.sha256(data.encode("utf-8")).hexdigest() @dataclass(frozen=True) class DiagnosticGates: gates_version: str minimum_steps: int maximum_steps: int divergence_ratio: float minimum_relative_improvement: float stagnation_update_ratio: float maximum_update_ratio: float oscillation_fraction: float def __post_init__(self) -> None: _text(self.gates_version, "gates_version") if type(self.minimum_steps) is not int or type(self.maximum_steps) is not int: _fail("step bounds must be concrete integers") if not 3 <= self.minimum_steps <= self.maximum_steps <= MAX_STEPS: _fail(f"step bounds must satisfy 3 <= minimum <= maximum <= {MAX_STEPS}") if _number(self.divergence_ratio, "divergence_ratio", allow_zero=False) <= 1.0: _fail("divergence_ratio must be greater than one") for field in ("minimum_relative_improvement", "stagnation_update_ratio", "maximum_update_ratio"): value = _number(getattr(self, field), field) if not 0.0 <= value <= 1.0: _fail(f"{field} must be between zero and one") oscillation = _number(self.oscillation_fraction, "oscillation_fraction", allow_zero=False) minimum_material_fraction = 1.0 / (self.maximum_steps - 2) if not minimum_material_fraction <= oscillation <= 1.0: _fail("oscillation_fraction must be a material observable fraction for the maximum window") if self.stagnation_update_ratio >= self.maximum_update_ratio: _fail("stagnation_update_ratio must be below maximum_update_ratio") @dataclass(frozen=True) class TrainingStep: step: int loss: float gradient_norm: float update_norm: float parameter_norm: float def __post_init__(self) -> None: if type(self.step) is not int or self.step < 0: _fail("step must be a concrete non-negative integer") _number(self.loss, "loss") _number(self.gradient_norm, "gradient_norm") _number(self.update_norm, "update_norm") _number(self.parameter_norm, "parameter_norm", allow_zero=False) @dataclass(frozen=True) class TrainingWindow: window_id: str run_id: str model_revision: str dataset_revision: str objective_revision: str optimizer_revision: str precision_policy: str source_version: str window_started_at: str window_ended_at: str observed_at: str owner: str steps: tuple[TrainingStep, ...] def __post_init__(self) -> None: object.__setattr__(self, "steps", tuple(self.steps)) for field in ( "window_id", "run_id", "model_revision", "dataset_revision", "objective_revision", "optimizer_revision", "precision_policy", "source_version", "owner", ): _text(getattr(self, field), field) started_at = _utc_timestamp(self.window_started_at, "window_started_at") ended_at = _utc_timestamp(self.window_ended_at, "window_ended_at") observed_at = _utc_timestamp(self.observed_at, "observed_at") if not started_at < ended_at <= observed_at: _fail("window timestamps must satisfy started_at < ended_at <= observed_at") if not 1 <= len(self.steps) <= MAX_STEPS: _fail(f"steps must contain between 1 and {MAX_STEPS} records") if any(type(item) is not TrainingStep for item in self.steps): _fail("steps must contain concrete TrainingStep records") for item in self.steps: item.__post_init__() indices = [item.step for item in self.steps] if len(indices) != len(set(indices)): _fail("step IDs must be unique") if any(right != left + 1 for left, right in zip(indices, indices[1:])): _fail("steps must be ordered and contiguous") @dataclass(frozen=True) class OptimizationContract: contract_version: str run_id: str model_revision: str dataset_revision: str objective_revision: str optimizer_revision: str precision_policy: str owner: str window_id: str source_version: str first_step: int last_step: int window_started_at: str window_ended_at: str observed_at: str gates: DiagnosticGates def __post_init__(self) -> None: for field in ( "contract_version", "run_id", "model_revision", "dataset_revision", "objective_revision", "optimizer_revision", "precision_policy", "owner", "window_id", "source_version", ): _text(getattr(self, field), field) if type(self.first_step) is not int or type(self.last_step) is not int: _fail("contract step bounds must be concrete integers") if self.first_step < 0 or not self.first_step < self.last_step: _fail("contract step bounds must be non-negative and increasing") if self.last_step - self.first_step + 1 > MAX_STEPS: _fail(f"contract step span must not exceed {MAX_STEPS}") started_at = _utc_timestamp(self.window_started_at, "window_started_at") ended_at = _utc_timestamp(self.window_ended_at, "window_ended_at") observed_at = _utc_timestamp(self.observed_at, "observed_at") if not started_at < ended_at <= observed_at: _fail("contract timestamps must satisfy started_at < ended_at <= observed_at") if type(self.gates) is not DiagnosticGates: _fail("gates must be a concrete DiagnosticGates record") self.gates.__post_init__() @dataclass(frozen=True) class OptimizationAudit: evidence_content_id: str relative_improvement: float maximum_update_ratio: float oscillation_fraction: float gate: str decision: str interpretation: str def diagnose_optimization(contract: OptimizationContract, window: TrainingWindow) -> OptimizationAudit: if type(contract) is not OptimizationContract or type(window) is not TrainingWindow: _fail("diagnostic requires concrete OptimizationContract and TrainingWindow records") contract.__post_init__() window.__post_init__() for field in ( "run_id", "model_revision", "dataset_revision", "objective_revision", "optimizer_revision", "precision_policy", "owner", "window_id", "source_version", "window_started_at", "window_ended_at", "observed_at", ): if getattr(window, field) != getattr(contract, field): _fail(f"window {field} does not match the diagnostic contract") gates = contract.gates if not gates.minimum_steps <= len(window.steps) <= gates.maximum_steps: _fail("training window violates the versioned step bounds") if window.steps[0].step != contract.first_step or window.steps[-1].step != contract.last_step: _fail("training window step bounds do not match the diagnostic contract") losses = [step.loss for step in window.steps] ratios = [_update_ratio(step.update_norm, step.parameter_norm) for step in window.steps] start = losses[0] scale = max(abs(start), sys.float_info.min) relative_improvement = (start - losses[-1]) / scale if not math.isfinite(relative_improvement): _fail("relative improvement overflowed the supported numeric range") if relative_improvement != 0.0 and abs(relative_improvement) < sys.float_info.min: _fail("relative improvement is below the supported numeric resolution") maximum_ratio = max(ratios) deltas = [right - left for left, right in zip(losses, losses[1:])] nonzero_signs = [1 if delta > 0.0 else -1 for delta in deltas if delta != 0.0] reversals = sum(left != right for left, right in zip(nonzero_signs, nonzero_signs[1:])) oscillation = reversals / max(1, len(nonzero_signs) - 1) if maximum_ratio > gates.maximum_update_ratio: gate, interpretation = "EXCESSIVE_UPDATE_RATIO", "Updates are too large relative to parameter scale; inspect rate, scaling, clipping, and precision." elif losses[-1] > start * gates.divergence_ratio: gate, interpretation = "DIVERGENCE", "Window loss increased beyond the versioned gate; inspect data, objective, rate, and numeric stability." elif relative_improvement < gates.minimum_relative_improvement and max(ratios) <= gates.stagnation_update_ratio: gate, interpretation = "STAGNATION", "Loss and parameter movement are both small; inspect signal, gradients, rate, and capacity." elif oscillation >= gates.oscillation_fraction and relative_improvement < gates.minimum_relative_improvement: gate, interpretation = "OSCILLATION", "Loss-direction reversals exceed the gate without enough net improvement; inspect rate, batches, and momentum." else: gate, interpretation = "WITHIN_GATES", "This bounded window clears the declared gates; it does not prove convergence or identify a cause." payload = { "contract": { "version": contract.contract_version, "run": contract.run_id, "model": contract.model_revision, "data": contract.dataset_revision, "objective": contract.objective_revision, "optimizer": contract.optimizer_revision, "precision": contract.precision_policy, "owner": contract.owner, "window": [ contract.window_id, contract.source_version, contract.first_step, contract.last_step, contract.window_started_at, contract.window_ended_at, contract.observed_at, ], "gates": [ gates.gates_version, gates.minimum_steps, gates.maximum_steps, gates.divergence_ratio, gates.minimum_relative_improvement, gates.stagnation_update_ratio, gates.maximum_update_ratio, gates.oscillation_fraction, ], }, "window": { "id": window.window_id, "run": window.run_id, "model": window.model_revision, "data": window.dataset_revision, "objective": window.objective_revision, "optimizer": window.optimizer_revision, "precision": window.precision_policy, "source": window.source_version, "window_started_at": window.window_started_at, "window_ended_at": window.window_ended_at, "observed_at": window.observed_at, "owner": window.owner, "steps": [(s.step, s.loss, s.gradient_norm, s.update_norm, s.parameter_norm) for s in window.steps], }, } return OptimizationAudit( _digest(payload), relative_improvement, maximum_ratio, oscillation, gate, "CONTINUE" if gate == "WITHIN_GATES" else "HOLD", interpretation, ) ILLUSTRATIVE_GATES = DiagnosticGates("optimization-gates-v1", 5, 64, 1.25, 0.05, 0.0001, 0.05, 0.75) ILLUSTRATIVE_CONTRACT = OptimizationContract( contract_version="optimization-diagnostic-v1", run_id="training-run-0042", model_revision="encoder-r8", dataset_revision="dataset-v12", objective_revision="cross-entropy-v2", optimizer_revision="adamw-lr3e-4-v4", precision_policy="bf16-loss-scale-v2", owner="training-platform", window_id="window-0100-0104", source_version="trainer-telemetry-v5", first_step=100, last_step=104, window_started_at="2026-08-25T08:59:00Z", window_ended_at="2026-08-25T08:59:04Z", observed_at="2026-08-25T09:00:00Z", gates=ILLUSTRATIVE_GATES, ) ILLUSTRATIVE_WINDOW = TrainingWindow( window_id="window-0100-0104", run_id="training-run-0042", model_revision="encoder-r8", dataset_revision="dataset-v12", objective_revision="cross-entropy-v2", optimizer_revision="adamw-lr3e-4-v4", precision_policy="bf16-loss-scale-v2", source_version="trainer-telemetry-v5", window_started_at="2026-08-25T08:59:00Z", window_ended_at="2026-08-25T08:59:04Z", observed_at="2026-08-25T09:00:00Z", owner="training-platform", steps=( TrainingStep(100, 2.00, 1.20, 0.020, 10.0), TrainingStep(101, 1.82, 1.10, 0.018, 10.0), TrainingStep(102, 1.70, 1.00, 0.017, 10.0), TrainingStep(103, 1.58, 0.94, 0.016, 10.0), TrainingStep(104, 1.50, 0.90, 0.015, 10.0), ), ) def format_example() -> str: audit = diagnose_optimization(ILLUSTRATIVE_CONTRACT, ILLUSTRATIVE_WINDOW) return "\n".join(( "example=illustrative_only", f"gates_version={ILLUSTRATIVE_GATES.gates_version}", f"evidence_id={audit.evidence_content_id}", f"relative_improvement={audit.relative_improvement:.3f}", f"max_update_ratio={audit.maximum_update_ratio:.4f}", f"gate={audit.gate}", f"decision={audit.decision}", )) if __name__ == "__main__": print(format_example())