GitNexus/eval/workflow_bench/evolution.py

"""Skill-candidate isolation, provenance, and deterministic promotion policy."""

from __future__ import annotations

import hashlib
import os
import secrets
import stat
import statistics
from collections.abc import Sequence
from pathlib import Path, PurePosixPath
from typing import Any

from .process_control import ManagedProcessError
from .proposer_sandbox import (
    SANDBOX_TMP,
    SANDBOX_WORKSPACE,
    SandboxSession,
    build_sandbox_environment,
)

CANDIDATE_ARMS = {
    "candidate_workflow": "workflow",
    "candidate_workflow_direct": "workflow_direct",
    "candidate_review": "review",
}
PROMOTION_SCHEMA_VERSION = 6


def promotion_policy(
    candidate_arms: Sequence[str],
    *,
    metric: str = "cost_usd",
    min_runs: int = 3,
    min_improvement_pct: float = 5.0,
    max_task_regression_pct: float = 20.0,
) -> dict[str, dict[str, Any]]:
    """The exact per-arm policy shared by evidence production and application."""
    if not candidate_arms or len(set(candidate_arms)) != len(candidate_arms):
        raise ValueError("promotion policy requires unique candidate arms")
    policies = {}
    for arm in candidate_arms:
        if arm not in CANDIDATE_ARMS:
            raise ValueError(f"unsupported candidate arm: {arm}")
        policies[arm] = (
            {
                "metric": "review_weighted_f1",
                "min_runs": min_runs,
                "min_improvement": 0.01,
                "quality_rule": "correct verdict on every repeat; minimum blocker recall; no clean-control regression",
            }
            if arm == "candidate_review"
            else {
                "metric": metric,
                "min_runs": min_runs,
                "min_improvement_pct": min_improvement_pct,
                "max_task_regression_pct": max_task_regression_pct,
                "max_failed_task_regression_pct": MAX_FAILED_TASK_REGRESSION_PCT,
                "min_gated_task_ratio": MIN_GATED_TASK_RATIO,
                "quality_rule": "no per-task resolution-rate regression",
            }
        )
    return policies


def promotion_evidence(
    results: dict[str, dict[str, dict[str, Any]]],
    *,
    policy: dict[str, dict[str, Any]],
    model: str | None,
    complete: bool,
) -> dict[str, Any]:
    """Produce discriminated, recomputable decisions, including partial reports."""
    decisions = []
    for candidate, rules in policy.items():
        common = {
            "incumbent_arm": CANDIDATE_ARMS[candidate],
            "candidate_arm": candidate,
            "model": model,
            "min_runs": rules["min_runs"],
        }
        if candidate == "candidate_review":
            decision = evaluate_review_candidate(results, **common, min_improvement=rules["min_improvement"])
        else:
            decision = evaluate_candidate(
                results,
                **common,
                **{
                    key: rules[key]
                    for key in (
                        "metric",
                        "min_improvement_pct",
                        "max_task_regression_pct",
                        "max_failed_task_regression_pct",
                    )
                },
            )
        if not complete:
            decision["decision"] = "insufficient_evidence"
            decision["reasons"].append("sweep aborted; partial evidence cannot promote")
        decisions.append(decision)
    return {
        "schema_version": PROMOTION_SCHEMA_VERSION,
        "run_status": "complete" if complete else "aborted",
        "policy": policy,
        "decisions": decisions,
    }


CANDIDATE_SKILLS = {
    "gitnexus-plan",
    "gitnexus-review",
    "gitnexus-work",
}
# Skills each incumbent arm actually loads in its sessions. An overlay that
# only touches other skills would never be exercised — the gate would decide
# from noise — so such overlays are rejected up front.
ARM_SKILLS = {
    "workflow": ("gitnexus-plan", "gitnexus-work"),
    "workflow_direct": ("gitnexus-work",),
    "review": ("gitnexus-review",),
}
# Repo-local prompts whose bytes are evidence for each executed arm. Keep this
# distinct from ``ARM_SKILLS``: that mapping defines which skills a promotable
# plan/work overlay must exercise, while this mapping also protects read-only
# review evaluation from task setup and review-phase prompt replacement.
EVALUATED_ARM_SKILLS = {
    **ARM_SKILLS,
    "review": ("gitnexus-review",),
}
PROMOTION_METRICS = ("output_tokens", "cost_usd", "duration_s", "num_turns")
# Token/turn metrics come from the CLI's top-level `usage`, which counts ONLY
# the main-loop session. `total_cost_usd` is the only reported number that
# includes subagent spend.
MAIN_LOOP_ONLY_METRICS = frozenset({"output_tokens", "num_turns"})
MAIN_LOOP_ONLY_WARNING = (
    "WARNING: token and turn metrics count only the main-loop session — subagent spend "
    "is invisible to them and systematically flatters subagent-heavy "
    "candidates. Prefer cost_usd (the only CLI-reported field that includes "
    "subagents), or sum usage from the digest-bound transcript_artifacts in "
    "each run output, deduplicating events "
    "that share one message.id."
)
# Failure kinds the prompts under test cause, not the task: the skill never
# ran at all. Both arms failing a task this way is evidence about the skills,
# so such a task stays inside the gate however unresolvable it looks.
SKILL_ATTRIBUTABLE_ERROR_KINDS = frozenset({"skill-not-invoked"})
# Leaving the quality gate is not leaving the spend gate. A candidate may fail
# the same oracle the incumbent fails, but not at a multiple of its cost — an
# ungated task is still real money and still ranks on the metric.
MAX_FAILED_TASK_REGRESSION_PCT = 100.0
# Promotion must rest on a real evidence base. Half the paired tasks is the
# loosest rule the three-task production set can carry: it tolerates the one
# scenario neither arm resolves and refuses a generation that has quietly
# decayed to a single gated task deciding everything.
MIN_GATED_TASK_RATIO = 0.5
EVIDENCE_MAX_AGE_DAYS = 90
MAX_CANDIDATE_OVERLAY_BYTES = 4 * 1024 * 1024
MAX_SKILL_FINGERPRINT_BYTES = 4 * 1024 * 1024
MAX_CANDIDATE_ENTRIES = 256
MAX_CANDIDATE_FILES = 64
MAX_CANDIDATE_PATH_BYTES = 512


def _require_real_directory(path: Path, *, label: str) -> None:
    try:
        metadata = path.lstat()
    except OSError as exc:
        raise ValueError(f"{label} is unavailable: {path}: {exc}") from exc
    if stat.S_ISLNK(metadata.st_mode) or not stat.S_ISDIR(metadata.st_mode):
        raise ValueError(f"{label} must be a real non-symlink directory: {path}")


def _require_directory_chain(root: Path, relative: Path, *, label: str) -> None:
    """Validate each lexical directory without erasing links via resolve()."""

    _require_real_directory(root, label=label)
    current = root
    for part in relative.parts:
        if part in {"", ".", ".."}:
            raise ValueError(f"{label} contains an unsafe path component: {relative}")
        current /= part
        _require_real_directory(current, label=label)


def _bounded_regular_bytes(path: Path, *, limit: int, label: str) -> bytes:
    """Read one bounded regular file without following its leaf link."""

    try:
        before = path.lstat()
    except OSError as exc:
        raise ValueError(f"{label} is unreadable: {path}: {exc}") from exc
    if stat.S_ISLNK(before.st_mode) or not stat.S_ISREG(before.st_mode):
        raise ValueError(f"{label} must be a regular non-symlink file: {path}")
    if before.st_size > limit:
        raise ValueError(f"{label} exceeds the bounded evidence limit")

    descriptor = os.open(path, os.O_RDONLY | getattr(os, "O_NOFOLLOW", 0))
    try:
        opened = os.fstat(descriptor)
        if not stat.S_ISREG(opened.st_mode) or opened.st_dev != before.st_dev or opened.st_ino != before.st_ino:
            raise ValueError(f"{label} changed while opening: {path}")
        chunks: list[bytes] = []
        remaining = limit + 1
        while remaining > 0:
            chunk = os.read(descriptor, min(64 * 1024, remaining))
            if not chunk:
                break
            chunks.append(chunk)
            remaining -= len(chunk)
        content = b"".join(chunks)
        if len(content) > limit:
            raise ValueError(f"{label} exceeds the bounded evidence limit")
        after = os.fstat(descriptor)
        if (
            opened.st_dev,
            opened.st_ino,
            opened.st_size,
            opened.st_mtime_ns,
        ) != (
            after.st_dev,
            after.st_ino,
            after.st_size,
            after.st_mtime_ns,
        ) or len(content) != opened.st_size:
            raise ValueError(f"{label} changed while being read: {path}")
        return content
    finally:
        os.close(descriptor)


def candidate_overlay_payload(overlay: Path) -> tuple[str, list[tuple[PurePosixPath, bytes]]]:
    """Return the sole validated, bounded candidate payload and its digest."""

    root = overlay.expanduser().absolute()
    payload: list[tuple[PurePosixPath, bytes]] = []
    remaining = MAX_CANDIDATE_OVERLAY_BYTES
    for source in candidate_overlay_files(root):
        relative = PurePosixPath(source.relative_to(root).as_posix())
        _require_directory_chain(
            root,
            Path(*relative.parent.parts),
            label="candidate overlay directory",
        )
        content = _bounded_regular_bytes(
            source,
            limit=remaining,
            label="candidate overlay file",
        )
        remaining -= len(content)
        payload.append((relative, content))
    return _fingerprint_payload(payload), payload


def _fingerprint_payload(payload: list[tuple[PurePosixPath, bytes]]) -> str:
    digest = hashlib.sha256()
    for relative_path, content in payload:
        relative = relative_path.as_posix().encode()
        digest.update(len(relative).to_bytes(8, "big"))
        digest.update(relative)
        digest.update(len(content).to_bytes(8, "big"))
        digest.update(content)
    return digest.hexdigest()


def _replace_regular_file(root: Path, relative: Path, content: bytes) -> None:
    """Replace a clone file through validated directory descriptors."""

    if relative.is_absolute() or not relative.parts or ".." in relative.parts:
        raise ValueError(f"candidate destination escapes the clone: {relative}")
    _require_real_directory(root, label="candidate destination root")
    directory_flags = os.O_RDONLY | getattr(os, "O_DIRECTORY", 0) | getattr(os, "O_NOFOLLOW", 0)
    descriptor = os.open(root, directory_flags)
    try:
        for part in relative.parts[:-1]:
            try:
                os.mkdir(part, mode=0o700, dir_fd=descriptor)
            except FileExistsError:
                pass
            try:
                child = os.open(part, directory_flags, dir_fd=descriptor)
            except OSError as exc:
                raise ValueError(
                    f"candidate destination parent must be a real directory: {relative.parent}: {exc}"
                ) from exc
            os.close(descriptor)
            descriptor = child

        leaf = relative.name
        try:
            existing = os.stat(leaf, dir_fd=descriptor, follow_symlinks=False)
        except FileNotFoundError:
            existing = None
        except OSError as exc:
            raise ValueError(f"candidate destination is unreadable: {relative}: {exc}") from exc
        if existing is not None and (stat.S_ISLNK(existing.st_mode) or not stat.S_ISREG(existing.st_mode)):
            raise ValueError(f"candidate destination must be a regular non-symlink file: {relative}")

        temporary = f".wfbench-overlay-{secrets.token_hex(12)}"
        temp_descriptor = os.open(
            temporary,
            os.O_WRONLY | os.O_CREAT | os.O_EXCL | getattr(os, "O_NOFOLLOW", 0),
            0o600,
            dir_fd=descriptor,
        )
        try:
            view = memoryview(content)
            while view:
                written = os.write(temp_descriptor, view)
                if written <= 0:
                    raise OSError("short write while staging candidate overlay")
                view = view[written:]
            os.fchmod(temp_descriptor, 0o644)
        except BaseException:
            try:
                os.unlink(temporary, dir_fd=descriptor)
            except OSError:
                pass
            raise
        finally:
            os.close(temp_descriptor)
        try:
            os.replace(
                temporary,
                leaf,
                src_dir_fd=descriptor,
                dst_dir_fd=descriptor,
            )
        except BaseException:
            try:
                os.unlink(temporary, dir_fd=descriptor)
            except OSError:
                pass
            raise
    finally:
        os.close(descriptor)


def _sandbox_overlay_git(
    sandbox: SandboxSession,
    args: list[str],
    *,
    extra_config: tuple[str, ...] = (),
) -> Any:
    hooks = f"{SANDBOX_TMP}/wfbench-empty-hooks"
    command = [
        "/usr/bin/git",
        "-c",
        "core.fsmonitor=false",
        "-c",
        f"core.hooksPath={hooks}",
        "-c",
        "commit.gpgsign=false",
    ]
    for item in extra_config:
        command.extend(("-c", item))
    command.extend(("-C", SANDBOX_WORKSPACE, *args))
    result = sandbox.run(
        command,
        timeout=60,
        env=build_sandbox_environment(),
    )
    return command, result


def candidate_overlay_files(overlay: Path) -> list[Path]:
    """Return a candidate's files after enforcing the benchmark trust boundary.

    Candidates may change only the canonical repo-local skill prompts. They
    cannot modify task code, tests, or verification commands and thereby game
    the promotion gate.
    """
    overlay = overlay.expanduser().absolute()
    try:
        resolved_overlay = overlay.resolve(strict=True)
    except OSError as exc:
        raise ValueError(f"candidate overlay is not a directory: {overlay}") from exc
    if resolved_overlay != overlay:
        raise ValueError(f"candidate overlay cannot traverse symlinks: {overlay}")
    _require_real_directory(overlay, label="candidate overlay")

    entries: list[Path] = []
    pending = [overlay]
    entry_count = 0
    while pending:
        directory = pending.pop()
        child_directories: list[Path] = []
        try:
            iterator = os.scandir(directory)
        except OSError as exc:
            raise ValueError(f"candidate overlay directory is unreadable: {directory}: {exc}") from exc
        with iterator:
            for item in iterator:
                entry_count += 1
                if entry_count > MAX_CANDIDATE_ENTRIES:
                    raise ValueError(f"candidate overlay exceeds the {MAX_CANDIDATE_ENTRIES}-entry limit")
                path = Path(item.path)
                relative = path.relative_to(overlay)
                if len(relative.as_posix().encode()) > MAX_CANDIDATE_PATH_BYTES:
                    raise ValueError(f"candidate overlay path exceeds {MAX_CANDIDATE_PATH_BYTES} bytes: {relative}")
                if item.is_symlink():
                    raise ValueError(f"candidate overlay cannot contain symlinks: {relative}")
                if item.is_dir(follow_symlinks=False):
                    child_directories.append(path)
                    continue
                if not item.is_file(follow_symlinks=False):
                    raise ValueError(f"candidate overlay entries must be regular files: {relative}")
                entries.append(path)
                if len(entries) > MAX_CANDIDATE_FILES:
                    raise ValueError(f"candidate overlay exceeds the {MAX_CANDIDATE_FILES}-file limit")
        pending.extend(child_directories)

    entries.sort(key=lambda path: path.relative_to(overlay).as_posix())
    if not entries:
        raise ValueError(f"candidate overlay contains no files: {overlay}")

    for path in entries:
        relative = path.relative_to(overlay)
        parts = relative.parts
        if (
            len(parts) < 4
            or parts[:2] != (".claude", "skills")
            or parts[2] not in CANDIDATE_SKILLS
            or path.suffix.lower() != ".md"
        ):
            raise ValueError(
                "candidate overlays may only contain Markdown files under "
                ".claude/skills/gitnexus-{plan,review,work}: "
                f"{relative}"
            )
    return entries


def required_candidate_arms(overlay: Path) -> list[str]:
    """Return the smallest candidate-arm set that exercises every change.

    Plan prompts are loaded only by the two-session workflow. Work prompts are
    loaded by both workflow shapes, so a work candidate must prove itself in
    both rather than inheriting a decision from an untested execution mode.
    """
    overlay = overlay.expanduser().absolute()
    touched = {path.relative_to(overlay).parts[2] for path in candidate_overlay_files(overlay)}
    required: list[str] = []
    if "gitnexus-plan" in touched or "gitnexus-work" in touched:
        required.append("candidate_workflow")
    if "gitnexus-work" in touched:
        required.append("candidate_workflow_direct")
    if "gitnexus-review" in touched:
        required.append("candidate_review")
    return required


def fingerprint_files(root: Path, files: list[Path]) -> str:
    digest = hashlib.sha256()
    for path in files:
        relative = path.relative_to(root).as_posix().encode()
        content = path.read_bytes()
        digest.update(len(relative).to_bytes(8, "big"))
        digest.update(relative)
        digest.update(len(content).to_bytes(8, "big"))
        digest.update(content)
    return digest.hexdigest()


def candidate_overlay_digest(overlay: Path) -> str:
    digest, _ = candidate_overlay_payload(overlay)
    return digest


def _commit_sandbox_paths(
    sandbox: SandboxSession,
    relative_paths: Sequence[str],
    *,
    message: str,
    require_change: bool,
) -> bool:
    """Stage and commit paths inside the outer sandbox.

    Returns True when a commit was created. ``require_change`` keeps the
    candidate-overlay contract: a no-op overlay is an error, while an
    incumbent skill seed may already match the historical tree.
    """

    mkdir_command = ["/bin/mkdir", "-p", f"{SANDBOX_TMP}/wfbench-empty-hooks"]
    mkdir_result = sandbox.run(
        mkdir_command,
        timeout=60,
        env=build_sandbox_environment(),
    )
    if not mkdir_result.ok:
        raise ManagedProcessError(mkdir_command, mkdir_result)
    if not relative_paths:
        if require_change:
            raise ValueError("candidate overlay is byte-identical to the incumbent skills")
        return False

    # Historical review SHAs gitignore `.claude/skills/*` and lack the current
    # per-skill allowlist. Force-add so a seed or overlay of harness-owned
    # skill bytes is not rejected as an ignored path.
    command, added = _sandbox_overlay_git(sandbox, ["add", "-f", "--", *relative_paths])
    if not added.ok:
        raise ManagedProcessError(command, added)
    command, changed = _sandbox_overlay_git(
        sandbox,
        ["diff", "--cached", "--quiet", "--no-ext-diff", "--no-textconv", "--"],
    )
    if changed.returncode == 0:
        if require_change:
            raise ValueError("candidate overlay is byte-identical to the incumbent skills")
        return False
    if changed.returncode != 1:
        raise ManagedProcessError(command, changed)

    command, committed = _sandbox_overlay_git(
        sandbox,
        [
            "commit",
            "--quiet",
            "--no-verify",
            "-m",
            message,
        ],
        extra_config=(
            "user.name=workflow-bench",
            "user.email=workflow-bench@invalid",
        ),
    )
    if not committed.ok:
        raise ManagedProcessError(command, committed)
    return True


def apply_candidate_overlay(
    overlay: Path,
    worktree: Path,
    *,
    sandbox: SandboxSession,
) -> str:
    """Safely copy and commit a prompt candidate inside its outer sandbox."""

    overlay = overlay.expanduser().absolute()
    expected_clone = Path(os.path.abspath(worktree.expanduser()))
    sandbox_clone = Path(os.path.abspath(sandbox.clone.expanduser()))
    if sandbox_clone != expected_clone:
        raise ValueError("candidate sandbox does not bind the requested clone")
    digest, payload = candidate_overlay_payload(overlay)
    relative_paths: list[str] = []
    for relative, content in payload:
        _replace_regular_file(worktree, relative, content)
        relative_paths.append(relative.as_posix())
    _commit_sandbox_paths(
        sandbox,
        relative_paths,
        message="benchmark candidate skill overlay",
        require_change=True,
    )
    return digest


def seed_evaluated_skills(
    source_repo: Path,
    worktree: Path,
    *,
    sandbox: SandboxSession,
    arm: str,
) -> None:
    """Install the current evaluated skill tree into a historical clone.

    Review evolution scores the current (or overlay) ``gitnexus-review`` skill
    against a historical PR checkout. Older SHAs predate that skill, and
    using whatever prose happened to exist at the reviewed commit would make
    the incumbent arm a moving target. Copy the harness checkout's skill
    bytes and commit them before setup so ``git status`` still shows only
    the task patch.
    """

    skill_names = EVALUATED_ARM_SKILLS.get(arm)
    if not skill_names:
        return

    source_repo = source_repo.expanduser().absolute()
    expected_clone = Path(os.path.abspath(worktree.expanduser()))
    sandbox_clone = Path(os.path.abspath(sandbox.clone.expanduser()))
    if sandbox_clone != expected_clone:
        raise ValueError("skill seed sandbox does not bind the requested clone")
    _require_real_directory(source_repo, label="incumbent skill repository")
    if source_repo.resolve(strict=True) != source_repo:
        raise ValueError(f"incumbent skill repository cannot traverse symlinks: {source_repo}")

    relative_paths: list[str] = []
    total = 0
    for skill_name in skill_names:
        _require_directory_chain(
            source_repo,
            Path(".claude") / "skills" / skill_name,
            label="incumbent skill root",
        )
        skill_root = source_repo / ".claude" / "skills" / skill_name
        pending = [skill_root]
        while pending:
            directory = pending.pop()
            try:
                children = list(os.scandir(directory))
            except OSError as exc:
                raise ValueError(f"incumbent skill directory is unreadable: {directory}: {exc}") from exc
            for item in children:
                path = Path(item.path)
                if item.is_symlink():
                    raise ValueError(
                        "incumbent skill seed cannot contain symlinks: "
                        f"{path.relative_to(source_repo)}"
                    )
                if item.is_dir(follow_symlinks=False):
                    pending.append(path)
                    continue
                if not item.is_file(follow_symlinks=False):
                    raise ValueError(
                        "incumbent skill seed entries must be regular files: "
                        f"{path.relative_to(source_repo)}"
                    )
                total += item.stat(follow_symlinks=False).st_size
                if total > MAX_SKILL_FINGERPRINT_BYTES:
                    raise ValueError("incumbent skill seed exceeds the bounded evidence limit")
                relative = Path(".claude") / "skills" / skill_name / path.relative_to(skill_root)
                content = _bounded_regular_bytes(
                    path,
                    limit=MAX_SKILL_FINGERPRINT_BYTES,
                    label="incumbent skill file",
                )
                _replace_regular_file(worktree, relative, content)
                relative_paths.append(PurePosixPath(relative.as_posix()).as_posix())

    _commit_sandbox_paths(
        sandbox,
        relative_paths,
        message="benchmark incumbent review skill",
        require_change=False,
    )


def unexercised_overlay_skills(overlay: Path, candidate_arms: list[str]) -> list[str]:
    """Overlay skills that no selected candidate arm would ever load.

    A gitnexus-lfg-only (or gitnexus-review-only) overlay paired with the
    workflow arms is never read by any benchmarked session, so any promotion
    decision about it would be noise.
    """
    overlay = overlay.expanduser().absolute()
    exercised = {skill for arm in candidate_arms for skill in ARM_SKILLS[CANDIDATE_ARMS[arm]]}
    touched = {path.relative_to(overlay).parts[2] for path in candidate_overlay_files(overlay)}
    return sorted(touched - exercised)


def skill_fingerprint(worktree: Path, arm: str) -> str | None:
    skill_names = EVALUATED_ARM_SKILLS.get(arm)
    if skill_names is None:
        return None

    worktree = worktree.expanduser().absolute()
    _require_real_directory(worktree, label="skill fingerprint worktree")
    _require_directory_chain(
        worktree,
        Path(".claude") / "skills",
        label="skill fingerprint parent",
    )
    for skill_name in skill_names:
        _require_directory_chain(
            worktree,
            Path(".claude") / "skills" / skill_name,
            label="skill fingerprint root",
        )

    entries = sorted(
        (path for skill_name in skill_names for path in (worktree / ".claude" / "skills" / skill_name).rglob("*")),
        key=lambda path: path.relative_to(worktree).as_posix(),
    )
    files: list[Path] = []
    total = 0
    for path in entries:
        metadata = path.lstat()
        if stat.S_ISDIR(metadata.st_mode):
            continue
        if stat.S_ISLNK(metadata.st_mode) or not stat.S_ISREG(metadata.st_mode):
            raise ValueError(f"skill fingerprint input must be a regular non-symlink file: {path}")
        total += metadata.st_size
        if total > MAX_SKILL_FINGERPRINT_BYTES:
            raise ValueError("skill fingerprint input exceeds the bounded evidence limit")
        files.append(path)
    return fingerprint_files(worktree, files)


def evaluate_review_candidate(
    results: dict[str, dict[str, dict[str, Any]]],
    *,
    incumbent_arm: str,
    candidate_arm: str,
    model: str | None,
    min_runs: int = 3,
    min_improvement: float = 0.01,
) -> dict[str, Any]:
    """Quality-first promotion gate for paired read-only review arms."""

    reasons: list[str] = []
    task_rows: list[dict[str, Any]] = []
    insufficient = not model
    regression = False
    improvement = False
    if not model:
        reasons.append("a named --model is required so review evidence cannot drift")

    for task_id, arms in sorted(results.items()):
        if incumbent_arm not in arms or candidate_arm not in arms:
            insufficient = True
            reasons.append(f"{task_id}: both {incumbent_arm} and {candidate_arm} are required")
            continue
        incumbent = arms[incumbent_arm]
        candidate = arms[candidate_arm]
        incumbent_runs = int(incumbent.get("valid_runs", 0))
        candidate_runs = int(candidate.get("valid_runs", 0))
        incumbent_score = incumbent.get("review_weighted_f1")
        candidate_score = candidate.get("review_weighted_f1")
        incumbent_blockers = incumbent.get("review_blocker_recall")
        candidate_blockers = candidate.get("review_blocker_recall")
        incumbent_fp = incumbent.get("review_false_positives")
        candidate_fp = candidate.get("review_false_positives")
        clean = bool(incumbent.get("review_clean_control", candidate.get("review_clean_control", False)))
        incumbent_clean_pass = incumbent.get("review_clean_pass")
        candidate_clean_pass = candidate.get("review_clean_pass")
        task_rows.append(
            {
                "task": task_id,
                "class": incumbent.get("class", ""),
                "incumbent_weighted_f1": incumbent_score,
                "incumbent": dict(incumbent),
                "candidate": dict(candidate),
                "gated": True,
                "candidate_weighted_f1": candidate_score,
                "incumbent_blocker_recall": incumbent_blockers,
                "candidate_blocker_recall": candidate_blockers,
                "incumbent_false_positives": incumbent_fp,
                "candidate_false_positives": candidate_fp,
                "clean_control": clean,
                "incumbent_clean_pass": incumbent_clean_pass,
                "candidate_clean_pass": candidate_clean_pass,
            }
        )
        if (
            incumbent_runs < min_runs
            or candidate_runs < min_runs
            or incumbent_runs != candidate_runs
            or incumbent.get("excluded_runs")
            or candidate.get("excluded_runs")
        ):
            insufficient = True
            reasons.append(
                f"{task_id}: needs {min_runs} valid paired runs with zero exclusions "
                f"(got {incumbent_runs}/{candidate_runs})"
            )
        required_values = (
            (incumbent_fp, candidate_fp, incumbent_clean_pass, candidate_clean_pass)
            if clean
            else (incumbent_score, candidate_score, incumbent_fp, candidate_fp)
        )
        if any(value is None for value in required_values):
            insufficient = True
            reasons.append(f"{task_id}: structured review quality metrics are incomplete")
            continue
        if candidate.get("review_verdict_correct") is None:
            insufficient = True
            reasons.append(f"{task_id}: candidate verdict evidence is incomplete")
        elif candidate["review_verdict_correct"] is not True:
            regression = True
            reasons.append(f"{task_id}: candidate verdict was incorrect on a valid repeat")
        if (incumbent_blockers is None) != (candidate_blockers is None):
            insufficient = True
            reasons.append(f"{task_id}: blocker recall evidence is incomplete")
        if incumbent_blockers is not None and candidate_blockers is not None and float(candidate_blockers) < float(
            incumbent_blockers
        ):
            regression = True
            reasons.append(f"{task_id}: blocker recall regressed")
        if clean and float(candidate_fp) > float(incumbent_fp):
            regression = True
            reasons.append(f"{task_id}: false positives increased on a clean control")
        if clean and bool(incumbent_clean_pass) and not bool(candidate_clean_pass):
            regression = True
            reasons.append(f"{task_id}: clean-control verdict regressed")
        if not clean and float(candidate_score) + 1e-9 < float(incumbent_score):
            regression = True
            reasons.append(f"{task_id}: weighted review score regressed")
        if not clean and float(candidate_score) >= float(incumbent_score) + min_improvement:
            improvement = True

    if not task_rows:
        insufficient = True
        reasons.append("no paired review task results were found")
    if insufficient:
        decision = "insufficient_evidence"
    elif regression:
        decision = "keep_incumbent"
    elif not improvement:
        decision = "keep_incumbent"
        reasons.append("candidate did not improve weighted review quality on any corpus case")
    else:
        decision = "promote"
        reasons.append("candidate improved weighted review quality without blocker or clean-control regression")
    return {
        "candidate_arm": candidate_arm,
        "incumbent_arm": incumbent_arm,
        "decision": decision,
        "metric": "review_weighted_f1",
        "model": model,
        "tasks": task_rows,
        "ungated_tasks": [],
        "reasons": reasons,
    }


def evaluate_candidate(
    results: dict[str, dict[str, dict[str, Any]]],
    *,
    incumbent_arm: str,
    candidate_arm: str,
    model: str | None,
    metric: str = "cost_usd",
    min_runs: int = 3,
    min_improvement_pct: float = 5.0,
    max_task_regression_pct: float = 20.0,
    max_failed_task_regression_pct: float = MAX_FAILED_TASK_REGRESSION_PCT,
) -> dict[str, Any]:
    """Deterministically decide whether a prompt candidate is promotable.

    Resolution is lexicographically primary: a cheaper candidate that fails
    more tasks never wins. With equal quality, the candidate must clear the
    configured median efficiency gain without a large per-task regression.

    A task neither arm can resolve leaves the quality gate, but only on
    evidence: a comparable metric, no skill-not-invoked run, and enough tasks
    left inside the gate to decide anything. It still ranks against the
    failed-task spend cap.
    """
    if metric not in PROMOTION_METRICS:
        raise ValueError(f"unsupported promotion metric: {metric}")

    reasons: list[str] = []
    task_rows: list[dict[str, Any]] = []
    insufficient = False
    quality_regression = False
    quality_floor_failed = False
    efficiency_regression = False

    if not model:
        insufficient = True
        reasons.append("a named --model is required so prompt evidence cannot drift")

    for task_id, arms in sorted(results.items()):
        if incumbent_arm not in arms or candidate_arm not in arms:
            insufficient = True
            reasons.append(f"{task_id}: both {incumbent_arm} and {candidate_arm} are required")
            continue

        incumbent = arms[incumbent_arm]
        candidate = arms[candidate_arm]
        # Session/infra-error rows carry no measured evidence: only VALID runs
        # count toward the run minimum, the pairing check, and resolve rates.
        incumbent_runs = int(incumbent.get("valid_runs", incumbent["runs"]))
        candidate_runs = int(candidate.get("valid_runs", candidate["runs"]))
        incumbent_excluded = int(incumbent.get("excluded_runs", 0))
        candidate_excluded = int(candidate.get("excluded_runs", 0))
        incumbent_rate = incumbent["resolved"] / incumbent_runs if incumbent_runs else 0.0
        candidate_rate = candidate["resolved"] / candidate_runs if candidate_runs else 0.0
        # cost_usd is None when a run's cost was never measured (see
        # runner_sessions.measured_cost): the arm's aggregate cost is then
        # unavailable and must not be ranked on, or a candidate could "win"
        # cheapness it never actually demonstrated.
        raw_incumbent_metric = incumbent.get(metric)
        raw_candidate_metric = candidate.get(metric)
        metric_unavailable = raw_incumbent_metric is None or raw_candidate_metric is None
        incumbent_metric = None if raw_incumbent_metric is None else float(raw_incumbent_metric)
        candidate_metric = None if raw_candidate_metric is None else float(raw_candidate_metric)
        improvement = (
            round(100 * (incumbent_metric - candidate_metric) / incumbent_metric, 1)
            if (not metric_unavailable and incumbent_metric)
            else None
        )
        # A task that both arms measured cleanly and neither ever resolved sits
        # outside both arms' current capability. It carries no quality signal
        # about the candidate, and its metric compares who spent more while
        # failing the same oracle — so gating on it measures the task, not the
        # candidate, and one such task vetoes every future promotion for as
        # long as it stays in the set. Keep it in the evidence, out of the gate,
        # and name it as task health instead.
        #
        # Ungating is itself a claim, so it needs evidence: the failures must
        # be the task's (not a skill that never loaded) and the metric must be
        # comparable, otherwise the task stays gated and the checks below name
        # what is missing.
        fully_measured = (
            incumbent_runs >= min_runs
            and candidate_runs >= min_runs
            and incumbent_runs == candidate_runs
            and not incumbent_excluded
            and not candidate_excluded
        )
        skill_attributable = bool(
            (set(incumbent.get("error_kinds", {})) | set(candidate.get("error_kinds", {})))
            & SKILL_ATTRIBUTABLE_ERROR_KINDS
        )
        mutually_unresolved = fully_measured and not incumbent["resolved"] and not candidate["resolved"]
        gated = not (mutually_unresolved and not skill_attributable and improvement is not None)
        # The floor asks the candidate to be reliable where the incumbent is.
        # On a task the incumbent never resolves there is no reliability to
        # match, and holding partial candidate progress to it punished a
        # candidate for resolving 1 of 3 runs while excusing it for resolving
        # none — the strictly worse result. A skill that never loaded is the
        # exception: those failures belong to the prompts, so the floor applies
        # even with nothing on the incumbent's side to match.
        quality_floor_enforced = bool(incumbent["resolved"]) or skill_attributable
        task_rows.append(
            {
                "task": task_id,
                "class": incumbent.get("class", ""),
                "incumbent_resolved": f"{incumbent['resolved']}/{incumbent_runs}",
                "incumbent": dict(incumbent),
                "candidate": dict(candidate),
                "candidate_resolved": f"{candidate['resolved']}/{candidate_runs}",
                "incumbent_excluded_runs": incumbent_excluded,
                "candidate_excluded_runs": candidate_excluded,
                "candidate_quality_floor_met": candidate_runs > 0 and candidate["resolved"] == candidate_runs,
                "quality_floor_enforced": quality_floor_enforced,
                "incumbent_metric": incumbent_metric,
                "candidate_metric": candidate_metric,
                "improvement_pct": improvement,
                "gated": gated,
                "skill_attributable_failure": skill_attributable,
            }
        )
        if not gated:
            if improvement < -max_failed_task_regression_pct:
                efficiency_regression = True
                reasons.append(
                    f"{task_id}: {metric} regressed {-improvement:.1f}% on a task neither arm resolved, "
                    f"above the {max_failed_task_regression_pct:.1f}% failed-task cap"
                )
            continue

        if incumbent_runs < min_runs or candidate_runs < min_runs:
            insufficient = True
            reasons.append(
                f"{task_id}: needs at least {min_runs} valid runs per arm (got {incumbent_runs}/{candidate_runs})"
            )
        if incumbent_excluded or candidate_excluded:
            insufficient = True
            reasons.append(
                f"{task_id}: promotion requires zero excluded runs in both paired arms "
                f"(got {incumbent_excluded}/{candidate_excluded})"
            )
        if incumbent_runs != candidate_runs:
            insufficient = True
            reasons.append(
                f"{task_id}: paired arms have different valid run counts "
                f"({incumbent_runs}/{candidate_runs} valid; {incumbent_excluded}/{candidate_excluded} excluded)"
            )
        if candidate_rate < incumbent_rate:
            quality_regression = True
            reasons.append(f"{task_id}: resolution regressed from {incumbent_rate:.0%} to {candidate_rate:.0%}")
        if quality_floor_enforced and candidate_runs > 0 and candidate["resolved"] != candidate_runs:
            quality_floor_failed = True
            floor_trigger = (
                "a run never invoked the skill under test"
                if skill_attributable
                else f"the incumbent resolves {incumbent['resolved']}/{incumbent_runs}"
            )
            reasons.append(
                f"{task_id}: candidate must resolve every valid run for the oracle-backed quality floor "
                f"({floor_trigger}; got {candidate['resolved']}/{candidate_runs})"
            )
        if metric_unavailable:
            insufficient = True
            reasons.append(
                f"{task_id}: {metric} was not measured on every run in both paired arms; "
                "cannot rank on it (fix cost capture or choose another metric)"
            )
        elif improvement is None:
            insufficient = True
            reasons.append(f"{task_id}: incumbent {metric} is zero; choose a metric with signal")
        elif improvement < -max_task_regression_pct:
            efficiency_regression = True
            reasons.append(
                f"{task_id}: {metric} regressed {-improvement:.1f}%, above the {max_task_regression_pct:.1f}% task cap"
            )

    ungated_tasks = [row["task"] for row in task_rows if not row["gated"]]
    gated_tasks = [row["task"] for row in task_rows if row["gated"]]
    if ungated_tasks:
        # One line, not one per task: `reasons` is truncated to three entries
        # when it is fed back to the proposer (evolve.summarize_gate), and a
        # growing set of unsolvable tasks must not crowd out the reason the
        # candidate actually won or lost. The full list ships structurally.
        reasons.append(
            f"not gated on {len(ungated_tasks)} task(s) neither arm resolved: {', '.join(ungated_tasks)} "
            f"(evidence base: {len(gated_tasks)}/{len(task_rows)} paired tasks gated)"
        )
    if not task_rows:
        insufficient = True
        reasons.append("no paired task results were found")
    elif not gated_tasks:
        # Every paired task was ungated, so nothing in this generation says
        # anything about candidate quality. Refuse rather than fall through to
        # an efficiency-only verdict on runs that all failed their oracle.
        insufficient = True
        reasons.append("no task supplied quality signal: neither arm resolved a run anywhere in the set")
    elif len(gated_tasks) < MIN_GATED_TASK_RATIO * len(task_rows):
        # Ungating one unsolvable task keeps promotion reachable; ungating most
        # of the set turns "promote" into a verdict from whatever is left.
        insufficient = True
        reasons.append(
            f"promotion evidence base is too thin: {len(gated_tasks)}/{len(task_rows)} paired tasks are gated "
            f"(at least {MIN_GATED_TASK_RATIO:.0%} required)"
        )

    improvements = [row["improvement_pct"] for row in task_rows if row["gated"] and row["improvement_pct"] is not None]
    median_improvement = round(statistics.median(improvements), 1) if improvements else None
    incumbent_resolved = sum(
        arms[incumbent_arm]["resolved"] for arms in results.values() if incumbent_arm in arms and candidate_arm in arms
    )
    candidate_resolved = sum(
        arms[candidate_arm]["resolved"] for arms in results.values() if incumbent_arm in arms and candidate_arm in arms
    )

    resolution_margin = candidate_resolved - incumbent_resolved
    if insufficient:
        decision = "insufficient_evidence"
    elif quality_regression or quality_floor_failed or efficiency_regression:
        decision = "keep_incumbent"
    elif resolution_margin >= 2:
        decision = "promote"
        reasons.append(
            f"candidate improves total task resolution by {resolution_margin} runs "
            "(at least 2 required) with no task regression"
        )
    else:
        if resolution_margin == 1:
            reasons.append(
                "total resolution improved by only 1 run — within the noise floor "
                "(2 required); deciding on efficiency instead"
            )
        if median_improvement is not None and median_improvement >= min_improvement_pct:
            decision = "promote"
            reasons.append(
                f"median {metric} improvement is {median_improvement:.1f}% (required {min_improvement_pct:.1f}%)"
            )
        else:
            decision = "keep_incumbent"
            reasons.append(
                f"median {metric} improvement is {median_improvement or 0.0:.1f}% (required {min_improvement_pct:.1f}%)"
            )

    return {
        "incumbent_arm": incumbent_arm,
        "candidate_arm": candidate_arm,
        "decision": decision,
        "metric": metric,
        "metric_warning": (MAIN_LOOP_ONLY_WARNING if metric in MAIN_LOOP_ONLY_METRICS else None),
        "median_improvement_pct": median_improvement,
        "ungated_tasks": ungated_tasks,
        "gated_tasks": gated_tasks,
        "reasons": reasons,
        "tasks": task_rows,
    }