GitNexus/eval/workflow_bench/runner.py

"""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 .baseline_guidance import GITNEXUS_UNAVAILABLE_NOTE
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,
    )
    # All arms load ordinary repository context (CLAUDE.md) through the same
    # startup mode. The no-MCP arm instead disables skills/commands and
    # restricts its exact tool surface, with an empty MCP config and no
    # supplied graph/runtime/CLI mounts.
    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),
        "settings_json": sandbox.settings_json,
        "strict_mcp_config": True,
        "mcp_config_json": host_text(sandbox_mcp_config(unsafe_host=backend == "host-unsafe")),
        "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"]) + "\n\n" + GITNEXUS_UNAVAILABLE_NOTE,
                worktree,
                disallowed_tools=["Skill", "mcp__gitnexus"],
                **{
                    **common,
                    "disable_slash_commands": True,
                    "mcp_config_json": '{"mcpServers":{}}',
                    "allowed_tools": [
                        tool for tool in allowed_agent_tools(implementation=True, include_mcp=False) if tool != "Skill"
                    ],
                },
            )
        )
    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
    execution_arm = CANDIDATE_ARMS.get(arm, arm)
    gitnexus_available = execution_arm != "baseline_nomcp"
    try:
        if ctx.asset_snapshot_error is not None:
            raise RuntimeError(f"task asset snapshot preparation failed: {ctx.asset_snapshot_error}")
        if gitnexus_available:
            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)
        if gitnexus_available:
            assert ctx.graph_snapshot is not None
            ctx.graph_snapshot.materialize(worktree, sanitized_head=sanitized_head)
        dependency_mounts = stage_task_assets(
            task,
            repo=ctx.repo,
            clone=worktree,
            snapshot=ctx.asset_snapshot,
        )
        gitnexus_mounts = (
            [*ctx.runtime_mounts, isolated_gitnexus_registry_mount(worktree, ctx.trees_dir)]
            if gitnexus_available
            else []
        )
        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,
                *gitnexus_mounts,
                *ce_mounts,
            ],
            preflight=False,
            backend=ctx.sandbox_backend,
            gitnexus_available=gitnexus_available,
        ) 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 task SHAs may predate the selected release skills.
            # Seed them before fingerprinting/setup so every task evaluates
            # the same release prose, with candidate overlays applied later.
            if execution_arm in ("review", "workflow", "workflow_direct"):
                seed_evaluated_skills(
                    getattr(args, "gitnexus_root", None) or 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 gitnexus_available and ctx.graph_snapshot is not None else None
            ),
            "sanitized_graph_manifest_digest": (
                ctx.graph_snapshot.manifest_digest if gitnexus_available and 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(
        "--gitnexus-root",
        type=Path,
        default=HARNESS_ROOT,
        help="checkout supplying the pinned GitNexus runtime and evaluated skills (defaults to the harness checkout)",
    )
    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(root=args.gitnexus_root)
    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(getattr(args, "gitnexus_root", None) or 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()