Rolling coldkey-deduped meaningful-failure pool (S5.6 fix)

tests/test_mf_pool.py

@@ -0,0 +1,61 @@
+"""Rolling meaningful-failure pool: recent contributors keep sharing the 10% until
+they age out, and a sybil (many hotkeys / one coldkey) gets ONE share."""
+from __future__ import annotations
+
+import sys
+from pathlib import Path
+
+sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
+
+from validator.mf_pool import active_hotkeys, update_pool  # noqa: E402
+
+
+def test_recent_failure_persists_across_quiet_epochs():
+    # epoch at block 1000: hkA fails. Quiet epochs follow (no new failures).
+    pool = update_pool([], ["hkA"], block=1000)
+    # 5 blocks later, no new failures -> hkA still in the window
+    pool = update_pool(pool, [], block=1005)
+    assert active_hotkeys(pool) == ["hkA"]
+
+
+def test_ages_out_after_window():
+    pool = update_pool([], ["hkA"], block=1000)
+    # far beyond RALPH_MF_WINDOW_BLOCKS (default 7200)
+    pool = update_pool(pool, [], block=1000 + 7201)
+    assert active_hotkeys(pool) == []
+
+
+def test_sybil_deduped_by_coldkey_one_share():
+    # star-dust9023-style: 3 hotkeys, ONE coldkey, all fail -> 1 share
+    pool = update_pool([], ["hkA", "hkB", "hkC"], block=1000)
+    pool = update_pool(pool, ["hkD"], block=1001)  # different operator
+    h2c = {"hkA": "ckX", "hkB": "ckX", "hkC": "ckX", "hkD": "ckY"}
+    act = active_hotkeys(pool, h2c)
+    assert len(act) == 2  # one per coldkey, NOT 4
+    assert "hkD" in act
+    # the ckX share goes to exactly one of its hotkeys
+    assert len([h for h in act if h in ("hkA", "hkB", "hkC")]) == 1
+
+
+def test_no_coldkey_map_dedups_by_hotkey():
+    pool = update_pool([], ["hkA", "hkB"], block=1000)
+    assert sorted(active_hotkeys(pool)) == ["hkA", "hkB"]
+
+
+def test_deregistered_hotkeys_excluded():
+    pool = update_pool([], ["hkA", "hkB"], block=1000)
+    assert active_hotkeys(pool, registered={"hkA"}) == ["hkA"]
+
+
+def test_window_max_caps_size():
+    pool = []
+    for b in range(200):
+        pool = update_pool(pool, [f"hk{b}"], block=1000 + b)
+    assert len(pool) <= 128  # RALPH_MF_WINDOW_MAX default
+
+
+def test_most_recent_hotkey_wins_per_coldkey():
+    pool = update_pool([], ["hkOld"], block=1000)
+    pool = update_pool(pool, ["hkNew"], block=1050)  # same operator, newer
+    h2c = {"hkOld": "ck1", "hkNew": "ck1"}
+    assert active_hotkeys(pool, h2c) == ["hkNew"]

validator/mf_pool.py

@@ -0,0 +1,96 @@
+"""Rolling meaningful-failure reward pool (§5.6 fix).
+
+The 10% "informative dead-end" pool used to be split among the CURRENT epoch's
+meaningful failures only, so a contributor earned for one (rate-limited) weight
+window and then the 10% reverted to the king every quiet epoch — effectively no
+sustained incentive. This keeps a persisted rolling window of recent meaningful
+failures so recent contributors keep sharing the 10% until they age out.
+
+SYBIL-RESISTANT: the split is deduped by COLDKEY (one share per operator, the most
+recent hotkey per coldkey). A single operator running many hotkeys (e.g. the
+star-dust9023 6-hotkey fleet) therefore gets ONE share, not one-per-hotkey, so it
+cannot farm the pool by spamming meaningful failures across hotkeys.
+"""
+from __future__ import annotations
+
+import json
+import os
+from pathlib import Path
+
+
+def _envi(name: str, default: int) -> int:
+    try:
+        return int(os.environ.get(name, default))
+    except (TypeError, ValueError):
+        return default
+
+
+def window_blocks() -> int:
+    # ~1 day at 12s/block. A meaningful failure keeps earning for this long.
+    return _envi("RALPH_MF_WINDOW_BLOCKS", 7200)
+
+
+def window_max() -> int:
+    return _envi("RALPH_MF_WINDOW_MAX", 128)
+
+
+def _path(chain_dir) -> Path | None:
+    # chain_dir is chain.chain_dir (the `chain/` dir holding king.json etc.); may be
+    # None on a LocalChain — then the pool is stateless (empty each call).
+    return (Path(chain_dir) / "meaningful_pool.json") if chain_dir else None
+
+
+def load_pool(chain_dir) -> list[dict]:
+    p = _path(chain_dir)
+    if p is None:
+        return []
+    try:
+        d = json.loads(p.read_text())
+        return d if isinstance(d, list) else []
+    except (OSError, ValueError):
+        return []
+
+
+def save_pool(chain_dir, pool: list[dict]) -> None:
+    p = _path(chain_dir)
+    if p is None:
+        return
+    try:
+        p.parent.mkdir(parents=True, exist_ok=True)
+        tmp = p.with_suffix(".json.tmp")
+        tmp.write_text(json.dumps(pool, indent=1))
+        tmp.replace(p)
+    except OSError:
+        pass
+
+
+def update_pool(pool: list[dict], new_hotkeys, block: int, val_bpbs: dict | None = None) -> list[dict]:
+    """Append this epoch's meaningful-failure hotkeys stamped with the current block,
+    prune to the rolling window (age + max size), return newest-first."""
+    val_bpbs = val_bpbs or {}
+    out = list(pool)
+    for hk in new_hotkeys:
+        out.append({"hotkey": hk, "block": int(block), "val_bpb": val_bpbs.get(hk)})
+    cutoff = int(block) - window_blocks()
+    out = [e for e in out if int(e.get("block", 0)) >= cutoff]
+    out.sort(key=lambda e: -int(e.get("block", 0)))  # newest first
+    return out[: window_max()]
+
+
+def active_hotkeys(pool: list[dict], hotkey_to_coldkey: dict | None = None,
+                   registered: set | None = None) -> list[str]:
+    """Hotkeys currently sharing the 10% pool. Deduped by coldkey (one per operator,
+    most-recent hotkey wins) when a hotkey->coldkey map is supplied. If `registered`
+    is given, only hotkeys still on the subnet are returned (weighting a deregistered
+    hotkey wastes the share)."""
+    by_key: dict = {}
+    for e in pool:  # newest-first: first-seen per key is the most recent
+        hk = e.get("hotkey")
+        if not hk:
+            continue
+        if registered is not None and hk not in registered:
+            continue
+        key = (hotkey_to_coldkey or {}).get(hk, hk)  # unknown coldkey -> own operator
+        if key not in by_key:
+            by_key[key] = hk
+    return list(by_key.values())

validator/service.py

@@ -1244,8 +1244,28 @@ def run_epoch(
         archive_bundle(bundle_dir, queue_dir, "scored")
 
     # §5.6 pool split: 90% king, 10% to meaningful_failures (equal split).
-    # If no meaningful_failures this epoch, king gets 100%.
-    round_scores = _apply_pool_split(chain, king_change_hotkey, meaningful_failure_hotkeys)
+    # ROLLING pool: recent meaningful failures keep sharing the 10% until they age
+    # out (RALPH_MF_WINDOW_BLOCKS) so a contributor isn't dropped the instant a quiet
+    # epoch re-asserts king=100%. Deduped by COLDKEY so a sybil (many hotkeys / one
+    # coldkey) gets ONE share, not one-per-hotkey. RALPH_MF_ROLLING_OFF=1 -> per-epoch.
+    if os.environ.get("RALPH_MF_ROLLING_OFF") == "1":
+        _mf_for_split = meaningful_failure_hotkeys
+    else:
+        from validator.mf_pool import active_hotkeys, load_pool, save_pool, update_pool
+        _mfp = update_pool(load_pool(chain_dir), meaningful_failure_hotkeys, epoch_start_block)
+        save_pool(chain_dir, _mfp)
+        _h2c, _reg = {}, None
+        _mg = getattr(chain, "metagraph", None)
+        if _mg is not None:
+            try:
+                _h2c = {h: c for h, c in zip(list(_mg.hotkeys), list(_mg.coldkeys))}
+                _reg = set(_mg.hotkeys)
+            except Exception:  # noqa: BLE001
+                _h2c, _reg = {}, None
+        _mf_for_split = active_hotkeys(_mfp, _h2c, _reg)
+        if _mf_for_split:
+            log_info(f"meaningful-failure pool: {len(_mf_for_split)} operator(s) share 10% (rolling window)")
+    round_scores = _apply_pool_split(chain, king_change_hotkey, _mf_for_split)
 
     # Merge any weights recovered from a previous rate-limited epoch so we don't
     # lose an unlanded meaningful_failure credit — but NEVER resurrect a king the