mirror of
https://github.com/abhigyanpatwari/GitNexus.git
synced 2026-10-04 02:31:36 +00:00
fix(eval): review artifact path, evidence classification, comparator reuse
Extracted from the combined skill-evolution branch. This is the runtime change set: everything that alters how a sweep executes and what it records. The packed scheduler and its measurement harness were separated onto perf/skill-evolution-packed-scheduler, which is purely additive. Correctness. The review artifact was mounted as a writable FILE inside a read-only workspace while the agent's Write tool writes atomically - temp file beside the target, then rename - so the temp create failed EROFS and the artifact was never written. Four layers gate that path and three named the old location: the CLI's own allowWrite/allowRead policy, the task corpus verify command run in its own sandbox invocation, and host_text/host_path for the host-unsafe backend. Fixing the translator exposed a second defect, since "/review-output" appears twice in "/review-output/review-output.json"; matching is now anchored to a path boundary. parse_review_output folded OSError, UnicodeError and JSONDecodeError into one message, so an artifact that was never written looked like an encoding fault; each cause now names itself. Health classification. broken_incumbent_arms inferred a broken environment from an arm resolving zero tasks, which a reviewer facing a hard corpus falsifies - Actions run 33962002890 is exactly that shape. Arms are now classified from fresh execution and evidence outcomes as UNKNOWN, OBSERVED_OK, DEGRADED or UNUSABLE, and only UNUSABLE aborts. Resolution count is not consulted. The guard names no cause: an empty artifact establishes unusable evidence, not that a mount rejected the write. Comparator reuse. Reuse accepted evidence it should have rejected: a row without a runtime_digest passed the drift lock, recorded_at was overwritten with the copy time so a row could outlive its own max_age, and the binding ignored task-asset and dependency digests although this change alters sandbox_dependencies in the review corpus. Closing the last one required preparing asset snapshots before the reuse decision, which also removes the concurrent TaskAssetCache.prepare the prefetch thread could race. These three concerns share aggregate() and _run_sweep, which is why they ship together: separating them further would mean hunk-level surgery on a function all three modify, and the risk of a silent omission outweighs the reviewability gain. 592 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. Known gap, and the reason this is not ready to merge: no test drives _run_sweep end to end. enforce_measurement_health is unit-tested including the below-breaker unusable case, and the caller wiring is pinned structurally by reading _run_sweep's compiled code object, but interruption semantics, exit precedence and persisted artifacts are not exercised through the real path.
This commit is contained in:
Gergo Magyar
parent
46ce838e8f
commit
1506a250dc
6 changed files with 0 additions and 1424 deletions
|
|
@ -1,122 +0,0 @@
|
|||
"""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
|
||||
|
|
@ -680,97 +680,6 @@ def test_graph_build_env_ready_keys_covers_successes_and_failures():
|
|||
}
|
||||
|
||||
|
||||
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,
|
||||
)
|
||||
|
||||
|
||||
def _cell(**overrides) -> dict[str, Any]:
|
||||
"""One results.jsonl row, healthy unless told otherwise."""
|
||||
|
||||
|
|
|
|||
|
|
@ -1,311 +0,0 @@
|
|||
#!/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())
|
||||
|
|
@ -766,26 +766,6 @@ CHURN_FIELDS = ("diff_files", "diff_insertions", "diff_deletions")
|
|||
EXCLUDED_ERROR_KINDS = REUSE_EXCLUDED_ERROR_KINDS
|
||||
|
||||
|
||||
# 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
|
||||
|
||||
# Health classification. These answer "did the harness work", which is a
|
||||
# different question from "did the agent get the right answer" - a review can be
|
||||
|
|
@ -925,155 +905,6 @@ 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)
|
||||
|
|
|
|||
|
|
@ -1,69 +0,0 @@
|
|||
{
|
||||
"_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."
|
||||
}
|
||||
|
|
@ -1,662 +0,0 @@
|
|||
#!/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 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
|
||||
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."""
|
||||
|
||||
started = time.monotonic()
|
||||
executed = 0
|
||||
for cells in plan:
|
||||
time.sleep(graph_seconds)
|
||||
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 restores the wave's
|
||||
``workers - 1`` overrun bound while still packing across task boundaries.
|
||||
Defaults to ``workers``, the smallest value that keeps every worker fed.
|
||||
"""
|
||||
|
||||
if window is None:
|
||||
window = workers
|
||||
|
||||
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 _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
|
||||
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}
|
||||
for name in ("wave", "faithful", "production"):
|
||||
out = SCHEDULERS[name](
|
||||
plan, args.workers, outage_limit=limit, graph_seconds=args.graph_seconds
|
||||
)
|
||||
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"]
|
||||
)
|
||||
row["overrun_within_bound"] = (
|
||||
row["production"]["overrun"] is not None and row["production"]["overrun"] <= args.workers
|
||||
)
|
||||
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)
|
||||
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()
|
||||
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 (3, 6):
|
||||
kwargs, _arms = _plan_args(args, weekly=False, seed=args.seed)
|
||||
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]
|
||||
extra = {"window": 12} if name == "faithful" else {}
|
||||
measured[name] = statistics.median(
|
||||
fn(
|
||||
plan,
|
||||
workers,
|
||||
outage_limit=0,
|
||||
graph_seconds=args.graph_seconds,
|
||||
cpu_fraction=cpu_fraction,
|
||||
burn_rate=burn_rate,
|
||||
**extra,
|
||||
).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),
|
||||
"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):
|
||||
kwargs, arms = _plan_args(args, weekly=False, seed=args.seed)
|
||||
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).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())
|
||||
Loading…
Add table
Reference in a new issue