ouroboros/devtools/benchmarks/programbench/run_programbench_e2e.py

#!/usr/bin/env python3
"""End-to-end ProgramBench runner for Ouroboros.

Orchestrates docker cleanroom setup, Ouroboros task submission via the gateway API,
submission packaging, and optional official ``programbench eval``.
"""

from __future__ import annotations

import argparse
import dataclasses
import json
import os
import pathlib
import sys
import traceback
import urllib.parse
from typing import Any

if __package__ in {None, ""}:
    sys.path.insert(0, str(pathlib.Path(__file__).resolve().parents[3]))

from devtools.benchmarks.common.manifests import (
    MODEL_SLOT_KEYS,
    admit_benchmark_run,
    finalize_run_manifest,
    runtime_attestation,
    write_json,
)
from devtools.benchmarks.common.official_commands import programbench_command_for_manifest
from devtools.benchmarks.common.result_index import append_result_index, task_result_row
from devtools.benchmarks.common.run_roots import (
    default_settings_path,
    assert_outside_repo,
    run_root,
    safe_join_under,
)
from devtools.benchmarks.programbench.programbench_adapter import (
    _checkpoint_task_id,
    build_instruction,
    build_ouroboros_task_body,
    classify_infra_failure,
    container_name_for_instance,
    create_submission_tarball,
    default_protected_backend_paths,
    ouroboros_api_request,
    pull_cleanroom_image,
    run_official_eval,
    seed_workspace_from_image,
    start_cleanroom_container,
    stop_cleanroom_container,
    submit_and_wait,
    terminal_task_status,
)
from devtools.benchmarks.programbench.schemas import PROGRAMBENCH_TIMEOUT_SEC
from ouroboros.provider_models import migrate_model_value

# Model-carrying slots only; the OUROBOROS_EFFORT_* entries in MODEL_SLOT_KEYS are
# effort levels, not model ids.
_MODEL_ID_SLOT_KEYS = tuple(key for key in MODEL_SLOT_KEYS if not key.startswith("OUROBOROS_EFFORT_"))

TASK_CHECKPOINT_BASENAME = "ouroboros_task_checkpoint.json"

# A resume-skipped instance (prior submission already on disk) is successful
# prior work, not a failure — the run exit code and failed_count agree on this.
_SUCCESS_STATUSES = ("completed", "skipped")


def _row_successful(row: dict[str, Any]) -> bool:
    return row.get("status") in _SUCCESS_STATUSES


def _ensure_docker_host() -> None:
    if os.environ.get("DOCKER_HOST"):
        return
    colima_sock = pathlib.Path.home() / ".colima" / "default" / "docker.sock"
    if colima_sock.exists():
        os.environ["DOCKER_HOST"] = f"unix://{colima_sock}"


def _setting_or_env(settings: dict[str, Any], key: str) -> str:
    """Settings value with env fallback — the server treats settings.json as SSOT
    and falls back to the launch environment for keys it does not carry."""
    raw = settings.get(key)
    if raw is None or str(raw).strip() == "":
        raw = os.environ.get(key, "")
    return str(raw or "").strip()


def _active_direct_provider(settings: dict[str, Any]) -> str:
    """Mirror config._exclusive_direct_remote_provider_env over settings+env.

    '' means the OpenRouter-style route (or an ambiguous key set), where
    ``provider/model`` ids are canonical and must NOT be rewritten.
    """
    if any(
        _setting_or_env(settings, key)
        for key in ("OPENROUTER_API_KEY", "OPENAI_BASE_URL", "OPENAI_COMPATIBLE_BASE_URL")
    ):
        return ""
    direct = [
        provider
        for provider, key in (
            ("openai", "OPENAI_API_KEY"),
            ("anthropic", "ANTHROPIC_API_KEY"),
            ("cloudru", "CLOUDRU_FOUNDATION_MODELS_API_KEY"),
        )
        if _setting_or_env(settings, key)
    ]
    if _setting_or_env(settings, "GIGACHAT_CREDENTIALS") or (
        _setting_or_env(settings, "GIGACHAT_USER") and _setting_or_env(settings, "GIGACHAT_PASSWORD")
    ):
        direct.append("gigachat")
    return direct[0] if len(direct) == 1 else ""


def preflight_model_slots(settings_path: pathlib.Path, *, solve_model: str = "") -> dict[str, str]:
    """Validate configured model ids the way the runtime routes them.

    Direct-provider routes reject legacy ``provider/model`` ids (they need
    ``provider::model``); a prior debug run burned 5/5 tasks on exactly that
    400. The runner cannot patch the externally started server's environment, so
    a legacy-form id (or a --solve-model that disagrees with settings) is a
    launch blocker, not a warning. Returns the normalized slot snapshot.
    """
    settings: dict[str, Any] = {}
    try:
        loaded = json.loads(pathlib.Path(settings_path).read_text(encoding="utf-8"))
        if isinstance(loaded, dict):
            settings = loaded
    except (OSError, json.JSONDecodeError):
        settings = {}
    provider = _active_direct_provider(settings)
    problems: list[str] = []
    slots: dict[str, str] = {}
    for key in _MODEL_ID_SLOT_KEYS:
        raw = _setting_or_env(settings, key)
        if not raw:
            continue
        parts = [part.strip() for part in raw.split(",") if part.strip()]
        migrated = [migrate_model_value(provider, part) if provider else part for part in parts]
        if migrated != parts:
            problems.append(f"{key}: {raw!r} -> {','.join(migrated)!r}")
        slots[key] = ",".join(migrated)
    expected_solve = ""
    if solve_model:
        expected_solve = migrate_model_value(provider, solve_model) if provider else solve_model
        if expected_solve != solve_model:
            problems.append(f"--solve-model: {solve_model!r} -> {expected_solve!r}")
        configured = slots.get("OUROBOROS_MODEL", "")
        if configured and configured != expected_solve:
            problems.append(
                f"--solve-model {expected_solve!r} does not match settings OUROBOROS_MODEL "
                f"{configured!r}; the server would solve on the wrong model"
            )
    if problems:
        raise SystemExit(
            "model slot preflight failed for the "
            + (f"direct {provider!r}" if provider else "configured")
            + " route — fix the settings the server was started with (see "
            f"{settings_path}):\n  " + "\n  ".join(problems)
        )
    return slots


def _load_instances(
    *,
    difficulty: str,
    filter_spec: str,
    slice_spec: str,
    instance_id: str,
    shuffle: bool,
) -> list[dict[str, Any]]:
    from programbench.utils.instance_filters import filter_instances
    from programbench.utils.load_data import load_all_instances

    instances = load_all_instances(include_tests=False)
    if difficulty:
        instances = [item for item in instances if str(item.get("difficulty") or "") == difficulty]
    if instance_id:
        instances = [item for item in instances if str(item.get("instance_id") or "") == instance_id]
    return filter_instances(instances, filter_spec=filter_spec, slice_spec=slice_spec, shuffle=shuffle)


@dataclasses.dataclass(frozen=True)
class InstanceRunConfig:
    """Per-run knobs shared by every instance (keeps `_process_instance` within
    the DEVELOPMENT.md parameter budget)."""

    out_root: pathlib.Path
    ouroboros_url: str
    timeout_sec: float
    cpus: str
    memory: str
    protected_paths: list[str]
    dry_run: bool
    skip_pull: bool
    redo_existing: bool


def _process_instance(instance: dict[str, Any], cfg: InstanceRunConfig) -> dict[str, Any]:
    instance_id = str(instance["instance_id"])
    image_name = str(instance["image_name"])
    instance_dir = safe_join_under(cfg.out_root, instance_id)
    workspace = instance_dir / "workspace"
    container_name = container_name_for_instance(instance_id)
    instruction = build_instruction(instance)
    instruction_path = instance_dir / "instruction.txt"
    instance_dir.mkdir(parents=True, exist_ok=True)
    instruction_path.write_text(instruction, encoding="utf-8")
    checkpoint_path = instance_dir / TASK_CHECKPOINT_BASENAME
    if cfg.redo_existing:
        # An explicit redo means a fresh solve; a surviving checkpoint would
        # re-attach to the previous task instead. Cancel a still-live server-side
        # task FIRST — otherwise it keeps running and its docker-exec calls land
        # in the NEW same-named container, contaminating the fresh solve
        # (adversarial review r1).
        stale_task_id = _checkpoint_task_id(checkpoint_path)
        if stale_task_id:
            try:
                ouroboros_api_request(
                    cfg.ouroboros_url, "POST",
                    f"/api/tasks/{urllib.parse.quote(stale_task_id)}/cancel", timeout=30,
                )
            except Exception:
                pass  # best-effort: server may not know it / be unreachable
        checkpoint_path.unlink(missing_ok=True)

    harness: dict[str, Any] = {"container_name": container_name, "image_name": image_name}
    output_paths: dict[str, str] = {"instruction": str(instruction_path)}
    keep_container_for_reattach = False
    # Reattach honor-check BEFORE any cleanroom work: a fresh seed/start would
    # reset the workspace and (start_cleanroom_container stops any namesake
    # first) kill the very container a prior client_poll_timeout_reattachable
    # row deliberately left running. If the checkpoint names a task this server
    # still knows, we preserve the container/workspace and let submit_and_wait
    # reattach (a settled task returns immediately and is packaged as-is); only
    # a missing/stale checkpoint (or --redo-existing, which unlinked it above)
    # gets a fresh cleanroom.
    reattach_task_id = _checkpoint_task_id(checkpoint_path)
    if reattach_task_id:
        try:
            _reattach_payload = ouroboros_api_request(
                cfg.ouroboros_url, "GET",
                f"/api/tasks/{urllib.parse.quote(reattach_task_id)}", timeout=60,
            )
        except Exception:
            # ANY failure (RuntimeError from an HTTPError, or a connection-level
            # URLError/OSError the adapter does not map) means we cannot confirm
            # the task is live — fall to a fresh cleanroom rather than aborting
            # the whole run before the top-level ledger is written (r1 #14).
            reattach_task_id = ""
        else:
            # Reattach ONLY to an in-flight or COMPLETED task. A task that already
            # SETTLED as a failure/cancellation must NOT be reattached — replaying
            # it is zero work and defeats the resume-RETRY contract (r1 #12: a
            # failed instance gets NO tarball precisely so a resume run re-solves
            # it). Drop the stale checkpoint and fall to a fresh cleanroom solve
            # (adversarial review r2 #5). An in-flight task (status not settled)
            # returns "" from terminal_task_status and is reattached as a genuine
            # crash-resume; a settled "completed" is reattached and packaged as-is.
            _settled = terminal_task_status(_reattach_payload)
            if _settled and _settled != "completed":
                reattach_task_id = ""
                checkpoint_path.unlink(missing_ok=True)
    harness["reattached_task_id"] = reattach_task_id
    # While a reattachable task is live, NOTHING may stop its container — not a
    # dry-run early return, not a mid-poll transport error. Only a settled
    # terminal result (below) re-arms the normal teardown.
    keep_container_for_reattach = bool(reattach_task_id)
    try:
        if not reattach_task_id:
            if not cfg.skip_pull:
                harness["image"] = pull_cleanroom_image(image_name)
            harness["seed"] = seed_workspace_from_image(image_name, workspace)
            harness["container"] = start_cleanroom_container(
                container_name,
                image_name,
                workspace,
                cpus=cfg.cpus,
                memory=cfg.memory,
            )
        body = build_ouroboros_task_body(
            instruction=instruction,
            workspace_host_path=workspace,
            container_name=container_name,
            protected_backend_paths=cfg.protected_paths,
        )
        body.setdefault("metadata", {})["programbench_instance_id"] = instance_id
        body.setdefault("metadata", {})["cleanroom_preflight"] = (harness.get("container") or {}).get("preflight") or {}
        body["timeout_sec"] = float(cfg.timeout_sec)
        write_json(instance_dir / "ouroboros_task_body.json", body)
        output_paths["task_body"] = str(instance_dir / "ouroboros_task_body.json")
        output_paths["workspace"] = str(workspace)

        if cfg.dry_run:
            return task_result_row(
                benchmark="programbench",
                instance_id=instance_id,
                status="completed",
                reason_code="dry_run",
                official_eval_status="not_run",
                output_paths=output_paths,
                details=harness,
            )

        output_paths["task_checkpoint"] = str(checkpoint_path)
        try:
            task_result = submit_and_wait(
                cfg.ouroboros_url,
                body,
                timeout_sec=cfg.timeout_sec,
                checkpoint_path=checkpoint_path,
            )
        except Exception as exc:
            # A client-side poll failure (timeout OR a transient mid-poll transport
            # error) after a task was submitted leaves it RUNNING server-side with
            # its task_id in the checkpoint. Tearing the container down here would
            # make the next run reattach to a task whose Docker backend is gone —
            # keep the executor alive for reattach (r1 #10, generalized from the
            # timeout-only guard). If NO task_id was ever recorded (creation itself
            # failed), there is nothing to reattach: re-raise to the normal
            # teardown+failed-row path.
            if not _checkpoint_task_id(checkpoint_path):
                raise
            keep_container_for_reattach = True
            reason = "client_poll_timeout_reattachable" if isinstance(exc, TimeoutError) else "client_poll_error_reattachable"
            return task_result_row(
                benchmark="programbench",
                instance_id=instance_id,
                status="failed",
                reason_code=reason,
                official_eval_status="not_run",
                output_paths=output_paths,
                error=str(exc),
                details={"harness": harness, "container_left_running": True,
                         "reattach_checkpoint": str(checkpoint_path)},
            )
        keep_container_for_reattach = False  # settled: normal teardown applies
        write_json(instance_dir / "ouroboros_task_result.json", task_result)
        output_paths["task_result"] = str(instance_dir / "ouroboros_task_result.json")

        # Classification comes from the payload's EXPLICIT terminal status; a
        # stale provider reason_code on a completed task must never demote it.
        status = terminal_task_status(task_result)
        infra_failed = classify_infra_failure(task_result)
        if status != "completed":
            # A non-completed settled task produces NO submission tarball — else the
            # resume gate would see submission.tar.gz and skip a FAILED instance as a
            # skipped-success, never retrying it (r1 #12). Denominator stays honest
            # via the failed row below.
            return task_result_row(
                benchmark="programbench",
                instance_id=instance_id,
                status="failed",
                reason_code=str(task_result.get("reason_code") or "") or "task_not_completed",
                runtime_result=task_result,
                prediction_written=False,
                official_eval_status="not_run",
                output_paths=output_paths,
                error=str(task_result.get("result") or task_result.get("error") or status),
                details={"harness": harness, "task_status": status, "infra_failed": infra_failed},
            )
        # Only a COMPLETED task produces the source submission — its presence is
        # what the resume gate keys on, so it must mean "solved", not "attempted".
        submission = create_submission_tarball(
            workspace,
            instance_dir / "submission.tar.gz",
            protected_paths=cfg.protected_paths,
        )
        output_paths["submission"] = str(submission)
        return task_result_row(
            benchmark="programbench",
            instance_id=instance_id,
            status="completed",
            reason_code="submission_prepared",
            # The non-completed branch above already publishes the runtime `reason_code`; the
            # success branch dropped it, so a submission prepared by a cost-truncated run read
            # exactly like one from a run that finished. `infra_failed` collapses the runtime's
            # terminal reason to one bool and cannot carry `budget_exhausted` at all.
            runtime_result=task_result,
            prediction_written=True,
            official_eval_status="not_run",
            output_paths=output_paths,
            details={"harness": harness, "task_status": status, "infra_failed": infra_failed},
        )
    except Exception as exc:
        return task_result_row(
            benchmark="programbench",
            instance_id=instance_id,
            status="failed",
            reason_code=type(exc).__name__,
            official_eval_status="not_run",
            output_paths=output_paths,
            error=str(exc),
            details={"harness": harness, "traceback": traceback.format_exc()},
        )
    finally:
        if not keep_container_for_reattach:
            stop_cleanroom_container(container_name)


def main() -> int:
    parser = argparse.ArgumentParser(description="Run ProgramBench instances through Ouroboros")
    parser.add_argument("--repo-dir", default=str(pathlib.Path(__file__).resolve().parents[3]))
    parser.add_argument("--ouroboros-url", default="http://127.0.0.1:8770", help="Ouroboros source server URL (desktop sandbox on 8765 cannot see bench_runs)")
    parser.add_argument("--solve-model", default="", help="expected OUROBOROS_MODEL; validated/normalized against --settings-path before any task burns")
    parser.add_argument("--difficulty", default="", help="filter by task.yaml difficulty (default: all)")
    parser.add_argument("--filter", default="", dest="filter_spec", help="regex filter on instance_id")
    parser.add_argument("--slice", default="", dest="slice_spec", help="slice spec, e.g. 0:3 (default: all)")
    parser.add_argument(
        "--shuffle",
        action="store_true",
        help="shuffle instances (programbench uses seed 42 after sorting by instance_id)",
    )
    parser.add_argument("--instance-id", default="", help="run a single instance id")
    parser.add_argument("--run-id", default="")
    parser.add_argument("--timeout-sec", type=float, default=PROGRAMBENCH_TIMEOUT_SEC)
    parser.add_argument("--cpus", default="4")
    parser.add_argument("--memory", default="16g")
    parser.add_argument("--protected-path", action="append", default=[])
    parser.add_argument("--settings-path", default="")
    parser.add_argument("--redo-existing", action="store_true")
    parser.add_argument("--skip-pull", action="store_true")
    parser.add_argument("--dry-run", action="store_true", help="docker setup + task body only")
    parser.add_argument("--eval", action="store_true", help="run official programbench eval/info after all instances")
    parser.add_argument("--no-eval", action="store_true", help=argparse.SUPPRESS)
    parser.add_argument(
        "--allow-dirty-seed",
        action="store_true",
        help="record and proceed with an unclean/unidentifiable seed checkout instead of refusing",
    )
    args = parser.parse_args()
    if args.no_eval:
        args.eval = False

    _ensure_docker_host()
    repo_dir = pathlib.Path(args.repo_dir).expanduser().resolve(strict=False)
    settings_path = pathlib.Path(args.settings_path).expanduser() if args.settings_path else default_settings_path()
    out_root = assert_outside_repo(run_root("programbench", args.run_id), repo_dir)
    protected_paths = args.protected_path or default_protected_backend_paths()

    # The DECLARED selector, which is pure argv, is what admission can honestly record: the
    # RESOLVED ids require loading the ProgramBench dataset, and that load (plus the model-slot
    # preflight's settings read) can fail on the state of the world — a failure that used to
    # take the process down before any manifest existed, leaving the run invisible rather than
    # merely footprint-free. Both moved behind admission; `requested_task_ids` is amended with
    # the resolved ids there, and the selector below records what was ASKED FOR either way.
    selector = {
        "difficulty": str(args.difficulty or ""),
        "filter_spec": str(args.filter_spec or ""),
        "slice_spec": str(args.slice_spec or ""),
        "instance_id": str(args.instance_id or ""),
        "shuffle": bool(args.shuffle),
    }
    declared_ids = [str(args.instance_id)] if args.instance_id else []
    manifest_path = out_root / "run_manifest.json"
    ledger_path = out_root / "result_index.jsonl"
    # ADMISSION, before the first instance burns anything: the seed-provenance gate (built into
    # the manifest) plus the runtime identity of the server we are about to submit to. Until
    # v6.75.0 this manifest was written at the very END of the run — after every instance and
    # the official eval — so the gate physically could not stop an unreproducible run.
    # `admit_benchmark_run` persists the complete gate/refusal payload BEFORE enforcement raises.
    manifest = admit_benchmark_run(
        manifest_path,
        benchmark="programbench",
        run_root=out_root,
        repo_dir=repo_dir,
        requested_task_ids=declared_ids,
        require_clean=not args.allow_dirty_seed,
        argv=sys.argv,
        output_paths={
            "run_root": str(out_root),
            "ledger": str(ledger_path),
            "manifest": str(manifest_path),
        },
        dataset="programbench",
        harness={
            "ouroboros_url": str(args.ouroboros_url),
            "difficulty": str(args.difficulty or ""),
            "dry_run": bool(args.dry_run),
            "solve_model": str(args.solve_model or ""),
            "selector": selector,
            "model_slots_normalized": {"pending": "preflight_runs_after_admission"},
        },
        official_command=programbench_command_for_manifest(out_root, eval_requested=bool(args.eval)),
        isolated_data_root="",
        settings_path=settings_path,
        extra={
            "eval_requested": bool(args.eval),
            "official_eval_status": "not_run",
            "protected_paths": protected_paths,
            "shuffle": bool(args.shuffle),
            "shuffle_seed": 42 if args.shuffle else None,
            "instance_order": str(out_root / "instance_order.json"),
            "runtime_attestation": {"pending": "not_attested_yet"},
        },
    )

    with finalize_run_manifest(manifest_path, manifest, outcome="completed") as final:
        # PREFLIGHT + DISCOVERY, now inside the admitted run: every refusal below is recorded by
        # the finalization seam instead of vanishing with the process.
        model_slots = preflight_model_slots(settings_path, solve_model=str(args.solve_model or ""))
        manifest["harness"]["model_slots_normalized"] = model_slots
        instances = _load_instances(**selector)
        if not instances:
            final.update({"outcome": "refused", "exit_code": 1,
                          "refusal": {"stage": "instance_selection", "exit_code": 1,
                                      "reason": "no_instances_matched"}})
            raise SystemExit("no ProgramBench instances matched the selection")
        selected_ids = [str(item["instance_id"]) for item in instances]
        # AMEND the retained manifest with the ids the selector actually resolved to. The
        # declared selector stays in `harness`, so the record says both what was asked for and
        # what it meant.
        manifest["requested_task_ids"] = selected_ids
        manifest["requested_count"] = len(selected_ids)
        # Written AFTER persisted admission: a refused run must not leave run artefacts behind, and
        # nothing before the admission call may touch the filesystem.
        write_json(
            out_root / "instance_order.json",
            {
                "count": len(instances),
                "shuffle": bool(args.shuffle),
                "shuffle_seed": 42 if args.shuffle else None,
                "instance_ids": [str(item["instance_id"]) for item in instances],
            },
        )
        # Attestation runs INSIDE the finalization block, never in the admission call's argument
        # list: Python evaluates arguments BEFORE entering the callee, so an attestation refusal
        # there meant `admit_benchmark_run` never ran and no manifest ever reached disk — the
        # durable-refusal contract defeated by evaluation order. The refusal still propagates
        # (unchanged behaviour); it is now recorded as a typed refusal first.
        if args.dry_run:
            manifest["extra"]["runtime_attestation"] = {"skipped": "dry_run"}
        else:
            try:
                attested = runtime_attestation(str(args.ouroboros_url), repo_dir)
            except Exception as exc:
                # A refusal CARRIES the record it built (`RuntimeAttestationRefused.attestation`),
                # so the durable manifest keeps the exact typed reason plus the runtime_version /
                # repo_head / repo_version facts instead of a generic message. The fallback reason
                # covers a failure that carries no record at all.
                refused = getattr(exc, "attestation", None)
                if isinstance(refused, dict):
                    manifest["extra"]["runtime_attestation"] = refused
                final.update({
                    "outcome": "refused",
                    "exit_code": 3,
                    "refusal": {
                        "stage": "runtime_attestation",
                        "exit_code": 3,
                        "reason": str((refused or {}).get("reason") or "")
                                  or "runtime_attestation_failed",
                    },
                })
                # RETURN the recorded code, never re-raise it. An escaping RuntimeError exits the
                # process with status 1 (`raise SystemExit(main())` is never reached), so a bare
                # re-raise made the durable record say 3 while the process said 1 — the record and
                # reality disagreeing is the one thing this manifest exists to prevent. The message
                # the exception used to surface is logged here instead.
                print(f"[pb-e2e] FATAL: {exc}", file=sys.stderr)
                return 3
            manifest["extra"]["runtime_attestation"] = attested
        skipped_rows: list[dict[str, Any]] = []
        if not args.redo_existing:
            # Resume narrows the WORK, never the ledger denominator: instances with a
            # prior submission are recorded as explicit skipped rows, not dropped.
            pending = []
            for instance in instances:
                submission = safe_join_under(out_root, str(instance["instance_id"])) / "submission.tar.gz"
                if submission.is_file() and submission.stat().st_size > 0:
                    skipped_rows.append(task_result_row(
                        benchmark="programbench",
                        instance_id=str(instance["instance_id"]),
                        status="skipped",
                        reason_code="skipped_existing_submission",
                        prediction_written=True,
                        official_eval_status="not_run",
                        output_paths={"submission": str(submission)},
                    ))
                    continue
                pending.append(instance)
            instances = pending

        rows: list[dict[str, Any]] = list(skipped_rows)
        cfg = InstanceRunConfig(
            out_root=out_root,
            ouroboros_url=str(args.ouroboros_url),
            timeout_sec=float(args.timeout_sec),
            cpus=str(args.cpus),
            memory=str(args.memory),
            protected_paths=protected_paths,
            dry_run=bool(args.dry_run),
            skip_pull=bool(args.skip_pull),
            redo_existing=bool(args.redo_existing),
        )
        # APPEND-ONLY run ledger: one line per row, written the moment the row exists. The old
        # whole-file rewrite at the end of the run discarded every completed instance if the
        # process died mid-run, and a resumed run silently replaced the previous run's history.
        # Readers dedup by instance_id, LAST row wins (documented in programbench/README.md).
        # BOTH locations for EVERY row, skips included: a resumed instance whose skip event only
        # reached the run root has an instance-level history that silently omits the resume, and the
        # README states the two-location contract without qualification.
        for row in skipped_rows:
            append_result_index(safe_join_under(out_root, str(row["instance_id"])), row)
            append_result_index(out_root, row)
        for instance in instances:
            row = _process_instance(instance, cfg)
            rows.append(row)
            append_result_index(safe_join_under(out_root, str(instance["instance_id"])), row)
            append_result_index(out_root, row)

        eval_result = None
        official_eval_status = "not_run"
        if args.eval and not args.dry_run:
            try:
                eval_result = run_official_eval(out_root)
                # returncode != 0 = the eval RAN and reported test failures (a valid
                # partial-pass benchmark result, not a run error).
                official_eval_status = "completed" if eval_result.get("eval", {}).get("returncode") == 0 else "failed"
            except Exception as exc:
                # The eval could not RUN at all (harness/infra error) — distinct from a
                # ran-but-some-tests-failed result; this is a run error (r2 fable #3).
                official_eval_status = "error"
                write_json(
                    out_root / "programbench_eval_result.json",
                    {"error": str(exc), "traceback": traceback.format_exc()},
                )

        manifest["extra"].update(
            {
                "official_eval_status": official_eval_status,
                "completed_count": sum(1 for row in rows if row.get("status") == "completed"),
                "failed_count": sum(1 for row in rows if not _row_successful(row)),
            }
        )
        print(out_root)
        rows_ok = all(_row_successful(row) for row in rows)
        # A REQUESTED official eval that could NOT run (harness error) is a run failure
        # even when every solve completed — otherwise `run_...e2e --eval && ...` reads a
        # broken eval as success. A "failed" eval that merely reported test failures is
        # a valid partial-pass result and does NOT fail the run (r2 fable #3).
        eval_errored = bool(args.eval) and not args.dry_run and official_eval_status == "error"
        code = 0 if (rows_ok and not eval_errored) else 1
        if code:
            # A typed terminal outcome, not a bare exit code: the two reasons a completed run
            # still fails are distinguishable in its own record.
            final["outcome"] = "official_eval_error" if eval_errored else "instances_failed"
        final["exit_code"] = code
        return code


if __name__ == "__main__":
    raise SystemExit(main())