Hugging Face's logo

Buckets:
affine-god
/
1111

Files
xet
affine-god/1111 / preflight_submit.py
affine-god's picture
affine-god
download
raw
33.8 kB
"""Preflight every gate the SN120 validator runs — BEFORE burning the ticket.
READ-ONLY. This module never reveals, commits, or submits anything. There is no
code path here that writes to the chain; the reveal stays a manual step.
Why this exists
---------------
`State.enqueue` does `seen_hotkeys.add(hotkey)` — *"Slot burned at enqueue: one
reveal = one shot."* Every failure AFTER that point still consumes the slot, and
the slot never comes back. Measured on the last 100 validator records:
eval_infra_exhausted 9 validator-side, unavoidable
unservable: failed to load in vLLM 8 preventable
unpromptable: no parsable action in 3 turns 4 preventable
revision_not_found 2 preventable
repo_name_rejected 1 preventable
------------------------------------------------
24/100 tickets wasted, 15 of them avoidable
So ~9% is pure validator-side loss nothing can design around — budget for it —
and the remaining 15% is exactly what this script removes.
Each check mirrors the validator's own implementation rather than paraphrasing
it, so the two cannot drift:
A registration metagraph lookup (also supplies the ON-CHAIN coldkey,
which is what identity binding uses — not the local wallet)
B slot unburned live dashboard intake + history for this hotkey
C repo name model_store.validate_repo_name (pattern + identity pair)
D revision model_store.fetch_repo_info (raises -> revision_not_found)
E hygiene model_store.validate_repo_hygiene
F copy of king model_store.check_model_copy (blob digest MULTISETS)
G promptability evalsrv.dueling.probe_injectable (needs a served model)
H anon access model_store.fetch_repo_info_or_status — UNAUTHENTICATED
I contract fork [subnet].weight_version_key still what we screened under
J reveal window current block > [subnet].min_submission_block
K reveal payload chain.parse_reveal(build_reveal(...)) round-trips
L revision novelty never submitted by anyone (content-revision dedup)
M index complete every weight_map shard in model.safetensors.index.json exists
N crown bar measured margin > max(k_sigma·SE, min_margin), live values
G is the one gate that cannot be checked from metadata, and it is the one that
matches what fine-tuning changes: a checkpoint that emits thoughts but no
parsable closed ```bash action looks identical in every screen we run, because
`screen_l2.py generate` keeps `z` and discards `y`.
H is the one that cost us reign 7. Gates D/E/F run with `--hf-token`, so a repo
that is gated to everyone else still passes them for us. The validator fetches
with *its* credentials and gets a 401 → `revision_not_found`, and — worse — it
re-probes the KING's weights before every subsequent duel, so a repo gated after
a crown is dethroned by rollback (`fetch_repo_info_or_status` returns "gone").
That is not a submit-time gate, it is a forever condition: see `--watch-king`.
Usage
-----
# metadata gates only (free, no GPU)
python harness/preflight_submit.py --repo dent1s2/affine-5FjqRq3dGA-v1 \
--revision <40hex> --hotkey hotkey-1
# include the promptability probe against an already-served candidate
python harness/preflight_submit.py ... --port 8002
Exit code 0 only when every applicable gate passes.
"""
from __future__ import annotations
import argparse
import asyncio
import json
import re
import sys
import urllib.error
import urllib.request
# affine.io is the contract SSOT and carries [duel] (k_sigma, min_margin) and
# the live weight_version_key; the hippius mirror is the fallback. The repo's
# affine.toml is NOT authoritative — it still said wvk=3 after the live fork
# to 4 (2026-08-12, the min_margin=0.002 addition).
CONTRACT_API = "https://affine.io/api/v1/contract"
CONTRACT_TOML = "https://s3.hippius.com/affine-sn120/data/contract.json"
SNAPSHOT_API = "https://affine.io/api/v1/snapshot"
HISTORY_API = "https://affine.io/api/v1/history?limit=1000"
DASHBOARD = "https://s3.hippius.com/affine-sn120/data/dashboard.json"
HISTORY = "https://s3.hippius.com/affine-sn120/data/history.json"
HF_RESOLVE = "https://huggingface.co/{repo}/resolve/{rev}/{path}"
HF_MODEL_API = "https://huggingface.co/api/models/{repo}/revision/{rev}"
NETUID = 120
# Mirrors affine.toml [submission]. Used only when the live contract cannot be
# fetched; a stale cap here would be worse than no answer, so a fallback is
# always reported as such.
FALLBACK = {
"repo_pattern": r"^[^/]+/[Aa]ffine-.+$",
"coldkey_prefix_len": 5,
"coldkey_suffix_len": 5,
"max_model_size_gb": 90.0,
"max_total_repo_gb": 100.0,
"allow_python_files": False,
"allow_auto_map": False,
"max_repo_files": 5000,
"max_config_bytes": 1 << 20,
"min_submission_block": 8767079,
}
FALLBACK_DUEL = {"k_sigma": 2.0, "min_margin": 0.002, "n_turns": 2080}
REVEAL_PREFIX = "affine1"
EXPECT_WVK = 4 # live as of 2026-08-13; a bump invalidates every screen
SHARD_RE = re.compile(r"^model-\d{5}-of-\d{5}\.safetensors$")
SS58_RE = re.compile(r"^[1-9A-HJ-NP-Za-km-z]{47,48}$")
def _get(url: str, timeout: float = 60.0):
# affine.io sits behind Cloudflare, which 403s a bare urllib User-Agent.
req = urllib.request.Request(url, headers={"User-Agent": "affine-preflight",
"Accept": "application/json"})
with urllib.request.urlopen(req, timeout=timeout) as r:
return json.load(r)
class Report:
"""Collects gate outcomes. A gate that could not run is NOT a pass."""
def __init__(self) -> None:
self.rows: list[tuple[str, str, str]] = []
def add(self, gate: str, status: str, detail: str = "") -> None:
self.rows.append((gate, status, detail))
def ok(self, gate: str, detail: str = "") -> None:
self.add(gate, "PASS", detail)
def fail(self, gate: str, detail: str) -> None:
self.add(gate, "FAIL", detail)
def skip(self, gate: str, detail: str) -> None:
self.add(gate, "SKIP", detail)
def warn(self, gate: str, detail: str) -> None:
"""Not a validator gate, but a known cause of a burnt slot."""
self.add(gate, "WARN", detail)
@property
def failed(self) -> bool:
return any(s == "FAIL" for _, s, _ in self.rows)
@property
def skipped(self) -> bool:
return any(s == "SKIP" for _, s, _ in self.rows)
def render(self) -> None:
print("\n" + "=" * 74)
for gate, status, detail in self.rows:
mark = {"PASS": "ok ", "FAIL": "FAIL", "SKIP": "skip", "WARN": "warn"}[status]
print(f" [{mark}] {gate:<22} {detail[:44]}")
if status in ("FAIL", "WARN") and len(detail) > 44:
print(f" {detail}")
print("=" * 74)
def load_contract(report: Report) -> dict:
"""affine.io first, hippius mirror second, pinned fallback last.
Carries [duel] and weight_version_key alongside [submission]: the crown
arithmetic and the fork check both read from here, and both were unreachable
while this only pulled [submission] off the hippius mirror.
"""
c, src = None, ""
for url, name in ((CONTRACT_API, "affine.io"), (CONTRACT_TOML, "hippius")):
try:
c, src = _get(url), name
break
except Exception:
continue
if c is None:
report.skip("contract", "live fetch failed everywhere; using pinned fallback")
cfg = dict(FALLBACK)
cfg["_duel"] = dict(FALLBACK_DUEL)
cfg["_wvk"] = None
cfg["_reveal_prefix"] = REVEAL_PREFIX
return cfg
sub, subnet = c.get("submission") or {}, c.get("subnet") or {}
cfg = {k: sub.get(k, v) for k, v in FALLBACK.items()}
cfg["min_submission_block"] = subnet.get(
"min_submission_block", FALLBACK["min_submission_block"])
cfg["_duel"] = {**FALLBACK_DUEL, **(c.get("duel") or {})}
cfg["_wvk"] = subnet.get("weight_version_key")
cfg["_reveal_prefix"] = sub.get("reveal_prefix", REVEAL_PREFIX)
d = cfg["_duel"]
report.ok("contract", f"{src}: wvk={cfg['_wvk']} k={d['k_sigma']} "
f"delta={d['min_margin']}")
return cfg
def check_contract_fork(cfg: dict, expect_wvk: int, report: Report) -> None:
"""A weight_version_key bump is a scoring-rule fork: every margin we
screened under the old rule is unproven under the new one."""
live = cfg.get("_wvk")
if live is None:
report.skip("contract fork", "live weight_version_key unreadable")
elif live != expect_wvk:
report.fail("contract fork", f"weight_version_key is {live}, screens were run "
f"under {expect_wvk} — re-read score.py and "
f"re-screen before revealing")
else:
report.ok("contract fork", f"weight_version_key={live} unchanged")
def check_registration(hotkey_ss58: str, report: Report) -> tuple[int | None, str | None]:
"""uid + ON-CHAIN coldkey. Identity binding uses the chain's coldkey, not
whatever the local wallet file says."""
try:
import bittensor as bt
except ImportError:
report.skip("registration", "bittensor not importable in this interpreter")
return None, None
try:
mg = bt.Subtensor(network="finney").metagraph(netuid=NETUID)
except Exception as e:
report.skip("registration", f"chain unreachable: {type(e).__name__}")
return None, None
if hotkey_ss58 not in mg.hotkeys:
report.fail("registration", f"{hotkey_ss58[:12]}… NOT registered on netuid {NETUID}")
return None, None
i = mg.hotkeys.index(hotkey_ss58)
uid, ck = int(mg.uids[i]), mg.coldkeys[i]
report.ok("registration", f"uid {uid}, coldkey {ck[:10]}…")
return uid, ck
def check_slot(hotkey_ss58: str, report: Report) -> None:
"""A burned slot is terminal — nothing downstream matters if this fails."""
try:
intake = (_get(DASHBOARD).get("intake") or [])
hist = _get(HISTORY)
except Exception as e:
report.skip("slot unburned", f"cannot read validator state: {type(e).__name__}")
return
hits = [r for r in intake if r.get("hotkey") == hotkey_ss58]
hh = [r for r in hist if r.get("hotkey") == hotkey_ss58]
if hits or hh:
d = "; ".join(f"{r.get('decision') or r.get('event')}" for r in (hits + hh)[:3])
report.fail("slot unburned", f"hotkey already used its one eval slot: {d}")
else:
report.ok("slot unburned", "no intake/history records (window-limited)")
def _anon_status(url: str, timeout: float = 30.0) -> tuple[int, str]:
"""HTTP status for an explicitly unauthenticated GET. No HF token is read
from the environment, the git credential store, or ~/.cache/huggingface —
that is the whole point: it answers "can a stranger fetch this?"."""
req = urllib.request.Request(url, method="GET",
headers={"User-Agent": "affine-preflight"})
try:
with urllib.request.urlopen(req, timeout=timeout) as r:
return int(r.status), ""
except urllib.error.HTTPError as e:
return int(e.code), e.reason or ""
except Exception as e:
return -1, f"{type(e).__name__}: {e}"
def check_anon_access(repo: str, revision: str, report: Report,
gate: str = "anon access") -> bool:
"""Mirrors model_store.fetch_repo_info_or_status, which the validator calls
with its own credentials — not ours.
A gated/private repo returns 401/403 and the validator classifies it "gone":
at intake that is `revision_not_found` (slot burnt), and for a sitting king
it is a rollback to the previous king (reign 7, 2026-08-13). The probe HEADs
an uncached blob at the pinned revision because a repo gated *after* first
contact keeps a readable public tree.
"""
ok = True
for label, url in (("repo api", HF_MODEL_API.format(repo=repo, rev=revision)),
("config blob", HF_RESOLVE.format(repo=repo, rev=revision,
path="config.json"))):
code, why = _anon_status(url)
if code == 200:
continue
ok = False
if code in (401, 403):
report.fail(gate, f"{label} -> HTTP {code}: repo is GATED/PRIVATE to "
f"anonymous callers. The validator sees 'gone'.")
elif code == 404:
report.fail(gate, f"{label} -> 404: repo or revision does not exist "
f"({repo}@{revision[:12]})")
else:
report.skip(gate, f"{label} -> {code} {why} (inconclusive, not a pass)")
if ok:
report.ok(gate, "public to anonymous callers at the pinned revision")
return ok
def check_reveal_window(cfg: dict, report: Report) -> None:
"""Reveals at or below min_submission_block are ignored by intake."""
floor = cfg["min_submission_block"]
try:
import bittensor as bt
block = int(bt.Subtensor(network="finney").block)
except Exception as e:
report.skip("reveal window", f"current block unreadable: {type(e).__name__} "
f"(floor is {floor})")
return
if block <= floor:
report.fail("reveal window", f"current block {block} <= min_submission_block "
f"{floor}: the reveal would be ignored")
else:
report.ok("reveal window", f"block {block} > floor {floor}")
def check_reveal_payload(cfg: dict, repo: str, revision: str, hotkey: str,
report: Report) -> None:
"""Mirrors chain.build_reveal + chain.parse_reveal. The payload is
timelock-encrypted, so a malformed one is only discovered at reveal —
after the slot is gone."""
prefix = cfg["_reveal_prefix"]
payload = f"{prefix}|{repo}|{revision}|{hotkey}"
parts = payload.split("|")
if len(parts) != 4 or parts[0] != prefix:
return report.fail("reveal payload", f"expected {prefix}|repo|revision|hotkey")
if "|" in repo or not re.match(r"^[^/]+/[^/]+$", repo):
return report.fail("reveal payload", f"invalid repo in reveal: {repo!r}")
if not re.fullmatch(r"[0-9a-f]{40}", revision):
return report.fail("reveal payload", f"invalid revision in reveal: {revision!r}")
if not SS58_RE.match(hotkey):
return report.fail("reveal payload", f"invalid author hotkey in reveal: {hotkey!r}")
report.ok("reveal payload", f"{prefix}|…|{revision[:8]}…|{hotkey[:8]}… parses")
def check_revision_novel(repo: str, revision: str, report: Report) -> None:
""""Never resubmit a content revision that has ever been submitted by
anyone" — a duplicate is rejected and the slot is still gone."""
rows = []
for url in (HISTORY_API, HISTORY):
try:
d = _get(url)
rows = d.get("items", d) if isinstance(d, dict) else d
break
except Exception:
continue
try:
snap = _get(SNAPSHOT_API)
rows = list(rows) + list(snap.get("intake") or []) + list(snap.get("queue") or [])
except Exception:
pass
if not rows:
return report.skip("revision novelty", "validator history unreadable")
hits = [r for r in rows if r.get("revision") == revision]
if hits:
who = hits[0].get("repo", "?")
return report.fail("revision novelty", f"revision {revision[:12]}… was already "
f"submitted as {who} — duplicate reject")
same_repo = [r for r in rows if r.get("repo") == repo]
if same_repo:
report.warn("revision novelty", f"revision is new, but {repo} appears "
f"{len(same_repo)}x in history under other revisions")
else:
report.ok("revision novelty", f"{revision[:12]}… unseen in {len(rows)} records")
def check_index_complete(info: dict, report: Report) -> None:
"""Catches the `unservable: failed to load in vLLM` class (8/100 burnt
tickets) at metadata cost: every shard the index references must exist.
A CausalLM re-save of the multimodal Qwen MoE king silently drops
`model.visual.*`, which shows up here as an index/file mismatch or, when the
index was rewritten too, as a missing preprocessor_config.json.
"""
files = set(info["files"])
if "model.safetensors.index.json" not in files:
return report.ok("index complete", "single-file model.safetensors layout")
idx = info.get("index")
if not idx:
return report.skip("index complete", "index.json not readable")
wanted = set((idx.get("weight_map") or {}).values())
missing = sorted(wanted - files)
if missing:
return report.fail("index complete", f"index references {len(missing)} shard(s) "
f"not in the repo: {missing[:3]}")
orphan = sorted({f for f in files if SHARD_RE.match(f)} - wanted)
n_tensors = len(idx.get("weight_map") or {})
if orphan:
report.warn("index complete", f"{len(orphan)} shard(s) present but unreferenced "
f"by the index: {orphan[:3]}")
cfgj = info["config"]
multimodal = ("vision_config" in cfgj
or any("ConditionalGeneration" in a
for a in (cfgj.get("architectures") or [])))
if multimodal and "preprocessor_config.json" not in files:
report.warn("index complete", f"{n_tensors} tensors, but multimodal config with "
f"NO preprocessor_config.json — vLLM may refuse to load")
elif not orphan:
report.ok("index complete", f"{n_tensors} tensors over {len(wanted)} shards")
def check_crown_bar(margin: float | None, se: float | None, cfg: dict,
report: Report) -> None:
"""score.duel: wins iff mean > max(k_sigma·SE, min_margin). Both terms are
live values — k_sigma moved 3.0→2.0 and min_margin appeared at wvk=4."""
d = cfg["_duel"]
k, delta = float(d["k_sigma"]), float(d["min_margin"])
if margin is None or se is None:
return report.skip("crown bar", f"pass --margin/--se from the screen "
f"(bar = max({k}·SE, {delta}))")
bar = max(k * se, delta)
which = "k·SE" if k * se >= delta else "min_margin"
if margin <= bar:
report.fail("crown bar", f"margin {margin:+.5f} <= bar {bar:.5f} ({which}) — "
f"this loses the duel and burns the slot")
else:
report.ok("crown bar", f"margin {margin:+.5f} > bar {bar:.5f} ({which}), "
f"headroom {margin / bar:.2f}x")
def identity_pairs(hotkey: str, coldkey: str | None, cfg: dict) -> list[tuple[str, str]]:
keys = [hotkey] + ([coldkey] if coldkey else [])
p, s = cfg["coldkey_prefix_len"], cfg["coldkey_suffix_len"]
return [(k[:p].lower(), k[-s:].lower()) for k in keys]
def check_repo_name(repo: str, pairs: list[tuple[str, str]], cfg: dict,
report: Report) -> None:
if not re.match(cfg["repo_pattern"], repo):
report.fail("repo name", f"{repo!r} does not match {cfg['repo_pattern']!r}")
return
if not pairs:
report.skip("repo name", "identity pairs unknown (no chain access)")
return
r = repo.lower()
if not any(p in r and s in r for p, s in pairs):
want = " or ".join(f"{p}…{s}" for p, s in pairs)
report.fail("repo name", f"must embed {want} (matched against the FULL "
f"repo id, org included)")
return
report.ok("repo name", "pattern + identity binding")
def fetch_info(repo: str, revision: str, token: str, report: Report):
"""Mirrors model_store.fetch_repo_info. Failure here is `revision_not_found`,
which burns the slot for real."""
from huggingface_hub import HfApi
api = HfApi(token=token or None)
tree = list(api.list_repo_tree(repo, revision=revision, recursive=True, expand=True))
files, blobs = [], {}
total_st = total_all = 0
for t in tree:
path = getattr(t, "path", "")
size = int(getattr(t, "size", 0) or 0)
files.append(path)
total_all += size
if not path.endswith(".safetensors"):
continue
total_st += size
lfs = getattr(t, "lfs", None)
sha = getattr(lfs, "sha256", None) if lfs else None
if sha:
blobs[path] = str(sha).lower().removeprefix("sha256:")
cfg_path = api.hf_hub_download(repo, "config.json", revision=revision)
with open(cfg_path) as f:
config = json.load(f)
index = None
if "model.safetensors.index.json" in files:
try:
ip = api.hf_hub_download(repo, "model.safetensors.index.json",
revision=revision)
with open(ip) as f:
index = json.load(f)
except Exception:
pass
committed_at = None
try:
for c in api.list_repo_commits(repo, revision=revision):
if getattr(c, "commit_id", "") == revision:
committed_at = getattr(c, "created_at", None)
break
except Exception:
pass
report.ok("revision", f"{len(files)} files, {total_st/1e9:.1f} GB safetensors")
return {"files": files, "config": config, "blobs": blobs, "index": index,
"total_st": total_st, "total_all": total_all,
"committed_at": committed_at}
def check_hygiene(info: dict, cfg: dict, report: Report) -> None:
"""Mirrors model_store.validate_repo_hygiene, in the same order."""
files, config = info["files"], info["config"]
if len(files) > cfg["max_repo_files"]:
return report.fail("hygiene", f"{len(files)} files > {cfg['max_repo_files']}")
py = sorted(f for f in files if f.endswith(".py"))
if py and not cfg["allow_python_files"]:
return report.fail("hygiene", f"repo ships *.py files: {py[:3]}")
if "auto_map" in config and not cfg["allow_auto_map"]:
return report.fail("hygiene", "auto_map in config.json (custom code not allowed)")
cb = len(json.dumps(config))
if cb > cfg["max_config_bytes"]:
return report.fail("hygiene", f"config.json {cb} B > {cfg['max_config_bytes']}")
st = [f for f in files if f.endswith(".safetensors")]
if not st:
return report.fail("hygiene", "no .safetensors files in repo")
has_single = "model.safetensors" in files
has_index = "model.safetensors.index.json" in files
has_shards = any(SHARD_RE.match(f) for f in st)
if not (has_single or (has_index and has_shards)):
if has_shards and not has_index:
return report.fail("hygiene", "missing model.safetensors.index.json for shards")
return report.fail("hygiene", f"non-canonical safetensors layout: {st[:3]}")
gb = info["total_st"] / 1e9
if gb > cfg["max_model_size_gb"]:
return report.fail("hygiene", f"oversized: {gb:.1f} GB > {cfg['max_model_size_gb']:.0f} GB "
f"(fp32 weights? duplicated shards?)")
tgb = info["total_all"] / 1e9
if tgb > cfg["max_total_repo_gb"]:
return report.fail("hygiene", f"repo total {tgb:.1f} GB > {cfg['max_total_repo_gb']:.0f} GB")
report.ok("hygiene", f"{gb:.1f} GB weights, {len(files)} files, no .py/auto_map")
def check_copy(repo: str, revision: str, info: dict, token: str, report: Report) -> None:
"""Mirrors model_store.check_model_copy: digest MULTISETS, copy iff every
king blob is present in the challenger."""
try:
d = _get(DASHBOARD)
king = d.get("king") or {}
krepo, krev = king.get("repo"), king.get("revision")
except Exception as e:
return report.skip("not a copy of king", f"cannot read king: {type(e).__name__}")
if not krepo:
return report.skip("not a copy of king", "king unknown")
if repo == krepo and revision == krev:
return report.fail("not a copy of king", "identical to the reigning king (repo+revision)")
try:
kinfo = fetch_info(krepo, krev, token, Report()) # throwaway report
except Exception as e:
return report.skip("not a copy of king", f"king metadata unavailable: {type(e).__name__}")
cb, kb = info["blobs"], kinfo["blobs"]
if not cb or not kb:
return report.skip("not a copy of king", "blob digests unavailable (fails open upstream)")
from collections import Counter
cc, kc = Counter(cb.values()), Counter(kb.values())
if any(cc[dg] < n for dg, n in kc.items()):
shared = sum(min(cc[dg], n) for dg, n in kc.items())
return report.ok("not a copy of king", f"{shared}/{len(kb)} king blobs shared")
report.fail("not a copy of king", f"all {len(kb)} king blobs present -> copy reject")
async def _probe(port: int, repo: str, turns: int, temp: float,
max_thought: int, max_action: int) -> str | None:
"""Mirrors evalsrv.dueling.probe_injectable: needs one parsable closed
```bash action across `turns`, plus finite forced logprobs."""
import httpx
import math
import screen
from vllm_client import Served, VllmModel
corpus = screen.load_corpus()
tids = sorted(corpus)[:turns]
served = Served("cand", repo, None, port)
sem = asyncio.Semaphore(4)
async with httpx.AsyncClient(limits=httpx.Limits(max_connections=8)) as http:
model = VllmModel(served, http, sem)
any_action = False
for tid in tids:
prefix = corpus[tid]["prefix"]
try:
z, y = await model.sample(prefix, temp, max_thought + max_action)
except Exception as e:
return f"probe_sample_failed:{type(e).__name__}:{e}"
if not y:
continue
any_action = True
try:
scored = await model.score_action(prefix, z, y)
except Exception as e:
return f"probe_force_failed:{type(e).__name__}:{e}"
if not math.isfinite(scored["lp_per_byte"]):
return f"probe_nonfinite_logprob:{scored['lp_per_byte']}"
if scored["n_tokens"] == 0:
return "probe_empty_action_span"
if not any_action:
return f"probe_no_parsable_action_in_{len(tids)}_turns"
return None
def check_probe(args, report: Report) -> None:
if not args.port:
report.skip("promptability", "no --port; serve the candidate to run this "
"(it is the gate fine-tuning breaks)")
return
try:
reason = asyncio.run(_probe(args.port, args.served_name or args.repo,
args.probe_turns, args.temperature,
args.max_thought, args.max_action))
except Exception as e:
report.fail("promptability", f"probe crashed: {type(e).__name__}: {e}")
return
if reason:
report.fail("promptability", f"unpromptable:{reason}")
else:
report.ok("promptability", f"parsable ```bash action within {args.probe_turns} turns")
def watch_king(report: Report) -> int:
"""Post-crown liveness: the reigning king must stay anonymously fetchable
forever, because the validator re-probes it before every duel. Run this on a
timer for as long as we hold a slot in the reign chain."""
try:
snap = _get(SNAPSHOT_API)
except Exception as e:
print(f"cannot read snapshot: {type(e).__name__}: {e}")
return 2
king = snap.get("king") or {}
repo, rev = king.get("repo"), king.get("revision")
print(f"king {repo}@{(rev or '')[:12]} reign {king.get('reign_number')}")
ok = check_anon_access(repo, rev, report, gate="king anon access")
chain = (snap.get("reign") or {}).get("members") or []
earning = [m for m in chain if m.get("earning")]
print(f"reign chain: {len(earning)} earning of {len(chain)} known kings")
for m in earning:
print(f" reign {m.get('reign_number')} uid {m.get('uid')} "
f"{m.get('repo')} {m.get('weight_bps')} bps "
f"tao/day {m.get('tao_per_day')}")
if m.get("repo") != repo:
check_anon_access(m.get("repo"), m.get("revision"), report,
gate=f"reign {m.get('reign_number')} access")
report.render()
return 0 if ok and not report.failed else 1
def main() -> None:
ap = argparse.ArgumentParser(description=__doc__.split("\n")[0])
ap.add_argument("--repo", help="required unless --watch-king")
ap.add_argument("--revision", help="40-hex commit sha")
ap.add_argument("--hotkey-ss58", default="", help="submitter hotkey ss58")
ap.add_argument("--coldkey-ss58", default="",
help="fallback when bittensor is unavailable. The validator "
"binds identity to the ON-CHAIN coldkey; a local "
"coldkeypub that has since been swapped would be wrong.")
ap.add_argument("--wallet", default="subnet120")
ap.add_argument("--hotkey", default="", help="wallet hotkey name, e.g. hotkey-1")
ap.add_argument("--hf-token", default="")
ap.add_argument("--port", type=int, default=0, help="serve the candidate to probe it")
ap.add_argument("--served-name", default="")
ap.add_argument("--probe-turns", type=int, default=3)
ap.add_argument("--temperature", type=float, default=0.8)
ap.add_argument("--max-thought", type=int, default=1024)
ap.add_argument("--max-action", type=int, default=768)
ap.add_argument("--margin", type=float, default=None,
help="paired mean Reason margin from the screen")
ap.add_argument("--se", type=float, default=None,
help="SE of that margin, scaled to the duel's n_turns")
ap.add_argument("--expect-wvk", type=int, default=EXPECT_WVK,
help="weight_version_key the screens were run under")
ap.add_argument("--watch-king", action="store_true",
help="post-crown mode: check the live king and every earning "
"reign-chain member is still anonymously fetchable")
args = ap.parse_args()
if args.watch_king:
sys.exit(watch_king(Report()))
if not args.repo or not args.revision:
sys.exit("--repo and --revision are required (or use --watch-king)")
import pathlib
wdir = pathlib.Path.home() / ".bittensor/wallets" / args.wallet
hk = args.hotkey_ss58
if not hk and args.hotkey:
p = wdir / "hotkeys" / f"{args.hotkey}pub.txt"
if p.exists():
hk = json.load(open(p))["ss58Address"]
if not hk:
sys.exit("need --hotkey-ss58 or a resolvable --wallet/--hotkey")
local_ck = args.coldkey_ss58
if not local_ck and (wdir / "coldkeypub.txt").exists():
local_ck = json.load(open(wdir / "coldkeypub.txt")).get("ss58Address", "")
print(f"preflight repo={args.repo} rev={args.revision[:12]} hotkey={hk[:12]}…")
print("READ-ONLY: this script never reveals or submits.\n")
report = Report()
cfg = load_contract(report)
check_contract_fork(cfg, args.expect_wvk, report)
if not re.fullmatch(r"[0-9a-f]{40}", args.revision):
report.fail("revision format", f"{args.revision!r} is not a 40-hex commit sha")
uid, coldkey = check_registration(hk, report)
if coldkey is None and local_ck:
# The hotkey pair alone is always checkable, so this only widens the
# accepted set. Flag it: an on-chain coldkey swap would not show here.
coldkey = local_ck
report.skip("coldkey source", f"unverified local coldkeypub {local_ck[:10]}…")
check_slot(hk, report)
check_reveal_window(cfg, report)
check_reveal_payload(cfg, args.repo, args.revision, hk, report)
pairs = identity_pairs(hk, coldkey, cfg)
check_repo_name(args.repo, pairs, cfg, report)
check_anon_access(args.repo, args.revision, report)
check_revision_novel(args.repo, args.revision, report)
info = None
try:
info = fetch_info(args.repo, args.revision, args.hf_token, report)
except Exception as e:
report.fail("revision", f"revision_not_found: {type(e).__name__}: {e}")
if info:
check_hygiene(info, cfg, report)
check_index_complete(info, report)
check_copy(args.repo, args.revision, info, args.hf_token, report)
else:
report.skip("hygiene", "no repo metadata")
report.skip("index complete", "no repo metadata")
report.skip("not a copy of king", "no repo metadata")
check_crown_bar(args.margin, args.se, cfg, report)
check_probe(args, report)
report.render()
if report.failed:
print("\n DO NOT SUBMIT — a failing gate burns the slot permanently.\n")
sys.exit(1)
if report.skipped:
print("\n Gates were SKIPPED. A skipped gate is not a pass — resolve them\n"
" before revealing, especially promptability.\n")
sys.exit(2)
print("\n All gates pass. Residual risk remains ~9% eval_infra_exhausted,\n"
" which is validator-side and cannot be preflighted.\n"
" Submission is still a manual step — this script does not perform it.\n"
" AFTER a crown: keep the repo public forever and run --watch-king on a\n"
" timer. Re-gating a reigning king rolls the reign back (reign 7).\n")
if __name__ == "__main__":
main()

Xet Storage Details

Size:
33.8 kB
·
Xet hash:
d91acc7fa01c8607be04a2e4f9473a5027e021195f3ee8e7c60e69e8100996ab

Xet efficiently stores files, intelligently splitting them into unique chunks and accelerating uploads and downloads. More info.