📊 telemetry: restore round-level gauges on journal recovery (schema v3)

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)
 # ---------------------------------------------------------------------------

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}

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,

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(