mirror of
https://github.com/abhigyanpatwari/GitNexus.git
synced 2026-09-15 23:32:49 +00:00
e2da8d90ce
* feat(eval): a scriptable stand-in for Anthropic and OpenAI Every defect this harness shipped last round was invisible to its own tests for one reason: the tests exercised a layer BELOW where the code runs. The usage log was never written because the proxy is a subprocess with a constructed environment. The callback could not be imported because LiteLLM loads it by path, not as a package. Failures went unrecorded because only the async hook was overridden. CI or review caught all three; no unit test could, because each called the function directly instead of driving the path that calls it. This closes that gap without spending money. It speaks the two wire protocols the harness actually depends on - Anthropic Messages, streaming and not, and OpenAI Responses - so a run can go through the real sandbox, the real CLI, the real gateway and the real usage callback with only the model faked. The runner already supports pointing at it: --base-url is the same path the free-model proxy documentation uses. Scripted rather than simulated. A test decides what the model says, which tools it asks for, and exactly what usage it reports. That last part is what makes provider-native accounting testable at all: real cache hits are not reproducible on demand, but a declared cache_read of 44,000 is. One Reply served down both protocols is also the cleanest demonstration that the same billed work is stated as a sum on one side and as a whole on the other. Tool blocks are the mechanism for artifact-producing cells. The CLI runs what it is asked to run, so a scripted Write block makes it write that file inside the sandbox for real - no model deciding anything. The end-to-end test drives the real proxy against the mock and asserts the usage log records the provider's own arithmetic through the Anthropic-shaped translation. It SKIPS here, because litellm's console script is absent in this environment, so it is unverified until CI runs it - the same footing the bubblewrap canary started on, and that one found a real bug on its first CI run. Not yet built: driving a whole sweep against this. That needs a scripted reply sequence that carries a cell to a scored artifact, which is the next step and the point of the exercise. 668 eval tests pass, 17 skipped; the two test_model_gateway.py failures are the pre-existing environmental ones. * test(eval): run a real session against the scripted provider The mock only proves something once the harness runs against it. This adds the stand-in CLI and the first integration tests that use it, so a session goes through the real code with only the model faked. tests/fixtures/fake_claude.py does what the CLI does at the two boundaries the harness depends on: it calls ANTHROPIC_BASE_URL for a turn, EXECUTES the tool blocks that come back, and prints the stream-json sequence the parent parses. Everything between - the session runner, the event-stream parse, the usage extraction, the artifact capture, the scorer - stays real. Four tests, chosen for the layers that have actually broken here: the usage a provider reported survives to the row, a scripted Write produces an artifact parse_review_output accepts, the prompt the harness meant to send is what arrived, and an upstream 529 lands as a failed session rather than a usable measurement. Writing the stand-in found two things worth keeping. The prompt arrives on STDIN under "-p --input-format text"; scanning argv for a non-flag token picks up a flag's value instead, and the prompt-fidelity test is what caught it. And three of these tests had been holding a sandbox they never applied, since no command_prefix is passed - that implied coverage which was not there, so the sandbox is gone from them and stays only in the artifact test, which needs its review directory. What these do NOT cover, checked rather than assumed: making the stand-in write in place instead of atomically still passes. On the host-unsafe backend there is no read-only mount to refuse it, so the atomic-write requirement remains a bubblewrap mount property that only the real-sandbox canary can prove. Dropping cache_read from the recorded usage does fail, so that half is genuinely pinned. 672 eval tests pass, 17 skipped; the two test_model_gateway.py failures are the environmental ones. * fix(eval): the usage adapter read a shape the callback never receives Running the gateway against the scripted provider proved the accounting merged in #3220 does not work, and the same run showed why nothing had caught it. LiteLLM does not hand a logger the upstream body. It normalises usage into its own Chat-Completions-shaped object first, so an OpenAI Responses reply reaches the callback as prompt_tokens / prompt_tokens_details.cached_tokens - never the input_tokens / input_tokens_details the shipped adapter reads. Every field came back unknown. The observed call_type is "anthropic_messages" as well, because Claude Code calls the Anthropic-shaped endpoint, so canonical_provider returned None and normalize_usage would have refused outright. Both were assumptions about a boundary I had only read about. The unit tests agreed with them because their fixture was written in the same wrong shape, so producer and consumer were consistent and both wrong - the exact failure the producer/consumer round trip exists to catch, one layer further out. Adds a LITELLM_NORMALIZED adapter for the object that actually arrives. The arithmetic is still OpenAI's - prompt_tokens is the whole, the details are subsets - so ordinary input is recovered by subtraction. The Responses adapter stays for a raw upstream body, which the mock still serves and tests directly. An unrecognised provider is still refused rather than guessed. The fixtures now carry the measured shape, and the end-to-end test asserts it through a real proxy: 48k prompt tokens with 44k cached is read back as 3k ordinary rather than as silence. 676 eval tests pass, 16 skipped, none failing. * test(eval): run a whole sweep offline, with negative controls The layers between a model turn and a promotion decision had never been exercised together. Unit tests covered each alone, and the paid runs that would have covered the composition kept dying, so the contracts BETWEEN them went unverified - which is where this harness has repeatedly shipped bugs. Drives runner.main() the way the workflow does. Real task selection, hidden oracle capture, sandbox, CLI subprocess, artifact capture, scoring against the oracle, aggregation, health guard and promotion gate. Only the model is scripted. Getting to green meant satisfying nine real contracts nothing had exercised end to end, and each failure was the harness correctly refusing bad evidence: --unsafe-no-bwrap is restricted to the paired review arms; ce_* needs a plugin carrying ce-plan, ce-work and ce-code-review; candidate_* needs an overlay; the clone needs .gitnexus/meta.json with indexedAt and lastCommit; the evidence gate needs a Skill request with a non-error result; review findings need exactly ten fields with severity in critical/high/medium/low; and the hidden labels use a DIFFERENT schema from the review output - line_start/line_end, six fields. That last one only a real run surfaces. Three negative controls, because a scorer that cannot be wrong measures nothing. A finding in the wrong place is tp=0 fp=1 fn=1 and oracle-failed, while its evidence stays VALID - being wrong is a quality result, not a broken measurement. Approving defective code is a miss with no false positive, and precision is None rather than 0, because it is undefined with no predictions. One run cannot promote: the gate says it needs three valid paired runs. A fourth control exists because a mutation demanded it. Forcing skill_was_invoked_events to return True left every other test here passing, so nothing pinned the gate that separates measuring a SKILL from measuring a model. Writing it turned up behaviour worth recording rather than assuming: a skill-not-invoked row still carries its score AND still counts toward the arm median, because aggregate() drops EXCLUDED_ERROR_KINDS and evidence_valid=False and skill-not-invoked is neither. The health guard stops the sweep, so a single-run sweep cannot promote on it, but a mixed run's median would include a cell whose skill never ran. Pinned as-is so it cannot change silently in either direction; changing it is a promotion-semantics decision, not a test fix. Two provisioning steps are stubbed and neither is harness logic: the pinned runtime mounts (no node_modules in a worktree) and the sanitized graph build (needs the gitnexus CLI at a mounted path). Containment is host-unsafe here; bubblewrap stays with the real-sandbox canary. 681 eval tests pass, 16 skipped, none failing. Runs in ~18s. * fix(eval): an uninvoked skill must not move the arm's quality median Found by the offline sweep: a skill-not-invoked row still carried its score into the arm's quality median. aggregate()'s filter dropped EXCLUDED_ERROR_KINDS and evidence_valid=False, and skill-not-invoked is neither, so an arm could be credited for a review it never performed with the skill under test - which is the one thing an arm exists to measure. Excluded from the QUALITY metrics only. Cost and duration still count that row, because the session really ran and really was billed, and the promotion gate still sees it, because it has its own vocabulary for a candidate that never loaded its skill. Two wider fixes were tried and abandoned, both because the tests said so rather than because I reasoned it out first. Reusing the health guard's evidence_failed predicate also excluded transcript-missing rows, but test_aggregate_excludes_session_error_rows_from_medians pins those as counting: that session ran, only its transcript is unverifiable. Excluding the row from `valid` outright turned a candidate whose skill never loaded from keep_incumbent into insufficient_evidence - the safety property held either way, but the decision vocabulary is promotion semantics and not mine to change on a measurement fix. Mutation-checked: putting the rows back into the quality median fails the new test. Both directions asserted, since a filter that excludes everything would also pass - a wrong-but-valid review still moves quality, because being wrong is exactly what a quality median should reflect. 682 eval tests pass, 16 skipped. * test(eval): run the offline sweep unstubbed in the job that can, and probe CLI identity Items 5 and 6 turned out to be one change. The containment (ubuntu) job already installs bubblewrap, the pinned Claude CLI, node_modules and a built GitNexus - everything the sweep's two provisioning stubs stand in for. So the stubs are not a property of the test, only of a machine that lacks those things. GITNEXUS_REQUIRE_FULL_SWEEP=1 makes the sweep run with nothing stubbed: real containment instead of --unsafe-no-bwrap, the real runtime mounts, the real sanitized graph. Set in that job, following the GITNEXUS_REQUIRE_BWRAP_CANARY pattern already there. The gate FAILS on a missing piece rather than degrading to the stubbed path, which is the point - a green tick that silently tested less is what the bubblewrap canary was written to prevent. Verified both states here: default green, and gate-on fails on this machine rather than skipping, since it cannot create user namespaces. Item 7 is an experiment, not an answer. Per-cell attribution needs an identifier that travels WITH the request, because one proxy serves the whole sweep and anything read from its environment is identical for every call. What the real CLI sends is not documented anywhere I can check, and guessing a wire format is exactly how the last three accounting bugs happened. So the probe drives the REAL pinned CLI against the mock and records the identity-bearing headers and body keys that arrive. It asserts only that a request was made; the recorded evidence is the deliverable, and the job log preserves it. Skips without CLAUDE_CANARY_BIN. Two guards caught this rather than review: the repo pins the containment job's env and its exact test list, so both had to be updated deliberately - which is the guard working, not friction. 682 eval tests pass, 17 skipped. * test(eval): make the offline sweep cross-task, so a scheduler change is checkable The sweep fixture had one task, and a single task cannot show the thing a cross-task scheduler changes: waves are per-task, so ordering, packing and a breaker spanning a task boundary are all invisible with one. A second task with its defect in a DIFFERENT file, and its own hidden labels, makes per-task routing observable. The scripted reply is now task-aware, which matters for the same reason: replying with the first task's finding scores the second task wrong. The load-bearing assertion is that each task scored against ITS OWN oracle. That is the dangerous failure mode of interleaving cells from different tasks - a mis-routed context or artifact scores one task against another's labels, and every row still looks green. Mutation-checked: pointing every cell at the first task's oracle snapshot fails it. This is the safety net the packed-scheduler wiring needs. Measured earlier against the real sweep_packed_cells, that change is worth -27% on a cold sweep and -37% weekly, with breaker fidelity holding at three injected failure positions - but it restructures a 125-line loop across ~92 names that also holds graph prefetch, reuse selection, oracle staging and the canary drop. Landing that on top of a one-task fixture would have been unverifiable, which is why this comes first and separately. 682 eval tests pass, 17 skipped. * fix(eval): commit the stand-in CLI's executable bit The file was created and chmod +x'd locally, but committed 100644 - so the mode existed only in my working tree. Any fresh checkout, CI included, gets a non-executable file and every cell dies with "required executable is not an executable regular file". Found by accident: checking out origin/main and back to compare a flaky test restored the file from the index and stripped the bit, which turned 5 green tests into 9 failures. Without that detour this would have failed on the first CI run instead. Same shape as the bugs this branch exists to catch - something that works only because of local state, breaking where the code actually runs. * fix(eval): apply code review findings Seven local reviewers and an independent cross-model pass. The headline is that a fix I added in this branch was worse than the gap it closed. Reverted the aggregate() quality-median filter. Excluding skill-not-invoked rows from the quality metrics left valid_runs and excluded_runs still counting them, so the promotion gate saw N clean runs while the median came from fewer. The dropped rows are systematically an arm's worst, so it biased toward PROMOTING - reproduced: one real run at 0.9 plus two uninvoked rows at 0.0 gave the gate 3 valid runs, zero exclusions and a 0.9 median, flipping keep_incumbent to promote. Three verdict fields compounded it: they are all() reducers still reading the wider set, so one uninvoked cell flipped a whole arm. Five reviewers found the two halves independently. Closing it honestly needs a scored-run count plus a paired-equality check in the gate, which is promotion semantics rather than an aggregation fix. The gap is now pinned by a test that states why the half-fix was reverted. Stopped forging the absence of CI. The runner refuses --unsafe-no-bwrap when CI is set because that mode runs sessions with bypassPermissions behind a boundary its own docstring calls "not a security boundary"; the sweep test deleted CI to get past it, so eval / locked pytest ran an uncontained agent sweep on the runner holding the checkout and credentials. It skips under CI instead - the containment job still runs it for real with GITNEXUS_REQUIRE_FULL_SWEEP=1. The stand-in CLI was lying in three ways. It never set is_error, so a refused write read as a completed one. It had no Skill branch at all, so honoring is_error revealed the evidence gate had been satisfied by a tool the fixture never ran - the gate was measuring the fixture, not a skill. And a reply with no usage became four zero-valued fields plus a fabricated cost, which is exactly the unknown-is-not-zero confusion the accounting it feeds exists to prevent. A provider failure also crashed the subprocess with no terminal result event. The identity probe never ran anywhere. test_mock_provider.py was in no job's file list, and the only job setting CLAUDE_CANARY_BIN runs a fixed list. My commit message claimed the next containment run would produce the answer; it would not have. Now wired in, with the CI-shape test updated to pin it. Also: the regex-miss fallback wrote a predictable name in shared /tmp through a symlink-following stage, now scoped to the test's own directory; and the canonical_provider docstring plus the callback comment still asserted a call_type branch the code no longer has. Deferred as design decisions rather than review fixes: the containment sweep uses the stand-in CLI rather than the pinned real one, the full-sweep path bypasses the gateway so native usage accounting is unexercised there, _normalize_litellm duplicates the Responses algorithm, and OPENAI_RESPONSES is now unreachable from canonical_provider. 682 eval tests pass, 17 skipped, ruff clean. * fix(eval): carry scripted tools over the Responses protocol Review round on #3235. Three real items; five more were already fixed ina20f94e1cand are answered on their threads rather than re-fixed. `_openai_response` emitted only an `output_text` item and never read `reply.tools`, so a reply scripted with a Write or Skill crossed the gateway with the tool silently dropped. Responses is the protocol the gateway is configured for BECAUSE it carries tool use, so the mock was wrong about the wire on the one path that matters most. Function-call items now accompany the message. Mutation-checked: reverting the emit fails the new test on "the scripted tool must cross the Responses path". The artifact session now takes `command_prefix` and `require_pid_namespace` from the sandbox the way `run_arm` does instead of calling `run_claude` bare. On host-unsafe `command_prefix_for` returns `[]` by construction, so this pins the wiring, not the isolation - the comment says so rather than implying more. CodeQL's three unused-variable reports on one line were one finding: a call whose result is entirely discarded. Unpack nothing there. 683 passed, 17 skipped. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com> * fix(eval): forward the usage the provider reported instead of zero-filling it Review round on #3235; all three findings valid. The stand-in CLI defaulted absent cache fields to 0. That fabricated a complete measurement out of an incomplete reply, and the second-order effect was worse than the first: `runner_sessions` requires all four USAGE_FIELDS before it calls a session measured, so a stand-in that always emitted four fields made that guard unfirable from any offline test. It was always satisfied. It now forwards exactly what arrived. `Reply`'s cache fields accept None to script absence, since a consumer that cannot tell "omitted" from "zero" is the bug this harness exists to catch. Mutation-checked: restoring the zero-fill fails the new test. Also corrected a comment claiming aggregate() excludes skill-not-invoked rows from the quality median. It does not - that was the filter reverted ina20f94e1cfor inverting a promotion, and the comment survived the revert describing the opposite of what the test pins. Dropped an unused monkeypatch fixture arg. 684 passed, 17 skipped. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com> * fix(eval): keep an omitted cache field omitted on the Responses wire Review round on #3235. The main finding is a miss in my own previous commit: that one taught the Anthropic path to forward absence instead of zero-filling, but `_openai_response` still serialized both `input_tokens_details` keys unconditionally. Collapsing None to 0 is right for the arithmetic - an unreported field adds nothing to the total - and wrong on the wire, because `_int_or_none` reads an absent key as unknown and a present 0 as a measured zero. So a reply scripted with `cache_read_input_tokens=None` was indistinguishable from a provider-reported zero on exactly one of the two protocols. Half-applying the invariant was arguably worse than not applying it: the Anthropic test passing made the pair look covered. Mutation-checked: restoring the unconditional keys fails the new test. Also: the module docstring claimed the stand-in executes the tool blocks that come back, without noting Bash is stubbed; and dropped an unused tmp_path fixture arg. 685 passed, 17 skipped. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com> * fix(eval): validate every usage field the stand-in forwards Review round on #3235. The guard checked input_tokens and output_tokens for type and sign, but the forwarding comprehension passed the cache fields through unchecked whenever present. The parent's well_formed test only asks whether the four keys are PRESENT, so a negative, boolean, or non-integer cache value rode into a `success` result and was recorded as a usable measurement. Same shape as the previous two rounds: the required half of a pair was handled and the optional half was not. A field good enough to report is good enough to check. Mutation-checked: dropping the added clause fails all three parametrized cases (negative, boolean, string). 688 passed, 17 skipped. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com> * docs(eval): say which stand-in tools execute and which are modelled Review round on #3235. The previous commit's docstring fix said "Write and Skill really run" while correcting the Bash claim. Only Write really runs: Skill validates the request and returns a synthetic result. Fourth round of the same shape - the reported half of a pair gets fixed and the sibling keeps the overclaim. Both docstrings now name each of the three branches and what it actually does. 688 passed, 17 skipped. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com> --------- Co-authored-by: Gergo Magyar <gergomagyar0@gmail.com> Co-authored-by: Claude Opus 5 (1M context) <noreply@anthropic.com>
2741 lines
120 KiB
Python
2741 lines
120 KiB
Python
"""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",
|
||
)
|
||
# NOTE: a skill-not-invoked row still contributes to these medians. That is
|
||
# a real measurement gap - an arm exists to measure a SKILL, and a cell
|
||
# where the skill never ran did not measure it - but the narrow fix is
|
||
# WORSE than the gap, so it is deliberately not applied here.
|
||
#
|
||
# Filtering those rows out of the quality metrics alone leaves valid_runs
|
||
# and excluded_runs counting them, so the promotion gate sees N clean runs
|
||
# while the median was taken over fewer. Because the dropped rows are
|
||
# systematically an arm's worst, that biases toward PROMOTING: measured on
|
||
# one real run at 0.9 plus two uninvoked rows at 0.0, the gate flipped from
|
||
# keep_incumbent to promote. The three verdict fields below compound it -
|
||
# they are all() reducers, so one uninvoked cell flips a whole arm.
|
||
# Closing this honestly needs a scored-run count and a paired-equality
|
||
# check in the gate itself: a promotion-semantics change, not an
|
||
# aggregation fix.
|
||
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()
|