Extract shared crown-eligibility predicates so scoring and live-crown can't drift

allways/validator/scoring.py

@@ -10,6 +10,7 @@
 
 from dataclasses import dataclass, field
 from enum import IntEnum
+from functools import partial
 from typing import TYPE_CHECKING, Callable, Dict, List, Optional, Set, Tuple
 
 import bittensor as bt
@@ -512,6 +513,39 @@ def merge_replay_events(
     return events
 
 
+def crown_can_fund(hotkey, rate, from_chain, to_chain, min_swap_rao, max_swap_rao, collaterals):
+    """Boundary-squat gate: a miner whose own rate forces a TAO leg larger than
+    their collateral earns no crown. Fail open on unknown collateral (absent !=
+    zero) so a missing baseline doesn't silently drop them."""
+    if hotkey not in collaterals:
+        return True
+    min_leg = min_executable_tao_leg(rate, from_chain, to_chain, min_swap_rao, max_swap_rao)
+    return min_leg == 0 or collaterals[hotkey] >= min_leg
+
+
+def make_crown_predicates(from_chain, to_chain, min_swap_rao, max_swap_rao, collaterals):
+    """Crown-eligibility predicates ``(executable_check, can_fund)`` shared by the
+    scoring replay and the live snapshot, so the live crown view can never diverge
+    from the rewarded ledger. Both are the shared rate utils with this direction's
+    bounds/collateral bound in."""
+    executable_check = partial(
+        is_executable_rate,
+        from_chain=from_chain,
+        to_chain=to_chain,
+        min_swap_rao=min_swap_rao,
+        max_swap_rao=max_swap_rao,
+    )
+    can_fund = partial(
+        crown_can_fund,
+        from_chain=from_chain,
+        to_chain=to_chain,
+        min_swap_rao=min_swap_rao,
+        max_swap_rao=max_swap_rao,
+        collaterals=collaterals,
+    )
+    return executable_check, can_fund
+
+
 def replay_crown_time_window(
     store: ValidatorStateStore,
     event_watcher: ContractEventWatcher,
@@ -555,8 +589,7 @@ def replay_crown_time_window(
     canon_from, _ = canonical_pair(from_chain, to_chain)
     lower_rate_wins = from_chain != canon_from
 
-    def executable_check(rate: float) -> bool:
-        return is_executable_rate(rate, from_chain, to_chain, min_swap_rao, max_swap_rao)
+    executable_check, can_fund = make_crown_predicates(from_chain, to_chain, min_swap_rao, max_swap_rao, collaterals)
 
     crown_blocks: Dict[str, float] = {}
     cap_weighted_blocks: Dict[str, float] = {}
@@ -575,19 +608,6 @@ def effective_rates() -> Dict[str, float]:
 
     bounds_set = min_swap_rao > 0 or max_swap_rao > 0
 
-    def can_fund(hotkey: str, rate: float) -> bool:
-        # Boundary-squat per-block gate: a miner whose own rate forces a TAO
-        # leg larger than their collateral_at_block earns no crown for that
-        # block. Cascades to the next-best rate via crown_holders_at_instant.
-        # Fail open when collateral is *unknown* (no event ever recorded):
-        # absent != zero. The contract auto-deactivates anyone below
-        # min_collateral, so an active miner always holds enough; treating a
-        # missing baseline as 0 would silently drop them from crown.
-        if hotkey not in collaterals:
-            return True
-        min_leg = min_executable_tao_leg(rate, from_chain, to_chain, min_swap_rao, max_swap_rao)
-        return min_leg == 0 or collaterals[hotkey] >= min_leg
-
     def credit_interval(interval_start: int, interval_end: int) -> None:
         duration = interval_end - interval_start
         if duration <= 0:
@@ -706,20 +726,12 @@ def snapshot_current_crown_holders(
         if pinned_rates:
             rates = {**rates, **pinned_rates}
 
-        def executable_check(rate: float, from_chain=from_chain, to_chain=to_chain) -> bool:
-            return is_executable_rate(rate, from_chain, to_chain, min_swap_amount, max_swap_amount)
-
-        def can_fund(
-            hotkey: str, rate: float, from_chain=from_chain, to_chain=to_chain, collaterals=collaterals
-        ) -> bool:
-            # Mirror the scoring path's boundary-squat gate so the live table
-            # never credits a holder whose collateral can't fund their own
-            # smallest legal leg, which the ledger drops. Fail open on unknown
-            # collateral (absent != zero) to match the scoring path exactly.
-            if hotkey not in collaterals:
-                return True
-            min_leg = min_executable_tao_leg(rate, from_chain, to_chain, min_swap_amount, max_swap_amount)
-            return min_leg == 0 or collaterals[hotkey] >= min_leg
+        # Same predicates the scoring replay uses, so the live table never
+        # credits a holder the ledger drops. Built per direction so each
+        # closure captures the right chain pair.
+        executable_check, can_fund = make_crown_predicates(
+            from_chain, to_chain, min_swap_amount, max_swap_amount, collaterals
+        )
 
         holders = crown_holders_at_instant(
             rates,

tests/test_scoring_v1.py

@@ -12,12 +12,14 @@
     SCORING_WINDOW_BLOCKS,
     SUCCESS_EXPONENT,
 )
+from allways.utils.rate import is_executable_rate, min_executable_tao_leg
 from allways.validator.event_watcher import ActiveEvent, CollateralEvent, ContractEventWatcher
 from allways.validator.scoring import (
     calculate_miner_rewards,
     credibility_ramp,
     crown_holders_at_instant,
     due_for_scoring,
+    make_crown_predicates,
     replay_crown_time_window,
     score_and_reward_miners,
     scoring_window_bounds,
@@ -783,6 +785,52 @@ def test_boundary_squat_excluded_from_live_table(self, tmp_path: Path):
         v.state_store.close()
 
 
+class TestLedgerSnapshotAgreement:
+    """The #450 invariant end-to-end: the live snapshot and the scoring ledger
+    must resolve the crown to the same holder when fed identical state. Guards
+    against a future one-sided edit even if it bypassed make_crown_predicates."""
+
+    def test_squat_dropped_by_both_paths(self, tmp_path: Path):
+        # Squatter posts the best rate but can't fund the leg it forces; the
+        # funded runner-up is the only eligible holder. Both the per-forward
+        # snapshot and the windowed replay must agree on hk_funded and exclude
+        # hk_squat — the executability/funding gate applied identically.
+        v = make_validator(
+            tmp_path,
+            ['hk_squat', 'hk_funded'],
+            block=1100,
+            min_swap_amount=100_000_000,
+            max_swap_amount=500_000_000,
+            collaterals={'hk_squat': 150_000_000, 'hk_funded': 500_000_000},
+        )
+        conn = v.state_store.require_connection()
+        for hk, rate in (('hk_squat', 50000.0), ('hk_funded', 326.0)):
+            conn.execute(
+                'INSERT INTO rate_events (hotkey, from_chain, to_chain, rate, block) VALUES (?, ?, ?, ?, ?)',
+                (hk, 'btc', 'tao', rate, 0),
+            )
+        conn.commit()
+
+        snapshot_holders = [row[2] for row in snapshot_current_crown_holders(v)[('btc', 'tao')]]
+        ledger = replay_crown_time_window(
+            store=v.state_store,
+            event_watcher=v.event_watcher,
+            from_chain='btc',
+            to_chain='tao',
+            window_start=100,
+            window_end=1100,
+            rewardable_hotkeys={'hk_squat', 'hk_funded'},
+            min_swap_rao=100_000_000,
+            max_swap_rao=500_000_000,
+        )
+
+        assert snapshot_holders == ['hk_funded']
+        assert set(ledger) == {'hk_funded'}  # squatter credited zero blocks
+        # The whole point: live view and rewarded ledger name the same holder.
+        assert snapshot_holders == list(ledger.keys())
+        v.state_store.close()
+
+
 class TestPinnedRateDuringReservation:
     """Crown calculation must use the pinned rate during the reserved-not-busy
     window, not the live rate. Closes the bump-after-pin loophole."""
@@ -2805,3 +2853,59 @@ def test_fresh_seed_scores_one_trailing_window(self):
         seed = max(0, block - SCORING_WINDOW_BLOCKS)
         start, end = scoring_window_bounds(current_block=block, last_scored_block=seed)
         assert (start, end) == (block - SCORING_WINDOW_BLOCKS, block)
+
+
+class TestCrownPredicateParity:
+    """make_crown_predicates is the single source of crown eligibility for both
+    the scoring replay and the live snapshot. Lock its semantics to the shared
+    rate utils so a future edit can't let the live view drift from the ledger."""
+
+    # 0.1 / 0.5 TAO — the live on-chain swap bounds.
+    BOUNDS = [(0, 0), (100_000_000, 500_000_000)]
+    DIRECTIONS = [('btc', 'tao'), ('tao', 'btc')]
+    RATES = [0.00015, 1.0, 345.0, 50_000_000.0, 1e10, 0.0, -1.0, float('inf')]
+
+    def _reference(self, from_chain, to_chain, min_rao, max_rao, collaterals):
+        def exec_ref(rate):
+            return is_executable_rate(rate, from_chain, to_chain, min_rao, max_rao)
+
+        def fund_ref(hotkey, rate):
+            if hotkey not in collaterals:
+                return True
+            min_leg = min_executable_tao_leg(rate, from_chain, to_chain, min_rao, max_rao)
+            return min_leg == 0 or collaterals[hotkey] >= min_leg
+
+        return exec_ref, fund_ref
+
+    def test_matches_shared_rate_utils_across_matrix(self):
+        collaterals = {'hk_rich': 10_000_000_000, 'hk_poor': 1, 'hk_zero': 0}
+        probe_hotkeys = ['hk_rich', 'hk_poor', 'hk_zero', 'hk_absent']
+        for from_chain, to_chain in self.DIRECTIONS:
+            for min_rao, max_rao in self.BOUNDS:
+                executable_check, can_fund = make_crown_predicates(from_chain, to_chain, min_rao, max_rao, collaterals)
+                exec_ref, fund_ref = self._reference(from_chain, to_chain, min_rao, max_rao, collaterals)
+                for rate in self.RATES:
+                    assert executable_check(rate) == exec_ref(rate), (
+                        f'executable_check drift dir={from_chain}->{to_chain} bounds=({min_rao},{max_rao}) rate={rate}'
+                    )
+                    for hk in probe_hotkeys:
+                        assert can_fund(hk, rate) == fund_ref(hk, rate), (
+                            f'can_fund drift dir={from_chain}->{to_chain} '
+                            f'bounds=({min_rao},{max_rao}) hk={hk} rate={rate}'
+                        )
+
+    def test_fail_open_on_absent_collateral(self):
+        # absent != zero — a miner with no recorded baseline must not be dropped.
+        _, can_fund = make_crown_predicates('btc', 'tao', 100_000_000, 500_000_000, {})
+        assert can_fund('hk_unknown', 345.0) is True
+
+    def test_drops_holder_whose_collateral_cannot_fund_min_leg(self):
+        # 1-rao collateral can't cover any real in-band leg → boundary-squat drop;
+        # a richly-funded miner at the same rate passes.
+        collaterals = {'hk_poor': 1, 'hk_rich': 10_000_000_000}
+        _, can_fund = make_crown_predicates('btc', 'tao', 100_000_000, 500_000_000, collaterals)
+        rate = 345.0
+        min_leg = min_executable_tao_leg(rate, 'btc', 'tao', 100_000_000, 500_000_000)
+        assert min_leg > 0  # rate is executable, so the gate is live
+        assert can_fund('hk_poor', rate) is False
+        assert can_fund('hk_rich', rate) is True