Fix/g1 freshness gate

connito/shared/config.py

@@ -169,8 +169,8 @@ class CycleCfg(BaseConfig):
     cycle_length: int = 448 # 1.5 hr
     distribute_period: int = 20 # 4 mins
     train_period: int = 300 # 1 hr (will adjust to 500 mins when mature to align with Diloco)
-    commit_period: int = 10 # 2 mins
-    submission_period: int = 80 # 4 mins
+    commit_period: int = 11 # 2 mins
+    submission_period: int = 100 # 4 mins
     validate_period: int = 10 # 10 mins
     merge_period: int = 50 # 10 mins
 
@@ -989,6 +989,13 @@ class EvalCfg(BaseConfig):
     validation_group_c_size: int = 17
     group_a_min_consensus: int = 1               # ≥ 1 qualified validator
     group_a_min_weight_per_validator: float = 0.03   # > 3% from at least one validator
+    # Freshness gate on weight Group 1: a UID may only hold a G1 seat if
+    # its most recent aggregator point is no older than this many rounds
+    # (`round_id >= cur_rid - g1_max_stale_rounds * cycle_length`).
+    # 1 → the current round or the one before it. Set to a large value to
+    # restore the legacy behavior of letting a stale rolling average hold
+    # a G1 seat indefinitely.
+    g1_max_stale_rounds: int = 1
     # When a miner's committed HF repo/revision/file is definitively not
     # retrievable (deleted, private, gated, revision rewritten) AND an
     # unauthenticated probe confirms it is not publicly fetchable, treat

connito/sn_owner/docker/docker-compose.yml

@@ -0,0 +1,44 @@
+services:
+  phase-service:
+    image: ${IMAGE:-ghcr.io/connito-ai/connito-sn-owner:stable}
+    restart: unless-stopped
+    pull_policy: always
+    init: true
+
+    # Local development build
+    build:
+      context: ../../..
+      dockerfile: connito/validator/docker/Dockerfile
+
+    labels:
+      com.centurylinklabs.watchtower.enable: "true"
+
+    # Same pattern as the validator: host networking so the DHT port and
+    # the FastAPI service are reachable without per-port NAT. This also
+    # means uvicorn's 127.0.0.1 bind is host-localhost only — bind 0.0.0.0
+    # in the code if other machines need to reach :8080.
+    network_mode: "host"
+
+    # Match the entrypoint style: exec the module, pass through args.
+    entrypoint: ["/bin/sh", "-c", "exec python -m connito.sn_owner.phase_service \"$@\"", "--"]
+
+    # environment:
+    #   WALLET_NAME: ${WALLET_NAME:?set WALLET_NAME in .env}
+    #   HOTKEY_NAME: ${HOTKEY_NAME:?set HOTKEY_NAME in .env}
+
+    volumes:
+      # Wallet — read-only; only the hotkey is needed for signing (if at all)
+      - ${BITTENSOR_WALLET_PATH:-${HOME}/.bittensor/wallets}:/root/.bittensor/wallets:ro
+      # Persistent state: init_peer_ids.json lives under config.run.root_path
+      - ${DATA_DIR:-../../..}:/data
+      # Owner config YAML — editable on the host without rebuilding
+      - ${CONFIG_PATH:?set CONFIG_PATH in .env}:/data/configs/owner.yaml
+
+    command: ["--path", "/data/configs/owner.yaml"]
+
+    healthcheck:
+      test: ["CMD", "curl", "-fsS", "--max-time", "10", "http://localhost:8080/"]
+      interval: 30s
+      timeout: 15s
+      retries: 5
+      start_period: 60s

connito/test/test_build_submission_uid_weights.py

@@ -264,3 +264,156 @@ def test_payload_is_a_frozen_dataclass():
     p = WeightSubmissionPayload(uid_weights={1: 1.0})
     with pytest.raises((TypeError, AttributeError)):
         p.weight_group_1 = (1, 2, 3)   # type: ignore[misc]
+
+
+# ---------------------------------------------------------------------------
+# G1 freshness gate
+# ---------------------------------------------------------------------------
+
+
+def _agg_with_points(points: dict[int, list[tuple[int, float]]]) -> MinerScoreAggregator:
+    """Build an aggregator from `{uid: [(round_id, score), ...]}`."""
+    agg = MinerScoreAggregator(max_points=8, max_history_points=64)
+    ts = datetime(2026, 1, 1, tzinfo=timezone.utc)
+    micro = 1
+    for uid, entries in points.items():
+        for rid, score in entries:
+            agg.add_score(
+                uid=uid, hotkey=f"hk{uid}", score=score,
+                ts=ts.replace(microsecond=micro), round_id=rid,
+            )
+            micro += 1
+    return agg
+
+
+def test_g1_freshness_gate_drops_stale_uid_despite_higher_avg():
+    """A UID whose most recent point predates the freshness window loses
+    its G1 seat even when its rolling avg is the highest in A∪B.
+
+    This is the mainnet uid-158 shape: a big rank score earned several
+    rounds ago keeps the avg high because `avg` divides by *recorded
+    points*, not by rounds elapsed, so the seat survives until the point
+    ages out of retention.
+    """
+    cur_rid, cycle_length = 1000, 100
+    agg = _agg_with_points({
+        # stale: 3 distinct round_ids inside the ≥3-of-5 window [500, 1000],
+        # but nothing since round 800 — and the highest avg by far.
+        1: [(600, 2.25), (700, 2.25), (800, 2.25)],
+        # fresh: scored right up to the current round, much lower avg.
+        2: [(800, 0.5), (900, 0.5), (1000, 0.5)],
+    })
+    payload = build_submission_uid_weights(
+        score_aggregator=agg,
+        cohort_state=_cohort_state(a=(1, 2)),
+        round_id=cur_rid,
+        cycle_length=cycle_length,
+        eval_cfg=_eval_cfg(),
+    )
+    assert payload.weight_group_1 == (2,)
+    assert payload.g1_stale_excluded == (1,)
+    # Demoted, not erased: the stale UID falls through to the 2% G2 tier
+    # (which gates only on `record_count >= 1`). What it loses is the 98%
+    # seat it was holding on stale evidence.
+    assert payload.weight_group_2 == (1,)
+    assert pytest.approx(payload.uid_weights[2], abs=1e-6) == 0.98
+    assert pytest.approx(payload.uid_weights[1], abs=1e-6) == 0.02
+
+
+def test_g1_freshness_gate_admits_previous_round():
+    """`g1_max_stale_rounds=1` means the current round OR the one before
+    it — a UID last scored exactly `cur_rid - cycle_length` still holds."""
+    cur_rid, cycle_length = 1000, 100
+    agg = _agg_with_points({1: [(700, 1.0), (800, 1.0), (900, 1.0)]})
+    payload = build_submission_uid_weights(
+        score_aggregator=agg,
+        cohort_state=_cohort_state(a=(1,)),
+        round_id=cur_rid,
+        cycle_length=cycle_length,
+        eval_cfg=_eval_cfg(),
+    )
+    assert payload.weight_group_1 == (1,)
+    assert payload.g1_stale_excluded == ()
+
+
+def test_g1_freshness_gate_is_configurable():
+    """A large `g1_max_stale_rounds` restores the legacy behavior of
+    letting a stale rolling average hold a G1 seat."""
+    cur_rid, cycle_length = 1000, 100
+    points = {1: [(600, 2.25), (700, 2.25), (800, 2.25)]}
+    strict = build_submission_uid_weights(
+        score_aggregator=_agg_with_points(points),
+        cohort_state=_cohort_state(a=(1,)),
+        round_id=cur_rid, cycle_length=cycle_length,
+        eval_cfg=_eval_cfg(g1_max_stale_rounds=1),
+    )
+    assert strict.weight_group_1 == (0,)          # empty G1 → owner redirect
+    assert strict.g1_redirected_to_uid_zero is True
+
+    lax = build_submission_uid_weights(
+        score_aggregator=_agg_with_points(points),
+        cohort_state=_cohort_state(a=(1,)),
+        round_id=cur_rid, cycle_length=cycle_length,
+        eval_cfg=_eval_cfg(g1_max_stale_rounds=8),
+    )
+    assert lax.weight_group_1 == (1,)
+    assert lax.g1_stale_excluded == ()
+
+
+def test_g1_freshness_gate_all_stale_redirects_to_uid_zero():
+    """If every A∪B UID is stale the 98% share goes to uid=0 rather than
+    to a miner the validator has no current evidence about."""
+    cur_rid, cycle_length = 1000, 100
+    agg = _agg_with_points({
+        1: [(600, 2.25), (700, 2.25), (800, 1.0)],
+        2: [(500, 1.5), (600, 1.5), (700, 1.5)],
+    })
+    payload = build_submission_uid_weights(
+        score_aggregator=agg,
+        cohort_state=_cohort_state(a=(1, 2)),
+        round_id=cur_rid,
+        cycle_length=cycle_length,
+        eval_cfg=_eval_cfg(),
+    )
+    assert payload.g1_redirected_to_uid_zero is True
+    assert payload.weight_group_1 == (0,)
+    assert set(payload.g1_stale_excluded) == {1, 2}
+    assert pytest.approx(payload.uid_weights[0], abs=1e-6) == 0.98
+    assert pytest.approx(sum(payload.uid_weights.values()), abs=1e-6) == 1.0
+
+
+def test_g1_freshness_gate_counts_a_zero_score_as_evidence():
+    """A `0.0` written by `finalize_round_scores` is still evidence that
+    the validator looked at the miner this round, so it satisfies the
+    gate. The gate is about freshness, not about performance — the low
+    score is already reflected in the avg used for ranking."""
+    cur_rid, cycle_length = 1000, 100
+    agg = _agg_with_points({1: [(800, 2.25), (900, 0.0), (1000, 0.0)]})
+    payload = build_submission_uid_weights(
+        score_aggregator=agg,
+        cohort_state=_cohort_state(a=(1,)),
+        round_id=cur_rid,
+        cycle_length=cycle_length,
+        eval_cfg=_eval_cfg(),
+    )
+    assert payload.weight_group_1 == (1,)
+    assert payload.g1_stale_excluded == ()
+
+
+def test_g1_freshness_gate_does_not_affect_group_2():
+    """G2 keeps its `record_count >= 1` gate with no recency or freshness
+    requirement — the 2% tier is explicitly the wider net."""
+    cur_rid, cycle_length = 1000, 100
+    agg = _agg_with_points({
+        1: [(800, 0.5), (900, 0.5), (1000, 0.5)],   # fresh → G1
+        9: [(600, 0.4)],                             # stale, single record → G2
+    })
+    payload = build_submission_uid_weights(
+        score_aggregator=agg,
+        cohort_state=_cohort_state(a=(1,), c=(9,)),
+        round_id=cur_rid,
+        cycle_length=cycle_length,
+        eval_cfg=_eval_cfg(),
+    )
+    assert payload.weight_group_1 == (1,)
+    assert payload.weight_group_2 == (9,)

connito/validator/evaluator.py

@@ -378,12 +378,16 @@ class WeightSubmissionPayload:
     can log them without recomputing the selection.
     `g1_redirected_to_uid_zero` is set when the empty-G1 guard fires —
     the caller logs that case under its own info line.
+    `g1_stale_excluded` lists the A∪B UIDs that cleared the count and
+    recency gates but were dropped by the freshness gate, so the caller
+    can log exactly who lost a seat to staleness.
     """
     uid_weights: dict[int, float]
     weight_group_1: tuple[int, ...] = ()
     weight_group_2: tuple[int, ...] = ()
     cohort_emission: bool = False
     g1_redirected_to_uid_zero: bool = False
+    g1_stale_excluded: tuple[int, ...] = ()
 
 
 def build_submission_uid_weights(
@@ -414,9 +418,11 @@ def build_submission_uid_weights(
         round_ids within the last `5 * cycle_length` blocks — i.e.
         scored in at least 3 of the last 5 cycles. Tightens the prior
         2-of-5 gate so a miner needs sustained participation to anchor
-        the validator's top-N ballot. Empty-G1 guard: if no UID
-        clears, redirect to `uid = 0` (subnet owner) so the validator
-        stays at full emission.
+        the validator's top-N ballot. On top of that, a **freshness
+        gate**: the UID's most recent tagged point must be no older
+        than `cfg.g1_max_stale_rounds` rounds. Empty-G1 guard: if no
+        UID clears, redirect to `uid = 0` (subnet owner) so the
+        validator stays at full emission.
       * Group 2 (`cfg.weight_group_2_share`): top-`weight_group_2_size`
         of A∪B∪C \\ G1 by aggregator avg, restricted to UIDs with
         `record_count >= 1` (no recency gate).
@@ -441,13 +447,44 @@ def build_submission_uid_weights(
     cur_rid = int(round_id)
     g1_window_min_rid = cur_rid - 5 * int(cycle_length)
 
+    # Freshness gate. `avg_scores` is a rolling mean over *recorded
+    # points* with no notion of elapsed rounds, so a UID that stops
+    # being evaluated keeps its last average intact and holds its G1
+    # seat until those points age out of the retention window. Observed
+    # on mainnet: uid 158 took 31.5% of emission at round 8692978 on a
+    # 2.25 earned four rounds earlier, and its avg sat unchanged at
+    # 0.40625 across six consecutive rounds. Requiring a recent tagged
+    # point makes "no current evidence about this miner" disqualifying
+    # for a G1 seat, independent of what the stale average says.
+    #
+    # A 0.0 written by `finalize_round_scores` counts as evidence — the
+    # gate is about whether the validator looked at the miner this
+    # round, not about how well it did. `latest_round_id` returns None
+    # for a UID with no tagged points (schema v1 legacy state), which
+    # fails the gate: unattributable history cannot back a G1 seat.
+    max_stale = int(getattr(eval_cfg, "g1_max_stale_rounds", 1))
+    g1_min_fresh_rid = cur_rid - max_stale * int(cycle_length)
+
+    def _has_recent_history(uid: int) -> bool:
+        return (
+            score_aggregator.record_count(uid) >= 3
+            and score_aggregator.count_distinct_round_ids_in_range(
+                uid, g1_window_min_rid, cur_rid,
+            ) >= 3
+        )
+
+    def _is_fresh(uid: int) -> bool:
+        latest_rid = score_aggregator.latest_round_id(uid)
+        return latest_rid is not None and latest_rid >= g1_min_fresh_rid
+
     ab_qualified = [
-        u for u in ab_uids
-        if score_aggregator.record_count(u) >= 3
-        and score_aggregator.count_distinct_round_ids_in_range(
-            u, g1_window_min_rid, cur_rid,
-        ) >= 3
+        u for u in ab_uids if _has_recent_history(u) and _is_fresh(u)
     ]
+    # Recorded purely so the caller can log who lost a seat to staleness;
+    # does not affect selection.
+    g1_stale_excluded = tuple(
+        u for u in ab_uids if _has_recent_history(u) and not _is_fresh(u)
+    )
     g1 = _rg.select_top_n_by_local_score(
         ab_qualified,
         avg_scores,
@@ -481,6 +518,7 @@ def build_submission_uid_weights(
         weight_group_2=g2,
         cohort_emission=True,
         g1_redirected_to_uid_zero=g1_redirected,
+        g1_stale_excluded=g1_stale_excluded,
     )
 
 

connito/validator/run.py

@@ -1118,6 +1118,15 @@ def run(rank: int, world_size: int, config: ValidatorConfig, pkg_version: str =
                         weight_group_1=list(payload.weight_group_1),
                         weight_group_2=list(payload.weight_group_2),
                     )
+                if payload.g1_stale_excluded:
+                    logger.info(
+                        "(4) g1 freshness gate — dropped stale UIDs",
+                        round_id=pending_round.round_id,
+                        uids=list(payload.g1_stale_excluded),
+                        max_stale_rounds=int(getattr(
+                            config.evaluation, "g1_max_stale_rounds", 1,
+                        )),
+                    )
                 # Mirror the about-to-submit weights into Prometheus so
                 # external aggregators don't have to scrape `/v1/state.json`
                 # to learn what each validator votes on chain. Mirrors the