GitNexus/eval/workflow_bench/runner.py
Gergő Magyar b1d87c1f33
fix(eval): sweep evidence handling and measurement health, with guarded comparator reuse (#3207)
* fix(eval): cut skill-evolution wall clock without shrinking the gate

Reuse matching incumbent/CE cells, sanitize each SHA once, and default
dispatch workers to 3 so weekly review generations finish inside the
EventBridge window. Cap the sweep from leftover instance uptime so a
Friday dispatch still uploads evidence.

Co-authored-by: Cursor <cursoragent@cursor.com>
Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>

* perf(eval): pipeline graph setup and correct the wall-clock cost model

The evolution sweep paid `sanitize` + `analyze --pdg --index-only` for every
unique task SHA on the critical path, one at a time, with nothing overlapping.
`_run_sweep` now starts the next unpaid SHA's clone template and graph snapshot
on a prefetch thread as soon as the current task's cells are dispatched, so
every SHA but the first hides behind a paid session wave. The thread is joined
before that SHA is used and before the trees tempdir is torn down, and a
prefetch failure is recorded against the SHA exactly as an inline failure is.

Tasks whose cells are all reusable comparator rows are not prefetched: they
never build a graph, so priming one would be pure cost.

Adds `measure_evolution_cost.py`, the cost model behind these numbers. It reads
the review corpus, the evolve defaults, and the workflow's workers default —
it does not start a session. Its first version charged `copy_isolated_tree`
once per paid cell, serially. `run_cell` clones inside its own pool worker, so
the clones in a wave overlap and only one is on the critical path per wave;
the model now charges `ceil(cells / workers)` waves.

Estimated review generation at workers=3: cold 21570s, weekly 7710s.

Wall clock is quantised by `ceil(cells_per_task / workers)`. A cold review task
is 9 cells, so workers=4 buys the wall clock of workers=3 and pays host
contention for it. Documented in the workflow's rollout checklist.
Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>

* perf(eval): price the benchmark against measured cell durations

The cost model assumed every cell runs the 1140s mean. Cells are not uniform:
the 41 rows in Actions run 33912693948's artifact are 826s at the median,
1262s at the mean, 2976s at p90, with two pinned at the 5400s session ceiling.
A wave waits for its slowest cell, so a mean understates every concurrent
schedule — the previous model called workers=3 cold 5.99h when the same
schedule against real durations is 10.33h.

session_durations.json carries the sample in submission order with its
provenance and its caveat: every cell in that run returned unusable evidence,
so the durations are real but a clean run may sit lower. It is the only live
artifact; the 2026-07-22 green run's has expired.

The model now simulates the schedule cell by cell rather than multiplying a
mean by a wave count, averaged over all 41 rotations of the sample so no
single alignment between sample order and cell index decides the answer. It
prices today's barrier (wave_makespan) against a continuously fed pool
(fed_makespan) and reports both, and it charges the proposer session — one
per generation, measured at 344.7s — which it had been omitting entirely.

Measurement only; no runtime behaviour changes.
Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>

* perf(eval): price arms separately and stop inventing setup constants

Two errors in the model, both found by auditing it against the artifact it
claims to describe.

The arms are not interchangeable. `candidate_review` runs 1416s at the mean
against `review`'s 1204s and `ce_review`'s 1176s, and the weekly lane pays the
candidate arm and nothing else — reuse skips both incumbents. Pricing weekly
from a pooled sample charged it for arms it never runs: weekly is 4.59h, not
the 3.65h a pooled sample reported. Cells are also submitted run-major and
arm-minor, so at workers=3 every wave holds one cell of each arm and the
slowest arm sets the wave; the model now builds cells in that order.

The setup constants were invented. GRAPH_ANALYZE_SECONDS=600 and
TEMPLATE_SANITIZE_SECONDS=180 charged 3900s of per-SHA setup for a cold run —
more than the entire non-session time of the source run, which was 2541s for
41 cells and 5 SHAs. `duration_s` is the sum of a cell's Claude sessions
(runner_sessions.py), so that 2541s residual is every clone, graph build,
sandbox and teardown the sweep paid. The model now charges the measured
residual per cell, 62.0s, and no longer credits clone templates or graph
prefetch: both landed after that run and there is no measurement of them yet.
The residual bounds what they can be worth.

Cold 37452s (10.40h), weekly 16541s (4.59h), against a fed pool at 31683s and
16541s. Measurement only; no runtime behaviour changes.
Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>

* perf(eval): charge sweep overhead where more workers cannot dissolve it

Three defects, found by auditing the model against the artifact again.

The overhead was charged inside the schedule. session_durations.json claimed
the residual was charged "per cell and serially - the pessimistic reading",
but task_cells folded it into each cell's duration, where the pool then
divided it by the worker count. The residual mixes per-cell work the pool
really does divide with per-SHA graph setup it cannot, and the artifact cannot
separate them, so it now sits outside the schedule: cold 11.09h, not 10.40h.

Alignment averaging weighted the shortest sample twice. The arm samples are 13,
14 and 14 long and the average ran over max()=14 offsets, so candidate_review's
first cell was counted twice and its last never. Averaging over lcm()=182
offsets weights every arm's sample evenly.

The wall assumed all 54 cells run. Replaying the sample's own error_kind
sequence through today's systemic_outage_streak trips the outage breaker at
cell 5 of 41. The source run executed all 41, so its runner did not break on
that sequence, but the current one would: these numbers price a HEALTHY sweep,
and a sweep with the sample's failure profile never reaches them. Stated on
generation_seconds and recorded next to the sample it qualifies.

Measurement only; no runtime behaviour changes.
Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>

* fix(eval): give the review agent somewhere it can actually write

Every review cell in the last recorded generation returned unusable evidence.
Not some — all 41, across all three arms and all six tasks, at $3653 for the
run. The transcripts say why, 127 times across 35 of 35 sessions:

    EROFS: read-only file system,
    open '/workspace/review-output.json.tmp.2.90a76e583b0c'

The review arm mounted the artifact as a writable FILE at
/workspace/review-output.json while binding /workspace read-only. The Write
tool writes atomically: it creates `<target>.tmp.<n>.<hex>` beside the target
and renames it. The parent was read-only, so the temp create failed and the
artifact was never written. A writable file inside a read-only directory is
not writable to anything that writes atomically. Agents tried
/proc/self/root/workspace/... and /proc/1/root/workspace/... to get around it;
all 41 artifacts came back 0 bytes.

The artifact now lives in its own writable directory bound at /review-output,
outside the workspace. That is what a rename needs, and it lets the workspace
get stricter rather than looser: the review phase may now change nothing there
at all (enforce_phase_workspace gained allowed_artifact=None), where before it
was entitled to one path inside it. The file is no longer pre-created — the
agent writes it, and absence is now meaningful evidence.

parse_review_output reported every one of these as "review output is not valid
UTF-8 JSON". The file was empty, and its except folded OSError, UnicodeError
and JSONDecodeError into that one string, so a sandbox that made writing
impossible was indistinguishable from an encoding fault. That is why this read
as an agent-quality problem for fifteen consecutive non-green runs. Each cause
now names itself: never written, empty, not valid UTF-8, not valid JSON with
the decoder's position. run_arm also keeps the FIRST error_detail, as it
already did for error_kind, so a phase-boundary violation is no longer buried
under the parse failure it causes.

The test double conflated sandbox.private_root with the clone, which put the
artifact directory inside the workspace and would have hidden the stricter
check. Regression tests pin the mount shape in the generated bwrap argv, the
contract path in the prompt, the four parse diagnostics, and the
untouched-workspace contract.

Verified by unit tests only: this container has unprivileged user namespaces
disabled, so bwrap cannot run here and the mount was not exercised end to end.
Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>

* fix(review): close the artifact path in every layer that gates it

Code review of this branch found the relocated review artifact was fixed in the
bwrap mount and nowhere else. Four independent layers decide whether the agent
can write it, and three still named the old location.

Claude Code applies its own filesystem policy to its own tools, and
build_claude_settings listed only /workspace, /tmp and /home/agent under
allowWrite with denyRead ["/"]. The artifact used to live under /workspace, so
this list was correct until it moved. SANDBOX_REVIEW_OUTPUT is now in allowWrite
and allowRead; without it the bwrap bind grants a write the CLI then refuses.

The task corpus still ran `test -s review-output.json` from the workspace, in a
separate sandbox invocation that never sees the artifact mount. Every review
cell would have been stamped verify-failed with resolved=False no matter how
good the review was, which also made those rows permanently unreusable and so
silently disabled this branch's own comparator reuse for review arms. The verify
and hidden-oracle commands now read the location from
GITNEXUS_BENCH_REVIEW_OUTPUT and get the directory bound read-only, mirroring
the mount-plus-env-var shape _run_hidden_oracle already used.

host_text and host_path did not translate the new path, so the host-unsafe
backend told the agent to write somewhere that exists on neither backend.
Adding the mapping exposed a second defect: host_text substituted every
occurrence of a target, and "/review-output" appears twice in
"/review-output/review-output.json" - once as the directory and once inside the
filename. Matching is now anchored to a path boundary.

Comparator reuse had three ways to accept evidence it should have rejected. A
row with no runtime_digest passed the drift lock because the guard only compared
when both sides were bound, and the branch's own test asserted that as correct;
absence is now a mismatch and the test states the rule. materialize_reused_row
overwrote recorded_at with the copy time while the age check read that field, so
a row copied forward each generation refreshed its own clock and never aged out;
the first measurement time is now preserved and aged against. A future-dated
stamp passed a one-sided bound and is now rejected as corrupt.

RUNTIME_DIGEST never reached the runner at all: runner_environment builds a
fixed dict and process_control replaces the child environment wholesale, so the
digest the workflow exports was dropped and the lock it feeds was inert. The
instance-window deadline was also checked only after run_proposer returned,
buying a proposal the generation had no room to benchmark.

The graph prefetch thread was started without copy_context, so it never saw the
cancellation ContextVar the rest of the sweep shares, and the outage breaker
returned without setting cancel_event - together, a tripped breaker would block
on joining a prefetch that was never told to stop. Both fixed, with outage
checked before cancellation at the two exits so an outage keeps exit 1 instead
of becoming a Ctrl-C's 130.

Both bwrap canaries that actually execute a write still bound the pre-fix shape
against a file this branch no longer creates, so they would have errored rather
than caught anything. They now bind the directory and write atomically - temp
file beside the target, then rename - which is the exact operation that failed
with EROFS. A source-text assertion over inspect.getsource(run_arm) was replaced
with one that inspects the real mount, and a wall-clock assertion was pinned to
a fixed monotonic clock.

Not applied, and why: binding task-asset and dependency digests into comparator
reuse needs asset snapshots prepared before the reuse decision rather than
inside the per-task loop, and shipping the comparison without that would add a
guard that silently never fires. Forcing a paid canary cell per incumbent arm
and folding reused rows into the outage streak are behaviour decisions, not
fixes. Clone-template reuse still has no test. The cost model's per-cell
residual still shrinks with arm count, overstating weekly savings by at most the
2541s residual; the docstring now says so rather than inventing a split.

585 eval tests pass, ruff clean, 29 workflow contract tests pass. The two
test_model_gateway.py failures are pre-existing and fail on main.

* fix(review): bind reuse to its environment and keep the health canary real

Applies the five findings the previous review round left open.

Comparator reuse ignored the environment a row was measured in. TaskReuseBinding
carried the task and oracle identity but not the task-asset or sandbox-dependency
digests, and this branch itself changes sandbox_dependencies in the review
corpus - so a reused comparator could be measured against one dependency set and
compared against a candidate built on another, handing the gate a false
comparison. Closing it needed the digests to exist before the reuse decision, so
asset snapshots are now prepared for every task up front instead of lazily
inside the per-task loop. That also removes the concurrent TaskAssetCache.prepare
the prefetch thread could otherwise race, which the file's own "plain dict,
read-then-write race" comment warned about. Both digests fail closed on either
side, matching the runtime digest.

The broken-incumbent canary could not fire when reuse was working. It read
`resolved`, which counts reused rows, so an arm whose cells were all reused
always looked healthy - in precisely the run where a broken environment would go
unnoticed. aggregate now also reports `resolved_fresh` and the canary reads it.
That count would be vacuous if an arm were reused end to end, so the sweep keeps
one paid cell per incumbent arm and says which one it kept.

Reused rows did not participate in the outage streak, so a run of failures could
carry across them and trip on stale history. A reused success now resets the
streak the way a paid success does.

The cost model charged sweep overhead per cell, which credited a weekly
generation for shrinking work it still performs: it pays one arm instead of
three but builds exactly the same graphs. Overhead is charged per SHA now.
Weekly is 5.20h rather than the 4.80h the per-cell rate reported; cold is
10.86h. The residual still cannot be split between per-SHA and per-cell work
from one artifact, so session_durations.json records that assumption and the
direction it errs in, rather than leaving a number nobody can trace.

Clone-template reuse - the branch's core speedup, taken on essentially every
multi-cell sweep - now has a test that builds a real sanitized template, asserts
the cell runs against the copy with the template's HEAD, and fails if run_cell
re-clones. A second test asserting only on a namespace built inside the test was
written and deleted: it exercised nothing, which is the failure this review
round penalised elsewhere.

589 eval tests pass, ruff clean, 29 workflow contract tests pass. The two
test_model_gateway.py failures are pre-existing and fail on main.

* refactor(eval): consolidate duplicated harness logic after the review round

Simplification pass over the branch. Behavior-preserving throughout; three
reviewers, nine findings applied, two skipped.

The review-artifact block in _run_hidden_oracle was unreachable. That function
runs only in run_arm's non-review branch, while the directory it probes for is
created only in the review branch, and each sandbox serves exactly one arm - so
`review_artifact.parent.is_dir()` could never be true. It was added an hour
earlier to make the hidden oracle resolve the moved artifact; the oracle never
runs for review tasks, so the guard was dead on arrival. Deleting it also
removes the duplication it had with the verify-command wiring.

EXCLUDED_ERROR_KINDS is now one definition. runner.py and comparator_reuse.py
each carried the same six-member frozenset, kept in sync by a comment. Only one
direction is possible: runner already imports from comparator_reuse, so the
reverse import fails at module-init with a circular-import error. That is now
stated where the alias lives, so nobody tries it the other way.

ensure_task_graph and prefetch_next_graph shared ten keyword parameters, passed
through two call sites and forwarded whole between them. They now take a
GraphBuildEnv, mirroring TaskCellContext, which already bundles per-cell state
in this file. Its ready_keys() replaces an inline four-set union at the call
site.

Smaller consolidations: _sha256_file's hand-rolled chunk loop becomes
hashlib.file_digest (3.11+, already used in runner_artifacts); _copy_owner_only
reuses task_assets._write_all and COPY_CHUNK_BYTES instead of repeating the
short-write retry; its stat-then-open existence check becomes the O_EXCL failure
it was already relying on, which is atomic rather than merely narrow; and
runner_environment reads the digest through comparator_reuse.current_runtime_digest
instead of re-parsing the environment variable.

Three test docstrings summarised the branch's own history ("the branch's core
speedup", "the regression that produced fifteen runs") rather than the invariant
under test. Rewritten to state the constraint, which is what survives the merge.
Repaired the indentation left behind by the outage-streak edit and flattened the
prefetch dispatch from three nested conditionals to one.

Skipped: consolidating comparator_reuse._real_directory onto proposer_sandbox's
same-named helper - they differ, the sandbox one rejects any symlink in the
resolved path while this one checks only the leaf, so sharing it would tighten
behavior rather than preserve it. That needs a decision about which policy the
reuse path wants, not a simplification.

589 eval tests pass, ruff clean, 29 workflow contract tests pass. Unrelated and
pre-existing: two test_model_gateway.py failures, and
test_process_control.py::test_timeout_kills_term_ignoring_descendants_before_they_write,
which is a TERM-to-KILL timing flake (passes 2 of 3 in isolation) in a file this
branch does not touch.

* refactor(eval): name the reuse directory check for the promise it makes

The simplification pass left one finding open: comparator_reuse and
proposer_sandbox both defined `_real_directory`, same name and same shape, with
different guarantees. The sandbox one rejects every symlink hop in the path; the
reuse one checks only the leaf and resolves through parents. Sharing the name
invites a consolidation that would silently tighten one of them.

They should not be merged, so the name stops claiming they could be.
proposer_sandbox guards a mount root, where a symlink hop changes what an
untrusted session is handed. comparator_reuse guards a data directory whose
contents are already validated one file at a time - reads go through
_regular_file, which lstats and rejects symlinks, and writes through O_NOFOLLOW.
A symlinked parent therefore grants nothing those guards do not already cover,
while refusing one would reject a symlinked artifacts directory or macOS's /var
for no gain.

Renamed to _resolved_directory, with the reasoning recorded at the definition,
and a test that pins both halves: a symlinked parent is accepted and resolved, a
symlinked leaf is still refused. Behavior is unchanged.

591 eval tests pass, ruff clean. The two test_model_gateway.py failures are
pre-existing and fail on main.

* test(eval): measure the sweep scheduler instead of modelling it

measure_evolution_cost predicts wall clock from a model of what
sweep_task_cells does. This runs the real thing - real threads, the real wave
barrier, the real outage breaker - with only the paid agent session replaced by
a sleep, and times it.

Durations are the measured per-arm samples divided by 5000, so a 1416s cell
takes ~0.28s. The shape is kept on purpose: the median cell is 826s against a
5400s ceiling, and that spread is the entire reason a barrier costs anything.
Uniform random sleeps would erase the effect under test. All schedulers consume
one identical seeded plan, so a comparison cannot be an artifact of one of them
drawing luckier cells.

The model survives contact: it tracks real execution within about 10%, and
workers=1 - which runs without a pool at all - sits at 0.95, so the residual
above 1.0 at higher worker counts is per-wave thread overhead rather than a
modelling error. Two structural claims that were arithmetic are now observed.
Weekly is flat from workers=3: 3.59, 3.59, 3.59, 3.60, 3.59, 3.59 across w=3..8.
workers=4 buys nothing over workers=3 on cold, 7.68 against 7.78.

Two prototype schedulers are measured beside it, deliberately before any
production code exists. A continuously fed pool per task is worth more than the
model claimed on cold, -27.3% against a predicted -17.9%, and exactly nothing on
weekly, +0.0%, because a weekly task is one wave with nothing to feed. One pool
across all tasks beats both: -40.7% weekly and -42.9% cold at workers=3, rising
to -65.7% and -63.9% at workers=8. It also subsumes the fed pool, since packing
across tasks is a fed pool.

That reorders the backlog. Cross-task packing moves from second to first: it
dominates on both profiles, and it is the only thing that moves weekly at all.
Raising the worker count is worth nothing until it lands - under the barrier
weekly does not improve from w=3 to w=8, and speedup against serial is 1.58x for
three workers and only 2.40x for eight.

The bound on all of it: sleeping threads do not contend. Real sandboxed sessions
compete for CPU, page cache and disk, and the duration sample was itself
measured at workers=1, so it carries no contention either. These speedups are
upper bounds. The ordering is trustworthy because the schedulers were compared
under identical conditions; the magnitudes are not. The packed prototype is also
a bare ThreadPoolExecutor with no breaker folding, no per-task graph lifecycle
and no reuse binding - which is the actual cost of building it, and is not
measured here.

* test(eval): carry the sweep invariants into the packed prototype

The first packed prototype was a bare ThreadPoolExecutor. It reported -43% and
none of the invariants the shipped scheduler holds, so it priced an idea nobody
could ship. This one carries them: a global submission order continued across
task boundaries, in-order folding, the real outage breaker, and per-task graph
readiness gating behind a serial builder.

The fidelity check first reported the two schedulers tripping on different
cells, 17 against 16. That was my instrumentation, not a divergence -
sweep_task_cells folds an entire wave before it evaluates the breaker, so the
last cell folded is not the cell that tripped. With the harness mirroring the
breaker's own evaluation the two agree exactly, across failures starting at
cell 0, 4 and 12, with overrun inside the workers-1 bound the wave docstring
promises.

Two results worth the exercise.

Head-of-line blocking, not the barrier, is what a naive in-order design pays.
Holding submission to `workers` cells beyond the fold pointer leaves the
faithful scheduler at -8.1% cold and -2.7% weekly: one slow cell stalls the
pointer, the window cannot slide, and it reproduces the wave almost exactly.
That is the number to quote if anyone proposes the obvious implementation.

But the overrun bound turns out to be set by the worker count, not the window.
Only `workers` cells can be running when the breaker trips; everything queued
behind them short-circuits on the halt flag. Overrun is 3 at an unbounded
window exactly as at 6, and the trip cell never moves off 16. So H2 does not
have to trade breaker fidelity for speed - a wide window takes -42% with the
semantics intact. The tension I assumed was there is not, and window=12 already
captures 97% of it.

Still an upper bound: sleeping threads do not contend, and the sample was
measured at workers=1. What this establishes is that the invariants are
affordable, which was the thing blocking H2. Not built here: the trees tempdir
lifecycle, reuse-row binding, and the cancel_event path.

591 eval tests pass, ruff clean.

* test(eval): put the scheduler comparison under real CPU contention

Every Phase 2 number so far came from sleeping threads, which contend for
nothing, against a duration sample measured at workers=1, which contains no
contention either. That was the standing caveat on the whole result, so this
measures it.

A cell now waits for its API share and then burns a fixed number of sha256
rounds in a subprocess. Work-bounded rather than wall-clock bounded, so it takes
longer when cores are busy - that is the effect under test. A subprocess because
Python threads burning Python would measure the GIL rather than the machine.
Calibrated at 519k rounds/s, stable within 2% across three probes.

The first run of this was worthless and is recorded as such: on a 24-core host
with 3 to 6 workers nothing ever contends, since cpu_fraction 0.5 at 6 workers
is about 3 cores of demand out of 24. It measured an absence. Re-run pinned with
taskset to 4 and 2 cores.

The packing advantage survives. It holds between -40% and -47% across every host
size and CPU fraction tested, including a genuinely oversubscribed 2-core box at
cpu_fraction 0.5 with 6 workers.

But contention erodes packing more than it erodes waves, for a structural
reason: packing is what creates the concurrency. Moving from 24 cores to 2 at
cpu 0.5 and 6 workers, the faithful scheduler slows 13% while the wave slows
3.7%, and the gain narrows from 45.0% to 39.8%. Packing and a higher worker
count are therefore not independent wins - packing spends the contention
headroom first, so raising workers has to be re-argued after it lands rather
than added to it.

Three things this still does not measure, and they bound the result. The real
CPU fraction of a benchmark cell is a guess informed by roughly 180 tool calls
per session; nobody has profiled one. The evolution runner's core count decides
which column applies and is unknown here. And the burn is sha256, pure CPU,
while real cells run vitest and analyze, which are memory and IO heavy - so this
is a floor on contention, not a ceiling.

591 eval tests pass, ruff clean.

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

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

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

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

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

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

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

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

595 eval tests pass, ruff clean.

* fix(eval): judge harness health on execution, not on how many tasks resolved

broken_incumbent_arms infers "the environment is broken" from an arm resolving
zero tasks. That inference does not hold: a reviewer can be wrong about every
task in a hard corpus while every process, mount and capture worked perfectly.
Actions run 33962002890 is exactly that shape - 51 cells, all resolved=False
with error_kind=oracle-failed, median score 0.212, and a healthy harness.

Someone already knew this, and patched it by excluding review arms at the call
site. That leaves the unsound inference in place for workflow and
workflow_direct, and leaves review arms with no health check at all - so the
run that genuinely was broken, 33912693948, where the mount made an atomic
write impossible and all 41 artifacts came back empty, could not have been
caught here either.

So this replaces the inference rather than adding another exemption. aggregate
now classifies fresh rows into execution failures (the process or its tooling
did not complete), evidence failures (it completed but produced nothing
trustworthy or scoreable), and admissible measurements. An arm is unhealthy
only when it has fresh attempts, zero admissible measurements, and at least one
execution or evidence failure. Resolution count is no longer consulted. Arms
with only reused rows report current health as UNKNOWN rather than good.

With the inference corrected, review arms are checked again, which is what lets
the empty-artifact case be caught at all.

Deliberately unchanged: comparator reuse eligibility, quality denominators,
promotion thresholds, model settings, skill prompts and scheduler behaviour.
Failures that stop being called infrastructure failures still surface in the
counts and reasons - an agent-originated failure must not vanish from reporting
because it was reclassified. broken_incumbent_arms and its tests are left in
place; deleting behaviour belongs in its own change.

Seven regression tests, built from both runs' shapes and labelled as
reconstructed from logged observations, since 33962002890's results.jsonl did
not survive the instance shutdown. They pin: a badly-scoring reviewer is
healthy; an all-zero score is still a valid negative; empty artifacts are
unhealthy; one admissible cell keeps an arm healthy while its failures stay
visible; reused rows alone leave health unknown; reused successes do not mask
fresh failures; and a parseable artifact does not excuse a failed session.

602 eval tests pass, ruff clean.

* fix(eval): pin the health guard below the breaker, and stop calling mixed runs healthy

Two corrections to the health-classification patch.

The regression I wrote could not have proved what it claimed. A fixture of 41
empty artifacts aborts through the outage breaker long before finalization:
review-evidence-invalid is in SYSTEMIC_ERROR_KINDS and the limit is 5, so it
trips at cell 5 through the pre-existing path. It demonstrated failure
detection, not the new guard. The decisive test now uses ONE fresh unusable
cell, asserts the streak stays under the breaker threshold, and only then
requires finalization to abort - leaving the new check as the only thing that
can catch it. Removing the call makes that test fail; restoring it passes.

The accurate defect statement is narrower than the last message claimed. Review
arms were excluded from the final incumbent-health check while the consecutive-
failure breaker gave them separate, partial coverage. They were not unguarded.

Second: "one admissible cell plus two execution failures" was asserted as
healthy. That converts "not wholly unusable" into "ran reliably", which is how
a partly-broken sweep passes review. Arms now report UNKNOWN, OBSERVED_OK,
DEGRADED or UNUSABLE. Only UNUSABLE is fatal, so eligibility and promotion are
untouched - this changes what is reported, not what is allowed.

The guard is extracted as enforce_measurement_health so it can be driven
directly, and it now reports a status line per arm. It names no cause: an empty
artifact establishes that evidence is unusable, not that a mount rejected the
write, so it prints cause=undetermined rather than guessing EROFS. It still
runs after report.md and promotion.json are written, so a failing sweep leaves
its evidence behind.

ce_review is named explicitly at the call site. It is a comparator rather than
a candidate, so it is absent from CANDIDATE_ARMS.values(), and dropping the
review exclusion alone would have left it unclassified.

The wiring test reads _run_sweep's compiled code object for the referenced
global rather than matching source text. It is honest about its limit: it
proves the call exists and would catch its removal, but no test here drives
_run_sweep end to end, which needs bwrap and a sandbox.

broken_incumbent_arms is marked LEGACY and NON-AUTHORITATIVE with removal
tracked. It has no production caller.

608 eval tests pass, ruff clean. The two test_model_gateway.py failures are
test_locked_litellm_translates_messages_to_offline_responses and
test_openai_gateway_never_leaves_proxy_output_on_an_undrained_pipe; both fail
identically on origin/main in this environment, checked directly rather than
carried forward as an inherited label.

* 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.

* fix(eval): address PR review feedback (#3207)

- aggregate: count admissible rows directly instead of subtracting the
  execution and evidence counters, which double-charged a row that is both
  a session error and invalid review evidence and could report UNUSABLE for
  an arm holding real measurements.
- run_proposer: bound the session timeout by what is left of
  --max-runtime-seconds, so clearing the sweep minimum cannot start a
  full-length session past the instance window.
- comparator reuse: hold one O_NOFOLLOW descriptor for the size check,
  digest and copy, and prove it is the inode that was checked, closing the
  swap window a concurrent writer of the reuse directory had.
- Drive the review-artifact mount assertion through run_arm and the
  clone-template assertion through run_cell, instead of rebuilding the
  expected values in the tests (also removes the CodeQL unnecessary lambda).
- Assert the workflow invokes run-evolution.sh rather than that its YAML
  mentions --max-runtime-seconds, which only appears in a comment.
- Correct the parse_review_output failure-mode claim: the fold was empty
  artifacts reported as "not valid UTF-8 JSON"; a never-created file raised
  FileNotFoundError.
- prettier: wrap the over-long readFileSync call flagged by PR autofix.

Note: pre-existing failure in tests/test_model_gateway.py::test_locked_litellm_translates_messages_to_offline_responses (local LiteLLM proxy never becomes ready in this environment) not addressed by this PR.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>

* fix(eval): address the second round of PR review feedback (#3207)

- Refuse a symlinked `transcripts` component on both sides of comparator
  reuse. O_NOFOLLOW guards the leaf only, so a link there redirected the
  read or the copy out of the results directory; checked per component as
  evolution._require_directory_chain does.
- Base the paid incumbent canary on the cells this sweep PLANS. Reuse
  selection accepts any prior run index, so a results directory produced
  with more runs left extra keys, the equality never held, and the canary
  stopped firing. Extracted as drop_canary_reuse_key and unit-tested.
- Start the runtime clock in main(). --max-runtime-seconds is measured from
  /proc/uptime before exec, so parsing, task I/O, preflight and gateway
  setup were being handed back to the sweep out of the upload reserve.
- Do not fall back to shutil.copytree when the managed clone copy was
  cancelled or timed out; that fallback is for a filesystem that cannot
  reflink, and copytree cannot be cancelled.
- Assert the review session's writable mount, not only the verifier's
  read-only one: the EROFS bug is about the agent's write.
- Exercise ref isolation in the copy_isolated_tree test rather than
  comparing an initial HEAD a shared namespace would also match.
- Point the stale-symlink fixture at the sentinel via os.path.relpath, and
  skip the reuse symlink tests where symlink creation needs privilege.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>

* fix(eval): close the runtime-cap gap and pin the reuse directory

Both were left open on #3207 as approach decisions rather than nits.

Runtime cap: run-evolution.sh computed the budget in its own
`uv run python -c` and passed a number, so the script's remaining
provenance work and the CLI's own startup were spent by nobody and charged
to the sweep — out of the upload reserve the cap exists to protect. The
script now passes --max-runtime-from-instance-window and evolve reads
/proc/uptime itself, on the line after it starts the clock the budget is
measured against, so no interval exists to lose. Also removes an
interpreter start from the script and lets --dry-run print the real argv.

Reuse directory: _real_child_directory lstat-checked `transcripts` and
returned its pathname, so a concurrent writer could rename the directory
and leave a symlink before the name was used again — O_NOFOLLOW guards
only the leaf. Every artifact is now resolved against a held descriptor:
_open_real_directory opens with O_DIRECTORY|O_NOFOLLOW (check and open in
one syscall), and _open_regular / _copy_owner_only take dir_fd. The reuse
path is therefore POSIX-only; _require_openat says so and fails closed,
which the runner already treats as "run a paid cell". _resolved_directory
still tolerates a symlinked reuse root, unchanged and still tested.

evolution._require_directory_chain is still lstat-per-component. It guards
a different surface (candidate overlay reads) that neither review raised,
so it is left alone rather than widened into here.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>

* fix(eval): sample the proposer budget where it is spent, digest what is copied

Four findings against f0cdc9e7, all of them mine.

- run_proposer took a precomputed remaining_seconds, but it clones,
  sanitizes and builds a sandbox before the session starts, so the caller's
  reading was already stale. It takes started_monotonic now and samples the
  budget on the last line before run_claude. The caller's earlier reading
  still decides whether to start at all — it just no longer decides how
  long to allow.
- _copy_transcript_artifact hashed the source and then read it again to
  copy it. A held descriptor stops the pathname being substituted, not the
  inode being rewritten, so the row could record the expected digest while
  the destination held other bytes. _copy_owner_only now digests the same
  buffers it writes and returns (digest, bytes); a mismatch unlinks the
  destination and raises. One read instead of two.
- Explain the empty except in _open_real_directory: an existing directory
  is the ordinary case, and the O_DIRECTORY|O_NOFOLLOW open below is what
  proves what it is (CodeQL).
- Drop the unused uptime fixture from the runtime-cap test, and correct the
  cross-file contract described in the workflow-contract test and the
  workflow YAML: neither names --max-runtime-from-instance-window, the
  script does.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>

* test(eval): close the reuse producer/consumer contract through the real run_cell

Every comparator-reuse test built its rows by hand. That proves the predicate's
logic and nothing about the producer: a fixture can satisfy eligibility while a
row the runner actually emits never does, and a key-name inventory cannot tell
the difference. I checked that inventory first - all 25 keys the predicate reads
have a producer - which is exactly why it was not sufficient evidence.

This carries one record through the production path instead:

    real run_cell -> production JSONL writer -> load_result_rows
        -> row_is_reusable_comparator

Only the expensive dependencies are replaced: the model session, sandbox launch,
repository acquisition, graph preparation, git plumbing. The digest fields the
reuse binding compares are assembled by run_cell itself from its
TaskCellContext, so they stay real - they are the subject, not scaffolding.
Expectations are built from the sweep's own configuration rather than copied out
of the emitted row, since copying them back would make producer and consumer
agree because the test arranged it.

Two cases: a production-emitted row with matching bindings is eligible, and the
same evidence with one dependency binding changed is rejected. A test-controlled
digest set keeps both deterministic.

Mutation-checked. Replacing run_cell's task_asset_manifest_digest assignment
with None makes the positive case fail; restoring it passes. That is evidence
the inventory audit could not produce.

Building it also documented what a real review row must carry, which no fixture
had recorded: a scored review with review_weighted_f1 present, and at least one
transcript artifact whose source is PARENT_EVENT_STREAM_SOURCE. An empty
artifact list is not reusable. Those are production contracts my first double
got wrong, and the reader rejected it each time.

No production code changed. 607 eval tests pass, ruff clean.

* test(eval): drive the real sweep to its finalization decision

enforce_measurement_health was unit-tested and its call site pinned by reading
_run_sweep's compiled code object. Neither showed the guard running inside a
sweep. These drive the real _run_sweep with cell execution scripted and
everything downstream left alone: folding, aggregation, the artifact writers,
the health guard and the exit selection.

The below-breaker case is the decisive one. A fixture of many unusable cells
aborts through the pre-existing outage breaker instead - review-evidence-invalid
is systemic with a limit of five - and would pass whether or not the
finalization guard exists. One fresh unusable cell stays under that threshold,
so only the guard can catch it. The test asserts the arm and status the guard
reports, and that no systemic-outage line was printed, rather than accepting any
SystemExit: an unrelated setup failure must not satisfy it.

Mutation-checked at the runtime path. Removing the guard invocation fails the
below-breaker test because the expected finalization behaviour disappears, not
because a name went missing from a code object.

A zero score is covered separately. review_weighted_f1 = 0.0 must stay a present
valid negative measurement; a truthiness check would read it as absent and turn
a quality result into an execution-health failure. The third test asserts the
evidence survives - results.jsonl carries the scored row and report.md exists -
so the persisted artifacts tell the same story as the exit.

Only expensive setup is replaced: cell execution, graph preparation, asset
snapshots, and task-binding resolution, which clones the repository and verifies
the ref. Building the fixture also documented that the report renders the whole
review metric set, so an incomplete row fails in formatting rather than logic.

Still open on this track: interruption semantics and exit-reason precedence
(outage 1 before cancellation 130) are not yet exercised, and fold order and the
real sandbox mount contract remain on separate tracks.

No production code changed. 610 eval tests pass, ruff clean. The two
test_model_gateway.py failures reproduce on origin/main in this environment.

* fix(eval): an interrupted sweep is interrupted, not aborted

Driving the real _run_sweep to its exit revealed that cancellation without an
outage exits 1 and writes "Sweep aborted: partial evidence", where the contract
is 130 and "Sweep cancelled".

sweep_task_cells returns (streak, tripped), and its caller assigns that flag to
outage_tripped and turns it into exit 1. Both cancellation paths returned True
for it. The breaker's own return is also True, so the two became
indistinguishable one frame up and cancellation inherited the outage's exit and
wording. The fix is to return False from the cancellation paths: the flag means
the breaker tripped, and the caller already tests cancel_event itself for the
stop decision, so nothing stops running any later than before.

Reproduced before the fix, through the real caller, not from reading. The
failing assertion was the report line; the exit code was 1.

Two runtime tests cover it. Each begins with one admissible cell so the arm
classifies DEGRADED rather than UNUSABLE - otherwise enforce_measurement_health
supplies exit 1 first and a precedence test passes without ever reaching exit
selection. Cancellation is set from a completed cell rather than a sleep or a
real signal, so the interruption point is deterministic.

The precedence test is mutation-checked: moving the cancel_event check ahead of
the outage check fails it with "assert 130 == 1" - it observes the wrong exit
code, not merely some failure. My first attempt at that mutation silently
matched nothing and the suite passed; a no-op mutation proves nothing, so the
edit now asserts its own anchor.

test_process_control read the same flag as "stopped" and asserted it after a
cancellation. It now reports the event and the flag separately, which is what it
was really asserting: cancelled, and not an outage.

The scored-row shape moved into tests/bench_fixtures.py with zero values
written out, so the finalization and interruption tests share one definition
of what a real review row carries.

Impact analysis returned risk UNKNOWN - the index predates this branch and does
not carry the eval harness - so the callers were confirmed by text search as the
rules require for UNKNOWN: one production caller and three test sites, all
updated or verified. detect_changes reports zero for the same reason; that zero
is unseen, not unaffected.

612 eval tests pass, ruff clean. The two test_model_gateway.py failures
reproduce on origin/main in this environment.

* test(eval): prove the review artifact mount under a real sandbox

The orchestration tests replace the sandbox, so they say nothing about
isolation. This covers the filesystem contract the EROFS defect actually broke,
through the production configuration: the same command_prefix_for call run_arm
makes for a review cell, with read_only_workspace=True and the review-output
directory as the writable mount - not a hand-built mount tuple that merely looks
right.

A deterministic writer stands in for the agent. It writes a temp file beside the
destination and renames it into place, which is the operation that failed: an
atomic write needs a WRITABLE PARENT DIRECTORY, and binding the file itself left
nowhere to put the temp file. It then attempts a workspace write and must be
refused, and the production parse_review_output reads the bytes the sandbox left
behind. No model session and no credentials.

Placed in test_proposer_sandbox.py, which the "eval / containment (ubuntu)" job
already runs with GITNEXUS_REQUIRE_BWRAP_CANARY=1. That gate is the point: where
the variable is set, a missing namespace capability FAILS the job instead of
skipping into a green tick. Verified here - forcing the variable on this machine
fails with "bwrap: No permissions to create new namespace" rather than skipping.

What this does not establish: the assertions have never executed. This machine
cannot create user namespaces, so the test stops at the preflight. Imports,
signatures and the payload were checked statically instead - parse_review_output
returns a tuple rather than an object, and it rejects a body without "verdict",
so both of my first attempts were wrong and are fixed. The first CI run on a
namespace-capable runner is what will actually confirm it.

Scope is the filesystem and process contract only. A stand-in writer does not
establish that a particular agent CLI's own file-access policy permits the same
operation; that is a second, independent gate.

40 passed, 10 skipped locally; ruff clean.

* fix(eval): read the review artifact before the sandbox deletes it

The containment (ubuntu) job has been red since before the mount canary was
added, and both failures have the same cause.

This branch moved the review artifact out of the clone and into the session's
private root, which is the point of the change: the agent writes atomically, so
the artifact needs a writable parent DIRECTORY outside the read-only workspace.
prepare_sandbox removes that private root in a finally on scope exit. Two tests
read the artifact AFTER the with block, so they were asserting against a
directory the sandbox had already deleted - FileNotFoundError, reported as
"review output was never written".

Production is not affected, and this is the reason: run_arm holds the live
session and reads the artifact inside that scope, both to score it and to mount
it read-only into the verifier's separate sandbox invocation. The tests were the
only readers outside it.

Both now read inside the scope. The pre-existing canary
(test_read_only_review_workspace_exposes_only_one_writable_artifact) predates
this PR and passes on main, where the artifact still lived in the clone and
survived teardown; the relocation is what broke it, so the fix belongs here.

The new canary found this independently and agreed on the cause, which is what
it was written for - though only after CI ran it, since this machine cannot
create user namespaces.

40 passed, 10 skipped locally; ruff clean. The bwrap-gated tests still skip here
and remain unverified until the ubuntu job runs them.

* test(eval): pin what an interrupted sweep persists and refuses to promote

The completed-run persistence test could not show either of these: it never
interrupts, so it would pass even if the writers only ran on the clean path.

Evidence already paid for survives cancellation. The cell that completed before
the interruption is still in results.jsonl with its measurement intact, and
report.md is still written. Losing those rows would mean paying for evidence the
sweep then discards.

An interrupted run emits nothing that authorizes promotion. Asserted as the
semantic condition rather than the absence of a file: promotion.json IS still
written for an aborted run - it is the record of why nothing was promoted - so
the test requires run_status "aborted" and every decision reduced to
insufficient_evidence with a partial-evidence reason, rather than requiring the
artifact to disappear.

Mutation-checked. Forcing complete=True at the promotion_evidence call site
fails the test with "assert 'complete' == 'aborted'" - it observes an aborted
run claiming completeness, not merely some failure.

These were the last two unasserted items on this PR's finalization checklist.

614 eval tests pass, ruff clean. The two test_model_gateway.py failures are
environmental (litellm[proxy] console script absent) and predate this branch.

* Address PR review feedback (#3207)

An exhausted runtime cap no longer buys a second. remaining_runtime_seconds
floors at 0 and the proposer call site wrapped it in max(1, ...), so a cap fully
spent by the clone, the sanitize pass and the sandbox build started a paid
session with a one-second allowance instead of stopping before the upload
reserve the cap exists to protect. It now returns a not-ok record, which the
caller already treats as end-of-run. Pinned by a test driving the real
run_proposer with setup that consumes the whole budget; reverting to max(1, ...)
fails it.

Reuse copies are bounded before their size is validated, not after. The source
is a prior sweep directory this module already treats as concurrently writable,
so a transcript appended to after its metadata was recorded was streamed to EOF
and only then compared against its declared size - filling the destination, or
never reaching EOF, long before the drift check could reject it. _copy_owner_only
now takes max_bytes and stops one byte past the ceiling, which keeps the drift
comparison meaningful. Review and patch artifacts carry no recorded size, but
"no recorded size" is not "no limit": they get MAX_TRANSCRIPT_BYTES, the ceiling
the capture path already enforces.

A reused review row must carry its artifact. The predicate accepted a row on its
score and transcript metadata while materialize_reused_row copied the review
artifact only when the row named one, so a scored review could be carried
forward with nothing for a proposer to read. The shared row fixture was the
thing out of step here, not the requirement - production sets review_artifact
whenever the review source exists - so it now carries one too.

Test fixes, all against code this PR added:

unusable_review_row could not be overridden at all. It passed explicit keywords
beside **overrides, and Python rejects the duplicate in the call expression
before scored_review_row can apply its update, so unusable_review_row(error_kind=...)
raised TypeError. Merged into one mapping.

The finalization helper monkeypatched runner.prepare_ce_plugin_snapshot with
raising=False. No such symbol exists - the real one is staged_ce_plugin_snapshot
- so it silently added an attribute nothing reads. Removed.

The promotion test named candidate_review in candidate_arms but _run_sweep builds
cells only from args.arms, so no candidate cell ran and insufficient_evidence
could hold because nothing executed rather than because partial evidence is
barred. The candidate arm now runs, cancellation fires after its cell, and the
test asserts the candidate actually produced a row so it cannot silently return
to being vacuous.

The reuse round trip serialized with json.dumps and write_text while claiming to
cover the production writer, which applies redact_text over the row's own bytes.
It now writes the way the sweep does. Its fixture also writes the review
artifact, because run_cell records review_artifact only when the review source
exists and the emitted row was otherwise one production never emits.

Not addressing the duplicate-match nit on proposer_sandbox.py: the claim is that
"/review-output" occurs once in "/review-output/review-output.json". It occurs
twice, at offsets 0 and 14 - the separator before the basename forms it again -
so the boundary the comment describes is real. Verified with re.finditer.

The workspace-snapshot finding is parked for a human: excluding bootstrap noise
is a documented deliberate choice, and tightening it is a genuine tradeoff.

634 eval tests pass, ruff clean. The two test_model_gateway.py failures are
environmental (litellm[proxy] console script absent) and predate this branch.

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

Carry review_f1 in the scored-row fixture. score_review emits "f1"
(review_scoring.py:317), which the runner folds in as review_f1, so a real
scored row has it and the fixture did not - the same fidelity gap as the metrics
already added there. aggregate now reports 0.5 for it instead of None.

The reported failure mode was not real, and the distinction matters for anyone
reading the thread later. aggregate filters on record.get(metric) is not None
BEFORE indexing record[metric], so a missing key is skipped rather than raising
KeyError; the finalization tests were passing throughout. Verified by running
aggregate against the old fixture. Fixed because the fixture should match what
production emits, not because anything was crashing.

Drop the unused row parameter from the round-trip _expectation helper. It never
read the argument - deliberately, since the docstring says the bindings must be
derived from sweep configuration rather than copied out of the emitted row - but
passing the row anyway was dead plumbing that suggested the opposite.

The reuse-root TOCTOU finding is parked for a human rather than fixed: the
lstat-then-resolve window is real, but _resolved_directory documents the weaker
promise as deliberate and says not to merge it with proposer_sandbox's helper
"without first deciding which promise the reuse path should make". That decision
is the fix, and it is the same class as the reuse TOCTOU already parked on this
PR.

634 eval tests pass, ruff clean. test_process_control's TERM-ignoring-descendant
test flaked once under full-suite load and passes 3/3 in isolation; it is a
timing test and neither file changed here touches it. The two
test_model_gateway.py failures remain environmental.

* fix(eval): pin the reuse roots to the directory that was checked

Settles the reuse TOCTOU that has been parked twice on this PR. The docstring
asked to decide which promise the reuse path makes before merging its helper
with proposer_sandbox's, and the answer is that two separate things were being
conflated.

The symlink POLICY stays exactly as it was: parent hops remain allowed, so a
symlinked artifacts directory or macOS's /var still works, and a symlinked leaf
is still refused. Rejecting hops would break ordinary setups for no gain, which
is what the docstring argued and it is still right.

What is closed is the other thing - the gap between checking a name and using
it. _resolved_directory lstats a name and the later open re-walks that same
name, so a prior sweep that renames its results root and leaves something else
behind is opened somewhere else entirely, and O_NOFOLLOW cannot see a link that
resolve() already followed. _resolved_directory now returns the checked
directory's identity alongside its path, and _open_pinned_root fstats the
descriptor it opened and refuses a mismatch.

The two are separable, so there was no trade to make: comparing identity
rejects nothing that holds still, since a stable directory always matches
itself. Both cases are pinned - a replacement by a different REAL directory is
refused (the leaf-symlink rule does not cover that one), and an ordinary
unchanged root opens normally.

Why it matters here rather than as a general hardening: the failure is silent.
Rows would be copied out of some other directory and folded into a comparator
baseline as though they were this sweep's own evidence, which is the one thing
the reuse path exists to get right - a corrupted baseline decides promotions.

636 eval tests pass, ruff clean. The two test_model_gateway.py failures remain
environmental.

* refactor(eval): one prompt digest, and drop two pieces of dead scaffolding

Compute task_prompt_digest once. row_is_reusable_comparator compares the value a
prior row stored against the value this sweep derives, as exact strings, and the
two inline copies of that hash had already drifted apart - the expectation side
had picked up a str() cast the row side lacked. They agree today only because
tasks validate "prompt" as a string (runner_tasks.required_strings), so nothing
was broken; a later divergence on one side would have silently stopped rows
matching, with no test to catch it. Verified the extracted helper is
byte-identical to what it replaced.

Two comments named broken_incumbent_arms as the thing that would read stale
health. This branch replaced that call with the measurement-health path, so the
reasoning still holds but the name no longer does; they now name arm_health.
The function itself stays: origin/main still calls it at runner.py:2098, and
retiring it is its own change rather than a cleanup pass.

Removed instance_window_budget_from_proc and its one test. It composes two
helpers main() deliberately calls separately - there is a comment there
explaining why the read and the budget calculation stay decoupled - and nothing
outside its own test ever called it. Introduced on this branch, absent from
main, so it was never deployed, public, or consumed elsewhere.

Dropped an unused tmp_path parameter from a comparator-reuse test that does no
filesystem work.

Skipped three findings. Hoisting _assert_self_contained_git_objects to a
once-per-template check is a real saving - measured at 5.66s over 1551 objects -
but it verifies the OUTPUT of each copy operation, and git clone from a local
path hardlinks objects by default, which is exactly what its st_nlink check
catches. Checking a predecessor instead of the clone each cell runs against
thins an isolation guarantee for 0.7% of a median cell. Deleting
broken_incumbent_arms outright would remove a symbol main still calls. And a
fourth copy of the test-only _git helper extends a pattern that already exists
in three other test modules; centralising it means editing conftest.py, outside
this scope.

635 eval tests pass, ruff clean. The two test_model_gateway.py failures are the
environmental ones.

---------

Co-authored-by: Gergo Magyar <gergomagyar0@gmail.com>
Co-authored-by: Cursor <cursoragent@cursor.com>
Co-authored-by: Claude Opus 5 (1M context) <noreply@anthropic.com>
2026-09-08 09:25:45 +00:00

2726 lines
119 KiB
Python
Raw Blame History

This file contains ambiguous Unicode characters

This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.

"""Benchmark the gitnexus-plan/work workflow against a baseline agent.
Usage:
uv run --locked --extra dev python -m workflow_bench.runner \
--tasks workflow_bench/tasks.scenarios.yaml --runs 3 \
--model claude-sonnet-4-20250514
Each task runs in a fresh self-contained clone of the target repo, once per
arm per run:
* ``workflow`` — two headless Claude Code sessions: gitnexus-plan, then
gitnexus-work on the produced plan.
* ``candidate_workflow`` / ``candidate_workflow_direct`` — the matching
workflow arm with a prompt-only candidate overlay committed in its clone.
* ``baseline`` — one headless session with the same task text and the Skill
tool disallowed (so it cannot borrow the workflow), everything else equal.
Token usage, cost, duration, and turn counts come from the CLI's own
``--output-format json`` report — nothing is estimated. Caveat: the report's
top-level ``usage`` counts ONLY the main-loop session; ``total_cost_usd`` is
the only reported number that includes subagent spend. A task's model-visible
``verify`` command is retained as an authored-test quality signal; ``resolved``
also requires its harness-owned hidden behavioral oracle. Token savings on
unresolved runs are reported but flagged, because saving tokens by failing is
not a saving.
Trust model: task files and candidate prompts are executable input. Every
setup, verifier, and model session runs inside a preflighted Linux Bubblewrap
boundary with an allowlisted environment, isolated home, PID namespace,
self-contained clone, and task-declared read-only dependencies. Unsupported
or unavailable containment fails before model invocation (README § Trust
model).
"""
from __future__ import annotations
import argparse
from concurrent.futures import ThreadPoolExecutor, wait
import hashlib
import json
import os
import re
import secrets
import stat
import statistics
import sys
import tempfile
import threading
import time
from collections.abc import Callable, Mapping, Sequence
from functools import partial
from contextvars import copy_context
from dataclasses import dataclass, replace
from datetime import UTC, datetime, timedelta
from pathlib import Path
from typing import Any
import yaml
from .comparator_reuse import (
REUSE_EXCLUDED_ERROR_KINDS,
CellKey,
ComparatorReuseExpectation,
TaskReuseBinding,
current_runtime_digest,
default_reuse_max_age,
load_result_rows,
materialize_reused_row,
select_reusable_comparator_rows,
)
from .evolution import (
CANDIDATE_ARMS,
EVIDENCE_MAX_AGE_DAYS,
EVALUATED_ARM_SKILLS,
PROMOTION_METRICS,
apply_candidate_overlay,
candidate_overlay_digest,
promotion_policy,
promotion_evidence,
required_candidate_arms,
seed_evaluated_skills,
skill_fingerprint,
)
from .model_gateway import (
attach_openai_gateway,
anthropic_api_key_from_environ,
credential_secrets,
model_session_environment,
openai_api_key_from_environ,
)
from .oracle_assets import (
ORACLE_ENV_VAR,
TaskOracleSnapshot,
capture_task_oracles,
require_hidden_harness_absent,
sanitize_clone_for_hidden_oracles,
staged_task_oracle,
with_hidden_harness_apply_exclude,
)
from .process_control import cancellation_scope, ManagedProcessError
from .promotion_apply import committed_destination_base_digests
from .review_scoring import (
REVIEW_OUTPUT,
REVIEW_OUTPUT_ENV_VAR,
expected_findings,
parse_review_output,
score_review,
)
from .proposer_sandbox import (
SANDBOX_GITNEXUS as SANDBOX_GITNEXUS,
SANDBOX_GITNEXUS_REGISTRY,
SANDBOX_GITNEXUS_SHARED as SANDBOX_GITNEXUS_SHARED,
SANDBOX_NODE as SANDBOX_NODE,
SANDBOX_REVIEW_OUTPUT,
SANDBOX_WORKSPACE,
ReadOnlyMount,
SandboxError,
SandboxSession,
build_sandbox_environment,
preflight_bubblewrap,
preflight_unsafe_host,
prepare_sandbox,
prepare_review_workspace,
redact_text,
review_output_path,
require_claude_sandbox_helpers,
sandbox_workspace_write_boundary,
)
from .runner_artifacts import (
IMPLEMENTATION_ARMS,
MAX_PATCH_BYTES as MAX_PATCH_BYTES,
MAX_WORKSPACE_SNAPSHOT_ENTRIES as MAX_WORKSPACE_SNAPSHOT_ENTRIES,
MAX_WORKSPACE_SNAPSHOT_FILE_BYTES as MAX_WORKSPACE_SNAPSHOT_FILE_BYTES,
MAX_WORKSPACE_SNAPSHOT_PATH_BYTES as MAX_WORKSPACE_SNAPSHOT_PATH_BYTES,
_bounded_regular_bytes,
_prepare_untracked_for_diff,
_sandbox_git,
capture_patch,
diff_churn,
enforce_phase_workspace,
enforce_work_evidence,
implementation_diff_digest,
copy_isolated_tree,
make_worktree,
new_plan_doc,
parse_shortstat as parse_shortstat,
remove_clone,
require_skill_fingerprint,
run_verify,
snapshot_plan_docs,
VerificationResult,
workspace_snapshot,
)
from .runner_sessions import (
BUILTIN_AGENT_TOOLS as BUILTIN_AGENT_TOOLS,
GITNEXUS_MUTATING_TOOLS as GITNEXUS_MUTATING_TOOLS,
GITNEXUS_READ_ONLY_TOOLS as GITNEXUS_READ_ONLY_TOOLS,
MAX_TRANSCRIPT_BYTES as MAX_TRANSCRIPT_BYTES,
SANDBOX_GITNEXUS_ENTRYPOINT as SANDBOX_GITNEXUS_ENTRYPOINT,
SESSION_TIMEOUT_SECONDS,
USAGE_FIELDS,
_na,
allowed_agent_tools,
run_claude,
sandbox_mcp_config,
sum_sessions,
)
from .runner_tasks import (
normalized_model_identifier,
resolve_task_bindings,
select_tasks,
selected_task_bindings as selected_task_bindings,
)
from .sanitized_graph import (
SanitizedGraphSnapshot,
prepare_sanitized_graph,
validate_no_prebuilt_graph_assets,
)
from .runtime_mounts import (
CE_ARMS,
CePluginSnapshot,
HARNESS_ROOT as HARNESS_ROOT,
ce_plugin_dir_for_arm,
ce_plugin_mounts_for_arm,
staged_ce_plugin_snapshot,
trusted_gitnexus_runtime_mounts,
validate_ce_plugin_inputs,
)
from .task_assets import TaskAssetCache, TaskAssetSnapshot, stage_task_assets
PLAN_PROMPT = (
"Use the gitnexus-plan skill for: {task}\n"
"Headless run: make reasonable choices without asking; the plan document "
"is the deliverable."
)
# Appended to every work-arm prompt. In a headless `claude -p` session there
# is no later turn: backgrounded test runs and scheduled wakeups never come
# back, so a session that "waits" for verification ends unverified (observed:
# a work arm backgrounded its slow tests, scheduled three wakeups that never
# fired, and reported done while two tests failed).
HEADLESS_VERIFY = (
" Verification must be observed inside this session: run the typecheck "
"and test commands in the foreground to completion and report their "
"actual output — never background them or wait on scheduled wakeups."
)
WORK_PROMPT = (
"Use the gitnexus-work skill to execute the plan at {plan}.\n"
"Headless run: proceed without asking; report Definition of Done status "
"at the end." + HEADLESS_VERIFY
)
WORK_DIRECT_PROMPT = (
"Use the gitnexus-work skill for: {task}\n"
"Headless run: proceed without asking. The user explicitly declines a "
"separate planning pass — execute in direct mode with the skill's "
"execution discipline." + HEADLESS_VERIFY
)
BASELINE_PROMPT = (
"{task}\n\n"
"Implement the change in this repository and verify it by running the "
"relevant tests. Work autonomously without asking questions."
)
# External-comparator arms: the compound-engineering plugin's plan/work family,
# prompted with the same structure as the gitnexus arms so only the skill
# family differs. The plugin ships user-level, so clones need no repo files.
CE_PLAN_PROMPT = (
"Use the ce-plan skill (compound-engineering plugin) for: {task}\n"
"Headless run: make reasonable choices without asking; the plan document "
"is the deliverable."
)
CE_WORK_PROMPT = (
"Use the ce-work skill (compound-engineering plugin) to execute the plan "
"at {plan}.\n"
"Headless run: proceed without asking; report completion status at the "
"end." + HEADLESS_VERIFY
)
CE_WORK_DIRECT_PROMPT = (
"Use the ce-work skill (compound-engineering plugin) for: {task}\n"
"Headless run: proceed without asking. The user explicitly declines a "
"separate planning pass — execute directly with the skill's execution "
"discipline." + HEADLESS_VERIFY
)
# Review cell: setup applies a historical PR diff, then the model sees a
# read-only checkout. Both arms emit the same strict artifact so quality can be
# scored deterministically against labels that remain hidden until it exits.
# Concatenated, not an f-string: the JSON shape below keeps its braces doubled
# because the finished prompt is .format()-ed with the task text.
REVIEW_OUTPUT_CONTRACT = f"\nWrite {SANDBOX_REVIEW_OUTPUT}/{REVIEW_OUTPUT} " + """as UTF-8 JSON with exactly this shape:
{{"schema_version":1,"verdict":"approve|comment|request_changes","findings":[{{
"id":"unique stable id","severity":"critical|high|medium|low",
"path":"repository-relative changed file","line":1,"end_line":1,
"category":"correctness|security|compatibility|performance|tests|other",
"scenario":"specific failure scenario","evidence":"concrete code/graph evidence",
"recommendation":"bounded fix","blocking":true}}]}}
Use an empty findings list with verdict approve when there are no actionable
defects. Do not emit Markdown and do not edit any other file.
"""
REVIEW_PROMPT = (
"Use the gitnexus-review skill to review the local uncommitted changes "
"in this repository. {task}\n"
"Headless run: proceed without asking; do not post to GitHub or anywhere "
"external." + REVIEW_OUTPUT_CONTRACT
)
CE_REVIEW_PROMPT = (
"Use the ce-code-review skill (compound-engineering plugin) to review "
"the local uncommitted changes in this repository. {task}\n"
"Headless run: proceed without asking; do not post to GitHub or anywhere "
"external." + REVIEW_OUTPUT_CONTRACT
)
# Skill each arm's session(s) must actually invoke; a session that never ran
# its skill is a silent no-op arm, not a data point (checked via transcript).
ARM_EXPECTED_SKILLS: dict[str, tuple[str, ...]] = {
"workflow": ("gitnexus-plan", "gitnexus-work"),
"ce_workflow": ("ce-plan", "ce-work"),
"workflow_direct": ("gitnexus-work",),
"ce_workflow_direct": ("ce-work",),
"review": ("gitnexus-review",),
"ce_review": ("ce-code-review",),
}
def _require_implementation_fingerprint(
session: dict[str, Any],
worktree: Path,
arm: str,
expected: str | None,
) -> None:
"""Bind a just-finished implementation session to its original skill bytes."""
try:
require_skill_fingerprint(
worktree,
arm,
expected,
phase="implementation",
)
except ValueError as exc:
if session.get("error_kind") is None:
session["ok"] = False
session["error_kind"] = "implementation-evidence-invalid"
session["error_detail"] = str(exc)
else:
session.setdefault("evidence_diagnostics", []).append(str(exc))
def _verification_outcome(result: VerificationResult | tuple[bool, str]) -> tuple[bool, str]:
if isinstance(result, VerificationResult):
if result.process.state != "exited":
# Hidden-oracle output can contain mounted test bytes. Preserve
# terminal-state evidence without letting candidate-controlled
# stdout/stderr enter results.jsonl through the exception string.
safe_process = replace(
result.process,
stdout_tail="",
stderr_tail="",
detail=result.process.detail or "verifier infrastructure failed",
)
raise ManagedProcessError(result.command, safe_process)
return result.passed, result.output
return result
def _run_hidden_oracle(
snapshot: TaskOracleSnapshot,
worktree: Path,
args: argparse.Namespace,
sandbox: SandboxSession,
) -> tuple[bool, str]:
"""Stage a captured oracle after the model exits, execute it, then erase it."""
if worktree.expanduser().absolute() != sandbox.clone.expanduser().absolute():
raise SandboxError("hidden oracle sandbox does not bind the credited worktree")
mount_name = f".wfbench-oracle-{secrets.token_hex(16)}"
mount_point = worktree / mount_name
mount_point.mkdir(mode=0o700)
primary: BaseException | None = None
try:
host_unsafe = getattr(sandbox, "backend", "bwrap") == "host-unsafe"
# Host mode has no bind mounts. Keep relative candidate imports valid
# by staging beside the candidate, still only after the model exits.
stage_parent = worktree if host_unsafe else sandbox.private_root
with staged_task_oracle(stage_parent, snapshot) as stage_root:
oracle_env = build_sandbox_environment()
# A private RO bind at a random workspace sibling preserves each
# oracle's ../gitnexus import as the candidate implementation. The
# empty mountpoint exists only post-model and is removed before the
# credited patch is captured.
oracle_mount = f"{SANDBOX_WORKSPACE}/{mount_name}"
oracle_env[ORACLE_ENV_VAR] = str(stage_root) if host_unsafe else oracle_mount
passed, _output = _verification_outcome(
run_verify(
snapshot.command,
sandbox.clone,
args.timeout,
command_prefix=sandbox.command_prefix_for(
read_only_workspace=True,
unshare_network=True,
extra_read_only_mounts=()
if host_unsafe
else (ReadOnlyMount(source=stage_root, target=oracle_mount),),
),
env=oracle_env,
require_pid_namespace=getattr(sandbox, "require_pid_namespace", True),
)
)
# Candidate code executes in this process. Never persist its stdout
# or stderr: it can read the mounted hidden test bytes and print them.
return passed, "hidden oracle passed" if passed else "hidden oracle failed"
except BaseException as exc:
primary = exc
raise
finally:
try:
metadata = mount_point.lstat()
if stat.S_ISLNK(metadata.st_mode) or not stat.S_ISDIR(metadata.st_mode):
raise SandboxError("hidden oracle mountpoint changed type during verification")
mount_point.rmdir()
except (OSError, SandboxError) as cleanup:
if primary is None:
raise
primary.add_note(f"hidden oracle mountpoint cleanup also failed: {cleanup}")
def _evaluated_skill_roots(worktree: Path, arm: str) -> tuple[Path, ...]:
"""Repo-local prompt roots that must remain immutable during a session."""
return tuple(worktree / ".claude" / "skills" / name for name in EVALUATED_ARM_SKILLS.get(arm, ()))
def isolated_gitnexus_registry_mount(worktree: Path, parent: Path) -> ReadOnlyMount:
"""Create a one-clone registry that cannot route MCP to any host repo."""
metadata_path = worktree / ".gitnexus" / "gitnexus.json"
if not metadata_path.exists():
metadata_path = worktree / ".gitnexus" / "meta.json"
mode = metadata_path.lstat().st_mode
if stat.S_ISLNK(mode) or not stat.S_ISREG(mode):
raise SandboxError(f"benchmark index metadata must be regular and non-symlink: {metadata_path}")
raw = _bounded_regular_bytes(metadata_path, limit=2 * 1024 * 1024)
try:
metadata = json.loads(raw)
except json.JSONDecodeError as exc:
raise SandboxError(f"benchmark index metadata is malformed: {metadata_path}") from exc
if not isinstance(metadata, dict):
raise SandboxError(f"benchmark index metadata must be an object: {metadata_path}")
indexed_at = metadata.get("indexedAt")
last_commit = metadata.get("lastCommit")
if not isinstance(indexed_at, str) or not indexed_at or not isinstance(last_commit, str) or not last_commit:
raise SandboxError("benchmark index metadata is missing indexedAt or lastCommit")
parent = parent.expanduser().absolute()
registry = Path(tempfile.mkdtemp(prefix="wfbench-registry-", dir=parent))
registry.chmod(0o700)
entry: dict[str, Any] = {
"name": "benchmark-target",
"path": SANDBOX_WORKSPACE,
"storagePath": f"{SANDBOX_WORKSPACE}/.gitnexus",
"indexedAt": indexed_at,
"lastCommit": last_commit,
}
for field in ("remoteUrl", "stats", "branch"):
if field in metadata:
entry[field] = metadata[field]
registry_file = registry / "registry.json"
descriptor = os.open(
registry_file,
os.O_WRONLY | os.O_CREAT | os.O_EXCL | getattr(os, "O_NOFOLLOW", 0),
0o600,
)
try:
os.fchmod(descriptor, 0o600)
payload = (json.dumps([entry], sort_keys=True, separators=(",", ":")) + "\n").encode()
view = memoryview(payload)
while view:
written = os.write(descriptor, view)
view = view[written:]
finally:
os.close(descriptor)
return ReadOnlyMount(source=registry, target=SANDBOX_GITNEXUS_REGISTRY)
def _unchanged(value: Any) -> Any:
return value
def run_arm(
arm: str,
task: dict[str, Any],
worktree: Path,
args: argparse.Namespace,
*,
sandbox: SandboxSession,
transcript_output_dir: Path | None = None,
transcript_output_prefix: str | None = None,
expected_skill_digest: str | None = None,
enforce_phase_boundary: bool = False,
ce_plugin_dir: str | None = None,
oracle_snapshot: TaskOracleSnapshot | None = None,
) -> dict[str, Any]:
sessions: list[dict[str, Any]] = []
environment_builder = getattr(sandbox, "environment", build_sandbox_environment)
host_text = getattr(sandbox, "host_text", _unchanged)
host_path = getattr(sandbox, "host_path", lambda value: str(value))
backend = getattr(sandbox, "backend", "bwrap")
env = model_session_environment(
auth_token=args.auth_token,
base_url=args.base_url,
model=args.model,
build_sandbox_environment=environment_builder,
)
# --bare hard-disables the Skill tool and every mcp__* tool — by Claude
# Code design, not a bug (--allowedTools can't restore what --bare
# removes). Every arm except baseline_nomcp needs Skill and/or MCP tools,
# so only baseline_nomcp can keep --bare's tighter isolation; the rest
# rely on ANTHROPIC_API_KEY alone (the sandboxed HOME has no OAuth/
# keychain state to conflict with it).
bare = arm == "baseline_nomcp"
progress_label = transcript_output_prefix or f"{task.get('id', 'task')}-{arm}"
common = {
"progress_label": progress_label,
"claude_bin": sandbox.claude_bin,
"timeout": args.timeout,
"model": args.model,
"effort": args.effort,
"env": env,
"permission_mode": (
"bypassPermissions" if backend == "host-unsafe" else "dontAsk"
),
"command_prefix": sandbox.command_prefix_for(
read_only_paths=_evaluated_skill_roots(worktree, arm),
),
"require_pid_namespace": getattr(sandbox, "require_pid_namespace", True),
"bare": bare,
"settings_json": sandbox.settings_json,
"strict_mcp_config": True,
"mcp_config_json": host_text(sandbox_mcp_config()),
"transcript_projects": sandbox.transcript_projects,
"transcript_cwd": Path(SANDBOX_WORKSPACE),
"transcript_wait_seconds": 5,
"transcript_output_dir": transcript_output_dir,
"transcript_output_prefix": transcript_output_prefix,
"transcript_secrets": tuple(credential_secrets(args)),
}
if ce_plugin_dir is not None:
common["plugin_dirs"] = (host_path(ce_plugin_dir),)
expected_skills = ARM_EXPECTED_SKILLS.get(arm, ())
plan_doc: Path | None = None
if arm in ("workflow", "ce_workflow"):
plan_prompt = PLAN_PROMPT if arm == "workflow" else CE_PLAN_PROMPT
work_prompt = WORK_PROMPT if arm == "workflow" else CE_WORK_PROMPT
pre = snapshot_plan_docs(worktree)
phase_before = workspace_snapshot(worktree) if enforce_phase_boundary else None
plan_session = run_claude(
plan_prompt.format(task=task["prompt"]),
worktree,
expected_skill=expected_skills[0],
**{
**common,
"progress_label": f"{progress_label} plan",
"allowed_tools": allowed_agent_tools(implementation=False),
},
)
sessions.append(plan_session)
if plan_session["ok"]:
try:
plan_doc = new_plan_doc(worktree, pre)
if phase_before is not None:
enforce_phase_workspace(
worktree,
phase_before,
allowed_artifact=plan_doc,
)
require_skill_fingerprint(
worktree,
arm,
expected_skill_digest,
phase="planning",
)
except ValueError as exc:
plan_session["ok"] = False
plan_session["error_kind"] = "plan-evidence-invalid"
plan_session["error_detail"] = str(exc)
else:
work_session = run_claude(
work_prompt.format(plan=plan_doc.relative_to(worktree)),
worktree,
expected_skill=expected_skills[1],
**{
**common,
"progress_label": f"{progress_label} work",
"allowed_tools": allowed_agent_tools(implementation=True),
},
)
_require_implementation_fingerprint(
work_session,
worktree,
arm,
expected_skill_digest,
)
sessions.append(work_session)
elif arm == "ce_workflow_direct":
work_session = run_claude(
CE_WORK_DIRECT_PROMPT.format(task=task["prompt"]),
worktree,
expected_skill=expected_skills[0],
**{**common, "allowed_tools": allowed_agent_tools(implementation=True)},
)
_require_implementation_fingerprint(
work_session,
worktree,
arm,
expected_skill_digest,
)
sessions.append(work_session)
elif arm in ("review", "ce_review"):
review_prompt = REVIEW_PROMPT if arm == "review" else CE_REVIEW_PROMPT
review_output = prepare_review_workspace(sandbox, REVIEW_OUTPUT)
phase_before = workspace_snapshot(worktree) if enforce_phase_boundary else None
review_common = {
**common,
"allowed_tools": allowed_agent_tools(implementation=False, allow_edit=False),
"command_prefix": sandbox.command_prefix_for(
read_only_workspace=True,
read_only_paths=_evaluated_skill_roots(worktree, arm),
extra_writable_mounts=(
# The DIRECTORY, outside the workspace. Binding the file
# itself left the agent nowhere to put the temp file it
# renames into place, so every review artifact came back
# empty with EROFS in the transcript.
ReadOnlyMount(
source=review_output.parent,
target=SANDBOX_REVIEW_OUTPUT,
),
),
),
}
with sandbox_workspace_write_boundary(
sandbox,
read_only_workspace=True,
# Nothing in the workspace is writable now — the artifact left it.
writable=(),
):
review_session = run_claude(
host_text(review_prompt.format(task=task["prompt"])),
worktree,
expected_skill=expected_skills[0],
**review_common,
)
sessions.append(review_session)
if review_session["ok"] and phase_before is not None:
try:
# The artifact is no longer in the workspace, so the review
# phase may now change nothing there at all.
enforce_phase_workspace(worktree, phase_before, allowed_artifact=None)
require_skill_fingerprint(
worktree,
arm,
expected_skill_digest,
phase="review",
)
except ValueError as exc:
review_session["ok"] = False
review_session["error_kind"] = "review-evidence-invalid"
review_session["error_detail"] = str(exc)
elif arm == "workflow_direct":
work_session = run_claude(
WORK_DIRECT_PROMPT.format(task=task["prompt"]),
worktree,
expected_skill=expected_skills[0],
**{**common, "allowed_tools": allowed_agent_tools(implementation=True)},
)
_require_implementation_fingerprint(
work_session,
worktree,
arm,
expected_skill_digest,
)
sessions.append(work_session)
elif arm == "baseline_nomcp":
# Isolates the workflow-discipline question from the GitNexus-tools
# question: no skills AND no graph tools.
sessions.append(
run_claude(
BASELINE_PROMPT.format(task=task["prompt"]),
worktree,
disallowed_tools=["Skill", "mcp__gitnexus"],
**{
**common,
"mcp_config_json": '{"mcpServers":{}}',
"allowed_tools": allowed_agent_tools(
implementation=True,
include_mcp=False,
),
},
)
)
else:
sessions.append(
run_claude(
BASELINE_PROMPT.format(task=task["prompt"]),
worktree,
disallowed_tools=["Skill"],
**{**common, "allowed_tools": allowed_agent_tools(implementation=True)},
)
)
record = sum_sessions(sessions)
record["arm"] = arm
record["plan_produced"] = arm not in ("workflow", "ce_workflow") or plan_doc is not None
# The verify command runs in its OWN sandbox invocation, which knows nothing
# about the review session's writable mount. Expose the artifact read-only and
# name it through the environment, the same shape _run_hidden_oracle uses, so
# one command works on both backends instead of hardcoding either path.
verify_env = environment_builder()
verify_mounts: tuple[ReadOnlyMount, ...] = ()
if arm in ("review", "ce_review"):
review_artifact = review_output_path(sandbox, REVIEW_OUTPUT)
verify_env[REVIEW_OUTPUT_ENV_VAR] = host_text(f"{SANDBOX_REVIEW_OUTPUT}/{REVIEW_OUTPUT}")
verify_mounts = (
ReadOnlyMount(source=review_artifact.parent, target=SANDBOX_REVIEW_OUTPUT),
)
authored_tests_passed, authored_test_output = _verification_outcome(
run_verify(
task["verify"],
worktree,
args.timeout,
command_prefix=sandbox.command_prefix_for(
read_only_workspace=True,
unshare_network=True,
extra_read_only_mounts=verify_mounts,
),
env=verify_env,
require_pid_namespace=getattr(sandbox, "require_pid_namespace", True),
)
)
review_score: dict[str, Any] | None = None
if arm in ("review", "ce_review"):
try:
verdict, findings = parse_review_output(review_output_path(sandbox, REVIEW_OUTPUT))
labels = expected_findings(oracle_snapshot) if oracle_snapshot is not None else ()
review_score = score_review(verdict, findings, labels)
except (OSError, ValueError) as exc:
record["ok"] = False
record["error_kind"] = record["error_kind"] or "review-evidence-invalid"
# Keep the FIRST detail, as error_kind already does. A phase-
# boundary violation is why the artifact is unparseable; reporting
# the parse failure over it buries the cause under the symptom.
record["error_detail"] = record.get("error_detail") or str(exc)
record["review_score"] = review_score
record["review_evidence_valid"] = review_score is not None
if review_score is not None:
record.update({f"review_{key}": value for key, value in review_score.items()})
if oracle_snapshot is None:
oracle_passed, oracle_output = False, "hidden oracle snapshot unavailable"
elif arm in ("review", "ce_review"):
oracle_passed = bool(
review_score
and review_score["false_positives"] == 0
and review_score["false_negatives"] == 0
and review_score["verdict_correct"]
)
oracle_output = "hidden review labels matched" if oracle_passed else "hidden review labels not fully matched"
else:
oracle_passed, oracle_output = _run_hidden_oracle(
oracle_snapshot,
worktree,
args,
sandbox,
)
record["authored_tests_passed"] = authored_tests_passed
record["authored_test_output"] = authored_test_output
record["oracle_passed"] = oracle_passed
record["oracle_output"] = oracle_output
record["resolved"] = record["ok"] and authored_tests_passed and oracle_passed
# Compatibility alias for existing report consumers. The authored tests are
# now an explicit signal and can never self-certify resolution.
record["verify_output"] = authored_test_output
if oracle_snapshot is not None:
record.update(
{
"oracle_digest": oracle_snapshot.digest,
"oracle_command_digest": oracle_snapshot.command_digest,
"oracle_manifest_digest": oracle_snapshot.manifest_digest,
}
)
if record["error_kind"] is None and not authored_tests_passed:
# The sessions completed — the produced change just failed the task's
# verify command. Kept distinct from session-error so aggregates can
# exclude infrastructure deaths without hiding real failures.
record["error_kind"] = "verify-failed"
elif record["error_kind"] is None and not oracle_passed:
record["error_kind"] = "oracle-failed" if oracle_snapshot is not None else "oracle-unavailable"
return record
# ─── Pure aggregation/report helpers (unit-tested) ──────────────────────────
CHURN_FIELDS = ("diff_files", "diff_insertions", "diff_deletions")
# Rows where the session (or the harness) died carry no measured evidence and
# must not skew efficiency medians or resolve denominators. verify-failed and
# skill-not-invoked rows DO count: those sessions ran and spent real tokens.
# One definition, in comparator_reuse: reuse eligibility and aggregate
# exclusion must never drift apart. The dependency only runs this way -
# comparator_reuse importing back from runner is a circular import.
EXCLUDED_ERROR_KINDS = REUSE_EXCLUDED_ERROR_KINDS
# 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
# wrong about a hard corpus while every process, mount and capture behaved.
#
# EXECUTION: the process or its tooling did not complete. Nothing was measured.
# EVIDENCE: it completed, but what it produced cannot be trusted or scored.
# Everything else - including resolved=False and a zero score - is a VALID
# NEGATIVE: an admissible measurement that the quality gate then judges.
EXECUTION_FAILURE_KINDS = frozenset({"session-error", "infra-error", "cleanup-failure", "cancelled"})
EVIDENCE_FAILURE_KINDS = frozenset({"review-evidence-invalid", "evidence-unverified", "skill-not-invoked"})
def execution_failed(record: Mapping[str, Any]) -> bool:
"""The process or its tooling did not complete."""
return record.get("error_kind") in EXECUTION_FAILURE_KINDS
def evidence_failed(record: Mapping[str, Any]) -> bool:
"""It completed, but what it produced cannot be trusted or scored."""
return (
record.get("error_kind") in EVIDENCE_FAILURE_KINDS
or record.get("review_evidence_valid") is False
or record.get("transcript_missing") is True
)
def task_prompt_digest(task: Mapping[str, Any]) -> str:
"""The prompt digest, computed once for the row and the reuse expectation.
row_is_reusable_comparator compares the value a prior row stored against the
value this sweep derives, as exact strings. Two inline copies of this hash
had already drifted - one picked up a str() cast the other lacked - and a
further divergence (normalising whitespace on one side, say) would silently
stop rows matching, or match rows that should not.
"""
return hashlib.sha256(str(task["prompt"]).encode()).hexdigest()
# 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.
SYSTEMIC_ERROR_KINDS = frozenset({"session-error", "infra-error", "cleanup-failure", "review-evidence-invalid"})
DEFAULT_OUTAGE_STREAK = 5
# A cell is a full clone plus a sandboxed agent session, so the ceiling is the
# machine, not the flag. Past a handful of siblings the cells lose CPU to each
# other, sessions reach their timeout, and a timed-out session is an excluded
# run the promotion gate refuses to work with — a mistyped --workers must fail
# at the command line rather than a quarter-day later as unusable evidence.
MAX_WORKERS = 8
def systemic_outage_streak(error_kind: str | None, prior_streak: int) -> int:
"""Consecutive systemic-failure count: +1 on a systemic kind, else reset to 0."""
return prior_streak + 1 if error_kind in SYSTEMIC_ERROR_KINDS else 0
def _run_wave(
wave: Sequence[tuple[int, str]],
*,
workers: int,
run: Callable[[int, str], dict[str, Any]],
cancel_event: threading.Event,
) -> tuple[list[dict[str, Any] | BaseException], BaseException | None]:
"""Cancel active subprocesses, join workers, and retain settled outcomes."""
pool = ThreadPoolExecutor(max_workers=workers)
futures = []
interruption = None
try:
for run_idx, arm in wave:
futures.append(pool.submit(copy_context().run, run, run_idx, arm))
wait(futures)
except (Exception, KeyboardInterrupt, SystemExit) as exc:
interruption = exc
cancel_event.set()
finally:
pool.shutdown(wait=True, cancel_futures=interruption is not None)
outcomes: list[dict[str, Any] | BaseException] = []
for future in futures:
if future.cancelled():
outcomes.append({"resolved": False, "error_kind": "cancelled"})
else:
error = future.exception()
outcomes.append(error if error is not None else future.result())
# Submission itself can be interrupted. Preserve positional evidence for
# cells that never started without masking the original interruption.
outcomes.extend({"resolved": False, "error_kind": "cancelled"} for _ in wave[len(futures) :])
return outcomes, interruption
def sweep_task_cells(
cells: Sequence[tuple[int, str]],
*,
workers: int,
run: Callable[[int, str], dict[str, Any]],
on_start: Callable[[int, str], None],
on_record: Callable[[int, str, dict[str, Any]], None],
outage_streak: int,
outage_limit: int,
cancel_event: threading.Event | None = None,
) -> tuple[int, bool]:
"""Run one task's cells in waves of ``workers``; return (streak, tripped).
Waves rather than one fan-out, because the outage breaker counts
CONSECUTIVE systemic failures and "consecutive" only means anything in a
fixed order — completion order under concurrency is not one. Each wave is
folded in submission order once it has fully completed, and the next wave
starts only if the breaker held, so the breaker overruns its limit by at
most ``workers - 1`` cells: the ones already in flight when it tripped.
The serial default executes directly; parallel workers copy the run
context so every owned subprocess observes the same cancellation signal.
"""
with cancellation_scope(cancel_event) as cancel_event:
if workers < 1:
raise ValueError("workers must be positive")
for wave_start in range(0, len(cells), workers):
if cancel_event.is_set():
# False: this flag means the OUTAGE breaker tripped, and the
# caller turns it into exit 1 with "Sweep aborted". Cancellation
# stops the sweep too, but it is the operator's Ctrl-C, not a
# systemic failure - reporting True relabelled every interrupted
# run an outage and returned 1 where the contract says 130. The
# caller tests cancel_event itself for the stop decision.
return outage_streak, False
wave = list(cells[wave_start : wave_start + workers])
for run_idx, arm in wave:
on_start(run_idx, arm)
if workers == 1:
records = [run(run_idx, arm) for run_idx, arm in wave]
else:
outcomes, interruption = _run_wave(wave, workers=workers, run=run, cancel_event=cancel_event)
failure = interruption or next(
(outcome for outcome in outcomes if isinstance(outcome, BaseException)), None
)
if failure is not None:
# The siblings of the failing cell have already completed and
# spent their budget. Persist their rows, in submission order,
# before the harness bug takes the process down — otherwise a
# crash in one cell silently erases the evidence of the others.
for (run_idx, arm), outcome in zip(wave, outcomes, strict=True):
if not isinstance(outcome, BaseException):
on_record(run_idx, arm, outcome)
raise failure
records = [outcome for outcome in outcomes if not isinstance(outcome, BaseException)]
for (run_idx, arm), record in zip(wave, records, strict=True):
# Every future in this wave has already completed and incurred its
# cost. Persist all of them in canonical submission order even if
# an earlier row trips the breaker; only later waves are skipped.
on_record(run_idx, arm, record)
if cancel_event.is_set():
return outage_streak, False
for record in records:
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."
)
# Signal in-flight background work too. The breaker exists to
# SHORTEN a doomed run; without this a graph prefetch keeps
# building and the unconditional join blocks the abort for the
# length of a full clone and offline index.
cancel_event.set()
return outage_streak, True
return outage_streak, False
# How far ahead of the in-order fold pointer cells may be submitted, as a
# multiple of the worker count. This is the wall-clock/wasted-cell trade, and it
# is a real one - measured against the review corpus at workers=3, with failures
# injected at four different positions:
#
# window wall vs waves worst overrun
# 3 -8% 2 (the wave scheduler's own bound)
# 6 -27% 4
# 12 -42% 9
# 54 -44% 11
#
# Overrun is wasted paid sessions when the breaker trips, at roughly $70 each.
# 2 is the default because it keeps the worst case within 2x the wave bound
# while taking most of the gain; raise it if a run's wall clock costs more than
# an occasional handful of cells on an aborted sweep.
PACKED_WINDOW_MULTIPLIER = 2
def sweep_packed_cells(
cells: Sequence[tuple[str, int, str]],
*,
workers: int,
run: Callable[[str, int, str], dict[str, Any]],
on_start: Callable[[str, int, str], None],
on_record: Callable[[str, int, str, dict[str, Any]], None],
outage_streak: int,
outage_limit: int,
window: int | None = None,
await_ready: Callable[[str], bool] | None = None,
cancel_event: threading.Event | None = None,
) -> tuple[int, bool]:
"""Run cells from EVERY task through one pool; return (streak, tripped).
``sweep_task_cells`` finishes one task before starting the next and drains a
wave before refilling it, so a task with fewer cells than ``workers`` leaves
workers idle and a slow cell stalls its whole wave. Packing every task's
cells into one continuously fed pool removes both, which is worth about 40%
of a cold sweep's wall clock and is the only thing that moves a seeded
weekly run at all - there, a task is three cells and a wave is never full.
The breaker keeps its exact meaning. ``cells`` is a total submission order
(task-major, run-major, arm-minor - the same order waves fold in, continued
across task boundaries), a folder walks results in precisely that order, and
"consecutive systemic failures" is evaluated there. So the run aborts on the
same logical cell it would have aborted on under waves.
``window`` is what bounds the overrun, and it is load-bearing. The halt flag
alone is not enough: the folder walks in order, so a slow early cell lets
workers race ahead, and by the time the breaker trips those cells have
already paid for their sessions. Measured, an unbounded queue overran by 11
cells at ``workers=3`` where the wave scheduler overruns by 2. Holding
submission to ``window`` cells beyond the fold point caps it, trading
packing for wasted cells - see ``PACKED_WINDOW_MULTIPLIER`` for the curve.
``await_ready`` gates a task's first cell on whatever that task still needs
(a sanitized clone, a graph). It returns False to abandon the task, whose
cells are then skipped rather than run against missing assets. Cells are
submitted as their task becomes ready, so a later task's graph builds while
earlier cells are still paying for sessions.
"""
with cancellation_scope(cancel_event) as cancel_event:
if workers < 1:
raise ValueError("workers must be positive")
if not cells:
return outage_streak, False
if window is None:
window = max(workers * PACKED_WINDOW_MULTIPLIER, workers)
if window < workers:
raise ValueError("window must be at least workers, or the pool starves")
halt = threading.Event()
results: list[dict[str, Any] | None] = [None] * len(cells)
submitted: list[Any] = []
gate = threading.Condition()
producing = True
fold_pointer = 0
def execute(index: int) -> None:
if halt.is_set() or cancel_event.is_set():
return
task_id, run_idx, arm = cells[index]
on_start(task_id, run_idx, arm)
results[index] = run(task_id, run_idx, arm)
pool = ThreadPoolExecutor(max_workers=workers)
# cancellation_scope binds _CANCELLATION in the CALLING thread's
# context, and a new thread starts with an empty one - so the producer
# has to copy this context rather than its own, or every cell it
# submits loses the run's cancellation event. sweep_task_cells gets
# this for free by submitting from the thread that entered the scope.
caller_context = copy_context()
def produce() -> None:
nonlocal producing
ready_tasks: dict[str, bool] = {}
try:
for index, (task_id, _run_idx, _arm) in enumerate(cells):
if halt.is_set() or cancel_event.is_set():
break
if task_id not in ready_tasks:
ready_tasks[task_id] = True if await_ready is None else await_ready(task_id)
if not ready_tasks[task_id]:
with gate:
submitted.append(None)
gate.notify_all()
continue
with gate:
while index - fold_pointer >= window and not halt.is_set():
gate.wait(timeout=0.5)
if halt.is_set() or cancel_event.is_set():
break
worker_context = caller_context.run(copy_context)
submitted.append(pool.submit(worker_context.run, execute, index))
gate.notify_all()
finally:
with gate:
producing = False
gate.notify_all()
producer = threading.Thread(target=produce, name="packed-cell-producer", daemon=False)
producer.start()
tripped = False
try:
index = 0
while True:
with gate:
while index >= len(submitted) and producing:
gate.wait(timeout=0.5)
if index >= len(submitted):
break
future = submitted[index]
if future is not None:
try:
future.result()
except BaseException:
# Same contract as sweep_task_cells: the cells submitted
# after this one have already run and spent their budget,
# so persist their rows in submission order before the
# harness bug takes the process down. Without this, one
# crashing cell silently erases the paid evidence of
# every sibling that had already finished. The failing
# index itself has no row - execute() only assigns on
# success - so folding forward cannot duplicate it.
with gate:
settled = list(submitted)
for later in range(index + 1, len(settled)):
pending = settled[later]
if pending is not None and not pending.done():
continue
row = results[later]
if row is not None:
on_record(*cells[later], row)
raise
record = results[index]
if record is not None:
task_id, run_idx, arm = cells[index]
on_record(task_id, run_idx, arm, record)
kind = (
"review-evidence-invalid"
if record.get("review_evidence_valid") is False
else record.get("error_kind")
)
outage_streak = systemic_outage_streak(kind, outage_streak)
if outage_limit and outage_streak >= outage_limit:
print(
f"[systemic-outage] {outage_streak} consecutive unusable-evidence "
"failures — aborting the remaining sweep; report and promotion are "
"written from partial evidence and the run exits non-zero."
)
tripped = True
halt.set()
cancel_event.set()
break
index += 1
with gate:
fold_pointer = index
gate.notify_all()
if cancel_event.is_set():
tripped = True
break
finally:
halt.set()
with gate:
gate.notify_all()
producer.join()
for pending in submitted[index + 1 :]:
if pending is not None:
pending.cancel()
pool.shutdown(wait=True)
return outage_streak, tripped
@dataclass(frozen=True)
class TaskCellContext:
"""Everything one benchmark cell needs from its task, prepared once.
A cell is one (run, arm) pair: a private clone, a sandboxed session set, and
the row it produces. Cells of the same task share this context read-only, so
it is what makes them independent of each other — every per-cell mutable is
local to ``run_cell``. Holding the fields explicitly, rather than closing
over ``main``'s scope, is what lets a cell run off the main thread without
dragging the whole sweep's state along with it.
``args`` is treated as immutable: ``main`` finishes mutating it during
setup, well before any cell starts. ``argparse.Namespace`` cannot enforce
that, so it is stated here.
"""
task: dict[str, Any]
oracle_snapshot: TaskOracleSnapshot
repo: Path
task_sha: str
graph_snapshot: SanitizedGraphSnapshot | None
graph_snapshot_error: BaseException | None
asset_snapshot: TaskAssetSnapshot | None
asset_snapshot_error: BaseException | None
args: argparse.Namespace
out_dir: Path
ce_plugin_snapshot: CePluginSnapshot | None
trees_dir: Path
bwrap_bin: Path
runtime_mounts: tuple[ReadOnlyMount, ...]
candidate_overlay: Path | None
overlay_digest: str | None
sandbox_backend: str = "bwrap"
clone_template: Path | None = None
sanitized_head: str | None = None
def run_cell(ctx: TaskCellContext, run_idx: int, arm: str) -> dict[str, Any]:
"""Run one (run, arm) cell end to end and return its result row.
Owns its clone for the whole call, including teardown: the ``finally``
removes the worktree whatever happens, and an exception outside the five
expected kinds is deliberately left to propagate — a harness bug must not be
recorded as an ordinary infra-error and averaged into the evidence.
"""
args = ctx.args
task = ctx.task
worktree: Path | None = None
record: dict[str, Any] | None = None
cleanup_error: OSError | None = None
try:
if ctx.asset_snapshot_error is not None:
raise RuntimeError(f"task asset snapshot preparation failed: {ctx.asset_snapshot_error}")
if ctx.graph_snapshot_error is not None:
raise RuntimeError(f"sanitized graph snapshot preparation failed: {ctx.graph_snapshot_error}")
if ctx.graph_snapshot is None:
raise RuntimeError("sanitized graph snapshot is unavailable")
if ctx.asset_snapshot is None:
raise RuntimeError("task asset snapshot is unavailable")
if ctx.clone_template is not None:
if not ctx.sanitized_head:
raise RuntimeError("clone template is missing its sanitized HEAD")
worktree = copy_isolated_tree(ctx.clone_template, ctx.trees_dir)
sanitized_head = ctx.sanitized_head
else:
worktree = make_worktree(ctx.repo, ctx.task_sha, ctx.trees_dir)
sanitized_head = sanitize_clone_for_hidden_oracles(worktree)
ctx.graph_snapshot.materialize(worktree, sanitized_head=sanitized_head)
dependency_mounts = stage_task_assets(
task,
repo=ctx.repo,
clone=worktree,
snapshot=ctx.asset_snapshot,
)
registry_mount = isolated_gitnexus_registry_mount(worktree, ctx.trees_dir)
execution_arm = CANDIDATE_ARMS.get(arm, arm)
ce_mounts = ce_plugin_mounts_for_arm(execution_arm, ctx.ce_plugin_snapshot)
with prepare_sandbox(
clone=worktree,
claude_bin=args.claude_bin,
bwrap_bin=ctx.bwrap_bin,
read_only_mounts=[
*dependency_mounts,
*ctx.runtime_mounts,
registry_mount,
*ce_mounts,
],
preflight=False,
backend=ctx.sandbox_backend,
) as sandbox:
# Capture the BASE (pre-overlay) skill digest — identical
# for the incumbent and candidate arms — then run the
# task's untrusted setup against those base skills. The
# candidate overlay is applied only afterwards, so setup
# can never observe candidate prose and both arms share
# byte-identical pre-overlay state.
# Historical review SHAs may predate gitnexus-review. Seed
# the current evaluated skill first so fingerprinting and
# the model see the same incumbent prose on every case.
if execution_arm == "review":
seed_evaluated_skills(
HARNESS_ROOT,
worktree,
sandbox=sandbox,
arm=execution_arm,
)
base_skill_digest = skill_fingerprint(worktree, execution_arm)
if task.get("setup"):
# Sanitization already removed eval/workflow_bench. Review
# cells copy the historical PR patch back under that path so
# `git apply` can read it. Do not overlay an empty mask on
# the same tree first — that hides the patch file and every
# cell dies with `can't open patch`. A historical patch that
# still edits the harness (gitignored learnings.jsonl) must
# skip those hunks or apply fails closed.
setup_command = ["/bin/sh", "-lc", with_hidden_harness_apply_exclude(str(task["setup"]))]
setup = sandbox.run(
setup_command,
timeout=600,
env=build_sandbox_environment(),
)
if not setup.ok:
raise ManagedProcessError(setup_command, setup)
# Setup (or its absence) must leave the staged harness copy gone
# before the model session starts. Fail closed rather than hide
# the tree with a mask that would also hide the patch from apply.
require_hidden_harness_absent(worktree)
# Tamper-evidence: setup must not have rewritten the base
# skills, verified before any candidate overlay lands.
require_skill_fingerprint(
worktree,
execution_arm,
base_skill_digest,
phase="task setup",
)
if arm in CANDIDATE_ARMS:
if ctx.candidate_overlay is None:
raise RuntimeError("candidate overlay is unavailable")
applied_digest = apply_candidate_overlay(
ctx.candidate_overlay,
worktree,
sandbox=sandbox,
)
if applied_digest != ctx.overlay_digest:
raise RuntimeError("candidate overlay changed during the benchmark run")
# The digest the model must preserve during its run is the
# post-overlay skill surface (candidate skills for
# candidate arms; unchanged base skills otherwise).
expected_skill_digest = (
skill_fingerprint(worktree, execution_arm) if arm in CANDIDATE_ARMS else base_skill_digest
)
orig_sha = _sandbox_git(sandbox, ["rev-parse", "HEAD"]).strip()
if not re.fullmatch(r"[0-9a-fA-F]{40,64}", orig_sha):
raise RuntimeError("sandboxed candidate setup did not produce an immutable commit")
before_work_digest = (
implementation_diff_digest(sandbox, orig_sha) if execution_arm in IMPLEMENTATION_ARMS else ""
)
record = run_arm(
execution_arm,
task,
worktree,
args,
sandbox=sandbox,
transcript_output_dir=ctx.out_dir,
transcript_output_prefix=f"{task['id']}-{arm}-run{run_idx}",
expected_skill_digest=expected_skill_digest,
enforce_phase_boundary=True,
ce_plugin_dir=ce_plugin_dir_for_arm(execution_arm, ctx.ce_plugin_snapshot),
oracle_snapshot=ctx.oracle_snapshot,
)
if execution_arm in ("review", "ce_review"):
review_source = review_output_path(sandbox, REVIEW_OUTPUT)
if review_source.is_file() and not review_source.is_symlink():
review_artifact = ctx.out_dir / f"{task['id']}-{arm}-run{run_idx}.review.json"
review_artifact.write_bytes(_bounded_regular_bytes(review_source, limit=256 * 1024))
record["review_artifact"] = review_artifact.name
_prepare_untracked_for_diff(sandbox)
after_work_digest = (
implementation_diff_digest(
sandbox,
orig_sha,
prepare_untracked=False,
)
if execution_arm in IMPLEMENTATION_ARMS
else ""
)
record.update(
diff_churn(
sandbox,
orig_sha,
prepare_untracked=False,
)
)
enforce_work_evidence(
record,
arm=execution_arm,
before_digest=before_work_digest,
after_digest=after_work_digest,
)
patch_bytes = capture_patch(sandbox, worktree, orig_sha)
record["arm"] = arm
record.update(
{
"model": args.model,
"benchmark_model": args.model,
"proposer_model": args.proposer_model,
"effort": args.effort,
"task_ref": task.get("ref", "HEAD"),
"task_base_sha": ctx.task_sha,
"sanitized_task_sha": sanitized_head,
"variant_head_sha": orig_sha,
"task_prompt_digest": task_prompt_digest(task),
"skill_digest": expected_skill_digest,
"candidate_overlay_digest": (ctx.overlay_digest if arm in CANDIDATE_ARMS else None),
"runtime_digest": current_runtime_digest(),
"recorded_at": datetime.now(UTC).isoformat(),
}
)
# Final working-tree patch — the clone is destroyed, so
# this is the only artifact for diagnosing verify fails.
patch_path = ctx.out_dir / f"{task['id']}-{arm}-run{run_idx}.patch"
patch_path.write_bytes(patch_bytes)
except (
ManagedProcessError,
SandboxError,
OSError,
RuntimeError,
ValueError,
) as exc:
# One hung session or failed setup must not abort the
# sweep — record the run as infra-error and move on so
# report.md/promotion.json still get written.
record = infra_error_record(exc)
if isinstance(exc, ManagedProcessError) and exc.result.state == "cancelled":
record["error_kind"] = "cancelled"
record["arm"] = arm
# ManagedProcessError carries up to 1000 raw bytes of stderr_tail, and
# this line now streams live into the CI log (run_managed echoes the
# sweep's stdout). Redact it like every other sink this data reaches.
detail = redact_text(str(exc), credential_secrets(args))
print(f"[{task['id']}][{arm}][run {run_idx}] infra-error: {detail}")
finally:
if worktree is not None and worktree.exists():
try:
remove_clone(worktree)
except OSError as exc:
cleanup_error = exc
assert record is not None
if cleanup_error is not None:
primary_kind = record.get("error_kind")
primary_detail = record.get("error_detail")
record["resolved"] = False
record["ok"] = False
record["error_kind"] = "cleanup-failure"
record["error_detail"] = (
f"primary={primary_kind}: {primary_detail}; cleanup: {type(cleanup_error).__name__}: {cleanup_error}"
)[:2000]
record.update(
{
"task": task["id"],
"class": task.get("class", ""),
"run": run_idx,
"sandbox_backend": ctx.sandbox_backend,
"task_asset_snapshot_digest": (ctx.asset_snapshot.digest if ctx.asset_snapshot is not None else None),
"task_asset_manifest_digest": (
ctx.asset_snapshot.manifest_digest if ctx.asset_snapshot is not None else None
),
"sandbox_dependency_content_digest": (
ctx.asset_snapshot.dependency_content_digest if ctx.asset_snapshot is not None else None
),
"sandbox_dependency_manifest_digest": (
ctx.asset_snapshot.dependency_manifest_digest if ctx.asset_snapshot is not None else None
),
"sanitized_graph_snapshot_digest": (ctx.graph_snapshot.digest if ctx.graph_snapshot is not None else None),
"sanitized_graph_manifest_digest": (
ctx.graph_snapshot.manifest_digest if ctx.graph_snapshot is not None else None
),
"oracle_digest": ctx.oracle_snapshot.digest,
"oracle_command_digest": ctx.oracle_snapshot.command_digest,
"oracle_manifest_digest": ctx.oracle_snapshot.manifest_digest,
"ce_plugin_version": (
ctx.ce_plugin_snapshot.version if arm in CE_ARMS and ctx.ce_plugin_snapshot is not None else None
),
"ce_plugin_manifest_digest": (
ctx.ce_plugin_snapshot.manifest_digest
if arm in CE_ARMS and ctx.ce_plugin_snapshot is not None
else None
),
}
)
return record
def infra_error_record(exc: BaseException) -> dict[str, Any]:
"""Row for a run the harness itself killed (timeout, setup failure)."""
if isinstance(exc, ManagedProcessError):
process = exc.result
detail = f"{process.state}: {process.detail or process.stderr_tail[-1500:]}"
else:
detail = f"{type(exc).__name__}: {exc}"
record: dict[str, Any] = dict.fromkeys(USAGE_FIELDS, 0)
record.update(
{
"ok": False,
"resolved": False,
"error_kind": "infra-error",
"error_detail": detail[:2000],
"session_ids": [],
"cost_usd": 0.0,
"duration_s": 0.0,
"num_turns": 0,
"plan_produced": False,
"authored_tests_passed": False,
"authored_test_output": "",
"oracle_passed": False,
"oracle_output": "",
"verify_output": "",
"skill_invoked": None,
"transcript_missing": False,
}
)
return record
def cell_progress_line(task_id: str, arm: str, run_idx: int, record: dict[str, Any]) -> str:
"""The live one-line summary printed as each cell finishes.
An infra-error row carries 0.0 cost and 0.0 duration as placeholders: the
cell died before any session could report a number. Printed as bare zeros
next to real rows they read as a run that was instant and free — the exact
misreading ``_na`` exists to prevent — so they are rendered "n/a" instead.
The row on disk is untouched: results.jsonl is promotion evidence and its
field types stay as they are.
"""
measured = record.get("error_kind") not in {"infra-error", "cleanup-failure"}
cost_usd = record.get("cost_usd") if measured else None
duration_s = record.get("duration_s") if measured else None
quality = record.get("review_weighted_f1")
quality_text = "n/a" if quality is None else f"{quality:.3f}"
return (
f"[{task_id}][{arm}][run {run_idx}] resolved={record['resolved']} "
f"quality={quality_text} "
f"in={record['input_tokens']} out={record['output_tokens']} "
f"cost={'n/a' if cost_usd is None else f'${cost_usd}'} "
f"took={'n/a' if duration_s is None else f'{duration_s}s'} "
# An excluded run is what actually blocks promotion, so name it here
# instead of leaving it to results.jsonl.
f"error_kind={record.get('error_kind') or 'none'}"
)
# A failing cell's error_kind names the category; the detail names the cause.
# Bounded because a session-error detail carries stdout/stderr tails.
MAX_CELL_DETAIL_CHARS = 1200
def cell_failure_detail_line(
task_id: str,
arm: str,
run_idx: int,
record: Mapping[str, Any],
secrets: Sequence[str] = (),
) -> str | None:
"""The redacted reason a cell failed, or None when it succeeded.
Without this the log says only ``error_kind=plan-evidence-invalid`` and the
reason stays locked in results.jsonl, which is an uploaded artifact rather
than something a watcher can read while the sweep is still running.
"""
if not record.get("error_kind"):
return None
detail = record.get("error_detail")
if detail in (None, "", {}, []):
return None
rendered = detail if isinstance(detail, str) else json.dumps(detail, default=str, sort_keys=True)
rendered = redact_text(rendered, secrets).replace("\n", "")
if len(rendered) > MAX_CELL_DETAIL_CHARS:
rendered = f"{rendered[:MAX_CELL_DETAIL_CHARS]}…[truncated {len(rendered) - MAX_CELL_DETAIL_CHARS} chars]"
return f"[{task_id}][{arm}][run {run_idx}] detail: {rendered}"
def aggregate(records: list[dict[str, Any]]) -> dict[str, Any]:
"""Median metrics + resolve rate across repeated runs of one task+arm.
Session/infra-error rows are excluded from the medians (they measured
nothing); ``valid_runs``/``excluded_runs`` make the exclusion visible.
"""
valid = [
r
for r in records
if r.get("error_kind") not in EXCLUDED_ERROR_KINDS and r.get("review_evidence_valid") is not False
]
metrics = (*USAGE_FIELDS, "duration_s", "num_turns", *CHURN_FIELDS)
out: dict[str, Any] = {m: statistics.median(r.get(m, 0) for r in (valid or [{}])) for m in metrics}
# cost_usd can be None (unmeasured) on an otherwise-valid run; a single
# unmeasured run makes the whole median unavailable so the gate won't rank
# a candidate on a cost that was never actually captured.
valid_costs = [r.get("cost_usd") for r in valid]
out["cost_usd"] = (
None if (not valid or any(cost is None for cost in valid_costs)) else statistics.median(valid_costs)
)
fresh = [r for r in records if not r.get("reused")]
out["fresh_attempts"] = len(fresh)
out["execution_failures"] = sum(1 for r in fresh if execution_failed(r))
out["evidence_failures"] = sum(1 for r in fresh if evidence_failed(r))
# Admissible means the harness delivered a trustworthy measurement. It says
# nothing about whether the answer was right, which is the whole point.
#
# Count the rows that failed NEITHER way rather than subtracting both
# counters: run_arm keeps a pre-existing session error and still marks the
# review evidence invalid, so one row can land in both. Subtracting it twice
# drove an arm holding real measurements to admissible=0, which arm_health
# reads as UNUSABLE and enforce_measurement_health then fails the sweep on.
out["admissible"] = sum(1 for r in fresh if not execution_failed(r) and not evidence_failed(r))
out["health_reasons"] = sorted(
{
str(r.get("error_kind"))
for r in fresh
if r.get("error_kind") in EXECUTION_FAILURE_KINDS or r.get("error_kind") in EVIDENCE_FAILURE_KINDS
}
)
out["resolved"] = sum(1 for r in records if r["resolved"])
# Reused rows are last generation's measurement. The health canary below has
# to ask whether THIS environment worked, so it needs the freshly-run count.
out["resolved_fresh"] = sum(1 for r in records if r["resolved"] and not r.get("reused"))
out["runs"] = len(records)
out["valid_runs"] = len(valid)
out["excluded_runs"] = len(records) - len(valid)
out["transcripts_missing"] = sum(1 for r in records if r.get("transcript_missing"))
out["class"] = records[0].get("class", "")
error_kinds: dict[str, int] = {}
for r in records:
kind = r.get("error_kind")
if kind:
error_kinds[kind] = error_kinds.get(kind, 0) + 1
out["error_kinds"] = error_kinds
review_metrics = (
"review_true_positives",
"review_false_positives",
"review_false_negatives",
"review_precision",
"review_recall",
"review_f1",
"review_weighted_precision",
"review_weighted_recall",
"review_weighted_f1",
"review_blocker_recall",
"review_severity_accuracy",
"review_category_accuracy",
"review_grounded_evidence",
)
if any("review_weighted_f1" in record for record in valid):
for metric in review_metrics:
values = [record[metric] for record in valid if record.get(metric) is not None]
reducer = min if metric == "review_blocker_recall" else statistics.median
out[metric] = reducer(values) if values and len(values) == len(valid) else None
verdicts = [record.get("review_verdict_correct") for record in valid]
out["review_verdict_correct"] = (
all(value is True for value in verdicts)
if verdicts and all(isinstance(value, bool) for value in verdicts)
else None
)
controls = [record["review_clean_control"] for record in valid if "review_clean_control" in record]
out["review_clean_control"] = all(controls) if controls and len(controls) == len(valid) else None
clean_passes = [record["review_clean_pass"] for record in valid if "review_clean_pass" in record]
out["review_clean_pass"] = all(clean_passes) if clean_passes and len(clean_passes) == len(valid) else None
return out
def savings(baseline: dict[str, Any], workflow: dict[str, Any]) -> dict[str, Any]:
"""Percent saved by the workflow arm per metric (positive = cheaper)."""
out: dict[str, Any] = {}
for metric in (*USAGE_FIELDS, "cost_usd", "duration_s"):
base = baseline.get(metric)
arm = workflow.get(metric)
if base is None or arm is None:
out[metric] = None
else:
out[metric] = round(100 * (base - arm) / base, 1) if base else 0.0
return out
@dataclass(frozen=True)
class ArmHealth:
"""What the harness observed for one arm this sweep, before any judgement."""
arm: str
fresh_attempts: int
admissible: int
execution_failures: int
evidence_failures: int
reasons: tuple[str, ...]
@property
def measured(self) -> bool:
"""False when only reused rows exist - current health is UNKNOWN, not good."""
return self.fresh_attempts > 0
@property
def status(self) -> str:
"""UNKNOWN / OBSERVED_OK / DEGRADED / UNUSABLE.
DEGRADED is the distinction that matters: an arm with both admissible
measurements and observed failures produced usable evidence but did not
run reliably. Reporting that as healthy is how a partly-broken sweep
looks fine. It is diagnostic here - only UNUSABLE is fatal - so this
patch changes what is reported, not what is eligible.
"""
if not self.measured:
return "UNKNOWN"
failures = self.execution_failures + self.evidence_failures
if self.admissible == 0 and failures > 0:
return "UNUSABLE"
if failures > 0:
return "DEGRADED"
return "OBSERVED_OK"
@property
def unhealthy(self) -> bool:
"""Every fresh attempt failed to execute or to produce usable evidence.
Deliberately not "resolved zero tasks". A reviewer can be wrong about
every task in a hard corpus with the harness working perfectly; that is
a valid negative and belongs to the quality gate, not here.
"""
return self.status == "UNUSABLE"
def arm_health(results: dict[str, dict[str, dict[str, Any]]], arms: set[str]) -> dict[str, ArmHealth]:
"""Fold per-task aggregates into one health observation per arm."""
health: dict[str, ArmHealth] = {}
for arm in sorted(arms):
rows = [task_arms[arm] for task_arms in results.values() if arm in task_arms]
if not rows:
continue
reasons: set[str] = set()
for row in rows:
reasons.update(row.get("health_reasons") or ())
health[arm] = ArmHealth(
arm=arm,
fresh_attempts=sum(int(r.get("fresh_attempts", 0)) for r in rows),
admissible=sum(int(r.get("admissible", 0)) for r in rows),
execution_failures=sum(int(r.get("execution_failures", 0)) for r in rows),
evidence_failures=sum(int(r.get("evidence_failures", 0)) for r in rows),
reasons=tuple(sorted(reasons)),
)
return health
def unhealthy_arms(results: dict[str, dict[str, dict[str, Any]]], arms: set[str]) -> list[ArmHealth]:
"""Arms whose every fresh attempt failed to execute or to produce evidence."""
return [h for h in arm_health(results, arms).values() if h.unhealthy]
def unmeasured_arms(results: dict[str, dict[str, dict[str, Any]]], arms: set[str]) -> list[str]:
"""Arms with no fresh attempt at all - reported as unknown, never as healthy."""
return [h.arm for h in arm_health(results, arms).values() if not h.measured]
def enforce_measurement_health(
results: dict[str, dict[str, dict[str, Any]]], arms: set[str]
) -> dict[str, ArmHealth]:
"""Report every arm's measurement status; abort only on UNUSABLE.
Runs after report.md and promotion.json are written, so a failing sweep
still leaves its evidence behind. Reports cause as undetermined: an empty
artifact establishes that evidence is unusable, not why - naming a mount
failure here would be a guess the recorded rows do not support.
"""
health = arm_health(results, arms)
for arm in sorted(health):
observed = health[arm]
reasons = f" reason={','.join(observed.reasons)}" if observed.reasons else ""
print(
f"[measurement-health] {arm}: {observed.status} "
f"fresh_attempts={observed.fresh_attempts} admissible={observed.admissible} "
f"execution_failures={observed.execution_failures} "
f"evidence_failures={observed.evidence_failures}{reasons}"
)
unusable = [h for h in health.values() if h.unhealthy]
if unusable:
detail = "; ".join(f"{h.arm} ({h.fresh_attempts} fresh attempt(s))" for h in unusable)
print(
f"[measurement-health] {detail} produced no usable measurement this sweep. "
"cause=undetermined — see error_detail in results.jsonl. Exiting non-zero rather "
"than reporting a quiet no-promotion."
)
raise SystemExit(1)
return health
def broken_incumbent_arms(
results: dict[str, dict[str, dict[str, Any]]],
incumbent_arms: set[str],
) -> list[str]:
"""LEGACY, NON-AUTHORITATIVE. Superseded by ``enforce_measurement_health``.
Kept only so its historical behaviour stays documented and testable while
the replacement settles; it has no production caller. Do not wire it into a
health decision - it infers a broken environment from a resolution count,
which a reviewer facing a hard corpus falsifies. Remove once the
measurement-health path has run in CI.
Incumbent arms that resolved nothing across every task they ran.
An incumbent arm is the currently-shipped, presumably-working skill: if it
resolves NOTHING across every task it ran, that reads as an environment or
harness failure (missing trusted interpreter, stale skill fingerprint,
sandbox misconfiguration), not a skill regression. A candidate merely
underperforming is a normal, expected outcome and must not trip this —
only checking incumbents keeps that distinction.
Deliberately does NOT require valid_runs > 0 per task: an incumbent that
fails every run with an excluded-but-non-systemic error_kind (e.g.
"evidence-unverified", which the outage-streak breaker explicitly resets
on rather than accumulates) would otherwise never accumulate a single
valid run and sail through silently — the exact "quiet no-promotion"
outcome this guard exists to catch, and arguably worse than the
some-runs-resolved-zero case since here nothing completed at all.
aggregate() never marks an excluded/unverifiable row resolved=True, so
resolved == 0 alone already covers both cases.
A reused row proves last generation's environment worked, not this one's, so
the count consulted here is ``resolved_fresh``. Without that, an arm whose
cells were all reused always looks healthy and the canary can never fire —
which is exactly when a broken environment would go unnoticed. The sweep
keeps one paid cell per incumbent arm so this count is never vacuous.
"""
present = incumbent_arms & {arm for arms in results.values() for arm in arms}
return sorted(
arm
for arm in present
if all(arms[arm].get("resolved_fresh", arms[arm]["resolved"]) == 0 for arms in results.values() if arm in arms)
)
def _cost_cell(value: Any) -> str:
return "n/a" if value is None else f"{value:.4f}"
def _review_metric_cell(value: Any) -> str:
return "n/a" if value is None else f"{value:.3f}"
def render_report(results: dict[str, dict[str, dict[str, Any]]]) -> str:
"""results: {task_id: {arm: aggregate}} → markdown report."""
lines = [
"# gitnexus workflow benchmark",
"",
"Medians across runs; savings rows = (baseline arm) / baseline per arm.",
"A negative saving means that arm spent more than baseline. churn =",
"files/+insertions/deletions vs the worktree's starting commit.",
"",
"**WARNING:** token columns count only each arm's main-loop session —",
"subagent spend is invisible to them and flatters subagent-heavy arms.",
"cost $ is the only column that includes subagent spend; to rank token",
"efficiency, sum usage from the session transcripts instead",
"(dedup events sharing one message.id).",
"",
"| task | class | arm | resolved | input | cache_create | cache_read | output | cost $ | wall s | turns | churn | errors |",
"| --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- |",
]
for task_id, arms in results.items():
for arm, agg in arms.items():
excluded = agg.get("excluded_runs", 0)
resolved_cell = f"{agg['resolved']}/{agg.get('valid_runs', agg['runs'])}"
if excluded:
resolved_cell += f" ({excluded} excluded)"
error_cell = ", ".join(f"{kind}×{count}" for kind, count in sorted(agg.get("error_kinds", {}).items()))
lines.append(
f"| {task_id} | {agg['class']} | {arm} | {resolved_cell} "
f"| {agg['input_tokens']:.0f} | {agg['cache_creation_input_tokens']:.0f} "
f"| {agg['cache_read_input_tokens']:.0f} | {agg['output_tokens']:.0f} "
f"| {_cost_cell(agg['cost_usd'])} | {agg['duration_s']:.0f} | {agg['num_turns']:.0f} "
f"| {agg['diff_files']:.0f}/+{agg['diff_insertions']:.0f}/{agg['diff_deletions']:.0f} "
f"| {error_cell} |"
)
for arm in arms:
if arm != "baseline" and "baseline" in arms:
s = savings(arms["baseline"], arms[arm])
lines.append(
f"| {task_id} | {arms[arm]['class']} | **{arm} savings %** | — "
f"| {s['input_tokens']} | {s['cache_creation_input_tokens']} "
f"| {s['cache_read_input_tokens']} | {s['output_tokens']} "
f"| {_na(s['cost_usd'])} | {s['duration_s']} | — | — | — |"
)
lines.append("")
if any("review_clean_control" in agg for arms in results.values() for agg in arms.values()):
lines.extend(
[
"## Review quality",
"",
"| case | arm | TP | FP | FN | precision | recall | blocker recall | weighted F1 | grounding |",
"| --- | --- | ---: | ---: | ---: | ---: | ---: | ---: | ---: | ---: |",
]
)
for task_id, arms in results.items():
for arm, agg in arms.items():
if "review_clean_control" not in agg:
continue
lines.append(
f"| {task_id} | {arm} | {agg['review_true_positives']:.1f} "
f"| {agg['review_false_positives']:.1f} | {agg['review_false_negatives']:.1f} "
f"| {_review_metric_cell(agg['review_precision'])} "
f"| {_review_metric_cell(agg['review_recall'])} "
f"| {_review_metric_cell(agg['review_blocker_recall'])} "
f"| {_review_metric_cell(agg['review_weighted_f1'])} "
f"| {_review_metric_cell(agg['review_grounded_evidence'])} |"
)
lines.append("")
all_aggs = [agg for arms in results.values() for agg in arms.values()]
excluded_total = sum(agg.get("excluded_runs", 0) for agg in all_aggs)
if excluded_total:
lines.append(
f"{excluded_total} run(s) hit session/infra errors or had unverifiable "
"evidence and were excluded "
"from medians and resolve denominators — see error_kind in results.jsonl."
)
missing_total = sum(agg.get("transcripts_missing", 0) for agg in all_aggs)
if missing_total:
lines.append(
f"{missing_total} run(s) had no locatable session transcript or it was "
"unreadable, so they were excluded from promotion evidence "
"(skill_invoked=null in results.jsonl)."
)
lines.append(
"Session ids for every run are in results.jsonl — open the matching "
"transcript to see where each arm spent its tokens."
)
return "\n".join(lines)
# ─── Main ────────────────────────────────────────────────────────────────────
def worker_count(value: str) -> int:
"""``--workers`` as a 1..MAX_WORKERS int, rejected at parse time."""
workers = int(value)
if not 1 <= workers <= MAX_WORKERS:
raise argparse.ArgumentTypeError(f"must be between 1 and {MAX_WORKERS}")
return workers
def build_parser() -> argparse.ArgumentParser:
parser = argparse.ArgumentParser(description=__doc__)
parser.add_argument("--tasks", required=True, type=Path)
parser.add_argument("--runs", type=int, default=1)
parser.add_argument(
"--workers",
type=worker_count,
default=1,
help=f"cells of one task to run at once (default 1, fully serial; max "
f"{MAX_WORKERS}). Size this to the machine: a cell that loses CPU to "
"its siblings takes longer, and a session that reaches its timeout is "
"an excluded run the promotion gate refuses to work with. Above 1 the "
"cells run on worker threads, so Ctrl-C no longer reaches the code "
"owning a sandboxed process and abandons the running cells instead of "
"cleaning up after them.",
)
parser.add_argument(
"--outage-streak",
type=int,
default=DEFAULT_OUTAGE_STREAK,
help="abort the sweep after this many consecutive session/infra/cleanup "
"failures (0 disables the circuit breaker)",
)
parser.add_argument(
"--arms",
nargs="+",
default=["workflow", "workflow_direct", "baseline"],
choices=[
"workflow",
"candidate_workflow",
"workflow_direct",
"candidate_workflow_direct",
"candidate_review",
"ce_workflow",
"ce_workflow_direct",
"review",
"ce_review",
"baseline",
"baseline_nomcp",
],
)
parser.add_argument("--claude-bin", default="claude")
parser.add_argument(
"--ce-plugin-dir",
type=Path,
default=None,
help="operator-supplied Compound Engineering plugin directory; required for ce_* arms",
)
parser.add_argument(
"--ce-plugin-version",
default=None,
help="exact Compound Engineering plugin version; required for ce_* arms",
)
parser.add_argument(
"--timeout",
type=int,
default=SESSION_TIMEOUT_SECONDS,
help="per session, seconds",
)
parser.add_argument("--out", type=Path, default=None)
parser.add_argument(
"--model",
required=True,
help="named, versioned model passed to every `claude --model` invocation",
)
parser.add_argument(
"--effort",
choices=("low", "medium", "high", "xhigh", "max"),
default="xhigh",
help="reasoning effort passed to every `claude --effort` invocation",
)
parser.add_argument(
"--proposer-model",
default=None,
help="model that generated the candidate overlay (recorded for provenance)",
)
parser.add_argument(
"--base-url",
default=None,
help="ANTHROPIC_BASE_URL override — point at an Anthropic-compatible "
"proxy (see free-model.litellm.yaml) to run on a free model",
)
parser.add_argument(
"--anthropic-api-key",
"--auth-token",
dest="auth_token",
default=anthropic_api_key_from_environ(),
help="Anthropic API key for Claude Code sessions (prefer "
"GITNEXUS_BENCH_ANTHROPIC_API_KEY). Not a Claude Code OAuth token. "
"Legacy --auth-token / GITNEXUS_BENCH_AUTH_TOKEN is still accepted.",
)
parser.add_argument(
"--openai-api-key",
default=openai_api_key_from_environ(),
help="OpenAI API key; starts a loopback Anthropic-compatible proxy "
"(prefer GITNEXUS_BENCH_OPENAI_API_KEY). The key never enters the sandbox.",
)
parser.add_argument(
"--include-expensive",
action="store_true",
help="include scenarios marked expensive: true (excluded by default)",
)
parser.add_argument(
"--candidate-overlay",
type=Path,
default=None,
help="directory mirroring one promotable .claude/skills/gitnexus-* tree; applied only to candidate_* arms",
)
parser.add_argument(
"--promotion-metric",
choices=PROMOTION_METRICS,
default="cost_usd",
help="efficiency metric used by the deterministic candidate gate; "
"cost_usd (default) is the only CLI-reported number that includes "
"subagent spend — token metrics count only the main loop",
)
parser.add_argument("--promotion-min-runs", type=int, default=3)
parser.add_argument("--promotion-min-improvement", type=float, default=5.0)
parser.add_argument("--promotion-max-task-regression", type=float, default=20.0)
parser.add_argument("--task-bindings-json", default=None, help=argparse.SUPPRESS)
parser.add_argument("--promotion-target-bases-json", default=None, help=argparse.SUPPRESS)
parser.add_argument("--unsafe-no-bwrap", action="store_true", help=argparse.SUPPRESS)
parser.add_argument(
"--reuse-results",
type=Path,
default=None,
help="prior wfbench results dir whose incumbent/CE rows may be reused "
"when model, effort, tasks, oracles, skill bytes, and CE plugin still "
"match. Candidate arms always run. Used by evolve.py so a weekly "
"generation does not re-pay for an unchanged comparator.",
)
return parser
def main() -> None:
parser = build_parser()
args = parser.parse_args()
try:
args.model = normalized_model_identifier(args.model)
args.proposer_model = (
normalized_model_identifier(args.proposer_model, flag="--proposer-model")
if args.proposer_model is not None
else None
)
task_document = yaml.safe_load(args.tasks.read_text())
if not isinstance(task_document, Mapping) or not isinstance(task_document.get("tasks"), list):
raise ValueError("task file must contain a tasks list")
tasks, skipped_expensive = select_tasks(
task_document["tasks"],
include_expensive=args.include_expensive,
)
oracle_snapshots = capture_task_oracles(tasks)
expected_task_bindings = json.loads(args.task_bindings_json) if args.task_bindings_json else None
if expected_task_bindings is not None and not isinstance(expected_task_bindings, list):
raise ValueError("--task-bindings-json must contain a list")
supplied_promotion_target_bases = (
json.loads(args.promotion_target_bases_json) if args.promotion_target_bases_json else {}
)
if not isinstance(supplied_promotion_target_bases, dict) or not all(
isinstance(path, str) and isinstance(digest, str)
for path, digest in supplied_promotion_target_bases.items()
):
raise ValueError("--promotion-target-bases-json must contain a string mapping")
ce_plugin_config = validate_ce_plugin_inputs(
args.arms,
args.ce_plugin_dir,
args.ce_plugin_version,
)
except (OSError, SandboxError, ValueError, yaml.YAMLError) as exc:
parser.error(str(exc))
raise AssertionError("ArgumentParser.error() returned unexpectedly")
candidate_arms = [arm for arm in args.arms if arm in CANDIDATE_ARMS]
if candidate_arms and args.candidate_overlay is None:
parser.error("candidate_* arms require --candidate-overlay")
if args.candidate_overlay is not None and not candidate_arms:
parser.error("--candidate-overlay requires at least one candidate_* arm")
for candidate_arm in candidate_arms:
incumbent_arm = CANDIDATE_ARMS[candidate_arm]
if incumbent_arm not in args.arms:
parser.error(f"{candidate_arm} must be paired with {incumbent_arm}")
if args.runs < 1 or args.promotion_min_runs < 1:
parser.error("--runs and --promotion-min-runs must be positive")
candidate_overlay = args.candidate_overlay.expanduser().absolute() if args.candidate_overlay is not None else None
overlay_digest = candidate_overlay_digest(candidate_overlay) if candidate_overlay is not None else None
if candidate_overlay is not None:
required_candidates = required_candidate_arms(candidate_overlay)
required_arms = [arm for candidate in required_candidates for arm in (CANDIDATE_ARMS[candidate], candidate)]
if required_candidates == ["candidate_review"]:
required_arms.insert(0, "ce_review")
if args.arms != required_arms:
parser.error("candidate overlay requires exactly these paired arms: " + " ".join(required_arms))
try:
promotion_target_bases = committed_destination_base_digests(candidate_overlay)
except ValueError as exc:
# Overlay adds a promotion target with no committed base — a clean
# CLI error, not a traceback.
parser.error(str(exc))
raise AssertionError("ArgumentParser.error() returned unexpectedly")
if supplied_promotion_target_bases and supplied_promotion_target_bases != promotion_target_bases:
parser.error("--promotion-target-bases-json does not match the committed incumbent")
else:
if supplied_promotion_target_bases:
parser.error("--promotion-target-bases-json requires --candidate-overlay")
promotion_target_bases = {}
if args.unsafe_no_bwrap and os.environ.get("CI"):
parser.error("--unsafe-no-bwrap is forbidden when CI is set")
if args.unsafe_no_bwrap and args.arms != ["ce_review", "review", "candidate_review"]:
parser.error("--unsafe-no-bwrap is restricted to the paired review arms")
try:
if args.unsafe_no_bwrap:
bwrap_bin = preflight_unsafe_host()
sandbox_backend = "host-unsafe"
print(
"WARNING: --unsafe-no-bwrap runs sessions directly on the host with no "
"containment; model and verifier processes can access the host filesystem, "
"network, and credentials.",
file=sys.stderr,
)
else:
bwrap_bin = preflight_bubblewrap()
sandbox_backend = "bwrap"
require_claude_sandbox_helpers()
runtime_mounts = trusted_gitnexus_runtime_mounts()
except SandboxError as exc:
parser.error(str(exc))
raise AssertionError("ArgumentParser.error() returned unexpectedly")
gateway = attach_openai_gateway(args)
try:
gateway.__enter__()
except ValueError as exc:
parser.error(str(exc))
raise AssertionError("ArgumentParser.error() returned unexpectedly")
try:
with cancellation_scope(handle_signals=True) as cancel_event:
_run_sweep(
args,
cancel_event=cancel_event,
parser=parser,
tasks=tasks,
skipped_expensive=skipped_expensive,
oracle_snapshots=oracle_snapshots,
expected_task_bindings=expected_task_bindings,
ce_plugin_config=ce_plugin_config,
bwrap_bin=bwrap_bin,
sandbox_backend=sandbox_backend,
runtime_mounts=runtime_mounts,
candidate_arms=candidate_arms,
candidate_overlay=candidate_overlay,
overlay_digest=overlay_digest,
promotion_target_bases=promotion_target_bases,
)
finally:
gateway.__exit__(None, None, None)
def _comparator_reuse_expectation(
*,
args: argparse.Namespace,
tasks: Sequence[Any],
task_bindings: Sequence[Mapping[str, Any]],
oracle_snapshots: Sequence[Any],
asset_snapshots: Mapping[str, Any],
sandbox_backend: str,
ce_plugin_snapshot: CePluginSnapshot | None,
) -> ComparatorReuseExpectation:
"""Bind this sweep's immutable identity for comparator-row reuse."""
skill_digests: dict[str, str | None] = {}
for arm in args.arms:
execution = CANDIDATE_ARMS.get(arm, arm)
if execution in EVALUATED_ARM_SKILLS:
skill_digests[arm] = skill_fingerprint(HARNESS_ROOT, execution)
else:
skill_digests[arm] = None
task_locks: dict[str, TaskReuseBinding] = {}
for task, binding, oracle in zip(tasks, task_bindings, oracle_snapshots, strict=True):
task_locks[str(task["id"])] = TaskReuseBinding(
task_base_sha=str(binding["resolved_sha"]),
task_prompt_digest=task_prompt_digest(task),
oracle_digest=oracle.digest,
oracle_command_digest=oracle.command_digest,
oracle_manifest_digest=oracle.manifest_digest,
task_asset_manifest_digest=getattr(
asset_snapshots.get(str(task["id"])), "manifest_digest", None
),
sandbox_dependency_manifest_digest=getattr(
asset_snapshots.get(str(task["id"])), "dependency_manifest_digest", None
),
)
return ComparatorReuseExpectation(
model=args.model,
effort=args.effort,
sandbox_backend=sandbox_backend,
runtime_digest=current_runtime_digest(),
now=datetime.now(UTC),
max_age=default_reuse_max_age(),
tasks=task_locks,
skill_digests=skill_digests,
ce_plugin_version=ce_plugin_snapshot.version if ce_plugin_snapshot is not None else None,
ce_plugin_manifest_digest=(
ce_plugin_snapshot.manifest_digest if ce_plugin_snapshot is not None else None
),
)
def task_has_planned_paid_cells(
task: Mapping[str, Any],
*,
arms: Sequence[str],
runs: int,
reusable_rows: Mapping[tuple[str, str, int], object],
reuse_source: Path | None,
) -> bool:
"""True when at least one planned cell is not a reusable comparator row."""
task_id = str(task["id"])
for run_idx in range(runs):
for arm in arms:
if reuse_source is None or (task_id, arm, run_idx) not in reusable_rows:
return True
return False
def drop_canary_reuse_key(
reusable_rows: dict[CellKey, dict[str, Any]],
*,
arm: str,
tasks: Sequence[Mapping[str, Any]],
runs: int,
) -> CellKey | None:
"""Drop one reusable cell so an incumbent arm still measures THIS sweep.
An arm reused end to end measures nothing about today's environment, and
arm_health would then be reading last week's health.
Counted against the cells this sweep PLANS, not every key reuse selection
returned: selection accepts any non-negative prior run index, so a results
directory produced with more runs than this invocation leaves extra keys.
Comparing against those made the check false exactly when it mattered, and
the canary silently stopped firing while every planned cell stayed reused.
Returns the dropped key, or None when the arm already has a paid cell.
"""
planned_keys = [(str(task["id"]), arm, run_idx) for task in tasks for run_idx in range(runs)]
arm_keys = sorted(key for key in planned_keys if key in reusable_rows)
if not arm_keys or len(arm_keys) != len(planned_keys):
return None
dropped = arm_keys[0]
del reusable_rows[dropped]
return dropped
def next_graph_prefetch_target(
remaining: Sequence[tuple[Mapping[str, Any], Mapping[str, Any]]],
*,
arms: Sequence[str],
runs: int,
reusable_rows: Mapping[tuple[str, str, int], object],
reuse_source: Path | None,
ready_keys: set[tuple[str, str]],
) -> tuple[Mapping[str, Any], Mapping[str, Any], tuple[str, str]] | None:
"""Next later task that still needs a clone template and sanitized graph."""
for task, binding in remaining:
if not task_has_planned_paid_cells(
task,
arms=arms,
runs=runs,
reusable_rows=reusable_rows,
reuse_source=reuse_source,
):
continue
key = (str(binding["repo_identity"]), str(binding["resolved_sha"]))
if key in ready_keys:
continue
return task, binding, key
return None
@dataclass(frozen=True)
class GraphBuildEnv:
"""Per-sweep state every graph build shares, and the caches it fills.
The four dicts are the sweep's memo of what has already been built, keyed by
(repo, sha). They are mutable by design and are written by both the sweep
thread and the prefetch thread, which is safe only because a build is
started for a key exactly once and joined before that key is read.
"""
trees: Path
task_asset_cache: TaskAssetCache
claude_bin: Path | str
bwrap_bin: Path | str
sandbox_backend: str
runtime_mounts: Sequence[ReadOnlyMount]
clone_templates: dict[tuple[str, str], tuple[Path, str]]
clone_template_errors: dict[tuple[str, str], BaseException]
graph_snapshots: dict[tuple[str, str], SanitizedGraphSnapshot]
graph_snapshot_errors: dict[tuple[str, str], BaseException]
def ready_keys(self) -> set[tuple[str, str]]:
"""Keys whose build has already been attempted, successfully or not."""
return (
set(self.clone_templates)
| set(self.clone_template_errors)
| set(self.graph_snapshots)
| set(self.graph_snapshot_errors)
)
def ensure_task_graph(
*,
task: Mapping[str, Any],
repo: Path,
task_sha: str,
graph_key: tuple[str, str],
env: GraphBuildEnv,
) -> None:
"""Build one SHA's sanitized clone template and graph. Idempotent per key."""
if graph_key in env.graph_snapshots or graph_key in env.graph_snapshot_errors:
return
try:
validate_no_prebuilt_graph_assets(task)
if graph_key not in env.clone_templates and graph_key not in env.clone_template_errors:
template = make_worktree(repo, task_sha, env.trees)
template_head = sanitize_clone_for_hidden_oracles(template)
env.clone_templates[graph_key] = (template, template_head)
clone_template: Path | None = None
template_head: str | None = None
if graph_key in env.clone_templates:
clone_template, template_head = env.clone_templates[graph_key]
if graph_key in env.clone_template_errors:
env.graph_snapshot_errors[graph_key] = env.clone_template_errors[graph_key]
return
env.graph_snapshots[graph_key] = prepare_sanitized_graph(
task,
repo=repo,
resolved_sha=task_sha,
parent=env.trees,
cache=env.task_asset_cache,
claude_bin=env.claude_bin,
bwrap_bin=env.bwrap_bin,
sandbox_backend=env.sandbox_backend,
runtime_mounts=env.runtime_mounts,
clone_template=clone_template,
sanitized_head=template_head,
)
except (ManagedProcessError, OSError, SandboxError, RuntimeError, ValueError) as exc:
env.graph_snapshot_errors[graph_key] = exc
env.clone_template_errors.setdefault(graph_key, exc)
@dataclass
class GraphPrefetch:
"""In-flight clone+graph build for a later task SHA."""
key: tuple[str, str]
thread: threading.Thread
def join(self) -> None:
self.thread.join()
def prefetch_next_graph(
*,
task: Mapping[str, Any],
binding: Mapping[str, Any],
graph_key: tuple[str, str],
env: GraphBuildEnv,
cancel_event: threading.Event,
) -> GraphPrefetch:
"""Start clone+graph prep for the next unpaid SHA during paid sessions."""
repo = Path(binding["repo_identity"])
task_sha = str(binding["resolved_sha"])
def run() -> None:
if cancel_event.is_set():
return
print(f"[prefetch_next_graph] clone+graph for {task_sha}")
ensure_task_graph(task=task, repo=repo, task_sha=task_sha, graph_key=graph_key, env=env)
# copy_context, as the worker pool already does at _run_wave: a plain Thread
# does not inherit ContextVars, so without this every run_managed inside the
# graph build resolves _CANCELLATION to None and ignores the shared cancel.
thread = threading.Thread(target=copy_context().run, args=(run,), name="prefetch_next_graph", daemon=False)
thread.start()
return GraphPrefetch(key=graph_key, thread=thread)
def _run_sweep(
args: argparse.Namespace,
*,
parser: argparse.ArgumentParser,
tasks: list[Any],
skipped_expensive: list[str],
oracle_snapshots: Any,
expected_task_bindings: Any,
ce_plugin_config: Any,
bwrap_bin: Any,
sandbox_backend: str,
runtime_mounts: Any,
candidate_arms: list[str],
candidate_overlay: Path | None,
overlay_digest: str | None,
promotion_target_bases: dict[str, str],
cancel_event: threading.Event | None = None,
) -> None:
cancel_event = cancel_event or threading.Event()
out_dir = args.out or Path("results") / time.strftime("wfbench-%Y%m%d-%H%M%S")
out_dir.mkdir(parents=True, exist_ok=True)
results_path = out_dir / "results.jsonl"
selected_ids = [task["id"] for task in tasks]
print(
f"selected {len(selected_ids)} task(s): {', '.join(selected_ids)}; "
f"skipped {len(skipped_expensive)} expensive task(s): "
f"{', '.join(skipped_expensive) if skipped_expensive else 'none'}"
)
results: dict[str, dict[str, dict[str, Any]]] = {}
outage_streak = 0
outage_tripped = False
# Progress accounting for the sweep. A generation runs for hours and each
# cell is a full set of agent sessions, so the log needs to say what is in
# flight and how much is left, not only what already finished.
total_cells = len(tasks) * args.runs * len(args.arms)
started_cells = 0
sweep_started = time.monotonic()
with (
tempfile.TemporaryDirectory(prefix="wfbench-trees-") as trees,
TaskAssetCache(Path(trees) / ".task-assets") as task_asset_cache,
staged_ce_plugin_snapshot(
ce_plugin_config,
destination_parent=Path(trees),
) as ce_plugin_snapshot,
):
try:
task_bindings = resolve_task_bindings(
tasks,
expected_task_bindings,
oracle_snapshots=oracle_snapshots,
task_asset_cache=task_asset_cache,
)
except (OSError, SandboxError, ValueError) as exc:
parser.error(str(exc))
raise AssertionError("ArgumentParser.error() returned unexpectedly")
# Asset snapshots are built here, up front, for two reasons. Comparator
# reuse has to compare this sweep's task-asset and dependency digests
# against the prior row's, and those digests do not exist until the
# snapshot does. Building them all before any cell or prefetch thread
# starts also keeps TaskAssetCache single-threaded, which is what its own
# "plain dict, read-then-write race" comment asks for.
asset_snapshots: dict[str, TaskAssetSnapshot] = {}
asset_snapshot_errors: dict[str, BaseException] = {}
for _task, _binding in zip(tasks, task_bindings, strict=True):
try:
asset_snapshots[str(_task["id"])] = task_asset_cache.prepare(
_task,
repo=Path(_binding["repo_identity"]),
resolved_sha=_binding["resolved_sha"],
expected_dependency_binding=_binding,
)
except (OSError, SandboxError, ValueError) as exc:
asset_snapshot_errors[str(_task["id"])] = exc
reuse_source = args.reuse_results.expanduser().resolve() if args.reuse_results is not None else None
reusable_rows: dict[tuple[str, str, int], dict[str, Any]] = {}
if reuse_source is not None:
if reuse_source == out_dir.resolve():
parser.error("--reuse-results cannot be this sweep's --out directory")
raise AssertionError("ArgumentParser.error() returned unexpectedly")
results_file = reuse_source / "results.jsonl"
if results_file.is_symlink() or not results_file.is_file():
parser.error("--reuse-results must contain a regular results.jsonl")
raise AssertionError("ArgumentParser.error() returned unexpectedly")
reusable_rows = select_reusable_comparator_rows(
load_result_rows(results_file),
expected=_comparator_reuse_expectation(
args=args,
tasks=tasks,
task_bindings=task_bindings,
oracle_snapshots=oracle_snapshots,
asset_snapshots=asset_snapshots,
sandbox_backend=sandbox_backend,
ce_plugin_snapshot=ce_plugin_snapshot,
),
)
# Keep one paid cell per incumbent arm. A generation that reuses an
# arm end to end measures nothing about today's environment, and
# arm_health would then be reading last week's health.
# One cell per arm is the cheapest thing that keeps the canary real.
incumbent_arms = [arm for arm in args.arms if arm not in candidate_arms]
for arm in incumbent_arms:
dropped = drop_canary_reuse_key(reusable_rows, arm=arm, tasks=tasks, runs=args.runs)
if dropped is not None:
print(
f"reuse-results: keeping one paid {arm} cell "
f"({dropped[0]} run {dropped[2]}) so incumbent health is measured this sweep"
)
print(
f"reuse-results {reuse_source}: {len(reusable_rows)} comparator "
f"cell(s) match this sweep; candidate arms always run"
)
graph_env = GraphBuildEnv(
trees=Path(trees),
task_asset_cache=task_asset_cache,
claude_bin=args.claude_bin,
bwrap_bin=bwrap_bin,
sandbox_backend=sandbox_backend,
runtime_mounts=runtime_mounts,
clone_templates={},
clone_template_errors={},
graph_snapshots={},
graph_snapshot_errors={},
)
graph_prefetch: GraphPrefetch | None = None
sweep_rows = list(zip(tasks, task_bindings, oracle_snapshots, strict=True))
def _join_graph_prefetch() -> None:
nonlocal graph_prefetch
if graph_prefetch is not None:
graph_prefetch.join()
graph_prefetch = None
try:
for index, (task, task_binding, oracle_snapshot) in enumerate(sweep_rows):
if outage_tripped or cancel_event.is_set():
break
repo = Path(task_binding["repo_identity"])
task_sha = task_binding["resolved_sha"]
per_arm: dict[str, list[dict[str, Any]]] = {a: [] for a in args.arms}
planned = [(run_idx, arm) for run_idx in range(args.runs) for arm in args.arms]
reused_records: list[tuple[int, str, dict[str, Any]]] = []
paid_cells: list[tuple[int, str]] = []
for run_idx, arm in planned:
prior = reusable_rows.get((task["id"], arm, run_idx))
if prior is None or reuse_source is None:
paid_cells.append((run_idx, arm))
continue
try:
reused_records.append(
(
run_idx,
arm,
materialize_reused_row(prior, source_dir=reuse_source, dest_dir=out_dir),
)
)
except (OSError, SandboxError, ValueError) as exc:
print(
f"[{task['id']}][{arm}][run {run_idx}] comparator reuse "
f"failed ({exc}); running a paid cell"
)
paid_cells.append((run_idx, arm))
asset_snapshot = asset_snapshots.get(str(task["id"]))
asset_snapshot_error: BaseException | None = asset_snapshot_errors.get(str(task["id"]))
graph_key = (str(repo), task_sha)
if graph_prefetch is not None and graph_prefetch.key == graph_key:
_join_graph_prefetch()
if paid_cells:
ensure_task_graph(
task=task, repo=repo, task_sha=task_sha, graph_key=graph_key, env=graph_env
)
graph_snapshot = graph_env.graph_snapshots.get(graph_key)
graph_snapshot_error = graph_env.graph_snapshot_errors.get(graph_key)
clone_template, template_head = graph_env.clone_templates.get(graph_key, (None, None))
cell_context = TaskCellContext(
task=task,
oracle_snapshot=oracle_snapshot,
repo=repo,
task_sha=task_sha,
graph_snapshot=graph_snapshot,
graph_snapshot_error=graph_snapshot_error,
asset_snapshot=asset_snapshot,
asset_snapshot_error=asset_snapshot_error,
args=args,
out_dir=out_dir,
ce_plugin_snapshot=ce_plugin_snapshot,
trees_dir=Path(trees),
bwrap_bin=bwrap_bin,
sandbox_backend=sandbox_backend,
runtime_mounts=runtime_mounts,
candidate_overlay=candidate_overlay,
overlay_digest=overlay_digest,
clone_template=clone_template,
sanitized_head=template_head,
)
def announce(run_idx: int, arm: str) -> None:
nonlocal started_cells
started_cells += 1
print(
f"[{task['id']}][{arm}][run {run_idx}] starting "
f"({started_cells}/{total_cells}, {(time.monotonic() - sweep_started) / 60:.0f}m elapsed)"
)
def keep(run_idx: int, arm: str, record: dict[str, Any]) -> None:
per_arm[arm].append(record)
with results_path.open("a") as fh:
# Redact any API token a session-error stderr_tail echoed
# into error_detail before it enters the uploaded
# results.jsonl artifact (transcripts are redacted; this
# sink was not).
fh.write(redact_text(json.dumps(record), credential_secrets(args)) + "\n")
print(cell_progress_line(task["id"], arm, run_idx, record))
failure = cell_failure_detail_line(task["id"], arm, run_idx, record, credential_secrets(args))
if failure:
print(failure)
for run_idx, arm, record in reused_records:
started_cells += 1
print(
f"[{task['id']}][{arm}][run {run_idx}] reused comparator "
f"({started_cells}/{total_cells}, {(time.monotonic() - sweep_started) / 60:.0f}m elapsed)"
)
keep(run_idx, arm, record)
if paid_cells and graph_prefetch is None and not cancel_event.is_set():
target = next_graph_prefetch_target(
[(later_task, later_binding) for later_task, later_binding, _ in sweep_rows[index + 1 :]],
arms=args.arms,
runs=args.runs,
reusable_rows=reusable_rows,
reuse_source=reuse_source,
ready_keys=graph_env.ready_keys(),
)
if target is not None:
later_task, later_binding, later_key = target
graph_prefetch = prefetch_next_graph(
task=later_task,
binding=later_binding,
graph_key=later_key,
env=graph_env,
cancel_event=cancel_event,
)
# A reused success is evidence the pipeline can produce a good row,
# so it resets the consecutive-failure count the same way a paid
# success does. Leaving reused rows out let a streak carry across
# them and trip on stale history.
for _run_idx, _arm, _record in reused_records:
outage_streak = systemic_outage_streak(_record.get("error_kind"), outage_streak)
outage_streak, outage_tripped = sweep_task_cells(
paid_cells,
workers=args.workers,
run=partial(run_cell, cell_context),
on_start=announce,
on_record=keep,
outage_streak=outage_streak,
outage_limit=args.outage_streak,
cancel_event=cancel_event,
)
results[task["id"]] = {a: aggregate(rs) for a, rs in per_arm.items() if rs}
finally:
_join_graph_prefetch()
selection_report = [
"## Run provenance",
"",
f"Benchmark model: `{args.model}`",
f"Proposer model: `{args.proposer_model}`",
f"Reasoning effort: `{args.effort}`",
f"Selected tasks ({len(selected_ids)}): {', '.join(selected_ids)}",
(
f"Skipped expensive tasks ({len(skipped_expensive)}): "
+ (", ".join(skipped_expensive) if skipped_expensive else "none")
),
]
if ce_plugin_snapshot is not None:
selection_report.append(
f"Compound Engineering plugin: `{ce_plugin_snapshot.version}` (`{ce_plugin_snapshot.manifest_digest}`)"
)
# outage first: the breaker now sets cancel_event to stop in-flight background
# work, so testing cancellation first would relabel every outage a cancellation.
if outage_tripped:
selection_report.append("Sweep aborted: partial evidence; promotion is disabled.")
elif cancel_event.is_set():
selection_report.append("Sweep cancelled: partial evidence; promotion is disabled.")
report = render_report(results) + "\n\n" + "\n".join(selection_report) + "\n"
(out_dir / "report.md").write_text(report)
if candidate_arms:
promotion_generated_at = datetime.now(UTC)
promotion = {
"generated_at": promotion_generated_at.isoformat(),
"evidence_expires_at": (promotion_generated_at + timedelta(days=EVIDENCE_MAX_AGE_DAYS)).isoformat(),
"benchmark_model": args.model,
"proposer_model": args.proposer_model,
"effort": args.effort,
"candidate_origin": ("model-proposer" if args.proposer_model is not None else "manual-initial-overlay"),
"candidate_overlay": str(candidate_overlay),
"candidate_overlay_digest": overlay_digest,
"target_base_digests": promotion_target_bases,
"required_candidate_arms": candidate_arms,
"selected_tasks": task_bindings,
"ce_plugin": ce_plugin_snapshot.provenance if ce_plugin_snapshot is not None else None,
**promotion_evidence(
results,
model=args.model,
complete=not outage_tripped and not cancel_event.is_set(),
policy=promotion_policy(
candidate_arms,
metric=args.promotion_metric,
min_runs=args.promotion_min_runs,
min_improvement_pct=args.promotion_min_improvement,
max_task_regression_pct=args.promotion_max_task_regression,
),
),
}
(out_dir / "promotion.json").write_text(json.dumps(promotion, indent=2) + "\n")
print(f"\n{report}\n\nWritten to {out_dir}/")
# Health is judged on whether fresh attempts EXECUTED and produced usable
# evidence - never on how many tasks they resolved. Review arms used to be
# excluded here because "resolved zero" is quality signal for a reviewer
# facing a hard corpus; with the inference corrected they are included
# again, which is what lets an all-artifacts-empty run be caught at all.
# ce_review is named explicitly: it is a comparator, not a candidate, so it
# is absent from CANDIDATE_ARMS and would otherwise go unclassified.
enforce_measurement_health(results, set(CANDIDATE_ARMS.values()) | {"review", "ce_review"})
if outage_tripped:
# Non-zero exit so a driver (evolve.py) treats the partial benchmark as a
# failed run and halts instead of proposing from outage-truncated evidence.
# Checked before cancel_event because the breaker sets it (see above), and
# an outage must keep exit 1 rather than becoming the 130 of a Ctrl-C.
raise SystemExit(1)
if cancel_event.is_set():
raise SystemExit(130)
if __name__ == "__main__":
main()