From 28ef92369c1afbed10a9f2916beb5defc687e686 Mon Sep 17 00:00:00 2001 From: George Date: Mon, 20 Jul 2026 21:43:06 -0400 Subject: [PATCH 1/4] =?UTF-8?q?=E2=AC=86=EF=B8=8F=20deps:=20bittensor=2010?= =?UTF-8?q?.5=20+=20btcli=209.23=20for=20multi-mechanism=20chain=20runtime?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit bittensor 10.1.0 can no longer query the live chain: the multi-mechanism runtime migration wrapped netuid storage keys in a NetUid newtype, so Subtensor.metagraph(netuid) dies with "Invalid type for data: 102 of type " on both finney and archive. Validator bootstrap calls metagraph() (run._sync_lite_metagraph), so any :stable container restart crash-loops fleet-wide. Bump to bittensor 10.5.0 / bittensor-cli 9.23.2 and move the async-substrate-interface pin to the cyscale-based 2.2.1 (both new pins require it; the old scalecodec-era 1.6.4 hack is obsolete — btcli no longer pulls py-scale-codec). Verified in a clean python:3.10 container and layered over the :stable image on a live host (137.184.170.154): deps resolve with no scalecodec, all connito validator/shared modules import, metagraph returns n=256 on netuid 102, and the validator bootstraps through chain-commit fetch and model load. bittensor 11.0.0 was rejected: it removes Subtensor.metagraph and needs a real migration. Co-Authored-By: Claude Fable 5 --- requirements.txt | 15 +++++++++------ 1 file changed, 9 insertions(+), 6 deletions(-) diff --git a/requirements.txt b/requirements.txt index d38ea70..9b9ef0d 100644 --- a/requirements.txt +++ b/requirements.txt @@ -53,12 +53,15 @@ fsspec==2024.6.1 # ----------------------------- # Blockchain # ----------------------------- -bittensor==10.1.0 -bittensor-cli==9.18.0 -# Pinned to pre-cyscale release. Newer async-substrate-interface versions -# migrated scalecodec → cyscale and added an import-time check that fails -# when bittensor-cli's transitive substrate-interface also pulls scalecodec. -async-substrate-interface==1.6.4 +# 10.1 predates the chain's multi-mechanism runtime migration: its storage-key +# encoding sends netuid as a bare int where the runtime now expects the NetUid +# newtype, so metagraph()/btcli-overview fail against live finney+archive. +bittensor==10.5.0 +bittensor-cli==9.23.2 +# Both pins above require the cyscale-based async-substrate-interface 2.x, +# which fails at import if py-scale-codec is also installed — do not +# reintroduce a scalecodec pin alongside these. +async-substrate-interface==2.2.1 # ----------------------------- # Observability From d7e979c4df2db2f873910dbce941195acb572602 Mon Sep 17 00:00:00 2001 From: George Date: Tue, 21 Jul 2026 10:52:37 -0400 Subject: [PATCH 2/4] =?UTF-8?q?=F0=9F=94=92=20eval:=20close=20exp=5Flegal?= =?UTF-8?q?=20memorization=20holes=20(seeded=20pick=20+=20data-quality=20g?= =?UTF-8?q?ates)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Miners scored val_loss ≈ 0.04 vs baselines of 4.5–12.5 (perplexity ~1.04) with all miners within 0.0006 of each other. Root cause is the eval data layer, not the seed: the MinerCommit2 block-hash seed (PR #175/#177) was verified correct on chain for all completed rounds. Validated on the live validator by reproducing the round-8670446 eval stream with production code: - exp_legal still ran the legacy head-of-stream path (§2 shortcut of docs/exp-legal-migration-plan.md) — the whole draw came from the head of one sub-file; - 38% of streamed Multi_Legal_Pile rows have empty text → all-padding batches → NaN loss → silently excluded from the scored divisor; - 75% of non-empty rows share an identical 200-char prefix (templated boilerplate; zlib 0.364 vs c4 0.527); - truncation always scored each document's first 1024 tokens — its most templated region. Changes: - eval_shard_pick: row_count_source="verified_table" (frozen per-shard counts double as the shard allowlist), leaf_name_pattern override, and load_builder script-bypass (Multi_Legal_Pile's builder streams files from external repos and needs trust_remote_code; the generic json builder reads the pinned native shard directly). Register (joelniklaus/Multi_Legal_Pile, all_all) @ 911e1d21 with all 29 native shards above the 10k-row headroom floor (counted by full decompress). - dataloader: eval-only deterministic gates eval_min_text_chars=200 and eval_dedup_prefix_chars=200 (exact-prefix set — builtin hash() is per-process randomized and could break cross-validator consensus). - tokenize_windowed: long docs contribute a content-hash-derived window instead of the boilerplate prefix (no RNG, consensus-safe). Adopted by the default dataset class and exp_legal; exp_math left for follow-up. - exp_legal config: eval_source_seeded_shard_pick: true (shortcut retired). Miner training still streams the all_all builder mix, so eval ⊆ training distribution. - evaluate: scored_batches/nan_batches log promoted to INFO — the only production signal that an average is silently excluding NaN batches. Rollout: eval batches change; all validators must upgrade together (same discipline as the shuffle/skip bumps). Co-Authored-By: Claude Fable 5 --- connito/shared/config.py | 19 +++ connito/shared/dataloader.py | 112 ++++++++++++- connito/shared/eval_shard_pick.py | 162 +++++++++++++++++-- connito/shared/evaluate.py | 7 +- connito/test/test_eval_data_quality.py | 211 +++++++++++++++++++++++++ expert_groups/exp_legal/config.yaml | 18 ++- expert_groups/exp_legal/dataset.py | 21 ++- 7 files changed, 515 insertions(+), 35 deletions(-) create mode 100644 connito/test/test_eval_data_quality.py diff --git a/connito/shared/config.py b/connito/shared/config.py index 071ad82..a410e72 100644 --- a/connito/shared/config.py +++ b/connito/shared/config.py @@ -305,6 +305,25 @@ class DataCfg(BaseConfig): # target and a mid-rollout HF re-upload would cause two # validators to pick different rows for the same seed. eval_source_revision_pin: dict[str, str] | None = None + # --- Eval-stream data-quality gates (validator eval path only; the + # miner training stream passes seed=None and is unaffected). Both + # are deterministic, so all validators on the same version keep + # identical eval batches. Same coordinated-rollout discipline as + # the other eval_source_* knobs. + # + # Drop rows whose text is shorter than this many characters after + # strip(). Motivated by Multi_Legal_Pile all_all, where 38% of + # streamed rows have empty text: those tokenize to all-padding + # sequences whose loss is NaN, silently shrinking the scored eval + # sample. 0 disables. + eval_min_text_chars: int = 200 + # Keep only the first row per distinct text prefix of this many + # characters. Templated corpora repeat opening boilerplate across + # documents (measured: 75% of non-empty Multi_Legal_Pile rows share + # an identical 200-char prefix); scoring duplicates hands + # template-memorizing miners near-zero loss on "unseen" rows. + # 0 disables. + eval_dedup_prefix_chars: int = 200 @model_validator(mode="after") def _validate_dataset_sources(self): diff --git a/connito/shared/dataloader.py b/connito/shared/dataloader.py index e565bb9..3768298 100644 --- a/connito/shared/dataloader.py +++ b/connito/shared/dataloader.py @@ -45,6 +45,90 @@ def _fractional_index_filter(_example, idx: int, seed: str | int, threshold: int return score <= threshold +def _min_text_chars_filter(example: dict[str, Any], min_chars: int) -> bool: + """Drop rows whose text is empty or trivially short. + + Eval-path only. Empirically (Multi_Legal_Pile all_all, 2026-07-21) + 38% of streamed rows carry a completely empty `text`; those rows + tokenize to all-padding sequences whose labels are fully masked, so + they produce NaN eval loss and silently drop out of the scored-batch + divisor — shrinking an already tiny eval sample. The distribution is + bimodal (0 chars or ≥200), so a low threshold removes exactly the + degenerate rows without biasing content selection. + """ + return len(str(example.get("text", "")).strip()) >= min_chars + + +class _PrefixDedupFilter: + """Keep only the first row for each distinct text prefix. + + Eval-path only, and only sound single-worker: the `seen` set lives in + this instance, so the eval dataloader must iterate the stream in one + process (the eval loader runs `num_workers<=1`; see `get_dataloader`). + + Rationale: templated corpora repeat their opening boilerplate across + documents — measured 75% of non-empty Multi_Legal_Pile rows sharing an + identical 200-char prefix. Scoring many near-identical rows lets a + miner fine-tuned on the template reach near-zero loss on "unseen" + documents. Deduplicating by prefix keeps one representative per + template instead of a batch full of copies. Deterministic given a + deterministic input stream (same seed → same order → same survivors). + """ + + def __init__(self, prefix_chars: int): + self.prefix_chars = int(prefix_chars) + # Exact prefixes, not `hash()` digests: builtin str hashing is + # per-process randomized, so two validators could disagree on a + # collision. The eval stream retains only ~thousands of rows, so + # exact storage is a few hundred KB at worst. + self.seen: set[str] = set() + + def __call__(self, example: dict[str, Any]) -> bool: + prefix = str(example.get("text", ""))[: self.prefix_chars] + if prefix in self.seen: + return False + self.seen.add(prefix) + return True + + +def tokenize_windowed( + text: str, tokenizer: PreTrainedTokenizerBase, sequence_length: int +) -> dict[str, list]: + """Tokenize `text`, sampling a deterministic window from long documents. + + The previous behavior (`truncation=True`) always scored/trained on a + document's FIRST `sequence_length` tokens. For templated corpora + (legal filings, papers) the prefix is the most boilerplate-heavy, + most predictable region of the document — document bodies never + entered the pipeline at all. This helper instead: + + - short documents (≤ sequence_length tokens): pad to length, as before; + - long documents: take a `sequence_length` window whose start is + derived from the text's own content hash — deterministic for every + validator (consensus-safe: no RNG, no config), uniform-ish across + the document, and not influenceable by the validator. + + The raw text is capped at `sequence_length * 40` chars before + tokenizing to bound tokenizer cost on pathological documents (real + corpora run 3–8 chars/token, so the cap only bites degenerate input). + """ + capped = str(text)[: sequence_length * 40] + ids = tokenizer(capped, truncation=False, add_special_tokens=True)["input_ids"] + if len(ids) > sequence_length: + span = len(ids) - sequence_length + start = h256_int("token_window", capped) % (span + 1) + window = ids[start : start + sequence_length] + return {"input_ids": window, "attention_mask": [1] * sequence_length} + pad_id = tokenizer.pad_token_id + if pad_id is None: + pad_id = tokenizer.eos_token_id if tokenizer.eos_token_id is not None else 0 + n = len(ids) + return { + "input_ids": ids + [pad_id] * (sequence_length - n), + "attention_mask": [1] * n + [0] * (sequence_length - n), + } + + # ----------------------------- # Dataset # ----------------------------- @@ -86,7 +170,7 @@ def tokenize_and_format( example: dict[str, str], tokenizer: PreTrainedTokenizerBase, sequence_length: int ) -> dict[str, list]: text = example.get("text", "") - return tokenizer(text, truncation=True, max_length=sequence_length, padding="max_length") # type: ignore + return tokenize_windowed(text, tokenizer, sequence_length) @classmethod def get_tokenised_dataset( @@ -272,6 +356,20 @@ def ensure_string(example: dict[str, Any], source_text_column: str): features=common_features, ) + # Eval-path data-quality gate (seed is None on the miner + # training path, which stays byte-identical). Applied + # per-source and before interleave so the source weights + # keep describing *usable* rows — an unfiltered source with + # 38% empty rows would otherwise contribute 38% NaN batches + # at its configured weight. + eval_min_text_chars = int( + getattr(config.task.exp.data, "eval_min_text_chars", 0) or 0 + ) + if seed is not None and eval_min_text_chars > 0: + source_split = source_split.filter( + partial(_min_text_chars_filter, min_chars=eval_min_text_chars) + ) + if pick is not None: # In-shard offset goes here so the validator's read # window lands at a random depth inside the chosen @@ -351,6 +449,18 @@ def ensure_string(example: dict[str, Any], source_text_column: str): logger.debug("Interleaving dataset sources", probabilities=probabilities) split = interleave_datasets(dataset_splits, probabilities=probabilities, seed=int_seed) + # Eval-path template dedup: drop rows repeating an already-seen + # text prefix (templated corpora open millions of documents with + # the same boilerplate — see `_PrefixDedupFilter`). Runs after + # interleave so the dedup window spans the whole eval stream, and + # before the fractional filter so surviving indices stay + # deterministic for every validator. + eval_dedup_prefix_chars = int( + getattr(config.task.exp.data, "eval_dedup_prefix_chars", 0) or 0 + ) + if seed is not None and eval_dedup_prefix_chars > 0: + split = split.filter(_PrefixDedupFilter(eval_dedup_prefix_chars)) + # Optional deterministic subsampling based on (seed, fraction) # Applied *before* sharding on the streaming iterable. if seed is not None and fraction is not None and fraction < 1.0: diff --git a/connito/shared/eval_shard_pick.py b/connito/shared/eval_shard_pick.py index dcb0017..6902728 100644 --- a/connito/shared/eval_shard_pick.py +++ b/connito/shared/eval_shard_pick.py @@ -142,10 +142,25 @@ class _SourceShardPolicy: path_prefix: str path_suffix: tuple[str, ...] revision: str - row_count_source: str # "constant" | "parquet_footer" + row_count_source: str # "constant" | "parquet_footer" | "verified_table" safe_floor_rows: int | None = None min_headroom_rows: int = 10_000 verified_shard_rows: dict[str, int] = field(default_factory=dict) + # Override for `_SHARD_NAME_FILTER` when a source's data files don't + # follow the `train`/`part_` leaf naming convention (e.g. + # Multi_Legal_Pile ships `data///.jsonl.xz`). + # Only consulted for listing-based policies; `verified_table` + # policies take their shard list from the table and skip name + # filtering entirely. + leaf_name_pattern: str | None = None + # When set (e.g. "json"), `load_streaming_shard` loads the shard via + # the named generic builder against a resolved `hf_hub_url` instead + # of `load_dataset(repo_id, data_files=...)`. Required for repos + # that ship a custom loading script: passing the repo id would + # execute the script (defeating both the shard pick and the + # `trust_remote_code` opt-out), while the generic builder reads the + # raw file directly. + load_builder: str | None = None # Known sources. Add new entries here, NOT via config — the consensus @@ -184,6 +199,61 @@ class _SourceShardPolicy: # count per shard at pick time. min_headroom_rows=10_000, ), + ("joelniklaus/Multi_Legal_Pile", "all_all"): _SourceShardPolicy( + # The repo's NATIVE data files. Deliberately NOT the `all_all` + # builder-script mix: the script streams additional files from + # external repos (joelito/eurlex_resources, legal-mc4, …) that + # can't be pinned or row-counted here, and executing it requires + # `trust_remote_code`. The eval pool is therefore a (large) + # subset of the miner-training distribution — acceptable: eval ⊆ + # train, and the native corpus is tens of GB. + path_prefix="data/", + path_suffix=(".jsonl.xz",), + # Pinned at registration time (2026-07-21). Bump together with a + # re-count of the table if the dataset is ever re-uploaded. + revision="911e1d214162fd11d2c78d3f1428cbfcbe07782c", + row_count_source="verified_table", + min_headroom_rows=10_000, + # `.jsonl.xz` has no footer — rows counted once offline (full + # decompress of every shard at the pinned revision, 2026-07-21) + # and frozen here. The table doubles as the allowlist; four + # shards with ≤10k rows (denmark_ddsc caselaw 4 442, + # en switzerland_lexfind 147, belgium_jurportal 2 221, + # it switzerland_lexfind 5 642) are deliberately left out — no + # safe offset exists above the headroom floor. + load_builder="json", + verified_shard_rows={ + "data/bg/legislation/bulgaria_marcell.jsonl.xz": 29_549, + "data/cs/caselaw/czechia_constitutional_court.jsonl.xz": 73_086, + "data/cs/caselaw/czechia_supreme_administrative_court.jsonl.xz": 52_660, + "data/cs/caselaw/czechia_supreme_court.jsonl.xz": 111_977, + "data/da/legislation/denmark_ddsc.jsonl.xz": 64_043, + "data/de/caselaw/germany_openlegaldata.jsonl.xz": 201_676, + "data/de/caselaw/switzerland_entscheidsuche.jsonl.xz": 308_612, + "data/de/legislation/germany_openlegaldata.jsonl.xz": 52_918, + "data/de/legislation/switzerland_lexfind.jsonl.xz": 16_981, + "data/en/legislation/uk_uk_lex.jsonl.xz": 36_499, + "data/fr/caselaw/france_cass.jsonl.xz": 113_844, + "data/fr/caselaw/luxembourg_judoc.jsonl.xz": 37_902, + "data/fr/caselaw/switzerland_entscheidsuche.jsonl.xz": 237_734, + "data/fr/legislation/belgium_ejustice.jsonl.xz": 10_613, + "data/fr/legislation/switzerland_lexfind.jsonl.xz": 10_680, + "data/hu/legislation/hungary_marcell.jsonl.xz": 26_821, + "data/it/caselaw/switzerland_entscheidsuche.jsonl.xz": 69_653, + "data/nl/legislation/belgium_ejustice.jsonl.xz": 10_556, + "data/pl/legislation/poland_marcell.jsonl.xz": 27_485, + "data/pt/caselaw/brazil_cjpg_0.jsonl.xz": 3_489_624, + "data/pt/caselaw/brazil_cjpg_1.jsonl.xz": 3_213_178, + "data/pt/caselaw/brazil_cjpg_2.jsonl.xz": 3_094_216, + "data/pt/caselaw/brazil_cjpg_3.jsonl.xz": 3_019_375, + "data/pt/caselaw/brazil_cjpg_4.jsonl.xz": 1_252_241, + "data/pt/caselaw/brazil_creta.jsonl.xz": 3_128_292, + "data/pt/caselaw/brazil_rulingbr.jsonl.xz": 10_623, + "data/ro/legislation/romania_marcell.jsonl.xz": 163_264, + "data/sk/legislation/slovakia_marcell.jsonl.xz": 13_055, + "data/sl/legislation/slovenia_marcell.jsonl.xz": 24_445, + }, + ), } @@ -199,7 +269,7 @@ def _validate_policy(key: tuple[str, str | None], policy: _SourceShardPolicy) -> biased samples or collapses rounds. """ repo_id, name = key - if policy.row_count_source not in {"constant", "parquet_footer"}: + if policy.row_count_source not in {"constant", "parquet_footer", "verified_table"}: raise ValueError( f"Policy {repo_id}/{name}: unknown row_count_source " f"{policy.row_count_source!r}" @@ -208,6 +278,33 @@ def _validate_policy(key: tuple[str, str | None], policy: _SourceShardPolicy) -> raise ValueError( f"Policy {repo_id}/{name}: min_headroom_rows must be > 0" ) + if policy.row_count_source == "verified_table": + # The table IS the shard allowlist: every listed shard must have + # a row count that leaves at least one valid offset after the + # headroom is reserved. Shards too small for the eval pipeline's + # per-round consumption must be left out of the table, not + # zero-bounded at pick time. + if not policy.verified_shard_rows: + raise ValueError( + f"Policy {repo_id}/{name}: row_count_source='verified_table' " + f"requires a non-empty verified_shard_rows table (it doubles " + f"as the shard allowlist)" + ) + for shard_path, rows in policy.verified_shard_rows.items(): + if rows <= policy.min_headroom_rows: + raise ValueError( + f"Policy {repo_id}/{name}: verified shard {shard_path!r} " + f"has {rows} rows ≤ min_headroom_rows " + f"({policy.min_headroom_rows}); no safe offset exists. " + f"Remove it from the table." + ) + if not shard_path.startswith(policy.path_prefix) or not shard_path.endswith( + policy.path_suffix + ): + raise ValueError( + f"Policy {repo_id}/{name}: table entry {shard_path!r} does " + f"not match path_prefix/path_suffix — typo in the table?" + ) if policy.row_count_source == "constant": if policy.safe_floor_rows is None or policy.safe_floor_rows <= 0: raise ValueError( @@ -296,7 +393,19 @@ def _list_shards(repo_id: str, name: str | None, revision: str) -> tuple[str, .. the same revision gets the same tuple. """ policy = _policy_for(repo_id, name) + if policy.row_count_source == "verified_table": + # The frozen table doubles as the shard allowlist. Listing from + # the HF API here would re-introduce the consensus hazard the + # table exists to remove (a re-uploaded repo changing the list + # under our feet); the pinned revision + table are the source + # of truth. + return tuple(sorted(policy.verified_shard_rows)) info = HfApi().dataset_info(repo_id, revision=revision) + name_filter = ( + re.compile(policy.leaf_name_pattern, re.IGNORECASE) + if policy.leaf_name_pattern + else _SHARD_NAME_FILTER + ) filtered = [] for f in info.siblings: rf = f.rfilename @@ -309,7 +418,7 @@ def _list_shards(repo_id: str, name: str | None, revision: str) -> tuple[str, .. # (e.g. metadata, sidecar files). The "train" / "part_" check # is per-source-format and intentionally narrow. leaf = rf.split("/")[-1] - if not _SHARD_NAME_FILTER.search(leaf): + if not name_filter.search(leaf): continue filtered.append(rf) if not filtered: @@ -366,6 +475,13 @@ def _resolve_offset_bound( assert policy.safe_floor_rows is not None return policy.safe_floor_rows + if policy.row_count_source == "verified_table": + # Module-load validation guarantees the shard is in the table + # with rows > min_headroom_rows (the table is the allowlist + # `_list_shards` picks from). + rows = policy.verified_shard_rows[shard_path] + return rows - policy.min_headroom_rows + if policy.row_count_source == "parquet_footer": actual_rows = _shard_rows_via_parquet_footer(repo_id, revision, shard_path) bound = actual_rows - policy.min_headroom_rows @@ -406,6 +522,9 @@ class ShardPick: offset_bound: int shard_rows: int # for the constant path this equals offset_bound (we don't know the true count) in_shard_offset: int + # Propagated from the policy: when set, `load_streaming_shard` loads + # via this generic builder against a resolved URL (script-bypass). + load_builder: str | None = None def pick_shard_for_source( @@ -456,8 +575,8 @@ def pick_shard_for_source( # For constant policies we don't know the actual shard size; # surface `offset_bound` as `shard_rows` so older callers (notebook # / tests) that check "offset < shard_rows" still see the right - # invariant. The parquet path can fill in the real count. - if policy.row_count_source == "parquet_footer": + # invariant. The parquet and verified-table paths know the real count. + if policy.row_count_source in {"parquet_footer", "verified_table"}: actual_shard_rows = offset_bound + policy.min_headroom_rows else: actual_shard_rows = offset_bound @@ -470,6 +589,7 @@ def pick_shard_for_source( offset_bound=offset_bound, shard_rows=actual_shard_rows, in_shard_offset=offset, + load_builder=policy.load_builder, ) @@ -495,14 +615,30 @@ def load_streaming_shard( # helps unit tests. from datasets import load_dataset - load_kwargs: dict[str, Any] = { - "data_files": [pick.shard_path], - "streaming": True, - "revision": pick.revision, - } - if extra_load_kwargs: - load_kwargs.update(extra_load_kwargs) - ds = load_dataset(pick.repo_id, **load_kwargs) + if pick.load_builder: + # Script-bypass path: the repo ships a custom loading script, so + # `load_dataset(repo_id, ...)` would execute it (and for + # Multi_Legal_Pile, stream entirely different files from external + # repos). Loading the raw file through a generic builder reads + # exactly the picked shard and needs no `trust_remote_code`. + from huggingface_hub import hf_hub_url + + url = hf_hub_url( + pick.repo_id, pick.shard_path, repo_type="dataset", revision=pick.revision + ) + load_kwargs = {"data_files": [url], "streaming": True} + if extra_load_kwargs: + load_kwargs.update(extra_load_kwargs) + ds = load_dataset(pick.load_builder, **load_kwargs) + else: + load_kwargs = { + "data_files": [pick.shard_path], + "streaming": True, + "revision": pick.revision, + } + if extra_load_kwargs: + load_kwargs.update(extra_load_kwargs) + ds = load_dataset(pick.repo_id, **load_kwargs) if split_name in ds: return ds[split_name] # `data_files=` with a single file lands the rows under "train" by diff --git a/connito/shared/evaluate.py b/connito/shared/evaluate.py index 196f223..2d0c134 100644 --- a/connito/shared/evaluate.py +++ b/connito/shared/evaluate.py @@ -117,7 +117,12 @@ def evaluate_model( if max_eval_batches is not None and batch_step >= max_eval_batches: break - logger.debug( + # INFO, not debug: `nan_batches` is the only production-visible + # signal distinguishing "miner scored on all batches" from + # "miner's average silently excludes NaN batches" (degenerate + # eval rows and overflow-on-OOD submissions both land here). + # Ops must be able to read this off a live validator's logs. + logger.info( "eval loss", loss_sum=round(loss_sum, 4), aux_loss_sum=round(aux_loss_sum, 4), diff --git a/connito/test/test_eval_data_quality.py b/connito/test/test_eval_data_quality.py new file mode 100644 index 0000000..94575d8 --- /dev/null +++ b/connito/test/test_eval_data_quality.py @@ -0,0 +1,211 @@ +"""Unit tests for the eval-data-quality gates added for exp_legal. + +Covers the pure-function layer only — no HF network access: + + - `_min_text_chars_filter`: drops empty / trivially-short rows + (Multi_Legal_Pile all_all streams 38% empty-text rows; measured + 2026-07-21 on the first 800 rows). + - `_PrefixDedupFilter`: keeps only the first row per distinct text + prefix (75% of non-empty M_L_P rows share an identical 200-char + prefix with another row), and does so deterministically without + relying on per-process `hash()`. + - `tokenize_windowed`: content-hash window sampling for documents + longer than `sequence_length`; prefix + padding for short ones. + - `_SourceShardPolicy` `row_count_source="verified_table"`: table + doubles as allowlist, validation rejects under-headroom shards and + prefix typos, offset bound honors per-shard counts. + +Integration behavior (streaming filter placement, interleave order, +seeded shard-pick equivalence for Multi_Legal_Pile) is exercised by the +notebook-driven checks described in docs/exp-legal-migration-plan.md §2 +and the P1 determinism test in test_eval_source_skip.py. +""" +from __future__ import annotations + +import pytest + +from connito.shared.dataloader import ( + _min_text_chars_filter, + _PrefixDedupFilter, + tokenize_windowed, +) +from connito.shared.eval_shard_pick import ( + _SourceShardPolicy, + _resolve_offset_bound, + _validate_policy, +) + + +# --------------------------------------------------------------------------- +# _min_text_chars_filter +# --------------------------------------------------------------------------- + +def test_min_chars_drops_empty_and_whitespace_rows(): + assert not _min_text_chars_filter({"text": ""}, min_chars=200) + assert not _min_text_chars_filter({"text": " \n\t "}, min_chars=200) + assert not _min_text_chars_filter({}, min_chars=200) + + +def test_min_chars_keeps_substantial_rows(): + assert _min_text_chars_filter({"text": "x" * 200}, min_chars=200) + assert _min_text_chars_filter({"text": " " + "x" * 200 + " "}, min_chars=200) + + +def test_min_chars_boundary_is_inclusive(): + assert _min_text_chars_filter({"text": "x" * 200}, min_chars=200) + assert not _min_text_chars_filter({"text": "x" * 199}, min_chars=200) + + +# --------------------------------------------------------------------------- +# _PrefixDedupFilter +# --------------------------------------------------------------------------- + +def test_prefix_dedup_keeps_first_occurrence_only(): + f = _PrefixDedupFilter(prefix_chars=10) + boiler = "IN THE SUPREME COURT OF ..." + assert f({"text": boiler + " case one"}) + assert not f({"text": boiler + " case two"}) + assert f({"text": "completely different text"}) + + +def test_prefix_dedup_distinguishes_beyond_prefix_window(): + f = _PrefixDedupFilter(prefix_chars=100) + a = "shared start " + "a" * 200 + b = "shared start " + "b" * 200 + # Prefix window (100 chars) reaches into the differing region. + assert f({"text": a}) + assert f({"text": b}) + + +def test_prefix_dedup_state_is_per_instance(): + text = {"text": "same row"} + assert _PrefixDedupFilter(50)(text) + assert _PrefixDedupFilter(50)(text) # fresh instance, fresh set + + +# --------------------------------------------------------------------------- +# tokenize_windowed +# --------------------------------------------------------------------------- + + +class _StubTokenizer: + """1-token-per-word tokenizer; ids are stable per word.""" + + pad_token_id = 0 + eos_token_id = 2 + + def __call__(self, text, truncation=False, add_special_tokens=True, **_kw): + ids = [7 + (len(w) % 50) for w in text.split()] + return {"input_ids": ids} + + +def test_windowed_short_doc_pads_to_length(): + tok = _StubTokenizer() + out = tokenize_windowed("three word doc", tok, sequence_length=8) + assert len(out["input_ids"]) == 8 + assert out["attention_mask"] == [1, 1, 1, 0, 0, 0, 0, 0] + assert out["input_ids"][3:] == [tok.pad_token_id] * 5 + + +def test_windowed_long_doc_returns_full_window(): + tok = _StubTokenizer() + text = " ".join(f"w{i}" for i in range(500)) + out = tokenize_windowed(text, tok, sequence_length=64) + assert len(out["input_ids"]) == 64 + assert out["attention_mask"] == [1] * 64 + + +def test_windowed_is_deterministic_per_content(): + tok = _StubTokenizer() + text = " ".join(f"w{i}" for i in range(500)) + a = tokenize_windowed(text, tok, sequence_length=64) + b = tokenize_windowed(text, tok, sequence_length=64) + assert a == b + + +def test_windowed_start_varies_across_documents(): + class PositionTokenizer(_StubTokenizer): + # ids encode stream position, so the returned window is + # `range(start, start+seq)` and directly reveals the chosen start. + def __call__(self, text, truncation=False, add_special_tokens=True, **_kw): + return {"input_ids": list(range(len(text.split())))} + + tok = PositionTokenizer() + full = [" ".join(f"d{k}w{i}" for i in range(500)) for k in range(20)] + starts = {tokenize_windowed(t, tok, sequence_length=64)["input_ids"][0] for t in full} + # Content-hash-derived starts over a 437-position span: 20 documents + # should land on many distinct starts (expected ≈19.6 distinct). + # A constant-start implementation collapses this to 1. + assert len(starts) >= 10 + + +def test_windowed_missing_pad_token_falls_back_to_eos(): + class NoPad(_StubTokenizer): + pad_token_id = None + + out = tokenize_windowed("one two", NoPad(), sequence_length=4) + assert out["input_ids"][2:] == [NoPad.eos_token_id] * 2 + + +# --------------------------------------------------------------------------- +# verified_table shard policy +# --------------------------------------------------------------------------- + +_KEY = ("example/repo", "subset") + + +def _table_policy(**overrides): + kwargs = dict( + path_prefix="data/", + path_suffix=(".jsonl.xz",), + revision="deadbeef", + row_count_source="verified_table", + min_headroom_rows=5_000, + verified_shard_rows={ + "data/a/one.jsonl.xz": 50_000, + "data/b/two.jsonl.xz": 12_000, + }, + ) + kwargs.update(overrides) + return _SourceShardPolicy(**kwargs) + + +def test_verified_table_policy_validates(): + _validate_policy(_KEY, _table_policy()) # should not raise + + +def test_verified_table_rejects_empty_table(): + with pytest.raises(ValueError, match="non-empty verified_shard_rows"): + _validate_policy(_KEY, _table_policy(verified_shard_rows={})) + + +def test_verified_table_rejects_under_headroom_shard(): + with pytest.raises(ValueError, match="no safe offset"): + _validate_policy( + _KEY, + _table_policy( + verified_shard_rows={"data/a/one.jsonl.xz": 4_000}, + ), + ) + + +def test_verified_table_rejects_prefix_typo(): + with pytest.raises(ValueError, match="path_prefix/path_suffix"): + _validate_policy( + _KEY, + _table_policy( + verified_shard_rows={"wrong/one.jsonl.xz": 50_000}, + ), + ) + + +def test_verified_table_offset_bound_is_per_shard(): + policy = _table_policy() + bound = _resolve_offset_bound( + repo_id=_KEY[0], + name=_KEY[1], + revision="deadbeef", + shard_path="data/b/two.jsonl.xz", + policy=policy, + ) + assert bound == 12_000 - 5_000 diff --git a/expert_groups/exp_legal/config.yaml b/expert_groups/exp_legal/config.yaml index d1307ae..9c7265c 100644 --- a/expert_groups/exp_legal/config.yaml +++ b/expert_groups/exp_legal/config.yaml @@ -43,11 +43,13 @@ data: world_size: 10 rank: 1 dataset_class: "expert_groups.exp_legal.dataset:StreamingTorchDataset" - # TODO(exp-legal-launch): register (joelniklaus/Multi_Legal_Pile, all_all) - # in connito/shared/eval_shard_pick.py:_KNOWN_SOURCES with verified - # shard rows for all 33 files, then flip this back to true. Until then, - # the validator's legacy head-of-stream eval path is used for this - # group; consensus determinism across validators is weaker than the - # seeded-pick guarantee exp_math already enjoys. See - # docs/exp-legal-migration-plan.md §2. - eval_source_seeded_shard_pick: false + # (joelniklaus/Multi_Legal_Pile, all_all) is registered in + # connito/shared/eval_shard_pick.py:_KNOWN_SOURCES with a + # verified-row-count table over the repo's native data/ shards + # (row_count_source="verified_table", pinned revision, script-bypass + # json loading — the launch-period legacy-path shortcut from + # docs/exp-legal-migration-plan.md §2 is retired). Note the eval + # pool is the native Multi_Legal_Pile files; the miner TRAINING + # stream still goes through the all_all builder script, so eval + # remains a subset of the training distribution. + eval_source_seeded_shard_pick: true diff --git a/expert_groups/exp_legal/dataset.py b/expert_groups/exp_legal/dataset.py index 9739af3..2bba265 100644 --- a/expert_groups/exp_legal/dataset.py +++ b/expert_groups/exp_legal/dataset.py @@ -2,7 +2,7 @@ from transformers import PreTrainedTokenizerBase -from connito.shared.dataloader import DefaultStreamingTorchDataset +from connito.shared.dataloader import DefaultStreamingTorchDataset, tokenize_windowed # ------------------------------------------------------------- @@ -20,19 +20,16 @@ def tokenize_and_format( Processes raw text for Continuous Pre-Training (CPT). Bypasses the chat template since Multi_Legal_Pile and C4 are raw text corpora, not conversational data. + + Long documents get a content-hash-derived window instead of the + prefix: legal filings front-load their most templated text + (headers, procedural boilerplate), so always scoring the first + `sequence_length` tokens let template-memorizing miners reach + near-zero eval loss. See `tokenize_windowed` for the mechanism + and its consensus properties. """ - # 1) Safely extract the raw text we aligned in dataloader.py text = str(example.get("text", "")) - - # 2) Tokenize text directly (no chat template) - toks = tokenizer( - text, - truncation=True, - max_length=sequence_length, - padding="max_length", - add_special_tokens=True, # Important: Ensures BOS/EOS tokens are added - ) - + toks = tokenize_windowed(text, tokenizer, sequence_length) return { "input_ids": toks["input_ids"], "attention_mask": toks["attention_mask"], From 6250320f4895501d4b236ede9cdd54ef234b1d95 Mon Sep 17 00:00:00 2001 From: George Date: Tue, 21 Jul 2026 15:19:57 -0400 Subject: [PATCH 3/4] =?UTF-8?q?=F0=9F=9A=91=20deps:=20unblock=20fresh=20re?= =?UTF-8?q?solve=20for=20bittensor=2010.5=20(requests=202.33,=20typer=200.?= =?UTF-8?q?26)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit CI's clean-image build caught two conflicts the layered container test could not: bittensor 10.5.0 requires requests>=2.33 (repo pinned 2.32.3) and bittensor-cli 9.23.2 requires typer~=0.26 (repo pinned 0.24.1). Verified with a full pip dry-run resolve of requirements.txt in a clean python:3.10 container against the cu124 index. Co-Authored-By: Claude Fable 5 --- requirements.txt | 6 ++++-- 1 file changed, 4 insertions(+), 2 deletions(-) diff --git a/requirements.txt b/requirements.txt index 9b9ef0d..107ae60 100644 --- a/requirements.txt +++ b/requirements.txt @@ -11,7 +11,8 @@ structlog==24.1.0 prometheus-client==0.20.0 psutil==6.1.1 PyYAML==6.0.1 -typer==0.24.1 +# btcli 9.23 requires typer~=0.26 (fresh-resolve conflict; bumped with the bittensor 10.5 move) +typer==0.26.0 python-dotenv==1.1.0 cryptography==46.0.5 varint==1.0.2 @@ -35,7 +36,8 @@ unsloth @ git+https://github.com/unslothai/unsloth.git; platform_system == "Linu # ----------------------------- fastapi==0.110.1 uvicorn[standard]==0.30.0 -requests==2.32.3 +# bittensor 10.5 requires requests>=2.33 (fresh-resolve conflict caught by CI 2026-07-21) +requests==2.33.0 aiofiles==25.1.0 python-multipart==0.0.22 From 0fe0c9d50f5ae3ff7b68a1306bcad0bf4806e6f8 Mon Sep 17 00:00:00 2001 From: George Date: Thu, 23 Jul 2026 16:07:29 -0400 Subject: [PATCH 4/4] =?UTF-8?q?=F0=9F=93=8A=20telemetry:=20cycle-consisten?= =?UTF-8?q?t=20per-miner=20attribution=20for=20the=20dashboard?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Additive only — no changes to evaluation, scoring, or weights. The subnet-api-gateway feature-detects these metrics, so deploy order is free in both directions. New per-miner families (miner_uid-labeled, set from finalize): - validator_miner_last_scored_round_id — round_id of the last finalize verdict (scored / tie-zeroed / validation-failed / freeze-zero). - validator_miner_round_delta — raw (baseline − val_loss)^1.2 signal for evaluated uids; distinct from score_latest (finalized podium score) so the dashboard can show "how close was I" for non-podium miners. - validator_miner_evaluated_commit_info{hf_repo_id, hf_revision} — valued with the freeze round_id; strictly ≤1 labelset per uid (previous labelset evicted under a lock on commit change; KeyError-guarded for the post-restart case). Emission moved INTO finalize_round_scores (after the journal flip): score snapshots (latest/avg/samples/emitted_at) now publish for EVERY verdict uid, not just weight recipients (~1 uid before), and the journal-recovery replay re-publishes everything after a restart — closing the "restart blanks the dashboard until the next full round" gap observed live during the 2026-07-23 cycle-api outage. The redundant snapshot loop in run.py's weight block is removed (WEIGHT_SUBMITTED stays). RoundJournal bumped to v2: adds uid_to_commit (uid -> (hf_repo_id, hf_revision)) captured from uid_to_chain_checkpoint at freeze/finalize; v1 files still load (empty map); future versions rejected. _RecoveryRound hydrates checkpoint stubs so recovered finalizes emit commit info too. Per-round series eviction: round_id label values emitted on validator_round_lifecycle_step / _miners_pending / _scored / _failed / bg_eval_lock_leak_total are tracked in-process and evicted via evict_round_series_before() on run.py's existing journal-prune cutoff — previously every round leaked permanent series. Tests: 11 new (eviction invariant incl. restart simulation, journal v2/v1/future-version handling, finalize emission on live-shaped and recovery rounds, round-series eviction); full related suite (36) green in the staging image. Co-Authored-By: Claude Fable 5 --- connito/shared/telemetry.py | 148 +++++++++ .../test/test_cycle_consistent_telemetry.py | 291 ++++++++++++++++++ connito/validator/background_eval_worker.py | 3 + connito/validator/evaluator.py | 51 +++ connito/validator/round.py | 3 + connito/validator/round_journal.py | 51 ++- connito/validator/run.py | 60 ++-- 7 files changed, 574 insertions(+), 33 deletions(-) create mode 100644 connito/test/test_cycle_consistent_telemetry.py diff --git a/connito/shared/telemetry.py b/connito/shared/telemetry.py index c70847a..c57bcf7 100644 --- a/connito/shared/telemetry.py +++ b/connito/shared/telemetry.py @@ -243,6 +243,39 @@ def set_validator_identity(*, hotkey: str, uid: int | None, version: str, netuid ["miner_uid"], ) +# --- Cycle-consistent per-miner attribution (dashboard contract) ----------- +# The gateway attributes every per-miner sample to the exact evaluation +# round via these three families. All are set from `finalize_round_scores` +# (including its journal-recovery replay path) so a validator restart +# re-publishes them without waiting for a fresh round. +VALIDATOR_MINER_LAST_SCORED_ROUND_ID = Gauge( + "validator_miner_last_scored_round_id", + "round_id of the last round in which this validator wrote a finalize " + "verdict (scored, tie-zeroed, validation-failed, or freeze-zero) for the miner", + ["miner_uid"], +) +VALIDATOR_MINER_ROUND_DELTA = Gauge( + "validator_miner_round_delta", + "Raw per-round improvement signal ((baseline - val_loss) ** 1.2, >= 0) " + "from the miner's most recent evaluated round. Distinct from " + "validator_miner_score_latest, which is the finalized podium rank score.", + ["miner_uid"], +) +VALIDATOR_MINER_EVALUATED_COMMIT_INFO = Gauge( + "validator_miner_evaluated_commit_info", + "round_id in which the labeled (hf_repo_id, hf_revision) was frozen and " + "evaluated for the miner. At most one labelset per miner_uid (old " + "labelsets are evicted on change).", + ["miner_uid", "hf_repo_id", "hf_revision"], +) +# uid -> (hf_repo_id, hf_revision) currently exposed on +# VALIDATOR_MINER_EVALUATED_COMMIT_INFO. Guarded by _COMMIT_INFO_LOCK; used +# to evict the previous labelset when a miner's commit changes, keeping the +# "<= 1 labelset per uid" invariant. After a restart both this dict and the +# registry start empty, so correctness holds without persistence. +_COMMIT_INFO_LOCK = threading.Lock() +_COMMIT_INFO_LABELS: dict[str, tuple[str, str]] = {} + # Per-round lifecycle (background submission validation) VALIDATOR_ROUND_LIFECYCLE_STEP = Gauge( "validator_round_lifecycle_step", @@ -264,6 +297,13 @@ def set_validator_identity(*, hotkey: str, uid: int | None, version: str, netuid "Roster miners that failed download/eval for the round", ["round_id"], ) +# round_id label values ever emitted on the per-round families above (and on +# VALIDATOR_BG_EVAL_LOCK_LEAK_TOTAL). Call sites register via +# `note_round_series`; `evict_round_series_before` removes stale labelsets on +# the same cutoff run.py already uses to prune journals/aggregator entries — +# without this, every round leaves four-plus permanent series behind. +_ROUND_SERIES_LOCK = threading.Lock() +_EMITTED_ROUND_IDS: set[int] = set() VALIDATOR_BG_WORKER_PAUSED = Gauge( "validator_bg_worker_paused", "1 while a background worker is paused on merge_phase_active / eval_window / download_window", @@ -425,6 +465,114 @@ def set_miner_eval_status(miner_uid: int | str, reason: EvalFailureReason | str pass +def set_miner_last_scored_round(miner_uid: int | str, round_id: int) -> None: + """Record the round_id of the last finalize verdict for this miner. + + Best-effort — never raises. Telemetry must not influence scoring. + """ + try: + VALIDATOR_MINER_LAST_SCORED_ROUND_ID.labels(miner_uid=str(miner_uid)).set( + float(int(round_id)) + ) + except Exception: + pass + + +def set_miner_round_delta(miner_uid: int | str, delta: float) -> None: + """Record the raw per-round improvement signal for an evaluated miner. + + Best-effort — never raises. + """ + try: + VALIDATOR_MINER_ROUND_DELTA.labels(miner_uid=str(miner_uid)).set(float(delta)) + except Exception: + pass + + +def set_miner_evaluated_commit( + miner_uid: int | str, hf_repo_id: str, hf_revision: str, round_id: int +) -> None: + """Expose which (hf_repo_id, hf_revision) was frozen + evaluated for the + miner, valued with the round_id it belongs to. + + Enforces at most ONE labelset per miner_uid: when the commit changes, the + previous labelset is removed from the registry before the new one is set, + so the gateway never sees two competing commit rows for a uid. The + KeyError guard covers the post-restart case (tracking dict repopulated + while the registry series was already re-created) and double-eviction. + + Best-effort — never raises. + """ + try: + uid = str(miner_uid) + repo = str(hf_repo_id or "") + rev = str(hf_revision or "") + if not repo or not rev: + return + with _COMMIT_INFO_LOCK: + prev = _COMMIT_INFO_LABELS.get(uid) + if prev is not None and prev != (repo, rev): + try: + VALIDATOR_MINER_EVALUATED_COMMIT_INFO.remove(uid, prev[0], prev[1]) + except KeyError: + pass + VALIDATOR_MINER_EVALUATED_COMMIT_INFO.labels( + miner_uid=uid, hf_repo_id=repo, hf_revision=rev + ).set(float(int(round_id))) + _COMMIT_INFO_LABELS[uid] = (repo, rev) + except Exception: + pass + + +def note_round_series(round_id: int) -> None: + """Register a round_id whose label value was emitted on a per-round + family, so `evict_round_series_before` can remove it later. + + Best-effort — never raises. + """ + try: + with _ROUND_SERIES_LOCK: + _EMITTED_ROUND_IDS.add(int(round_id)) + except Exception: + pass + + +def evict_round_series_before(min_round_id: int) -> int: + """Remove per-round labelsets for every tracked round_id below the + cutoff. Called from run.py's journal/aggregator prune block with the + same cutoff, so metric retention matches on-disk retention. + + Only rounds emitted by THIS process are tracked (the set is in-memory); + series left over from a previous process incarnation don't exist in the + fresh registry either, so nothing is leaked across restarts. + + Returns the number of round_ids evicted. Best-effort — never raises. + """ + removed = 0 + try: + cutoff = int(min_round_id) + with _ROUND_SERIES_LOCK: + stale = [r for r in _EMITTED_ROUND_IDS if r < cutoff] + for r in stale: + rid = str(r) + for family in ( + VALIDATOR_ROUND_LIFECYCLE_STEP, + VALIDATOR_ROUND_MINERS_PENDING, + VALIDATOR_ROUND_MINERS_SCORED, + VALIDATOR_ROUND_MINERS_FAILED, + VALIDATOR_BG_EVAL_LOCK_LEAK_TOTAL, + ): + try: + family.remove(rid) + except KeyError: + pass + _EMITTED_ROUND_IDS.discard(r) + removed += 1 + except Exception: + return removed + return removed + + def set_miner_score_snapshot( miner_uid: int | str, *, diff --git a/connito/test/test_cycle_consistent_telemetry.py b/connito/test/test_cycle_consistent_telemetry.py new file mode 100644 index 0000000..0673030 --- /dev/null +++ b/connito/test/test_cycle_consistent_telemetry.py @@ -0,0 +1,291 @@ +"""Tests for the cycle-consistent miner-telemetry contract. + +Covers: + - `set_miner_evaluated_commit` eviction invariant (≤ 1 labelset per uid, + restart-simulation safe); + - RoundJournal v2 round-trip + v1 backward compatibility; + - `finalize_round_scores` emitting last_scored_round_id / score snapshots + for every verdict uid, round deltas for scored uids, and exactly one + evaluated-commit labelset per checkpointed uid — both on a live-shaped + round and again via the journal-recovery path; + - `evict_round_series_before` removing tracked per-round labelsets. + +The prometheus registry is process-global, so every test uses uids in a +dedicated 9xxx range to avoid colliding with other test modules. +""" +from __future__ import annotations + +import json +from types import SimpleNamespace + +from connito.shared import telemetry as T +from connito.validator import round_journal as RJ +from connito.validator.evaluator import finalize_round_scores + + +def _samples_for(gauge, family_name: str) -> list: + for metric in gauge.collect(): + return [s for s in metric.samples if s.name == family_name] + return [] + + +def _labelsets_for_uid(uid: int) -> list: + return [ + s + for s in _samples_for( + T.VALIDATOR_MINER_EVALUATED_COMMIT_INFO, + "validator_miner_evaluated_commit_info", + ) + if s.labels["miner_uid"] == str(uid) + ] + + +# --------------------------------------------------------------------------- +# Commit-info eviction invariant +# --------------------------------------------------------------------------- + +def test_evaluated_commit_evicts_previous_labelset(): + uid = 9001 + T.set_miner_evaluated_commit(uid, "acme/repo", "rev-a", 100) + T.set_miner_evaluated_commit(uid, "acme/repo", "rev-b", 200) + rows = _labelsets_for_uid(uid) + assert len(rows) == 1 + assert rows[0].labels["hf_revision"] == "rev-b" + assert rows[0].value == 200.0 + + +def test_evaluated_commit_same_labels_updates_value_in_place(): + uid = 9002 + T.set_miner_evaluated_commit(uid, "acme/repo", "rev-x", 100) + T.set_miner_evaluated_commit(uid, "acme/repo", "rev-x", 300) + rows = _labelsets_for_uid(uid) + assert len(rows) == 1 + assert rows[0].value == 300.0 + + +def test_evaluated_commit_restart_simulation_is_safe(): + uid = 9003 + T.set_miner_evaluated_commit(uid, "acme/repo", "rev-a", 100) + # Simulate a restart of the tracking dict only (registry keeps series in + # a real restart neither survives; here the tracking dict being empty + # while the registry still has the old row exercises the KeyError-free + # first write, and the follow-up change exercises eviction rebuilt from + # scratch). + with T._COMMIT_INFO_LOCK: + T._COMMIT_INFO_LABELS.clear() + T.set_miner_evaluated_commit(uid, "acme/repo", "rev-b", 200) + # Old row may linger (tracking was lost) but the invariant re-establishes + # on the NEXT change: + T.set_miner_evaluated_commit(uid, "acme/repo", "rev-c", 300) + rows = _labelsets_for_uid(uid) + revs = {r.labels["hf_revision"] for r in rows} + assert "rev-c" in revs + assert "rev-b" not in revs # evicted by the rev-b -> rev-c change + + +def test_evaluated_commit_skips_empty_repo_or_revision(): + uid = 9004 + T.set_miner_evaluated_commit(uid, "", "rev-a", 100) + T.set_miner_evaluated_commit(uid, "acme/repo", None, 100) # type: ignore[arg-type] + assert _labelsets_for_uid(uid) == [] + + +# --------------------------------------------------------------------------- +# Journal v2 / v1 compatibility +# --------------------------------------------------------------------------- + +def test_journal_v2_roundtrip(tmp_path): + j = RJ.RoundJournal( + round_id=123, + uid_to_hotkey={1: "hk1"}, + scores={1: 2.5}, + scored_uids=(1,), + uid_to_commit={1: ("acme/repo", "deadbeef")}, + finalized=True, + ) + p = tmp_path / "round_123.json" + RJ.write_atomic(p, j) + loaded = RJ.load(p) + assert loaded is not None + assert loaded.schema_version == RJ.SCHEMA_VERSION + assert loaded.uid_to_commit == {1: ("acme/repo", "deadbeef")} + assert loaded.scores == {1: 2.5} + + +def test_journal_v1_loads_with_empty_commit_map(tmp_path): + v1 = { + "round_id": 77, + "uid_to_hotkey": {"5": "hk5"}, + "scores": {"5": 1.25}, + "scored_uids": [5], + "failed_uids": [], + "validation_failed_uids": [], + "freeze_zero_uids": [], + "freeze_zero_hotkeys": {}, + "finalized": False, + "schema_version": 1, + } + p = tmp_path / "round_77.json" + p.write_text(json.dumps(v1), encoding="utf-8") + loaded = RJ.load(p) + assert loaded is not None + assert loaded.uid_to_commit == {} + assert loaded.scores == {5: 1.25} + + +def test_journal_future_version_rejected(tmp_path): + p = tmp_path / "round_88.json" + p.write_text( + json.dumps({"round_id": 88, "schema_version": RJ.SCHEMA_VERSION + 1}), + encoding="utf-8", + ) + try: + RJ.load(p) + except ValueError: + return + raise AssertionError("future schema_version must raise") + + +def test_commit_map_from_checkpoints_skips_incomplete(): + m = RJ.commit_map_from_checkpoints({ + 1: SimpleNamespace(hf_repo_id="a/r", hf_revision="v1"), + 2: SimpleNamespace(hf_repo_id=None, hf_revision="v2"), + 3: SimpleNamespace(hf_repo_id="a/r3", hf_revision=""), + }) + assert m == {1: ("a/r", "v1")} + + +# --------------------------------------------------------------------------- +# finalize_round_scores emission (live-shaped round + recovery path) +# --------------------------------------------------------------------------- + + +class _FakeAggregator: + def __init__(self): + self.rows: dict[int, list[float]] = {} + + def drop_round(self, round_id): + pass + + def add_score(self, *, uid, hotkey, score, round_id): + self.rows.setdefault(int(uid), []).append(float(score)) + + def uid_score_pairs(self, how="avg"): + if how == "latest": + return {u: v[-1] for u, v in self.rows.items()} + return {u: sum(v) / len(v) for u, v in self.rows.items()} + + def record_count(self, uid): + return len(self.rows.get(int(uid), [])) + + +def _make_round(round_id: int, journal_path=None): + return RJ._RecoveryRound( + round_id=round_id, + scores={9101: 2.5, 9102: 1.5, 9103: 0.0}, + scored_uids={9101, 9102, 9103}, + failed_uids=set(), + validation_failed_uids={9104}, + freeze_zero_uids={9105}, + freeze_zero_hotkeys={9105: "hk9105"}, + uid_to_hotkey={u: f"hk{u}" for u in (9101, 9102, 9103, 9104)}, + journal_path=journal_path, + uid_to_chain_checkpoint={ + 9101: SimpleNamespace(hf_repo_id="miner/one", hf_revision="aaa1"), + 9102: SimpleNamespace(hf_repo_id="miner/two", hf_revision="bbb2"), + }, + ) + + +def _gauge_value(gauge, family_name, uid): + for s in _samples_for(gauge, family_name): + if s.labels.get("miner_uid") == str(uid): + return s.value + return None + + +def test_finalize_emits_cycle_consistent_series(tmp_path): + rid = 524_000 + round_obj = _make_round(rid, journal_path=tmp_path / f"round_{rid}.json") + written = finalize_round_scores( + round_obj=round_obj, score_aggregator=_FakeAggregator(), score_path=None, + ) + # Verdict uids: 3 scored + 1 validation_failed + 1 freeze_zero. + assert set(written) == {9101, 9102, 9103, 9104, 9105} + for uid in written: + assert _gauge_value( + T.VALIDATOR_MINER_LAST_SCORED_ROUND_ID, + "validator_miner_last_scored_round_id", uid, + ) == float(rid) + assert _gauge_value( + T.VALIDATOR_MINER_SCORE_LATEST, "validator_miner_score_latest", uid, + ) is not None + # Deltas only for evaluated uids, raw values preserved. + assert _gauge_value( + T.VALIDATOR_MINER_ROUND_DELTA, "validator_miner_round_delta", 9101 + ) == 2.5 + assert _gauge_value( + T.VALIDATOR_MINER_ROUND_DELTA, "validator_miner_round_delta", 9103 + ) == 0.0 + assert _gauge_value( + T.VALIDATOR_MINER_ROUND_DELTA, "validator_miner_round_delta", 9105 + ) is None + # Exactly one commit labelset per checkpointed uid. + assert len(_labelsets_for_uid(9101)) == 1 + assert _labelsets_for_uid(9101)[0].value == float(rid) + assert len(_labelsets_for_uid(9102)) == 1 + + +def test_finalize_recovery_path_reemits_commits(tmp_path): + rid = 524_524 + # Write a v2 journal, hydrate a recovery round from it, finalize. + live = _make_round(rid, journal_path=tmp_path / f"round_{rid}.json") + RJ.write_atomic( + live.journal_path, + RJ.RoundJournal( + round_id=rid, + uid_to_hotkey=dict(live.uid_to_hotkey), + scores=dict(live.scores), + scored_uids=tuple(sorted(live.scored_uids)), + validation_failed_uids=tuple(sorted(live.validation_failed_uids)), + freeze_zero_uids=tuple(sorted(live.freeze_zero_uids)), + freeze_zero_hotkeys=dict(live.freeze_zero_hotkeys), + uid_to_commit=RJ.commit_map_from_checkpoints(live.uid_to_chain_checkpoint), + ), + ) + journal = RJ.load(live.journal_path) + assert journal is not None + recovered = RJ._RecoveryRound.from_journal(journal, live.journal_path) + finalize_round_scores( + round_obj=recovered, score_aggregator=_FakeAggregator(), score_path=None, + ) + rows = _labelsets_for_uid(9101) + assert len(rows) == 1 + assert rows[0].value == float(rid) + assert _gauge_value( + T.VALIDATOR_MINER_LAST_SCORED_ROUND_ID, + "validator_miner_last_scored_round_id", 9105, + ) == float(rid) + + +# --------------------------------------------------------------------------- +# Per-round series eviction +# --------------------------------------------------------------------------- + +def test_evict_round_series_before(): + old_rid, new_rid = 910_000, 910_524 + for rid in (old_rid, new_rid): + T.note_round_series(rid) + T.VALIDATOR_ROUND_MINERS_SCORED.labels(round_id=str(rid)).set(5) + removed = T.evict_round_series_before(new_rid) + assert removed >= 1 + rids = { + s.labels["round_id"] + for s in _samples_for( + T.VALIDATOR_ROUND_MINERS_SCORED, "validator_round_miners_scored" + ) + } + assert str(old_rid) not in rids + assert str(new_rid) in rids + # Idempotent / KeyError-safe on repeat. + assert T.evict_round_series_before(new_rid) == 0 diff --git a/connito/validator/background_eval_worker.py b/connito/validator/background_eval_worker.py index 8963045..da1c34a 100644 --- a/connito/validator/background_eval_worker.py +++ b/connito/validator/background_eval_worker.py @@ -31,6 +31,7 @@ VALIDATOR_ROUND_MINERS_FAILED, VALIDATOR_ROUND_MINERS_PENDING, VALIDATOR_ROUND_MINERS_SCORED, + note_round_series, ) from connito.validator.evaluator import ( EVAL_MAX_BATCHES, @@ -503,6 +504,7 @@ def _run_eval(): round_id=round_obj.round_id, ) try: + note_round_series(round_obj.round_id) VALIDATOR_BG_EVAL_LOCK_LEAK_TOTAL.labels( round_id=str(round_obj.round_id) ).inc() @@ -626,6 +628,7 @@ def _prune_non_top(self, round_obj) -> None: def _record_metrics(round_obj, *, scored_inc: bool) -> None: try: stats = round_obj.stats() + note_round_series(round_obj.round_id) VALIDATOR_ROUND_MINERS_SCORED.labels(round_id=str(round_obj.round_id)).set(stats["scored"]) VALIDATOR_ROUND_MINERS_FAILED.labels(round_id=str(round_obj.round_id)).set(stats["failed"]) VALIDATOR_ROUND_MINERS_PENDING.labels(round_id=str(round_obj.round_id)).set(stats["pending"]) diff --git a/connito/validator/evaluator.py b/connito/validator/evaluator.py index 4209003..df2f5f9 100644 --- a/connito/validator/evaluator.py +++ b/connito/validator/evaluator.py @@ -344,6 +344,9 @@ def finalize_round_scores( validation_failed_uids=tuple(sorted(validation_failed)), freeze_zero_uids=tuple(sorted(freeze_zero)), freeze_zero_hotkeys=dict(freeze_hotkeys), + uid_to_commit=_rj.commit_map_from_checkpoints( + getattr(round_obj, "uid_to_chain_checkpoint", None) or {} + ), finalized=True, ), ) @@ -353,6 +356,54 @@ def finalize_round_scores( round_id=round_obj.round_id, error=str(e), ) + # --- Cycle-consistent per-miner telemetry (dashboard contract). ------- + # Emitted HERE — not from run.py's weight loop — for two reasons: + # (1) the weight loop only iterates weight recipients, so the dashboard + # previously saw score snapshots for ~1 uid; every verdict uid gets + # one now; (2) the journal-recovery replay calls this function too, + # so a restart re-publishes the series without waiting for a fresh + # round. Best-effort throughout — telemetry must never block + # finalize or scoring. + try: + from connito.shared.telemetry import ( + set_miner_evaluated_commit, + set_miner_last_scored_round, + set_miner_round_delta, + set_miner_score_snapshot, + ) + + _rid = int(round_obj.round_id) + try: + _latest_scores = score_aggregator.uid_score_pairs(how="latest") + _avg_scores = score_aggregator.uid_score_pairs(how="avg") + except Exception: + _latest_scores, _avg_scores = {}, {} + for uid in written: + set_miner_last_scored_round(int(uid), _rid) + try: + _samples = score_aggregator.record_count(int(uid)) + except Exception: + _samples = None + set_miner_score_snapshot( + int(uid), + latest=_latest_scores.get(int(uid)), + avg=_avg_scores.get(int(uid)), + samples=_samples, + ) + for uid in scored: + set_miner_round_delta(int(uid), float(round_scores.get(uid, 0.0))) + _ckpt_map = getattr(round_obj, "uid_to_chain_checkpoint", None) or {} + for uid, _ckpt in _ckpt_map.items(): + _repo = getattr(_ckpt, "hf_repo_id", None) + _rev = getattr(_ckpt, "hf_revision", None) + if _repo and _rev: + set_miner_evaluated_commit(int(uid), str(_repo), str(_rev), _rid) + except Exception as e: + logger.warning( + "finalize_round_scores: telemetry emission failed", + round_id=round_obj.round_id, error=str(e), + ) + logger.info( "finalize_round_scores: round scored by rank", round_id=round_obj.round_id, diff --git a/connito/validator/round.py b/connito/validator/round.py index 9670264..a3c310d 100644 --- a/connito/validator/round.py +++ b/connito/validator/round.py @@ -575,6 +575,8 @@ def _journal_snapshot_locked(self) -> dict | None: """ if self.journal_path is None: return None + from connito.validator.round_journal import commit_map_from_checkpoints + return { "round_id": self.round_id, "uid_to_hotkey": dict(self.uid_to_hotkey), @@ -584,6 +586,7 @@ def _journal_snapshot_locked(self) -> dict | None: "validation_failed_uids": tuple(sorted(self.validation_failed_uids)), "freeze_zero_uids": tuple(sorted(self.freeze_zero_uids)), "freeze_zero_hotkeys": dict(self.freeze_zero_hotkeys), + "uid_to_commit": commit_map_from_checkpoints(self.uid_to_chain_checkpoint), "finalized": False, } diff --git a/connito/validator/round_journal.py b/connito/validator/round_journal.py index 2757922..41e2250 100644 --- a/connito/validator/round_journal.py +++ b/connito/validator/round_journal.py @@ -25,7 +25,11 @@ from dataclasses import asdict, dataclass, field from pathlib import Path -SCHEMA_VERSION = 1 +# v2 adds `uid_to_commit` (uid -> (hf_repo_id, hf_revision)) so the +# journal-recovery finalize can re-publish evaluated-commit telemetry. +# `from_json` accepts v1 files (missing map -> empty) so leftover journals +# written by an older build still recover. +SCHEMA_VERSION = 2 JOURNAL_DIR_NAME = "round_journal" JOURNAL_FILENAME_PREFIX = "round_" JOURNAL_FILENAME_SUFFIX = ".json" @@ -52,6 +56,9 @@ class RoundJournal: validation_failed_uids: tuple[int, ...] = () freeze_zero_uids: tuple[int, ...] = () freeze_zero_hotkeys: dict[int, str] = field(default_factory=dict) + # v2: uid -> (hf_repo_id, hf_revision) evaluated for the round. Only + # uids whose chain checkpoint carried BOTH values are recorded. + uid_to_commit: dict[int, tuple[str, str]] = field(default_factory=dict) finalized: bool = False schema_version: int = SCHEMA_VERSION @@ -65,16 +72,23 @@ def to_json(self) -> str: payload["failed_uids"] = list(self.failed_uids) payload["validation_failed_uids"] = list(self.validation_failed_uids) payload["freeze_zero_uids"] = list(self.freeze_zero_uids) + payload["uid_to_commit"] = { + str(k): [str(v[0]), str(v[1])] for k, v in self.uid_to_commit.items() + } return json.dumps(payload) @classmethod def from_json(cls, data: str) -> "RoundJournal": raw = json.loads(data) version = int(raw.get("schema_version", 1)) - if version != SCHEMA_VERSION: + # Accept every version up to the current one: v1 files simply lack + # `uid_to_commit` (defaults to empty). Reject only FUTURE versions — + # fields this build doesn't understand could change recovery + # semantics silently. + if version > SCHEMA_VERSION: raise ValueError( f"Unsupported RoundJournal schema_version={version}; " - f"expected {SCHEMA_VERSION}" + f"this build supports <= {SCHEMA_VERSION}" ) return cls( round_id=int(raw["round_id"]), @@ -87,11 +101,31 @@ def from_json(cls, data: str) -> "RoundJournal": freeze_zero_hotkeys={ int(k): str(v) for k, v in raw.get("freeze_zero_hotkeys", {}).items() }, + uid_to_commit={ + int(k): (str(v[0]), str(v[1])) + for k, v in raw.get("uid_to_commit", {}).items() + if isinstance(v, (list, tuple)) and len(v) == 2 + }, finalized=bool(raw.get("finalized", False)), schema_version=version, ) +def commit_map_from_checkpoints( + uid_to_chain_checkpoint: dict[int, object], +) -> dict[int, tuple[str, str]]: + """Extract the journal's `uid_to_commit` map from a round's + `uid_to_chain_checkpoint`. Skips uids missing either field. + """ + out: dict[int, tuple[str, str]] = {} + for uid, ckpt in (uid_to_chain_checkpoint or {}).items(): + repo = getattr(ckpt, "hf_repo_id", None) + rev = getattr(ckpt, "hf_revision", None) + if repo and rev: + out[int(uid)] = (str(repo), str(rev)) + return out + + def journal_dir(checkpoint_path: str | os.PathLike) -> Path: """Directory holding all per-round journal files.""" return Path(checkpoint_path) / JOURNAL_DIR_NAME @@ -179,10 +213,17 @@ class _RecoveryRound: freeze_zero_hotkeys: dict[int, str] uid_to_hotkey: dict[int, str] journal_path: Path + # Same shape finalize reads off a live Round: objects exposing + # `.hf_repo_id` / `.hf_revision`. Hydrated from the journal's v2 + # `uid_to_commit` map (empty for v1 journals) so a recovered finalize + # re-publishes evaluated-commit telemetry too. + uid_to_chain_checkpoint: dict[int, object] = field(default_factory=dict) _lock: threading.Lock = field(default_factory=threading.Lock, repr=False) @classmethod def from_journal(cls, journal: "RoundJournal", journal_path: str | os.PathLike) -> "_RecoveryRound": + from types import SimpleNamespace + return cls( round_id=int(journal.round_id), scores=dict(journal.scores), @@ -193,6 +234,10 @@ def from_journal(cls, journal: "RoundJournal", journal_path: str | os.PathLike) freeze_zero_hotkeys=dict(journal.freeze_zero_hotkeys), uid_to_hotkey=dict(journal.uid_to_hotkey), journal_path=Path(journal_path), + uid_to_chain_checkpoint={ + int(uid): SimpleNamespace(hf_repo_id=repo, hf_revision=rev) + for uid, (repo, rev) in journal.uid_to_commit.items() + }, ) def processed_uids_snapshot(self) -> tuple[set[int], set[int]]: diff --git a/connito/validator/run.py b/connito/validator/run.py index d6d6ff8..7edc947 100644 --- a/connito/validator/run.py +++ b/connito/validator/run.py @@ -187,8 +187,9 @@ def validate_hf_distribution_config(config: ValidatorConfig) -> tuple[str | None set_miner_assignment_role, set_miner_cohort_group, set_miner_last_observed_commit_block, - set_miner_score_snapshot, set_validator_identity, + note_round_series, + evict_round_series_before, track_metagraph_sync_latency, ) from datetime import datetime @@ -1092,6 +1093,22 @@ def run(rank: int, world_size: int, config: ValidatorConfig, pkg_version: str = "round_journal.prune_before_round failed", error=str(e), ) + # Evict per-round Prometheus labelsets on the same cutoff so + # metric retention matches on-disk retention (without this, + # every round leaves permanent {round_id} series behind). + try: + _series_evicted = evict_round_series_before(_min_round_id) + if _series_evicted: + logger.info( + "telemetry: evicted stale per-round series", + rounds=_series_evicted, + min_round_id=_min_round_id, + ) + except Exception as e: + logger.warning( + "telemetry.evict_round_series_before failed", + error=str(e), + ) logger.info( "(4) Handing weight submission to background submitter", round_id=pending_round.round_id, @@ -1120,26 +1137,16 @@ def run(rank: int, world_size: int, config: ValidatorConfig, pkg_version: str = ) # 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 - # semantics of `score_aggregator.uid_score_pairs(how="avg")` - # — entries are written only for UIDs we actually weight, - # so a miner the validator has never scored has *no* sample - # rather than a zero (preserves prior EMA semantics). + # to learn what each validator votes on chain. Entries are + # written only for UIDs we actually weight, so a miner the + # validator has never scored has *no* sample rather than a + # zero (preserves prior EMA semantics). # - # Same scrape also publishes the aggregator snapshot - # (latest / avg / sample count) for every UID we weight, - # so the gateway can render miner-facing telemetry without - # re-deriving from per-round samples. Best-effort throughout - # — a Prometheus failure must not block weight submission. - try: - _latest_scores = score_aggregator.uid_score_pairs(how="latest") - _avg_scores = score_aggregator.uid_score_pairs(how="avg") - except Exception as _e: - logger.warning( - "Failed to read aggregator snapshot for telemetry", - error=str(_e), - ) - _latest_scores, _avg_scores = {}, {} + # The per-miner score snapshots (latest / avg / samples / + # emitted_at) are NOT published here anymore — they moved to + # `finalize_round_scores`, which covers every verdict uid + # (not just weight recipients) and re-publishes via the + # journal-recovery replay after a restart. for _uid, _weight in uid_weights.items(): try: VALIDATOR_MINER_WEIGHT_SUBMITTED.labels( @@ -1147,16 +1154,6 @@ def run(rank: int, world_size: int, config: ValidatorConfig, pkg_version: str = ).set(float(_weight)) except Exception: pass - try: - _samples = score_aggregator.record_count(int(_uid)) - except Exception: - _samples = None - set_miner_score_snapshot( - int(_uid), - latest=_latest_scores.get(int(_uid)), - avg=_avg_scores.get(int(_uid)), - samples=_samples, - ) # Fire-and-forget. ChainSubmitter sets # pending_round.weights_submitted once the chain accepts the call. chain_submitter.async_submit_weight(pending_round, uid_weights) @@ -1418,6 +1415,7 @@ def run(rank: int, world_size: int, config: ValidatorConfig, pkg_version: str = round_ref.swap(new_current=new_round) download_window_closed.clear() try: + note_round_series(new_round.round_id) VALIDATOR_ROUND_LIFECYCLE_STEP.labels(round_id=str(new_round.round_id)).set(0) except Exception: pass @@ -1497,6 +1495,7 @@ async def _bounded_foreground_eval(): if eval_worker is not None and not eval_worker.has_eval_base_model(): eval_worker.set_eval_base_model(copy.deepcopy(global_model)) try: + note_round_series(new_round.round_id) VALIDATOR_ROUND_LIFECYCLE_STEP.labels(round_id=str(new_round.round_id)).set(2) except Exception: pass @@ -1728,6 +1727,7 @@ async def _bounded_foreground_eval(): # the post-Merge mutation of global_model does not affect it. eval_window_active.set() try: + note_round_series(new_round.round_id) VALIDATOR_ROUND_LIFECYCLE_STEP.labels(round_id=str(new_round.round_id)).set(3) except Exception: pass