| """Library Drift Engine.
|
|
|
| Manages library version snapshots and triggers version upgrades during
|
| training to create non-stationary verification. In simulation mode it
|
| just tracks the current snapshot index — that index influences
|
| breakage selection and is exposed in observations so the Repair Agent
|
| can adapt.
|
|
|
| Also exposes Chojecki GVU's SNR computation
|
| (https://arxiv.org/abs/2512.02731 Definition 4.4).
|
| """
|
| from __future__ import annotations
|
|
|
| import math
|
| from dataclasses import dataclass, field
|
|
|
| DEFAULT_VERSION_SNAPSHOTS: list[dict[str, str]] = [
|
| {"transformers": "4.36.0", "datasets": "2.14.0", "trl": "0.7.0"},
|
| {"transformers": "4.40.0", "datasets": "2.18.0", "trl": "0.8.0"},
|
| {"transformers": "4.45.0", "datasets": "3.0.0", "trl": "0.10.0"},
|
| {"transformers": "4.50.0", "datasets": "3.2.0", "trl": "0.12.0"},
|
| ]
|
|
|
|
|
| @dataclass
|
| class LibraryDriftEngine:
|
| snapshots: list[dict[str, str]] = field(
|
| default_factory=lambda: list(DEFAULT_VERSION_SNAPSHOTS)
|
| )
|
| current_index: int = 0
|
| drift_history: list[dict] = field(default_factory=list)
|
|
|
| def current_versions(self) -> dict[str, str]:
|
| return dict(self.snapshots[self.current_index])
|
|
|
| def maybe_drift(self, episode_num: int, drift_every: int = 50) -> bool:
|
| if (
|
| episode_num > 0
|
| and episode_num % drift_every == 0
|
| and self.current_index < len(self.snapshots) - 1
|
| ):
|
| prev = self.snapshots[self.current_index]
|
| self.current_index += 1
|
| self.drift_history.append(
|
| {
|
| "episode": episode_num,
|
| "from": prev,
|
| "to": self.snapshots[self.current_index],
|
| }
|
| )
|
| return True
|
| return False
|
|
|
| def reset(self) -> None:
|
| self.current_index = 0
|
| self.drift_history.clear()
|
|
|
| @staticmethod
|
| def compute_snr(
|
| recent_held_out: list[float], recent_visible: list[float]
|
| ) -> dict[str, float]:
|
| """SNR per Chojecki GVU Def 4.4: SNR = mean(rewards)^2 / variance(rewards)."""
|
|
|
| def snr(values: list[float]) -> float:
|
| if len(values) < 2:
|
| return 0.0
|
| mean = sum(values) / len(values)
|
| var = sum((v - mean) ** 2 for v in values) / len(values)
|
| return mean**2 / max(var, 1e-8)
|
|
|
| return {
|
| "snr_verifier": snr(recent_held_out),
|
| "snr_generator": snr(recent_visible),
|
| }
|
|
|
