diff --git a/connito/shared/config.py b/connito/shared/config.py index 071ad82..417bfd9 100644 --- a/connito/shared/config.py +++ b/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 diff --git a/connito/sn_owner/docker/docker-compose.yml b/connito/sn_owner/docker/docker-compose.yml new file mode 100644 index 0000000..1e5c760 --- /dev/null +++ b/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 \ No newline at end of file diff --git a/connito/test/test_build_submission_uid_weights.py b/connito/test/test_build_submission_uid_weights.py index aa359d4..ab7f2ff 100644 --- a/connito/test/test_build_submission_uid_weights.py +++ b/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,) diff --git a/connito/validator/evaluator.py b/connito/validator/evaluator.py index 4209003..e676423 100644 --- a/connito/validator/evaluator.py +++ b/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, ) diff --git a/connito/validator/run.py b/connito/validator/run.py index d6d6ff8..5019787 100644 --- a/connito/validator/run.py +++ b/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