"""Validator loop — manifest → eval → KOTH decision → weights. The validator never trains. Each round it: 1. Reads the current :class:`TrainingManifest` from the owner dataset repo and verifies its signature + that king and challenger share the contract digest (the controlled-experiment guarantee). 2. Pulls the king's and challenger's trained checkpoints and scores both on the *same* held-out eval windows. 3. Runs the paired-bootstrap KOTH verdict and folds it into the sticky champion state (``dethrone_cp`` consecutive wins to take the throne; ``dethrone_cp = 1`` makes it single-round). 4. Sets weights: equal share across the current king plus up to ``[scoring] reward_prior_kings`` registered prior kings (teutonic-style), collapsing to winner-take-all when ``reward_prior_kings = 0``. The pure orchestration in :meth:`ValidatorRunner.process_round` is testable by injecting ``evaluate_fn`` and ``windows``; HF + torch + chain are isolated behind the defaults. """ from __future__ import annotations import contextlib import hashlib import json import logging from collections.abc import Callable from dataclasses import dataclass, field, replace from datetime import UTC from pathlib import Path from typing import TYPE_CHECKING from ..eval.koth import RoundResult, evaluate_round from ..eval.scoring import WindowScore from ..eval.window import EvalWindow from ..shared.config import ChainConfig from ..shared.manifest import ( TrainedEntry, TrainingManifest, contract_digest, parse_trained_pointer, verify_signature, ) from ..shared.receipt import ( EntryScores, EvalContext, Participant, RoundReceipt, VerdictRecord, WindowScoreRecord, build_receipt, ) from . import state as state_mod from .state import ChampionState, StateTransition if TYPE_CHECKING: from ..trainer.remote import RemoteHost from .cascade import CascadeController log = logging.getLogger("cascade.validator") # Resolve a trained entry to its per-window scores on the eval set. EvaluateFn = Callable[[TrainedEntry, list[EvalWindow]], list[WindowScore]] # Resolve a trained entry to its gift-eval ratio rows for the public-benchmark # gate: ``{"status", "rows", "revision"}`` (see ``eval.benchmarks.run_gift_rows``) # or ``None`` when the sidecar produced nothing. GiftRowsFn = Callable[[TrainedEntry], dict | None] def participants_from_commitments(commitments: list, cutoff_block: int, floor_block: int = 0) -> tuple[Participant, ...]: """The round's eligible participant set, for the public receipt. Mirrors the trainer's eligibility rule (``trainer.loop.resolve_commitments``): parseable generator pointers revealed STRICTLY BEFORE the epoch boundary and at/after the go-live ``floor_block``, latest commit per hotkey among the eligible ones — but keeps ``commit_block`` so an auditor can re-check every entrant against the cutoff. Sorted by UID for a deterministic receipt body. ``commitments`` should carry each hotkey's FULL reveal history (``poll_commitments(include_history=True)``): with only the latest reveal, a miner who re-committed after the boundary vanishes from the record even though their eligible pre-cutoff entry fielded the round. """ from ..interface.validation import parse_commit best: dict[str, Participant] = {} for c in commitments: if floor_block and c.commit_block < floor_block: continue if c.commit_block >= cutoff_block: continue parsed = parse_commit(c.payload) if parsed is None: continue prev = best.get(c.hotkey) if prev is None or c.commit_block >= prev.commit_block: best[c.hotkey] = Participant( hotkey=c.hotkey, uid=c.uid, gen_ref=parsed.ref, commit_block=c.commit_block ) return tuple(sorted(best.values(), key=lambda p: p.uid)) @dataclass(frozen=True) class RoundOutcome: result: RoundResult transition: StateTransition # Every per-window score that fed the verdict, one record per evaluated # (role, size) entry in evaluation order — threaded out so the live loop can # publish them in the round's signed public receipt (cascade.shared.receipt). entry_scores: tuple[EntryScores, ...] = () # The king's tenure AT decision time (it set the margin); recorded in the # receipt so an auditor can recompute margin_for_tenure without validator state. king_tenure_rounds: int = 0 # How long the live loop keeps re-trying a round whose eval-pool index cannot # be READ at the pin gate (auth/network/5xx — not absence) before rejecting it # for real. Sized for observed Hippius blips (seconds-to-minutes) with a wide # margin — ~15 polls at the default 120s manifest cadence. A persistent failure # (bad credentials, dead endpoint) still ends in the loud reject receipt, just # this much later. POOL_PIN_READ_GRACE_SECONDS = 1800.0 @dataclass class ValidatorRunner: cfg: ChainConfig state: ChampionState = field(default_factory=ChampionState) evaluate_fn: EvaluateFn | None = None # injected in tests; defaults to registry+torch gift_rows_fn: GiftRowsFn | None = None # injected in tests; defaults to the sidecar bridge cache_dir: Path | None = None device: str = "cpu" # Optional resolver for the GPU eval-offload pod, called AT EACH offloaded # eval (see cascade.validator.eval_offload.make_eval_host_fn): the pod is # elastic — rented per round by the provisioner, torn down on the receipt — # so it must be re-resolved lazily, not captured at startup. ``None`` (or a # call returning ``None``) ⇒ that eval runs on ``device`` locally. The # wallet and every consensus decision stay on this box either way; the pod # is never used for the private-pool duel. eval_host_fn: Callable[[], RemoteHost | None] | None = None verify_signatures: bool = True # gate manifests on the trainer-hotkey signature # Cascade — king-reign promotion (see cascade.validator.cascade). When wired, # the reign clock is reset on each dethrone, every reigning-king checkpoint is # scored (GIFT-Eval + TIME) and logged, and once per round the clock is checked; # a fired Cascade installs the promoted warm-start init and re-crowns the # same king (DEC-CA-0004). None ⇒ Cascade is disabled (pure KOTH). cascade: CascadeController | None = None # Block of the last successful (or attempted re-assert) weight-set; drives # the between-rounds freshness push in _maybe_reassert_weights. None ⇒ # never set this process, so the first live-loop tick re-asserts # immediately (a restart is also a manual "refresh last_update now"). _last_weight_block: int | None = None # First-failure clock (time.monotonic) per round for an UNREADABLE eval-pool # index at the pin gate. Within POOL_PIN_READ_GRACE_SECONDS the round is # re-tried on the next manifest poll (no latch, no reject receipt); once the # grace expires the reject goes through, loudly. In-memory on purpose: a # restart merely restarts the grace clock, which is harmless. _pin_read_first_failure: dict[str, float] = field(default_factory=dict, repr=False) # ── manifest gating ───────────────────────────────────────────────────── def check_manifest(self, manifest: TrainingManifest) -> str | None: """Return a rejection reason string, or None if the manifest is usable. Enforces (1) the trainer-hotkey signature, (2) the contract-digest match (king and challenger trained under the same terms), and (3) that the manifest targets our configured base architecture and eval dataset. """ if self.verify_signatures and not verify_signature(manifest, self.cfg.manifest.trainer_hotkey): return "signature_invalid" want_contract = contract_digest(self.cfg.training) if manifest.contract_digest != want_contract: return f"contract_digest_mismatch: {manifest.contract_digest} != {want_contract}" if manifest.base_arch_digest != self.cfg.training.base_arch_digest: return "base_arch_digest_mismatch" if manifest.eval_dataset != self.cfg.eval.eval_dataset: return "eval_dataset_mismatch" gpu_reason = self._check_gpu(manifest) if gpu_reason is not None: return gpu_reason ws_reason = self._check_warm_start(manifest) if ws_reason is not None: return ws_reason return None def _check_warm_start(self, manifest: TrainingManifest) -> str | None: """Warm-start pin gate (Cascade, DEC-CA-0005): the manifest's signed ``warm_start_ckpt`` must equal the init THIS validator's deterministic promotion installed (its own ``warm_start_init_path`` file; "" before any promotion). Every validator computes the same promotion (block-anchored clock + trainer-signed bench scores), so agreement is fleet-wide. A mismatch — trainer trained from random when a promotion is live, or from a stale/foreign init — rejects the round rather than silently scoring runs trained off-baseline; the trainer re-syncs by the next round. Only enforced when Cascade is wired (off ⇒ pure KOTH, field ignored).""" if self.cascade is None: return None expected = "" p = Path(self.cfg.validator.warm_start_init_path) if p.is_file(): try: expected = str(json.loads(p.read_text(encoding="utf-8")).get("checkpoint_id") or "") except Exception as e: # noqa: BLE001 — unreadable pin must fail LOUD, not open return f"warm_start_state_unreadable: {p}: {e}" if manifest.warm_start_ckpt != expected: return (f"warm_start_mismatch: manifest trained from " f"{manifest.warm_start_ckpt or ''!r}, this validator " f"expects {expected or ''!r}") return None @staticmethod def check_pool_pin( manifest: TrainingManifest, window_source: object, *, block: int | None ) -> str | None: """Verify this round's eval pool against the trainer-signed pin. A pinned manifest carries the ``(key, sha256)`` of the snapshot the trainer screened on; the validator's own deterministic selection (same epoch block, same rule) must resolve to the identical pair. The pin is inside the signed body, so pool integrity descends from the trainer signature rather than the unsigned ``pool/index.json`` — a poisoned index or tampered tar surfaces here as a loud reject instead of scoring on attacker-chosen data. Unpinned manifests (older trainers) keep the legacy index-trust behaviour. Returns a reject reason or ``None``. Raises :class:`~cascade.shared.hippius.StorageError` when the pool index could not be READ (auth/network/5xx) — that is a transient, not a verdict, and the caller decides retry-vs-reject (the live loop retries within :data:`POOL_PIN_READ_GRACE_SECONDS`). Every other failure is a reject reason. """ if not (manifest.eval_pool_key and manifest.eval_pool_sha256): return None prov_fn = getattr(window_source, "provenance_for_round", None) if prov_fn is None: return ("pool_pin_unverifiable: manifest pins the eval pool but this " "validator's pool source reports no provenance") try: key, sha = prov_fn(int(manifest.round_id), block=block) except Exception as e: # noqa: BLE001 — an unverifiable pin must reject, not crash from ..shared.hippius import StorageError if isinstance(e, StorageError): raise # read failure, not a verdict — caller retries (see above) return f"pool_pin_unverifiable: provenance lookup failed: {e}" if not key or not sha: return ("pool_pin_unverifiable: manifest pins the eval pool but this " "validator resolved no snapshot for the round") if (key, sha) != (manifest.eval_pool_key, manifest.eval_pool_sha256): return (f"pool_pin_mismatch: manifest signed {manifest.eval_pool_key}@" f"{manifest.eval_pool_sha256[:12]}…, this validator resolved " f"{key}@{sha[:12]}…") return None def _pool_pin_read_failed( self, round_id: str, err: Exception, *, now: float | None = None ) -> str | None: """Grace bookkeeping for an UNREADABLE pool index at the pin gate. Returns ``None`` while ``round_id``'s read failures span less than :data:`POOL_PIN_READ_GRACE_SECONDS` — the caller then skips the cycle with no latch and no receipt, so the next manifest poll re-attempts the round from scratch. Once the grace expires, returns the terminal reject reason (and forgets the round, so a later re-publish starts a fresh grace window). """ import time now = time.monotonic() if now is None else now first = self._pin_read_first_failure.setdefault(str(round_id), now) waited = now - first if waited < POOL_PIN_READ_GRACE_SECONDS: log.warning( "pool index unreadable at pin gate for round=%s (%s); retrying " "next poll (%.0fs into %.0fs grace, no reject latched)", round_id, err, waited, POOL_PIN_READ_GRACE_SECONDS, ) return None self._pin_read_first_failure.pop(str(round_id), None) return ( f"pool_pin_unverifiable: provenance lookup failed persistently " f"({waited:.0f}s > {POOL_PIN_READ_GRACE_SECONDS:.0f}s grace): {err}" ) def _check_gpu(self, manifest: TrainingManifest) -> str | None: """Matched-hardware gate for byte-exact re-derivation. If ``[training] expected_gpu`` is pinned, every entry must report that GPU. Otherwise require only that king and challenger ran the same GPU (when both report one) — equal compute is already guaranteed by the token budget, but a byte-exact audit needs the comparison run on one SKU. """ pinned = self.cfg.training.expected_gpu gpus = {e.gpu_name for e in manifest.entries if e.gpu_name} if pinned: bad = sorted(g for g in gpus if g != pinned) if bad or any(not e.gpu_name for e in manifest.entries): return f"gpu_mismatch: expected {pinned!r}, manifest has {sorted(gpus)!r}" elif len(gpus) > 1: return f"gpu_mismatch: king/challenger on different GPUs {sorted(gpus)!r}" return None # ── per-round decision ────────────────────────────────────────────────── def _fetch_checkpoint_dir(self, entry: TrainedEntry) -> Path: """Fetch a trained checkpoint from the Hippius Hub registry to a local dir and return it. The OCI digest in the ref pins the bytes, so the fetch is self-verifying; repeated fetches of the same ref land in the same digest-named dir (cheap to reuse).""" from ..shared.hippius import HubConfig, HubRef, fetch_from_hub ref = parse_trained_pointer(entry.trained_pointer) if ref is None: raise ValueError(f"malformed trained_pointer: {entry.trained_pointer!r}") hub = HubConfig.from_storage(self.cfg.storage) dest = Path(self.cache_dir or "./_eval_ckpts") / HubRef.parse(ref).digest.replace(":", "-") fetch_from_hub(ref, dest, hub) return dest def _evaluate(self, entry: TrainedEntry, windows: list[EvalWindow]) -> list[WindowScore]: if self.evaluate_fn is not None: return self.evaluate_fn(entry, windows) # Default path: fetch the checkpoint from the Hippius Hub registry and # score it (registry + torch). from .evaluator import evaluate_checkpoint dest = self._fetch_checkpoint_dir(entry) return evaluate_checkpoint( dest, windows, num_samples=self.cfg.eval.num_samples, device=self.device ) # ── public-benchmark no-regression gate ───────────────────────────────── def _eval_host(self) -> RemoteHost | None: """The offload pod for THIS eval — resolved fresh every time so an elastic pod (rented per round manifest, torn down on the receipt) appears and disappears without a validator restart.""" return self.eval_host_fn() if self.eval_host_fn is not None else None def _gift_rows(self, entry: TrainedEntry) -> dict | None: """Gift-eval ratio rows for one entry — injected in tests, else the sidecar bridge on the fetched checkpoint dir.""" if self.gift_rows_fn is not None: return self.gift_rows_fn(entry) ec = self.cfg.eval dest = self._fetch_checkpoint_dir(entry) num_samples = ec.gift_gate_num_samples or ec.num_samples eval_host = self._eval_host() if eval_host is not None: # Offload the (heavy, paired) gift-eval to the GPU pod; the paired # bootstrap and every consensus decision stay on this box. from .eval_offload import gift_rows_via_host return gift_rows_via_host( eval_host, dest, datasets=ec.gift_gate_datasets, num_samples=num_samples, data_dir=(ec.gift_gate_data_dir or None), device="cuda", timeout_s=ec.gift_gate_timeout_s, ) from ..eval.benchmarks import run_gift_rows return run_gift_rows( dest, project_dir=ec.benchmark_project_dir, datasets=ec.gift_gate_datasets, num_samples=num_samples, device=self.device, data_dir=(ec.gift_gate_data_dir or None), timeout_s=ec.gift_gate_timeout_s, ) def _run_gift_gate( self, result: RoundResult, king_entry: TrainedEntry, chal_entry: TrainedEntry, *, seed: int | str, round_id: str, ) -> RoundResult: """Fold the public-benchmark gate into a *winning* round result. Scores both sides on gift-eval (via the sidecar bridge) and runs the paired no-regression bootstrap. The gate is uncomputable — and the round therefore inconclusive under ``enforce`` — when either sidecar run fails, gift-eval was skipped/errored, or the two runs scored against different pinned data revisions (a consensus-safety check: king and challenger must be judged on identical public data). """ from ..eval.gift_gate import evaluate_gift_gate, uncomputable_gate from ..eval.koth import apply_gift_gate p = self.cfg.koth_params() mode = p.gift_gate_mode king_run = self._gift_rows(king_entry) chal_run = self._gift_rows(chal_entry) if ( king_run is None or chal_run is None or king_run.get("status") != "ok" or chal_run.get("status") != "ok" ): gate = uncomputable_gate(p.gift_gate_tolerance, "gift-eval sidecar unavailable/errored") elif king_run.get("revision") != chal_run.get("revision"): gate = uncomputable_gate( p.gift_gate_tolerance, f"data-revision mismatch: king {king_run.get('revision')} != " f"chal {chal_run.get('revision')}", ) else: gate = evaluate_gift_gate( king_run["rows"], chal_run["rows"], tolerance=p.gift_gate_tolerance, alpha=p.bootstrap_alpha, B=p.bootstrap_B, seed=seed, min_configs=p.gift_gate_min_configs, ) log.info( "gift-gate round=%s mode=%s computed=%s passed=%s lcb=%s tol=%.4f " "n_configs=%d king_agg=%.5f chal_agg=%.5f%s", round_id, mode, gate.computed, gate.passed, f"{gate.lcb:.5f}" if gate.computed else "n/a", gate.tolerance, gate.n_configs, gate.king_agg, gate.chal_agg, "" if gate.computed else f" reason={gate.reason!r}", ) return apply_gift_gate(result, gate, mode=mode) def _maybe_run_benchmarks( self, manifest: TrainingManifest, outcome: RoundOutcome | None ) -> None: # pragma: no cover — exercised only in the live loop """Log public-benchmark numbers for a newly crowned king (log-only). Best-effort and strictly off the consensus path: it runs only when a challenger just dethroned the king, scores that new king's checkpoint via the isolated sidecar, and logs whatever comes back. Any failure is swallowed — a benchmark hiccup must never disturb weights or KOTH state. """ ec = self.cfg.eval if not ec.run_benchmarks: return if outcome is None or not outcome.transition.dethroned: return new_king = manifest.entry_for_role("challenger") if new_king is None: return try: from ..eval.benchmarks import format_report, run_benchmarks ckpt = self._fetch_checkpoint_dir(new_king) report = run_benchmarks( ckpt, project_dir=ec.benchmark_project_dir, suites=ec.benchmark_suites or ("gift-eval", "boom", "time"), num_samples=ec.benchmark_num_samples or ec.num_samples, max_series=ec.benchmark_max_series, device=self.device, ) if report is not None: log.info( "benchmarks round=%s king=%s %s", manifest.round_id, self.state.king_hotkey, format_report(report), ) except Exception as e: # noqa: BLE001 — log-only, never fatal log.warning("benchmark hook failed for round=%s: %s", manifest.round_id, e) # ── Cascade: king-reign promotion ──────────────────────────────────────── def _current_king_entry(self, manifest: TrainingManifest) -> TrainedEntry | None: """The manifest checkpoint the reigning champion produced this round. Cascade times the *validator's champion*, not the manifest's (lagging) king role — so the checkpoint to score is the entry whose miner hotkey is the champion. Prefers the primary throne size (what the benchmark sidecar scores) and falls back to any size that hotkey trained.""" hk = self.state.king_hotkey if hk is None: return None matches = [e for e in manifest.entries if e.miner_hotkey == hk] if not matches: return None primary = self.cfg.throne_contracts()[0].arch_preset return next((e for e in matches if e.size == primary), matches[0]) @staticmethod def _bench_scores_dict(entry: TrainedEntry) -> dict | None: """The six Cascade numbers off a manifest entry's trainer-signed ``bench_scores`` (GIFT-Eval / BOOM / TIME CRPS+MASE), or ``None`` when the entry carries none. This is the authoritative, consensus-safe source: every validator reads the identical signed numbers.""" bs = entry.bench_scores if bs is None: return None return { "gifteval_crps": bs.gifteval_crps, "gifteval_mase": bs.gifteval_mase, "boom_crps": bs.boom_crps, "boom_mase": bs.boom_mase, "time_crps": bs.time_crps, "time_mase": bs.time_mase, } def _bench_metrics_via_sidecar(self, entry: TrainedEntry) -> dict | None: # pragma: no cover — sidecar glue """Fallback: score one checkpoint on GIFT-Eval, BOOM, and TIME via the out-of-process sidecar, returning the six numbers or ``None`` when any suite is missing/errored. Used only when the manifest carries no ``bench_scores`` (e.g. a trainer that predates the Cascade hook). NOTE: independently-run GPU sweeps are not bit-reproducible, so this path is not consensus-safe across validators — prefer the trainer-signed numbers.""" from ..eval.benchmarks import extract_bench_scores, run_benchmarks ec = self.cfg.eval ckpt = self._fetch_checkpoint_dir(entry) num_samples = ec.benchmark_num_samples or ec.num_samples eval_host = self._eval_host() if eval_host is not None: # Offload the cascade bench (GIFT-Eval+BOOM+TIME) to the GPU pod, # same seam as the gift-eval gate; the wallet stays on this box. from .eval_offload import bench_scores_via_host metrics = bench_scores_via_host( eval_host, ckpt, num_samples=num_samples, max_series=ec.cascade_bench_max_series, # 0 = full battery data_dir=(ec.gift_gate_data_dir or None), device="cuda", timeout_s=ec.gift_gate_timeout_s, ) else: report = run_benchmarks( ckpt, project_dir=ec.benchmark_project_dir, suites=("gift-eval", "boom", "time"), num_samples=num_samples, max_series=ec.cascade_bench_max_series, # 0 = full battery device=self.device, ) metrics = extract_bench_scores(report) if metrics is None: log.warning("cascade: incomplete GIFT-Eval/BOOM/TIME metrics for king checkpoint %s; " "not recording this round", entry.trained_pointer) return metrics def _record_king_checkpoint( self, manifest: TrainingManifest, now: float ) -> None: # pragma: no cover — sidecar glue """Add the reigning king's checkpoint to the reign log so a later Cascade selection is a lookup, not a re-eval. Prefers the trainer's signed ``bench_scores`` on the manifest (consensus-safe); falls back to scoring via the local sidecar only when the manifest carries none. Best-effort: a miss just means this round's checkpoint isn't a promotion candidate.""" if self.cascade is None or self.state.king_hotkey is None: return entry = self._current_king_entry(manifest) if entry is None: return metrics = self._bench_scores_dict(entry) or self._bench_metrics_via_sidecar(entry) if metrics is None: return self.cascade.record_checkpoint(entry.trained_pointer, now=now, size=entry.size, **metrics) def _cascade_round( self, manifest: TrainingManifest, outcome: RoundOutcome | None ) -> None: # pragma: no cover — live-loop glue; the controller is unit-tested """One Cascade step, run at the end of a round (after weights/receipts). Resets the reign clock on a dethrone, records the reigning king's checkpoint, then checks the clock — a fired Cascade installs the promoted init and re-crowns the SAME king (DEC-CA-0004: the champion throne is never touched). Fully guarded: Cascade never disturbs KOTH.""" if self.cascade is None: return import time now = time.time() try: # The reign clock runs on the round's epoch block — identical for every # validator (from the signed manifest), so all fire on the same round. block = self._epoch_start_block(manifest) # Reuse KOTH's dethrone signal to reset the clock (never reimplement it); # on genesis, crown the first champion so the reign clock starts ticking. if outcome is not None and outcome.transition.dethroned and outcome.transition.new_king_hotkey: self.cascade.note_dethrone(outcome.transition.new_king_hotkey, block=block) elif self.cascade.state.king_hotkey is None and self.state.king_hotkey is not None: self.cascade.note_dethrone(self.state.king_hotkey, block=block) self._record_king_checkpoint(manifest, now) event = self.cascade.cascade_check(block=block, now=now) if event is not None: self._apply_cascade(event) except Exception as e: # noqa: BLE001 — Cascade must never disturb a round log.warning("cascade step failed for round=%s: %s", manifest.round_id, e) def _apply_cascade(self, event: object) -> None: # pragma: no cover — live-loop glue """Log a fired Cascade. The champion throne is deliberately untouched (DEC-CA-0004): the king persists — vacating had no benefit (both roles train from the shared init) and a vacant throne refillable only via the dethrone branch froze the reign clock when the incumbent kept winning.""" winner = getattr(event, "winner", None) king = getattr(event, "old_king", None) log.info( "cascade: promotion installed (king %s persists); field trains from " "checkpoint %s next round", (king or "?")[:12], getattr(winner, "checkpoint_id", "?"), ) def process_round( self, manifest: TrainingManifest, windows: list[EvalWindow], base_seed: int | str, ) -> RoundOutcome | None: """Evaluate one manifest against the eval windows and update state. A round carries one (king, challenger) pair PER trained size (the primary plus any ``[[training.sizes]]``). Each size's pair is scored on the SAME windows, then the per-size scores are POOLED — king's across sizes vs challenger's across sizes, in identical order — and a single paired bootstrap decides ONE throne on the combined score (scaling-aware KOTH). Pooling preserves pairing because each size's king and challenger share the window ``abs_target``. Returns None (king holds, no state change) when the manifest carries no size with both a king and a challenger, or fails the contract gate. Otherwise returns the round outcome with the (already-applied) transition. """ reason = self.check_manifest(manifest) if reason is not None: log.warning("rejecting manifest round=%s: %s", manifest.round_id, reason) return None king_by_size = {e.size: e for e in manifest.entries_for_role("king")} chal_by_size = {e.size: e for e in manifest.entries_for_role("challenger")} paired_sizes = [s for s in manifest.sizes() if s in king_by_size and s in chal_by_size] if not paired_sizes: log.info("manifest round=%s has no king/challenger pair; king holds", manifest.round_id) return None king_scores: list[WindowScore] = [] chal_scores: list[WindowScore] = [] score_records: list[EntryScores] = [] for size in paired_sizes: import time as _time _t0 = _time.perf_counter() ks = self._evaluate(king_by_size[size], windows) _t_king = _time.perf_counter() - _t0 _t1 = _time.perf_counter() cs = self._evaluate(chal_by_size[size], windows) _t_chal = _time.perf_counter() - _t1 log.info( "round=%s eval-timing size=%s device=%s n_windows=%d num_samples=%d " "king=%.1fs challenger=%.1fs total=%.1fs", manifest.round_id, size, self.device, len(windows), self.cfg.eval.num_samples, _t_king, _t_chal, _t_king + _t_chal, ) king_scores += ks chal_scores += cs for entry, scores in ((king_by_size[size], ks), (chal_by_size[size], cs)): score_records.append(EntryScores( role=entry.role, size=size, hotkey=entry.miner_hotkey, uid=entry.miner_uid, scores=tuple(WindowScoreRecord.from_score(s) for s in scores), )) # One challenger generator competes at every size, so any size's entry # carries its identity for the KOTH state machine. chal_entry = chal_by_size[paired_sizes[0]] tenure_at_decision = self.state.tenure_rounds result = evaluate_round( king_scores, chal_scores, self.cfg.koth_params(), seed=base_seed, king_tenure_rounds=tenure_at_decision, ) # Public-benchmark no-regression gate: only on a private-pool win, and # only when enabled. It can block a dethrone (or, uncomputable, hold the # round) but never grant one. Gated on the primary size's checkpoint # pair (the pooled decision spans sizes; the gate screens on one). if self.cfg.scoring.gift_gate_mode != "off" and result.challenger_wins_round: gate_size = ( self.cfg.training.arch_preset if self.cfg.training.arch_preset in king_by_size and self.cfg.training.arch_preset in chal_by_size else paired_sizes[0] ) result = self._run_gift_gate( result, king_by_size[gate_size], chal_by_size[gate_size], seed=base_seed, round_id=manifest.round_id, ) transition = state_mod.apply_round( self.state, challenger_hotkey=chal_entry.miner_hotkey, challenger_uid=chal_entry.miner_uid, result=result, dethrone_cp=self.cfg.scoring.dethrone_cp, keep_former_kings=self.cfg.scoring.reward_prior_kings, ) self.state = transition.state log.info( "round=%s lcb=%.4f margin=%.4f win=%s %s king=%s tenure=%d", manifest.round_id, result.lcb, result.margin, result.challenger_wins_round, transition.note, self.state.king_hotkey, self.state.tenure_rounds, ) # Shadow diagnostics: never gate the verdict. A rank-based view that # disagrees with the LCB, or a per-domain win-rate sign flip, means the # pool composition is doing the deciding — alert-worthy, not decisive. if result.win_rate is not None: log.info( "round=%s diag n_clusters=%d win_rate=%.3f wilcoxon_p=%s per_domain=%s", manifest.round_id, result.n_clusters, result.win_rate, f"{result.wilcoxon_p:.4g}" if result.wilcoxon_p is not None else "n/a", {d: f"{wr:.2f}/n{n}" for d, (wr, n) in (result.per_domain_win_rate or {}).items()}, ) return RoundOutcome( result=result, transition=transition, entry_scores=tuple(score_records), king_tenure_rounds=tenure_at_decision, ) def _epoch_start_block(self, manifest: TrainingManifest) -> int: """The round's epoch-boundary block: ``created_block`` floored to the epoch grid. Monotonic and identical for every validator (from the shared manifest), so it is the consensus key for daily eval-pool snapshot selection — unlike the round id, which is a block *hash* (non-monotonic). """ epoch_blocks = max(1, self.cfg.round.epoch_blocks) return (manifest.created_block // epoch_blocks) * epoch_blocks # ── public round receipts ──────────────────────────────────────────────── def build_round_receipt( self, manifest: TrainingManifest, *, base_seed: int, epoch_start_block: int, epoch_block_hash: str, outcome: RoundOutcome | None = None, windows: list[EvalWindow] | None = None, reject_reason: str | None = None, participants: tuple[Participant, ...] = (), pool_provenance: tuple[str, str] = ("", ""), reward_uids: tuple[int, ...] = (), weights: tuple[float, ...] = (), validator_hotkey: str = "", ) -> RoundReceipt: """Assemble the round's public receipt (pure — no I/O, no signing). A gated-out manifest — or one with no (king, challenger) pair to score — yields a ``rejected`` receipt carrying the reason; a scored round yields the full record: chain context, embedded manifest, participant set, the eval slice, every per-window score, the verdict, and the weight vector. """ from ..trainer.contract import RoundSeeds seeds = RoundSeeds.derive(base_seed, self.cfg.training) if reject_reason is not None: return build_receipt( round_id=manifest.round_id, status="rejected", epoch_start_block=epoch_start_block, epoch_block_hash=epoch_block_hash, base_seed=base_seed, seeds=seeds, manifest=manifest, participants=participants, reject_reason=reject_reason, reward_uids=reward_uids, weights=weights, validator_hotkey=validator_hotkey, ) if outcome is None or windows is None: raise ValueError("a scored receipt needs both outcome and windows") eval_context = EvalContext( pool_ref=pool_provenance[0], pool_digest=pool_provenance[1], window_ids=tuple(w.series_id for w in windows), n_windows=len(windows), num_samples=self.cfg.eval.num_samples, ) verdict = VerdictRecord.from_round( outcome.result, outcome.transition, params=self.cfg.koth_params(), bootstrap_seed=base_seed, king_tenure_rounds=outcome.king_tenure_rounds, ) return build_receipt( round_id=manifest.round_id, status="scored", epoch_start_block=epoch_start_block, epoch_block_hash=epoch_block_hash, base_seed=base_seed, seeds=seeds, manifest=manifest, participants=participants, eval_context=eval_context, entry_scores=outcome.entry_scores, verdict=verdict, reward_uids=reward_uids, weights=weights, validator_hotkey=validator_hotkey, ) def _publish_round_receipt( self, client: object, manifest: TrainingManifest, base_seed: int, *, outcome: RoundOutcome | None = None, windows: list[EvalWindow] | None = None, reject_reason: str | None = None, window_source: object = None, reward_uids: tuple[int, ...] = (), weights: tuple[float, ...] = (), ) -> None: # pragma: no cover — live-loop glue; assembly is unit-tested """Gather chain context, sign, and publish the round receipt. Best-effort by design: a receipt failure must never disturb weights or KOTH state (they are already committed), so every chain lookup degrades to an empty field and any publish error is logged and swallowed. The audit CLI treats missing context as WARN, not PASS. """ from datetime import datetime from ..shared.hippius import ( open_manifest_store, publish_receipt, update_receipt_index, ) from ..shared.receipt import dump_receipt, sign_receipt, summarize_receipt try: epoch_start = self._epoch_start_block(manifest) epoch_hash = "" current_block: int | None = None participants: tuple[Participant, ...] = () try: epoch_hash = client.block_hash(epoch_start) participants = participants_from_commitments( client.poll_commitments(include_history=True), cutoff_block=epoch_start, floor_block=self.cfg.round.commit_floor_block, ) # Anchor for the dashboard's next-round countdown (best-effort; # the client extrapolates block→wall-clock from this + as_of). current_block = int(client.current_block()) except Exception as e: # noqa: BLE001 — chain context is best-effort log.warning("receipt chain context unavailable for round=%s: %s", manifest.round_id, e) provenance = ("", "") prov_fn = getattr(window_source, "provenance_for_round", None) if prov_fn is not None: try: # Same epoch block that selected the round's windows, so the # recorded provenance is the pool actually scored. provenance = tuple(prov_fn(base_seed, block=epoch_start)) except Exception as e: # noqa: BLE001 log.warning("pool provenance unavailable for round=%s: %s", manifest.round_id, e) wallet = getattr(client, "wallet", lambda: None)() hotkey_ss58 = str(getattr(getattr(wallet, "hotkey", None), "ss58_address", "") or "") receipt = self.build_round_receipt( manifest, base_seed=base_seed, epoch_start_block=epoch_start, epoch_block_hash=str(epoch_hash), outcome=outcome, windows=windows, reject_reason=reject_reason, participants=participants, pool_provenance=(provenance[0], provenance[1]), reward_uids=reward_uids, weights=weights, validator_hotkey=hotkey_ss58, ) if wallet is not None: receipt = sign_receipt(receipt, wallet) else: log.warning("publishing an UNSIGNED receipt (no wallet) for round=%s", manifest.round_id) store = open_manifest_store(self.cfg.storage) key = publish_receipt(store, dump_receipt(receipt), manifest.round_id, validator_hotkey=hotkey_ss58) log.info("published %s receipt round=%s signed=%s → s3://%s/%s", receipt.status, manifest.round_id, receipt.signature is not None, self.cfg.storage.manifest_bucket, key) # Refresh the dashboard-facing rolling index (best-effort, and inside # the outer guard: a listing convenience must never disturb a round). try: now_iso = datetime.now(UTC).isoformat(timespec="seconds") # Schedule anchor for the "time until next round" countdown. The # next round begins at the next epoch boundary; the client turns # blocks into wall-clock via block_time_s, extrapolating current_block # from `as_of`. Bittensor blocks are ~12s regardless of epoch_blocks # (testnet shortens epochs, not block time), so it is a constant. chain = { "as_of": now_iso, "current_block": current_block, "epoch_start_block": epoch_start, "epoch_blocks": int(self.cfg.round.epoch_blocks), "block_time_s": 12.0, } update_receipt_index( store, summarize_receipt(receipt), updated_at=now_iso, subnet={"netuid": self.cfg.subnet.netuid, "name": self.cfg.subnet.name}, chain=chain, ) except Exception as e: # noqa: BLE001 log.warning("receipt index update failed for round=%s: %s", manifest.round_id, e) except Exception as e: # noqa: BLE001 — receipts must never disturb the round log.warning("receipt publication failed for round=%s: %s", manifest.round_id, e) # ── live loop ──────────────────────────────────────────────────────────── def _publish_chain_status(self, client: object, store: object) -> None: # pragma: no cover """Publish the dashboard's live ``status/chain.json`` (current block, epoch grid, stage windows, revealed submissions) on the poll cadence. Purely presentational and best-effort — it feeds the web dashboard's round-stage strip and live submissions panel between receipts, and any failure (chain flake, storage outage) is swallowed: status telemetry must never disturb a round. """ from datetime import datetime from ..shared.chain_status import build_chain_status, publish_chain_status try: status = build_chain_status( self.cfg, current_block=int(client.current_block()), # type: ignore[attr-defined] commitments=client.poll_commitments(), # type: ignore[attr-defined] network=str(getattr(client, "network", "")), as_of=datetime.now(UTC).isoformat(timespec="seconds"), ) publish_chain_status(store, status) except Exception as e: # noqa: BLE001 — telemetry only log.debug("chain status publish skipped: %s", e) def run_forever(self, client: object, *, window_source: object) -> None: # pragma: no cover """Poll the manifest bucket → evaluate → set weights, once per round. ``window_source`` is a :class:`cascade.validator.windows.WindowSource` (the loaded private pool). Each new manifest's ``round_id`` is the base seed; the same seed drives the rotating window slice so every validator scores the identical set. """ import time from ..shared.hippius import StorageError, open_manifest_store, read_latest_manifest from ..shared.manifest import load_manifest store = open_manifest_store(self.cfg.storage) poll = self.cfg.manifest.poll_seconds # Dedup on CONTENT, not round_id: a re-published manifest for an # already-seen round id (same-round-id rerun, e.g. after a contract # fix) must be re-judged, not silently skipped (2026-07-15: the # round_id-only latch ignored the rerun manifest with no log line). last_round: str | None = None last_digest: str | None = None while True: try: # Live dashboard telemetry first, every poll: between receipts # this is the page's only fresh view of the chain (stage strip # + live submissions). Best-effort; never affects the round. self._publish_chain_status(client, store) raw = read_latest_manifest(store) digest = hashlib.sha256(raw.encode("utf-8")).hexdigest() if digest == last_digest: log.debug("manifest unchanged (round=%s sha=%s…); skipping", last_round, digest[:12]) else: manifest = load_manifest(raw) base_seed = int(manifest.round_id) if manifest.round_id == last_round: log.warning( "manifest for already-handled round=%s RE-PUBLISHED " "with different content (sha %s… -> %s…); re-judging", manifest.round_id, (last_digest or "")[:12], digest[:12], ) log.info( "new manifest round=%s entries=%d (%s); gating + scoring …", manifest.round_id, len(manifest.entries), ",".join(f"{e.role}:uid{e.miner_uid}" for e in manifest.entries), ) # Gate first so a rejected manifest never moves weights. reason = self.check_manifest(manifest) retry_pin = False if reason is None: # Pool-pin gate: the signed snapshot pin must match this # validator's own deterministic selection for the round. try: reason = self.check_pool_pin( manifest, window_source, block=self._epoch_start_block(manifest), ) except StorageError as e: # The index could not be READ (auth/network/5xx) — # a transient, not a verdict. Within the grace # window: no latch, no receipt, retry next poll. reason = self._pool_pin_read_failed(manifest.round_id, e) retry_pin = reason is None else: self._pin_read_first_failure.pop(str(manifest.round_id), None) if retry_pin: # In-grace read failure: neither last_round nor # last_digest move, so the next manifest poll re-judges # this round from scratch. Falls through to the weight # re-assert + sleep below. pass elif reason is not None: log.warning("rejecting manifest round=%s: %s", manifest.round_id, reason) last_round, last_digest = manifest.round_id, digest # A rejected round still gets a public receipt carrying # the gate's reason — visible, not silently absent. self._publish_round_receipt( client, manifest, base_seed, reject_reason=reason, window_source=window_source, ) elif not self.king_synced(manifest): # The trainer trained the OLD king (incentive lags a # dethrone). Hold the KOTH state and keep voting the champion # so incentive migrates and the trainer re-syncs — bounded by # the safety valve (see _resync_step). A public receipt # records why, not a silent skip. last_round, last_digest = manifest.round_id, digest self.state, reject_reason = self._resync_step(manifest) self._persist_state() reward_uids = self._reward_uids(manifest, None, client) weights_vec = self._apply_weights(client, manifest.round_id, reward_uids) self._publish_round_receipt( client, manifest, base_seed, reject_reason=reject_reason, window_source=window_source, reward_uids=tuple(reward_uids), weights=weights_vec, ) else: # The epoch block selects the daily snapshot; base_seed # rotates the window slice within it. windows = window_source.windows_for_round( base_seed, self.cfg.eval.n_windows, block=self._epoch_start_block(manifest), ) # process_round mutates the sticky KOTH state atomically (it # raises before any mutation on a transient eval/fetch error, # leaving state untouched for a clean retry). Mark the round # consumed as soon as it returns, so a later weight-set failure # can NEVER re-run it and double-count the streak/tenure. outcome = self.process_round(manifest, windows, base_seed) # Back in sync — clear any accumulated resync holds so a # future desync starts the safety-valve count from zero. if self.state.resync_holds or self.state.last_resync_round_id: self.state = replace( self.state, resync_holds=0, last_resync_round_id=None) last_round, last_digest = manifest.round_id, digest self._persist_state() reward_uids = self._reward_uids(manifest, outcome, client) weights_vec = self._apply_weights(client, manifest.round_id, reward_uids) # The public receipt — strictly after weights, so it # records what was actually set (empty vector = the # weight extrinsic failed this round). if outcome is not None: self._publish_round_receipt( client, manifest, base_seed, outcome=outcome, windows=windows, window_source=window_source, reward_uids=tuple(reward_uids), weights=weights_vec, ) else: # Gated in but nothing to score (no king/challenger # pair at any size): a public record still exists. self._publish_round_receipt( client, manifest, base_seed, reject_reason="no_king_challenger_pair", window_source=window_source, reward_uids=tuple(reward_uids), weights=weights_vec, ) # Log-only public benchmarks for a freshly crowned king. # Strictly after weights are decided; never affects them. self._maybe_run_benchmarks(manifest, outcome) # Cascade step — strictly last, so this round's weights and # receipt already recorded the outgoing king. Resets the # reign clock on a dethrone, records the king's checkpoint, # and fires the promotion when the clock is ripe. self._cascade_round(manifest, outcome) except Exception as e: # noqa: BLE001 — a service loop must not die on one round log.exception("round processing failed; retrying after poll: %s", e) try: self._maybe_reassert_weights(client) except Exception as e: # noqa: BLE001 log.warning("weight re-assert check failed: %s", e) time.sleep(poll) def _maybe_reassert_weights(self, client: object) -> None: """Re-push the standing weight vector every ``weight_set_interval_blocks``. Subtensor treats a validator whose ``last_update`` is older than the subnet's ``activity_cutoff`` (5000 blocks ≈ 16.7 h on netuid 91) as inactive in Yuma consensus. A mainnet round is 7200 blocks, so voting only when a manifest lands blows through the cutoff every round — and a stalled trainer silences the validator entirely. The vector needs no manifest: it is recomputed from the persisted champion state (king + registered prior kings; no champion burns to ``burn_uid``), so this is a pure freshness signal that can never move the throne. The interval must stay ≥ the subnet's ``weights_rate_limit`` (100 blocks on netuid 91) or the chain silently no-ops the extrinsic; ≤ 0 disables. """ interval = int(self.cfg.validator.weight_set_interval_blocks) if interval <= 0: return cur = int(client.current_block()) # type: ignore[attr-defined] if self._last_weight_block is not None and cur - self._last_weight_block < interval: return uids: list[int] = [] if self.state.king_hotkey is not None: uid = client.uid_for_hotkey(self.state.king_hotkey) # type: ignore[attr-defined] if uid is not None: uids.append(uid) for hk in self.state.former_kings: uid = client.uid_for_hotkey(hk) # type: ignore[attr-defined] if uid is not None and uid not in uids: uids.append(uid) log.info("re-asserting weights (last set %s, block %d): reward_uids=%s", "never" if self._last_weight_block is None else f"{cur - self._last_weight_block} blocks ago", cur, uids or [self.cfg.scoring.burn_uid]) self._apply_weights(client, "weight-reassert", uids) # Stamp even when the extrinsic failed: the next attempt comes after # one interval (~27 more before the cutoff), not every poll tick. self._last_weight_block = cur @staticmethod def _manifest_king_hotkey(manifest: TrainingManifest) -> str | None: e = manifest.entry_for_role("king") return e.miner_hotkey if e is not None else None @staticmethod def _manifest_king_uid(manifest: TrainingManifest) -> int | None: e = manifest.entry_for_role("king") return e.miner_uid if e is not None else None def king_synced(self, manifest: TrainingManifest) -> bool: """Whether the round's *trained* king matches the validator's champion. The trainer picks the king it trains from on-chain incentive, which lags the validator's dethrone verdicts. Until the champion the validator crowned actually becomes the highest-incentive UID — and so the king the trainer trains — the two disagree, and a round trained against the *old* king must not have its verdict applied to the *new* champion. Synced when the champion is unset (bootstrap) or the trained king is the champion. """ if self.state.king_hotkey is None: return True return self._manifest_king_hotkey(manifest) == self.state.king_hotkey def _resync_step(self, manifest: TrainingManifest) -> tuple[ChampionState, str]: """Next champion state + receipt reason for a king-resync round. Called when the trained king != champion (``king_synced`` is False). By default it holds the throne, bumping the consecutive-hold counter, and the caller keeps voting the champion so incentive migrates and the trainer re-syncs. SAFETY VALVE: once the champion has stayed un-synced for ``scoring.king_resync_max_rounds`` consecutive rounds it can never be the king the trainer trains (e.g. it has no usable commitment), so holding forever would wedge the subnet — the valve abandons it and adopts the trainer's trained king (:func:`state.demote_to_trained`), and normal scoring resumes next round. ``king_resync_max_rounds <= 0`` disables the valve (hold indefinitely). The counter advances once per DISTINCT un-synced round: a restart re-gates the same stale manifest, and counting those re-gates let five restarts during a pause trip the valve and demote a healthy champion (2026-07-22). Pure: returns the new state; the caller persists, votes, and publishes. """ champ = self.state.king_hotkey trained = self._manifest_king_hotkey(manifest) round_id = str(manifest.round_id) same_round = self.state.last_resync_round_id == round_id holds = self.state.resync_holds if same_round else self.state.resync_holds + 1 cap = self.cfg.scoring.king_resync_max_rounds if 0 < cap <= holds and trained is not None: log.warning( "round=%s king_resync SAFETY VALVE: champion %s un-synced %d rounds " "(cap=%d) — demoting to trained king %s, resuming normal scoring next round", manifest.round_id, (champ or "?")[:12], holds, cap, (trained or "?")[:12], ) return ( state_mod.demote_to_trained( self.state, trained_hotkey=trained, trained_uid=self._manifest_king_uid(manifest), ), f"king_resync_demoted: champion {champ} un-synced {holds} rounds " f"(cap={cap}); adopted trained king {trained}", ) log.warning( "round=%s trainer king %s != champion %s; voting champion to re-sync " "incentive, KOTH state held (%d/%s)", manifest.round_id, (trained or "?")[:12], (champ or "?")[:12], holds, cap if cap > 0 else "∞", ) return ( replace(self.state, resync_holds=holds, last_resync_round_id=round_id), f"king_resyncing: champion {champ} != trained king {trained}", ) def _king_uid_to_vote(self, manifest: TrainingManifest, *, client: object | None = None) -> int | None: """The UID to put the king's weight on this round. The **validator's champion state** is the authority on who holds the throne, so vote *that* king every round — not the (lagging) king the trainer happened to train. This is what makes a dethrone STICK: the new champion keeps the weight, incentive migrates to it, and next round the trainer trains it as king (they re-sync). Voting the trained/manifest king instead — the old behaviour — reverted a dethrone the moment the trainer lagged one round, orphaning the champion. The champion hotkey is resolved to its current UID via the metagraph (robust to re-registration); the manifest king is used only to bootstrap when there is no champion yet. """ if self.state.king_hotkey is not None: if client is not None: resolved = client.uid_for_hotkey(self.state.king_hotkey) # type: ignore[attr-defined] if resolved is not None: return resolved return self.state.king_uid king_entry = manifest.entry_for_role("king") return king_entry.miner_uid if king_entry is not None else None def _reward_uids( self, manifest: TrainingManifest, outcome: RoundOutcome | None, client: object ) -> list[int]: """UIDs that share this round's weight: the current king plus any ``former_kings`` still registered (teutonic-style equal-share payout). Returns an empty list when there is no king to vote for at all (no champion and no manifest king); the loop hands that to ``set_equal_share_weights``, which burns to ``burn_uid`` rather than reverting. The list is otherwise deduped/range-checked there too. ``[validator] force_burn`` empties the list HERE — not just at the weight push — so the published receipt's ``reward_uids`` agree with the burn vector actually set (``cascade-audit`` recomputes one from the other and fails on a mismatch). """ if self.cfg.validator.force_burn: return [] uids: list[int] = [] king_uid = self._king_uid_to_vote(manifest, client=client) if king_uid is not None: uids.append(king_uid) for hk in self.state.former_kings: uid = client.uid_for_hotkey(hk) # type: ignore[attr-defined] if uid is not None: uids.append(uid) return uids def _apply_weights(self, client: object, round_id: str, reward_uids: list[int]) -> tuple[float, ...]: """Set the equal-share weight vector on chain; return it (empty on failure). Shared by the scored path and the king-resync path. Always sets weights — an empty ``reward_uids`` burns to ``burn_uid`` so emission still leaves the network. A failed extrinsic is logged and retried next round (the empty vector is recorded truthfully in the receipt). ``[validator] force_burn`` overrides the vector to a burn HERE — the single choke point every push flows through (scored, resync, re-assert) — so the receipt records the burn that was actually set. Champion state is never touched by this override.""" from ..shared.chain import decayed_share_vector if self.cfg.validator.force_burn: log.warning( "FORCE-BURN active (round=%s): %sburning to uid %d (champion state " "untouched — unset [validator] force_burn and restart to resume voting)", round_id, f"dropping reward_uids={reward_uids}; " if reward_uids else "", self.cfg.scoring.burn_uid, ) reward_uids = [] decay = self.cfg.scoring.king_decay try: n_uids = client.n_uids() # type: ignore[attr-defined] client.set_equal_share_weights( # type: ignore[attr-defined] reward_uids, n_uids, decay=decay, burn_uid=self.cfg.scoring.burn_uid, ) vec = tuple(decayed_share_vector( reward_uids, n_uids, decay=decay, burn_uid=self.cfg.scoring.burn_uid)) log.info("round=%s weights set: reward_uids=%s (n_uids=%d, burn_uid=%d)", round_id, reward_uids or [self.cfg.scoring.burn_uid], n_uids, self.cfg.scoring.burn_uid) # Reset the re-assert timer (fake test clients lack current_block). with contextlib.suppress(Exception): self._last_weight_block = int(client.current_block()) # type: ignore[attr-defined] return vec except Exception as e: # noqa: BLE001 — retried next round log.warning("weight set failed for round=%s (king holds, retried next round): %s", round_id, e) return () def _persist_state(self) -> None: # pragma: no cover from . import state as state_mod try: Path(self.cfg.validator.state_db_path).write_text( state_mod.dumps(self.state), encoding="utf-8" ) except Exception as e: # noqa: BLE001 log.warning("failed to persist validator state: %s", e) def _load_state(path: str) -> ChampionState: """Load persisted champion state from ``state_db_path`` (JSON), or a fresh state if the file is absent/unreadable.""" p = Path(path) if not p.is_file(): return ChampionState() try: return state_mod.loads(p.read_text(encoding="utf-8")) except Exception as e: # noqa: BLE001 log.warning("could not load validator state from %s (%s); starting fresh", path, e) return ChampionState() def _bootstrap_state_from_receipts(store: object, anchor: str) -> ChampionState | None: """Champion inherited from the signed public receipt trail, or ``None``. First-boot inheritance for a validator with no local state: the throne otherwise lives only in each validator's private state DB, so a validator joining mid-reign would judge the next manifest blind (``king_synced`` treats an unset champion as synced) and crown whichever king it happened to see win first — a different champion than every validator that witnessed the real dethrone (OPSLOG 2026-07-17). ``anchor`` is the pinned receipt-signing ss58 (``[manifest] validator_hotkey``, falling back to ``trainer_hotkey``) — the same trust anchor the validator already applies to manifests, extended once, at first boot, to the receipt trail. Reads the anchor's ``receipts//latest.json`` (legacy shared pointer as fallback) and adopts the throne recorded by a *scored* receipt whose signature verifies. When ``latest.json`` is a hold/rejected receipt (verdict-less by construction), the receipt *index* is consulted — but only as an UNTRUSTED pointer to candidate round ids: nothing is adopted except from a per-round receipt whose signature verifies against the anchor. Anything short of that — missing objects, unreadable JSON, a bad signature, a genesis throne (``king_hotkey`` unset) — returns ``None`` and the caller proceeds with the stock blank-slate behaviour. Storage faults must never block validator startup. """ if not anchor: return None from ..shared.hippius import ( RECEIPT_INDEX_KEY, RECEIPT_LATEST_KEY, receipt_latest_key, receipt_round_key, ) from ..shared.receipt import load_receipt, verify_receipt_signature def _adopt_from(key: str) -> ChampionState | None: try: text = store.get_text(key) except Exception: # noqa: BLE001 — absent/unreachable key ⇒ next candidate return None try: receipt = load_receipt(text) except Exception as e: # noqa: BLE001 log.warning("receipt bootstrap: unreadable receipt at %s (%s); skipped", key, e) return None if not verify_receipt_signature(receipt, anchor): log.warning("receipt bootstrap: receipt at %s is not signed by the pinned " "hotkey %s…; skipped", key, anchor[:8]) return None v = receipt.verdict if receipt.status != "scored" or v is None or not v.king_hotkey or v.king_uid is None: return None log.info("receipt bootstrap: adopting champion %s (uid %d) from signed scored " "receipt round=%s", v.king_hotkey, int(v.king_uid), receipt.round_id) return ChampionState(king_hotkey=str(v.king_hotkey), king_uid=int(v.king_uid)) adopted = _adopt_from(receipt_latest_key(anchor)) or _adopt_from(RECEIPT_LATEST_KEY) if adopted is not None: return adopted try: rows = json.loads(store.get_text(RECEIPT_INDEX_KEY)).get("rounds", []) except Exception: # noqa: BLE001 — no index ⇒ nothing more to try rows = [] # Index rows are chronological (oldest first, capped at most-recent); # walk newest-first and adopt the first scored round that verifies. for row in reversed(rows): if str(row.get("status")) != "scored": continue adopted = _adopt_from(receipt_round_key(str(row.get("round_id", "")), anchor)) if adopted is not None: return adopted return None def _warm_start_installer(path: Path) -> Callable[[object], None]: """The default Cascade installer: promote the winning checkpoint by writing its pointer (and its eval numbers) to ``warm_start_init_path`` — the seam the trainer reads to warm-start every subsequent round from. Promotes AS-IS; no retrain/fine-tune.""" def _install(winner: object) -> None: # pragma: no cover — file glue import time path.parent.mkdir(parents=True, exist_ok=True) path.write_text( json.dumps( { "checkpoint_id": getattr(winner, "checkpoint_id", None), "size": getattr(winner, "size", ""), "score": getattr(winner, "score", None), "gifteval_crps": getattr(winner, "gifteval_crps", None), "gifteval_mase": getattr(winner, "gifteval_mase", None), "boom_crps": getattr(winner, "boom_crps", None), "boom_mase": getattr(winner, "boom_mase", None), "time_crps": getattr(winner, "time_crps", None), "time_mase": getattr(winner, "time_mase", None), "installed_at": time.time(), }, sort_keys=True, ), encoding="utf-8", ) log.info("cascade: warm-start init written to %s (checkpoint %s)", path, getattr(winner, "checkpoint_id", "?")) return _install def _build_cascade(cfg: ChainConfig) -> CascadeController: """Construct the Cascade controller from config, restoring the persisted reign clock + checkpoint log so it resumes across restarts.""" from .cascade import CascadeController, load_state state_path = Path(cfg.validator.cascade_state_db_path) return CascadeController( reign_days=cfg.scoring.cascade_reign_days, state=load_state(state_path), install_fn=_warm_start_installer(Path(cfg.validator.warm_start_init_path)), state_path=state_path, ) def build_runner( *, chain_toml: Path | None = None, cache_dir: Path | None = None, device: str = "cpu", eval_host_fn: Callable[[], RemoteHost | None] | None = None, ) -> ValidatorRunner: """Construct a runner from ``chain.toml``, restoring persisted champion state. Wallet/chain wiring for live weight-setting is attached by ``cascade-validator`` (see main.py). ``eval_host_fn`` (optional) resolves the GPU pod to offload heavy evals to — re-invoked per eval, so an elastic provisioner-rented pod is picked up lazily; the wallet stays on this box.""" from ..shared.config import load_chain_config cfg = load_chain_config(chain_toml) state = _load_state(cfg.validator.state_db_path) if (cfg.validator.bootstrap_from_receipts and state.king_hotkey is None and state.rounds_seen == 0 and not state.former_kings): # Truly fresh validator (no champion, no history): inherit the throne # from the signed receipt trail before the first manifest is judged. from ..shared.hippius import open_manifest_store anchor = cfg.manifest.validator_hotkey or cfg.manifest.trainer_hotkey try: adopted = _bootstrap_state_from_receipts( open_manifest_store(cfg.storage), anchor) except Exception as e: # noqa: BLE001 — storage must never block startup log.warning("receipt bootstrap skipped (%s); starting blank", e) adopted = None if adopted is not None: state = adopted # Cascade is opt-in ([scoring] cascade_enabled); off ⇒ no controller is wired # and the runner is pure KOTH. cascade = _build_cascade(cfg) if cfg.scoring.cascade_enabled else None return ValidatorRunner( cfg=cfg, state=state, cache_dir=cache_dir, device=device, cascade=cascade, eval_host_fn=eval_host_fn, )