feat(eval): run a task's benchmark cells in waves instead of one at a time

18 cells at ~48 min each, strictly serial, is 97.4% of a generation's
14.7h. The cells are independent — the wall clock was a scheduling
choice, not a measurement requirement.

`--workers` (default 1) runs the cells of one task concurrently; tasks
stay sequential, so the sanitized graph snapshot each task already builds
before its cells stays a single-writer affair. Threads, not processes:
cells are subprocess-bound and `run_managed` keeps every piece of
ownership state local to its own call, so nothing is shared to race on.

Waves, not one fan-out. The outage breaker counts CONSECUTIVE systemic
failures, and "consecutive" means nothing in completion order — folding
as futures landed would make the trip point flaky between identical runs.
Each wave is folded in submission order once complete, and the next wave
starts only if the breaker held, so the breaker overruns by at most
`workers - 1` cells (the ones already in flight) rather than by a whole
task.

`--workers 1` calls the cell directly rather than using a pool of one.
That is not an optimisation: an async KeyboardInterrupt is delivered only
to the main thread, so a cell on a worker thread is outside the reach of
the ownership cleanup that kills its sandboxed process tree. The default
therefore stays exactly what it is today, Ctrl-C included, and above 1
the flag's help says what is given up.

All bookkeeping stays on the main thread — the results.jsonl append, the
per-arm accumulation, the progress prints, the streak fold. No lock is
needed anywhere, the progress counter cannot race, prints do not
interleave, and results.jsonl keeps its canonical order.

Every future is read. An exception a cell did not expect stays parked
inside its Future until something asks for it; unread, a harness bug
would become a silently missing run instead of a crash.

eval/tests/test_runner_hardening.py

@@ -551,3 +551,116 @@ def test_run_cell_fails_closed_when_a_per_task_snapshot_never_materialized(tmp_p
 
     assert record["error_kind"] == "infra-error"
     assert "no assets" in str(record["error_detail"])
+
+
+def _sweep(cells, *, workers, run, outage_limit=5, streak=0):
+    """Drive sweep_task_cells, recording what it started and kept."""
+    started: list[tuple[int, str]] = []
+    kept: list[tuple[int, str]] = []
+    ending_streak, tripped = runner.sweep_task_cells(
+        cells,
+        workers=workers,
+        run=run,
+        on_start=lambda run_idx, arm: started.append((run_idx, arm)),
+        on_record=lambda run_idx, arm, _record: kept.append((run_idx, arm)),
+        outage_streak=streak,
+        outage_limit=outage_limit,
+    )
+    return SimpleNamespace(started=started, kept=kept, streak=ending_streak, tripped=tripped)
+
+
+def _row(error_kind=None):
+    return {"resolved": error_kind is None, "error_kind": error_kind}
+
+
+CELLS = [(run_idx, arm) for run_idx in range(3) for arm in ("workflow", "candidate_workflow")]
+
+
+def test_sweep_keeps_rows_in_submission_order_whatever_order_they_finish(tmp_path):
+    # Cells finish in whatever order the machine allows, but a wave is folded
+    # in submission order — the outage streak counts consecutive failures, and
+    # "consecutive" in completion order would make the trip point flaky.
+    import time as _time
+
+    def run(run_idx, arm):
+        _time.sleep(0.02 if run_idx == 0 else 0.0)
+        return _row()
+
+    result = _sweep(CELLS, workers=3, run=run)
+
+    assert result.kept == CELLS
+    assert result.started == CELLS
+    assert result.tripped is False
+
+
+@pytest.mark.parametrize("workers", [1, 2, 3])
+def test_sweep_trips_the_breaker_within_one_wave_of_the_serial_point(workers):
+    # Serial stops after the 5th consecutive systemic failure. Cells already in
+    # flight when the breaker trips cannot be recalled, so the overrun is
+    # bounded by the wave — the point of waves is that it is never the whole
+    # task. Ten cells, so the bound is visible rather than hidden by the end.
+    long_task = [(run_idx, arm) for run_idx in range(5) for arm in ("workflow", "candidate_workflow")]
+
+    result = _sweep(long_task, workers=workers, run=lambda *_: _row("session-error"))
+
+    assert result.tripped is True
+    assert len(result.kept) == 5
+    assert 5 <= len(result.started) <= 5 + workers - 1
+    assert len(result.started) < len(long_task)
+
+
+def test_sweep_reads_a_real_failure_as_signal_rather_than_an_outage():
+    # resolved=False with no systemic error_kind is the benchmark working, not
+    # the harness failing; it must reset the streak instead of tripping.
+    result = _sweep(CELLS, workers=3, run=lambda *_: {"resolved": False, "error_kind": None})
+
+    assert result.tripped is False
+    assert result.streak == 0
+    assert result.kept == CELLS
+
+
+def test_sweep_surfaces_an_unexpected_worker_failure_instead_of_dropping_the_cell():
+    def run(run_idx, arm):
+        if (run_idx, arm) == (0, "candidate_workflow"):
+            raise KeyError("harness bug")
+        return _row()
+
+    # A Future holds its exception until read. Unread, this cell would vanish
+    # from the evidence with no crash and no row — fewer runs in an arm's
+    # aggregate, silently.
+    with pytest.raises(KeyError):
+        _sweep(CELLS, workers=3, run=run)
+
+
+def test_sweep_runs_cells_of_a_wave_at_the_same_time():
+    import threading
+
+    barrier = threading.Barrier(3, timeout=10)
+
+    def run(run_idx, arm):
+        # Deadlocks unless all three cells of the wave are genuinely in flight
+        # together — a pool that serialised them would time out here.
+        barrier.wait()
+        return _row()
+
+    result = _sweep(CELLS, workers=3, run=run)
+
+    assert result.kept == CELLS
+
+
+def test_sweep_of_one_worker_never_leaves_the_calling_thread():
+    import threading
+
+    caller = threading.current_thread()
+    seen: list[threading.Thread] = []
+
+    def run(run_idx, arm):
+        seen.append(threading.current_thread())
+        return _row()
+
+    # Ctrl-C reaches only the main thread, so the serial default has to stay on
+    # it: a cell on a worker thread is outside the reach of the cleanup that
+    # kills its sandboxed process tree.
+    _sweep(CELLS, workers=1, run=run)
+
+    assert seen == [caller] * len(CELLS)

eval/workflow_bench/runner.py

@@ -35,6 +35,7 @@ model).
 from __future__ import annotations
 
 import argparse
+from concurrent.futures import ThreadPoolExecutor, wait
 import hashlib
 import json
 import os
@@ -44,7 +45,8 @@ import stat
 import statistics
 import tempfile
 import time
-from collections.abc import Mapping
+from collections.abc import Callable, Mapping, Sequence
+from functools import partial
 from dataclasses import dataclass, replace
 from datetime import UTC, datetime, timedelta
 from pathlib import Path
@@ -639,6 +641,64 @@ def systemic_outage_streak(error_kind: str | None, prior_streak: int) -> int:
     return prior_streak + 1 if error_kind in SYSTEMIC_ERROR_KINDS else 0
 
 
+def sweep_task_cells(
+    cells: Sequence[tuple[int, str]],
+    *,
+    workers: int,
+    run: Callable[[int, str], dict[str, Any]],
+    on_start: Callable[[int, str], None],
+    on_record: Callable[[int, str, dict[str, Any]], None],
+    outage_streak: int,
+    outage_limit: int,
+) -> tuple[int, bool]:
+    """Run one task's cells in waves of ``workers``; return (streak, tripped).
+
+    Waves rather than one fan-out, because the outage breaker counts
+    CONSECUTIVE systemic failures and "consecutive" only means anything in a
+    fixed order — completion order under concurrency is not one. Each wave is
+    folded in submission order once it has fully completed, and the next wave
+    starts only if the breaker held, so the breaker overruns its limit by at
+    most ``workers - 1`` cells: the ones already in flight when it tripped.
+
+    ``workers == 1`` runs the cell directly instead of through a pool of one.
+    That is not an optimisation — an async KeyboardInterrupt is delivered only
+    to the main thread, so a cell on a worker thread is outside the reach of
+    the ownership cleanup that kills its sandboxed process tree.
+    """
+    if workers < 1:
+        raise ValueError("workers must be positive")
+    for wave_start in range(0, len(cells), workers):
+        wave = list(cells[wave_start : wave_start + workers])
+        for run_idx, arm in wave:
+            on_start(run_idx, arm)
+        if workers == 1:
+            records = [run(run_idx, arm) for run_idx, arm in wave]
+        else:
+            with ThreadPoolExecutor(max_workers=workers) as pool:
+                futures = [pool.submit(run, run_idx, arm) for run_idx, arm in wave]
+                try:
+                    wait(futures)
+                except KeyboardInterrupt:
+                    pool.shutdown(wait=False, cancel_futures=True)
+                    raise
+                # Every future has to be read. An exception a cell did not
+                # expect stays parked inside its Future until something asks
+                # for it, so skipping this would turn a harness bug into a
+                # silently missing run rather than a crash.
+                records = [future.result() for future in futures]
+        for (run_idx, arm), record in zip(wave, records, strict=True):
+            on_record(run_idx, arm, record)
+            outage_streak = systemic_outage_streak(record.get("error_kind"), outage_streak)
+            if outage_limit and outage_streak >= outage_limit:
+                print(
+                    f"[systemic-outage] {outage_streak} consecutive session/infra/cleanup "
+                    "failures — aborting the remaining sweep; report and promotion are written "
+                    "from partial evidence and the run exits non-zero."
+                )
+                return outage_streak, True
+    return outage_streak, False
+
+
 @dataclass(frozen=True)
 class TaskCellContext:
     """Everything one benchmark cell needs from its task, prepared once.
@@ -1075,6 +1135,17 @@ def build_parser() -> argparse.ArgumentParser:
     parser = argparse.ArgumentParser(description=__doc__)
     parser.add_argument("--tasks", required=True, type=Path)
     parser.add_argument("--runs", type=int, default=1)
+    parser.add_argument(
+        "--workers",
+        type=int,
+        default=1,
+        help="cells of one task to run at once (default 1, fully serial). Size "
+        "this to the machine: a cell that loses CPU to its siblings takes "
+        "longer, and a session that reaches its timeout is an excluded run the "
+        "promotion gate refuses to work with. Above 1 the cells run on worker "
+        "threads, so Ctrl-C no longer reaches the code owning a sandboxed "
+        "process and an abort waits for the running cells to finish.",
+    )
     parser.add_argument(
         "--outage-streak",
         type=int,
@@ -1215,6 +1286,8 @@ def main() -> None:
             parser.error(f"{candidate_arm} must be paired with {incumbent_arm}")
     if args.runs < 1 or args.promotion_min_runs < 1:
         parser.error("--runs and --promotion-min-runs must be positive")
+    if args.workers < 1:
+        parser.error("--workers must be positive")
 
     candidate_overlay = args.candidate_overlay.expanduser().absolute() if args.candidate_overlay is not None else None
     overlay_digest = candidate_overlay_digest(candidate_overlay) if candidate_overlay is not None else None
@@ -1350,43 +1423,49 @@ def main() -> None:
                 overlay_digest=overlay_digest,
             )
             per_arm: dict[str, list[dict[str, Any]]] = {a: [] for a in args.arms}
-            for run_idx in range(args.runs):
-                if outage_tripped:
-                    break
-                for arm in args.arms:
-                    if outage_tripped:
-                        break
-                    started_cells += 1
-                    print(
-                        f"[{task['id']}][{arm}][run {run_idx}] starting "
-                        f"({started_cells}/{total_cells}, {(time.monotonic() - sweep_started) / 60:.0f}m elapsed)"
-                    )
-                    record = run_cell(cell_context, run_idx, arm)
-                    per_arm[arm].append(record)
-                    with results_path.open("a") as fh:
-                        # Redact any API token a session-error stderr_tail
-                        # echoed into error_detail before it enters the uploaded
-                        # results.jsonl artifact (transcripts are redacted; this
-                        # sink was not).
-                        fh.write(redact_text(json.dumps(record), [args.auth_token or ""]) + "\n")
-                    print(
-                        f"[{task['id']}][{arm}][run {run_idx}] resolved={record['resolved']} "
-                        f"in={record['input_tokens']} out={record['output_tokens']} "
-                        f"cost=${_na(record['cost_usd'])} "
-                        f"took={_na(record.get('duration_s'))}s "
-                        # An excluded run is what actually blocks promotion, so
-                        # name it here instead of leaving it to results.jsonl.
-                        f"error_kind={record.get('error_kind') or 'none'}"
-                    )
-                    outage_streak = systemic_outage_streak(record.get("error_kind"), outage_streak)
-                    if args.outage_streak and outage_streak >= args.outage_streak:
-                        outage_tripped = True
-                        print(
-                            f"[systemic-outage] {outage_streak} consecutive session/infra/cleanup "
-                            "failures — aborting the remaining sweep; report and promotion are written "
-                            "from partial evidence and the run exits non-zero."
-                        )
-                        break
+            cells = [(run_idx, arm) for run_idx in range(args.runs) for arm in args.arms]
+
+            def announce(run_idx: int, arm: str, task_id: str = task["id"]) -> None:
+                nonlocal started_cells
+                started_cells += 1
+                print(
+                    f"[{task_id}][{arm}][run {run_idx}] starting "
+                    f"({started_cells}/{total_cells}, {(time.monotonic() - sweep_started) / 60:.0f}m elapsed)"
+                )
+
+            def keep(
+                run_idx: int,
+                arm: str,
+                record: dict[str, Any],
+                task_id: str = task["id"],
+                rows: dict[str, list[dict[str, Any]]] = per_arm,
+            ) -> None:
+                rows[arm].append(record)
+                with results_path.open("a") as fh:
+                    # Redact any API token a session-error stderr_tail echoed
+                    # into error_detail before it enters the uploaded
+                    # results.jsonl artifact (transcripts are redacted; this
+                    # sink was not).
+                    fh.write(redact_text(json.dumps(record), [args.auth_token or ""]) + "\n")
+                print(
+                    f"[{task_id}][{arm}][run {run_idx}] resolved={record['resolved']} "
+                    f"in={record['input_tokens']} out={record['output_tokens']} "
+                    f"cost=${_na(record['cost_usd'])} "
+                    f"took={_na(record.get('duration_s'))}s "
+                    # An excluded run is what actually blocks promotion, so name
+                    # it here instead of leaving it to results.jsonl.
+                    f"error_kind={record.get('error_kind') or 'none'}"
+                )
+
+            outage_streak, outage_tripped = sweep_task_cells(
+                cells,
+                workers=args.workers,
+                run=partial(run_cell, cell_context),
+                on_start=announce,
+                on_record=keep,
+                outage_streak=outage_streak,
+                outage_limit=args.outage_streak,
+            )
             results[task["id"]] = {a: aggregate(rs) for a, rs in per_arm.items() if rs}
 
     selection_report = [