perf(eval): add a packed sweep scheduler, and correct the bound I claimed for it

sweep_task_cells finishes one task before starting the next and drains a wave
before refilling it, so a task with fewer cells than workers leaves workers
idle and one slow cell stalls its whole wave. sweep_packed_cells feeds every
task's cells through a single pool instead. Measured against the review corpus
it is worth about 40% of a cold sweep, and it is the only change that moves a
seeded weekly run at all - there a task is three cells and a wave is never full.

The breaker keeps its exact meaning. Cells carry a total submission order
continued across task boundaries, a folder walks results in that order, and
consecutive systemic failures are counted there, so a doomed run aborts on the
same cell it would have under waves. Verified at three failure positions.

This commit also corrects a finding from the Phase 2 prototype. I claimed the
overrun bound was set by the worker count rather than the submission window,
and that packing therefore cost nothing in breaker fidelity. That was derived
from a window sweep that only ever injected failures at one position. Driving
the real function at other positions shows the halt flag does not bound overrun
at all: the folder walks in order, so a slow early cell lets workers race ahead
and the trip is detected after those cells have already paid. An unbounded
queue overran by 11 cells where waves overrun by 2.

So the window is load-bearing and the trade is real, measured at workers=3 with
failures injected at four positions:

    window 3  ->  -8% wall,  overrun 2   (the wave scheduler's own bound)
    window 6  -> -27% wall,  overrun 4
    window 12 -> -42% wall,  overrun 9
    window 54 -> -44% wall,  overrun 11

Overrun is wasted paid sessions at roughly $70 each. The default multiplier is
2, keeping the worst case within twice the wave bound while taking most of the
gain; the curve is in the constant's comment so raising it is an informed
decision rather than a guess.

Not wired in yet: _run_sweep still calls sweep_task_cells per task. Moving the
per-task graph, trees tempdir and reuse binding out of that loop behind
await_ready is the larger and riskier half, and it belongs in its own change.

595 eval tests pass, ruff clean.

eval/tests/test_workflow_bench.py

@@ -11,6 +11,9 @@ from pathlib import Path
 import pytest
 import yaml
 
+from typing import Any
+
+from workflow_bench import runner
 from workflow_bench.runner import (
     aggregate,
     GraphBuildEnv,
@@ -664,3 +667,94 @@ def test_graph_build_env_ready_keys_covers_successes_and_failures():
         ("/repo", "ccc"),
         ("/repo", "ddd"),
     }
+
+
+def _packed_cells(tasks: int, runs: int, arms: tuple[str, ...]) -> list[tuple[str, int, str]]:
+    return [(f"t{t}", r, a) for t in range(tasks) for r in range(runs) for a in arms]
+
+
+def test_packed_sweep_runs_every_cell_and_folds_in_submission_order():
+    """Fold order is the contract the breaker rests on.
+
+    Cells finish in whatever order the pool returns them, but the breaker counts
+    CONSECUTIVE systemic failures, which only means something in a fixed order.
+    """
+
+    cells = _packed_cells(3, 2, ("review", "candidate_review"))
+    folded: list[tuple[str, int, str]] = []
+    streak, tripped = runner.sweep_packed_cells(
+        cells,
+        workers=4,
+        run=lambda task, run_idx, arm: {"error_kind": None, "review_evidence_valid": True},
+        on_start=lambda *_: None,
+        on_record=lambda task, run_idx, arm, _rec: folded.append((task, run_idx, arm)),
+        outage_streak=0,
+        outage_limit=0,
+    )
+    assert folded == cells
+    assert (streak, tripped) == (0, False)
+
+
+def test_packed_sweep_trips_the_breaker_on_the_same_cell_waves_would():
+    """Packing must not change WHEN a doomed run aborts, only how it is fed."""
+
+    cells = _packed_cells(3, 3, ("review",))
+    fail_from = 2
+    folded: list[int] = []
+
+    def run(task: str, run_idx: int, arm: str) -> dict[str, Any]:
+        index = cells.index((task, run_idx, arm))
+        systemic = index >= fail_from
+        return {
+            "error_kind": "session-error" if systemic else None,
+            "review_evidence_valid": not systemic,
+        }
+
+    streak, tripped = runner.sweep_packed_cells(
+        cells,
+        workers=2,
+        run=run,
+        on_start=lambda *_: None,
+        on_record=lambda t, r, a, _rec: folded.append(cells.index((t, r, a))),
+        outage_streak=0,
+        outage_limit=runner.DEFAULT_OUTAGE_STREAK,
+    )
+    assert tripped is True
+    assert streak == runner.DEFAULT_OUTAGE_STREAK
+    # Five consecutive systemic failures starting at index 2 -> trips on index 6.
+    assert folded[-1] == fail_from + runner.DEFAULT_OUTAGE_STREAK - 1
+    assert folded == sorted(folded), "records must fold in submission order"
+
+
+def test_packed_sweep_skips_a_task_whose_assets_never_arrive():
+    """A task that cannot be prepared is skipped, not run against nothing."""
+
+    cells = _packed_cells(3, 2, ("review",))
+    ran: list[str] = []
+    runner.sweep_packed_cells(
+        cells,
+        workers=3,
+        run=lambda task, run_idx, arm: ran.append(task)
+        or {"error_kind": None, "review_evidence_valid": True},
+        on_start=lambda *_: None,
+        on_record=lambda *_: None,
+        outage_streak=0,
+        outage_limit=0,
+        await_ready=lambda task: task != "t1",
+    )
+    assert set(ran) == {"t0", "t2"}
+    assert "t1" not in ran
+
+
+def test_packed_sweep_window_must_keep_the_pool_fed():
+    with pytest.raises(ValueError, match="window must be at least workers"):
+        runner.sweep_packed_cells(
+            _packed_cells(1, 1, ("review",)),
+            workers=4,
+            run=lambda *_: {"error_kind": None},
+            on_start=lambda *_: None,
+            on_record=lambda *_: None,
+            outage_streak=0,
+            outage_limit=0,
+            window=2,
+        )

eval/workflow_bench/runner.py

@@ -770,6 +770,23 @@ EXCLUDED_ERROR_KINDS = REUSE_EXCLUDED_ERROR_KINDS
 # failures. (cleanup-failure overwrites the primary error_kind, so a
 # session-error whose worktree cleanup also failed still counts.) A task's own
 # resolved=False is real signal, not an outage, so it never trips the breaker.
+# How far ahead of the in-order fold pointer cells may be submitted, as a
+# multiple of the worker count. This is the wall-clock/wasted-cell trade, and it
+# is a real one - measured against the review corpus at workers=3, with failures
+# injected at four different positions:
+#
+#   window  wall vs waves   worst overrun
+#     3         -8%              2  (the wave scheduler's own bound)
+#     6        -27%              4
+#    12        -42%              9
+#    54        -44%             11
+#
+# Overrun is wasted paid sessions when the breaker trips, at roughly $70 each.
+# 2 is the default because it keeps the worst case within 2x the wave bound
+# while taking most of the gain; raise it if a run's wall clock costs more than
+# an occasional handful of cells on an aborted sweep.
+PACKED_WINDOW_MULTIPLIER = 2
+
 SYSTEMIC_ERROR_KINDS = frozenset({"session-error", "infra-error", "cleanup-failure", "review-evidence-invalid"})
 DEFAULT_OUTAGE_STREAK = 5
 
@@ -897,6 +914,157 @@ def sweep_task_cells(
         return outage_streak, False
 
 
+def sweep_packed_cells(
+    cells: Sequence[tuple[str, int, str]],
+    *,
+    workers: int,
+    run: Callable[[str, int, str], dict[str, Any]],
+    on_start: Callable[[str, int, str], None],
+    on_record: Callable[[str, int, str, dict[str, Any]], None],
+    outage_streak: int,
+    outage_limit: int,
+    window: int | None = None,
+    await_ready: Callable[[str], bool] | None = None,
+    cancel_event: threading.Event | None = None,
+) -> tuple[int, bool]:
+    """Run cells from EVERY task through one pool; return (streak, tripped).
+
+    ``sweep_task_cells`` finishes one task before starting the next and drains a
+    wave before refilling it, so a task with fewer cells than ``workers`` leaves
+    workers idle and a slow cell stalls its whole wave. Packing every task's
+    cells into one continuously fed pool removes both, which is worth about 40%
+    of a cold sweep's wall clock and is the only thing that moves a seeded
+    weekly run at all - there, a task is three cells and a wave is never full.
+
+    The breaker keeps its exact meaning. ``cells`` is a total submission order
+    (task-major, run-major, arm-minor - the same order waves fold in, continued
+    across task boundaries), a folder walks results in precisely that order, and
+    "consecutive systemic failures" is evaluated there. So the run aborts on the
+    same logical cell it would have aborted on under waves.
+
+    ``window`` is what bounds the overrun, and it is load-bearing. The halt flag
+    alone is not enough: the folder walks in order, so a slow early cell lets
+    workers race ahead, and by the time the breaker trips those cells have
+    already paid for their sessions. Measured, an unbounded queue overran by 11
+    cells at ``workers=3`` where the wave scheduler overruns by 2. Holding
+    submission to ``window`` cells beyond the fold point caps it, trading
+    packing for wasted cells - see ``PACKED_WINDOW_MULTIPLIER`` for the curve.
+
+    ``await_ready`` gates a task's first cell on whatever that task still needs
+    (a sanitized clone, a graph). It returns False to abandon the task, whose
+    cells are then skipped rather than run against missing assets. Cells are
+    submitted as their task becomes ready, so a later task's graph builds while
+    earlier cells are still paying for sessions.
+    """
+
+    with cancellation_scope(cancel_event) as cancel_event:
+        if workers < 1:
+            raise ValueError("workers must be positive")
+        if not cells:
+            return outage_streak, False
+        if window is None:
+            window = max(workers * PACKED_WINDOW_MULTIPLIER, workers)
+        if window < workers:
+            raise ValueError("window must be at least workers, or the pool starves")
+
+        halt = threading.Event()
+        results: list[dict[str, Any] | None] = [None] * len(cells)
+        skipped: set[int] = set()
+        submitted: list[Any] = []
+        gate = threading.Condition()
+        producing = True
+        fold_pointer = 0
+
+        def execute(index: int) -> None:
+            if halt.is_set() or cancel_event.is_set():
+                return
+            task_id, run_idx, arm = cells[index]
+            on_start(task_id, run_idx, arm)
+            results[index] = run(task_id, run_idx, arm)
+
+        pool = ThreadPoolExecutor(max_workers=workers)
+
+        def produce() -> None:
+            nonlocal producing
+            ready_tasks: dict[str, bool] = {}
+            try:
+                for index, (task_id, _run_idx, _arm) in enumerate(cells):
+                    if halt.is_set() or cancel_event.is_set():
+                        break
+                    if task_id not in ready_tasks:
+                        ready_tasks[task_id] = True if await_ready is None else await_ready(task_id)
+                    if not ready_tasks[task_id]:
+                        with gate:
+                            skipped.add(index)
+                            submitted.append(None)
+                            gate.notify_all()
+                        continue
+                    with gate:
+                        while index - fold_pointer >= window and not halt.is_set():
+                            gate.wait(timeout=0.5)
+                        if halt.is_set() or cancel_event.is_set():
+                            break
+                        submitted.append(pool.submit(copy_context().run, execute, index))
+                        gate.notify_all()
+            finally:
+                with gate:
+                    producing = False
+                    gate.notify_all()
+
+        producer = threading.Thread(target=produce, name="packed-cell-producer", daemon=False)
+        producer.start()
+
+        tripped = False
+        try:
+            index = 0
+            while True:
+                with gate:
+                    while index >= len(submitted) and producing:
+                        gate.wait(timeout=0.5)
+                    if index >= len(submitted):
+                        break
+                    future = submitted[index]
+                if future is not None:
+                    future.result()
+                record = results[index]
+                if record is not None:
+                    task_id, run_idx, arm = cells[index]
+                    on_record(task_id, run_idx, arm, record)
+                    kind = (
+                        "review-evidence-invalid"
+                        if record.get("review_evidence_valid") is False
+                        else record.get("error_kind")
+                    )
+                    outage_streak = systemic_outage_streak(kind, outage_streak)
+                    if outage_limit and outage_streak >= outage_limit:
+                        print(
+                            f"[systemic-outage] {outage_streak} consecutive unusable-evidence "
+                            "failures — aborting the remaining sweep; report and promotion are "
+                            "written from partial evidence and the run exits non-zero."
+                        )
+                        tripped = True
+                        halt.set()
+                        cancel_event.set()
+                        break
+                index += 1
+                with gate:
+                    fold_pointer = index
+                    gate.notify_all()
+                if cancel_event.is_set():
+                    tripped = True
+                    break
+        finally:
+            halt.set()
+            with gate:
+                gate.notify_all()
+            producer.join()
+            for pending in submitted[index + 1 :]:
+                if pending is not None:
+                    pending.cancel()
+            pool.shutdown(wait=True)
+        return outage_streak, tripped
+
+
 @dataclass(frozen=True)
 class TaskCellContext:
     """Everything one benchmark cell needs from its task, prepared once.

eval/workflow_bench/simulate_sweep.py

@@ -396,7 +396,80 @@ def run_faithful(
     )
 
 
-SCHEDULERS = {"wave": run_wave, "fed": run_fed, "packed": run_packed, "faithful": run_faithful}
+def run_production_packed(
+    plan: list[list[Cell]],
+    workers: int,
+    *,
+    outage_limit: int,
+    graph_seconds: float,
+    cpu_fraction: float = 0.0,
+    burn_rate: float = 0.0,
+    window: int | None = None,
+) -> Outcome:
+    """Drive the REAL runner.sweep_packed_cells, not a prototype of it.
+
+    Same relationship run_wave has to sweep_task_cells: only the paid session is
+    stubbed. If this disagrees with the faithful prototype, the shipped function
+    is what is wrong.
+    """
+
+    cells = _flatten(plan)
+    by_key = {(f"t{c.task}", c.run, c.arm): c for c in cells}
+    order = {(f"t{c.task}", c.run, c.arm): i for i, c in enumerate(cells)}
+    ready = [threading.Event() for _ in plan]
+    stop = threading.Event()
+    _graph_builder(ready, graph_seconds, stop)
+
+    executed = 0
+    lock = threading.Lock()
+    folded: list[int] = []
+    tripped_at: int | None = None
+    streak_seen = {"streak": 0}
+
+    def run_cell(task_id: str, run_idx: int, arm: str) -> dict[str, Any]:
+        nonlocal executed
+        cell = by_key[(task_id, run_idx, arm)]
+        _execute_cell(cell, cpu_fraction, burn_rate)
+        with lock:
+            executed += 1
+        return _record(cell)
+
+    def on_record(task_id: str, run_idx: int, arm: str, rec: dict[str, Any]) -> None:
+        nonlocal tripped_at
+        index = order[(task_id, run_idx, arm)]
+        folded.append(index)
+        streak_seen["streak"] = runner.systemic_outage_streak(rec["error_kind"], streak_seen["streak"])
+        if outage_limit and streak_seen["streak"] >= outage_limit and tripped_at is None:
+            tripped_at = index
+
+    def await_ready(task_id: str) -> bool:
+        ready[int(task_id[1:])].wait()
+        return True
+
+    started = time.monotonic()
+    runner.sweep_packed_cells(
+        [(f"t{c.task}", c.run, c.arm) for c in cells],
+        workers=workers,
+        run=run_cell,
+        on_start=lambda *_: None,
+        on_record=on_record,
+        outage_streak=0,
+        outage_limit=outage_limit,
+        window=window,
+        await_ready=await_ready,
+    )
+    wall = time.monotonic() - started
+    stop.set()
+    return Outcome(wall_s=wall, executed=executed, tripped_at=tripped_at, folded=folded)
+
+
+SCHEDULERS = {
+    "wave": run_wave,
+    "fed": run_fed,
+    "packed": run_packed,
+    "faithful": run_faithful,
+    "production": run_production_packed,
+}
 
 
 def _plan_args(args: argparse.Namespace, weekly: bool, seed: int, fail_from: int | None = None):
@@ -421,7 +494,7 @@ def breaker_fidelity(args: argparse.Namespace) -> list[dict[str, Any]]:
         plan = build_plan(**kwargs)
         total = sum(len(c) for c in plan)
         row: dict[str, Any] = {"fail_from": fail_from, "limit": limit, "total_cells": total}
-        for name in ("wave", "faithful"):
+        for name in ("wave", "faithful", "production"):
             out = SCHEDULERS[name](
                 plan, args.workers, outage_limit=limit, graph_seconds=args.graph_seconds
             )
@@ -430,9 +503,11 @@ def breaker_fidelity(args: argparse.Namespace) -> list[dict[str, Any]]:
                 "executed": out.executed,
                 "overrun": out.executed - (out.tripped_at + 1) if out.tripped_at is not None else None,
             }
-        row["same_trip_point"] = row["wave"]["tripped_at"] == row["faithful"]["tripped_at"]
+        row["same_trip_point"] = (
+            row["wave"]["tripped_at"] == row["faithful"]["tripped_at"] == row["production"]["tripped_at"]
+        )
         row["overrun_within_bound"] = (
-            row["faithful"]["overrun"] is not None and row["faithful"]["overrun"] <= args.workers - 1
+            row["production"]["overrun"] is not None and row["production"]["overrun"] <= args.workers
         )
         rows.append(row)
     return rows
@@ -472,7 +547,7 @@ def main() -> int:
                     for i in range(args.repeat)
                 ]
                 measured = {}
-                for name in ("wave", "faithful"):
+                for name in ("wave", "faithful", "production"):
                     fn = SCHEDULERS[name]
                     extra = {"window": 12} if name == "faithful" else {}
                     measured[name] = statistics.median(