feat(#000057): sweep --resume (skip-complete + last-wins dedupe)

fox: 'not 11 hours it shouldn't take that long'. Probed endpoints —
Qwen-27B absorbs 8 concurrent with 0 errors (0.5->2.1s); the
bottleneck is the serialized claude -p Opus judge, so more workers
≈ near-linear speedup. control_sweep.py gains --resume PATH: appends
to an existing JSONL, skips items already COMPLETE (full
models×variants for solo + variants for arborist if i<=arborist-n),
re-runs partial items; _aggregate now dedupes (i,arm,model,variant)
last-wins so a killed-mid-unit restart never double-counts, and
_load_recs tolerates a truncated trailing line from the kill.
Makefile control-sweep gains CONTROL_SWEEP_WORKERS / _RESUME / _ARB_N
so make stays the interface. 6-worker run killed cleanly (specific
pids, no pkill), relaunched resume @ 12 workers — 12 done items
preserved, 374 to run, ~5h -> ~2-2.5h.

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)
 

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")