"""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()