scriptoriumadmin/GitNexus
1
0
Fork 0
mirror of https://github.com/abhigyanpatwari/GitNexus.git synced 2026-09-12 23:02:45 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 10

feat(eval): Add bounded packed-scheduler primitives and offline replay benchmarks (#3206)

Browse source 
* perf(eval): packed sweep scheduler and the harness that measured it

Extracted from the combined skill-evolution branch so it can be reviewed on its
own. Purely additive against main: no existing function changes behaviour, and
sweep_packed_cells has no production caller yet.

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, keeping the breaker's meaning: a
total submission order continued across task boundaries, a folder walking
results in that order, and consecutive systemic failures counted there, so a
doomed run aborts on the same cell it would have under waves.

simulate_sweep.py is what produced the numbers. It drives the real schedulers
with only the paid agent session stubbed, using the measured per-arm durations
in session_durations.json divided by a scale factor. The distribution's shape
is kept deliberately - median 826s against a 5400s ceiling - because that
spread is the entire reason a barrier costs anything, and uniform sleeps would
erase the effect under test. All schedulers consume one identical seeded plan.

Measured at workers=3 against the review corpus, packing 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 submission window is
a real trade, measured 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 on an aborted sweep. The default multiplier is
2; the curve lives in the constant's comment so raising it is an informed
decision. Contention was measured separately by burning real CPU in
subprocesses under taskset: the advantage holds between -40% and -47% from 24
cores down to an oversubscribed 2, though packing erodes faster than waves do
because packing is what creates the concurrency.

measure_evolution_cost.py is the offline cost model, with no runtime caller. It
reports workers from the workflow's current default, which on this base is 1.

Limits worth stating: sleeping threads do not contend and the duration sample
was itself recorded at workers=1, so the speedups are upper bounds; the ordering
of the schedulers is trustworthy because they were compared under identical
conditions, the magnitudes are not.

562 eval tests pass at this base. The two test_model_gateway.py failures,
test_locked_litellm_translates_messages_to_offline_responses and
test_openai_gateway_never_leaves_proxy_output_on_an_undrained_pipe, fail
identically on origin/main in this environment.

* fix(eval): compare the shipped window and bound the overrun by it

Address PR review feedback (#3206).

run_faithful defaulted its submission window to `workers` while
runner.sweep_packed_cells defaults to `max(workers * PACKED_WINDOW_MULTIPLIER,
workers)`, so every run that named no window compared a prototype queued twice
as tightly as the shipped scheduler and presented it as the production
invariant. The faithful default now reads the same constant. Measured at
workers=3, faithful and production agreed on nothing before and agree exactly
now: breaker overrun 2/1/2 vs 2/4/3 becomes 2/4/3 vs 2/4/3 across the three
failure positions.

The contention sweep hard-coded `window=12` for faithful only, which the
production run never saw - masked at workers=6 where both are 12. Removed, and
the production measurement it was already paying for is now reported as
`production_s` instead of being discarded.

breaker_fidelity checked the overrun against `args.workers`. The bound the
producer actually enforces is `window - 1` cells past the fold pointer, which
is the wave scheduler's own `workers - 1` when window == workers; against the
shipped default of 6 the old predicate reported a failure for an in-bound run.
The window is now passed explicitly, reported in each row, and checked against
its own bound.

--window was parsed and never read. Wired into the schedulers that hold one.
Dropped two unused plan constructions CodeQL flagged, and the `skipped` set in
sweep_packed_cells that nothing reads - the None appended to `submitted` is the
skip representation the fold loop consumes.

Verification: 562 passed, 15 skipped, 2 failed (the two test_model_gateway.py
failures the PR description documents as reproducing on origin/main), ruff
clean.

* fix(eval): carry the cancellation scope into packed cells, reject the args that hang

Address PR review feedback (#3206).

sweep_packed_cells submits from a producer THREAD, and a new thread starts with
an empty context, so `copy_context()` there copied the producer's context rather
than the one cancellation_scope had just bound _CANCELLATION in. Every packed
cell therefore ran with no cancellation event, and run_managed falls back to
_CANCELLATION when none is passed - so a cancelled run's subprocesses would
never have learned about it. sweep_task_cells gets this right for free by
submitting from the thread that entered the scope. Reproduced directly: packed
workers observed [False, False], wave workers [True, True]. The caller's context
is now captured before the producer starts and copied per submission; the new
test fails without the fix.

Three CLI arguments were accepted and then wedged the run:

  --scale 0                       ZeroDivisionError before any scheduler starts
  --graph-seconds -1              hangs: the builder thread dies on a negative
                                  sleep, every scheduler waits on a readiness
                                  event nobody sets
  --window 0 (faithful)           hangs: submitted - fold_pointer >= 0 holds
                                  before the first submission, so the producer
                                  and the consumer wait on each other

The first two are rejected at the parser, which is the only layer that runs
before a thread exists. run_faithful now enforces the same window >= workers
rule sweep_packed_cells already had, so the prototype rejects exactly what the
shipped function rejects. All three were confirmed to crash or hang first.

Verification: 563 passed, 15 skipped, 2 failed (the two test_model_gateway.py
failures the PR description documents as reproducing on origin/main), ruff
clean.

* chore(autofix): apply prettier + eslint fixes via /autofix command

* Address PR review feedback (#3206)

Preserve settled sibling rows when a packed cell raises. run_cell deliberately
lets unexpected harness exceptions propagate, and sweep_task_cells answers that
by folding every non-failing sibling before it re-raises - the cells already ran
and already spent their budget, so dropping their rows means paying for evidence
the sweep then discards. sweep_packed_cells called future.result() bare, so the
fold stopped at the failing index and every later cell that had already
completed was silently lost. It now folds forward over the settled futures
before re-raising. The failing index itself has no row, since execute() assigns
only on success, so folding forward cannot duplicate it.

Pinned by a regression test that fails without the fix: the later cell is made
to finish first, so there is real settled evidence to lose at the moment cell 0
raises.

Reject arguments that cannot produce a run, at the boundary rather than deep
inside a thread. NaN defeats every comparison it appears in, so the existing
"> 0" and ">= 0" checks admitted --scale nan and --graph-seconds nan; the NaN
then reached time.sleep in a worker or the graph thread, raised there, and left
every scheduler waiting forever on a readiness event nobody would set. Infinity
was worse than a crash: it scaled all durations to zero and the run reported a
sweep that took no time. Both flags now require a finite value.

The count flags are indexed or handed straight to a thread pool, so a zero
surfaced as an IndexError on plans[0], a median over an empty sequence, or
ThreadPoolExecutor's own error - none naming the flag responsible. --workers,
--repeat and --runs now require at least 1.

Two flags were not in the review but carry the same invariant and the same
one-line treatment, so they are fixed with the class rather than left to
resurface: --runs (same empty-plan path as --repeat) and --window, where zero
admits no cell at all because the producer waits for a fold pointer to move past
a cell it was never allowed to submit.

Verified each guard fires with its own message rather than a stack trace.

563 eval tests pass. Note: pre-existing failures in test_model_gateway.py not
addressed by this PR - litellm[proxy]'s console script is absent in this
environment, and neither test touches the files changed here.

* Address PR review feedback (#3206), round 2

Stop charging the fed baseline for overlap the wave scheduler gets free.
run_fed is documented as pricing the barrier alone, but it slept graph_seconds
serially before every task, while run_wave starts one background builder that
prepares task N+1 while task N's cells run. The fed-versus-wave delta therefore
mixed the loss of that overlap into what was reported as the price of the
barrier. run_fed now uses the same builder, started before the clock, so the
barrier is the only remaining difference.

This moved the numbers. On the weekly profile fed was 4.203s and is now 3.694s,
exactly equal to wave - which is the answer that profile should give. On cold,
fed was 5.995s and is now 5.487s, so the measured price of the barrier widens
from 1.844s to 2.352s: the old arrangement understated it by about a quarter.
No committed results file or PR-body figure quotes these, so there is nothing
stale to regenerate.

Enforce the window bound the schedulers actually hold. Last round's guard
required only >= 1, but run_faithful and sweep_packed_cells both refuse a window
below the worker count, so --scheduler faithful --workers 3 --window 1 passed
validation and then died on an uncaught ValueError. The check now uses the
worker count.

It also uses the LARGEST worker count the invocation will really use.
--contention-sweep runs its own counts irrespective of --workers, so validating
against --workers alone let the three-worker measurements finish and then raised
on the six-worker one, losing the run partway through. Those counts are now a
named constant the validator can see.

Verified: --scheduler faithful --workers 3 --window 1 is rejected naming 3, and
--workers 3 --window 3 --contention-sweep is rejected naming 6.

No regression test for the graph-overlap fix. Discriminating it from the old
behaviour requires cell work to overlap graph work, which makes the assertion a
timing comparison, and this project does not take non-deterministic tests. It is
verified by the before/after measurement above instead.

564 eval tests pass. Note: pre-existing failures in test_model_gateway.py not
addressed by this PR - litellm[proxy]'s console script is absent here.

---------

Co-authored-by: Gergo Magyar <gergomagyar0@gmail.com>
Co-authored-by: github-actions[bot] <41898282+github-actions[bot]@users.noreply.github.com>
This commit is contained in:
Gergő Magyar 2026-09-08 08:22:12 +01:00 • committed by GitHub
parent 95858e7549
commit d1463977c8
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
6 changed files with 1568 additions and 0 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

122
eval/tests/test_measure_evolution_cost.py Normal file
View file

@ -0,0 +1,122 @@
"""Cost model for the evolution wall clock: measured cells, real schedules."""
from __future__ import annotations
import pytest
from workflow_bench.measure_evolution_cost import (
CANDIDATE_ARM,
SHA_OVERHEAD_SECONDS,
DURATIONS_BY_ARM,
PROPOSER_SECONDS,
REVIEW_ARMS,
expected_task_seconds,
fed_makespan,
fed_pool_enabled,
generation_seconds,
graph_pipeline_enabled,
paid_arms,
task_cells,
wave_makespan,
)
def test_every_arm_has_its_own_unsorted_sample():
assert set(DURATIONS_BY_ARM) == set(REVIEW_ARMS)
for arm, sample in DURATIONS_BY_ARM.items():
assert len(sample) >= 10, arm
# Sorting would hand each task a uniform block and hide the variance
# the whole model exists to price.
assert list(sample) != sorted(sample), arm
assert PROPOSER_SECONDS > 0
assert SHA_OVERHEAD_SECONDS > 0
def test_weekly_reuse_pays_the_candidate_arm_only():
assert paid_arms(weekly=True, reuse_enabled=True) == (CANDIDATE_ARM,)
assert paid_arms(weekly=False, reuse_enabled=True) == REVIEW_ARMS
assert paid_arms(weekly=True, reuse_enabled=False) == REVIEW_ARMS
def test_cells_are_submitted_run_major_arm_minor():
# runner.py: [(run_idx, arm) for run_idx in range(runs) for arm in arms].
# At workers=3 that puts one cell of each arm in every wave.
cells = task_cells(2, REVIEW_ARMS, 0)
assert len(cells) == 6
expected = [DURATIONS_BY_ARM[arm][run] for run in range(2) for arm in REVIEW_ARMS]
assert cells == expected
def test_overhead_is_charged_per_sha_and_outside_the_pool():
# Two properties at once: the residual sits outside the schedule, where more
# workers cannot dissolve it, and it scales with SHAs rather than cells.
assert task_cells(1, (CANDIDATE_ARM,), 0) == [DURATIONS_BY_ARM[CANDIDATE_ARM][0]]
wide = generation_seconds(
task_count=1, runs=3, arms=REVIEW_ARMS, workers=9, fed_pool=True, unique_shas=5
)
assert wide >= PROPOSER_SECONDS + 5 * SHA_OVERHEAD_SECONDS
def test_sweep_overhead_does_not_shrink_with_the_arm_count():
"""The bias that made weekly look cheaper than it is.
A seeded weekly generation pays one arm instead of three but builds exactly
the same graphs. Charging the residual per cell billed it a third of a cost
the real sweep still pays; per SHA, the two attribute the same setup.
"""
kwargs = dict(task_count=6, runs=3, workers=3, fed_pool=False, unique_shas=5)
weekly = generation_seconds(arms=(CANDIDATE_ARM,), **kwargs)
cold = generation_seconds(arms=REVIEW_ARMS, **kwargs)
weekly_sessions = 6 * expected_task_seconds(3, (CANDIDATE_ARM,), 3, fed_pool=False)
cold_sessions = 6 * expected_task_seconds(3, REVIEW_ARMS, 3, fed_pool=False)
# Whatever each wall is, the non-session part is identical.
assert round(weekly - weekly_sessions) == round(cold - cold_sessions)
# Cycling wraps, so a task can ask for more runs than the sample holds.
long_sample = task_cells(len(DURATIONS_BY_ARM[CANDIDATE_ARM]) + 2, (CANDIDATE_ARM,), 0)
assert len(long_sample) == len(DURATIONS_BY_ARM[CANDIDATE_ARM]) + 2
def test_a_wave_costs_its_slowest_cell_and_a_fed_pool_does_not():
slow = [10.0, 1.0, 1.0, 10.0, 1.0, 1.0]
assert wave_makespan(slow, 3) == 20.0
# Fed: one worker takes the first 10; the second 10 lands on a worker that
# has already cleared a 1, and the remaining 1s fill the third.
assert fed_makespan(slow, 3) == 11.0
assert fed_makespan(slow, 1) == wave_makespan(slow, 1) == 24.0
def test_expected_task_seconds_is_alignment_averaged_and_deterministic():
waved = expected_task_seconds(3, REVIEW_ARMS, 3, fed_pool=False)
assert waved == expected_task_seconds(3, REVIEW_ARMS, 3, fed_pool=False)
assert expected_task_seconds(0, REVIEW_ARMS, 3, fed_pool=False) == 0.0
assert expected_task_seconds(3, (), 3, fed_pool=False) == 0.0
# The barrier can only cost time, never save it.
assert waved >= expected_task_seconds(3, REVIEW_ARMS, 3, fed_pool=True)
def test_a_generation_pays_one_proposer_session_on_top_of_its_tasks():
one = generation_seconds(
task_count=1, runs=3, arms=REVIEW_ARMS, workers=3, fed_pool=False, unique_shas=1
)
two = generation_seconds(
task_count=2, runs=3, arms=REVIEW_ARMS, workers=3, fed_pool=False, unique_shas=1
)
# Each extra task adds exactly one task's makespan. The proposer and the
# per-SHA sweep overhead are both paid once, not per task.
assert two - one == pytest.approx(
one - PROPOSER_SECONDS - SHA_OVERHEAD_SECONDS, abs=2.0
)
def test_feature_flags_read_the_runner_not_the_wish():
assert graph_pipeline_enabled("def _run_sweep(): pass") == 0
assert graph_pipeline_enabled("graph_prefetch = GraphPrefetch(...)") == 1
assert fed_pool_enabled("def _run_wave(): pass") == 0
assert fed_pool_enabled("def _run_fed_pool(): pass") == 1
@pytest.mark.parametrize("workers", [1, 3, 8])
def test_more_workers_never_lengthen_a_task(workers):
serial = expected_task_seconds(3, REVIEW_ARMS, 1, fed_pool=True)
assert expected_task_seconds(3, REVIEW_ARMS, workers, fed_pool=True) <= serial

157
eval/tests/test_workflow_bench.py
View file

@ -5,11 +5,16 @@ import os
import re
import shlex
import subprocess
import threading
from pathlib import Path
import pytest
import yaml
from typing import Any
from workflow_bench import runner
from workflow_bench.process_control import _CANCELLATION, cancellation_scope
from workflow_bench.runner import (
aggregate,
broken_incumbent_arms,
@ -512,3 +517,155 @@ def test_run_evolution_script_is_the_shared_ci_and_local_entrypoint():
assert "--include-expensive" in argv
assert "claude-sonnet-5" not in argv
assert printed.stderr # rewrite notice goes to stderr
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_workers_inherit_the_runs_cancellation_event():
"""A worker that cannot see the event runs on after the sweep is cancelled.
The cells are submitted from a producer THREAD, and a new thread starts with
an empty context - so copying the context at submission copies the wrong one
unless the caller's is captured first. run_managed falls back to
_CANCELLATION when no event is passed, which is how a cell's subprocesses
learn the run was cancelled at all.
"""
seen: list[threading.Event | None] = []
event = threading.Event()
with cancellation_scope(event):
runner.sweep_packed_cells(
_packed_cells(2, 1, ("review",)),
workers=2,
run=lambda *_: seen.append(_CANCELLATION.get()) or {"error_kind": None},
on_start=lambda *_: None,
on_record=lambda *_: None,
outage_streak=0,
outage_limit=0,
)
assert seen and all(observed is event for observed in seen)
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,
)
def test_a_raising_packed_cell_still_persists_its_settled_siblings():
"""A crash in one cell must not erase the evidence of cells that finished.
run_cell deliberately lets unexpected harness exceptions propagate, and the
wave scheduler answers that by folding every non-failing sibling before it
re-raises. The packed scheduler has to hold the same contract: the later
cells already ran and already cost money, so losing their rows would mean
paying for evidence the sweep then throws away.
"""
folded: list[tuple[int, str]] = []
started = threading.Event()
def run(task_id: str, run_idx: int, arm: str) -> dict[str, Any]:
if run_idx == 0:
# Let the later cell finish first, so there is settled evidence to
# lose at the moment this one raises.
started.wait(timeout=5)
raise RuntimeError("harness bug in cell 0")
started.set()
return {"error_kind": None}
with pytest.raises(RuntimeError, match="harness bug in cell 0"):
runner.sweep_packed_cells(
_packed_cells(1, 2, ("review",)),
workers=2,
run=run,
on_start=lambda *_: None,
on_record=lambda task_id, run_idx, arm, _rec: folded.append((run_idx, arm)),
outage_streak=0,
outage_limit=0,
)
assert (1, "review") in folded, "the sibling that completed was never recorded"

311
eval/workflow_bench/measure_evolution_cost.py Normal file
View file

@ -0,0 +1,311 @@
#!/usr/bin/env python3
"""Cheap cost model for the skill-evolution review generation.
This is the ce-optimize measurement harness. It does not start Claude and it
does not replay a run. It reads the review corpus, the evolve defaults and the
workflow's workers default, then schedules the measured cell durations in
``session_durations.json`` the way ``sweep_task_cells`` schedules real cells.
Everything priced here is measured. Cell durations and the proposer session
come from a real artifact, and the work outside the agent sessions comes from
that run's own step wall minus the time its sessions and proposer account for.
Weekly assumes a matching seed, so every reusable comparator cell is skipped
and only the candidate arm is paid. Cold assumes an empty seed.
"""
from __future__ import annotations
import json
import math
import re
import statistics as st
import subprocess
import sys
from pathlib import Path
REPO_ROOT = Path(__file__).resolve().parents[2]
EVAL_ROOT = REPO_ROOT / "eval"
REVIEW_TASKS = EVAL_ROOT / "workflow_bench" / "tasks.review.scenarios.yaml"
EVOLVE_PY = EVAL_ROOT / "workflow_bench" / "evolve.py"
RUNNER_PY = EVAL_ROOT / "workflow_bench" / "runner.py"
ARTIFACTS_PY = EVAL_ROOT / "workflow_bench" / "runner_artifacts.py"
REUSE_PY = EVAL_ROOT / "workflow_bench" / "comparator_reuse.py"
WORKFLOW = REPO_ROOT / ".github" / "workflows" / "gitnexus-skill-evolution.yml"
MEASURED = json.loads(
(Path(__file__).resolve().parent / "session_durations.json").read_text(encoding="utf-8")
)
# Per arm, because the arms are not interchangeable and the weekly lane pays
# only the candidate one. Cells are submitted run-major and arm-minor
# (runner.py ``planned``), so at workers=3 every wave holds one cell of each
# arm and the slowest arm sets the wave.
DURATIONS_BY_ARM: dict[str, tuple[float, ...]] = {
arm: tuple(values) for arm, values in MEASURED["cell_duration_s_by_arm"].items()
}
PROPOSER_SECONDS: float = MEASURED["proposer_duration_s"]
_RESIDUAL = MEASURED["residual"]
# Clone, graph build, sandbox, teardown: the sweep's own time, taken as that
# run's step wall minus what its sessions and proposer account for. Charged
# SERIALLY, outside the pool, and charged PER SHA rather than per cell. The
# residual mixes per-cell work with per-SHA graph setup and the artifact cannot
# separate them; per-SHA is the direction that refuses to credit a run for
# shrinking work it still performs, which per-cell did - a weekly generation
# pays one arm instead of three but builds exactly the same graphs. See
# session_durations.json residual._split_assumption.
SHA_OVERHEAD_SECONDS: float = _RESIDUAL["sha_overhead_s"]
# runner.py CANDIDATE_ARMS derives the candidate arm from its incumbent, and
# only an incumbent row can be reused from a prior generation.
CANDIDATE_ARM = "candidate_review"
REVIEW_ARMS = ("ce_review", "review", CANDIDATE_ARM)
SUITE_FILES = (
"tests/test_measure_evolution_cost.py",
"tests/test_comparator_reuse.py",
"tests/test_evolve.py",
"tests/test_sanitized_graph.py",
"tests/test_workflow_bench.py",
"tests/test_workflow_bench_sessions.py",
"tests/test_session_progress.py",
)
def _read(path: Path) -> str:
return path.read_text(encoding="utf-8")
def review_tasks(text: str) -> list[dict[str, str]]:
tasks: list[dict[str, str]] = []
current: dict[str, str] | None = None
for raw in text.splitlines():
line = raw.strip()
if line.startswith("id:"):
if current is not None:
tasks.append(current)
current = {"id": line.split(":", 1)[1].strip()}
elif line.startswith("ref:") and current is not None:
current["ref"] = line.split(":", 1)[1].strip()
if current is not None:
tasks.append(current)
return tasks
def evolve_default(name: str, text: str) -> int:
match = re.search(rf'add_argument\("--{re.escape(name)}".*?default=(\d+)', text, flags=re.S)
if match is None:
raise ValueError(f"evolve.py is missing --{name} default")
return int(match.group(1))
def workflow_dispatch_workers(text: str) -> int:
match = re.search(r"^\s+workers:\n(?:.*\n)*?^\s+default: '(\d+)'", text, flags=re.M)
if match is None:
raise ValueError("workflow_dispatch workers default is missing")
return int(match.group(1))
def feature_enabled() -> tuple[int, int]:
evolve = _read(EVOLVE_PY)
runner = _read(RUNNER_PY)
artifacts = _read(ARTIFACTS_PY)
reuse = int(
REUSE_PY.is_file()
and "--reuse-results" in evolve
and "select_reusable_comparator_rows" in runner
and "CANDIDATE" in _read(REUSE_PY)
)
templates = int("def copy_isolated_tree" in artifacts and "clone_templates" in runner)
return reuse, templates
def graph_pipeline_enabled(runner_text: str) -> int:
"""True when the runner prefetches the next SHA during paid sessions."""
return int("prefetch_next_graph" in runner_text or "GraphPrefetch" in runner_text)
def fed_pool_enabled(runner_text: str) -> int:
"""True when the sweep feeds a live pool instead of waiting on waves."""
return int("def _run_fed_pool" in runner_text)
def paid_arms(weekly: bool, reuse_enabled: bool) -> tuple[str, ...]:
"""Arms a generation actually pays for."""
if weekly and reuse_enabled:
return (CANDIDATE_ARM,)
return REVIEW_ARMS
def task_cells(runs: int, arms: tuple[str, ...], offset: int) -> list[float]:
"""One task's cell durations in submission order: run-major, arm-minor.
Each arm draws from its own measured sample, cycled from ``offset`` so the
caller can average over every alignment instead of trusting one.
"""
cells: list[float] = []
for run_idx in range(runs):
for arm in arms:
sample = DURATIONS_BY_ARM[arm]
cells.append(sample[(offset + run_idx) % len(sample)])
return cells
def wave_makespan(durations: list[float], workers: int) -> float:
"""Today's scheduler: fixed waves of ``workers``, with a barrier between."""
return sum(
max(durations[start : start + workers]) for start in range(0, len(durations), workers)
)
def fed_makespan(durations: list[float], workers: int) -> float:
"""Continuously fed pool: a free worker takes the next cell immediately."""
busy_until = [0.0] * workers
for duration in durations:
first = min(range(workers), key=busy_until.__getitem__)
busy_until[first] += duration
return max(busy_until)
def expected_task_seconds(
runs: int, arms: tuple[str, ...], workers: int, *, fed_pool: bool
) -> float:
"""Mean makespan of one task over every alignment of the measured samples.
One fixed alignment would let an accident of the source run - its slowest
cells happen to come first - decide the answer. Averaging keeps the real
multiset and the real ordering effects without that artifact, and stays
deterministic.
"""
if runs < 1 or not arms:
return 0.0
makespan = fed_makespan if fed_pool else wave_makespan
# lcm, not max: with samples of 13 and 14, max would wrap the shorter one
# and count its first entry twice.
alignments = math.lcm(*(len(DURATIONS_BY_ARM[arm]) for arm in arms))
return (
sum(makespan(task_cells(runs, arms, offset), workers) for offset in range(alignments))
/ alignments
)
def generation_seconds(
*,
task_count: int,
runs: int,
arms: tuple[str, ...],
workers: int,
fed_pool: bool,
unique_shas: int,
) -> int:
"""Whole generation: proposer, then the tasks back to back, plus overhead.
Prices a HEALTHY sweep. A run whose cells return unusable evidence does not
reach this wall at all: the outage breaker aborts after
``DEFAULT_OUTAGE_STREAK`` consecutive systemic failures, which for the
sample's own error sequence is cell 5 of 41.
Sweep overhead is charged per SHA, so it does not shrink with the arm count.
Weekly pays one arm instead of three but builds the same graphs, and billing
that per cell credited it for a saving the real run never makes.
"""
return round(
PROPOSER_SECONDS
+ task_count * expected_task_seconds(runs, arms, workers, fed_pool=fed_pool)
+ unique_shas * SHA_OVERHEAD_SECONDS
)
def _pytest_python() -> list[str]:
venv_python = EVAL_ROOT / ".venv" / "bin" / "python"
if venv_python.is_file():
return [str(venv_python)]
if (EVAL_ROOT / "uv.lock").is_file():
return ["uv", "run", "--locked", "--extra", "dev", "python"]
return [sys.executable]
def suite_passed() -> int:
files = [name for name in SUITE_FILES if (EVAL_ROOT / name).is_file()]
if not files:
return 0
cmd = [*_pytest_python(), "-m", "pytest", *files, "-q", "--tb=no", "--no-header"]
try:
completed = subprocess.run(
cmd, cwd=EVAL_ROOT, check=False, capture_output=True, text=True, timeout=240
)
except (OSError, subprocess.TimeoutExpired):
return 0
return int(completed.returncode == 0)
def main() -> int:
tasks = review_tasks(_read(REVIEW_TASKS))
evolve = _read(EVOLVE_PY)
runner = _read(RUNNER_PY)
runs = evolve_default("runs", evolve)
workers = workflow_dispatch_workers(_read(WORKFLOW))
reuse_enabled, clone_templates_enabled = feature_enabled()
fed_pool = fed_pool_enabled(runner)
# Both walls build the same graphs; the arm count does not change that.
unique_shas = len({t.get("ref", "") for t in tasks if t.get("ref")})
payload: dict[str, object] = {}
for label, weekly in (("weekly", True), ("cold", False)):
arms = paid_arms(weekly, bool(reuse_enabled))
payload[f"estimated_{label}_wall_seconds"] = generation_seconds(
task_count=len(tasks),
runs=runs,
arms=arms,
workers=workers,
fed_pool=bool(fed_pool),
unique_shas=unique_shas,
)
payload[f"paid_{label}_cells"] = len(tasks) * runs * len(arms)
# What the wave barrier costs: the same cells, continuously fed.
payload[f"fed_pool_{label}_wall_seconds"] = generation_seconds(
task_count=len(tasks),
runs=runs,
arms=arms,
workers=workers,
fed_pool=True,
unique_shas=unique_shas,
)
all_durations = [d for sample in DURATIONS_BY_ARM.values() for d in sample]
payload.update(
{
"suite_passed": suite_passed(),
"promotion_min_runs": evolve_default("promotion-min-runs", evolve),
"review_task_count": len(tasks),
"candidate_cells": len(tasks) * runs,
"workers": workers,
"unique_task_shas": len({t.get("ref", "") for t in tasks if t.get("ref")}),
"reuse_enabled": reuse_enabled,
"clone_templates_enabled": clone_templates_enabled,
"graph_pipeline_enabled": graph_pipeline_enabled(runner),
"fed_pool_enabled": fed_pool,
"measured_cell_count": len(all_durations),
"median_cell_seconds": round(st.median(all_durations)),
"mean_cell_seconds": round(st.mean(all_durations)),
"max_cell_seconds": round(max(all_durations)),
"median_candidate_cell_seconds": round(st.median(DURATIONS_BY_ARM[CANDIDATE_ARM])),
"mean_candidate_cell_seconds": round(st.mean(DURATIONS_BY_ARM[CANDIDATE_ARM])),
"proposer_seconds": round(PROPOSER_SECONDS),
"sha_overhead_seconds": round(SHA_OVERHEAD_SECONDS, 1),
}
)
json.dump(payload, sys.stdout, sort_keys=True)
sys.stdout.write("\n")
return 0
if __name__ == "__main__":
raise SystemExit(main())

198
eval/workflow_bench/runner.py
View file

@ -851,6 +851,204 @@ def sweep_task_cells(
return outage_streak, False
# A sustained upstream outage shows up as a run of session/infra/cleanup
# 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
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)
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)
# cancellation_scope binds _CANCELLATION in the CALLING thread's
# context, and a new thread starts with an empty one - so the producer
# has to copy this context rather than its own, or every cell it
# submits loses the run's cancellation event. sweep_task_cells gets
# this for free by submitting from the thread that entered the scope.
caller_context = copy_context()
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:
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
worker_context = caller_context.run(copy_context)
submitted.append(pool.submit(worker_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:
try:
future.result()
except BaseException:
# Same contract as sweep_task_cells: the cells submitted
# after this one have already run and spent their budget,
# so persist their rows in submission order before the
# harness bug takes the process down. Without this, one
# crashing cell silently erases the paid evidence of
# every sibling that had already finished. The failing
# index itself has no row - execute() only assigns on
# success - so folding forward cannot duplicate it.
with gate:
settled = list(submitted)
for later in range(index + 1, len(settled)):
pending = settled[later]
if pending is not None and not pending.done():
continue
row = results[later]
if row is not None:
on_record(*cells[later], row)
raise
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.

33
eval/workflow_bench/session_durations.json Normal file
View file

@ -0,0 +1,33 @@
{
"_provenance": "Actions run 33912693948 (2026-09-04), review profile, gen-0, workers=1. Artifact gitnexus-evolution-33912693948-1: gen-0/bench/results.jsonl and gen-0/proposer-session.json. Step wall from the Actions API.",
"_caveat": "Every cell in that run returned unusable evidence (32 review-evidence-invalid, 6 session-error, 3 skill-not-invoked); two hit the 5400s ceiling and it cost 653. Durations are real, but a run that resolves cleanly may sit lower. It is the only live artifact - the 2026-07-22 green run's has expired.",
"_order": "Submission order, deliberately unsorted: the model cycles these, so sorting would hand each task a uniform block and hide the variance being measured.",
"_duration_scope": "duration_s is the sum of the cell's Claude session durations (runner_sessions.py). It excludes the clone, graph materialize, asset staging, sandbox setup and teardown - those live in the residual below.",
"session_ceiling_s": 5400,
"proposer_duration_s": 344.7,
"cell_duration_s_by_arm": {
"candidate_review": [
2485.6, 1338.0, 3075.2, 702.5, 1240.4, 5400.0, 762.1, 342.9, 826.3, 489.7, 675.4, 337.8, 734.3
],
"ce_review": [
3744.6, 2140.4, 1418.4, 436.1, 653.3, 1022.3, 851.2, 1191.1, 902.1, 847.2, 502.6, 991.3,
1222.7, 544.0
],
"review": [
5400.0, 2976.4, 1162.8, 627.1, 901.3, 436.9, 704.3, 963.4, 631.8, 627.9, 663.5, 662.4, 361.3,
741.1
]
},
"residual": {
"benchmark_step_wall_s": 54623,
"session_seconds": 51737.7,
"proposer_seconds": 344.7,
"unaccounted_s": 2540.6,
"cells": 41,
"unique_shas": 5,
"_note": "Everything the sweep spent outside the agent sessions: per-SHA sanitize and `analyze --pdg --index-only`, plus each cell's clone, materialize, staging, sandbox and teardown. That run predates clone templates and graph prefetch, so this is an upper bound for the current code. The split between per-SHA and per-cell is not recoverable from the artifact, so the model charges it per cell and serially, outside the pool - the pessimistic reading of an already-small term.",
"_split_assumption": "The residual mixes per-SHA graph setup with per-cell clone/sandbox/teardown and the artifact cannot separate them. The model charges it per SHA, not per cell, because only that direction refuses to credit a run for shrinking work it still performs: a weekly generation pays one arm instead of three but builds the same graphs. This overstates cold slightly and refuses to understate weekly. Replace with measured per-SHA and per-cell times when a run records them separately.",
"sha_overhead_s": 508.1
},
"_breaker": "Replaying this sample's error_kind sequence through today's systemic_outage_streak trips the outage breaker at cell 5 of 41 (DEFAULT_OUTAGE_STREAK=5). The source run executed all 41, so its runner did not break on this sequence. The durations stay valid as per-cell timings; what they cannot describe is a 54-cell sweep with this failure profile, because the current code would never run one."
}

747
eval/workflow_bench/simulate_sweep.py Normal file
View file

@ -0,0 +1,747 @@
#!/usr/bin/env python3
"""Run the real sweep scheduler against stub sessions and time it.
``measure_evolution_cost`` is arithmetic: it predicts wall clock from a model of
what ``sweep_task_cells`` does. This runs the actual function - real threads,
the real wave barrier, the real outage breaker - and replaces only the paid
agent session with a sleep. If the two disagree, the model is wrong.
Durations are the measured per-arm samples from ``session_durations.json``
divided by ``--scale``, so a cell that really took 1416s takes ~0.28s here. The
shape is preserved deliberately: the median cell is 826s against a 5400s
ceiling, and that spread is the whole reason a barrier costs anything. Uniform
random sleeps would erase the effect under test.
Schedulers, all consuming one identical seeded plan:
``wave`` the shipped ``sweep_task_cells`` - fixed waves of ``workers``, a
barrier between them, one task at a time.
``fed`` a continuously fed pool per task (H1). Naive: no breaker, no graph
gating. Present to price the barrier alone.
``packed`` one pool across every task (H2). Naive, same caveat.
``faithful``H2 carrying the invariants the shipped scheduler actually holds:
a global submission order, in-order folding, the outage breaker, and
per-task graph readiness gating. This is the one to believe.
python3 -m workflow_bench.simulate_sweep --compare --repeat 5
python3 -m workflow_bench.simulate_sweep --breaker-fidelity
"""
from __future__ import annotations
import argparse
import json
import math
import random
import statistics
import subprocess
import sys
import threading
import time
from concurrent.futures import ThreadPoolExecutor
from dataclasses import dataclass, field
from typing import Any
from . import runner
from .measure_evolution_cost import (
CANDIDATE_ARM,
DURATIONS_BY_ARM,
REVIEW_ARMS,
REVIEW_TASKS,
SHA_OVERHEAD_SECONDS,
_read,
expected_task_seconds,
review_tasks,
)
DEFAULT_SCALE = 5000.0
# --contention-sweep measures both of these regardless of --workers, so the
# window has to be valid for the LARGEST of them, not for the parsed value.
CONTENTION_WORKERS = (3, 6)
SYSTEMIC_KIND = "session-error"
# A cell is mostly a model session waiting on the network, but its tool calls -
# git, vitest, analyze - burn real CPU in real subprocesses. Sleeping threads
# model the wait and nothing else, so every speedup measured that way is an
# upper bound. This burns WORK, not wall clock: a fixed number of sha256 rounds
# in a subprocess, which takes longer when cores are contended. That is the
# effect under test, and it has to be a subprocess - Python threads burning
# Python would measure the GIL rather than the machine.
_BURN_SRC = (
"import hashlib,sys\n"
"n=int(sys.argv[1]); b=b'x'*4096; h=hashlib.sha256()\n"
"for _ in range(n): h.update(b)\n"
"sys.stdout.write(h.hexdigest()[:8])\n"
)
def calibrate_burn(probe_rounds: int = 400_000) -> float:
"""sha256 rounds per second, one uncontended subprocess. Measured, not assumed."""
started = time.monotonic()
subprocess.run(
[sys.executable, "-c", _BURN_SRC, str(probe_rounds)],
check=True,
capture_output=True,
)
return probe_rounds / (time.monotonic() - started)
def _execute_cell(cell: Cell, cpu_fraction: float, burn_rate: float) -> None:
"""The stub session: wait for the API, then do the tool-call work."""
if cpu_fraction <= 0:
time.sleep(cell.seconds)
return
time.sleep(cell.seconds * (1.0 - cpu_fraction))
rounds = int(cell.seconds * cpu_fraction * burn_rate)
if rounds > 0:
subprocess.run(
[sys.executable, "-c", _BURN_SRC, str(rounds)], check=True, capture_output=True
)
@dataclass(frozen=True)
class Cell:
task: int
run: int
arm: str
seconds: float
systemic: bool = False
@dataclass
class Outcome:
wall_s: float
executed: int
tripped_at: int | None = None
folded: list[int] = field(default_factory=list)
def build_plan(
*,
task_count: int,
runs: int,
arms: tuple[str, ...],
scale: float,
seed: int,
fail_from: int | None = None,
) -> list[list[Cell]]:
"""Per-task cells in submission order, with durations drawn once.
Shared by every scheduler so a comparison cannot be an artifact of one of
them drawing luckier cells. ``fail_from`` marks every cell at or after that
global index systemic, which is what the breaker-fidelity mode needs.
"""
rng = random.Random(seed)
plan: list[list[Cell]] = []
index = 0
for task in range(task_count):
cells: list[Cell] = []
for run_idx in range(runs):
for arm in arms:
sample = DURATIONS_BY_ARM[arm]
cells.append(
Cell(
task=task,
run=run_idx,
arm=arm,
seconds=sample[rng.randrange(len(sample))] / scale,
systemic=fail_from is not None and index >= fail_from,
)
)
index += 1
plan.append(cells)
return plan
def _flatten(plan: list[list[Cell]]) -> list[Cell]:
return [cell for cells in plan for cell in cells]
def _record(cell: Cell) -> dict[str, Any]:
kind = SYSTEMIC_KIND if cell.systemic else None
return {
"run": cell.run,
"arm": cell.arm,
"ok": not cell.systemic,
"resolved": not cell.systemic,
"error_kind": kind,
"review_evidence_valid": not cell.systemic,
}
def _graph_builder(
ready: list[threading.Event], graph_seconds: float, stop: threading.Event
) -> threading.Thread:
"""One graph at a time, in task order - they are CPU and IO heavy."""
def build() -> None:
for event in ready:
if stop.is_set():
return
time.sleep(graph_seconds)
event.set()
thread = threading.Thread(target=build, name="graph-builder", daemon=True)
thread.start()
return thread
def run_wave(
plan: list[list[Cell]],
workers: int,
*,
outage_limit: int,
graph_seconds: float,
cpu_fraction: float = 0.0,
burn_rate: float = 0.0,
) -> Outcome:
"""The shipped scheduler, driven for real, task after task."""
ready = [threading.Event() for _ in plan]
stop = threading.Event()
_graph_builder(ready, graph_seconds, stop)
executed = 0
lock = threading.Lock()
streak = 0
tripped_at: int | None = None
folded: list[int] = []
base = 0
started = time.monotonic()
for task, cells in enumerate(plan):
ready[task].wait()
by_key = {(c.run, c.arm): c for c in cells}
def fake_run(run_idx: int, arm: str) -> dict[str, Any]:
nonlocal executed
cell = by_key[(run_idx, arm)]
_execute_cell(cell, cpu_fraction, burn_rate)
with lock:
executed += 1
return _record(cell)
order = {(c.run, c.arm): base + i for i, c in enumerate(cells)}
def on_record(run_idx: int, arm: str, rec: dict[str, Any]) -> None:
# Mirror the breaker's own evaluation so the reported trip point is
# the cell that crossed the limit, not merely the last one folded -
# sweep_task_cells folds a whole wave before it evaluates.
nonlocal streak, tripped_at
index = order[(run_idx, arm)]
folded.append(index)
streak = runner.systemic_outage_streak(rec["error_kind"], streak)
if outage_limit and streak >= outage_limit and tripped_at is None:
tripped_at = index
streak, tripped = runner.sweep_task_cells(
[(c.run, c.arm) for c in cells],
workers=workers,
run=fake_run,
on_start=lambda *_: None,
on_record=on_record,
outage_streak=streak,
outage_limit=outage_limit,
)
base += len(cells)
if tripped:
break
stop.set()
return Outcome(wall_s=time.monotonic() - started, executed=executed, tripped_at=tripped_at, folded=folded)
def _drain_naive(cells: list[Cell], workers: int) -> int:
with ThreadPoolExecutor(max_workers=workers) as pool:
list(pool.map(lambda c: time.sleep(c.seconds), cells))
return len(cells)
def run_fed(plan: list[list[Cell]], workers: int, *, outage_limit: int, graph_seconds: float) -> Outcome:
"""H1 without invariants: fed pool per task. Prices the barrier alone.
Graph building is deliberately identical to ``run_wave`` - the same
background builder, started before the clock - because that is what makes
the claim in the first line true. Sleeping ``graph_seconds`` serially before
each task instead, as this did, charged fed for overlap that wave gets for
free: the wave builder prepares task N+1 while task N's cells run. The
fed-versus-wave delta then mixed the loss of that overlap into what was
reported as the price of the barrier.
"""
ready = [threading.Event() for _ in plan]
stop = threading.Event()
_graph_builder(ready, graph_seconds, stop)
executed = 0
started = time.monotonic()
for task, cells in enumerate(plan):
ready[task].wait()
executed += _drain_naive(cells, workers)
return Outcome(wall_s=time.monotonic() - started, executed=executed)
def run_packed(plan: list[list[Cell]], workers: int, *, outage_limit: int, graph_seconds: float) -> Outcome:
"""H2 without invariants. Upper bound, not a design."""
started = time.monotonic()
time.sleep(graph_seconds)
executed = _drain_naive(_flatten(plan), workers)
return Outcome(wall_s=time.monotonic() - started, executed=executed)
def run_faithful(
plan: list[list[Cell]],
workers: int,
*,
outage_limit: int,
graph_seconds: float,
window: int | None = None,
cpu_fraction: float = 0.0,
burn_rate: float = 0.0,
) -> Outcome:
"""H2 carrying the invariants the shipped scheduler holds.
Global submission order is task-major, run-major, arm-minor - the same total
order the wave scheduler folds in, just continued across task boundaries. A
folder walks results in exactly that order, so "consecutive systemic
failures" keeps its meaning; the breaker trips on the same logical cell it
would have in waves. Cells already in flight when it trips are the overrun,
bounded by ``workers - 1`` exactly as the wave docstring promises.
A task's cells are not submitted until its graph is ready, which is what
makes this a schedule rather than a wish: the graph builder is serial, so
packing cannot outrun it.
``window`` is the design question. Queue every cell at once and workers race
far ahead of the fold pointer, so a breaker trip has already paid for cells
nobody has looked at - measured at 5 against a bound of 2. Holding
submission to ``window`` cells beyond the fold point caps the overrun at
``window - 1``, which is the wave's own ``workers - 1`` bound when the two
are equal, while still packing across task boundaries. Defaults to whatever
``runner.sweep_packed_cells`` defaults to, so a run that names no window
compares the shipped policy rather than a more tightly queued prototype.
"""
if window is None:
window = max(workers * runner.PACKED_WINDOW_MULTIPLIER, workers)
if window < workers:
# Same rule sweep_packed_cells enforces. Without it a window below 1
# never lets the producer past its own gate and the run hangs.
raise ValueError("window must be at least workers, or the pool starves")
cells = _flatten(plan)
ready = [threading.Event() for _ in plan]
stop = threading.Event()
_graph_builder(ready, graph_seconds, stop)
results: list[dict[str, Any] | None] = [None] * len(cells)
executed = 0
lock = threading.Lock()
halt = threading.Event()
def work(index: int) -> None:
nonlocal executed
if halt.is_set():
return
cell = cells[index]
_execute_cell(cell, cpu_fraction, burn_rate)
with lock:
executed += 1
results[index] = _record(cell)
gate = threading.Condition()
fold_pointer = 0
futures: list[Any] = []
producer_done = threading.Event()
started = time.monotonic()
pool = ThreadPoolExecutor(max_workers=workers)
def produce() -> None:
submitted = 0
for task, task_cells in enumerate(plan):
ready[task].wait()
for _ in task_cells:
with gate:
while submitted - fold_pointer >= window and not halt.is_set():
gate.wait(timeout=0.5)
if halt.is_set():
producer_done.set()
return
futures.append(pool.submit(work, submitted))
submitted += 1
gate.notify_all()
producer_done.set()
producer = threading.Thread(target=produce, name="cell-producer", daemon=True)
producer.start()
streak = 0
tripped_at: int | None = None
folded: list[int] = []
try:
index = 0
while True:
with gate:
while index >= len(futures) and not producer_done.is_set():
gate.wait(timeout=0.5)
if index >= len(futures):
break
future = futures[index]
future.result()
record = results[index]
if record is not None:
folded.append(index)
streak = runner.systemic_outage_streak(record["error_kind"], streak)
if outage_limit and streak >= outage_limit:
tripped_at = index
halt.set()
with gate:
gate.notify_all()
for pending in futures[index + 1 :]:
pending.cancel()
break
index += 1
with gate:
fold_pointer = index
gate.notify_all()
finally:
halt.set()
with gate:
gate.notify_all()
stop.set()
producer.join(timeout=5)
pool.shutdown(wait=True)
return Outcome(
wall_s=time.monotonic() - started, executed=executed, tripped_at=tripped_at, folded=folded
)
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 _window_kwargs(name: str, window: int | None) -> dict[str, int]:
"""``--window`` only means anything to the two schedulers that hold one."""
return {"window": window} if window is not None and name in ("faithful", "production") else {}
def _plan_args(args: argparse.Namespace, weekly: bool, seed: int, fail_from: int | None = None):
arms = (CANDIDATE_ARM,) if weekly else REVIEW_ARMS
return {
"task_count": len(review_tasks(_read(REVIEW_TASKS))),
"runs": args.runs,
"arms": arms,
"scale": args.scale,
"seed": seed,
"fail_from": fail_from,
}, arms
def breaker_fidelity(args: argparse.Namespace) -> list[dict[str, Any]]:
"""Does packing still trip where waves trip, and overrun no further?"""
rows: list[dict[str, Any]] = []
limit = runner.DEFAULT_OUTAGE_STREAK
window = args.window if args.window is not None else max(
args.workers * runner.PACKED_WINDOW_MULTIPLIER, args.workers
)
for fail_from in (0, 4, 12):
kwargs, _arms = _plan_args(args, weekly=False, seed=args.seed, fail_from=fail_from)
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, "window": window
}
for name in ("wave", "faithful", "production"):
out = SCHEDULERS[name](
plan, args.workers, outage_limit=limit, graph_seconds=args.graph_seconds,
**_window_kwargs(name, window),
)
row[name] = {
"tripped_at": out.tripped_at,
"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["production"]["tripped_at"]
)
# The producer holds submission to ``window`` cells beyond the fold
# pointer, so at most ``window - 1`` cells past the tripping one can
# already be in flight. At ``window == workers`` that is exactly the
# wave scheduler's own ``workers - 1`` bound.
row["overrun_within_bound"] = (
row["production"]["overrun"] is not None
and row["production"]["overrun"] <= window - 1
)
rows.append(row)
return rows
def main() -> int:
parser = argparse.ArgumentParser(description=__doc__)
parser.add_argument("--workers", type=int, default=3)
parser.add_argument("--scale", type=float, default=DEFAULT_SCALE)
parser.add_argument("--seed", type=int, default=1729)
parser.add_argument("--repeat", type=int, default=1)
parser.add_argument("--runs", type=int, default=3)
parser.add_argument("--scheduler", choices=sorted(SCHEDULERS), default="wave")
parser.add_argument("--compare", action="store_true")
parser.add_argument("--breaker-fidelity", action="store_true")
parser.add_argument("--window-sweep", action="store_true", help="wall clock vs breaker overrun")
parser.add_argument("--contention-sweep", action="store_true", help="does the gain survive real CPU?")
parser.add_argument(
"--window",
type=int,
default=None,
help="submission window for the packed schedulers; defaults to the shipped policy",
)
parser.add_argument(
"--graph-seconds",
type=float,
default=None,
help="per-task graph build; defaults to the measured per-SHA overhead, scaled",
)
args = parser.parse_args()
# Both are checked here rather than where they are used: a bad --scale
# divides by zero before anything runs, and a negative --graph-seconds
# kills the graph-builder thread, after which every scheduler waits on a
# readiness event nobody will ever set.
# NaN defeats every comparison it appears in, so "> 0" and ">= 0" both admit
# it and the failure surfaces far from the flag: NaN durations reach
# time.sleep in a worker or the graph thread and raise there, after which the
# schedulers wait forever on a readiness event nobody will set. Infinity is
# worse than a crash - it silently scales every duration to zero and the run
# reports a sweep that took no time.
if not math.isfinite(args.scale) or args.scale <= 0:
parser.error("--scale must be a finite positive number")
if args.graph_seconds is not None and (not math.isfinite(args.graph_seconds) or args.graph_seconds < 0):
parser.error("--graph-seconds must be a finite non-negative number")
# Counts are indexed or handed to a thread pool without further checking, so
# a zero turns into an IndexError on plans[0], a median over an empty
# sequence, or ThreadPoolExecutor's own error - none of which name the flag
# that caused them.
if args.workers < 1:
parser.error("--workers must be at least 1")
if args.repeat < 1:
parser.error("--repeat must be at least 1")
if args.runs < 1:
parser.error("--runs must be at least 1")
# run_faithful and sweep_packed_cells both refuse a window below the worker
# count - a smaller one starves the pool, because the producer waits for a
# fold pointer to pass a cell it was never allowed to submit. Enforcing it
# here turns an uncaught ValueError partway through a measurement into an
# argument error before anything runs. Checked against the largest worker
# count this invocation will actually use: --contention-sweep runs its own
# counts irrespective of --workers, so validating against --workers alone
# let the 3-worker measurements finish and then raised on the 6-worker one.
window_workers = args.workers
if args.contention_sweep:
window_workers = max(window_workers, max(CONTENTION_WORKERS))
if args.window is not None and args.window < window_workers:
parser.error(f"--window must be at least the worker count ({window_workers}); a smaller window starves the pool")
if args.graph_seconds is None:
args.graph_seconds = SHA_OVERHEAD_SECONDS / args.scale
if args.contention_sweep:
burn_rate = statistics.median(calibrate_burn() for _ in range(3))
rows = []
for cpu_fraction in (0.0, 0.25, 0.5):
for workers in CONTENTION_WORKERS:
plans = [
build_plan(**_plan_args(args, False, args.seed + i)[0])
for i in range(args.repeat)
]
measured = {}
for name in ("wave", "faithful", "production"):
fn = SCHEDULERS[name]
measured[name] = statistics.median(
fn(
plan,
workers,
outage_limit=0,
graph_seconds=args.graph_seconds,
cpu_fraction=cpu_fraction,
burn_rate=burn_rate,
**_window_kwargs(name, args.window),
).wall_s
for plan in plans
)
serial = statistics.median(
sum(c.seconds for c in _flatten(plan)) for plan in plans
)
rows.append(
{
"cpu_fraction": cpu_fraction,
"workers": workers,
"wave_s": round(measured["wave"], 2),
"faithful_s": round(measured["faithful"], 2),
"production_s": round(measured["production"], 2),
"packing_gain_pct": round(
(measured["faithful"] - measured["wave"]) / measured["wave"] * 100, 1
),
"wave_speedup": round(serial / measured["wave"], 2),
"faithful_speedup": round(serial / measured["faithful"], 2),
}
)
print(json.dumps({"burn_rate": round(burn_rate), "nproc": __import__("os").cpu_count(), "rows": rows}, indent=2))
return 0
if args.window_sweep:
total = len(review_tasks(_read(REVIEW_TASKS))) * args.runs * len(REVIEW_ARMS)
rows = []
for window in (args.workers, args.workers * 2, args.workers * 4, total):
clean = [build_plan(**_plan_args(args, False, args.seed + i)[0]) for i in range(args.repeat)]
wall = statistics.median(
run_faithful(
p, args.workers, outage_limit=0, graph_seconds=args.graph_seconds, window=window
).wall_s
for p in clean
)
failing = build_plan(**_plan_args(args, weekly=False, seed=args.seed, fail_from=12)[0])
trip = run_faithful(
failing,
args.workers,
outage_limit=runner.DEFAULT_OUTAGE_STREAK,
graph_seconds=args.graph_seconds,
window=window,
)
rows.append(
{
"window": window,
"cold_wall_s": round(wall, 3),
"tripped_at": trip.tripped_at,
"executed": trip.executed,
"overrun_cells": trip.executed - (trip.tripped_at + 1)
if trip.tripped_at is not None
else None,
}
)
print(json.dumps({"workers": args.workers, "rows": rows}, indent=2))
return 0
if args.breaker_fidelity:
print(
json.dumps(
{"workers": args.workers, "graph_seconds": round(args.graph_seconds, 4),
"rows": breaker_fidelity(args)},
indent=2,
)
)
return 0
names = sorted(SCHEDULERS) if args.compare else [args.scheduler]
rows: list[dict[str, Any]] = []
for label, weekly in (("weekly", True), ("cold", False)):
plans = []
for i in range(args.repeat):
kwargs, arms = _plan_args(args, weekly, args.seed + i)
plans.append(build_plan(**kwargs))
serial = statistics.median(sum(c.seconds for c in _flatten(p)) for p in plans)
predicted = (
len(plans[0])
* expected_task_seconds(args.runs, arms, args.workers, fed_pool=False)
/ args.scale
)
for name in names:
observed = statistics.median(
SCHEDULERS[name](
p,
args.workers,
outage_limit=0,
graph_seconds=args.graph_seconds,
**_window_kwargs(name, args.window),
).wall_s
for p in plans
)
rows.append(
{
"profile": label,
"scheduler": name,
"workers": args.workers,
"observed_s": round(observed, 3),
"wave_model_s": round(predicted, 3),
"serial_s": round(serial, 3),
"speedup_vs_serial": round(serial / observed, 3) if observed else None,
}
)
print(json.dumps({"scale": args.scale, "repeat": args.repeat, "rows": rows}, indent=2))
return 0
if __name__ == "__main__":
raise SystemExit(main())