mirror of
https://github.com/abhigyanpatwari/GitNexus.git
synced 2026-09-20 00:11:37 +00:00
d1463977c8
* 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>
671 lines
25 KiB
Python
671 lines
25 KiB
Python
"""Unit tests for workflow benchmark aggregation, reporting, task, and CI contracts."""
|
||
|
||
import json
|
||
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,
|
||
build_parser,
|
||
infra_error_record,
|
||
normalized_model_identifier,
|
||
parse_shortstat,
|
||
render_report,
|
||
savings,
|
||
select_tasks,
|
||
systemic_outage_streak,
|
||
)
|
||
|
||
|
||
def record(**overrides):
|
||
base = {
|
||
"input_tokens": 1000,
|
||
"cache_creation_input_tokens": 200,
|
||
"cache_read_input_tokens": 5000,
|
||
"output_tokens": 400,
|
||
"cost_usd": 0.5,
|
||
"duration_s": 60.0,
|
||
"num_turns": 10,
|
||
"diff_files": 2,
|
||
"diff_insertions": 30,
|
||
"diff_deletions": 5,
|
||
"class": "demo",
|
||
"resolved": True,
|
||
}
|
||
base.update(overrides)
|
||
return base
|
||
|
||
|
||
def test_aggregate_takes_medians_and_counts_resolved():
|
||
records = [
|
||
record(input_tokens=1000, resolved=True),
|
||
record(input_tokens=3000, resolved=False),
|
||
record(input_tokens=2000, resolved=True),
|
||
]
|
||
agg = aggregate(records)
|
||
assert agg == {
|
||
"input_tokens": 2000,
|
||
"cache_creation_input_tokens": 200,
|
||
"cache_read_input_tokens": 5000,
|
||
"output_tokens": 400,
|
||
"cost_usd": 0.5,
|
||
"duration_s": 60.0,
|
||
"num_turns": 10,
|
||
"diff_files": 2,
|
||
"diff_insertions": 30,
|
||
"diff_deletions": 5,
|
||
"class": "demo",
|
||
"resolved": 2,
|
||
"runs": 3,
|
||
"valid_runs": 3,
|
||
"excluded_runs": 0,
|
||
"transcripts_missing": 0,
|
||
"error_kinds": {},
|
||
}
|
||
|
||
|
||
def test_savings_is_positive_when_workflow_is_cheaper():
|
||
baseline = aggregate([record(input_tokens=2000, output_tokens=800, cost_usd=1.0)])
|
||
workflow = aggregate([record(input_tokens=1000, output_tokens=400, cost_usd=0.4)])
|
||
s = savings(baseline, workflow)
|
||
assert s["input_tokens"] == 50.0
|
||
assert s["output_tokens"] == 50.0
|
||
assert s["cost_usd"] == 60.0
|
||
|
||
|
||
def task_row(task_id: str, **overrides):
|
||
task = {
|
||
"id": task_id,
|
||
"class": "demo",
|
||
"repo": "/repo",
|
||
"prompt": "do it",
|
||
"verify": "true",
|
||
"oracle": {
|
||
"command": "true",
|
||
"files": [
|
||
{
|
||
"source": "trivial-status-json-alias.oracle.test.ts",
|
||
"target": "oracle.test.ts",
|
||
}
|
||
],
|
||
},
|
||
}
|
||
task.update(overrides)
|
||
return task
|
||
|
||
|
||
def test_expensive_tasks_are_opt_in_and_reported_as_skipped():
|
||
tasks = [task_row("default"), task_row("large", expensive=True)]
|
||
selected, skipped = select_tasks(tasks, include_expensive=False)
|
||
assert [task["id"] for task in selected] == ["default"]
|
||
assert skipped == ["large"]
|
||
|
||
selected, skipped = select_tasks(tasks, include_expensive=True)
|
||
assert [task["id"] for task in selected] == ["default", "large"]
|
||
assert skipped == []
|
||
|
||
|
||
@pytest.mark.parametrize("value", ["true", 1, None, [], {}])
|
||
def test_expensive_metadata_must_be_boolean(value):
|
||
with pytest.raises(ValueError, match="expensive.*boolean"):
|
||
select_tasks([task_row("bad", expensive=value)], include_expensive=False)
|
||
|
||
|
||
def test_task_selection_rejects_duplicate_ids_and_empty_selection():
|
||
with pytest.raises(ValueError, match="duplicate task id"):
|
||
select_tasks([task_row("same"), task_row("same")], include_expensive=True)
|
||
with pytest.raises(ValueError, match="no tasks selected"):
|
||
select_tasks([task_row("large", expensive=True)], include_expensive=False)
|
||
|
||
|
||
def test_runner_requires_a_named_model_and_supports_expensive_opt_in():
|
||
with pytest.raises(SystemExit):
|
||
build_parser().parse_args(["--tasks", "tasks.yaml"])
|
||
args = build_parser().parse_args(
|
||
[
|
||
"--tasks",
|
||
"tasks.yaml",
|
||
"--model",
|
||
"claude-sonnet-4-20250514",
|
||
"--include-expensive",
|
||
]
|
||
)
|
||
assert args.include_expensive is True
|
||
with pytest.raises(ValueError, match="nonblank"):
|
||
normalized_model_identifier(" ")
|
||
|
||
|
||
@pytest.mark.parametrize(
|
||
"alias",
|
||
["Auto", "AUTO", "latest", "provider/latest", "provider:Latest", "provider@LATEST"],
|
||
)
|
||
def test_runner_rejects_mutable_model_aliases(alias):
|
||
with pytest.raises(ValueError, match="mutable auto/latest"):
|
||
normalized_model_identifier(alias)
|
||
assert normalized_model_identifier("free-coder") == "free-coder"
|
||
assert normalized_model_identifier("claude-sonnet-4-20250514") == "claude-sonnet-4-20250514"
|
||
|
||
|
||
def test_eval_ci_uses_locked_uv_and_blocking_native_containment_jobs():
|
||
repo_root = Path(__file__).resolve().parents[2]
|
||
workflow = (repo_root / ".github" / "workflows" / "ci-tests.yml").read_text()
|
||
workflow_document = yaml.safe_load(workflow)
|
||
containment = workflow_document["jobs"]["eval-containment-linux"]
|
||
containment_steps = {step.get("name"): step for step in containment["steps"] if "name" in step}
|
||
containment_node_setup = next(
|
||
step for step in containment["steps"] if str(step.get("uses", "")).startswith("actions/setup-node@")
|
||
)
|
||
claude_lock = json.loads((repo_root / ".github" / "claude-canary-runtime" / "package-lock.json").read_text())
|
||
setup_uv = "astral-sh/setup-uv@11f9893b081a58869d3b5fccaea48c9e9e46f990"
|
||
assert workflow.count(setup_uv) >= 3
|
||
assert workflow.count("version: '0.11.23'") >= 3
|
||
assert workflow.count("uv run --locked --extra dev python -m pytest") >= 3
|
||
assert "eval-containment-linux:" in workflow
|
||
assert "GITNEXUS_REQUIRE_BWRAP_CANARY: '1'" in workflow
|
||
assert "GITNEXUS_REQUIRE_CLAUDE_CANARY: '1'" in workflow
|
||
assert containment["env"] == {
|
||
"GITNEXUS_REQUIRE_BWRAP_CANARY": "1",
|
||
"GITNEXUS_REQUIRE_CLAUDE_CANARY": "1",
|
||
}
|
||
assert containment["timeout-minutes"] == 20
|
||
assert containment_node_setup["with"] == {
|
||
"node-version": "22.18.0",
|
||
"cache": "npm",
|
||
"cache-dependency-path": "gitnexus/package-lock.json\ngitnexus-shared/package-lock.json\n",
|
||
}
|
||
assert (
|
||
"CLAUDE_CANARY_BIN: ${{ runner.temp }}/claude-canary/node_modules/@anthropic-ai/claude-code-linux-x64/claude"
|
||
in workflow
|
||
)
|
||
assert ".github/claude-canary-runtime/package-lock.json" in workflow
|
||
assert "npm ci" in workflow
|
||
assert "--package-lock=false" not in workflow
|
||
assert claude_lock["packages"]["node_modules/@anthropic-ai/claude-code"]["version"] == "2.1.214"
|
||
assert claude_lock["packages"]["node_modules/@anthropic-ai/claude-code"]["integrity"].startswith("sha512-")
|
||
assert "if(p.version!=='2.1.214') process.exit(1)" in workflow
|
||
assert "'2.1.214 (Claude Code)'" in workflow
|
||
# Shared is compiled by gitnexus `npm run build` (scripts/build.js runTsc).
|
||
# A dedicated npm ci in gitnexus-shared pulls TypeScript 7 and stalls CI.
|
||
assert "Build pinned shared runtime" not in containment_steps
|
||
assert not any(
|
||
step.get("working-directory") == "gitnexus-shared" and "npm ci" in str(step.get("run", ""))
|
||
for step in containment["steps"]
|
||
)
|
||
assert containment_steps["Install and build pinned GitNexus runtime"]["working-directory"] == "gitnexus"
|
||
assert containment_steps["Install and build pinned GitNexus runtime"]["run"].splitlines() == [
|
||
"npm ci",
|
||
"npm run build",
|
||
]
|
||
selected_containment_tests = containment_steps["Prove process-tree and sandbox containment"]["run"].split()
|
||
assert selected_containment_tests == [
|
||
"uv",
|
||
"run",
|
||
"--locked",
|
||
"--extra",
|
||
"dev",
|
||
"python",
|
||
"-m",
|
||
"pytest",
|
||
"tests/test_process_control.py",
|
||
"tests/test_proposer_sandbox.py",
|
||
"tests/test_workflow_bench_sessions.py",
|
||
"tests/test_ce_plugin_runtime.py",
|
||
"-q",
|
||
]
|
||
bwrap_canary_marker = re.compile(
|
||
r'@pytest\.mark\.skipif\(\s*os\.environ\.get\("GITNEXUS_REQUIRE_BWRAP_CANARY"\)',
|
||
re.MULTILINE,
|
||
)
|
||
bwrap_canary_files = sorted(
|
||
path.name
|
||
for path in (repo_root / "eval" / "tests").glob("test_*.py")
|
||
if bwrap_canary_marker.search(path.read_text())
|
||
)
|
||
assert bwrap_canary_files == ["test_proposer_sandbox.py", "test_workflow_bench_sessions.py"]
|
||
assert all(f"tests/{name}" in selected_containment_tests for name in bwrap_canary_files)
|
||
assert "eval-containment-windows:" in workflow
|
||
|
||
|
||
def test_shipped_scenarios_opt_out_the_cross_module_cell_and_rebuild_graph_assets():
|
||
task_file = Path(__file__).resolve().parents[1] / "workflow_bench" / "tasks.scenarios.yaml"
|
||
tasks = yaml.safe_load(task_file.read_text())["tasks"]
|
||
selected, skipped = select_tasks(tasks, include_expensive=False)
|
||
assert [task["id"] for task in selected] == [
|
||
"trivial-status-json-alias",
|
||
"inv-bug-c-system-include",
|
||
"inv-feature-list-repos-filter",
|
||
]
|
||
assert skipped == ["cross-module-parse-retry"]
|
||
assert all(not task.get("sandbox_copy") for task in tasks)
|
||
assert all(task["sandbox_dependencies"] for task in tasks)
|
||
assert all(task["oracle"]["command"] and task["oracle"]["files"] for task in tasks)
|
||
assert all("./node_modules/.bin/vitest run" in task["oracle"]["command"] for task in tasks)
|
||
assert all("npx vitest" not in task["oracle"]["command"] for task in tasks)
|
||
assert all(
|
||
'--config "$GITNEXUS_BENCH_ORACLE_ROOT/vitest.config.mts"' in task["oracle"]["command"] for task in tasks
|
||
)
|
||
assert all({item["target"] for item in task["oracle"]["files"]} >= {"vitest.config.mts"} for task in tasks)
|
||
|
||
|
||
def test_savings_handles_zero_baseline_without_dividing():
|
||
baseline = aggregate([record(cost_usd=0.0)])
|
||
workflow = aggregate([record(cost_usd=0.0)])
|
||
assert savings(baseline, workflow)["cost_usd"] == 0.0
|
||
|
||
|
||
def test_parse_shortstat_full_and_empty():
|
||
full = parse_shortstat(" 3 files changed, 120 insertions(+), 7 deletions(-)")
|
||
assert full == {"diff_files": 3, "diff_insertions": 120, "diff_deletions": 7}
|
||
assert parse_shortstat("") == {
|
||
"diff_files": 0,
|
||
"diff_insertions": 0,
|
||
"diff_deletions": 0,
|
||
}
|
||
singular = parse_shortstat(" 1 file changed, 1 insertion(+)")
|
||
assert singular == {"diff_files": 1, "diff_insertions": 1, "diff_deletions": 0}
|
||
|
||
|
||
def test_render_report_emits_arm_rows_and_per_arm_savings_rows():
|
||
results = {
|
||
"demo-task": {
|
||
"workflow": aggregate([record(input_tokens=1000)]),
|
||
"workflow_direct": aggregate([record(input_tokens=1500)]),
|
||
"baseline": aggregate([record(input_tokens=2000)]),
|
||
}
|
||
}
|
||
report = render_report(results)
|
||
assert "| demo-task | demo | workflow | 1/1 | 1000 |" in report
|
||
assert "| demo-task | demo | baseline | 1/1 | 2000 |" in report
|
||
assert "| demo-task | demo | **workflow savings %** | — | 50.0 |" in report
|
||
assert "| demo-task | demo | **workflow_direct savings %** | — | 25.0 |" in report
|
||
assert "2/+30/−5" in report
|
||
assert "results.jsonl" in report
|
||
assert "subagent spend" in report # token columns are main-loop-only
|
||
|
||
|
||
def test_aggregate_excludes_session_error_rows_from_medians():
|
||
records = [
|
||
record(cost_usd=1.0),
|
||
record(cost_usd=3.0, transcript_missing=True),
|
||
record(cost_usd=100.0, resolved=False, error_kind="session-error"),
|
||
]
|
||
agg = aggregate(records)
|
||
assert agg["cost_usd"] == 2.0
|
||
assert agg["runs"] == 3
|
||
assert agg["valid_runs"] == 2
|
||
assert agg["excluded_runs"] == 1
|
||
assert agg["transcripts_missing"] == 1
|
||
assert agg["resolved"] == 2
|
||
|
||
|
||
def test_aggregate_excludes_unverified_transcript_evidence():
|
||
agg = aggregate(
|
||
[
|
||
record(cost_usd=1.0),
|
||
record(
|
||
cost_usd=100.0,
|
||
resolved=False,
|
||
error_kind="evidence-unverified",
|
||
transcript_missing=True,
|
||
),
|
||
]
|
||
)
|
||
assert agg["cost_usd"] == 1.0
|
||
assert agg["valid_runs"] == 1
|
||
assert agg["excluded_runs"] == 1
|
||
|
||
|
||
def test_aggregate_excludes_invalid_review_artifacts_from_quality_metrics():
|
||
scored = record(
|
||
cost_usd=1.0,
|
||
review_weighted_f1=0.8,
|
||
review_true_positives=2,
|
||
review_false_positives=0,
|
||
review_false_negatives=1,
|
||
review_precision=1.0,
|
||
review_recall=0.67,
|
||
review_f1=0.8,
|
||
review_weighted_precision=0.8,
|
||
review_weighted_recall=0.8,
|
||
review_blocker_recall=1.0,
|
||
review_severity_accuracy=1.0,
|
||
review_category_accuracy=1.0,
|
||
review_grounded_evidence=1.0,
|
||
review_clean_control=False,
|
||
)
|
||
agg = aggregate(
|
||
[
|
||
scored,
|
||
record(cost_usd=2.0, resolved=False, error_kind="review-evidence-invalid"),
|
||
]
|
||
)
|
||
assert agg["valid_runs"] == 1
|
||
assert agg["excluded_runs"] == 1
|
||
assert agg["review_weighted_f1"] == 0.8
|
||
assert agg["review_true_positives"] == 2
|
||
|
||
|
||
def test_render_report_surfaces_excluded_and_unverified_runs():
|
||
results = {
|
||
"t": {
|
||
"workflow": aggregate(
|
||
[
|
||
record(transcript_missing=True),
|
||
record(resolved=False, error_kind="session-error"),
|
||
]
|
||
)
|
||
}
|
||
}
|
||
report = render_report(results)
|
||
assert "| t | demo | workflow | 1/1 (1 excluded) |" in report
|
||
assert "session/infra errors" in report
|
||
assert "no locatable session transcript" in report
|
||
|
||
|
||
def test_render_report_surfaces_why_each_row_failed():
|
||
results = {
|
||
"t": {
|
||
"workflow": aggregate(
|
||
[record(resolved=False, error_kind="plan-evidence-invalid")],
|
||
),
|
||
}
|
||
}
|
||
report = render_report(results)
|
||
assert "plan-evidence-invalid×1" in report
|
||
|
||
|
||
def test_broken_incumbent_arms_flags_an_incumbent_that_resolved_nothing():
|
||
results = {
|
||
"t1": {"workflow": aggregate([record(resolved=False, error_kind="plan-evidence-invalid")])},
|
||
"t2": {"workflow": aggregate([record(resolved=False, error_kind="plan-evidence-invalid")])},
|
||
}
|
||
assert broken_incumbent_arms(results, {"workflow"}) == ["workflow"]
|
||
|
||
|
||
def test_broken_incumbent_arms_ignores_a_merely_underperforming_candidate():
|
||
# The incumbent works fine; only the candidate arm fails. That's a normal,
|
||
# expected "bad candidate" outcome and must not read as a broken harness.
|
||
results = {
|
||
"t1": {
|
||
"workflow": aggregate([record(resolved=True)]),
|
||
"candidate_workflow": aggregate([record(resolved=False, error_kind="verify-failed")]),
|
||
},
|
||
}
|
||
assert broken_incumbent_arms(results, {"workflow"}) == []
|
||
|
||
|
||
def test_broken_incumbent_arms_flags_an_incumbent_with_zero_valid_runs():
|
||
# Every run excluded via an excluded-but-non-systemic error_kind
|
||
# ("evidence-unverified"): valid_runs == 0 for every task, which the old
|
||
# `valid_runs > 0` guard let sail through silently, and which the outage
|
||
# streak breaker also doesn't catch (it resets rather than accumulates
|
||
# on this exact error_kind -- see test_systemic_outage_streak_resets_on_non_outage).
|
||
results = {
|
||
"t1": {"workflow": aggregate([record(resolved=False, error_kind="evidence-unverified")])},
|
||
"t2": {"workflow": aggregate([record(resolved=False, error_kind="evidence-unverified")])},
|
||
}
|
||
assert results["t1"]["workflow"]["valid_runs"] == 0
|
||
assert broken_incumbent_arms(results, {"workflow"}) == ["workflow"]
|
||
|
||
|
||
def test_broken_incumbent_arms_ignores_partial_incumbent_failure():
|
||
# Resolved in at least one task — struggling, not broken.
|
||
results = {
|
||
"t1": {"workflow": aggregate([record(resolved=False, error_kind="verify-failed")])},
|
||
"t2": {"workflow": aggregate([record(resolved=True)])},
|
||
}
|
||
assert broken_incumbent_arms(results, {"workflow"}) == []
|
||
|
||
|
||
def test_infra_error_record_captures_the_failure_and_is_excluded():
|
||
exc = subprocess.TimeoutExpired(cmd="claude -p", timeout=5)
|
||
rec = infra_error_record(exc)
|
||
assert rec["ok"] is False
|
||
assert rec["resolved"] is False
|
||
assert rec["error_kind"] == "infra-error"
|
||
assert "TimeoutExpired" in rec["error_detail"]
|
||
assert rec["output_tokens"] == 0
|
||
agg = aggregate([record(cost_usd=2.0), rec])
|
||
assert agg["cost_usd"] == 2.0
|
||
assert agg["valid_runs"] == 1
|
||
assert agg["excluded_runs"] == 1
|
||
|
||
|
||
def test_systemic_outage_streak_counts_consecutive_systemic_failures():
|
||
# session/infra/cleanup failures accumulate; a cleanup-failure that masked a
|
||
# session-error still counts toward the streak.
|
||
streak = 0
|
||
for kind in ("session-error", "infra-error", "cleanup-failure"):
|
||
streak = systemic_outage_streak(kind, streak)
|
||
assert streak == 3
|
||
assert systemic_outage_streak("cleanup-failure", 4) == 5
|
||
|
||
|
||
def test_systemic_outage_streak_resets_on_non_outage():
|
||
# A real task failure (resolved=False → error_kind None) or an unverifiable
|
||
# evidence run is not an outage and resets the streak.
|
||
assert systemic_outage_streak(None, 4) == 0
|
||
assert systemic_outage_streak("evidence-unverified", 4) == 0
|
||
|
||
|
||
def test_outage_streak_flag_defaults_and_disables():
|
||
base = ["--tasks", "tasks.yaml", "--model", "claude-sonnet-4-20250514"]
|
||
assert build_parser().parse_args(base).outage_streak == 5
|
||
assert build_parser().parse_args([*base, "--outage-streak", "0"]).outage_streak == 0
|
||
|
||
|
||
def test_run_evolution_script_is_the_shared_ci_and_local_entrypoint():
|
||
eval_dir = Path(__file__).resolve().parents[1]
|
||
script = eval_dir / "workflow_bench" / "run-evolution.sh"
|
||
workflow = eval_dir.parent / ".github" / "workflows" / "gitnexus-skill-evolution.yml"
|
||
assert script.is_file()
|
||
assert script.stat().st_mode & 0o111
|
||
workflow_text = workflow.read_text()
|
||
assert "./workflow_bench/run-evolution.sh --apply" in workflow_text
|
||
assert "python -m workflow_bench.evolve" not in workflow_text
|
||
|
||
env = {
|
||
"PATH": os.environ.get("PATH", "/usr/bin"),
|
||
"MODEL": "claude-sonnet-5",
|
||
"PROPOSER_MODEL": "claude-opus-4-8",
|
||
"EFFORT": "xhigh",
|
||
"GENERATIONS": "1",
|
||
"RUNS": "3",
|
||
"WORKERS": "2",
|
||
"PROVIDER": "openai",
|
||
"INCLUDE_EXPENSIVE": "1",
|
||
"SEED_RESULTS": "/tmp/seed-bench",
|
||
"CLAUDE_BIN": "/opt/claude",
|
||
"OUT_ROOT": "/tmp/wfevolve",
|
||
"CE_PLUGIN_DIR": "/tmp/ce-plugin",
|
||
"CE_PLUGIN_VERSION": "3.24.0",
|
||
"HOME": os.environ.get("HOME", "/tmp"),
|
||
}
|
||
printed = subprocess.run(
|
||
[str(script), "--dry-run", "--apply"],
|
||
check=True,
|
||
capture_output=True,
|
||
text=True,
|
||
env=env,
|
||
)
|
||
argv = shlex.split(printed.stdout)
|
||
assert argv[:7] == ["uv", "run", "--locked", "--extra", "dev", "python", "-m"]
|
||
assert argv[7] == "workflow_bench.evolve"
|
||
assert argv[argv.index("--tasks") + 1] == "workflow_bench/tasks.review.scenarios.yaml"
|
||
assert argv[argv.index("--arms") + 1] == "review"
|
||
assert argv[argv.index("--ce-plugin-version") + 1] == "3.24.0"
|
||
assert argv[argv.index("--model") + 1] == "gpt-5.6-sol"
|
||
assert argv[argv.index("--proposer-model") + 1] == "gpt-5.6-sol"
|
||
assert argv[argv.index("--effort") + 1] == "xhigh"
|
||
assert argv[argv.index("--workers") + 1] == "2"
|
||
assert argv[argv.index("--claude-bin") + 1] == "/opt/claude"
|
||
assert argv[argv.index("--out-root") + 1] == "/tmp/wfevolve"
|
||
assert argv[argv.index("--seed-results") + 1] == "/tmp/seed-bench"
|
||
assert "--apply" in argv
|
||
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"
|