diff --git a/eval/tests/test_runner_hardening.py b/eval/tests/test_runner_hardening.py
index f1d012d99..f72c4b797 100644
--- a/eval/tests/test_runner_hardening.py
+++ b/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)
diff --git a/eval/workflow_bench/runner.py b/eval/workflow_bench/runner.py
index 62ab0c1aa..b86a1a132 100644
--- a/eval/workflow_bench/runner.py
+++ b/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 = [