validator: stop orphan raw scores inflating rolling avg / weights

connito/validator/aggregator.py

@@ -15,6 +15,21 @@
 # so a restart mid-round can drop the in-flight round's partial scores.
 SCHEMA_VERSION = 2
 
+# The only score values a *finalized* round ever writes (see
+# `finalize_round_scores`): the geometric rank mapping 2.25/1.5/1.0 for the
+# top-3 plus 0.0 for everyone else. Any other value in the window is a *raw*
+# in-cycle score (`delta ** 1.2`) that finalize is supposed to replace via
+# `drop_round`. A raw value that outlives its round (an "orphan", e.g. from a
+# late background eval that landed after finalize, or a round whose finalize
+# never completed) rides the rolling avg as an illegal non-rank score and
+# distorts weights. `sweep_orphan_raw_scores` removes such stale points.
+_RANK_SCORE_VALUES: tuple[float, ...] = (0.0, 1.0, 1.5, 2.25)
+_RANK_SCORE_EPS: float = 1e-9
+
+
+def _is_rank_score(value: float) -> bool:
+    return any(abs(float(value) - r) <= _RANK_SCORE_EPS for r in _RANK_SCORE_VALUES)
+
 
 def _utc_now() -> datetime:
     return datetime.now(timezone.utc)
@@ -478,6 +493,43 @@ def prune_before_round(self, min_round_id: int) -> int:
                 dropped += state.series.prune_before_round(min_round_id)
         return dropped
 
+    def sweep_orphan_raw_scores(self) -> int:
+        """Drop every *raw* (non-rank) score point across all miners.
+
+        A finalized round only ever stores rank values (2.25/1.5/1.0/0.0).
+        Any other value is a raw in-cycle `delta ** 1.2` point that
+        `finalize_round_scores`'s `drop_round` was supposed to replace. One
+        that outlives its round is an orphan: it rides the rolling avg as an
+        illegal non-rank score and inflates the on-chain weight derived from
+        it (e.g. a single 5.179 point pinning a uid's avg at ~1.29).
+
+        Every non-rank point is dropped unconditionally — including one tagged
+        with the miner's highest round_id. We do NOT special-case a
+        "still in-flight" raw: this runs at startup right before the
+        round-journal recovery pass (see run.py), which replays any
+        unfinalized round through `finalize_round_scores`. That re-derives the
+        round's rank scores from the journal (`drop_round` + re-add), entirely
+        independent of the aggregator's pre-existing points, so a genuinely
+        in-flight raw is reconstructed there rather than preserved here.
+        Keeping the in-flight raw instead would leave a hole: an orphan that
+        happens to sit at the miner's newest round_id (e.g. a miner not
+        evaluated since) would never be swept.
+
+        Belt-and-suspenders cleanup for orphans persisted before the
+        finalized-round race fix landed. Returns the number of points dropped.
+        """
+        dropped = 0
+        with self._lock:
+            for state in self._miners.values():
+                pts = state.series.points
+                if not pts:
+                    continue
+                kept = [p for p in pts if _is_rank_score(p[1])]
+                if len(kept) != len(pts):
+                    dropped += len(pts) - len(kept)
+                    state.series.points = kept
+        return dropped
+
     def record_count(self, uid: int) -> int:
         """Number of recorded score points for ``uid`` (0 if unknown)."""
         uid = int(uid)

connito/validator/evaluator.py

@@ -205,12 +205,24 @@ def finalize_round_scores(
     """
     # Snapshot all sets under the round's lock so the worker threads
     # cannot race a mark_scored / mark_failed against the read.
-    scored, _failed = round_obj.processed_uids_snapshot()
+    # Mark the round finalized FIRST, atomically with snapshotting its
+    # scores. `mark_scored` checks `round.finalized` under this same lock and
+    # drops any later (e.g. background) eval result, so no in-flight eval can
+    # add a fresh raw point to the aggregator after the `drop_round` below.
+    # Without this gate a late bg-eval writes a raw `delta ** 1.2` tagged with
+    # this round_id *after* drop_round has run; the point is never re-ranked
+    # and rides the rolling avg as an illegal non-rank score (the root cause
+    # of the inflated `validator_miner_score_avg` / weight distortion).
+    # `_RecoveryRound` (startup replay) has no `finalized` field; setattr
+    # just creates it, and recovery is single-threaded so the gate is a no-op
+    # there.
     # `round.scores` is mutated under the same lock; copy it explicitly
     # rather than alias.
     with round_obj._lock:  # noqa: SLF001 — same module family
+        round_obj.finalized = True
         round_scores = dict(round_obj.scores)
         validation_failed = set(round_obj.validation_failed_uids)
+    scored, _failed = round_obj.processed_uids_snapshot()
     freeze_zero = set(round_obj.freeze_zero_uids)
     freeze_hotkeys = dict(round_obj.freeze_zero_hotkeys)
 

connito/validator/round.py

@@ -93,6 +93,18 @@ 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
+    # Set True by `finalize_round_scores` (under `_lock`, atomically with its
+    # score snapshot) the instant it begins finalizing this round. Once set,
+    # `mark_scored` drops any further result for the round instead of
+    # recording it. This closes the late-eval race: a background eval that
+    # finishes after the eval window closed and `finalize_round_scores` has
+    # already run would otherwise write a *raw* in-cycle score
+    # (`delta ** 1.2`) to the aggregator tagged with this round_id. Finalize's
+    # `drop_round(round_id)` has already passed, so that point is never
+    # re-ranked — it survives as an illegal non-rank value in the rolling
+    # window, inflating the miner's `score_avg` and its on-chain weight for up
+    # to the full history window. See `finalize_round_scores` and `mark_scored`.
+    finalized: bool = False
 
     # Round-group construction scheme (gated by
     # `config.evaluation.enable_round_group_construction`). All default
@@ -622,6 +634,11 @@ def _record_in_cycle_score(self, uid: int, hotkey: str, score: float) -> None:
         """
         if self.score_aggregator is None:
             return
+        if self.finalized:
+            # Defense-in-depth: never feed the aggregator once the round is
+            # finalized. `mark_scored` already gates on this, but guard here
+            # too so any future caller cannot reintroduce an orphan raw point.
+            return
         try:
             self.score_aggregator.add_score(
                 uid=uid, hotkey=hotkey, score=float(score), round_id=self.round_id,
@@ -648,6 +665,21 @@ def mark_scored(self, uid: int, score: float = 0.0) -> None:
         """
         score_f = float(score)
         with self._lock:
+            if self.finalized:
+                # The round was finalized (scores snapshotted, ranks written,
+                # drop_round already run) while this evaluation was still in
+                # flight. Recording it now would leave an un-ranked raw
+                # `delta ** 1.2` point in the aggregator that no drop_round
+                # ever removes — it would ride the rolling avg as an illegal
+                # non-rank score and inflate the miner's on-chain weight.
+                # Drop the stale result; release the claim so bookkeeping is
+                # consistent. The miner keeps its prior, correctly-ranked EMA.
+                self.claimed_uids.discard(uid)
+                logger.warning(
+                    "Round: dropping late score for finalized round",
+                    uid=uid, round_id=self.round_id, score=round(score_f, 6),
+                )
+                return
             self.scored_uids.add(uid)
             self.scores[uid] = score_f
             self.claimed_uids.discard(uid)

connito/validator/run.py

@@ -799,6 +799,23 @@ def run(rank: int, world_size: int, config: ValidatorConfig, pkg_version: str =
                     max_points=score_window,
                     max_history_points=score_history_window,
                 )
+            # Sweep stale raw (non-rank) score points left on disk — orphans
+            # from a late background eval that landed after finalize, or from a
+            # round whose finalize never completed (e.g. a scoring failure +
+            # restart). These ride the rolling avg as illegal non-rank scores
+            # and inflate on-chain weights until they age out of the window.
+            _swept = score_aggregator.sweep_orphan_raw_scores()
+            if _swept:
+                logger.warning(
+                    "Dropped orphan raw (non-rank) score points on load",
+                    dropped=_swept,
+                )
+                try:
+                    score_aggregator.persist_atomic(score_path)
+                except Exception as e:
+                    logger.warning(
+                        "persist_atomic after orphan sweep failed", error=str(e),
+                    )
             _loaded_latest = score_aggregator.uid_score_pairs(how="latest")
             _loaded_avg = score_aggregator.uid_score_pairs(how="avg")
             logger.info(