Buckets:
| """Stage-1 screen: estimate a candidate's duel margin on 2 GPUs, no teacher, no king. | |
| Why this is possible at all | |
| --------------------------- | |
| The validator publishes complete duel dumps. Two things in them are reusable: | |
| * `teacher_refs` — the teacher's sampled (z_C, y_C) per turn, so the teacher | |
| never has to be served again for these turns; | |
| * the reigning king's own per-turn pair components, recoverable from the | |
| `challenger_rows` of the duel it won. | |
| The ranking term is S = Λ2 + w·clip(L1lift). Λ2 needs teacher logprobs, but | |
| L1lift = lpA(y_C|z_A) − lpA(y_C|∅) | |
| is entirely challenger-side. ΔL1lift alone predicts the paired duel margin at | |
| R² ≈ 0.90 over the 79 published scored duels, so a candidate can be screened | |
| by serving ONE model on 2 GPUs instead of three models on eight. | |
| This is a filter, not the verdict. Promote survivors to a full duel. | |
| Serve the candidate first, byte-matching affine.toml [miner_serving] — tp=2 and | |
| max_model_len=65536. Neither is cosmetic. A MoE routes differently under a | |
| different tensor-parallel reduction order, so a tp=1 measurement compared | |
| against a tp=2 dump baseline carries a systematic offset; and slice prefixes | |
| reach ~30k tokens with sample() asking 1792 more, which 400s mid-duel at 32768. | |
| vllm serve <repo> --revision <sha> --served-model-name <repo> \ | |
| --tensor-parallel-size 2 --max-model-len 65536 \ | |
| --gpu-memory-utilization 0.80 --max-num-batched-tokens 8192 --port 8002 | |
| Then: | |
| python harness/screen.py --repo <repo> --revision <sha> | |
| """ | |
| from __future__ import annotations | |
| import argparse | |
| import asyncio | |
| import gzip | |
| import json | |
| import pathlib | |
| import statistics as st | |
| import time | |
| import score | |
| from calibrate import fit, predict | |
| # httpx / transformers / vllm_client are imported inside run() so that the | |
| # aggregation and decision logic stay importable (and testable) on a machine | |
| # with no GPU stack — see selftest.py. | |
| ROOT = pathlib.Path(__file__).resolve().parent.parent | |
| EVALS = ROOT / "evals" | |
| CORPUS = ROOT / "corpus" / "turns.jsonl" | |
| def load_corpus() -> dict[str, dict]: | |
| """Turn index, keyed `traj_id:turn_idx`, across both published layouts. | |
| The corpus switched shape upstream on 2026-08-10: it used to ship one row | |
| per turn with the prefix already materialized, and now ships one row per | |
| TRAJECTORY carrying the full `messages` list plus a `turns` descriptor of | |
| `msg_pos` cut points. Rebuilding the prefix as `messages[:msg_pos]` | |
| reproduces all 18,384 rows of the old file byte-for-byte, so the two are | |
| the same data at different granularity — and the new one covers 11,476 | |
| turns more. Keep both readable: the dumps on disk span both eras. | |
| """ | |
| idx = {} | |
| with open(CORPUS) as fh: | |
| for line in fh: | |
| r = json.loads(line) | |
| if "turns" not in r: # legacy: already flat | |
| idx[f"{r['traj_id']}:{r['turn_idx']}"] = r | |
| continue | |
| msgs = r["messages"] | |
| for t in r["turns"]: | |
| idx[f"{r['traj_id']}:{t['turn_idx']}"] = { | |
| **{k: v for k, v in r.items() if k not in ("messages", "turns")}, | |
| **t, "prefix": msgs[: t["msg_pos"]], | |
| } | |
| return idx | |
| def load_reference(name: str) -> dict: | |
| """Reference duel: supplies the turn slice, teacher refs, and the incumbent. | |
| In the dump the *challenger* is the model that won, i.e. the current king, | |
| so its rows are the baseline a new candidate must beat. | |
| """ | |
| with gzip.open(EVALS / f"{name}.json.gz") as fh: | |
| d = json.load(fh) | |
| if not d["verdict"]["challenger_wins"]: | |
| raise SystemExit( | |
| f"{name} is not a crowning duel — its challenger never became king. " | |
| "Pass --ref <the duel the current king won>.") | |
| gates = d["verdict"].get("gates", {}) | |
| clip = gates.get("l1_clip", score.DEFAULT_L1_CLIP) | |
| king = {} | |
| for row in d["challenger_rows"]: | |
| if not row.get("valid"): | |
| continue | |
| king[row["turn_id"]] = { | |
| "l1": st.mean(score.clipped_l1_lift(p, clip) for p in row["pairs"]), | |
| "baseline": st.mean(abs(p["lpA_yc_e"]) for p in row["pairs"]), | |
| } | |
| return {"turn_ids": d["turn_ids"], "refs": d["teacher_refs"], "king": king, | |
| "king_repo": d["request"]["challenger_repo"], "clip": clip, | |
| "w": gates.get("l1_weight", score.DEFAULT_L1_WEIGHT)} | |
| async def score_turn(model: VllmModel, prefix: list[dict], refs: list[dict], | |
| n: int, temperature: float, max_thought: int, | |
| max_action: int, gate_check: bool) -> dict | None: | |
| """Mirror of terms.miner_terms, restricted to the challenger-side calls.""" | |
| rollouts = await asyncio.gather(*[ | |
| model.sample(prefix, temperature, max_thought + max_action) | |
| for _ in range(n) | |
| ]) | |
| rollouts = [(z, y) for z, y in rollouts if y] | |
| if not rollouts or not refs: | |
| return None | |
| m = min(len(refs), len(rollouts)) | |
| refs, rollouts = refs[:m], rollouts[:m] | |
| jobs = [] | |
| for i in range(m): | |
| za, ya = rollouts[i] | |
| yc = refs[i]["y"] | |
| jobs.append(model.score_action(prefix, za, yc)) # lpA_yc_za | |
| jobs.append(model.score_action(prefix, "", yc)) # lpA_yc_e | |
| if gate_check: | |
| jobs.append(model.score_action(prefix, za, ya)) # lpA_ya_za | |
| jobs.append(model.score_action(prefix, "", ya)) # lpA_ya_e | |
| res = await asyncio.gather(*jobs) | |
| per = 4 if gate_check else 2 | |
| pairs = [] | |
| for i in range(m): | |
| b = i * per | |
| p = {"lpA_yc_za": res[b]["lp_per_byte"], "lpA_yc_e": res[b + 1]["lp_per_byte"], | |
| "z_a": rollouts[i][0], "y_a": rollouts[i][1]} | |
| if gate_check: | |
| p["lpA_ya_za"] = res[b + 2]["lp_per_byte"] | |
| p["lpA_ya_e"] = res[b + 3]["lp_per_byte"] | |
| pairs.append(p) | |
| return {"pairs": pairs, "n_pairs": m} | |
| async def run(args) -> dict: | |
| import httpx | |
| from vllm_client import Served, VllmModel | |
| ref = load_reference(args.ref) | |
| corpus = load_corpus() | |
| tids = ref["turn_ids"][: args.turns] if args.turns else ref["turn_ids"] | |
| missing = [t for t in tids if t not in corpus] | |
| if missing: | |
| raise SystemExit(f"{len(missing)} reference turns missing from corpus " | |
| f"(is it synced to the right epoch?): {missing[:3]}") | |
| served = Served("challenger", args.repo, args.revision or None, args.port) | |
| sem = asyncio.Semaphore(args.concurrency) | |
| limits = httpx.Limits(max_connections=args.concurrency + 8) | |
| t0 = time.time() | |
| rows, done = {}, 0 | |
| async with httpx.AsyncClient(limits=limits) as http: | |
| model = VllmModel(served, http, sem) | |
| async def one(tid): | |
| nonlocal done | |
| out = await score_turn( | |
| model, corpus[tid]["prefix"], ref["refs"].get(tid, []), | |
| args.samples, args.temperature, args.max_thought, | |
| args.max_action, not args.no_gate) | |
| done += 1 | |
| if done % 10 == 0: | |
| print(f" ... {done}/{len(tids)} turns " | |
| f"({time.time() - t0:.0f}s)", flush=True) | |
| if out: | |
| rows[tid] = out | |
| await asyncio.gather(*[one(t) for t in tids]) | |
| args._elapsed = time.time() - t0 | |
| return aggregate(rows, ref, args) | |
| def aggregate(rows: dict, ref: dict, args) -> dict: | |
| paired = sorted(set(rows) & set(ref["king"])) | |
| if len(paired) < 2: | |
| raise SystemExit("too few paired turns to estimate anything") | |
| clip, w = ref["clip"], ref["w"] | |
| diffs, cand_l1, base = [], [], [] | |
| for tid in paired: | |
| ps = rows[tid]["pairs"] | |
| mine = st.mean(score.clipped_l1_lift(p, clip) for p in ps) | |
| cand_l1.append(mine) | |
| base.append(st.mean(abs(p["lpA_yc_e"]) for p in ps)) | |
| diffs.append(w * (mine - ref["king"][tid]["l1"])) | |
| allp = [p for tid in paired for p in rows[tid]["pairs"]] | |
| dl1 = st.mean(diffs) | |
| f = fit() | |
| pred = predict(dl1, f) | |
| king_base = st.mean(ref["king"][t]["baseline"] for t in paired) | |
| my_base = st.mean(base) | |
| return { | |
| "repo": args.repo, "revision": args.revision, "ref": args.ref, | |
| "king_repo": ref["king_repo"], "n_paired_turns": len(paired), | |
| "delta_l1lift": dl1, | |
| "delta_l1_se": st.stdev(diffs) / len(diffs) ** 0.5, | |
| "margin_hat": pred["margin_hat"], "lo": pred["lo"], "hi": pred["hi"], | |
| "calib_r": score.calibration_ratio(allp), | |
| "gate_rate": (None if args.no_gate else | |
| st.mean(1.0 if score.gate_pass(p) else 0.0 for p in allp)), | |
| "baseline_ratio": my_base / king_base if king_base else None, | |
| "elapsed_s": getattr(args, "_elapsed", 0.0), | |
| "fit_r2": f["r2"], "crown_margin": 3 * f["se_typical"], | |
| "screen_gate": args.screen_gate, | |
| } | |
| def report(r: dict, submit_gate: float) -> None: | |
| crown = r["crown_margin"] | |
| print("\n" + "=" * 68) | |
| print(f" {r['repo']}@{(r['revision'] or 'HEAD')[:12]}") | |
| print(f" vs king {r['king_repo']} slice {r['ref']} " | |
| f"({r['n_paired_turns']} paired turns, {r['elapsed_s']:.0f}s)") | |
| print("=" * 68) | |
| print(f" ΔL1lift {r['delta_l1lift']:+.5f} ± {r['delta_l1_se']:.5f}") | |
| print(f" predicted margin {r['margin_hat']:+.5f} 80% band " | |
| f"[{r['lo']:+.4f}, {r['hi']:+.4f}] (fit R²={r['fit_r2']:.2f})") | |
| print(f" crown threshold {crown:+.5f} (3·se) submit gate {submit_gate:+.4f}") | |
| print() | |
| r_ = r["calib_r"] | |
| warn = [] | |
| print(f" calibration r {r_:.3f} {'OK' if 0.55 <= r_ <= 0.85 else 'OUTSIDE 0.55-0.85'}") | |
| if not 0.55 <= r_ <= 0.85: | |
| warn.append(f"r={r_:.2f} outside 0.55-0.85; crowns cluster at 0.66-0.81 but r varies ~0.09 by slice") | |
| if not 0.3 <= r_ <= 4.0: | |
| warn.append(f"r={r_:.2f} FAILS the hard calibration gate [0.3, 4.0] -> auto-invalid") | |
| if r["gate_rate"] is not None: | |
| print(f" causality gate {r['gate_rate']:.3f} " | |
| f"{'OK' if r['gate_rate'] >= 0.30 else 'FAILS gamma=0.30 -> auto-invalid'}") | |
| if r["gate_rate"] < 0.30: | |
| warn.append("causality pass rate below gamma=0.30 -> miner INVALID") | |
| if r["baseline_ratio"] is not None: | |
| br = r["baseline_ratio"] | |
| print(f" baseline / king {br:.3f} " | |
| f"{'OK' if br <= 1.25 else 'EXCEEDS baseline_band=1.25 -> auto-invalid'}") | |
| if br > 1.25: | |
| warn.append("empty-baseline band exceeded -> challenger gated out") | |
| elif br > 1.15: | |
| warn.append(f"baseline ratio {br:.2f} is close to the 1.25 band") | |
| print() | |
| if warn: | |
| print(" WARNINGS") | |
| for w in warn: | |
| print(f" ! {w}") | |
| print() | |
| # The screen is a recall filter, NOT the submit decision. Backtested | |
| # leave-one-out on 79 duels, a 0.020 gate keeps both known crowns while | |
| # removing 92% of the field; the predictor's mean |error| (0.0071) is too | |
| # close to the crown threshold to make the final call here. The 0.035 | |
| # submit gate applies to the FULL duel result, never to this estimate. | |
| if r["margin_hat"] >= r["screen_gate"]: | |
| verdict = ("PROMOTE — run the full 8-GPU duel next; submit only if the " | |
| f"real duel margin clears {submit_gate:+.3f}") | |
| else: | |
| verdict = "REJECT — do not spend a duel or a hotkey on this checkpoint" | |
| print(f" screen gate {r['screen_gate']:+.3f} " | |
| f"(recall-tuned; 92% of the field falls below it)") | |
| print(f"\n VERDICT: {verdict}") | |
| print("=" * 68) | |
| def main() -> None: | |
| ap = argparse.ArgumentParser(description=__doc__.split("\n")[0]) | |
| ap.add_argument("--repo", required=True) | |
| ap.add_argument("--revision", default="") | |
| ap.add_argument("--ref", default="chal-00388", | |
| help="crowning duel of the current king (default: chal-00388)") | |
| ap.add_argument("--port", type=int, default=8002) | |
| ap.add_argument("--samples", type=int, default=4, help="rollouts per turn") | |
| 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("--concurrency", type=int, default=16) | |
| ap.add_argument("--turns", type=int, default=0, help="limit turns (smoke test)") | |
| ap.add_argument("--no-gate", action="store_true", | |
| help="skip causality-gate calls (2 echo calls/pair instead of 4)") | |
| ap.add_argument("--screen-gate", type=float, default=0.020, | |
| help="promote to a full duel above this predicted margin " | |
| "(recall-tuned; see harness/backtest.py)") | |
| ap.add_argument("--submit-gate", type=float, default=0.035, | |
| help="gate applied to the FULL duel result, not to this estimate") | |
| ap.add_argument("--out", default="") | |
| args = ap.parse_args() | |
| r = asyncio.run(run(args)) | |
| report(r, args.submit_gate) | |
| if args.out: | |
| pathlib.Path(args.out).write_text(json.dumps(r, indent=2)) | |
| print(f"\nwrote {args.out}") | |
| if __name__ == "__main__": | |
| main() | |
Xet Storage Details
- Size:
- 13.2 kB
- Xet hash:
- 44af45304dd990b6a7b33570cedca964177222f63b3767889d2fd0d0cafb0713
·
Xet efficiently stores files, intelligently splitting them into unique chunks and accelerating uploads and downloads. More info.