From c5d35112105f5a3f2b1061c923543c9ecca5450d Mon Sep 17 00:00:00 2001 From: Landyn Date: Fri, 24 Jul 2026 15:34:57 -0500 Subject: [PATCH 1/2] Add validator-voted repo weight consensus via commitments --- README.md | 4 + gittensor/utils/config.py | 14 ++ gittensor/validator/forward.py | 6 +- gittensor/validator/utils/config.py | 10 ++ gittensor/validator/utils/storage.py | 51 +++++- .../validator/weight_consensus/__init__.py | 7 + gittensor/validator/weight_consensus/chain.py | 87 ++++++++++ gittensor/validator/weight_consensus/codec.py | 103 ++++++++++++ .../validator/weight_consensus/consensus.py | 114 +++++++++++++ .../validator/weight_consensus/manager.py | 157 ++++++++++++++++++ .../validator/weight_consensus/publisher.py | 48 ++++++ neurons/validator.py | 29 +++- tests/validator/test_consensus_aggregate.py | 144 ++++++++++++++++ tests/validator/test_consensus_apply.py | 41 +++++ tests/validator/test_consensus_codec.py | 85 ++++++++++ tests/validator/test_consensus_manager.py | 154 +++++++++++++++++ tests/validator/test_consensus_publisher.py | 95 +++++++++++ 17 files changed, 1145 insertions(+), 4 deletions(-) create mode 100644 gittensor/validator/weight_consensus/__init__.py create mode 100644 gittensor/validator/weight_consensus/chain.py create mode 100644 gittensor/validator/weight_consensus/codec.py create mode 100644 gittensor/validator/weight_consensus/consensus.py create mode 100644 gittensor/validator/weight_consensus/manager.py create mode 100644 gittensor/validator/weight_consensus/publisher.py create mode 100644 tests/validator/test_consensus_aggregate.py create mode 100644 tests/validator/test_consensus_apply.py create mode 100644 tests/validator/test_consensus_codec.py create mode 100644 tests/validator/test_consensus_manager.py create mode 100644 tests/validator/test_consensus_publisher.py diff --git a/README.md b/README.md index 0225c89fa..fb54bbee5 100644 --- a/README.md +++ b/README.md @@ -71,6 +71,10 @@ docker-compose -f docker-compose.vali.yml up -d See full guide **[here](https://docs.gittensor.io/validator.html)** +### Repository Weight Consensus + +Repository emission shares are voted by validators, not hardcoded: each validator publishes a basket of up to 10 repos to the chain (commitments pallet), and every validator applies the stake-weighted mean of all eligible baskets (vpermit + 30k alpha), aggregated at a fixed snapshot block twice a day, so all validators derive identical weights. Validators without a basket run on the aggregate with a warning. Recommended: connect to an archive endpoint (`wss://archive.chain.opentensor.ai:443`) so snapshot state is always queryable; lite nodes gracefully fall back to the last-good aggregate. + ## Reward Algorithm ### Important Structures diff --git a/gittensor/utils/config.py b/gittensor/utils/config.py index a7a4d9892..e41ffab33 100644 --- a/gittensor/utils/config.py +++ b/gittensor/utils/config.py @@ -140,6 +140,20 @@ def add_validator_args(cls, parser): default=4096, ) + parser.add_argument( + '--neuron.consensus_prefs_path', + type=str, + help='Path to the repo weight preferences JSON voted on chain. Defaults to /repo_weight_prefs.json.', + default=None, + ) + + parser.add_argument( + '--neuron.disable_weight_consensus', + action='store_true', + help='Disables validator-voted repo weight consensus; scoring uses the baked-in repository weights.', + default=False, + ) + parser.add_argument( '--wandb.project_name', type=str, diff --git a/gittensor/validator/forward.py b/gittensor/validator/forward.py index 1347b0bff..7fca228ef 100644 --- a/gittensor/validator/forward.py +++ b/gittensor/validator/forward.py @@ -23,6 +23,7 @@ load_programming_language_weights, load_token_config, ) +from gittensor.validator.weight_consensus import run_weight_consensus if TYPE_CHECKING: from neurons.validator import Validator @@ -40,14 +41,17 @@ async def forward(self: 'Validator') -> None: Emission blending: - Combined scoring pool: 90%, allocated by repository emission_share + (validator-voted consensus aggregate when active, baked-in weights otherwise) - Maintainer cut: per-repo carve-out routed to maintainer miner neurons - Issue treasury: 10%, flat to UID 111 - Recycle: registry slack and inactive repo slices to UID 0 """ + self.consensus_manager.maybe_refresh(self.subtensor, self.block) + if self.step % VALIDATOR_STEPS_INTERVAL == 0: miner_uids = get_all_uids(self) - master_repositories = load_master_repo_weights() + master_repositories = run_weight_consensus(self, load_master_repo_weights()) programming_languages = load_programming_language_weights() token_config = load_token_config() diff --git a/gittensor/validator/utils/config.py b/gittensor/validator/utils/config.py index 39d11f755..a7fb53b26 100644 --- a/gittensor/validator/utils/config.py +++ b/gittensor/validator/utils/config.py @@ -5,6 +5,16 @@ VALIDATOR_WAIT = 60 # 60 seconds VALIDATOR_STEPS_INTERVAL = 120 # 2 hours, every time a scoring round happens +# Weight consensus: validator-voted repository emission shares via chain commitments. +# Tunables below are provisional team defaults — each is a one-line change. +CONSENSUS_SNAPSHOT_INTERVAL_BLOCKS = 3600 # ~12h @ 12s/block, aggregate recomputed 2x/day +CONSENSUS_MIN_VALIDATOR_STAKE_RAO = 30_000 * 10**9 # 30k alpha voter threshold +CONSENSUS_MAX_REPOS = 10 # max repos per validator basket +CONSENSUS_MAX_PAYLOAD_BYTES = 512 # chain BigRaw field limit +CONSENSUS_FRESH_WINDOW_BLOCKS = 200 # lite nodes prune state ~256 blocks back +CONSENSUS_CACHE_KEEP = 8 # snapshots retained in the disk cache +CONSENSUS_WEIGHT_PRECISION = 10**12 # fixed-point scale for deterministic int math + # required env vars GITTENSOR_VALIDATOR_PAT = os.getenv('GITTENSOR_VALIDATOR_PAT') WANDB_API_KEY = os.getenv('WANDB_API_KEY') diff --git a/gittensor/validator/utils/storage.py b/gittensor/validator/utils/storage.py index ded4bf0ec..0a8748080 100644 --- a/gittensor/validator/utils/storage.py +++ b/gittensor/validator/utils/storage.py @@ -1,14 +1,18 @@ +import json from dataclasses import dataclass from datetime import datetime, timezone -from typing import Dict, List +from typing import Dict, List, Tuple import bittensor as bt from gittensor.classes import Miner, MinerEvaluation from gittensor.validator.storage.database import create_database_connection from gittensor.validator.storage.repository import Repository +from gittensor.validator.utils.config import CONSENSUS_SNAPSHOT_INTERVAL_BLOCKS from gittensor.validator.utils.load_weights import RepositoryConfig +WEIGHT_CONSENSUS_SNAPSHOT_RETENTION = 30 # snapshots kept for dashboard history / mirror hysteresis + @dataclass class StorageResult: @@ -28,6 +32,51 @@ def __init__(self): def is_enabled(self) -> bool: return self.db_connection is not None + def store_weight_consensus( + self, snapshot_block: int, baskets: List[Tuple[str, int, Dict[str, int]]], result + ) -> None: + """Persist one snapshot's validator baskets + aggregate shares. + + baskets: (hotkey, stake_rao, decoded prefs) per eligible voter. Rows are + replaced per snapshot and pruned past the retention window. Failures + are the caller's to swallow — consensus never depends on this. + """ + if not self.is_enabled(): + return + assert self.db_connection is not None + + cutoff = snapshot_block - WEIGHT_CONSENSUS_SNAPSHOT_RETENTION * CONSENSUS_SNAPSHOT_INTERVAL_BLOCKS + try: + with self.db_connection.cursor() as cur: + cur.execute( + 'DELETE FROM validator_weight_baskets WHERE snapshot_block = %s OR snapshot_block < %s', + (snapshot_block, cutoff), + ) + cur.executemany( + 'INSERT INTO validator_weight_baskets (snapshot_block, hotkey, stake_rao, basket) VALUES (%s, %s, %s, %s)', + [(snapshot_block, hotkey, stake_rao, json.dumps(prefs)) for hotkey, stake_rao, prefs in baskets], + ) + cur.execute( + 'DELETE FROM repo_weight_consensus WHERE snapshot_block = %s OR snapshot_block < %s', + (snapshot_block, cutoff), + ) + cur.execute( + 'INSERT INTO repo_weight_consensus (snapshot_block, gate_passed, shares, eligible_stake_rao, valid_stake_rao, voter_count) ' + 'VALUES (%s, %s, %s, %s, %s, %s)', + ( + snapshot_block, + result.shares is not None, + json.dumps(result.shares or {}), + result.eligible_stake_rao, + result.valid_stake_rao, + result.voter_count, + ), + ) + self.db_connection.commit() + except Exception: + self.db_connection.rollback() + raise + def store_evaluation( self, miner_eval: MinerEvaluation, master_repositories: Dict[str, RepositoryConfig] ) -> StorageResult: diff --git a/gittensor/validator/weight_consensus/__init__.py b/gittensor/validator/weight_consensus/__init__.py new file mode 100644 index 000000000..57d099bdd --- /dev/null +++ b/gittensor/validator/weight_consensus/__init__.py @@ -0,0 +1,7 @@ +# The MIT License (MIT) +# Copyright © 2025 Entrius + +from gittensor.validator.weight_consensus.consensus import apply_consensus +from gittensor.validator.weight_consensus.manager import ConsensusManager, run_weight_consensus + +__all__ = ['ConsensusManager', 'apply_consensus', 'run_weight_consensus'] diff --git a/gittensor/validator/weight_consensus/chain.py b/gittensor/validator/weight_consensus/chain.py new file mode 100644 index 000000000..55cf044b2 --- /dev/null +++ b/gittensor/validator/weight_consensus/chain.py @@ -0,0 +1,87 @@ +# The MIT License (MIT) +# Copyright © 2025 Entrius +"""Raw chain access for commitment payloads. + +The SDK's convenience readers (``get_all_commitments`` / ``decode_metadata``) +decode fields as utf-8 with ``errors='ignore'`` and silently corrupt binary +payloads, so reads go through ``query_map``/``query`` and the bytes are +extracted here. Writes use the low-level pallet builder because the high-level +``set_commitment`` only supports Raw0-128 strings. +""" + +from typing import Any, Dict, List, Optional, cast + +import bittensor as bt +from bittensor.core.extrinsics.pallets.commitments import Commitments + +from gittensor.validator.weight_consensus.codec import decode_prefs + + +def _to_bytes(value: Any) -> Optional[bytes]: + if isinstance(value, (bytes, bytearray)): + return bytes(value) + if isinstance(value, str) and value.startswith('0x'): + try: + return bytes.fromhex(value[2:]) + except ValueError: + return None + if isinstance(value, (list, tuple)) and all(isinstance(b, int) and 0 <= b <= 255 for b in value): + return bytes(value) + return None + + +def extract_payload_candidates(commitment_value: Any) -> List[bytes]: + """Collect every Raw/BigRaw field's bytes from a decoded CommitmentOf value, + tolerating the varying nesting SCALE decoding produces.""" + candidates: List[bytes] = [] + + def walk(node: Any) -> None: + if isinstance(node, dict): + for key, value in node.items(): + if isinstance(key, str) and (key == 'BigRaw' or key.startswith('Raw')): + payload = _to_bytes(value) + if payload is not None: + candidates.append(payload) + else: + walk(value) + elif isinstance(node, (list, tuple)): + for item in node: + walk(item) + + walk(getattr(commitment_value, 'value', commitment_value)) + return candidates + + +def extract_prefs(commitment_value: Any) -> Optional[bytes]: + """Return the first field payload that decodes as a valid preference vector.""" + return next((c for c in extract_payload_candidates(commitment_value) if decode_prefs(c) is not None), None) + + +def fetch_all_commitments(subtensor: 'bt.Subtensor', netuid: int, block: int) -> Dict[str, bytes]: + """All hotkeys' valid preference payloads at a block: hotkey -> payload.""" + records = subtensor.query_map(module='Commitments', name='CommitmentOf', params=[netuid], block=block) + commitments: Dict[str, bytes] = {} + for hotkey, value in cast(Any, records) or []: + payload = extract_prefs(value) + if payload is not None: + commitments[str(getattr(hotkey, 'value', hotkey))] = payload + return commitments + + +def fetch_own_prefs(subtensor: 'bt.Subtensor', netuid: int, hotkey_ss58: str) -> Optional[Dict[str, int]]: + """The validator's currently published preference vector, if any.""" + value = subtensor.substrate.query( + module='Commitments', storage_function='CommitmentOf', params=[netuid, hotkey_ss58] + ) + payload = extract_prefs(value) + return decode_prefs(payload) if payload is not None else None + + +def publish_payload(subtensor: 'bt.Subtensor', wallet: 'bt.Wallet', netuid: int, payload: bytes) -> bool: + """Publish one BigRaw commitment field signed by the validator hotkey.""" + # The pallet builder is typed for sync and async subtensors; sync returns the call directly. + call = cast(Any, Commitments(subtensor).set_commitment(netuid=netuid, info={'fields': [[{'BigRaw': payload}]]})) + response = subtensor.sign_and_send_extrinsic( + call=call, wallet=wallet, sign_with='hotkey', wait_for_inclusion=True, wait_for_finalization=False + ) + return bool(response.success) diff --git a/gittensor/validator/weight_consensus/codec.py b/gittensor/validator/weight_consensus/codec.py new file mode 100644 index 000000000..bf7f87ded --- /dev/null +++ b/gittensor/validator/weight_consensus/codec.py @@ -0,0 +1,103 @@ +# The MIT License (MIT) +# Copyright © 2025 Entrius +"""Payload codec for repo weight preference vectors published to the chain. + +Wire format (v1): zlib("v1|owner/repo:weight|...") — names lowercase, weights +u16 relative shares, at most CONSENSUS_MAX_REPOS entries, one BigRaw field. +""" + +import re +import zlib +from typing import Dict, Mapping, Optional + +from gittensor.validator.utils.config import CONSENSUS_MAX_PAYLOAD_BYTES, CONSENSUS_MAX_REPOS + +PAYLOAD_VERSION = 'v1' +U16_MAX = 65535 +_MAX_PLAINTEXT_BYTES = 4096 +_REPO_NAME_RE = re.compile(r'^[a-z0-9][a-z0-9\-_.]*/[a-z0-9\-_.]+$') + + +class CodecError(ValueError): + """Raised when local preferences cannot be encoded into a valid payload.""" + + +def canonicalize_prefs(raw: Mapping[str, float]) -> Dict[str, int]: + """Normalize local preferences into the canonical on-chain form. + + Lowercases names (merging duplicates), keeps the top CONSENSUS_MAX_REPOS by + weight, and quantizes to u16 integers summing exactly U16_MAX via largest + remainder. Deterministic: all ties break lexicographically. + """ + merged: Dict[str, float] = {} + for name, weight in raw.items(): + if not isinstance(weight, (int, float)) or weight <= 0: + continue + key = str(name).strip().lower() + merged[key] = merged.get(key, 0.0) + float(weight) + + for name in merged: + if not _REPO_NAME_RE.match(name): + raise CodecError(f'Invalid repository name: {name!r}') + + top = sorted(merged.items(), key=lambda kv: (-kv[1], kv[0]))[:CONSENSUS_MAX_REPOS] + if not top: + raise CodecError('No repositories with positive weight to encode') + + total = sum(w for _, w in top) + floors = {name: int(w * U16_MAX // total) for name, w in top} + remainders = sorted(top, key=lambda kv: (-(kv[1] * U16_MAX / total - floors[kv[0]]), kv[0])) + for name, _ in remainders[: U16_MAX - sum(floors.values())]: + floors[name] += 1 + + return {name: floors[name] for name in sorted(floors) if floors[name] > 0} + + +def encode_prefs(prefs: Mapping[str, int]) -> bytes: + """Encode canonical preferences into the compressed wire payload.""" + if not prefs or len(prefs) > CONSENSUS_MAX_REPOS: + raise CodecError(f'Basket must contain 1-{CONSENSUS_MAX_REPOS} repositories, got {len(prefs)}') + for name, weight in prefs.items(): + if not _REPO_NAME_RE.match(name): + raise CodecError(f'Invalid repository name: {name!r}') + if not isinstance(weight, int) or not 0 < weight <= U16_MAX: + raise CodecError(f'Weight for {name} must be an int in [1, {U16_MAX}], got {weight!r}') + + plaintext = PAYLOAD_VERSION + '|' + '|'.join(f'{name}:{prefs[name]}' for name in sorted(prefs)) + payload = zlib.compress(plaintext.encode('ascii'), 9) + if len(payload) > CONSENSUS_MAX_PAYLOAD_BYTES: + raise CodecError(f'Encoded payload is {len(payload)} bytes, exceeds {CONSENSUS_MAX_PAYLOAD_BYTES}') + return payload + + +def decode_prefs(payload: bytes) -> Optional[Dict[str, int]]: + """Decode a wire payload into canonical preferences. + + Returns None on ANY invalidity — a bad payload disqualifies the voter, + never raises. Zero weights are dropped; an empty result is invalid. + """ + try: + decompressor = zlib.decompressobj() + plaintext = decompressor.decompress(payload, _MAX_PLAINTEXT_BYTES).decode('ascii') + if not decompressor.eof: + return None + + version, *entries = plaintext.split('|') + if version != PAYLOAD_VERSION or not entries or len(entries) > CONSENSUS_MAX_REPOS: + return None + + prefs: Dict[str, int] = {} + seen = set() + for entry in entries: + name, _, weight_str = entry.rpartition(':') + name = name.lower() + weight = int(weight_str) + if not _REPO_NAME_RE.match(name) or not 0 <= weight <= U16_MAX or name in seen: + return None + seen.add(name) + if weight > 0: + prefs[name] = weight + + return {name: prefs[name] for name in sorted(prefs)} if prefs else None + except Exception: + return None diff --git a/gittensor/validator/weight_consensus/consensus.py b/gittensor/validator/weight_consensus/consensus.py new file mode 100644 index 000000000..5d8be84aa --- /dev/null +++ b/gittensor/validator/weight_consensus/consensus.py @@ -0,0 +1,114 @@ +# The MIT License (MIT) +# Copyright © 2025 Entrius +"""Deterministic aggregation of validator repo weight preferences. + +Every validator computes this over the identical chain snapshot, so the math +runs entirely in python big-ints — floats appear only in the final division of +identical integers, guaranteeing byte-identical shares on every machine. +""" + +from dataclasses import dataclass, replace +from typing import Dict, Mapping, Optional + +from gittensor.validator.utils.config import ( + CONSENSUS_MIN_VALIDATOR_STAKE_RAO, + CONSENSUS_SNAPSHOT_INTERVAL_BLOCKS, + CONSENSUS_WEIGHT_PRECISION, +) +from gittensor.validator.utils.load_weights import RepositoryConfig +from gittensor.validator.weight_consensus.codec import decode_prefs + + +@dataclass(frozen=True) +class AggregateResult: + """Outcome of one snapshot aggregation. + + ``shares`` is None when the activation gate failed — callers fall back to + the baked-in repository weights. ``numerators`` are the exact integers the + disk cache persists so a reload reproduces identical float shares. + """ + + shares: Optional[Dict[str, float]] + numerators: Dict[str, int] + eligible_stake_rao: int + valid_stake_rao: int + voter_count: int + + +def compute_snapshot_block(block: int) -> int: + return block - (block % CONSENSUS_SNAPSHOT_INTERVAL_BLOCKS) + + +def aggregate_preferences( + commitments: Mapping[str, bytes], + stakes_rao: Mapping[str, int], + validator_permits: Mapping[str, bool], +) -> AggregateResult: + """Stake-weighted mean of eligible validators' preference vectors. + + Eligible voters hold a validator permit and >= the stake threshold at the + snapshot block. Each valid vector is normalized to sum 1.0 in fixed-point + (``S_rao * w * PREC // W_v``) before the stake-weighted accumulation, so a + voter's influence is exactly proportional to stake. The activation gate + passes when valid voters hold at least half the eligible stake. + """ + eligible_stake = 0 + valid_stake = 0 + voter_count = 0 + numerators: Dict[str, int] = {} + + for hotkey in sorted(stakes_rao): + stake = stakes_rao[hotkey] + if not validator_permits.get(hotkey) or stake < CONSENSUS_MIN_VALIDATOR_STAKE_RAO: + continue + eligible_stake += stake + + payload = commitments.get(hotkey) + prefs = decode_prefs(payload) if payload else None + if prefs is None: + continue + + valid_stake += stake + voter_count += 1 + basket_total = sum(prefs.values()) + for repo, weight in prefs.items(): + numerators[repo] = numerators.get(repo, 0) + stake * weight * CONSENSUS_WEIGHT_PRECISION // basket_total + + gate_passed = eligible_stake > 0 and valid_stake * 2 >= eligible_stake and numerators + return AggregateResult( + shares=shares_from_numerators(numerators) if gate_passed else None, + numerators=numerators, + eligible_stake_rao=eligible_stake, + valid_stake_rao=valid_stake, + voter_count=voter_count, + ) + + +def shares_from_numerators(numerators: Mapping[str, int]) -> Dict[str, float]: + """Convert integer numerators to float shares summing ~1.0. Deterministic: + identical ints divide to identical doubles on every platform.""" + total = sum(numerators.values()) + return {repo: numerators[repo] / total for repo in sorted(numerators)} + + +def apply_consensus( + master: Dict[str, RepositoryConfig], + shares: Optional[Mapping[str, float]], +) -> Dict[str, RepositoryConfig]: + """Overlay consensus shares onto the baked repository registry. + + None shares (gate failed / no aggregate) leaves the registry untouched. + Otherwise the aggregate is the complete share vector: registry repos keep + their tuned config with the share overridden (0.0 when unvoted — the + aggregate sums to 1.0, keeping baked shares would double-count), and novel + repos enter with default config. + """ + if shares is None: + return master + + return { + name: replace(master[name], emission_share=shares.get(name, 0.0)) + if name in master + else RepositoryConfig(emission_share=shares[name]) + for name in sorted(set(master) | set(shares)) + } diff --git a/gittensor/validator/weight_consensus/manager.py b/gittensor/validator/weight_consensus/manager.py new file mode 100644 index 000000000..bdb4acbd9 --- /dev/null +++ b/gittensor/validator/weight_consensus/manager.py @@ -0,0 +1,157 @@ +# The MIT License (MIT) +# Copyright © 2025 Entrius +"""Snapshot lifecycle for the repo weight consensus. + +Aggregates are computed opportunistically right after each snapshot boundary +(while lite nodes still serve that state) and persisted to disk as integer +numerators, so restarts and pruned state fall back to the last-good aggregate +instead of diverging or crashing. Chain problems degrade, never raise. +""" + +import json +import os +from pathlib import Path +from typing import Any, Callable, Dict, Optional + +import bittensor as bt + +from gittensor.validator.utils.config import ( + CONSENSUS_CACHE_KEEP, + CONSENSUS_FRESH_WINDOW_BLOCKS, +) +from gittensor.validator.utils.load_weights import RepositoryConfig +from gittensor.validator.weight_consensus.chain import fetch_all_commitments +from gittensor.validator.weight_consensus.consensus import ( + AggregateResult, + aggregate_preferences, + apply_consensus, + compute_snapshot_block, + shares_from_numerators, +) +from gittensor.validator.weight_consensus.publisher import maybe_publish_prefs, resolve_local_prefs + +StoreHook = Callable[[int, Dict[str, bytes], Dict[str, int], Dict[str, bool], AggregateResult], None] + + +class ConsensusManager: + """Computes, caches, and serves the per-snapshot aggregate.""" + + def __init__(self, netuid: int, cache_dir: Path, store_hook: Optional[StoreHook] = None): + self.netuid = netuid + self.cache_path = Path(cache_dir) / 'weight_consensus_cache.json' + self.store_hook = store_hook + self._failed_snapshots: set = set() + self._cache: Dict[str, Any] = self._load_cache() + + def maybe_refresh(self, subtensor: 'bt.Subtensor', block: int) -> None: + """Cheap per-step tick: compute the current snapshot once, while fresh.""" + snapshot = compute_snapshot_block(block) + if str(snapshot) in self._cache or snapshot in self._failed_snapshots: + return + try: + self._compute_and_persist(subtensor, snapshot) + except Exception as e: + if block - snapshot > CONSENSUS_FRESH_WINDOW_BLOCKS: + self._failed_snapshots.add(snapshot) + bt.logging.warning( + f'weight_consensus: snapshot {snapshot} state unavailable ({e}); ' + f'using last-good aggregate until the next boundary' + ) + else: + bt.logging.debug(f'weight_consensus: snapshot {snapshot} refresh failed ({e}); retrying next step') + + def get_shares(self, subtensor: 'bt.Subtensor', block: int) -> Optional[Dict[str, float]]: + """Aggregate shares for the current snapshot, last-good on pruned state, + None when nothing is available or the activation gate failed.""" + snapshot = compute_snapshot_block(block) + entry = self._cache.get(str(snapshot)) + + if entry is None and snapshot not in self._failed_snapshots: + try: + entry = self._compute_and_persist(subtensor, snapshot) + except Exception as e: + self._failed_snapshots.add(snapshot) + bt.logging.warning(f'weight_consensus: snapshot {snapshot} compute failed ({e})') + + if entry is None: + previous = [int(b) for b in self._cache if int(b) < snapshot] + if previous: + entry = self._cache[str(max(previous))] + bt.logging.warning(f'weight_consensus: using last-good aggregate from snapshot {max(previous)}') + + if entry is None or not entry['gate_passed']: + return None + return shares_from_numerators({repo: int(n) for repo, n in entry['numerators'].items()}) + + def _compute_and_persist(self, subtensor: 'bt.Subtensor', snapshot: int) -> Dict[str, Any]: + commitments = fetch_all_commitments(subtensor, self.netuid, snapshot) + metagraph = subtensor.metagraph(self.netuid, block=snapshot, lite=True) + stakes_rao = {hk: int(round(float(metagraph.S[uid]) * 1e9)) for uid, hk in enumerate(metagraph.hotkeys)} + permits = {hk: bool(metagraph.validator_permit[uid]) for uid, hk in enumerate(metagraph.hotkeys)} + + result = aggregate_preferences(commitments, stakes_rao, permits) + entry = { + 'gate_passed': result.shares is not None, + 'numerators': result.numerators, + 'eligible_stake_rao': result.eligible_stake_rao, + 'valid_stake_rao': result.valid_stake_rao, + 'voter_count': result.voter_count, + } + self._cache[str(snapshot)] = entry + self._save_cache() + bt.logging.info( + f'weight_consensus: snapshot {snapshot} aggregated — {result.voter_count} voters, ' + f'gate {"passed" if entry["gate_passed"] else "FAILED (using baked weights)"}' + ) + + if self.store_hook is not None: + try: + self.store_hook(snapshot, commitments, stakes_rao, permits, result) + except Exception as e: + bt.logging.warning(f'weight_consensus: snapshot store hook failed ({e})') + return entry + + def _load_cache(self) -> Dict[str, Any]: + try: + if self.cache_path.exists(): + return json.loads(self.cache_path.read_text()) + except (OSError, ValueError) as e: + bt.logging.warning(f'weight_consensus: corrupt cache {self.cache_path} ({e}); starting fresh') + return {} + + def _save_cache(self) -> None: + keep = sorted(self._cache, key=int)[-CONSENSUS_CACHE_KEEP:] + self._cache = {block: self._cache[block] for block in keep} + try: + tmp_path = self.cache_path.with_suffix('.tmp') + tmp_path.write_text(json.dumps(self._cache)) + os.replace(tmp_path, self.cache_path) + except OSError as e: + bt.logging.warning(f'weight_consensus: cache write failed ({e})') + + +def run_weight_consensus(validator, master: Dict[str, RepositoryConfig]) -> Dict[str, RepositoryConfig]: + """Forward-seam orchestrator: publish own vote, fetch the aggregate, overlay + it on the baked registry. Any failure returns the registry untouched.""" + if getattr(validator.config.neuron, 'disable_weight_consensus', False): + return master + + try: + prefs_path = Path( + validator.config.neuron.consensus_prefs_path + or Path(validator.config.neuron.full_path) / 'repo_weight_prefs.json' + ) + prefs = resolve_local_prefs(prefs_path, master) + if not maybe_publish_prefs(validator.subtensor, validator.wallet, validator.config.netuid, prefs): + bt.logging.warning( + 'weight_consensus: no preference vector on chain — running as bystander on the ' + 'aggregate. Future releases may require participation.' + ) + + shares = validator.consensus_manager.get_shares(validator.subtensor, validator.block) + if shares is None: + bt.logging.info('weight_consensus: no active aggregate; using baked-in repository weights') + return apply_consensus(master, shares) + except Exception as e: + bt.logging.error(f'weight_consensus: unexpected failure ({e}); using baked-in repository weights') + return master diff --git a/gittensor/validator/weight_consensus/publisher.py b/gittensor/validator/weight_consensus/publisher.py new file mode 100644 index 000000000..ea2d2f0a2 --- /dev/null +++ b/gittensor/validator/weight_consensus/publisher.py @@ -0,0 +1,48 @@ +# The MIT License (MIT) +# Copyright © 2025 Entrius +"""Publishes the validator's own repo weight preferences to the chain.""" + +import json +from pathlib import Path +from typing import Dict, Optional + +import bittensor as bt + +from gittensor.validator.utils.load_weights import RepositoryConfig +from gittensor.validator.weight_consensus.chain import fetch_own_prefs, publish_payload +from gittensor.validator.weight_consensus.codec import CodecError, canonicalize_prefs, encode_prefs + + +def resolve_local_prefs(prefs_path: Optional[Path], master: Dict[str, RepositoryConfig]) -> Dict[str, int]: + """The validator's desired vote in canonical form. + + Reads ``{"version": 1, "repos": {"owner/repo": weight}}`` from prefs_path; + a missing or invalid file falls back to voting the baked-in registry shares. + """ + if prefs_path is not None and prefs_path.exists(): + try: + data = json.loads(prefs_path.read_text()) + return canonicalize_prefs(data['repos']) + except (OSError, ValueError, KeyError, TypeError, CodecError) as e: + bt.logging.warning(f'weight_consensus: invalid prefs file {prefs_path} ({e}); voting baked-in shares') + + return canonicalize_prefs({name: cfg.emission_share for name, cfg in master.items() if cfg.emission_share > 0}) + + +def maybe_publish_prefs(subtensor: 'bt.Subtensor', wallet: 'bt.Wallet', netuid: int, prefs: Dict[str, int]) -> bool: + """Publish prefs unless the identical vector is already on chain. + + Compares decoded vectors, never compressed bytes (zlib output can differ + across builds). Returns True when the on-chain state matches the desired + prefs on exit; failures warn and retry next round. + """ + try: + if fetch_own_prefs(subtensor, netuid, wallet.hotkey.ss58_address) == prefs: + return True + if publish_payload(subtensor, wallet, netuid, encode_prefs(prefs)): + bt.logging.info(f'weight_consensus: published preference vector ({len(prefs)} repos)') + return True + bt.logging.warning('weight_consensus: commitment publish rejected; will retry next round') + except Exception as e: + bt.logging.warning(f'weight_consensus: commitment publish failed ({e}); will retry next round') + return False diff --git a/neurons/validator.py b/neurons/validator.py index 4dffa3bf5..a90e2355c 100644 --- a/neurons/validator.py +++ b/neurons/validator.py @@ -19,11 +19,12 @@ import os import time from functools import partial +from pathlib import Path from typing import Dict, List, Set import bittensor as bt -import wandb +import wandb from gittensor import __version__ from gittensor.classes import MinerEvaluation, MinerEvaluationCache from gittensor.validator import pat_storage @@ -36,9 +37,16 @@ priority_pat_broadcast, priority_pat_check, ) -from gittensor.validator.utils.config import STORE_DB_RESULTS, WANDB_PROJECT, WANDB_VALIDATOR_NAME +from gittensor.validator.utils.config import ( + CONSENSUS_MIN_VALIDATOR_STAKE_RAO, + STORE_DB_RESULTS, + WANDB_PROJECT, + WANDB_VALIDATOR_NAME, +) from gittensor.validator.utils.load_weights import RepositoryConfig from gittensor.validator.utils.storage import DatabaseStorage +from gittensor.validator.weight_consensus import ConsensusManager +from gittensor.validator.weight_consensus.codec import decode_prefs from neurons.base.validator import BaseValidatorNeuron @@ -83,6 +91,12 @@ def __init__(self, config=None): bt.logging.warning('Validation result storage enabled.') self.db_storage = DatabaseStorage() + self.consensus_manager = ConsensusManager( + netuid=self.config.netuid, + cache_dir=Path(self.config.neuron.full_path), + store_hook=self._store_weight_consensus if self.db_storage else None, + ) + # Initialize wandb only if disable_set_weights is False if not self.config.neuron.disable_set_weights: try: @@ -99,6 +113,17 @@ def __init__(self, config=None): bt.logging.info('load_state()') self.load_state() + def _store_weight_consensus(self, snapshot_block, commitments, stakes_rao, permits, result) -> None: + """Persist eligible voters' baskets and the aggregate for the dashboards.""" + baskets = [ + (hotkey, stakes_rao[hotkey], prefs) + for hotkey, payload in sorted(commitments.items()) + if permits.get(hotkey) + and stakes_rao.get(hotkey, 0) >= CONSENSUS_MIN_VALIDATOR_STAKE_RAO + and (prefs := decode_prefs(payload)) is not None + ] + self.db_storage.store_weight_consensus(snapshot_block, baskets, result) + async def bulk_store_evaluation( self, miner_evals: Dict[int, MinerEvaluation], diff --git a/tests/validator/test_consensus_aggregate.py b/tests/validator/test_consensus_aggregate.py new file mode 100644 index 000000000..825f12dbf --- /dev/null +++ b/tests/validator/test_consensus_aggregate.py @@ -0,0 +1,144 @@ +# The MIT License (MIT) +# Copyright © 2025 Entrius +"""Tests for deterministic stake-weighted aggregation of repo weight preferences.""" + +import json +import random + +from gittensor.validator.utils.config import ( + CONSENSUS_MIN_VALIDATOR_STAKE_RAO, + CONSENSUS_SNAPSHOT_INTERVAL_BLOCKS, + CONSENSUS_WEIGHT_PRECISION, +) +from gittensor.validator.weight_consensus.codec import encode_prefs +from gittensor.validator.weight_consensus.consensus import aggregate_preferences, compute_snapshot_block + +STAKE_30K = CONSENSUS_MIN_VALIDATOR_STAKE_RAO + + +def _aggregate(voters): + """voters: list of (hotkey, stake, permit, prefs-or-raw-bytes-or-None).""" + commitments, stakes, permits = {}, {}, {} + for hotkey, stake, permit, prefs in voters: + stakes[hotkey] = stake + permits[hotkey] = permit + if isinstance(prefs, bytes): + commitments[hotkey] = prefs + elif prefs is not None: + commitments[hotkey] = encode_prefs(prefs) + return aggregate_preferences(commitments, stakes, permits) + + +class TestSnapshotBlock: + def test_boundaries(self): + interval = CONSENSUS_SNAPSHOT_INTERVAL_BLOCKS + assert compute_snapshot_block(0) == 0 + assert compute_snapshot_block(interval - 1) == 0 + assert compute_snapshot_block(interval) == interval + assert compute_snapshot_block(2 * interval - 1) == interval + + +class TestAggregation: + def test_stake_weighted_mean_golden_vector(self): + result = _aggregate( + [ + ('hk1', 3 * STAKE_30K, True, {'a/b': 65535}), + ('hk2', STAKE_30K, True, {'a/b': 32768, 'c/d': 32767}), + ] + ) + s1, s2, prec = 3 * STAKE_30K, STAKE_30K, CONSENSUS_WEIGHT_PRECISION + expected_ab = s1 * 65535 * prec // 65535 + s2 * 32768 * prec // 65535 + expected_cd = s2 * 32767 * prec // 65535 + assert result.numerators == {'a/b': expected_ab, 'c/d': expected_cd} + total = expected_ab + expected_cd + assert result.shares == {'a/b': expected_ab / total, 'c/d': expected_cd / total} + assert result.voter_count == 2 + + def test_filters_no_permit_and_low_stake(self): + result = _aggregate( + [ + ('miner', 100 * STAKE_30K, False, {'m/spam': 65535}), + ('small', STAKE_30K - 1, True, {'s/small': 65535}), + ('vali', STAKE_30K, True, {'a/b': 65535}), + ] + ) + assert set(result.shares) == {'a/b'} + assert result.eligible_stake_rao == STAKE_30K + + def test_invalid_payload_counts_toward_eligible_not_valid(self): + result = _aggregate( + [ + ('bad', STAKE_30K, True, b'\x00junk'), + ('good', STAKE_30K, True, {'a/b': 65535}), + ] + ) + assert result.eligible_stake_rao == 2 * STAKE_30K + assert result.valid_stake_rao == STAKE_30K + assert result.shares == {'a/b': 1.0} + + def test_activation_gate_below_half_returns_none_shares(self): + result = _aggregate( + [ + ('silent', 3 * STAKE_30K, True, None), + ('voter', STAKE_30K, True, {'a/b': 65535}), + ] + ) + assert result.shares is None + assert result.numerators # still computed for the cache + + def test_activation_gate_exact_half_passes(self): + result = _aggregate( + [ + ('silent', STAKE_30K, True, None), + ('voter', STAKE_30K, True, {'a/b': 65535}), + ] + ) + assert result.shares == {'a/b': 1.0} + + def test_zero_eligible_stake_gate_fails(self): + result = _aggregate([('miner', STAKE_30K, False, {'a/b': 65535})]) + assert result.shares is None + assert result.eligible_stake_rao == 0 + + def test_per_voter_normalization(self): + # Same stake, same single repo, wildly different basket totals — equal influence. + result = _aggregate( + [ + ('hk1', STAKE_30K, True, {'a/b': 10}), + ('hk2', STAKE_30K, True, {'c/d': 65535}), + ] + ) + assert abs(result.shares['a/b'] - result.shares['c/d']) < 1e-12 + + def test_shares_sum_to_one(self): + random.seed(3) + voters = [ + ( + f'hk{i}', + STAKE_30K * random.randint(1, 20), + True, + {f'o/r{j}': random.randint(1, 65535) for j in range(random.randint(1, 10))}, + ) + for i in range(12) + ] + result = _aggregate(voters) + assert abs(sum(result.shares.values()) - 1.0) < 1e-9 + + def test_determinism_under_shuffled_input_order(self): + random.seed(11) + voters = [ + ( + f'hk{i:02d}', + STAKE_30K * random.randint(1, 50), + random.random() > 0.2, + {f'own{j}/repo{j}': random.randint(1, 65535) for j in range(random.randint(1, 10))}, + ) + for i in range(20) + ] + baseline = None + for _ in range(5): + random.shuffle(voters) + result = _aggregate(voters) + serialized = json.dumps({'shares': result.shares, 'numerators': result.numerators}, sort_keys=True) + assert baseline is None or serialized == baseline + baseline = serialized diff --git a/tests/validator/test_consensus_apply.py b/tests/validator/test_consensus_apply.py new file mode 100644 index 000000000..e6103c74e --- /dev/null +++ b/tests/validator/test_consensus_apply.py @@ -0,0 +1,41 @@ +# The MIT License (MIT) +# Copyright © 2025 Entrius +"""Tests for overlaying consensus shares onto the baked repository registry.""" + +from gittensor.validator.utils.load_weights import RepositoryConfig +from gittensor.validator.weight_consensus.consensus import apply_consensus + + +def _master(): + return { + 'a/tuned': RepositoryConfig(emission_share=0.6, maintainer_cut=0.2, default_label_multiplier=1.5), + 'b/plain': RepositoryConfig(emission_share=0.4), + } + + +class TestApplyConsensus: + def test_overrides_share_keeps_tuned_config(self): + result = apply_consensus(_master(), {'a/tuned': 0.7, 'b/plain': 0.3}) + assert result['a/tuned'].emission_share == 0.7 + assert result['a/tuned'].maintainer_cut == 0.2 + assert result['a/tuned'].default_label_multiplier == 1.5 + + def test_zeroes_master_repos_absent_from_aggregate(self): + result = apply_consensus(_master(), {'a/tuned': 1.0}) + assert result['b/plain'].emission_share == 0.0 + + def test_adds_novel_repo_with_defaults(self): + result = apply_consensus(_master(), {'new/comer': 0.5, 'a/tuned': 0.5}) + novel = result['new/comer'] + assert novel.emission_share == 0.5 + assert novel.maintainer_cut == 0.0 + assert novel.default_label_multiplier == 1.0 + + def test_none_shares_returns_master_unchanged(self): + master = _master() + assert apply_consensus(master, None) is master + + def test_does_not_mutate_master_configs(self): + master = _master() + apply_consensus(master, {'a/tuned': 1.0}) + assert master['a/tuned'].emission_share == 0.6 diff --git a/tests/validator/test_consensus_codec.py b/tests/validator/test_consensus_codec.py new file mode 100644 index 000000000..92013d93d --- /dev/null +++ b/tests/validator/test_consensus_codec.py @@ -0,0 +1,85 @@ +# The MIT License (MIT) +# Copyright © 2025 Entrius +"""Tests for the weight consensus payload codec.""" + +import zlib + +import pytest + +from gittensor.validator.weight_consensus.codec import ( + U16_MAX, + CodecError, + canonicalize_prefs, + decode_prefs, + encode_prefs, +) + + +def _encode_plaintext(plaintext: str) -> bytes: + return zlib.compress(plaintext.encode('ascii'), 9) + + +class TestEncodeDecode: + def test_roundtrip(self): + prefs = canonicalize_prefs({'entrius/gittensor': 0.5, 'autovara/kata': 0.3, 'a/b': 0.2}) + assert decode_prefs(encode_prefs(prefs)) == prefs + + def test_decode_rejects_bad_version(self): + assert decode_prefs(_encode_plaintext('v2|a/b:100')) is None + + def test_decode_rejects_undecompressable_bytes(self): + assert decode_prefs(b'\x00garbage') is None + + @pytest.mark.parametrize('entry', ['a/b', 'a/b:', 'a/b:x', 'a/b:1.5', 'a/b:-1', 'a/b:65536', ':100', 'noslash:100']) + def test_decode_rejects_malformed_entries(self, entry): + assert decode_prefs(_encode_plaintext(f'v1|{entry}')) is None + + def test_decode_rejects_too_many_repos(self): + plaintext = 'v1|' + '|'.join(f'a/repo{i}:100' for i in range(11)) + assert decode_prefs(_encode_plaintext(plaintext)) is None + + def test_decode_lowercases_repo_names(self): + assert decode_prefs(_encode_plaintext('v1|Owner/Repo:100')) == {'owner/repo': 100} + + def test_decode_drops_zero_weights_and_rejects_empty_vector(self): + assert decode_prefs(_encode_plaintext('v1|a/b:0|c/d:100')) == {'c/d': 100} + assert decode_prefs(_encode_plaintext('v1|a/b:0')) is None + + def test_decode_rejects_duplicates_after_lowercasing(self): + assert decode_prefs(_encode_plaintext('v1|a/b:100|A/B:200')) is None + assert decode_prefs(_encode_plaintext('v1|a/b:0|a/b:200')) is None + + def test_decode_rejects_oversized_plaintext(self): + assert decode_prefs(zlib.compress(b'v1|' + b'a' * 100_000)) is None + + def test_encode_raises_on_too_many_repos(self): + with pytest.raises(CodecError): + encode_prefs({f'a/repo{i}': 100 for i in range(11)}) + + def test_encode_raises_on_invalid_weight(self): + with pytest.raises(CodecError): + encode_prefs({'a/b': 0}) + with pytest.raises(CodecError): + encode_prefs({'a/b': U16_MAX + 1}) + + +class TestCanonicalize: + def test_top10_by_weight_quantized_to_u16_sum(self): + raw = {f'a/repo{i:02d}': float(i + 1) for i in range(15)} + prefs = canonicalize_prefs(raw) + assert len(prefs) == 10 + assert sum(prefs.values()) == U16_MAX + assert set(prefs) == {f'a/repo{i:02d}' for i in range(5, 15)} + + def test_deterministic_ties_and_duplicate_merge(self): + assert canonicalize_prefs({'A/B': 1.0, 'a/b': 1.0, 'c/d': 2.0}) == canonicalize_prefs({'a/b': 2.0, 'c/d': 2.0}) + + def test_raises_on_empty_or_nonpositive(self): + with pytest.raises(CodecError): + canonicalize_prefs({}) + with pytest.raises(CodecError): + canonicalize_prefs({'a/b': 0.0, 'c/d': -1.0}) + + def test_raises_on_invalid_name(self): + with pytest.raises(CodecError): + canonicalize_prefs({'not a repo': 1.0}) diff --git a/tests/validator/test_consensus_manager.py b/tests/validator/test_consensus_manager.py new file mode 100644 index 000000000..c2afa2997 --- /dev/null +++ b/tests/validator/test_consensus_manager.py @@ -0,0 +1,154 @@ +# The MIT License (MIT) +# Copyright © 2025 Entrius +"""Tests for snapshot lifecycle: warm compute, numerator cache, fallbacks.""" + +import json +from types import SimpleNamespace + +from gittensor.validator.utils.config import ( + CONSENSUS_CACHE_KEEP, + CONSENSUS_FRESH_WINDOW_BLOCKS, + CONSENSUS_MIN_VALIDATOR_STAKE_RAO, + CONSENSUS_SNAPSHOT_INTERVAL_BLOCKS, +) +from gittensor.validator.utils.load_weights import RepositoryConfig +from gittensor.validator.weight_consensus.codec import encode_prefs +from gittensor.validator.weight_consensus.manager import ConsensusManager, run_weight_consensus + +INTERVAL = CONSENSUS_SNAPSHOT_INTERVAL_BLOCKS +STAKE_ALPHA = CONSENSUS_MIN_VALIDATOR_STAKE_RAO / 1e9 + + +class FakeSubtensor: + """Serves canned commitments/metagraph per block; raises for pruned blocks.""" + + def __init__(self, voters, prunable_before: int = -1): + # voters: list of (hotkey, stake_alpha, permit, prefs-or-None) + self.voters = voters + self.prunable_before = prunable_before + self.query_calls = 0 + + def _check_pruned(self, block): + if block < self.prunable_before: + raise RuntimeError(f'State discarded for block {block}') + + def query_map(self, module, name, params, block): + self._check_pruned(block) + self.query_calls += 1 + return [ + (hotkey, {'info': {'fields': [[{'BigRaw': '0x' + encode_prefs(prefs).hex()}]]}}) + for hotkey, _, _, prefs in self.voters + if prefs is not None + ] + + def metagraph(self, netuid, block, lite): + self._check_pruned(block) + return SimpleNamespace( + hotkeys=[hotkey for hotkey, _, _, _ in self.voters], + S=[stake for _, stake, _, _ in self.voters], + validator_permit=[permit for _, _, permit, _ in self.voters], + ) + + +def _voters(): + return [ + ('hk1', 3 * STAKE_ALPHA, True, {'a/b': 65535}), + ('hk2', STAKE_ALPHA, True, {'c/d': 65535}), + ] + + +class TestConsensusManager: + def test_fresh_compute_persists_and_reload_is_byte_identical(self, tmp_path): + subtensor = FakeSubtensor(_voters()) + manager = ConsensusManager(netuid=74, cache_dir=tmp_path) + shares = manager.get_shares(subtensor, INTERVAL + 10) + + reloaded = ConsensusManager(netuid=74, cache_dir=tmp_path) + assert json.dumps(reloaded.get_shares(FakeSubtensor([], prunable_before=10**9), INTERVAL + 50)) == json.dumps( + shares + ) + + def test_cache_hit_makes_no_chain_calls(self, tmp_path): + subtensor = FakeSubtensor(_voters()) + manager = ConsensusManager(netuid=74, cache_dir=tmp_path) + manager.maybe_refresh(subtensor, INTERVAL) + calls = subtensor.query_calls + manager.maybe_refresh(subtensor, INTERVAL + 5) + assert manager.get_shares(subtensor, INTERVAL + 10) is not None + assert subtensor.query_calls == calls + + def test_pruned_state_falls_back_to_last_good(self, tmp_path): + manager = ConsensusManager(netuid=74, cache_dir=tmp_path) + manager.maybe_refresh(FakeSubtensor(_voters()), INTERVAL) + pruned = FakeSubtensor(_voters(), prunable_before=10 * INTERVAL) + shares = manager.get_shares(pruned, 2 * INTERVAL + CONSENSUS_FRESH_WINDOW_BLOCKS + 1) + assert shares == {'a/b': 0.75, 'c/d': 0.25} + + def test_pruned_state_no_cache_returns_none(self, tmp_path): + manager = ConsensusManager(netuid=74, cache_dir=tmp_path) + assert manager.get_shares(FakeSubtensor([], prunable_before=10**9), INTERVAL + 10) is None + + def test_gate_failed_interval_cached_and_returns_none(self, tmp_path): + voters = [('hk1', STAKE_ALPHA, True, None), ('hk2', STAKE_ALPHA / 3, True, {'a/b': 65535})] + subtensor = FakeSubtensor(voters) + manager = ConsensusManager(netuid=74, cache_dir=tmp_path) + assert manager.get_shares(subtensor, INTERVAL) is None + calls = subtensor.query_calls + assert manager.get_shares(subtensor, INTERVAL + 1) is None + assert subtensor.query_calls == calls + + def test_cache_trims_and_tolerates_corrupt_file(self, tmp_path): + (tmp_path / 'weight_consensus_cache.json').write_text('{corrupt') + manager = ConsensusManager(netuid=74, cache_dir=tmp_path) + subtensor = FakeSubtensor(_voters()) + for i in range(CONSENSUS_CACHE_KEEP + 3): + manager.maybe_refresh(subtensor, (i + 1) * INTERVAL) + assert len(manager._cache) == CONSENSUS_CACHE_KEEP + + def test_maybe_refresh_retries_in_window_marks_failed_after(self, tmp_path): + manager = ConsensusManager(netuid=74, cache_dir=tmp_path) + pruned = FakeSubtensor([], prunable_before=10**9) + manager.maybe_refresh(pruned, INTERVAL + 10) + assert INTERVAL not in manager._failed_snapshots # transient: retry next step + manager.maybe_refresh(pruned, INTERVAL + CONSENSUS_FRESH_WINDOW_BLOCKS + 1) + assert INTERVAL in manager._failed_snapshots + + def test_store_hook_failure_does_not_break_compute(self, tmp_path): + def broken_hook(*args): + raise RuntimeError('db down') + + manager = ConsensusManager(netuid=74, cache_dir=tmp_path, store_hook=broken_hook) + assert manager.get_shares(FakeSubtensor(_voters()), INTERVAL) is not None + + +class TestRunWeightConsensus: + def _validator(self, tmp_path, subtensor): + manager = ConsensusManager(netuid=74, cache_dir=tmp_path) + return SimpleNamespace( + config=SimpleNamespace( + netuid=74, + neuron=SimpleNamespace( + disable_weight_consensus=False, consensus_prefs_path=None, full_path=str(tmp_path) + ), + ), + subtensor=subtensor, + wallet=SimpleNamespace(hotkey=SimpleNamespace(ss58_address='hk1')), + block=INTERVAL + 10, + consensus_manager=manager, + ) + + def test_never_raises_and_falls_back_to_master(self, tmp_path): + class ExplodingSubtensor: + def __getattr__(self, name): + raise RuntimeError('chain down') + + master = {'a/b': RepositoryConfig(emission_share=1.0)} + validator = self._validator(tmp_path, FakeSubtensor([], prunable_before=10**9)) + validator.subtensor = ExplodingSubtensor() + assert run_weight_consensus(validator, master) is master + + def test_disable_flag_bypasses_everything(self, tmp_path): + master = {'a/b': RepositoryConfig(emission_share=1.0)} + validator = self._validator(tmp_path, FakeSubtensor(_voters())) + validator.config.neuron.disable_weight_consensus = True + assert run_weight_consensus(validator, master) is master diff --git a/tests/validator/test_consensus_publisher.py b/tests/validator/test_consensus_publisher.py new file mode 100644 index 000000000..acf814bf5 --- /dev/null +++ b/tests/validator/test_consensus_publisher.py @@ -0,0 +1,95 @@ +# The MIT License (MIT) +# Copyright © 2025 Entrius +"""Tests for local preference resolution and chain publishing.""" + +import json +import zlib +from types import SimpleNamespace +from unittest.mock import MagicMock + +from gittensor.validator.utils.load_weights import RepositoryConfig +from gittensor.validator.weight_consensus.chain import extract_payload_candidates, extract_prefs +from gittensor.validator.weight_consensus.codec import encode_prefs +from gittensor.validator.weight_consensus.publisher import maybe_publish_prefs, resolve_local_prefs + +MASTER = { + 'a/big': RepositoryConfig(emission_share=0.6), + 'b/small': RepositoryConfig(emission_share=0.4), + 'c/zero': RepositoryConfig(emission_share=0.0), +} + + +def _commitment_value(payload: bytes) -> dict: + return {'info': {'fields': [[{'BigRaw': '0x' + payload.hex()}]]}} + + +def _subtensor_with_own_commitment(payload): + subtensor = MagicMock() + subtensor.substrate.query.return_value = _commitment_value(payload) if payload else None + subtensor.sign_and_send_extrinsic.return_value = SimpleNamespace(success=True) + return subtensor + + +def _wallet(): + return SimpleNamespace(hotkey=SimpleNamespace(ss58_address='hk-self')) + + +class TestResolveLocalPrefs: + def test_default_vote_from_baked_master_shares(self, tmp_path): + prefs = resolve_local_prefs(tmp_path / 'missing.json', MASTER) + assert set(prefs) == {'a/big', 'b/small'} + assert prefs['a/big'] > prefs['b/small'] + + def test_prefs_file_parsed(self, tmp_path): + path = tmp_path / 'prefs.json' + path.write_text(json.dumps({'version': 1, 'repos': {'X/Y': 3, 'z/w': 1}})) + prefs = resolve_local_prefs(path, MASTER) + assert set(prefs) == {'x/y', 'z/w'} + + def test_invalid_file_falls_back_to_default(self, tmp_path): + path = tmp_path / 'prefs.json' + path.write_text('{not json') + assert set(resolve_local_prefs(path, MASTER)) == {'a/big', 'b/small'} + + +class TestExtractPayload: + def test_extracts_bigraw_hex_bytes_and_int_lists(self): + payload = encode_prefs({'a/b': 100}) + for encoded in ('0x' + payload.hex(), payload, list(payload)): + assert extract_payload_candidates({'info': {'fields': [[{'BigRaw': encoded}]]}}) == [payload] + + def test_prefers_first_valid_prefs_field(self): + payload = encode_prefs({'a/b': 100}) + value = {'info': {'fields': [[{'Raw16': '0x' + b'not a vector 123'.hex()}, {'BigRaw': '0x' + payload.hex()}]]}} + assert extract_prefs(value) == payload + + def test_no_fields_returns_none(self): + assert extract_prefs({'info': {'fields': []}}) is None + assert extract_prefs(None) is None + + +class TestMaybePublish: + def test_skips_when_onchain_decodes_equal(self): + prefs = {'a/b': 60000, 'c/d': 5535} + subtensor = _subtensor_with_own_commitment(encode_prefs(prefs)) + assert maybe_publish_prefs(subtensor, _wallet(), 74, prefs) is True + subtensor.sign_and_send_extrinsic.assert_not_called() + + def test_skip_compares_decoded_prefs_not_compressed_bytes(self): + prefs = {'a/b': 65535} + plaintext = 'v1|a/b:65535' + subtensor = _subtensor_with_own_commitment(zlib.compress(plaintext.encode(), 1)) + assert maybe_publish_prefs(subtensor, _wallet(), 74, prefs) is True + subtensor.sign_and_send_extrinsic.assert_not_called() + + def test_publishes_bigraw_double_nested_info_signed_by_hotkey(self): + prefs = {'a/b': 65535} + subtensor = _subtensor_with_own_commitment(None) + assert maybe_publish_prefs(subtensor, _wallet(), 74, prefs) is True + _, kwargs = subtensor.sign_and_send_extrinsic.call_args + assert kwargs['sign_with'] == 'hotkey' + + def test_publish_failure_warns_and_returns_false(self): + subtensor = _subtensor_with_own_commitment(None) + subtensor.sign_and_send_extrinsic.side_effect = RuntimeError('rate limited') + assert maybe_publish_prefs(subtensor, _wallet(), 74, {'a/b': 65535}) is False From 894a1441ac2ce98063ed57b22efc3848eeeb8a6d Mon Sep 17 00:00:00 2001 From: Landyn Date: Fri, 24 Jul 2026 15:54:25 -0500 Subject: [PATCH 2/2] Add mirror tracked-repo reconciliation from snapshots --- gittensor/validator/utils/config.py | 6 + .../validator/weight_consensus/mirror_sync.py | 122 ++++++++++++++++++ neurons/validator.py | 7 +- tests/validator/test_mirror_sync.py | 102 +++++++++++++++ 4 files changed, 236 insertions(+), 1 deletion(-) create mode 100644 gittensor/validator/weight_consensus/mirror_sync.py create mode 100644 tests/validator/test_mirror_sync.py diff --git a/gittensor/validator/utils/config.py b/gittensor/validator/utils/config.py index a7fb53b26..ae1590520 100644 --- a/gittensor/validator/utils/config.py +++ b/gittensor/validator/utils/config.py @@ -24,6 +24,12 @@ # optional env vars STORE_DB_RESULTS = os.getenv('STORE_DB_RESULTS', 'false').lower() == 'true' +# Mirror reconciliation (team validator only; disabled without the API key) +MIRROR_ADMIN_API_KEY = os.getenv('MIRROR_ADMIN_API_KEY', '') +MIRROR_DEREG_SNAPSHOTS = 4 # consecutive absent snapshots (~2 days) before deregistration +MIRROR_MAX_TRACKED_REPOS = 300 # hard cap on repos synced to the mirror +MIRROR_BACKFILL_DAYS = 40 # deep backfill window for newly registered repos + # log values bt.logging.info(f'VALIDATOR_WAIT: {VALIDATOR_WAIT}') bt.logging.info(f'VALIDATOR_STEPS_INTERVAL: {VALIDATOR_STEPS_INTERVAL}') diff --git a/gittensor/validator/weight_consensus/mirror_sync.py b/gittensor/validator/weight_consensus/mirror_sync.py new file mode 100644 index 000000000..d0227df44 --- /dev/null +++ b/gittensor/validator/weight_consensus/mirror_sync.py @@ -0,0 +1,122 @@ +# The MIT License (MIT) +# Copyright © 2025 Entrius +"""Reconciles the GT mirror's tracked-repo set with the consensus snapshot. + +Runs inside the team validator only (enabled by MIRROR_ADMIN_API_KEY) after +each snapshot is stored. The tracked target is the union of eligible voters' +baskets — bounded by 10 x voters, changing at most 2x/day — so the mirror sees +a stable, definitive list: registrations happen on first appearance, +deregistrations only after a repo is absent for a full hysteresis window. +""" + +from typing import Dict, List, Optional, Set + +import bittensor as bt +import requests + +from gittensor.constants import GITTENSOR_MIRROR_DEFAULT_URL +from gittensor.validator.utils.config import ( + MIRROR_ADMIN_API_KEY, + MIRROR_BACKFILL_DAYS, + MIRROR_DEREG_SNAPSHOTS, + MIRROR_MAX_TRACKED_REPOS, +) + +_TIMEOUT = 30 + + +def sync_mirror_repos( + db_connection, + snapshot_block: int, + voted_repos: Set[str], + aggregate_shares: Optional[Dict[str, float]], +) -> None: + """Diff the voted-repo union against the mirror's registry and reconcile. + + Registers (+deep backfill) voted repos whose GitHub App row exists, warns + on repos pending App install, and deregisters repos absent from the last + MIRROR_DEREG_SNAPSHOTS snapshots — only while the consensus gate is active, + so pre-consensus tracking is never torn down during rollout. + """ + if not MIRROR_ADMIN_API_KEY: + return + + if len(voted_repos) > MIRROR_MAX_TRACKED_REPOS: + shares = aggregate_shares or {} + kept = sorted(voted_repos, key=lambda r: (-shares.get(r, 0.0), r))[:MIRROR_MAX_TRACKED_REPOS] + bt.logging.warning( + f'mirror_sync: {len(voted_repos)} voted repos exceed cap {MIRROR_MAX_TRACKED_REPOS}; ' + f'registering top by aggregate share' + ) + voted_repos = set(kept) + + registry = {entry['repoFullName'].lower(): entry for entry in _admin_get('/api/v1/admin/repos')} + + for repo in sorted(voted_repos): + entry = registry.get(repo) + if entry is None: + bt.logging.warning(f'mirror_sync: {repo} is voted but pending GitHub App install — cannot track yet') + elif not entry['registered']: + _register(repo) + + if aggregate_shares is None: + return # gate inactive — consensus does not govern the list yet + + stale_candidates = [ + entry['repoFullName'] + for entry in registry.values() + if entry['registered'] and entry['repoFullName'].lower() not in voted_repos + ] + for repo in sorted(_absent_for_window(db_connection, stale_candidates)): + _deregister(repo) + + +def _absent_for_window(db_connection, candidates: List[str]) -> List[str]: + """Candidates absent from every basket in the last MIRROR_DEREG_SNAPSHOTS + snapshots. Requires a full window of history so a fresh deployment never + mass-deregisters.""" + if not candidates: + return [] + with db_connection.cursor() as cur: + cur.execute( + 'SELECT DISTINCT snapshot_block FROM validator_weight_baskets ORDER BY snapshot_block DESC LIMIT %s', + (MIRROR_DEREG_SNAPSHOTS,), + ) + recent = [row[0] for row in cur.fetchall()] + if len(recent) < MIRROR_DEREG_SNAPSHOTS: + return [] + cur.execute( + 'SELECT DISTINCT jsonb_object_keys(basket) FROM validator_weight_baskets WHERE snapshot_block = ANY(%s)', + (recent,), + ) + recently_voted = {row[0] for row in cur.fetchall()} + return [repo for repo in candidates if repo.lower() not in recently_voted] + + +def _register(repo: str) -> None: + _admin_post('/api/v1/admin/repos/register', {'repoFullName': repo}) + _admin_post('/api/v1/admin/backfill', {'repoFullName': repo, 'days': MIRROR_BACKFILL_DAYS}) + bt.logging.info(f'mirror_sync: registered {repo} (+{MIRROR_BACKFILL_DAYS}d backfill)') + + +def _deregister(repo: str) -> None: + _admin_post('/api/v1/admin/repos/deregister', {'repoFullName': repo}) + bt.logging.info(f'mirror_sync: deregistered {repo} (absent {MIRROR_DEREG_SNAPSHOTS} snapshots)') + + +def _admin_get(path: str) -> list: + response = requests.get( + f'{GITTENSOR_MIRROR_DEFAULT_URL}{path}', headers={'x-api-key': MIRROR_ADMIN_API_KEY}, timeout=_TIMEOUT + ) + response.raise_for_status() + return response.json() + + +def _admin_post(path: str, body: dict) -> None: + response = requests.post( + f'{GITTENSOR_MIRROR_DEFAULT_URL}{path}', + json=body, + headers={'x-api-key': MIRROR_ADMIN_API_KEY}, + timeout=_TIMEOUT, + ) + response.raise_for_status() diff --git a/neurons/validator.py b/neurons/validator.py index a90e2355c..a197c14eb 100644 --- a/neurons/validator.py +++ b/neurons/validator.py @@ -47,6 +47,7 @@ from gittensor.validator.utils.storage import DatabaseStorage from gittensor.validator.weight_consensus import ConsensusManager from gittensor.validator.weight_consensus.codec import decode_prefs +from gittensor.validator.weight_consensus.mirror_sync import sync_mirror_repos from neurons.base.validator import BaseValidatorNeuron @@ -114,7 +115,8 @@ def __init__(self, config=None): self.load_state() def _store_weight_consensus(self, snapshot_block, commitments, stakes_rao, permits, result) -> None: - """Persist eligible voters' baskets and the aggregate for the dashboards.""" + """Persist eligible voters' baskets and the aggregate for the dashboards, + then reconcile the mirror's tracked repos (team validator only).""" baskets = [ (hotkey, stakes_rao[hotkey], prefs) for hotkey, payload in sorted(commitments.items()) @@ -124,6 +126,9 @@ def _store_weight_consensus(self, snapshot_block, commitments, stakes_rao, permi ] self.db_storage.store_weight_consensus(snapshot_block, baskets, result) + voted_repos = {repo for _, _, prefs in baskets for repo in prefs} + sync_mirror_repos(self.db_storage.db_connection, snapshot_block, voted_repos, result.shares) + async def bulk_store_evaluation( self, miner_evals: Dict[int, MinerEvaluation], diff --git a/tests/validator/test_mirror_sync.py b/tests/validator/test_mirror_sync.py new file mode 100644 index 000000000..14fa85441 --- /dev/null +++ b/tests/validator/test_mirror_sync.py @@ -0,0 +1,102 @@ +# The MIT License (MIT) +# Copyright © 2025 Entrius +"""Tests for mirror tracked-repo reconciliation.""" + +from unittest.mock import MagicMock + +import pytest + +from gittensor.validator.utils.config import MIRROR_DEREG_SNAPSHOTS, MIRROR_MAX_TRACKED_REPOS +from gittensor.validator.weight_consensus import mirror_sync +from gittensor.validator.weight_consensus.mirror_sync import sync_mirror_repos + + +@pytest.fixture +def admin(monkeypatch): + """Enable sync and capture admin API traffic.""" + monkeypatch.setattr(mirror_sync, 'MIRROR_ADMIN_API_KEY', 'key') + calls = {'posts': []} + + def fake_get(url, **kwargs): + response = MagicMock() + response.json.return_value = calls['registry'] + return response + + def fake_post(url, json=None, **kwargs): + calls['posts'].append((url.split('/api/v1/admin')[1], json)) + return MagicMock() + + monkeypatch.setattr(mirror_sync.requests, 'get', fake_get) + monkeypatch.setattr(mirror_sync.requests, 'post', fake_post) + return calls + + +def _db(recent_snapshots, recently_voted): + """Fake psycopg connection serving the two hysteresis queries.""" + cursor = MagicMock() + cursor.__enter__ = lambda self: cursor + cursor.__exit__ = MagicMock(return_value=False) + results = [[(b,) for b in recent_snapshots], [(r,) for r in recently_voted]] + cursor.fetchall.side_effect = results + connection = MagicMock() + connection.cursor.return_value = cursor + return connection + + +def test_disabled_without_api_key(monkeypatch): + monkeypatch.setattr(mirror_sync, 'MIRROR_ADMIN_API_KEY', '') + get = MagicMock() + monkeypatch.setattr(mirror_sync.requests, 'get', get) + sync_mirror_repos(MagicMock(), 3600, {'a/b'}, {'a/b': 1.0}) + get.assert_not_called() + + +def test_registers_unregistered_voted_repos_with_backfill(admin): + admin['registry'] = [ + {'repoFullName': 'a/b', 'registered': False, 'hasInstallation': True}, + {'repoFullName': 'c/d', 'registered': True, 'hasInstallation': True}, + ] + sync_mirror_repos(_db([], []), 3600, {'a/b', 'c/d'}, {'a/b': 0.5, 'c/d': 0.5}) + assert ('/repos/register', {'repoFullName': 'a/b'}) in admin['posts'] + assert any(path == '/backfill' and body['repoFullName'] == 'a/b' for path, body in admin['posts']) + assert not any(body.get('repoFullName') == 'c/d' for _, body in admin['posts']) + + +def test_pending_app_install_only_warns(admin): + admin['registry'] = [] + sync_mirror_repos(_db([], []), 3600, {'new/repo'}, {'new/repo': 1.0}) + assert admin['posts'] == [] + + +def test_deregisters_after_full_absence_window(admin): + admin['registry'] = [{'repoFullName': 'old/repo', 'registered': True, 'hasInstallation': True}] + db = _db(recent_snapshots=list(range(MIRROR_DEREG_SNAPSHOTS)), recently_voted=['a/b']) + sync_mirror_repos(db, 3600, {'a/b'}, {'a/b': 1.0}) + assert ('/repos/deregister', {'repoFullName': 'old/repo'}) in admin['posts'] + + +def test_no_dereg_with_short_history_or_recent_vote_or_inactive_gate(admin): + registry = [{'repoFullName': 'old/repo', 'registered': True, 'hasInstallation': True}] + + admin['registry'] = registry + sync_mirror_repos(_db([1], []), 3600, {'a/b'}, {'a/b': 1.0}) # short history + assert not any(path == '/repos/deregister' for path, _ in admin['posts']) + + admin['registry'] = registry + sync_mirror_repos(_db(list(range(MIRROR_DEREG_SNAPSHOTS)), ['old/repo']), 3600, {'a/b'}, {'a/b': 1.0}) + assert not any(path == '/repos/deregister' for path, _ in admin['posts']) + + admin['registry'] = registry + sync_mirror_repos(_db(list(range(MIRROR_DEREG_SNAPSHOTS)), []), 3600, {'a/b'}, None) # gate inactive + assert not any(path == '/repos/deregister' for path, _ in admin['posts']) + + +def test_cap_registers_top_by_aggregate_share(admin): + repos = {f'o/r{i}': i + 1 for i in range(MIRROR_MAX_TRACKED_REPOS + 5)} + shares = {name: weight / sum(repos.values()) for name, weight in repos.items()} + admin['registry'] = [{'repoFullName': name, 'registered': False, 'hasInstallation': True} for name in repos] + sync_mirror_repos(_db([], []), 3600, set(repos), shares) + registered = [body['repoFullName'] for path, body in admin['posts'] if path == '/repos/register'] + assert len(registered) == MIRROR_MAX_TRACKED_REPOS + assert f'o/r{MIRROR_MAX_TRACKED_REPOS + 4}' in registered # highest share kept + assert 'o/r0' not in registered # lowest share dropped