Buckets:
| """Re-pull the published eval dumps and the turn corpus from the public bucket. | |
| Fail-closed, mirroring evalsrv/corpus.py: the manifest pointer must match its | |
| immutable published copy, and every shard's uncompressed sha256 must verify. | |
| Run before every screening session — the king turns over roughly daily and the | |
| corpus is refreshed as an anti-memorisation measure. | |
| """ | |
| from __future__ import annotations | |
| import concurrent.futures as cf | |
| import gzip | |
| import hashlib | |
| import json | |
| import pathlib | |
| import re | |
| import urllib.parse | |
| import urllib.request | |
| BASE = "https://s3.hippius.com/affine-sn120" | |
| ROOT = pathlib.Path(__file__).resolve().parent.parent | |
| EVALS = ROOT / "evals" | |
| CORPUS = ROOT / "corpus" | |
| def get(url: str, timeout: int = 300) -> bytes: | |
| with urllib.request.urlopen(url, timeout=timeout) as r: | |
| return r.read() | |
| def list_keys(prefix: str) -> list[str]: | |
| keys, token = [], "" | |
| while True: | |
| url = f"{BASE}?list-type=2&prefix={prefix}&max-keys=1000" | |
| if token: | |
| url += f"&continuation-token={urllib.parse.quote(token)}" | |
| body = get(url, 60).decode() | |
| keys += re.findall(rf"<Key>({re.escape(prefix)}[^<]+)</Key>", body) | |
| m = re.search(r"<NextContinuationToken>([^<]+)</NextContinuationToken>", body) | |
| if not m: | |
| return keys | |
| token = m.group(1) | |
| def sync_evals() -> int: | |
| EVALS.mkdir(exist_ok=True) | |
| keys = list_keys("evals/") | |
| def fetch(key: str) -> int: | |
| dest = EVALS / pathlib.PurePosixPath(key).name | |
| if dest.exists() and dest.stat().st_size > 1000: | |
| return 0 | |
| dest.write_bytes(get(f"{BASE}/{key}")) | |
| return 1 | |
| with cf.ThreadPoolExecutor(16) as ex: | |
| new = sum(ex.map(fetch, keys)) | |
| print(f"evals: {len(keys)} published, {new} newly downloaded") | |
| return new | |
| def sync_corpus() -> None: | |
| CORPUS.mkdir(exist_ok=True) | |
| raw = get(f"{BASE}/turns/manifest.json", 60) | |
| digest = hashlib.sha256(raw).hexdigest() | |
| if get(f"{BASE}/turns/manifests/{digest}.json", 60) != raw: | |
| raise SystemExit("manifest pointer does not match its immutable copy — refusing") | |
| man = json.loads(raw) | |
| total = 0 | |
| tmp = CORPUS / "turns.jsonl.tmp" | |
| with open(tmp, "wb") as out: | |
| for shard in man["shards"]: | |
| if not shard.get("active"): | |
| continue | |
| data = gzip.decompress(get(f"{BASE}/{shard['key']}")) | |
| got = hashlib.sha256(data).hexdigest() | |
| if got != shard["sha256"]: | |
| raise SystemExit(f"shard {shard['key']} sha256 mismatch — refusing") | |
| out.write(data) | |
| total += shard["n_turns"] | |
| print(f" {shard['key']:<40} n={shard['n_turns']:>6} sha256 OK") | |
| tmp.replace(CORPUS / "turns.jsonl") | |
| print(f"corpus: epoch {man.get('corpus_epoch')}, {total} turns, manifest " | |
| f"{digest[:16]}") | |
| def latest_crowning_duel() -> None: | |
| """Print the --ref to use: the duel the current king won.""" | |
| idx = EVALS / "index.jsonl" | |
| if not idx.exists(): | |
| return | |
| wins = [json.loads(x) for x in idx.read_text().splitlines()] | |
| wins = [w for w in wins if w.get("challenger_wins")] | |
| if not wins: | |
| print("\nno crowning duel in index.jsonl — pass --ref explicitly") | |
| return | |
| last = max(wins, key=lambda w: w["at"]) | |
| print(f"\ncurrent king: {last['repo']}") | |
| print(f" crowned {last['at'][:19]} margin {last['margin']:+.4f}") | |
| print(f" use: --ref {last['challenge_id']}") | |
| if __name__ == "__main__": | |
| sync_evals() | |
| sync_corpus() | |
| latest_crowning_duel() | |
Xet Storage Details
- Size:
- 3.59 kB
- Xet hash:
- 3fb9f503ccf3ca2e840d4e3d19fd1a7252185f04ff7eaaeb741914371aef50f3
·
Xet efficiently stores files, intelligently splitting them into unique chunks and accelerating uploads and downloads. More info.