diff --git a/eval/tests/test_workflow_bench.py b/eval/tests/test_workflow_bench.py index 2bd18214b..4e46bc5da 100644 --- a/eval/tests/test_workflow_bench.py +++ b/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, + ) diff --git a/eval/workflow_bench/runner.py b/eval/workflow_bench/runner.py index 70b1e476f..eeeb056d3 100644 --- a/eval/workflow_bench/runner.py +++ b/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. diff --git a/eval/workflow_bench/simulate_sweep.py b/eval/workflow_bench/simulate_sweep.py index d3e64b006..233147cbc 100644 --- a/eval/workflow_bench/simulate_sweep.py +++ b/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(