From ba8a878db9a78d116c245f3fd1a27fb57dff2478 Mon Sep 17 00:00:00 2001 From: Bitzy Date: Wed, 24 Jun 2026 11:48:29 +0000 Subject: [PATCH] validator: optional wandb telemetry per epoch - new validator/telemetry.py (opt-in via RALPH_WANDB=1) - logs per-epoch: submissions/accepted/rejected/meaningful_failures, king val_bpb + benchmark, standing uid/stake/vtrust/vpermit, block - single resumed run (fixed id) so restarts continue history; auto-step so a reset epoch counter can't break wandb - fully best-effort: missing/unauthed/erroring wandb disables it, never disrupts validation - wired init/log_epoch/finish into service.main loop - tests for the disabled-by-default + never-raise contract Co-Authored-By: Claude Opus 4.8 (1M context) --- tests/test_telemetry.py | 44 ++++++++++++++ validator/service.py | 7 +++ validator/telemetry.py | 125 ++++++++++++++++++++++++++++++++++++++++ 3 files changed, 176 insertions(+) create mode 100644 tests/test_telemetry.py create mode 100644 validator/telemetry.py diff --git a/tests/test_telemetry.py b/tests/test_telemetry.py new file mode 100644 index 0000000..072e14d --- /dev/null +++ b/tests/test_telemetry.py @@ -0,0 +1,44 @@ +"""Tests for the opt-in wandb validator telemetry — focus on the defensive +contract: disabled by default, and never raises regardless of chain shape.""" + +import validator.telemetry as telemetry + + +def test_disabled_by_default(monkeypatch): + monkeypatch.delenv("RALPH_WANDB", raising=False) + assert telemetry.init(netuid=40, epoch_seconds=120) is False + + +def test_log_epoch_noop_when_disabled(): + # No run initialised → must be a silent no-op even with a None chain. + telemetry._enabled = False + telemetry._run = None + telemetry.log_epoch(None, 1, {"submissions": 7, "accepted": 0, "rejected": 7}) # no raise + + +def test_standing_best_effort_on_bad_chain(): + class Bad: + pass + + assert telemetry._standing(Bad()) == {} + + +def test_standing_reads_metagraph(): + class MG: + hotkeys = ["aa", "bb", "cc"] + S = {1: 296.0} + Tv = {1: 0.5} + validator_permit = {1: True} + + class Chain: + metagraph = MG() + + class wallet: # noqa: N801 + class hotkey: # noqa: N801 + ss58_address = "bb" + + out = telemetry._standing(Chain()) + assert out["standing/uid"] == 1 + assert out["standing/stake"] == 296.0 + assert out["standing/vtrust"] == 0.5 + assert out["standing/vpermit"] == 1 diff --git a/validator/service.py b/validator/service.py index 4ef9cc3..5184c5a 100644 --- a/validator/service.py +++ b/validator/service.py @@ -44,6 +44,7 @@ from chain_layer.config import get_chain from validator.hf_poller import DEFAULT_REPO as DEFAULT_HF_REPO from validator.hf_poller import poll_hub +from validator import telemetry from validator.scoring import score_bundle from validator.validator import judge_submission @@ -1167,11 +1168,14 @@ def main(): log_info(f"submit bundles to: {args.queue_dir / 'pending' / '/'}") log_info("") + telemetry.init(netuid=int(os.environ.get("BT_NETUID", "40")), epoch_seconds=args.epoch_seconds) + epoch = 0 while not SHUTDOWN: epoch += 1 log_info(f"--- epoch {epoch} ---") + result = {"submissions": 0, "accepted": 0, "rejected": 0} try: result = run_epoch( chain, @@ -1203,6 +1207,8 @@ def main(): pass _log_validator_standing(chain) + telemetry.log_epoch(chain, epoch, result) + if args.once: break @@ -1212,6 +1218,7 @@ def main(): break time.sleep(1) + telemetry.finish() log_info("validator service stopped") diff --git a/validator/telemetry.py b/validator/telemetry.py new file mode 100644 index 0000000..54583c5 --- /dev/null +++ b/validator/telemetry.py @@ -0,0 +1,125 @@ +"""Optional Weights & Biases telemetry for the validator. + +Opt-in via ``RALPH_WANDB=1`` (needs ``WANDB_API_KEY`` in the environment). Logs +one wandb point per epoch: submission counts, the current king's val_bpb, and +this validator's on-chain standing (uid / stake / vTrust / permit / block). + +Entirely best-effort — if wandb is missing, unauthenticated, or errors, the +telemetry silently disables itself and NEVER disrupts validation. A single +resumed run (fixed id) is used so restarts continue the same history; steps are +auto-incremented (block is logged as a metric for use as the dashboard x-axis) +so an epoch counter that resets on restart can't break wandb's monotonic step. +""" + +from __future__ import annotations + +import os + +_run = None +_enabled = False + + +def _truthy(v: str | None) -> bool: + return str(v).strip().lower() in {"1", "true", "yes", "on"} + + +def init(netuid: int, epoch_seconds: int) -> bool: + """Start (or resume) the wandb run. Returns True if telemetry is active.""" + global _run, _enabled + if not _truthy(os.environ.get("RALPH_WANDB", "0")): + return False + try: + import wandb + + os.environ.setdefault("WANDB_SILENT", "true") + run_id = os.environ.get("RALPH_WANDB_RUN") or f"validator-sn{netuid}" + _run = wandb.init( + project=os.environ.get("WANDB_PROJECT", "ralph-validator"), + entity=os.environ.get("WANDB_ENTITY") or None, + name=run_id, + id=run_id, + resume="allow", + config={"netuid": netuid, "epoch_seconds": epoch_seconds}, + ) + _enabled = True + print(f"[telemetry] wandb enabled: {getattr(_run, 'url', '?')}", flush=True) + except Exception as e: # noqa: BLE001 — telemetry must never break the validator + _enabled = False + print(f"[telemetry] wandb disabled ({type(e).__name__}: {e})", flush=True) + return _enabled + + +def _standing(chain) -> dict: + """This validator's on-chain standing, best-effort (empty dict on any miss).""" + out: dict = {} + try: + mg = getattr(chain, "metagraph", None) + wallet = getattr(chain, "wallet", None) + if mg is None or wallet is None: + return out + ss58 = wallet.hotkey.ss58_address + hotkeys = list(getattr(mg, "hotkeys", [])) + if ss58 not in hotkeys: + return out + uid = hotkeys.index(ss58) + out["standing/uid"] = uid + try: + out["standing/stake"] = float(mg.S[uid]) + except Exception: + pass + for attr in ("Tv", "validator_trust"): + try: + out["standing/vtrust"] = float(getattr(mg, attr)[uid]) + break + except Exception: + continue + try: + out["standing/vpermit"] = int(bool(mg.validator_permit[uid])) + except Exception: + pass + except Exception: + pass + return out + + +def log_epoch(chain, epoch: int, result: dict) -> None: + """Log one epoch's metrics. No-op unless telemetry is active.""" + if not _enabled or _run is None: + return + try: + import wandb + + m = { + "epoch": epoch, + "epoch/submissions": result.get("submissions", 0), + "epoch/accepted": result.get("accepted", 0), + "epoch/rejected": result.get("rejected", 0), + "epoch/meaningful_failures": result.get("meaningful_failures", 0), + } + try: + m["chain/block"] = chain.get_current_block() + except Exception: + pass + try: + king = chain.get_king() + if king is not None: + m["king/val_bpb"] = float(king.val_bpb) + m["king/benchmark_accuracy"] = float(getattr(king, "benchmark_accuracy", 0.0) or 0.0) + except Exception: + pass + m.update(_standing(chain)) + wandb.log(m) + except Exception as e: # noqa: BLE001 + print(f"[telemetry] log_epoch failed (non-fatal): {e}", flush=True) + + +def finish() -> None: + global _run + if _run is not None: + try: + import wandb + + wandb.finish() + except Exception: + pass + _run = None