diff --git a/Makefile b/Makefile index 2cea8ed..430744f 100644 --- a/Makefile +++ b/Makefile @@ -243,10 +243,16 @@ control-ab: judge-self-test ## #000057 v1: Hermes-solo vs Arborist, blinded Opus # make control-sweep CONTROL_SWEEP_N=5 CONTROL_SWEEP_FIXTURE ?= bench/qa_questions_stale_map.json CONTROL_SWEEP_N ?= 3 -control-sweep: ## #000057: model×framing control sweep, review table [CONTROL_SWEEP_N=3 ...] +CONTROL_SWEEP_ARB_N ?= 40 +CONTROL_SWEEP_WORKERS ?= 6 +CONTROL_SWEEP_RESUME ?= +control-sweep: ## #000057: model×framing control sweep [CONTROL_SWEEP_N / _WORKERS / _RESUME=jsonl ...] $(PY) bench/control_sweep.py \ --fixture $(CONTROL_SWEEP_FIXTURE) \ --n $(CONTROL_SWEEP_N) \ + --arborist-n $(CONTROL_SWEEP_ARB_N) \ + --max-workers $(CONTROL_SWEEP_WORKERS) \ + $(if $(CONTROL_SWEEP_RESUME),--resume $(CONTROL_SWEEP_RESUME),) \ --shards-dir $(SHARDS_DIR) \ --out-dir $(BENCH_QA_OUT) diff --git a/bench/control_sweep.py b/bench/control_sweep.py index ed02372..4466fd6 100644 --- a/bench/control_sweep.py +++ b/bench/control_sweep.py @@ -217,21 +217,38 @@ def _bucket(c: Counter) -> str: f"JE={c['JUDGE_ERROR']} (n={tot})") -def _aggregate(jsonl_path: Path): - tally: dict[tuple, Counter] = {} - seen_items: set = set() - n_recs = 0 +def _load_recs(jsonl_path: Path) -> list[dict]: + """Tolerant: a killed-mid-write run can leave a truncated final + line — skip it rather than crash the aggregator/resume scan.""" + recs = [] for ln in jsonl_path.read_text().splitlines(): if not ln.strip(): continue - r = json.loads(ln) + try: + recs.append(json.loads(ln)) + except json.JSONDecodeError: + continue # truncated trailing line from a kill + return recs + + +def _aggregate(jsonl_path: Path): + # Dedupe by (i, arm, model, variant), last-wins: a resume re-runs + # any item that was incomplete when the prior run was killed, so + # the same cell can appear twice — the later (complete-run) record + # is authoritative. Without this, a restart double-counts. + latest: dict[tuple, dict] = {} + for r in _load_recs(jsonl_path): if r.get("arm") == "skip": continue - n_recs += 1 + k = (r["i"], r["arm"], r["model"], r["variant"]) + latest[k] = r + tally: dict[tuple, Counter] = {} + seen_items: set = set() + for r in latest.values(): seen_items.add(r["i"]) key = (r["arm"], r["model"], r["variant"]) tally.setdefault(key, Counter())[r["verdict_eff"]] += 1 - return tally, len(seen_items), n_recs + return tally, len(seen_items), len(latest) def _report(tally, n_items, n_recs, args, jsonl_path, ts, @@ -319,6 +336,12 @@ def main() -> int: ap.add_argument("--report-only", default="", help="aggregate a (partial) JSONL with NO spend " "and exit — the interim check-in path") + ap.add_argument("--resume", default="", + help="append to an existing JSONL, skipping items " + "already COMPLETE in it (re-runs partial " + "items; aggregator dedupes last-wins) — for " + "restarting at higher --max-workers without " + "losing finished units") ap.add_argument("--skip-self-test", action="store_true") a = ap.parse_args() @@ -352,21 +375,45 @@ def main() -> int: print("ABORT: judge unreliable") return 1 - ts = time.strftime("%Y-%m-%dT%H-%M-%SZ", time.gmtime()) - outp = Path(a.out_dir) / f"control_sweep_{ts}.jsonl" + skip_items: set[int] = set() + if a.resume: + outp = Path(a.resume) + ts = outp.stem.replace("control_sweep_", "") + exp_solo = len(models) * len(variants) + per_item: dict[int, Counter] = {} + for r in _load_recs(outp): + if r.get("arm") == "skip": + continue + per_item.setdefault(r["i"], Counter())[r["arm"]] += 1 + for i in range(1, len(items) + 1): + c = per_item.get(i) + if not c: + continue + need_arb = len(variants) if i <= arb_units else 0 + if c["solo"] >= exp_solo and c["arborist"] >= need_arb: + skip_items.add(i) + open_mode = "a" + print(f" RESUME {outp}: {len(skip_items)} complete items " + f"skipped, {len(items)-len(skip_items)} to run " + f"(dedupe is last-wins at aggregation)") + else: + ts = time.strftime("%Y-%m-%dT%H-%M-%SZ", time.gmtime()) + outp = Path(a.out_dir) / f"control_sweep_{ts}.jsonl" + open_mode = "w" outp.parent.mkdir(parents=True, exist_ok=True) shards_dir = Path(a.shards_dir) lock = threading.Lock() t0 = time.time() done_units = 0 - with open(outp, "w") as log, ThreadPoolExecutor( + with open(outp, open_mode) as log, ThreadPoolExecutor( max_workers=a.max_workers) as ex: futs = { ex.submit(_process_item, i, it, variants, models, shards_dir, i <= arb_units, a.arborist_ref, ts): i for i, it in enumerate(items, 1) + if i not in skip_items } for fut in as_completed(futs): i = futs[fut] @@ -381,9 +428,10 @@ def main() -> int: log.flush() done_units += 1 el = time.time() - t0 + to_run = len(items) - len(skip_items) rate = done_units / el if el else 0 - eta = (len(items) - done_units) / rate / 60 if rate else 0 - print(f" unit {done_units}/{len(items)} (item {i}) " + eta = (to_run - done_units) / rate / 60 if rate else 0 + print(f" unit {done_units}/{to_run} (item {i}) " f"· {el/60:.1f}m elapsed · ETA {eta:.0f}m") print(f"\nsweep complete in {(time.time()-t0)/60:.1f}m")