Mirror tracked-repo reconciliation from consensus snapshots

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

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 <neuron.full_path>/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,

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()
 

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')
@@ -14,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}')

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:

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']

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)

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