diff --git a/connito/test/test_cycle_consistent_telemetry.py b/connito/test/test_cycle_consistent_telemetry.py index e563dc1..6fd4c4c 100644 --- a/connito/test/test_cycle_consistent_telemetry.py +++ b/connito/test/test_cycle_consistent_telemetry.py @@ -155,6 +155,78 @@ def test_commit_map_from_checkpoints_skips_incomplete(): assert m == {1: ("a/r", "v1")} +# --------------------------------------------------------------------------- +# Journal v3: roster_size / lifecycle_step for round-gauge recovery +# --------------------------------------------------------------------------- + +def test_journal_v3_roundtrips_roster_and_step(tmp_path): + j = RJ.RoundJournal( + round_id=555, + scored_uids=(1, 2, 3), + failed_uids=(4,), + roster_size=165, + lifecycle_step=3, + finalized=True, + ) + p = tmp_path / "round_555.json" + RJ.write_atomic(p, j) + loaded = RJ.load(p) + assert loaded is not None + assert loaded.schema_version == 3 + assert loaded.roster_size == 165 + assert loaded.lifecycle_step == 3 + + +def test_journal_v2_loads_with_zero_roster(tmp_path): + # A v2 file (no roster_size/lifecycle_step) must still load, defaulting + # both to 0 so recovery degrades gracefully rather than crashing. + v2 = { + "round_id": 42, "schema_version": 2, + "scored_uids": [1, 2], "failed_uids": [3], + "uid_to_commit": {}, "finalized": False, + } + p = tmp_path / "round_42.json" + p.write_text(json.dumps(v2), encoding="utf-8") + loaded = RJ.load(p) + assert loaded is not None + assert loaded.roster_size == 0 + assert loaded.lifecycle_step == 0 + + +def test_recovery_round_carries_v3_fields(tmp_path): + j = RJ.RoundJournal(round_id=777, roster_size=100, lifecycle_step=2) + stub = RJ._RecoveryRound.from_journal(j, tmp_path / "round_777.json") + assert stub.roster_size == 100 + assert stub.lifecycle_step == 2 + + +def test_finalize_preserves_roster_in_journal_rewrite(tmp_path): + # finalize re-writes the journal with finalized=True; the v3 fields must + # survive that round-trip (they feed the round-gauge restore next boot). + rid = 8_888 + jpath = tmp_path / f"round_{rid}.json" + RJ.write_atomic( + jpath, + RJ.RoundJournal( + round_id=rid, + uid_to_hotkey={9201: "hk", 9202: "hk2"}, + scores={9201: 2.5}, + scored_uids=(9201,), + failed_uids=(9202,), + roster_size=165, + lifecycle_step=3, + ), + ) + stub = RJ._RecoveryRound.from_journal(RJ.load(jpath), jpath) + finalize_round_scores( + round_obj=stub, score_aggregator=_FakeAggregator(), score_path=None, + ) + reloaded = RJ.load(jpath) + assert reloaded.finalized is True + assert reloaded.roster_size == 165 + assert reloaded.lifecycle_step == 3 + + # --------------------------------------------------------------------------- # finalize_round_scores emission (live-shaped round + recovery path) # --------------------------------------------------------------------------- diff --git a/connito/test/test_journal_telemetry_republish.py b/connito/test/test_journal_telemetry_republish.py new file mode 100644 index 0000000..5c87b25 --- /dev/null +++ b/connito/test/test_journal_telemetry_republish.py @@ -0,0 +1,221 @@ +"""Tests for the metrics-only telemetry republish on startup recovery. + +Startup recovery restores telemetry by *replaying* `finalize_round_scores`, +which marks the journal finalized — and the replay loop skips finalized +journals. So recovery only ever restored telemetry ONCE: a second restart +found nothing to replay and emitted nothing, leaving the dashboard blank +for the entire last completed cycle (observed 2026-07-31, two Watchtower +restarts 25 minutes apart). + +`republish_telemetry_from_journal` closes that. It must be metrics-only: +re-running finalize would keep the aggregator's point *set* correct +(`drop_round` runs first) but re-stamp those points with fresh timestamps, +reshuffling the "last N by timestamp" rolling average that drives weight +submission. + +The prometheus registry is process-global, so these tests use uids in a +dedicated 94xx range and round ids in a 94xxxx range. +""" +from __future__ import annotations + +import json + +from connito.shared import telemetry as T +from connito.validator import round_journal as RJ + + +def _sample(gauge, family: str, **labels) -> float | None: + for metric in gauge.collect(): + for s in metric.samples: + if s.name == family and all(s.labels.get(k) == v for k, v in labels.items()): + return s.value + return None + + +def _journal(round_id: int, **overrides) -> RJ.RoundJournal: + kwargs = dict( + round_id=round_id, + uid_to_hotkey={9401: "hk1", 9402: "hk2", 9403: "hk3", 9404: "hk4"}, + scores={9401: 2.5, 9402: 1.5, 9403: 0.0}, + scored_uids=(9401, 9402, 9403), + failed_uids=(9404,), # operational failure — no verdict + validation_failed_uids=(), + freeze_zero_uids=(9405,), + freeze_zero_hotkeys={9405: "hk5"}, + uid_to_commit={9401: ("miner/one", "aaa1")}, + uid_to_val_loss={9401: 1.25, 9402: 1.75}, + roster_size=10, + lifecycle_step=3, + finalized=True, + ) + kwargs.update(overrides) + return RJ.RoundJournal(**kwargs) + + +class _RecordingAggregator: + """Read-only probe: records every mutating call so a test can assert + the republish pass made none.""" + + def __init__(self): + self.mutations: list[str] = [] + + # --- read side (what republish is allowed to use) --- + def uid_score_pairs(self, how="avg"): + return {9401: 2.25 if how == "latest" else 1.1, 9402: 1.5} + + def record_count(self, uid): + return 4 + + # --- write side (must never be called) --- + def add_score(self, **kwargs): + self.mutations.append("add_score") + + def drop_round(self, round_id): + self.mutations.append("drop_round") + + def persist_atomic(self, path): + self.mutations.append("persist_atomic") + + def prune_before_round(self, min_round_id): + self.mutations.append("prune_before_round") + + +# --------------------------------------------------------------------------- +# The guard the backend specifically asked for +# --------------------------------------------------------------------------- + +def test_republish_does_not_mutate_the_aggregator(): + agg = _RecordingAggregator() + RJ.republish_telemetry_from_journal(_journal(940_001), score_aggregator=agg) + assert agg.mutations == [] + + +def test_republish_works_without_an_aggregator(): + # Snapshot gauges are skipped, everything else still emits. + n = RJ.republish_telemetry_from_journal(_journal(940_002), score_aggregator=None) + assert n == 4 # 3 scored + 1 freeze_zero (the failed uid gets no verdict) + + +# --------------------------------------------------------------------------- +# What gets emitted +# --------------------------------------------------------------------------- + +def test_republish_emits_last_scored_round_for_verdict_uids(): + rid = 940_010 + RJ.republish_telemetry_from_journal(_journal(rid), score_aggregator=_RecordingAggregator()) + for uid in (9401, 9402, 9403, 9405): + assert _sample( + T.VALIDATOR_MINER_LAST_SCORED_ROUND_ID, + "validator_miner_last_scored_round_id", + miner_uid=str(uid), + ) == float(rid) + + +def test_republish_excludes_operational_failures(): + """`failed_uids` minus validation failures get NO aggregator entry at + finalize, so telemetry must not imply they were judged.""" + j = _journal(940_011, uid_to_hotkey={9410: "hk"}, scored_uids=(), scores={}, + failed_uids=(9410,), freeze_zero_uids=(), freeze_zero_hotkeys={}, + uid_to_commit={}, uid_to_val_loss={}) + assert RJ.verdict_uids(j) == set() + assert RJ.republish_telemetry_from_journal(j) == 0 + + +def test_republish_emits_val_loss(): + rid = 940_020 + RJ.republish_telemetry_from_journal(_journal(rid)) + assert _sample( + T.VALIDATOR_MINER_VAL_LOSS, "validator_miner_val_loss", miner_uid="9401" + ) == 1.25 + assert _sample( + T.VALIDATOR_MINER_VAL_LOSS, "validator_miner_val_loss", miner_uid="9402" + ) == 1.75 + + +def test_republish_emits_round_counters_and_lifecycle(): + rid = 940_030 + RJ.republish_telemetry_from_journal(_journal(rid)) + label = {"round_id": str(rid)} + assert _sample(T.VALIDATOR_ROUND_MINERS_SCORED, "validator_round_miners_scored", **label) == 3.0 + assert _sample(T.VALIDATOR_ROUND_MINERS_FAILED, "validator_round_miners_failed", **label) == 1.0 + # roster 10 - 3 scored - 1 failed + assert _sample(T.VALIDATOR_ROUND_MINERS_PENDING, "validator_round_miners_pending", **label) == 6.0 + assert _sample( + T.VALIDATOR_ROUND_LIFECYCLE_STEP, "validator_round_lifecycle_step", **label + ) == 3.0 + + +def test_republish_emits_round_delta_and_commit(): + rid = 940_040 + RJ.republish_telemetry_from_journal(_journal(rid)) + assert _sample( + T.VALIDATOR_MINER_ROUND_DELTA, "validator_miner_round_delta", miner_uid="9401" + ) == 2.5 + assert _sample( + T.VALIDATOR_MINER_EVALUATED_COMMIT_INFO, + "validator_miner_evaluated_commit_info", + miner_uid="9401", hf_repo_id="miner/one", hf_revision="aaa1", + ) == float(rid) + + +def test_republish_pre_v3_journal_clamps_pending_and_skips_val_loss(): + """A v2 journal has no roster_size and no losses — republish must not + invent a denominator or crash.""" + rid = 940_050 + j = _journal(rid, roster_size=0, lifecycle_step=0, uid_to_val_loss={}) + assert RJ.republish_telemetry_from_journal(j) == 4 + assert _sample( + T.VALIDATOR_ROUND_MINERS_PENDING, "validator_round_miners_pending", + round_id=str(rid), + ) == 0.0 + + +def test_republish_is_idempotent(): + rid = 940_060 + a = RJ.republish_telemetry_from_journal(_journal(rid)) + b = RJ.republish_telemetry_from_journal(_journal(rid)) + assert a == b + assert _sample( + T.VALIDATOR_ROUND_MINERS_SCORED, "validator_round_miners_scored", + round_id=str(rid), + ) == 3.0 + + +def test_republish_registers_round_for_eviction(): + rid = 940_070 + RJ.republish_telemetry_from_journal(_journal(rid)) + assert rid in T._EMITTED_ROUND_IDS + + +def test_republish_never_raises_on_malformed_journal(): + class Broken: + round_id = "not-an-int" + assert RJ.republish_telemetry_from_journal(Broken()) == 0 # type: ignore[arg-type] + + +# --------------------------------------------------------------------------- +# val_loss journalling (Gap 1) +# --------------------------------------------------------------------------- + +def test_journal_v3_roundtrips_val_loss(tmp_path): + j = _journal(940_080) + p = tmp_path / "round.json" + RJ.write_atomic(p, j) + loaded = RJ.load(p) + assert loaded is not None + assert loaded.uid_to_val_loss == {9401: 1.25, 9402: 1.75} + + +def test_journal_v2_file_loads_with_empty_val_loss(tmp_path): + v2 = {"round_id": 42, "schema_version": 2, "scored_uids": [1], "finalized": True} + p = tmp_path / "round_42.json" + p.write_text(json.dumps(v2), encoding="utf-8") + loaded = RJ.load(p) + assert loaded is not None + assert loaded.uid_to_val_loss == {} + assert loaded.roster_size == 0 + + +def test_recovery_round_carries_val_losses(tmp_path): + stub = RJ._RecoveryRound.from_journal(_journal(940_090), tmp_path / "r.json") + assert stub.val_losses == {9401: 1.25, 9402: 1.75} diff --git a/connito/validator/background_eval_worker.py b/connito/validator/background_eval_worker.py index d198a22..a6f9704 100644 --- a/connito/validator/background_eval_worker.py +++ b/connito/validator/background_eval_worker.py @@ -518,7 +518,7 @@ def _run_eval(): self._prune_non_top(round_obj) return - round_obj.mark_scored(uid, evaluated.score) + round_obj.mark_scored(uid, evaluated.score, val_loss=evaluated.val_loss) logger.info( "bg-eval: success", round_id=round_obj.round_id, diff --git a/connito/validator/evaluator.py b/connito/validator/evaluator.py index 9123264..7201955 100644 --- a/connito/validator/evaluator.py +++ b/connito/validator/evaluator.py @@ -347,6 +347,19 @@ def finalize_round_scores( uid_to_commit=_rj.commit_map_from_checkpoints( getattr(round_obj, "uid_to_chain_checkpoint", None) or {} ), + # v3 round-level gauge inputs. A live Round exposes + # foreground/background uids; the recovery stub carries + # `roster_size` forward from the journal it was hydrated + # from — so a re-finalize preserves them. + roster_size=int( + getattr(round_obj, "roster_size", 0) + or ( + len(getattr(round_obj, "foreground_uids", ())) + + len(getattr(round_obj, "background_uids", ())) + ) + ), + lifecycle_step=int(getattr(round_obj, "lifecycle_step", 0)), + uid_to_val_loss=dict(getattr(round_obj, "val_losses", None) or {}), finalized=True, ), ) @@ -628,6 +641,14 @@ class MinerEvalJob: model_path: str step: int score: float = 0.0 + # Raw evaluation loss for this miner this round. Carried alongside the + # delta-based `score` so the caller can journal it: `val_loss` is + # published to Prometheus at eval time and is NOT recoverable from + # `score` alone, because `delta = max(0.0, baseline - val_loss)` clamps + # at zero — every miner scoring 0 (the majority in many rounds) would + # be underivable. Without journaling it, a mid-round restart loses the + # cycle's losses permanently. + val_loss: float | None = None # -------------------------- Pipeline Config ----------------------------------- @@ -876,6 +897,7 @@ def evaluate_one_miner_sync( model_path=str(model_path), step=int(step), score=float(score), + val_loss=float(val_loss), ) except EvalDeadlineExceeded as e: logger.warning( @@ -1193,7 +1215,7 @@ async def evaluate_foreground_round( round_obj=round_obj, ) continue - round_obj.mark_scored(uid, evaluated.score) + round_obj.mark_scored(uid, evaluated.score, val_loss=evaluated.val_loss) round_obj.publish_progress() completed.append(evaluated) _prune_non_top_after_eval( diff --git a/connito/validator/round.py b/connito/validator/round.py index aa16db3..1eb394b 100644 --- a/connito/validator/round.py +++ b/connito/validator/round.py @@ -77,6 +77,11 @@ class Round: # rounds and would let history pull a non-top-this-round miner into # the keep set). scores: dict[int, float] = field(default_factory=dict) + # Per-uid raw evaluation loss, recorded by `mark_scored` alongside the + # score. Journaled so a mid-round restart doesn't lose the cycle's + # losses — `validator_miner_val_loss` is emitted at eval time and is + # not derivable from `scores` (the delta clamps at 0). + val_losses: dict[int, float] = field(default_factory=dict) claimed_uids: set[int] = field(default_factory=set) failed_uids: set[int] = field(default_factory=set) # UIDs the miner is at fault for: explicit validation failures @@ -96,6 +101,11 @@ class Round: freeze_zero_uids: set[int] = field(default_factory=set) freeze_zero_hotkeys: dict[int, str] = field(default_factory=dict) weights_submitted: bool = False + # Last live lifecycle step this round reached (set by run.py alongside + # the VALIDATOR_ROUND_LIFECYCLE_STEP gauge: 0 freeze / 2 post-foreground + # / 3 eval-window). Persisted to the journal so startup recovery can + # restore the round-level gauges that only the live loop writes. + lifecycle_step: int = 0 # Round-group construction scheme (gated by # `config.evaluation.enable_round_group_construction`). All default @@ -588,6 +598,9 @@ def _journal_snapshot_locked(self) -> dict | None: "freeze_zero_uids": tuple(sorted(self.freeze_zero_uids)), "freeze_zero_hotkeys": dict(self.freeze_zero_hotkeys), "uid_to_commit": commit_map_from_checkpoints(self.uid_to_chain_checkpoint), + "uid_to_val_loss": dict(self.val_losses), + "roster_size": len(self.foreground_uids) + len(self.background_uids), + "lifecycle_step": int(self.lifecycle_step), "finalized": False, } @@ -640,12 +653,20 @@ def _record_in_cycle_score(self, uid: int, hotkey: str, score: float) -> None: error=str(e), uid=uid, round_id=self.round_id, ) - def mark_scored(self, uid: int, score: float = 0.0) -> None: + def mark_scored( + self, uid: int, score: float = 0.0, val_loss: float | None = None + ) -> None: """Record a successful evaluation. `score` is this-round's score (e.g. ``delta ** 1.2`` from `evaluate_one_miner`); it is stored in `self.scores` so per-round ranking — used by post-eval submission cleanup — never has to consult the global aggregator. + `val_loss` is the raw evaluation loss. It is recorded purely so the + journal can carry it: `validator_miner_val_loss` is published at + eval time and cannot be reconstructed from `score`, because + ``delta = max(0.0, baseline - val_loss)`` clamps at zero. Optional + so existing callers and test fixtures keep working. + Also writes to the per-round journal and (if configured) the score aggregator with the raw delta tagged with this round_id, so a kill before `finalize_round_scores` runs does not lose the @@ -656,6 +677,8 @@ def mark_scored(self, uid: int, score: float = 0.0) -> None: with self._lock: self.scored_uids.add(uid) self.scores[uid] = score_f + if val_loss is not None: + self.val_losses[uid] = float(val_loss) self.claimed_uids.discard(uid) hotkey = self.uid_to_hotkey.get(uid) self._persist_journal() diff --git a/connito/validator/round_journal.py b/connito/validator/round_journal.py index 41e2250..ec4450b 100644 --- a/connito/validator/round_journal.py +++ b/connito/validator/round_journal.py @@ -27,9 +27,18 @@ # v2 adds `uid_to_commit` (uid -> (hf_repo_id, hf_revision)) so the # journal-recovery finalize can re-publish evaluated-commit telemetry. -# `from_json` accepts v1 files (missing map -> empty) so leftover journals -# written by an older build still recover. -SCHEMA_VERSION = 2 +# v3 adds `roster_size` + `lifecycle_step` so recovery can also restore the +# round-level gauges (validator_round_miners_{scored,pending,failed}, +# lifecycle_step, current_round_id) — those are written only by the live +# eval workers, so without persistence the dashboard's "Evaluated N of M" +# column blanks for a full cycle after every restart. v3 also adds +# `uid_to_val_loss`, the only per-miner field with no recovery path at all: +# `validator_miner_val_loss` is emitted at eval time and cannot be derived +# from `scores`, because `delta = max(0.0, baseline - val_loss)` clamps at +# zero, so every miner scoring 0 would be underivable. +# `from_json` accepts v1/v2 files (missing fields default) so leftover +# journals written by an older build still recover. +SCHEMA_VERSION = 3 JOURNAL_DIR_NAME = "round_journal" JOURNAL_FILENAME_PREFIX = "round_" JOURNAL_FILENAME_SUFFIX = ".json" @@ -59,6 +68,17 @@ class RoundJournal: # v2: uid -> (hf_repo_id, hf_revision) evaluated for the round. Only # uids whose chain checkpoint carried BOTH values are recorded. uid_to_commit: dict[int, tuple[str, str]] = field(default_factory=dict) + # v3: round-level gauge inputs. `roster_size` = len(foreground_uids) + + # len(background_uids) at freeze; `lifecycle_step` = the last live + # lifecycle step the round reached (0 freeze / 2 post-foreground / + # 3 eval-window). Both let startup recovery restore the round-level + # gauges. Default 0 for v1/v2 journals (recovery then leaves pending at + # 0 rather than inventing a roster). + roster_size: int = 0 + lifecycle_step: int = 0 + # v3: uid -> raw evaluation loss, recorded at `mark_scored`. Only + # populated for uids actually evaluated this round. + uid_to_val_loss: dict[int, float] = field(default_factory=dict) finalized: bool = False schema_version: int = SCHEMA_VERSION @@ -75,6 +95,9 @@ def to_json(self) -> str: payload["uid_to_commit"] = { str(k): [str(v[0]), str(v[1])] for k, v in self.uid_to_commit.items() } + payload["uid_to_val_loss"] = { + str(k): float(v) for k, v in self.uid_to_val_loss.items() + } return json.dumps(payload) @classmethod @@ -106,11 +129,139 @@ def from_json(cls, data: str) -> "RoundJournal": for k, v in raw.get("uid_to_commit", {}).items() if isinstance(v, (list, tuple)) and len(v) == 2 }, + roster_size=int(raw.get("roster_size", 0)), + lifecycle_step=int(raw.get("lifecycle_step", 0)), + uid_to_val_loss={ + int(k): float(v) for k, v in raw.get("uid_to_val_loss", {}).items() + }, finalized=bool(raw.get("finalized", False)), schema_version=version, ) +def verdict_uids(journal: "RoundJournal") -> set[int]: + """UIDs that received a finalize verdict for this round. + + Mirrors which uids `finalize_round_scores` writes an aggregator entry + for: every scored uid, every explicit validation failure, and every + freeze-zero uid not already in those sets. Operational failures + (`failed_uids` minus `validation_failed_uids`) are deliberately + excluded — finalize writes nothing for them so the miner keeps its + prior EMA, and telemetry must not imply otherwise. + + A uid with no known hotkey is skipped, matching finalize's `continue`. + """ + scored = set(journal.scored_uids) + validation_failed = set(journal.validation_failed_uids) + freeze_zero = set(journal.freeze_zero_uids) - scored - validation_failed + out: set[int] = set() + for uid in scored | validation_failed | freeze_zero: + if uid in journal.uid_to_hotkey or uid in journal.freeze_zero_hotkeys: + out.add(int(uid)) + return out + + +def republish_telemetry_from_journal( + journal: "RoundJournal", score_aggregator=None +) -> int: + """Re-emit a finalized round's Prometheus series from its journal. + + **Metrics only — this must never mutate scoring state.** It exists + because startup recovery works by *replaying* `finalize_round_scores`, + which marks the journal finalized; the recovery loop then skips + finalized journals, so a second restart re-emits nothing and the + dashboard loses the whole last completed cycle (observed 2026-07-31: + two Watchtower restarts 25 minutes apart left every per-miner family at + zero series for 17 minutes). + + Re-running finalize instead would be actively harmful, and not for the + obvious reason: `finalize_round_scores` calls `drop_round` first, so the + aggregator's *point set* would stay correct — but `add_score` stamps + `_utc_now()`, so the round's points would get fresh timestamps and jump + to the end of the time-ordered series. The rolling average is "last + `max_points` **by timestamp**", and that average is what drives weight + submission, so a re-finalize would silently reshuffle the scoring + window. Hence: read the journal, set gauges, touch nothing else. + + `score_aggregator` is read (never written) for the latest/avg/samples + snapshot, which lives in the aggregator rather than the journal. Pass + `None` to skip those three gauges. + + Returns the number of uids republished. Best-effort — never raises. + """ + from connito.shared.telemetry import ( + VALIDATOR_CURRENT_ROUND_ID, + VALIDATOR_MINER_VAL_LOSS, + VALIDATOR_ROUND_LIFECYCLE_STEP, + set_miner_evaluated_commit, + set_miner_last_scored_round, + set_miner_round_delta, + set_miner_score_snapshot, + set_round_progress, + ) + + try: + rid = int(journal.round_id) + uids = verdict_uids(journal) + + latest_scores: dict[int, float] = {} + avg_scores: dict[int, float] = {} + if score_aggregator is not None: + try: + latest_scores = score_aggregator.uid_score_pairs(how="latest") + avg_scores = score_aggregator.uid_score_pairs(how="avg") + except Exception: + latest_scores, avg_scores = {}, {} + + for uid in uids: + set_miner_last_scored_round(uid, rid) + samples = None + if score_aggregator is not None: + try: + samples = score_aggregator.record_count(uid) + except Exception: + samples = None + set_miner_score_snapshot( + uid, + latest=latest_scores.get(uid), + avg=avg_scores.get(uid), + samples=samples, + ) + + # Raw per-round delta, for every uid actually evaluated. + for uid in journal.scored_uids: + set_miner_round_delta(int(uid), float(journal.scores.get(uid, 0.0))) + + # val_loss — the field with no other recovery path (v3+ journals). + for uid, loss in journal.uid_to_val_loss.items(): + try: + VALIDATOR_MINER_VAL_LOSS.labels(miner_uid=str(int(uid))).set(float(loss)) + except Exception: + pass + + for uid, (repo, rev) in journal.uid_to_commit.items(): + set_miner_evaluated_commit(int(uid), repo, rev, rid) + + # Round-level counters. `roster_size` is 0 on pre-v3 journals, so + # pending clamps to 0 rather than inventing a denominator. + scored_n = len(journal.scored_uids) + failed_n = len(journal.failed_uids) + set_round_progress( + rid, + scored=scored_n, + failed=failed_n, + pending=max(0, int(journal.roster_size) - scored_n - failed_n), + ) + if journal.lifecycle_step: + VALIDATOR_ROUND_LIFECYCLE_STEP.labels(round_id=str(rid)).set( + int(journal.lifecycle_step) + ) + VALIDATOR_CURRENT_ROUND_ID.set(float(rid)) + return len(uids) + except Exception: + return 0 + + def commit_map_from_checkpoints( uid_to_chain_checkpoint: dict[int, object], ) -> dict[int, tuple[str, str]]: @@ -218,6 +369,12 @@ class _RecoveryRound: # `uid_to_commit` map (empty for v1 journals) so a recovered finalize # re-publishes evaluated-commit telemetry too. uid_to_chain_checkpoint: dict[int, object] = field(default_factory=dict) + # v3: carried through so the finalize re-write (finalized=True) preserves + # them, and so the recovery pass can restore the round-level gauges. + roster_size: int = 0 + lifecycle_step: int = 0 + # Carried so the finalize journal-rewrite preserves the losses. + val_losses: dict[int, float] = field(default_factory=dict) _lock: threading.Lock = field(default_factory=threading.Lock, repr=False) @classmethod @@ -238,6 +395,9 @@ def from_journal(cls, journal: "RoundJournal", journal_path: str | os.PathLike) int(uid): SimpleNamespace(hf_repo_id=repo, hf_revision=rev) for uid, (repo, rev) in journal.uid_to_commit.items() }, + roster_size=int(journal.roster_size), + lifecycle_step=int(journal.lifecycle_step), + val_losses=dict(journal.uid_to_val_loss), ) def processed_uids_snapshot(self) -> tuple[set[int], set[int]]: diff --git a/connito/validator/run.py b/connito/validator/run.py index 925b80a..a529b6c 100644 --- a/connito/validator/run.py +++ b/connito/validator/run.py @@ -927,6 +927,48 @@ def run(rank: int, world_size: int, config: ValidatorConfig, pkg_version: str = journals_finalized=_recovered, journals_seen=len(_journals), ) + + # Re-emit the most recent finalized round's telemetry. + # + # Runs unconditionally, and this is the point: the replay loop above + # only touches *unfinalized* journals, and replaying one marks it + # finalized. So a second restart finds nothing to replay and used to + # emit nothing at all — leaving the dashboard blank for the whole + # last completed cycle until the next round's evaluations arrived + # (observed 2026-07-31: two Watchtower restarts 25 min apart, every + # per-miner family at zero series for 17 minutes). + # + # This is a METRICS-ONLY pass — it never re-runs finalize. Re-running + # finalize would keep the aggregator's point set correct (drop_round + # runs first) but would re-stamp those points with fresh timestamps, + # reshuffling the "last N by timestamp" rolling average that drives + # weight submission. See `republish_telemetry_from_journal`. + # + # Journals are scanned ascending, so the last finalized one is the + # most recent round — which is also the one just replayed on a first + # restart, making the re-emit a harmless idempotent gauge write. + try: + _newest_finalized = None + for _journal_file in reversed(_journals): + _j = _rj_recover.load(_journal_file) + if _j is not None and _j.finalized: + _newest_finalized = _j + break + if _newest_finalized is not None: + _republished = _rj_recover.republish_telemetry_from_journal( + _newest_finalized, score_aggregator=score_aggregator, + ) + logger.info( + "Startup recovery: republished telemetry for last finalized round", + round_id=_newest_finalized.round_id, + uids=_republished, + schema_version=_newest_finalized.schema_version, + val_losses=len(_newest_finalized.uid_to_val_loss), + ) + except Exception as e: + logger.warning( + "Startup recovery: telemetry republish failed", error=str(e), + ) except Exception as e: logger.warning( "Startup recovery: scan failed", error=str(e), @@ -1473,6 +1515,7 @@ def run(rank: int, world_size: int, config: ValidatorConfig, pkg_version: str = download_window_closed.clear() try: note_round_series(new_round.round_id) + new_round.lifecycle_step = 0 VALIDATOR_ROUND_LIFECYCLE_STEP.labels(round_id=str(new_round.round_id)).set(0) # Seed the progress counters at freeze so the round exists in # the metric from the moment it is frozen (scored=0, failed=0, @@ -1559,6 +1602,7 @@ async def _bounded_foreground_eval(): eval_worker.set_eval_base_model(copy.deepcopy(global_model)) try: note_round_series(new_round.round_id) + new_round.lifecycle_step = 2 VALIDATOR_ROUND_LIFECYCLE_STEP.labels(round_id=str(new_round.round_id)).set(2) except Exception: pass @@ -1791,6 +1835,7 @@ async def _bounded_foreground_eval(): eval_window_active.set() try: note_round_series(new_round.round_id) + new_round.lifecycle_step = 3 VALIDATOR_ROUND_LIFECYCLE_STEP.labels(round_id=str(new_round.round_id)).set(3) except Exception: pass