Close parser-poison free-rider hole + harden sentinel-rate handling

allways/commitments.py

@@ -1,25 +1,36 @@
 """Shared commitment parsing logic — used by validator, miner, and CLI."""
 
 import math
-from typing import List, Optional
+from typing import List, Optional, Set
 
 import bittensor as bt
 from bittensor.utils import ss58_encode
 
 from allways.chains import SUPPORTED_CHAINS, canonical_pair
 from allways.classes import MinerPair
 from allways.constants import COMMITMENT_VERSION
-from allways.utils.rate import normalize_rate
+from allways.utils.rate import is_executable_rate, normalize_rate
 
 SS58_PREFIX = 42
 
 
-def parse_commitment_data(raw: str, uid: int = 0, hotkey: str = '') -> Optional[MinerPair]:
+def parse_commitment_data(
+    raw: str,
+    uid: int = 0,
+    hotkey: str = '',
+    *,
+    min_swap_rao: int = 0,
+    max_swap_rao: int = 0,
+) -> Optional[MinerPair]:
     """Parse a commitment string into a MinerPair.
 
     Format: v{VERSION}:{src_chain}:{src_addr}:{dst_chain}:{dst_addr}:{rate}:{counter_rate}
     Both rates are 'canonical_dest per 1 canonical_source'. rate is for source→dest, counter_rate for dest→source.
     Example: v1:btc:bc1q...:tao:5C...:340:350
+
+    When ``min_swap_rao`` / ``max_swap_rao`` are non-zero, any positive rate that
+    is not executable under those bounds drops the entire pair. Zero stays
+    opt-out semantics (one direction disabled), not sentinel.
     """
     try:
         parts = raw.split(':')
@@ -65,6 +76,14 @@ def parse_commitment_data(raw: str, uid: int = 0, hotkey: str = '') -> Optional[
             rate, counter_rate = counter_rate, rate
             rate_str, counter_rate_str = counter_rate_str, rate_str
 
+        if min_swap_rao > 0 or max_swap_rao > 0:
+            if rate > 0 and not is_executable_rate(rate, src_chain, dst_chain, min_swap_rao, max_swap_rao):
+                return None
+            if counter_rate > 0 and not is_executable_rate(
+                counter_rate, dst_chain, src_chain, min_swap_rao, max_swap_rao
+            ):
+                return None
+
         return MinerPair(
             uid=uid,
             hotkey=hotkey,
@@ -125,6 +144,9 @@ def read_miner_commitment(
     hotkey: str,
     block: Optional[int] = None,
     metagraph: Optional['bt.Metagraph'] = None,
+    *,
+    min_swap_rao: int = 0,
+    max_swap_rao: int = 0,
 ) -> Optional[MinerPair]:
     """Read a single miner's commitment, optionally at a specific block.
 
@@ -142,19 +164,36 @@ def read_miner_commitment(
         uid = resolved
     commitment = get_commitment(subtensor, netuid, hotkey, block=block)
     if commitment:
-        return parse_commitment_data(commitment, uid=uid, hotkey=hotkey)
+        return parse_commitment_data(
+            commitment,
+            uid=uid,
+            hotkey=hotkey,
+            min_swap_rao=min_swap_rao,
+            max_swap_rao=max_swap_rao,
+        )
     return None
 
 
-def read_miner_commitments(subtensor: bt.Subtensor, netuid: int) -> List[MinerPair]:
+def read_miner_commitments(
+    subtensor: bt.Subtensor,
+    netuid: int,
+    *,
+    min_swap_rao: int = 0,
+    max_swap_rao: int = 0,
+) -> List[MinerPair]:
     """Read all miner commitments for the netuid in a single RPC call.
 
     Uses substrate-interface's ``query_map`` over the ``CommitmentOf`` double map
     keyed by ``(netuid, hotkey)``. One RPC round-trip returns every committed
     hotkey on the subnet — cheaper than the old N-RPC for-loop, matters most
     on full validator polling cadence.
+
+    When ``min_swap_rao`` / ``max_swap_rao`` are non-zero, pairs with any
+    unexecutable positive rate are dropped at the parser layer so the validator
+    never sees them.
     """
     pairs: List[MinerPair] = []
+    dropped = 0
     try:
         metagraph = subtensor.metagraph(netuid)
         hotkey_to_uid = {metagraph.hotkeys[uid]: uid for uid in range(metagraph.n.item())}
@@ -182,11 +221,83 @@ def read_miner_commitments(subtensor: bt.Subtensor, netuid: int) -> List[MinerPa
             commitment = decode_commitment_field(metadata)
             if not commitment:
                 continue
-            pair = parse_commitment_data(commitment, uid=uid, hotkey=hotkey)
+            pair = parse_commitment_data(
+                commitment,
+                uid=uid,
+                hotkey=hotkey,
+                min_swap_rao=min_swap_rao,
+                max_swap_rao=max_swap_rao,
+            )
             if pair:
                 pairs.append(pair)
+            elif min_swap_rao > 0 or max_swap_rao > 0:
+                # Re-parse permissively to distinguish "unexecutable under bounds"
+                # from "malformed/garbage". Only the former counts as dropped.
+                if parse_commitment_data(commitment, uid=uid, hotkey=hotkey) is not None:
+                    dropped += 1
     except (ConnectionError, TimeoutError) as e:
         bt.logging.warning(f'Transient error reading commitments: {e}')
     except Exception as e:
         bt.logging.error(f'Error reading commitments: {e}')
+    if dropped > 0 and (min_swap_rao > 0 or max_swap_rao > 0):
+        bt.logging.info(
+            f'Commitments: dropped {dropped} pair(s) with unexecutable rates '
+            f'under bounds [{min_swap_rao}, {max_swap_rao}]'
+        )
     return pairs
+
+
+def read_unexecutable_commitments(
+    subtensor: bt.Subtensor,
+    netuid: int,
+    min_swap_rao: int,
+    max_swap_rao: int,
+) -> Set[str]:
+    """Hotkeys whose commitment parses permissively but drops under bounds.
+
+    Distinct from malformed/garbage commitments — those don't return either way.
+    Staged for the follow-up auto-deactivate streak tracker; no live caller in
+    this PR.
+    """
+    unexecutable: Set[str] = set()
+    if min_swap_rao <= 0 and max_swap_rao <= 0:
+        return unexecutable
+    try:
+        metagraph = subtensor.metagraph(netuid)
+        hotkey_to_uid = {metagraph.hotkeys[uid]: uid for uid in range(metagraph.n.item())}
+        result = subtensor.substrate.query_map(
+            module='Commitments',
+            storage_function='CommitmentOf',
+            params=[netuid],
+        )
+        for key, metadata in result:
+            raw = key.value if hasattr(key, 'value') else key
+            if isinstance(raw, tuple) and len(raw) == 1:
+                raw = raw[0]
+            if isinstance(raw, (tuple, list)):
+                raw = bytes(raw)
+            if isinstance(raw, (bytes, bytearray)) and len(raw) == 32:
+                hotkey = ss58_encode(bytes(raw), SS58_PREFIX)
+            else:
+                hotkey = str(raw)
+            if hotkey not in hotkey_to_uid:
+                continue
+            commitment = decode_commitment_field(metadata)
+            if not commitment:
+                continue
+            permissive = parse_commitment_data(commitment, hotkey=hotkey)
+            if permissive is None:
+                continue
+            bounded = parse_commitment_data(
+                commitment,
+                hotkey=hotkey,
+                min_swap_rao=min_swap_rao,
+                max_swap_rao=max_swap_rao,
+            )
+            if bounded is None:
+                unexecutable.add(hotkey)
+    except (ConnectionError, TimeoutError) as e:
+        bt.logging.warning(f'Transient error reading commitments: {e}')
+    except Exception as e:
+        bt.logging.error(f'Error reading commitments: {e}')
+    return unexecutable

allways/utils/rate.py

@@ -193,6 +193,33 @@ def _has_integer_routable_source(forward_rate: float, src_chain: str) -> bool:
     return True
 
 
+def min_executable_tao_leg(
+    rate: float,
+    from_chain: str,
+    to_chain: str,
+    min_swap_rao: int,
+    max_swap_rao: int,
+) -> int:
+    """Smallest TAO leg (rao) the rate produces among in-band fundable swaps.
+
+    Shares band math with is_executable_rate. Returns 0 when no in-band
+    fundable swap exists (rate unexecutable) — caller treats as "no constraint".
+    """
+    if not is_executable_rate(rate, from_chain, to_chain, min_swap_rao, max_swap_rao):
+        return 0
+    if from_chain == 'tao':
+        return max(get_chain('tao').min_onchain_amount, max(0, min_swap_rao))
+    if to_chain == 'tao':
+        src = get_chain(from_chain)
+        decimal_factor = 10 ** (get_chain('tao').decimals - src.decimals)
+        denom = rate * decimal_factor
+        if not math.isfinite(denom) or denom <= 0:
+            return 0
+        min_source = max(src.min_onchain_amount, math.ceil(max(1, min_swap_rao) / denom))
+        return int(min_source * denom)
+    return 0
+
+
 def check_swap_viability(
     tao_amount_rao: int,
     miner_collateral_rao: int,

allways/validator/axon_handlers.py

@@ -23,7 +23,12 @@
 from allways.synapses import MinerActivateSynapse, SwapConfirmSynapse, SwapReserveSynapse
 from allways.utils.logging import miner_label as _miner_label
 from allways.utils.proofs import reserve_proof_message, swap_proof_message
-from allways.utils.rate import calculate_to_amount, derive_tao_leg, quote_within_slippage
+from allways.utils.rate import (
+    calculate_to_amount,
+    derive_tao_leg,
+    is_executable_rate,
+    quote_within_slippage,
+)
 from allways.utils.scale import encode_bytes, encode_str, encode_u128
 from allways.validator.state_store import PendingConfirm, ReservationPin
 
@@ -227,9 +232,11 @@ async def handle_miner_activate(
                 netuid=validator.config.netuid,
                 hotkey=miner_hotkey,
                 metagraph=validator.metagraph,
+                min_swap_rao=validator.bounds_cache.min_swap_amount(),
+                max_swap_rao=validator.bounds_cache.max_swap_amount(),
             )
             if commitment is None:
-                reject_synapse(synapse, 'No commitment found', ctx)
+                reject_synapse(synapse, 'No valid commitment (missing, malformed, or rate not executable)', ctx)
                 return synapse
 
             collateral, active, _, _, _ = contract.get_miner_snapshot(miner_hotkey)
@@ -358,6 +365,11 @@ async def handle_swap_reserve(
             if reserve_rate <= 0:
                 reject_synapse(synapse, 'Miner does not support this swap direction', ctx)
                 return synapse
+            min_swap = validator.bounds_cache.min_swap_amount()
+            max_swap = validator.bounds_cache.max_swap_amount()
+            if not is_executable_rate(reserve_rate, synapse.from_chain, synapse.to_chain, min_swap, max_swap):
+                reject_synapse(synapse, 'Miner rate is not executable under current swap bounds', ctx)
+                return synapse
             bt.logging.info(
                 f'{ctx}: commitment ok — miner_rate={reserve_rate_str or reserve_rate} '
                 f'miner_from={commitment.from_address} miner_to={commitment.to_address}'
@@ -413,8 +425,6 @@ async def handle_swap_reserve(
                 reject_synapse(synapse, 'Miner collateral below minimum', ctx)
                 return synapse
 
-            min_swap = validator.bounds_cache.min_swap_amount()
-            max_swap = validator.bounds_cache.max_swap_amount()
             if min_swap > 0 and synapse.tao_amount < min_swap:
                 reject_synapse(synapse, f'Swap amount below minimum ({synapse.tao_amount} < {min_swap} rao)', ctx)
                 return synapse

allways/validator/forward.py

@@ -378,12 +378,29 @@ def poll_commitments(self: Validator) -> None:
 
 def refresh_miner_rates(self: Validator) -> None:
     try:
-        pairs = read_miner_commitments(self.subtensor, self.config.netuid)
+        max_swap_amount = int(self.bounds_cache.max_swap_amount())
+    except Exception as e:
+        bt.logging.warning(f'max_swap_amount read failed: {e}')
+        max_swap_amount = 0
+    try:
+        min_swap_amount = int(self.bounds_cache.min_swap_amount())
+    except Exception as e:
+        bt.logging.warning(f'min_swap_amount read failed: {e}')
+        min_swap_amount = 0
+
+    try:
+        pairs = read_miner_commitments(
+            self.subtensor,
+            self.config.netuid,
+            min_swap_rao=min_swap_amount,
+            max_swap_rao=max_swap_amount,
+        )
     except Exception as e:
         bt.logging.warning(f'Commitment poll failed: {e}')
         return
 
     current_hotkeys = set(self.metagraph.hotkeys)
+    admitted_keys: set[tuple[str, str, str]] = set()
 
     for pair in pairs:
         if pair.hotkey not in current_hotkeys:
@@ -406,6 +423,7 @@ def refresh_miner_rates(self: Validator) -> None:
                     )
                     self.last_known_rates[key] = 0.0
                 continue
+            admitted_keys.add(key)
             if self.last_known_rates.get(key) == r:
                 continue
             self.state_store.insert_rate_event(
@@ -417,6 +435,38 @@ def refresh_miner_rates(self: Validator) -> None:
             )
             self.last_known_rates[key] = r
 
+    # SECOND SWEEP: terminate previously-positive directions that vanished from
+    # this poll. Covers parser-poison (commitment overwritten with garbage) and
+    # bounds-tighten exits (rate dropped below executability). Without this a
+    # miner's stale positive rate keeps earning crown until deregistration.
+    #
+    # Guard: read_miner_commitments swallows transient RPC errors and returns
+    # an empty list. If pairs is empty, we can't distinguish "RPC dead" from
+    # "nobody posting" — either way, terminating every miner is wrong. Skip
+    # the sweep; the next successful poll catches whatever genuinely vanished.
+    if not pairs:
+        return
+    for key, rate in list(self.last_known_rates.items()):
+        if rate <= 0:
+            continue
+        hk, from_c, to_c = key
+        if hk not in current_hotkeys:
+            continue  # purge_deregistered_hotkeys handles dereg
+        if key in admitted_keys:
+            continue
+        latest = self.state_store.get_latest_rate_before(hk, from_c, to_c, self.block)
+        if latest is None or latest[0] <= 0:
+            continue
+        self.state_store.insert_rate_event(
+            hotkey=hk,
+            from_chain=from_c,
+            to_chain=to_c,
+            rate=0.0,
+            block=self.block,
+        )
+        self.last_known_rates[key] = 0.0
+        bt.logging.info(f'forward: terminating rate for {hk[:8]} {from_c}->{to_c} — commitment dropped')
+
 
 def purge_deregistered_hotkeys(self: Validator) -> None:
     current_hotkeys = set(self.metagraph.hotkeys)

allways/validator/scoring.py

@@ -27,7 +27,7 @@
     SUCCESS_EXPONENT,
     VOLUME_WEIGHT_ALPHA,
 )
-from allways.utils.rate import is_executable_rate
+from allways.utils.rate import is_executable_rate, min_executable_tao_leg
 from allways.validator.event_watcher import ContractEventWatcher
 from allways.validator.scoring_trace import WeightingTrace, log_scoring_trace
 from allways.validator.state_store import ValidatorStateStore
@@ -571,6 +571,15 @@ def effective_rates() -> Dict[str, float]:
         merged.update(pinned_rates)
         return merged
 
+    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.
+        min_leg = min_executable_tao_leg(rate, from_chain, to_chain, min_swap_rao, max_swap_rao)
+        return min_leg == 0 or collaterals.get(hotkey, 0) >= min_leg
+
     def credit_interval(interval_start: int, interval_end: int) -> None:
         duration = interval_end - interval_start
         if duration <= 0:
@@ -584,6 +593,7 @@ def credit_interval(interval_start: int, interval_end: int) -> None:
             active=active_set,
             lower_rate_wins=lower_rate_wins,
             executable_rate_check=executable_check,
+            can_fund_at_rate=can_fund if bounds_set else None,
         )
         if not holders:
             if trace is not None:
@@ -735,6 +745,7 @@ def crown_holders_at_instant(
     active: Optional[Set[str]] = None,
     lower_rate_wins: bool = False,
     executable_rate_check: Optional[Callable[[float], bool]] = None,
+    can_fund_at_rate: Optional[Callable[[str, float], bool]] = None,
 ) -> List[str]:
     """Take the miners posting the best rate, but only if they satisfy every
     other condition (rewardable, active, not busy, rate > 0, executable).
@@ -772,6 +783,8 @@ def qualifies(hotkey: str) -> bool:
             return False
         if executable_rate_check is not None and not executable_rate_check(rate):
             return False
+        if can_fund_at_rate is not None and not can_fund_at_rate(hotkey, rate):
+            return False
         return hotkey in rewardable and hotkey not in busy
 
     by_rate: Dict[float, List[str]] = {}