Burn KoTH discriminator rewards until GAS enables emissions

gas/koth_weights.py

@@ -1,5 +1,6 @@
 """Build on-chain weights for King-of-the-Hill discriminator lanes."""
 
+from datetime import datetime, timezone
 from typing import Callable, Dict, Iterable, List, Optional
 
 import numpy as np
@@ -16,6 +17,28 @@
 KOTH_CHAIN_ROLES = ("current", "previous", "two_back")
 
 
+def discriminator_emissions_enabled(
+    payload: Optional[dict], now: Optional[datetime] = None
+) -> bool:
+    """Require explicit GAS activation and an elapsed, timezone-aware boundary.
+
+    A cached warm-up response stays disabled after the boundary. GAS must first
+    publish an enabled response; the validator never starts its own timer.
+    """
+    if not isinstance(payload, dict) or payload.get("emissions_enabled") is not True:
+        return False
+    raw_start = payload.get("emissions_start_at")
+    if not isinstance(raw_start, str):
+        return False
+    try:
+        start = datetime.fromisoformat(raw_start.replace("Z", "+00:00"))
+    except ValueError:
+        return False
+    if start.tzinfo is None:
+        return False
+    return start <= (now or datetime.now(timezone.utc))
+
+
 def kings_by_modality(payload: Optional[dict]) -> Dict[str, str]:
     """Map modality -> hotkey from a /kings response."""
     out: Dict[str, str] = {}
@@ -86,14 +109,16 @@ def build_koth_weights(
     burn_uid: Optional[int] = None,
     split: Optional[Dict[str, float]] = None,
     chains: Optional[Dict[str, List[Dict[str, object]]]] = None,
+    emissions_enabled: bool = True,
 ) -> np.ndarray:
     """Return a length-n weight vector. Missing kings go to burn_uid.
 
     `uid_for_hotkey` must resolve at current chain head. Escrow addresses are
     never used. Each discriminator lane is 85/10/5 across the current king and
     the previous two distinct kings. Unused residual slots roll to the current
     king. An unresolvable current king burns its share; unresolvable previous
-    kings roll to the current king when that UID resolved.
+    kings roll to the current king when that UID resolved. When emissions are
+    disabled, all discriminator lanes burn; generator rewards are unchanged.
     """
     split = dict(KOTH_SPLIT if split is None else split)
     try:
@@ -114,6 +139,9 @@ def build_koth_weights(
     burned = 0.0
     for modality in ("image", "video", "audio"):
         pct = split[modality]
+        if not emissions_enabled:
+            burned += pct
+            continue
         members = list((chains or {}).get(modality) or [])
         if not members:
             hotkey = kings.get(modality)

neurons/validator/validator.py

@@ -11,7 +11,12 @@
 
 from gas import __spec_version__ as spec_version
 from gas.protocol.validator_requests import get_benchmark_results, get_current_kings
-from gas.koth_weights import build_koth_weights, chains_by_modality, kings_by_modality
+from gas.koth_weights import (
+    build_koth_weights,
+    chains_by_modality,
+    discriminator_emissions_enabled,
+    kings_by_modality,
+)
 from gas.utils.autoupdater import autoupdate
 from gas.cache import ContentManager
 from gas.utils.metagraph import create_set_weights
@@ -251,6 +256,7 @@ def uid_for_hotkey(hotkey_ss58: str):
                     burn_uid=burn_uid,
                     split=split,
                     chains=chains,
+                    emissions_enabled=discriminator_emissions_enabled(kings_payload),
                 )
 
                 total_weight = float(np.sum(normed_weights))

tests/test_koth_warmup.py

@@ -0,0 +1,69 @@
+"""Activation metadata gates discriminator payouts without changing generators."""
+
+from datetime import datetime, timedelta, timezone
+
+import numpy as np
+import pytest
+
+from gas.koth_weights import (
+    build_koth_weights,
+    chains_by_modality,
+    discriminator_emissions_enabled,
+    kings_by_modality,
+)
+
+
+@pytest.mark.parametrize("metadata", [
+    None,
+    {},
+    {"emissions_enabled": True},
+    {"emissions_enabled": "true", "emissions_start_at": "2026-09-01T00:00:00Z"},
+    {"emissions_enabled": True, "emissions_start_at": "invalid"},
+    {"emissions_enabled": True, "emissions_start_at": "2026-09-01T00:00:00"},
+    {"emissions_enabled": True, "emissions_start_at": 123},
+])
+def test_missing_or_malformed_activation_keeps_burning(metadata):
+    assert not discriminator_emissions_enabled(metadata)
+
+
+def test_activation_requires_elapsed_boundary_and_explicit_enabled_response():
+    boundary = datetime(2026, 9, 1, tzinfo=timezone.utc)
+    cached = {"emissions_enabled": False, "emissions_start_at": boundary.isoformat()}
+    assert not discriminator_emissions_enabled(cached, now=boundary + timedelta(days=1))
+    enabled = {**cached, "emissions_enabled": True}
+    assert not discriminator_emissions_enabled(enabled, now=boundary - timedelta(seconds=1))
+    assert discriminator_emissions_enabled(enabled, now=boundary)
+    assert discriminator_emissions_enabled(enabled, now=boundary + timedelta(seconds=1))
+
+
+def test_warmup_burns_kings_and_residuals_then_resumes_without_changing_generators():
+    boundary = datetime(2026, 9, 1, tzinfo=timezone.utc)
+    payload = {
+        "emissions_start_at": boundary.isoformat(),
+        "emissions_enabled": False,
+        "kings": [{"modality": "image", "ss58_address": "current"}],
+        "chain": {"image": ["current", "previous"]},
+    }
+    args = dict(
+        n=5, scores=np.array([0., 0., 0., 2., 1.]), generator_uids=[3, 4],
+        kings=kings_by_modality(payload), chains=chains_by_modality(payload),
+        uid_for_hotkey={"current": 1, "previous": 2}.get, burn_uid=0,
+        split={"image": .3, "video": .2, "audio": .1, "generator": .4},
+    )
+    warmup = build_koth_weights(
+        **args, emissions_enabled=discriminator_emissions_enabled(payload, now=boundary)
+    )
+    payload["emissions_enabled"] = True
+    live = build_koth_weights(
+        **args, emissions_enabled=discriminator_emissions_enabled(payload, now=boundary)
+    )
+    assert warmup[1] == warmup[2] == 0
+    assert warmup[0] == pytest.approx(.6)
+    assert live[1] > 0 and live[2] > 0
+    assert live[1] + live[2] == pytest.approx(.3)
+    assert live[0] == pytest.approx(.3)  # Vacant video/audio lanes still burn.
+    np.testing.assert_allclose(warmup[3:], live[3:])
+    assert warmup.sum() == pytest.approx(1)
+    assert live.sum() == pytest.approx(1)
+    with pytest.raises(ValueError, match="burn UID is unavailable"):
+        build_koth_weights(**{**args, "burn_uid": None}, emissions_enabled=False)