From 161f6a273c721ab2f759c375b7b82c13e7a7856c Mon Sep 17 00:00:00 2001 From: Bitzy Date: Sat, 20 Jun 2026 12:14:03 +0000 Subject: [PATCH] feat: auditor sets weights on a block cadence - counter-weight is now periodic (~300 blocks), not one-shot per report: a validator must keep setting weights every epoch or vTrust decays - chain.py: get_current_block + blocks_since_weight_set (hotkey -> uid -> blocks_since_last_update) - weights.py: is_weight_set_due + weight_set_interval_blocks (env AUDITOR_WEIGHT_INTERVAL_BLOCKS, default 300) + auditor_hotkey_ss58 - main.py: maybe_counter_weight re-sets from the latest CLEAN epoch's replayed scores when due; runs every --once/--loop tick, decoupled from whether a new report dropped - default 300 blocks clears the subnet weights_rate_limit (~100) - 25 tests: pure cadence decision + orchestration flow (no chain/wallet) --- auditor/chain.py | 25 +++++ auditor/main.py | 114 +++++++++++++++++----- auditor/weights.py | 56 ++++++++++- tests/test_auditor_weights.py | 173 ++++++++++++++++++++++++++++++++++ 4 files changed, 342 insertions(+), 26 deletions(-) create mode 100644 tests/test_auditor_weights.py diff --git a/auditor/chain.py b/auditor/chain.py index a8f8b24..179ada9 100644 --- a/auditor/chain.py +++ b/auditor/chain.py @@ -83,6 +83,31 @@ def get_commitment_hash(self, at_block: int, hotkey: str | None = None) -> str | decoded = _decode_commitment(raw) return _normalize_hex(decoded) if decoded else None + def get_current_block(self) -> int: + """Current chain-head block (drives the counter-weight cadence).""" + return int(self._connect().get_current_block()) + + def blocks_since_weight_set(self, hotkey: str) -> int | None: + """How many blocks since `hotkey` last set weights on this netuid. + + Resolves hotkey -> uid -> `blocks_since_last_update` (= current_block - + LastUpdate[uid]). Returns None if the hotkey isn't registered or the + query fails — the caller treats None as "due / unknown" so a brand-new + (never-set) auditor sets weights on its first cadence tick. + """ + sub = self._connect() + try: + uid = sub.get_uid_for_hotkey_on_subnet(hotkey, self.netuid) + except Exception: + return None + if uid is None: + return None + try: + blocks = sub.blocks_since_last_update(self.netuid, int(uid)) + except Exception: + return None + return None if blocks is None else int(blocks) + def close(self) -> None: sub = self._subtensor if sub is not None: diff --git a/auditor/main.py b/auditor/main.py index dd9e44f..875132c 100644 --- a/auditor/main.py +++ b/auditor/main.py @@ -27,6 +27,12 @@ HF_TOKEN only for a private report repo (public needs none) AUDIT_INTERVAL_SECONDS --loop sleep (default 300) AUDITOR_SET_WEIGHTS_ENABLED opt-in counter-weight (default off) + AUDITOR_WEIGHT_INTERVAL_BLOCKS re-set weights every N blocks (default 300) + +Counter-weight (when enabled) runs on a BLOCK cadence, not per-report: each pass +reads how many blocks since the auditor's hotkey last set weights and re-sets +from the latest clean epoch's replayed scores once ~300 blocks have elapsed, so +the auditor-validator keeps its weights (and vTrust) fresh every epoch. """ from __future__ import annotations @@ -60,6 +66,7 @@ STATE_FILE = Path(".audit_state") PUBLISHED_FILE = Path(".audit_published") +LAST_CLEAN_EPOCH_FILE = Path(".audit_last_clean_epoch") # epoch_id of the most recent clean epoch def _read_int_file(path: Path) -> int | None: @@ -75,6 +82,13 @@ def _write_int_file(path: Path, value: int) -> None: path.write_text(str(value)) +def _read_str_file(path: Path) -> str | None: + if not path.exists(): + return None + val = path.read_text().strip() + return val or None + + def _setup_logging(verbose: bool) -> None: logging.basicConfig( level=logging.DEBUG if verbose else logging.INFO, @@ -145,13 +159,12 @@ def audit_epoch(epoch_id: str, chain: ChainClient, api: ReportClient) -> int: def audit_new_epochs(chain: ChainClient, api: ReportClient) -> int: """Audit every epoch newer than the local watermark. Returns the worst code. - On a clean epoch the `.audit_state` watermark advances. If - AUDITOR_SET_WEIGHTS_ENABLED, also counter-weights from the most recent clean - epoch's replayed scores using the auditor's OWN wallet. + On a clean epoch the `.audit_state` watermark advances and the epoch_id is + recorded as the latest-clean epoch (consumed by `maybe_counter_weight`). + Weight-setting is decoupled from this pass — it runs on a block cadence (see + `maybe_counter_weight`) so the auditor keeps setting weights every epoch even + when no new report dropped this cycle. """ - from auditor.weights import is_enabled as cw_enabled - from auditor.weights import submit_weights - last_audited = _read_int_file(STATE_FILE) try: reports = api.list_reports() @@ -161,8 +174,6 @@ def audit_new_epochs(chain: ChainClient, api: ReportClient) -> int: sorted_reports = sorted(reports, key=lambda r: r.get("epoch_end_block") or 0) worst = EXIT_CLEAN - last_clean_epoch_id: str | None = None - last_clean_end_block: int | None = None for r in sorted_reports: end_block = r.get("epoch_end_block") @@ -172,26 +183,76 @@ def audit_new_epochs(chain: ChainClient, api: ReportClient) -> int: worst = max(worst, code) if code == EXIT_CLEAN and end_block is not None: _write_int_file(STATE_FILE, end_block) - last_clean_epoch_id = r["epoch_id"] - last_clean_end_block = end_block - - if last_clean_epoch_id and cw_enabled(): - try: - envelope = api.get_report(last_clean_epoch_id) - replayed = replay_scoring(envelope.get("report_json") or {}) - ok = submit_weights( - subtensor_url=chain.subtensor_url, - netuid=chain.netuid, - weights_by_hotkey=replayed, - ) - if ok and last_clean_end_block is not None: - _write_int_file(PUBLISHED_FILE, last_clean_end_block) - except Exception: - logger.exception("counter-weight step failed for %s", last_clean_epoch_id) + LAST_CLEAN_EPOCH_FILE.write_text(str(r["epoch_id"])) return worst +def maybe_counter_weight(chain: ChainClient, api: ReportClient) -> None: + """Set the auditor's OWN weights on a block cadence (off unless enabled). + + Continuous epoch-cadence process, NOT one-shot: each call reads how many + blocks since the auditor last set weights and, if at least + `weight_set_interval_blocks` have elapsed (≈300 blocks ≈ 1h), re-sets weights + from the latest CLEAN epoch's independently-replayed scores — shadowing the + honest validator and keeping the auditor's own vTrust from decaying. Re-uses + the last clean scores when no new report appeared this cycle. Never raises + into the loop. + """ + from auditor.weights import ( + auditor_hotkey_ss58, + is_weight_set_due, + submit_weights, + weight_set_interval_blocks, + ) + from auditor.weights import ( + is_enabled as cw_enabled, + ) + + if not cw_enabled(): + return + hotkey = auditor_hotkey_ss58() + if hotkey is None: + return # no wallet configured → stay read-only + + interval = weight_set_interval_blocks() + try: + blocks_since = chain.blocks_since_weight_set(hotkey) + current = chain.get_current_block() + except Exception: + logger.exception("counter-weight: chain query failed; skipping this tick") + return + + if not is_weight_set_due(blocks_since, interval): + logger.info( + "counter-weight: %s/%s blocks since last set (block %s) — not due", + blocks_since, interval, current, + ) + return + + epoch_id = _read_str_file(LAST_CLEAN_EPOCH_FILE) + if not epoch_id: + logger.info("counter-weight: due but no clean epoch audited yet — skipping") + return + + logger.info( + "counter-weight: due (%s≥%s blocks since last set, block %s) — setting from %s", + blocks_since, interval, current, epoch_id, + ) + try: + envelope = api.get_report(epoch_id) + replayed = replay_scoring(envelope.get("report_json") or {}) + ok = submit_weights( + subtensor_url=chain.subtensor_url, + netuid=chain.netuid, + weights_by_hotkey=replayed, + ) + if ok: + _write_int_file(PUBLISHED_FILE, current) + except Exception: + logger.exception("counter-weight step failed for %s", epoch_id) + + def main(argv: list[str] | None = None) -> None: p = argparse.ArgumentParser( prog="auditor", @@ -237,12 +298,15 @@ def main(argv: list[str] | None = None) -> None: while True: try: audit_new_epochs(chain, api) + maybe_counter_weight(chain, api) # block-cadence weight-set each tick except Exception: logger.exception("audit loop iteration failed") time.sleep(interval) # default (and --once) -> single pass. - sys.exit(audit_new_epochs(chain, api)) + code = audit_new_epochs(chain, api) + maybe_counter_weight(chain, api) + sys.exit(code) if __name__ == "__main__": diff --git a/auditor/weights.py b/auditor/weights.py index ed9c6b7..f0dae5a 100644 --- a/auditor/weights.py +++ b/auditor/weights.py @@ -23,6 +23,12 @@ logger = logging.getLogger("ralph-auditor.weights") +# Counter-weight cadence: a validator must keep setting weights every epoch or +# its weights go stale and vTrust decays. The subnet enforces a minimum gap +# (weights_rate_limit ≈ 100 blocks on netuid 40); we default comfortably above it. +DEFAULT_WEIGHT_SET_INTERVAL_BLOCKS = 300 # ≈ 1h at 12s/block + + def is_enabled() -> bool: return os.environ.get("AUDITOR_SET_WEIGHTS_ENABLED", "false").strip().lower() in { "1", @@ -32,6 +38,47 @@ def is_enabled() -> bool: } +def weight_set_interval_blocks() -> int: + """Blocks between counter-weight sets (env AUDITOR_WEIGHT_INTERVAL_BLOCKS, + default 300). Invalid/non-positive values fall back to the default.""" + raw = os.environ.get("AUDITOR_WEIGHT_INTERVAL_BLOCKS", "").strip() + if raw: + try: + v = int(raw) + if v > 0: + return v + except ValueError: + pass + return DEFAULT_WEIGHT_SET_INTERVAL_BLOCKS + + +def is_weight_set_due(blocks_since: int | None, interval_blocks: int) -> bool: + """True if the auditor should (re)set weights now. + + `blocks_since` is blocks elapsed since the auditor's hotkey last set weights + (None = never set / unknown → due). Due once at least `interval_blocks` have + elapsed. This is what makes counter-weighting a continuous epoch-cadence + process rather than a one-shot. + """ + if interval_blocks <= 0: + raise ValueError(f"interval_blocks must be > 0; got {interval_blocks}") + if blocks_since is None: + return True + return blocks_since >= interval_blocks + + +def auditor_hotkey_ss58() -> str | None: + """The auditor's OWN hotkey ss58, read-only, for cadence queries + (blocks-since-last-weight-set). None if no wallet is configured.""" + wallet = _load_wallet() + if wallet is None: + return None + try: + return wallet.hotkey.ss58_address + except Exception: + return None + + def _load_wallet(): """Load the auditor's OWN bittensor wallet from env-supplied NAMES. @@ -133,4 +180,11 @@ def submit_weights( pass -__all__ = ["is_enabled", "submit_weights"] +__all__ = [ + "DEFAULT_WEIGHT_SET_INTERVAL_BLOCKS", + "auditor_hotkey_ss58", + "is_enabled", + "is_weight_set_due", + "submit_weights", + "weight_set_interval_blocks", +] diff --git a/tests/test_auditor_weights.py b/tests/test_auditor_weights.py new file mode 100644 index 0000000..ee5822a --- /dev/null +++ b/tests/test_auditor_weights.py @@ -0,0 +1,173 @@ +"""Counter-weight cadence tests — the block-cadence weight-setting that makes +the auditor a continuous epoch-cadence validator, not a one-shot. + +Pure logic only (is_weight_set_due / interval / due decision); no chain, no +wallet, no torch. The submit_weights extrinsic itself needs a live subtensor + +wallet and is exercised operationally, not here. +""" +from __future__ import annotations + +import sys +from pathlib import Path + +import pytest + +sys.path.insert(0, str(Path(__file__).resolve().parent.parent)) + +import ralph_bootstrap # noqa: F401 +from auditor.weights import ( + DEFAULT_WEIGHT_SET_INTERVAL_BLOCKS, + is_enabled, + is_weight_set_due, + weight_set_interval_blocks, +) + + +# --- is_weight_set_due --------------------------------------------------- +def test_due_when_never_set(): + # None = never set / unknown → set on the first tick + assert is_weight_set_due(None, 300) is True + + +def test_not_due_before_interval(): + assert is_weight_set_due(0, 300) is False + assert is_weight_set_due(299, 300) is False + + +def test_due_at_and_after_interval(): + assert is_weight_set_due(300, 300) is True + assert is_weight_set_due(450, 300) is True + + +def test_due_nonpositive_interval_raises(): + with pytest.raises(ValueError, match="interval_blocks must be > 0"): + is_weight_set_due(500, 0) + + +# --- weight_set_interval_blocks (env) ------------------------------------ +def test_interval_default(monkeypatch): + monkeypatch.delenv("AUDITOR_WEIGHT_INTERVAL_BLOCKS", raising=False) + assert weight_set_interval_blocks() == DEFAULT_WEIGHT_SET_INTERVAL_BLOCKS == 300 + + +def test_interval_env_override(monkeypatch): + monkeypatch.setenv("AUDITOR_WEIGHT_INTERVAL_BLOCKS", "120") + assert weight_set_interval_blocks() == 120 + + +@pytest.mark.parametrize("bad", ["0", "-5", "abc", ""]) +def test_interval_bad_env_falls_back(monkeypatch, bad): + monkeypatch.setenv("AUDITOR_WEIGHT_INTERVAL_BLOCKS", bad) + assert weight_set_interval_blocks() == DEFAULT_WEIGHT_SET_INTERVAL_BLOCKS + + +def test_default_interval_above_subnet_rate_limit(): + # netuid 40 weights_rate_limit is ~100 blocks; the default must clear it so + # cadence sets are never rejected for setting too often. + assert DEFAULT_WEIGHT_SET_INTERVAL_BLOCKS > 100 + + +# --- is_enabled (opt-in gate) -------------------------------------------- +@pytest.mark.parametrize("val,expected", [ + ("1", True), ("true", True), ("YES", True), ("on", True), + ("0", False), ("false", False), ("", False), +]) +def test_is_enabled(monkeypatch, val, expected): + monkeypatch.setenv("AUDITOR_SET_WEIGHTS_ENABLED", val) + assert is_enabled() is expected + + +def test_is_enabled_default_off(monkeypatch): + monkeypatch.delenv("AUDITOR_SET_WEIGHTS_ENABLED", raising=False) + assert is_enabled() is False + + +# --- maybe_counter_weight orchestration (cadence flow) ------------------- +class _FakeChain: + subtensor_url = "ws://x" + netuid = 40 + + def __init__(self, blocks_since, current=1000): + self._bs = blocks_since + self._cur = current + + def blocks_since_weight_set(self, hotkey): + return self._bs + + def get_current_block(self): + return self._cur + + +class _FakeApi: + def __init__(self): + self.fetched = [] + + def get_report(self, epoch_id): + self.fetched.append(epoch_id) + return {"report_json": {"epoch_id": epoch_id}} + + +def _wire(monkeypatch, tmp_path, *, enabled, hotkey, clean_epoch): + import auditor.main as m + import auditor.weights as w + + monkeypatch.setattr(w, "is_enabled", lambda: enabled) + monkeypatch.setattr(w, "auditor_hotkey_ss58", lambda: hotkey) + submitted = {} + monkeypatch.setattr(w, "submit_weights", + lambda **kw: submitted.update(kw) or True) + monkeypatch.setattr(m, "replay_scoring", lambda rj: {"5Fminer": 1.0}) + clean = tmp_path / "clean" + if clean_epoch is not None: + clean.write_text(clean_epoch) + monkeypatch.setattr(m, "LAST_CLEAN_EPOCH_FILE", clean) + monkeypatch.setattr(m, "PUBLISHED_FILE", tmp_path / "pub") + return m, submitted + + +def test_counter_weight_sets_when_due(monkeypatch, tmp_path): + m, submitted = _wire(monkeypatch, tmp_path, enabled=True, + hotkey="5Faudit", clean_epoch="40-123") + api = _FakeApi() + m.maybe_counter_weight(_FakeChain(blocks_since=300, current=1000), api) + assert api.fetched == ["40-123"] + assert submitted["weights_by_hotkey"] == {"5Fminer": 1.0} + assert submitted["netuid"] == 40 + assert (tmp_path / "pub").read_text() == "1000" # records the block we set at + + +def test_counter_weight_sets_when_never_set(monkeypatch, tmp_path): + m, submitted = _wire(monkeypatch, tmp_path, enabled=True, + hotkey="5Faudit", clean_epoch="40-9") + m.maybe_counter_weight(_FakeChain(blocks_since=None), _FakeApi()) + assert submitted # None blocks_since → due + + +def test_counter_weight_skips_when_not_due(monkeypatch, tmp_path): + m, submitted = _wire(monkeypatch, tmp_path, enabled=True, + hotkey="5Faudit", clean_epoch="40-123") + api = _FakeApi() + m.maybe_counter_weight(_FakeChain(blocks_since=50), api) # < 300 + assert not submitted and api.fetched == [] + + +def test_counter_weight_noop_when_disabled(monkeypatch, tmp_path): + m, submitted = _wire(monkeypatch, tmp_path, enabled=False, + hotkey="5Faudit", clean_epoch="40-123") + m.maybe_counter_weight(_FakeChain(blocks_since=500), _FakeApi()) + assert not submitted + + +def test_counter_weight_skips_without_wallet(monkeypatch, tmp_path): + m, submitted = _wire(monkeypatch, tmp_path, enabled=True, + hotkey=None, clean_epoch="40-123") + m.maybe_counter_weight(_FakeChain(blocks_since=500), _FakeApi()) + assert not submitted + + +def test_counter_weight_due_but_no_clean_epoch(monkeypatch, tmp_path): + m, submitted = _wire(monkeypatch, tmp_path, enabled=True, + hotkey="5Faudit", clean_epoch=None) + api = _FakeApi() + m.maybe_counter_weight(_FakeChain(blocks_since=500), api) + assert not submitted and api.fetched == []