scripts/platform_lifecycle.py · oak/benchmarks

Viewing files at commit 62758990fe31Commit details View latest
Browse files
scripts/platform_lifecycle.py 4263 lines · 161.5 KB · python
#!/usr/bin/env python3
"""The platform lane: hosted integration workflows, end to end.

Measures GitHub-the-platform vs oak hosted workflows from "task branch exists
locally" to "change merged in main, present in local checkout" — including
the async waits agents actually experience (GitHub computes PR mergeability
asynchronously; that settling is a measured operation, observed by polling on
one local monotonic clock).

Drivers per platform:
  cli         shells out to gh/git/oak via run_timed — full token/output
              accounting from captured output, like every other CLI lane.
  harness-api stdlib urllib REST against the GitHub API — exact api_metrics
              (request_count/bytes, rate-limit header, request ids); token
              fields stay null because API JSON is not an agent transcript
              and token costs would be fiction (ADR-0002).
  fake-provider
              deterministic in-process branch triage provider for oracle
              coverage when no hosted platform is available; not comparable
              to real platform timings.

This lane runs against real hosted services and is EXCLUDED from devloop:
external variance must never poison changeset verdicts.

Without credentials/configured repos (the only path CI exercises) every
operation becomes a skip row with an explicit skip_reason and the process
exits 3 (all-skipped), never silent absence.

Setup:
  GITHUB_TOKEN=... python3 scripts/platform_lifecycle.py \\
    --platform github --driver harness-api --scenario pr_single_anatomy \\
    --repo your-org/disposable-bench-repo --runs 3
"""

from __future__ import annotations

import argparse
import concurrent.futures
import json
import os
import platform as platform_module
import re
import shlex
import shutil
import subprocess
import sys
import tempfile
import threading
import time
from dataclasses import dataclass
from datetime import datetime, timezone
from pathlib import Path
from typing import Any, Callable, Optional

from oakbench import platform_clock
from oakbench import tokens as oakbench_tokens
from oakbench.environment import ENV_ISOLATION_VERSION, base_env as oakbench_base_env
from oakbench.environment import command_display
from oakbench.execution import run_timed
from oakbench.platform_api import GitHubApiError, GitHubClient, RateLimitBudgeter
from oakbench.remotes import disposable_branch
from oakbench.results import ResultsStore
from oakbench.rows import row_returncode
from oakbench.runlock import measurement_lock
from oakbench.runner import runner_fields, stamp_row

from mount_probe import parse_yaml_subset

ROOT = Path(__file__).resolve().parents[1]
DEFAULT_SPEC = ROOT / "scenarios" / "platform.yaml"
DEFAULT_SEMANTICS = ROOT / "config" / "platform_semantics.json"
DEFAULT_WORKDIR = Path(tempfile.gettempdir()) / "oak-platform-lane"
SKIP_RETURNCODE = 77
EXIT_ALL_SKIPPED = 3
PR_URL_RE = re.compile(r"/pull/(\d+)")
STRUCTURAL_SOURCE = "structural_platform_semantics_zero_commands"

# integration_race_nN: N task branches each through publish -> PR -> merge,
# serially or as fast as RateLimitBudgeter pacing allows (project plan:
# default 500 content ops/hour). Injected pacing waits are REPORTED as
# race.pacing.injected, never subtracted from operation durations.
RACE_SCENARIO_RE = re.compile(r"^integration_race_n(\d+)$")
BRANCH_TRIAGE_SCENARIO_RE = re.compile(r"^branch_triage_n(\d+)$")
BRANCH_FLEET_SCENARIO_RE = re.compile(r"^branch_fleet_n(\d+)$")
DEFAULT_RACE_OPS_PER_HOUR = 500
# gh-cli race orchestration (N clones + interleaved gh polling under one
# budgeter) is deferred to Phase 4: run_timed's one-command-at-a-time
# accounting cannot yet attribute interleaved waits honestly across branches.
RACE_CLI_DEFER_REASON = "race_cli_driver_deferred"
# check-instant protection variant: the HARNESS posts this required status
# instantly via the commit status API (deterministic, free, no Actions).
CHECK_INSTANT_CONTEXT = "oakbench/check-instant"
CHECKS_GREEN_OPERATION = "integration.checks.green.settle"
BRANCH_TRIAGE_FAKE_DRIVER = "fake-provider"
BRANCH_TRIAGE_CLASSES = ("clean", "stale", "duplicate", "conflicting")
BRANCH_TRIAGE_ACTIONS = {
    "clean": "merge",
    "stale": "refresh_then_merge",
    "duplicate": "close_duplicate",
    "conflicting": "keep_open_manual_resolution",
}
BRANCH_FLEET_PROVIDER_FAKE = "fake"
BRANCH_FLEET_PROVIDER_GITHUB_API = "github:harness-api"
BRANCH_FLEET_PROVIDER_OAK_CLI = "oak:cli"
BRANCH_FLEET_LIVE_LARGE_LIMIT = 100
BRANCH_FLEET_LIVE_LARGE_ENV = "OAK_BENCH_BRANCH_FLEET_LIVE_LARGE"
BRANCH_FLEET_CLASSIFY_PARSE_STDOUT_BYTES = 50 * 1024 * 1024
BRANCH_FLEET_SEED_TRANSIENT_RETRIES = 2
BRANCH_FLEET_SEED_RETRY_DELAY_S = 2.0
BRANCH_FLEET_SEED_TIMEOUT_RETCODE = 124
BRANCH_FLEET_SEED_WORKERS_ENV = "OAK_BENCH_BRANCH_FLEET_SEED_WORKERS"
BRANCH_FLEET_CLEANUP_WORKERS_ENV = "OAK_BENCH_BRANCH_FLEET_CLEANUP_WORKERS"
BRANCH_FLEET_SEED_WORKERS_DEFAULT = 4
BRANCH_FLEET_CLEANUP_WORKERS_DEFAULT = 4
BRANCH_FLEET_WORKERS_MAX = 16
# Posting a commit status needs the REST API; gh/git in this lane cannot do
# it, so a cli-driver run under check-instant would merge on a protected repo
# without the required check ever existing (or fail outright) while its rows
# claimed the variant held. Honest gap instead: every operation row skips.
CHECK_INSTANT_CLI_SKIP_REASON = "protection_variant_requires_api_driver:check-instant"


# --------------------------------------------------------------------------
# Configuration
# --------------------------------------------------------------------------


def parse_args(argv: Optional[list[str]] = None) -> argparse.Namespace:
    parser = argparse.ArgumentParser(description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter)
    parser.add_argument("--platform", choices=("github", "oak"), default="github")
    parser.add_argument("--driver", choices=("cli", "harness-api", BRANCH_TRIAGE_FAKE_DRIVER), default="cli")
    parser.add_argument("--scenario", default="platform_capability_probe", help="Scenario name from scenarios/platform.yaml")
    parser.add_argument("--repo", default="", help="org/repo for github; oak repo for oak. Must be DISPOSABLE.")
    parser.add_argument("--results", type=Path, default=ROOT / "results" / "platform")
    parser.add_argument("--runs", type=int, default=1)
    parser.add_argument("--protection-variant", default="protection-none")
    parser.add_argument("--semantics-class", default="integration-default")
    parser.add_argument("--token-env", default="GITHUB_TOKEN", help="NAME of the env var holding the PAT")
    parser.add_argument("--spec", type=Path, default=DEFAULT_SPEC)
    parser.add_argument("--semantics-config", type=Path, default=DEFAULT_SEMANTICS)
    parser.add_argument("--workdir", type=Path, default=DEFAULT_WORKDIR)
    parser.add_argument("--admitted-output-chars", type=int, default=20_000)
    parser.add_argument("--keep-workdirs", action="store_true")
    return parser.parse_args(argv)


def load_platform_spec(path: Path) -> dict[str, Any]:
    return parse_yaml_subset(path.read_text())


def scenario_by_name(spec: dict[str, Any], name: str) -> Optional[dict[str, Any]]:
    for scenario in spec.get("scenarios") or []:
        if isinstance(scenario, dict) and scenario.get("name") == name:
            return scenario
    return None


def load_semantics(path: Path) -> dict[str, Any]:
    return json.loads(path.read_text())


def compute_semantic_contract_match(
    semantics: dict[str, Any], semantics_class: str, protection_variant: str
) -> bool:
    """True when this run executes a declared semantics class under an
    implemented protection variant — the precondition for any cross-platform
    comparison of its rows."""
    classes = semantics.get("classes") or {}
    implemented = semantics.get("implemented_protection_variants") or []
    return semantics_class in classes and protection_variant in implemented


def platform_comparison_key(
    semantics_version: Any, semantics_class: str, protection_variant: str
) -> str:
    """Stable join key for cross-platform platform-lane comparisons.

    The protection variant is part of the key by design: a
    protection-none row and a check-instant row may share a semantics class,
    but they are not the same hosted workflow contract.
    """
    return f"{semantics_version}:{semantics_class}:{protection_variant}"


def platform_rows_comparable(row_a: dict[str, Any], row_b: dict[str, Any]) -> bool:
    """Pure guard for consumers: compare only same semantics+variant rows."""
    if row_a.get("semantic_contract_match") is not True:
        return False
    if row_b.get("semantic_contract_match") is not True:
        return False
    key_a = row_a.get("platform_comparison_key")
    key_b = row_b.get("platform_comparison_key")
    if key_a is not None or key_b is not None:
        return key_a == key_b
    # Backward-compatible fallback for older rows that predate the explicit key.
    return (
        row_a.get("platform_semantics_version") or row_a.get("semantics_version"),
        row_a.get("platform_semantics_class"),
        row_a.get("protection_variant"),
    ) == (
        row_b.get("platform_semantics_version") or row_b.get("semantics_version"),
        row_b.get("platform_semantics_class"),
        row_b.get("protection_variant"),
    )


def protection_variant_executable(
    platform_name: str, driver: str, protection_variant: str
) -> bool:
    """Whether this driver can actually EXECUTE the protection variant.

    check-instant means the harness posts the required commit status via the
    REST API; on GitHub only the harness-api driver can do that, so a
    cli-driver run never executes the variant and its rows must never claim
    semantic_contract_match (the cli runner emits
    CHECK_INSTANT_CLI_SKIP_REASON skip rows instead). Everything else is
    unchanged: implementedness is compute_semantic_contract_match's job."""
    if driver == BRANCH_TRIAGE_FAKE_DRIVER:
        return False
    if (
        platform_name == "github"
        and driver == "cli"
        and protection_variant == "check-instant"
    ):
        return False
    return True


@dataclass(frozen=True)
class Settings:
    interval_s: float
    settle_timeout_s: float
    timeout_s: float
    admitted: int


def scenario_settings(spec: dict[str, Any], scenario: dict[str, Any], admitted: int) -> Settings:
    defaults = spec.get("defaults") or {}
    interval = scenario.get("poll_interval_s", defaults.get("settle_interval_s", 2))
    return Settings(
        interval_s=float(interval),
        settle_timeout_s=float(defaults.get("settle_timeout_s", 300)),
        timeout_s=float(defaults.get("timeout_seconds", 120)),
        admitted=admitted,
    )


# --------------------------------------------------------------------------
# Readiness: missing PAT/repos/binaries become skip rows, never failures.
# --------------------------------------------------------------------------


def readiness_skip_reason(
    platform_name: str,
    driver: str,
    repo: str,
    token_env: str,
    *,
    requires_remote: bool,
    which: Callable[[str], Optional[str]] = shutil.which,
    environ: Optional[dict[str, str]] = None,
) -> Optional[str]:
    if driver == BRANCH_TRIAGE_FAKE_DRIVER:
        return None
    env = os.environ if environ is None else environ
    if platform_name == "github":
        if driver == "cli" and which("gh") is None:
            return "gh_cli_missing"
        if not str(env.get(token_env, "")).strip():
            return f"github_token_missing:{token_env}"
        if requires_remote and not repo:
            return "github_repo_not_configured"
        return None
    if driver == "harness-api":
        return "oak_harness_api_not_implemented"
    if which("oak") is None:
        return "oak_cli_missing"
    if requires_remote and not repo:
        return "oak_remote_missing"
    if requires_remote and not oak_credentials_path(env).is_file():
        return "oak_credentials_missing"
    return None


def oak_cli_commit_push_skip_reason(
    env: dict[str, str],
    *,
    runner: Callable[..., subprocess.CompletedProcess[str]] = subprocess.run,
) -> Optional[str]:
    """Return a skip reason when the live Oak CLI cannot publish explicitly."""
    try:
        probe = runner(
            ["oak", "commit", "--help"],
            capture_output=True,
            text=True,
            env=env,
            timeout=15,
            check=False,
        )
    except FileNotFoundError:
        return "oak_cli_missing"
    except subprocess.TimeoutExpired:
        return "oak_commit_push_capability_timeout"
    help_text = (probe.stdout or "") + "\n" + (probe.stderr or "")
    if probe.returncode != 0:
        return "oak_commit_push_capability_failed"
    if "--push" not in help_text:
        return "oak_commit_push_not_supported:requires_current_oak"
    return None


# --------------------------------------------------------------------------
# Row builders
# --------------------------------------------------------------------------


def base_env() -> dict[str, str]:
    return oakbench_base_env(
        author_name="Oak Platform Lane",
        author_email="[email protected]",
        oak_author="oak-platform-lane",
    )


def oak_credentials_path(environ: Optional[dict[str, str]] = None) -> Path:
    env = os.environ if environ is None else environ
    override = str(env.get("OAK_CREDENTIALS_FILE") or "").strip()
    if override:
        return Path(override).expanduser()
    return Path(str(env.get("HOME") or "")).expanduser() / ".oak" / "credentials"


def install_oak_credentials(env: dict[str, str]) -> bool:
    source = oak_credentials_path()
    if not source.is_file():
        return False
    dest = Path(env["HOME"]) / ".oak" / "credentials"
    dest.parent.mkdir(mode=0o700, parents=True, exist_ok=True)
    shutil.copy2(source, dest)
    dest.chmod(0o600)
    return True


def cli_env(args: argparse.Namespace) -> dict[str, str]:
    env = base_env()
    token = os.environ.get(args.token_env, "")
    if args.platform == "github" and token:
        # gh reads GH_TOKEN; git over https uses it via gh's credential helper.
        env.setdefault("GH_TOKEN", token)
        env.setdefault("GITHUB_TOKEN", token)
    if args.platform == "oak":
        install_oak_credentials(env)
    return env


def null_token_fields() -> dict[str, Any]:
    return {
        "estimated_tokens_total": None,
        "estimated_tokens_input": None,
        "estimated_tokens_output": None,
        "estimated_tokens_agent_emitted": None,
        "estimated_tokens_agent_ingested": None,
        "estimated_cost_weighted_tokens": None,
        "token_fields_note": (
            "harness-api driver: REST JSON is not an agent transcript; "
            "token costs would be fiction. Null means unmeasured (ADR-0002)."
        ),
    }


def fake_provider_token_fields() -> dict[str, Any]:
    return {
        "estimated_tokens_total": None,
        "estimated_tokens_input": None,
        "estimated_tokens_output": None,
        "estimated_tokens_agent_emitted": None,
        "estimated_tokens_agent_ingested": None,
        "estimated_cost_weighted_tokens": None,
        "raw_output_bytes": None,
        "stdout_bytes": None,
        "stderr_bytes": None,
        "token_fields_note": (
            "fake-provider driver: in-process deterministic oracle fixture; "
            "no terminal command or agent-visible command output exists."
        ),
    }


def skip_row(
    meta: dict[str, Any],
    scenario: str,
    operation: str,
    run_index: int,
    reason: str,
    *,
    settings: Optional[Settings] = None,
) -> dict[str, Any]:
    row = {
        **meta,
        "scenario": scenario,
        "run": run_index,
        "operation": operation,
        "elapsed_ms": 0.0,
        "returncode": SKIP_RETURNCODE,
        "command": [],
        "skipped": True,
        "skip_reason": reason,
        "tool_call_count": 0,
    }
    if settings is not None and operation.endswith(".settle"):
        row.update(
            {
                "settled": None,
                "settle_elapsed_ms": None,
                "observed_wait_ms": 0.0,
                "poll_iterations": 0,
                "poll_interval_s": settings.interval_s,
                "poll_quantization_ms": settings.interval_s * 1000.0,
            }
        )
    return row


def cli_step_row(
    meta: dict[str, Any],
    scenario: str,
    operation: str,
    command: list[str],
    cwd: Path,
    run_index: int,
    env: dict[str, str],
    admitted: int,
    full_stdout_bytes: Optional[int] = None,
) -> dict[str, Any]:
    capture = run_timed(command, cwd, env, admitted, full_output_bytes=full_stdout_bytes)
    row = {
        **meta,
        "scenario": scenario,
        "run": run_index,
        "operation": operation,
        "elapsed_ms": round(capture.elapsed_ms, 3),
        "returncode": capture.returncode,
        "command": command,
        "tool_call_count": 1,
        "terminal_tool_call_count": 1,
        **oakbench_tokens.interaction_token_fields(
            command_display(command),
            capture.stdout_text,
            capture.stderr_text,
            capture.stdout_bytes,
            capture.stderr_bytes,
            capture.stdout_truncated,
            capture.stderr_truncated,
        ),
        "_stdout": capture.stdout_full_text
        if capture.stdout_full_text is not None
        else capture.stdout_text,
    }
    if capture.returncode != 0:
        row["stderr"] = capture.stderr_text[-4000:]
    return row


def cli_sequence_row(
    meta: dict[str, Any],
    scenario: str,
    operation: str,
    commands: list[list[str]],
    cwd: Path,
    run_index: int,
    env: dict[str, str],
    admitted: int,
) -> dict[str, Any]:
    start = platform_clock.monotonic_ms()
    stdout_parts: list[str] = []
    stderr_parts: list[str] = []
    failed: Optional[subprocess.CompletedProcess[str]] = None
    for command in commands:
        proc = run_untimed(command, cwd, env)
        stdout_parts.append(proc.stdout)
        stderr_parts.append(proc.stderr)
        if proc.returncode != 0:
            failed = proc
            break
    elapsed_ms = platform_clock.monotonic_ms() - start
    stdout = "\n".join(part for part in stdout_parts if part)
    stderr = "\n".join(part for part in stderr_parts if part)
    command_text = "\n".join(command_display(command) for command in commands)
    row = {
        **meta,
        "scenario": scenario,
        "run": run_index,
        "operation": operation,
        "elapsed_ms": round(elapsed_ms, 3),
        "returncode": 0 if failed is None else int(failed.returncode),
        "command": commands,
        "tool_call_count": len(commands if failed is None else commands[: len(stdout_parts)]),
        "terminal_tool_call_count": len(commands if failed is None else commands[: len(stdout_parts)]),
        **oakbench_tokens.interaction_token_fields(
            command_text,
            stdout[:admitted],
            stderr[:admitted],
            len(stdout.encode("utf-8")),
            len(stderr.encode("utf-8")),
            len(stdout) > admitted,
            len(stderr) > admitted,
        ),
        "_stdout": stdout,
    }
    if failed is not None:
        row["stderr"] = stderr[-4000:]
    return row


def settle_fields(result: dict[str, Any]) -> dict[str, Any]:
    return {
        "settled": result["settled"],
        "settle_elapsed_ms": result["elapsed_ms"],
        "observed_wait_ms": round(result["observed_wait_ms"], 3),
        "poll_iterations": result["poll_iterations"],
        "poll_interval_s": result["poll_interval_s"],
        "poll_quantization_ms": result["poll_quantization_ms"],
    }


def settle_row(
    meta: dict[str, Any],
    scenario: str,
    operation: str,
    run_index: int,
    poll_fn: Callable[[], Any],
    settings: Settings,
    command: list[str],
    poll_is_tool_call: bool,
    clock: Callable[[], float] = time.monotonic,
    sleep: Callable[[float], None] = time.sleep,
) -> dict[str, Any]:
    result = platform_clock.settle(
        poll_fn,
        interval_s=settings.interval_s,
        timeout_s=settings.settle_timeout_s,
        clock=clock,
        sleep=sleep,
    )
    # Print the resolution bound the way devloop prints detection limits: a
    # settle observed at this cadence can never claim finer precision.
    print(
        f"[quantization] {operation} poll_interval_s={settings.interval_s} "
        f"poll_quantization_ms={result['poll_quantization_ms']}",
        flush=True,
    )
    return {
        **meta,
        "scenario": scenario,
        "run": run_index,
        "operation": operation,
        # The wall time spent on this operation is always measured; the
        # settle duration itself is null on timeout (platform_clock policy).
        "elapsed_ms": round(result["observed_wait_ms"], 3),
        "returncode": 0 if result["settled"] else 1,
        "command": command,
        "tool_call_count": result["poll_iterations"] if poll_is_tool_call else None,
        **settle_fields(result),
    }


def structural_row(
    meta: dict[str, Any],
    scenario: str,
    operation: str,
    run_index: int,
    note: str,
) -> dict[str, Any]:
    """A platform-semantics zero: the operation has no equivalent because the
    platform completes it synchronously elsewhere. Zero commands were run
    (measured zero); the settle value is null (no equivalent, ADR-0002)."""
    return {
        **meta,
        "scenario": scenario,
        "run": run_index,
        "operation": operation,
        "elapsed_ms": 0.0,
        "returncode": 0,
        "command": [],
        "structural": True,
        "note": note,
        "settled": True,
        "settle_elapsed_ms": None,
        "poll_iterations": 0,
        "tool_call_count": 0,
        "measurement_source": STRUCTURAL_SOURCE,
    }


def metrics_snapshot(client: GitHubClient) -> dict[str, Any]:
    metrics = client.api_metrics
    return {**metrics, "request_ids": list(metrics["request_ids"])}


def metrics_delta(before: dict[str, Any], after: dict[str, Any]) -> dict[str, Any]:
    delta_count = after["request_count"] - before["request_count"]
    ids = after["request_ids"][-delta_count:] if delta_count else []
    return {
        "request_count": delta_count,
        "request_bytes": after["request_bytes"] - before["request_bytes"],
        "response_bytes": after["response_bytes"] - before["response_bytes"],
        "rate_limit_remaining": after["rate_limit_remaining"],
        "request_ids": ids,
    }


def total_row(
    meta: dict[str, Any],
    scenario: str,
    run_index: int,
    rows: list[dict[str, Any]],
    driver: str,
) -> dict[str, Any]:
    failures = [
        row for row in rows if row_returncode(row) not in (0, SKIP_RETURNCODE)
    ]
    row: dict[str, Any] = {
        **meta,
        "scenario": scenario,
        "run": run_index,
        "operation": "integration.total",
        "elapsed_ms": round(sum(float(row.get("elapsed_ms") or 0.0) for row in rows), 3),
        "returncode": 1 if failures else 0,
        "command": [row["command"] for row in rows],
        "summarized_operations": [row["operation"] for row in rows],
        "poll_iterations_total": sum(int(row.get("poll_iterations") or 0) for row in rows),
    }
    if driver == "harness-api":
        row.update(null_token_fields())
        row["tool_call_count"] = None
        deltas = [row_["api_metrics"] for row_ in rows if isinstance(row_.get("api_metrics"), dict)]
        row["api_metrics"] = {
            "request_count": sum(delta["request_count"] for delta in deltas),
            "request_bytes": sum(delta["request_bytes"] for delta in deltas),
            "response_bytes": sum(delta["response_bytes"] for delta in deltas),
            "rate_limit_remaining": deltas[-1]["rate_limit_remaining"] if deltas else None,
        }
    else:
        row["tool_call_count"] = sum(int(row_.get("tool_call_count") or 0) for row_ in rows)
        row.update(
            oakbench_tokens.summed_token_fields(
                rows, "sum_of_platform_lifecycle_steps_command_plus_admitted_output_chars_div_4"
            )
        )
    return row


# --------------------------------------------------------------------------
# With-credentials runners (not exercised in CI; kept simple and linear).
# Each runner is driven by the scenario's operation list so emitted
# operations always match scenarios/platform.yaml (ADR-0005).
# --------------------------------------------------------------------------


def run_untimed(
    command: list[str], cwd: Optional[Path], env: dict[str, str], timeout: float = 300.0
) -> subprocess.CompletedProcess[str]:
    return subprocess.run(
        command,
        cwd=cwd,
        env=env,
        text=True,
        stdout=subprocess.PIPE,
        stderr=subprocess.PIPE,
        timeout=timeout,
        check=False,
    )


def run_operations(
    operations: list[str],
    handlers: dict[str, Callable[[], dict[str, Any]]],
    meta: dict[str, Any],
    scenario_name: str,
    run_index: int,
    driver: str,
) -> list[dict[str, Any]]:
    """Run the scenario's declared operations in order; a failed step turns
    the remaining steps into skip rows (a recorded gap, never a cascade of
    misleading failures)."""
    rows: list[dict[str, Any]] = []
    aborted_at: Optional[str] = None
    for operation in operations:
        if operation == "integration.total":
            rows.append(total_row(meta, scenario_name, run_index, rows, driver))
            continue
        if aborted_at is not None:
            rows.append(
                skip_row(meta, scenario_name, operation, run_index, f"prior_step_failed:{aborted_at}")
            )
            continue
        handler = handlers.get(operation)
        if handler is None:
            rows.append(skip_row(meta, scenario_name, operation, run_index, "operation_not_implemented"))
            continue
        row = handler()
        row.pop("_stdout", None)
        rows.append(row)
        if row_returncode(row) not in (0, SKIP_RETURNCODE):
            aborted_at = operation
    return rows


def run_capability_probe(
    args: argparse.Namespace, meta: dict[str, Any], run_index: int, settings: Settings
) -> list[dict[str, Any]]:
    scenario_name = "platform_capability_probe"
    if args.driver == "harness-api":
        if not args.repo:
            return [skip_row(meta, scenario_name, "capability.probe", run_index, "github_repo_not_configured")]
        client = GitHubClient(token=os.environ.get(args.token_env))
        return [
            api_step_row(
                meta, scenario_name, "capability.probe", run_index, client,
                lambda: client.get_repo(args.repo), [f"api:GET /repos/{args.repo}"],
            )
        ]
    command = ["gh", "auth", "status"] if args.platform == "github" else ["oak", "--version"]
    return [
        cli_step_row(
            meta, scenario_name, "capability.probe", command, Path.cwd(), run_index,
            cli_env(args), settings.admitted,
        )
    ]


def api_step_row(
    meta: dict[str, Any],
    scenario: str,
    operation: str,
    run_index: int,
    client: GitHubClient,
    func: Callable[[], Any],
    command: list[str],
) -> dict[str, Any]:
    before = metrics_snapshot(client)
    start = platform_clock.monotonic_ms()
    error: Optional[GitHubApiError] = None
    try:
        func()
    except GitHubApiError as exc:
        error = exc
    elapsed_ms = platform_clock.monotonic_ms() - start
    row = {
        **meta,
        "scenario": scenario,
        "run": run_index,
        "operation": operation,
        "elapsed_ms": round(elapsed_ms, 3),
        "returncode": 0 if error is None else 1,
        "command": command,
        "tool_call_count": None,
        "api_metrics": metrics_delta(before, metrics_snapshot(client)),
        **null_token_fields(),
    }
    if error is not None:
        row["stderr"] = str(error)[:4000]
    return row


def api_settle_row(
    meta: dict[str, Any],
    scenario: str,
    operation: str,
    run_index: int,
    client: GitHubClient,
    poll_fn: Callable[[], Any],
    settings: Settings,
    command: list[str],
    clock: Callable[[], float] = time.monotonic,
    sleep: Callable[[float], None] = time.sleep,
) -> dict[str, Any]:
    before = metrics_snapshot(client)
    row = settle_row(
        meta, scenario, operation, run_index, poll_fn, settings, command,
        poll_is_tool_call=False, clock=clock, sleep=sleep,
    )
    row["api_metrics"] = metrics_delta(before, metrics_snapshot(client))
    row.update(null_token_fields())
    return row


def api_check_instant_checks_green_row(
    meta: dict[str, Any],
    scenario_name: str,
    run_index: int,
    client: GitHubClient,
    repo: str,
    head_sha: str,
    settings: Settings,
    *,
    pace: Callable[[], None] = lambda: None,
    clock: Callable[[], float] = time.monotonic,
    sleep: Callable[[float], None] = time.sleep,
) -> dict[str, Any]:
    """check-instant, executed: the harness posts the required status itself
    via the commit status API (deterministic, free, no Actions), then settles
    on the combined status going green before merge. Shared by the race
    runner and the non-race harness-api runner so the variant means the same
    thing wherever it is executed (ADR-0005)."""
    operation = CHECKS_GREEN_OPERATION
    command = [
        f"api:POST /repos/{repo}/statuses/<head-sha>",
        f"api:GET /repos/{repo}/commits/<head-sha>/status (poll)",
    ]
    before = metrics_snapshot(client)
    error_text: Optional[str] = None
    post_start = platform_clock.monotonic_ms()
    if not head_sha:
        error_text = "head_sha_unknown_cannot_post_status"
    else:
        try:
            pace()
            client.post_status(
                repo, head_sha, "success", CHECK_INSTANT_CONTEXT,
                description="oakbench harness-posted instant check",
            )
        except GitHubApiError as exc:
            error_text = str(exc)[:4000]
    post_elapsed_ms = platform_clock.monotonic_ms() - post_start
    if error_text is not None:
        return {
            **meta,
            "scenario": scenario_name,
            "run": run_index,
            "operation": operation,
            "elapsed_ms": round(post_elapsed_ms, 3),
            "returncode": 1,
            "command": command,
            "tool_call_count": None,
            "stderr": error_text,
            "api_metrics": metrics_delta(before, metrics_snapshot(client)),
            **null_token_fields(),
        }

    def poll() -> bool:
        try:
            combined = client.get_combined_status(repo, head_sha)
        except GitHubApiError:
            return False
        return combined.get("state") == "success"

    result = platform_clock.settle(
        poll, interval_s=settings.interval_s, timeout_s=settings.settle_timeout_s,
        clock=clock, sleep=sleep,
    )
    return {
        **meta,
        "scenario": scenario_name,
        "run": run_index,
        "operation": operation,
        # post + settle: the harness-posted status is part of the
        # operation's measured cost, never hidden setup.
        "elapsed_ms": round(post_elapsed_ms + result["observed_wait_ms"], 3),
        "returncode": 0 if result["settled"] else 1,
        "command": command,
        "tool_call_count": None,
        **settle_fields(result),
        "status_post_elapsed_ms": round(post_elapsed_ms, 3),
        "status_context": CHECK_INSTANT_CONTEXT,
        "api_metrics": metrics_delta(before, metrics_snapshot(client)),
        **null_token_fields(),
    }


def run_github_api_scenario(
    args: argparse.Namespace,
    scenario: dict[str, Any],
    meta: dict[str, Any],
    run_index: int,
    settings: Settings,
    *,
    client: Optional[GitHubClient] = None,
    clock: Callable[[], float] = time.monotonic,
    sleep: Callable[[float], None] = time.sleep,
) -> list[dict[str, Any]]:
    scenario_name = str(scenario["name"])
    operations = [str(op) for op in scenario.get("operations") or []]
    check_instant = args.protection_variant == "check-instant"
    if check_instant and CHECKS_GREEN_OPERATION not in operations:
        # check-instant is only executed when the required status is actually
        # posted and settled green BEFORE the merge request; the non-race
        # anatomy scenarios do not declare the operation, so it is inserted
        # ahead of integration.merge.request (the race ordering).
        anchor = (
            operations.index("integration.merge.request")
            if "integration.merge.request" in operations
            else len(operations)
        )
        operations = operations[:anchor] + [CHECKS_GREEN_OPERATION] + operations[anchor:]
    repo = args.repo
    if client is None:
        client = GitHubClient(token=os.environ.get(args.token_env))
    branch = f"oakbench-platform-{meta['bench_id'].lower()}-r{run_index}"
    payload = f"platform lane {meta['bench_id']} run={run_index}\n".encode("utf-8")
    state: dict[str, Any] = {}

    try:
        state["base_sha"] = client.get_main_sha(repo)
    except (GitHubApiError, KeyError, json.JSONDecodeError) as exc:
        return [
            skip_row(meta, scenario_name, op, run_index, f"setup_failed:{str(exc)[:160]}")
            for op in operations
        ]

    def publish() -> dict[str, Any]:
        def do() -> None:
            client.create_ref(repo, f"refs/heads/{branch}", state["base_sha"])
            put = client.put_file(
                repo,
                path=f".oak-bench/platform-{run_index}.md",
                message=f"bench platform lane {meta['bench_id']} r{run_index}",
                content=payload,
                branch=branch,
            )
            commit = put.get("commit") if isinstance(put, dict) else None
            if isinstance(commit, dict) and commit.get("sha"):
                state["head_sha"] = str(commit["sha"])
        return api_step_row(
            meta, scenario_name, "branch.publish", run_index, client, do,
            [f"api:POST /repos/{repo}/git/refs", f"api:PUT /repos/{repo}/contents/..."],
        )

    def open_pr() -> dict[str, Any]:
        def do() -> None:
            pr = client.create_pr(
                repo, title=f"bench platform lane r{run_index}", head=branch, base="main"
            )
            state["pr_number"] = int(pr["number"])
            head = pr.get("head") or {}
            if not state.get("head_sha") and head.get("sha"):
                state["head_sha"] = str(head["sha"])
        return api_step_row(
            meta, scenario_name, "integration.open", run_index, client, do,
            [f"api:POST /repos/{repo}/pulls"],
        )

    def get_pr_quiet() -> Optional[dict[str, Any]]:
        try:
            return client.get_pr(repo, state["pr_number"])
        except GitHubApiError:
            return None

    def visible() -> dict[str, Any]:
        return api_settle_row(
            meta, scenario_name, "integration.visible.settle", run_index, client,
            lambda: get_pr_quiet() is not None, settings,
            [f"api:GET /repos/{repo}/pulls/<n> (poll)"], clock=clock, sleep=sleep,
        )

    def mergeable() -> dict[str, Any]:
        def poll() -> bool:
            pr = get_pr_quiet()
            return pr is not None and pr.get("mergeable") is not None
        return api_settle_row(
            meta, scenario_name, "integration.mergeable.settle", run_index, client,
            poll, settings, [f"api:GET /repos/{repo}/pulls/<n> mergeable (poll)"],
            clock=clock, sleep=sleep,
        )

    def checks_green() -> dict[str, Any]:
        # Reached only under check-instant (the operation is inserted above);
        # the status is posted after the PR is open and must settle green
        # before integration.merge.request runs.
        return api_check_instant_checks_green_row(
            meta, scenario_name, run_index, client, repo,
            str(state.get("head_sha") or ""), settings, clock=clock, sleep=sleep,
        )

    def merge() -> dict[str, Any]:
        return api_step_row(
            meta, scenario_name, "integration.merge.request", run_index, client,
            lambda: client.merge_pr(repo, state["pr_number"]),
            [f"api:PUT /repos/{repo}/pulls/<n>/merge"],
        )

    def merged() -> dict[str, Any]:
        def poll() -> bool:
            pr = get_pr_quiet()
            return bool(pr and pr.get("merged"))
        return api_settle_row(
            meta, scenario_name, "integration.merged.settle", run_index, client,
            poll, settings, [f"api:GET /repos/{repo}/pulls/<n> merged (poll)"],
            clock=clock, sleep=sleep,
        )

    def main_settle() -> dict[str, Any]:
        def poll() -> bool:
            try:
                return client.get_main_sha(repo) != state["base_sha"]
            except GitHubApiError:
                return False
        return api_settle_row(
            meta, scenario_name, "integration.main.settle", run_index, client,
            poll, settings, [f"api:GET /repos/{repo}/git/ref/heads/main (poll)"],
            clock=clock, sleep=sleep,
        )

    def sync_local() -> dict[str, Any]:
        row = api_step_row(
            meta, scenario_name, "integration.sync.local", run_index, client,
            lambda: client.get_main_sha(repo),
            [f"api:GET /repos/{repo}/git/ref/heads/main"],
        )
        row["note"] = (
            "harness-api driver has no local working copy; this is the API-side "
            "equivalent (observe merged main ref). Local-checkout sync cost is a "
            "cli-driver measurement."
        )
        return row

    handlers = {
        "branch.publish": publish,
        "integration.open": open_pr,
        "integration.visible.settle": visible,
        "integration.mergeable.settle": mergeable,
        CHECKS_GREEN_OPERATION: checks_green,
        "integration.merge.request": merge,
        "integration.merged.settle": merged,
        "integration.main.settle": main_settle,
        "integration.sync.local": sync_local,
    }
    rows = run_operations(operations, handlers, meta, scenario_name, run_index, "harness-api")
    try:
        client.delete_ref(repo, f"heads/{branch}")
    except GitHubApiError:
        pass  # untimed cleanup; merged-branch deletion failures are not data
    return rows


def run_github_cli_scenario(
    args: argparse.Namespace,
    scenario: dict[str, Any],
    meta: dict[str, Any],
    run_index: int,
    run_root: Path,
    settings: Settings,
) -> list[dict[str, Any]]:
    scenario_name = str(scenario["name"])
    operations = [str(op) for op in scenario.get("operations") or []]
    if args.protection_variant == "check-instant":
        # gh/git cannot post the harness commit status, so the cli driver
        # cannot execute this variant: running anyway would merge (or fail)
        # while the rows claimed the protection held. Honest gap instead.
        return [
            skip_row(
                meta,
                scenario_name,
                op,
                run_index,
                CHECK_INSTANT_CLI_SKIP_REASON,
                settings=settings,
            )
            for op in operations
        ]
    repo = args.repo
    env = cli_env(args)
    # Time the real Git, not the macOS xcrun trampoline, including the git
    # that gh spawns internally (ADR-0011): put the resolved Git first on PATH.
    from oakbench.subjects import which_measured_git
    real_git = which_measured_git()
    if real_git:
        env = {**env, "PATH": str(Path(real_git).parent) + os.pathsep + env.get("PATH", os.environ.get("PATH", ""))}
    git = real_git or "git"
    branch = f"oakbench-platform-{meta['bench_id'].lower()}-r{run_index}"
    workdir = run_root / f"github-cli-r{run_index}"
    state: dict[str, Any] = {}

    # Untimed setup: the lane's precondition is "task branch exists locally".
    def setup() -> Optional[str]:
        clone = run_untimed(["gh", "repo", "clone", repo, str(workdir)], run_root, env)
        if clone.returncode != 0:
            return f"setup_failed:clone:{clone.stderr.strip()[:120]}"
        for command in (
            [git, "switch", "-c", branch],
            ["/bin/zsh", "-c", f"mkdir -p .oak-bench && echo 'platform lane r{run_index}' > .oak-bench/platform-{run_index}.md"],
            [git, "add", "."],
            [git, "commit", "-q", "-m", f"bench platform lane {meta['bench_id']} r{run_index}"],
        ):
            proc = run_untimed(command, workdir, env)
            if proc.returncode != 0:
                return f"setup_failed:{command[0]}:{proc.stderr.strip()[:120]}"
        ls = run_untimed([git, "ls-remote", "origin", "refs/heads/main"], workdir, env)
        state["base_sha"] = ls.stdout.split()[0] if ls.returncode == 0 and ls.stdout.split() else ""
        return None

    setup_error = setup()
    if setup_error:
        return [skip_row(meta, scenario_name, op, run_index, setup_error) for op in operations]

    def remote_main_sha() -> str:
        ls = run_untimed([git, "ls-remote", "origin", "refs/heads/main"], workdir, env)
        parts = ls.stdout.split()
        return parts[0] if ls.returncode == 0 and parts else ""

    def pr_number() -> str:
        return str(state.get("pr_number", ""))

    def publish() -> dict[str, Any]:
        return cli_step_row(
            meta, scenario_name, "branch.publish",
            [git, "push", "-q", "-u", "origin", branch], workdir, run_index, env, settings.admitted,
        )

    def open_pr() -> dict[str, Any]:
        row = cli_step_row(
            meta, scenario_name, "integration.open",
            [
                "gh", "pr", "create", "--repo", repo, "--head", branch, "--base", "main",
                "--title", f"bench platform lane r{run_index}",
                "--body", f"Disposable platform-lane benchmark PR ({meta['bench_id']}).",
            ],
            workdir, run_index, env, settings.admitted,
        )
        match = PR_URL_RE.search(row.get("_stdout", ""))
        if match:
            state["pr_number"] = int(match.group(1))
        elif row["returncode"] == 0:
            row["returncode"] = 1
            row["stderr"] = "pr_number_unparsed_from_gh_output"
        return row

    def gh_pr_json(field: str) -> str:
        proc = run_untimed(
            ["gh", "pr", "view", pr_number(), "--repo", repo, "--json", field, "-q", f".{field}"],
            workdir, env,
        )
        return proc.stdout.strip() if proc.returncode == 0 else ""

    def visible() -> dict[str, Any]:
        return settle_row(
            meta, scenario_name, "integration.visible.settle", run_index,
            lambda: bool(gh_pr_json("state")), settings,
            ["gh", "pr", "view", "<n>", "--json", "state", "(poll)"], poll_is_tool_call=True,
        )

    def mergeable() -> dict[str, Any]:
        return settle_row(
            meta, scenario_name, "integration.mergeable.settle", run_index,
            lambda: gh_pr_json("mergeable") not in ("", "UNKNOWN"), settings,
            ["gh", "pr", "view", "<n>", "--json", "mergeable", "(poll)"], poll_is_tool_call=True,
        )

    def merge() -> dict[str, Any]:
        return cli_step_row(
            meta, scenario_name, "integration.merge.request",
            ["gh", "pr", "merge", pr_number(), "--repo", repo, "--merge"],
            workdir, run_index, env, settings.admitted,
        )

    def merged() -> dict[str, Any]:
        return settle_row(
            meta, scenario_name, "integration.merged.settle", run_index,
            lambda: gh_pr_json("state") == "MERGED", settings,
            ["gh", "pr", "view", "<n>", "--json", "state", "(poll)"], poll_is_tool_call=True,
        )

    def main_settle() -> dict[str, Any]:
        return settle_row(
            meta, scenario_name, "integration.main.settle", run_index,
            lambda: remote_main_sha() not in ("", state["base_sha"]), settings,
            ["git", "ls-remote", "origin", "refs/heads/main", "(poll)"], poll_is_tool_call=True,
        )

    def sync_local() -> dict[str, Any]:
        run_untimed([git, "switch", "-q", "main"], workdir, env)
        return cli_step_row(
            meta, scenario_name, "integration.sync.local",
            [git, "pull", "-q", "--ff-only", "origin", "main"],
            workdir, run_index, env, settings.admitted,
        )

    handlers = {
        "branch.publish": publish,
        "integration.open": open_pr,
        "integration.visible.settle": visible,
        "integration.mergeable.settle": mergeable,
        "integration.merge.request": merge,
        "integration.merged.settle": merged,
        "integration.main.settle": main_settle,
        "integration.sync.local": sync_local,
    }
    rows = run_operations(operations, handlers, meta, scenario_name, run_index, "cli")
    run_untimed([git, "push", "-q", "origin", "--delete", branch], workdir, env)  # untimed cleanup
    return rows


def run_oak_cli_scenario(
    args: argparse.Namespace,
    scenario: dict[str, Any],
    meta: dict[str, Any],
    run_index: int,
    run_root: Path,
    settings: Settings,
) -> list[dict[str, Any]]:
    scenario_name = str(scenario["name"])
    operations = [str(op) for op in scenario.get("operations") or []]
    env = cli_env(args)
    workdir = run_root / f"oak-cli-r{run_index}"

    # Untimed setup: mount and commit so the task branch exists locally.
    mount = run_untimed(["oak", "mount", args.repo, str(workdir)], run_root, env)
    if mount.returncode != 0:
        reason = f"setup_failed:oak_mount:{mount.stderr.strip()[:120]}"
        return [skip_row(meta, scenario_name, op, run_index, reason) for op in operations]
    payload_dir = workdir / ".oak-bench"
    payload_dir.mkdir(parents=True, exist_ok=True)
    (payload_dir / f"platform-{run_index}.md").write_text(
        f"platform lane {meta['bench_id']} run={run_index}\n"
    )
    commit = run_untimed(["oak", "commit", "--no-verify"], workdir, env)
    if commit.returncode != 0:
        run_untimed(["oak", "mount", "end", str(workdir), "-f"], run_root, env)
        reason = f"setup_failed:oak_commit:{commit.stderr.strip()[:120]}"
        return [skip_row(meta, scenario_name, op, run_index, reason) for op in operations]

    def cli(operation: str, command: list[str]) -> Callable[[], dict[str, Any]]:
        return lambda: cli_step_row(
            meta, scenario_name, operation, command, workdir, run_index, env, settings.admitted
        )

    def structural(operation: str, note: str) -> Callable[[], dict[str, Any]]:
        return lambda: structural_row(meta, scenario_name, operation, run_index, note)

    # oak merge is synchronous: zero polls is structural (a measured zero —
    # no commands run), and the settle value is null because the async
    # concept has no equivalent on this platform (ADR-0002).
    handlers = {
        "branch.publish": cli("branch.publish", ["oak", "push"]),
        "integration.open": structural(
            "integration.open",
            "integration object exists at publish on oak; no separate open call",
        ),
        "integration.visible.settle": structural(
            "integration.visible.settle",
            "oak push returns after the branch is visible on the server; nothing to poll",
        ),
        "integration.mergeable.settle": structural(
            "integration.mergeable.settle", "no async mergeability concept"
        ),
        "integration.merge.request": cli("integration.merge.request", ["oak", "merge"]),
        "integration.merged.settle": structural(
            "integration.merged.settle", "oak merge is synchronous; merged state holds at return"
        ),
        "integration.main.settle": structural(
            "integration.main.settle", "oak merge returns after main is updated; nothing to poll"
        ),
        "integration.sync.local": cli("integration.sync.local", ["oak", "switch", "main"]),
    }
    rows = run_operations(operations, handlers, meta, scenario_name, run_index, "cli")
    run_untimed(["oak", "mount", "end", str(workdir), "-f"], run_root, env)  # untimed cleanup
    return rows


# --------------------------------------------------------------------------
# integration_race_nN: the flagship race family. N task branches each go
# through publish -> PR -> merge (per-branch anatomy rows reuse the
# vocabulary above), landing serially or as fast as RateLimitBudgeter pacing
# allows, plus race-level rows: race.publish.window (first publish start ->
# last publish end), race.integrate.window (first integration.open -> last
# integration.main.settle), race.pacing.injected (sum of budgeter-injected
# waits - REPORTED, never subtracted), race.total. clock/sleep are injectable
# so tests never wait on real time (fake transports only, no network).
# --------------------------------------------------------------------------


def is_race_scenario(scenario: dict[str, Any]) -> bool:
    return RACE_SCENARIO_RE.match(str(scenario.get("name", ""))) is not None


def race_branch_count(scenario: dict[str, Any]) -> int:
    if scenario.get("branch_count") is not None:
        return int(scenario["branch_count"])
    match = RACE_SCENARIO_RE.match(str(scenario.get("name", "")))
    return int(match.group(1)) if match else 1


def race_branch_skip(
    meta: dict[str, Any],
    scenario_name: str,
    operation: str,
    run_index: int,
    reason: str,
    branch_index: int,
) -> dict[str, Any]:
    row = skip_row(meta, scenario_name, operation, run_index, reason)
    row["race_branch"] = branch_index
    return row


def race_window_ms(
    timeline: dict[str, list[tuple[float, float]]], start_op: str, end_op: str
) -> Optional[float]:
    """First start of ``start_op`` -> last end of ``end_op``, on the race's
    single monotonic clock. None when either endpoint was never observed."""
    starts = timeline.get(start_op) or []
    ends = timeline.get(end_op) or []
    if not starts or not ends:
        return None
    return (max(end for _, end in ends) - min(start for start, _ in starts)) * 1000.0


def race_summary_rows(
    meta: dict[str, Any],
    scenario_name: str,
    run_index: int,
    driver: str,
    race_operations: list[str],
    branch_rows: list[dict[str, Any]],
    timeline: dict[str, list[tuple[float, float]]],
    budgeter: RateLimitBudgeter,
    race_start: float,
    race_end: float,
    branch_count: int,
) -> list[dict[str, Any]]:
    failures = [
        row for row in branch_rows
        if row_returncode(row) not in (0, SKIP_RETURNCODE)
    ]
    failed_operations = {str(row.get("operation")) for row in failures}

    def base_row(operation: str, elapsed_ms: float, returncode: int, note: str) -> dict[str, Any]:
        row: dict[str, Any] = {
            **meta,
            "scenario": scenario_name,
            "run": run_index,
            "operation": operation,
            "elapsed_ms": round(elapsed_ms, 3),
            "returncode": returncode,
            "command": [],
            "tool_call_count": None,
            "race_branch_count": branch_count,
            "note": note,
        }
        if driver == "harness-api":
            row.update(null_token_fields())
        return row

    def window_row(operation: str, start_op: str, end_op: str, note: str) -> dict[str, Any]:
        window = race_window_ms(timeline, start_op, end_op)
        if window is None:
            # The window's endpoints were never observed (e.g. every branch
            # aborted before reaching them): a recorded gap, never a zero.
            row = skip_row(
                meta, scenario_name, operation, run_index,
                f"race_window_unobserved:{start_op}->{end_op}",
            )
            row["race_branch_count"] = branch_count
            return row
        complete = (
            len(timeline.get(start_op) or []) == branch_count
            and len(timeline.get(end_op) or []) == branch_count
            and start_op not in failed_operations
            and end_op not in failed_operations
        )
        row = base_row(operation, window, 0 if complete else 1, note)
        row["window_start_operation"] = start_op
        row["window_end_operation"] = end_op
        return row

    def pacing_row() -> dict[str, Any]:
        row = base_row(
            "race.pacing.injected",
            budgeter.injected_wait_s_total * 1000.0,
            0,
            "sum of RateLimitBudgeter-injected pacing waits across the race; "
            "REPORTED here, never subtracted from any operation duration",
        )
        row["budget_ops_per_hour"] = budgeter.ops_per_hour
        return row

    def total_row_race() -> dict[str, Any]:
        row = base_row(
            "race.total",
            (race_end - race_start) * 1000.0,
            1 if failures else 0,
            "whole-race wall clock on one local monotonic clock: before the "
            "first branch's publish to after the last branch's final operation",
        )
        row["summarized_operations"] = [r["operation"] for r in branch_rows]
        row["poll_iterations_total"] = sum(
            int(r.get("poll_iterations") or 0) for r in branch_rows
        )
        if driver == "harness-api":
            deltas = [
                r["api_metrics"] for r in branch_rows if isinstance(r.get("api_metrics"), dict)
            ]
            row["api_metrics"] = {
                "request_count": sum(delta["request_count"] for delta in deltas),
                "request_bytes": sum(delta["request_bytes"] for delta in deltas),
                "response_bytes": sum(delta["response_bytes"] for delta in deltas),
                "rate_limit_remaining": deltas[-1]["rate_limit_remaining"] if deltas else None,
            }
        else:
            row["tool_call_count"] = sum(
                int(r.get("tool_call_count") or 0) for r in branch_rows
            )
            row.update(
                oakbench_tokens.summed_token_fields(
                    branch_rows,
                    "sum_of_platform_lifecycle_steps_command_plus_admitted_output_chars_div_4",
                )
            )
        return row

    emitters: dict[str, Callable[[], dict[str, Any]]] = {
        "race.publish.window": lambda: window_row(
            "race.publish.window", "branch.publish", "branch.publish",
            "first branch.publish start -> last branch.publish end across all branches",
        ),
        "race.integrate.window": lambda: window_row(
            "race.integrate.window", "integration.open", "integration.main.settle",
            "first integration.open start -> last integration.main.settle end across all branches",
        ),
        "race.pacing.injected": pacing_row,
        "race.total": total_row_race,
    }
    rows: list[dict[str, Any]] = []
    for operation in race_operations:
        emitter = emitters.get(operation)
        if emitter is None:
            rows.append(
                skip_row(meta, scenario_name, operation, run_index, "operation_not_implemented")
            )
            continue
        rows.append(emitter())
    return rows


# A branch factory does the branch's untimed setup and returns
# (setup_error_or_None, operation handlers, untimed cleanup callable).
RaceBranchFactory = Callable[
    [int, str],
    tuple[Optional[str], dict[str, Callable[[], dict[str, Any]]], Callable[[], None]],
]


def run_race_operations(
    scenario: dict[str, Any],
    meta: dict[str, Any],
    run_index: int,
    driver: str,
    branch_count: int,
    make_branch: RaceBranchFactory,
    budgeter: RateLimitBudgeter,
    clock: Callable[[], float],
) -> list[dict[str, Any]]:
    """Race orchestration shared by drivers: loop branch-0..N-1 serially,
    run each branch's declared per-branch operations, record a start/end
    timeline for the race windows, then emit the race-level rows. A failed
    step turns every remaining per-branch step into a skip row (a recorded
    gap, never a cascade of misleading failures)."""
    scenario_name = str(scenario["name"])
    operations = [str(op) for op in scenario.get("operations") or []]
    branch_operations = [op for op in operations if not op.startswith("race.")]
    race_operations = [op for op in operations if op.startswith("race.")]

    rows: list[dict[str, Any]] = []
    timeline: dict[str, list[tuple[float, float]]] = {}
    cleanups: list[Callable[[], None]] = []
    aborted_at: Optional[str] = None
    race_start = clock()

    for branch_index in range(branch_count):
        branch = disposable_branch(meta["bench_id"], scenario_name, f"b{branch_index}", run_index)
        if aborted_at is not None:
            rows.extend(
                race_branch_skip(
                    meta, scenario_name, operation, run_index,
                    f"prior_step_failed:{aborted_at}", branch_index,
                )
                for operation in branch_operations
            )
            continue
        setup_error, handlers, cleanup = make_branch(branch_index, branch)
        cleanups.append(cleanup)
        if setup_error is not None:
            aborted_at = f"branch-{branch_index}:setup"
            rows.extend(
                race_branch_skip(meta, scenario_name, operation, run_index, setup_error, branch_index)
                for operation in branch_operations
            )
            continue
        for operation in branch_operations:
            if aborted_at is not None:
                rows.append(
                    race_branch_skip(
                        meta, scenario_name, operation, run_index,
                        f"prior_step_failed:{aborted_at}", branch_index,
                    )
                )
                continue
            handler = handlers.get(operation)
            if handler is None:
                rows.append(
                    race_branch_skip(
                        meta, scenario_name, operation, run_index,
                        "operation_not_implemented", branch_index,
                    )
                )
                continue
            injected_before = budgeter.injected_wait_s_total
            start = clock()
            row = handler()
            end = clock()
            row.pop("_stdout", None)
            row["race_branch"] = branch_index
            row["race_branch_name"] = branch
            injected_s = budgeter.injected_wait_s_total - injected_before
            if injected_s > 0:
                row["pacing_injected_ms"] = round(injected_s * 1000.0, 3)
                row["pacing_note"] = (
                    "elapsed_ms includes this budgeter-injected wait; it is "
                    "reported here and summed into race.pacing.injected, never subtracted"
                )
            timeline.setdefault(operation, []).append((start, end))
            rows.append(row)
            if row_returncode(row) not in (0, SKIP_RETURNCODE):
                aborted_at = f"branch-{branch_index}:{operation}"

    race_end = clock()
    rows.extend(
        race_summary_rows(
            meta, scenario_name, run_index, driver, race_operations, list(rows),
            timeline, budgeter, race_start, race_end, branch_count,
        )
    )
    for cleanup in cleanups:
        cleanup()  # untimed best-effort cleanup; cleanup failures are not data
    return rows


def run_github_api_race(
    args: argparse.Namespace,
    scenario: dict[str, Any],
    meta: dict[str, Any],
    run_index: int,
    settings: Settings,
    *,
    client: Optional[GitHubClient] = None,
    budgeter: Optional[RateLimitBudgeter] = None,
    sleep: Callable[[float], None] = time.sleep,
    clock: Callable[[], float] = time.monotonic,
) -> list[dict[str, Any]]:
    scenario_name = str(scenario["name"])
    repo = args.repo
    if client is None:
        client = GitHubClient(token=os.environ.get(args.token_env))
    if budgeter is None:
        budgeter = RateLimitBudgeter(DEFAULT_RACE_OPS_PER_HOUR)
    check_instant = args.protection_variant == "check-instant"

    def pace_content_op() -> None:
        # Pace before every content (write) op; the budgeter accumulates the
        # injected waits, which race.pacing.injected reports.
        wait = budgeter.pace()
        if wait > 0:
            sleep(wait)

    def make_branch(branch_index: int, branch: str) -> tuple[
        Optional[str], dict[str, Callable[[], dict[str, Any]]], Callable[[], None]
    ]:
        state: dict[str, Any] = {}
        try:
            # Serial landing: each branch baselines whatever main is now.
            state["base_sha"] = client.get_main_sha(repo)
        except (GitHubApiError, KeyError, json.JSONDecodeError) as exc:
            return f"setup_failed:{str(exc)[:160]}", {}, lambda: None

        payload = (
            f"platform race {meta['bench_id']} run={run_index} branch={branch_index}\n"
        ).encode("utf-8")

        def publish() -> dict[str, Any]:
            def do() -> None:
                pace_content_op()
                client.create_ref(repo, f"refs/heads/{branch}", state["base_sha"])
                pace_content_op()
                put = client.put_file(
                    repo,
                    path=f".oak-bench/race-{run_index}-b{branch_index}.md",
                    message=f"bench race {meta['bench_id']} r{run_index} b{branch_index}",
                    content=payload,
                    branch=branch,
                )
                commit = put.get("commit") if isinstance(put, dict) else None
                if isinstance(commit, dict) and commit.get("sha"):
                    state["head_sha"] = str(commit["sha"])
            return api_step_row(
                meta, scenario_name, "branch.publish", run_index, client, do,
                [f"api:POST /repos/{repo}/git/refs", f"api:PUT /repos/{repo}/contents/..."],
            )

        def open_pr() -> dict[str, Any]:
            def do() -> None:
                pace_content_op()
                pr = client.create_pr(
                    repo,
                    title=f"bench race {scenario_name} r{run_index} b{branch_index}",
                    head=branch,
                    base="main",
                )
                state["pr_number"] = int(pr["number"])
                head = pr.get("head") or {}
                if not state.get("head_sha") and head.get("sha"):
                    state["head_sha"] = str(head["sha"])
            return api_step_row(
                meta, scenario_name, "integration.open", run_index, client, do,
                [f"api:POST /repos/{repo}/pulls"],
            )

        def get_pr_quiet() -> Optional[dict[str, Any]]:
            try:
                return client.get_pr(repo, state["pr_number"])
            except GitHubApiError:
                return None

        def visible() -> dict[str, Any]:
            return api_settle_row(
                meta, scenario_name, "integration.visible.settle", run_index, client,
                lambda: get_pr_quiet() is not None, settings,
                [f"api:GET /repos/{repo}/pulls/<n> (poll)"], clock=clock, sleep=sleep,
            )

        def mergeable() -> dict[str, Any]:
            def poll() -> bool:
                pr = get_pr_quiet()
                return pr is not None and pr.get("mergeable") is not None
            return api_settle_row(
                meta, scenario_name, "integration.mergeable.settle", run_index, client,
                poll, settings, [f"api:GET /repos/{repo}/pulls/<n> mergeable (poll)"],
                clock=clock, sleep=sleep,
            )

        def checks_green() -> dict[str, Any]:
            if not check_instant:
                return structural_row(
                    meta, scenario_name, CHECKS_GREEN_OPERATION, run_index,
                    "protection-none: no required status checks; nothing to wait for",
                )
            return api_check_instant_checks_green_row(
                meta, scenario_name, run_index, client, repo,
                str(state.get("head_sha") or ""), settings,
                pace=pace_content_op, clock=clock, sleep=sleep,
            )

        def merge() -> dict[str, Any]:
            def do() -> None:
                pace_content_op()
                client.merge_pr(repo, state["pr_number"])
            return api_step_row(
                meta, scenario_name, "integration.merge.request", run_index, client, do,
                [f"api:PUT /repos/{repo}/pulls/<n>/merge"],
            )

        def merged() -> dict[str, Any]:
            def poll() -> bool:
                pr = get_pr_quiet()
                return bool(pr and pr.get("merged"))
            return api_settle_row(
                meta, scenario_name, "integration.merged.settle", run_index, client,
                poll, settings, [f"api:GET /repos/{repo}/pulls/<n> merged (poll)"],
                clock=clock, sleep=sleep,
            )

        def main_settle() -> dict[str, Any]:
            def poll() -> bool:
                try:
                    return client.get_main_sha(repo) != state["base_sha"]
                except GitHubApiError:
                    return False
            return api_settle_row(
                meta, scenario_name, "integration.main.settle", run_index, client,
                poll, settings, [f"api:GET /repos/{repo}/git/ref/heads/main (poll)"],
                clock=clock, sleep=sleep,
            )

        def cleanup() -> None:
            try:
                client.delete_ref(repo, f"heads/{branch}")
            except GitHubApiError:
                pass  # untimed cleanup; merged-branch deletion failures are not data

        handlers = {
            "branch.publish": publish,
            "integration.open": open_pr,
            "integration.visible.settle": visible,
            "integration.mergeable.settle": mergeable,
            "integration.checks.green.settle": checks_green,
            "integration.merge.request": merge,
            "integration.merged.settle": merged,
            "integration.main.settle": main_settle,
        }
        return None, handlers, cleanup

    return run_race_operations(
        scenario, meta, run_index, "harness-api", race_branch_count(scenario),
        make_branch, budgeter, clock,
    )


def run_oak_cli_race(
    args: argparse.Namespace,
    scenario: dict[str, Any],
    meta: dict[str, Any],
    run_index: int,
    run_root: Path,
    settings: Settings,
    *,
    budgeter: Optional[RateLimitBudgeter] = None,
    sleep: Callable[[float], None] = time.sleep,
    clock: Callable[[], float] = time.monotonic,
) -> list[dict[str, Any]]:
    scenario_name = str(scenario["name"])
    env = cli_env(args)
    if budgeter is None:
        budgeter = RateLimitBudgeter(DEFAULT_RACE_OPS_PER_HOUR)

    def pace_content_op() -> None:
        wait = budgeter.pace()
        if wait > 0:
            sleep(wait)

    def make_branch(branch_index: int, branch: str) -> tuple[
        Optional[str], dict[str, Callable[[], dict[str, Any]]], Callable[[], None]
    ]:
        workdir = run_root / f"oak-cli-race-r{run_index}-b{branch_index}"
        mount = run_untimed(["oak", "mount", args.repo, str(workdir)], run_root, env)
        if mount.returncode != 0:
            return (
                f"setup_failed:oak_mount:{mount.stderr.strip()[:120]}", {}, lambda: None
            )

        def cleanup() -> None:
            run_untimed(["oak", "mount", "end", str(workdir), "-f"], run_root, env)

        payload_dir = workdir / ".oak-bench"
        payload_dir.mkdir(parents=True, exist_ok=True)
        (payload_dir / f"race-{run_index}-b{branch_index}.md").write_text(
            f"platform race {meta['bench_id']} run={run_index} branch={branch_index}\n"
        )
        commit = run_untimed(["oak", "commit", "--no-verify"], workdir, env)
        if commit.returncode != 0:
            return f"setup_failed:oak_commit:{commit.stderr.strip()[:120]}", {}, cleanup

        def paced_cli(operation: str, command: list[str]) -> Callable[[], dict[str, Any]]:
            def handler() -> dict[str, Any]:
                pace_content_op()
                return cli_step_row(
                    meta, scenario_name, operation, command, workdir, run_index, env,
                    settings.admitted,
                )
            return handler

        def structural(operation: str, note: str) -> Callable[[], dict[str, Any]]:
            return lambda: structural_row(meta, scenario_name, operation, run_index, note)

        handlers = {
            "branch.publish": paced_cli("branch.publish", ["oak", "push"]),
            "integration.open": structural(
                "integration.open",
                "integration object exists at publish on oak; no separate open call",
            ),
            "integration.visible.settle": structural(
                "integration.visible.settle",
                "oak push returns after the branch is visible on the server; nothing to poll",
            ),
            "integration.mergeable.settle": structural(
                "integration.mergeable.settle", "no async mergeability concept"
            ),
            "integration.checks.green.settle": structural(
                "integration.checks.green.settle",
                "no hosted status-check API on oak; check-instant is a GitHub-side "
                "harness construct with no equivalent operation here (ADR-0002)",
            ),
            "integration.merge.request": paced_cli(
                "integration.merge.request", ["oak", "merge"]
            ),
            "integration.merged.settle": structural(
                "integration.merged.settle",
                "oak merge is synchronous; merged state holds at return",
            ),
            "integration.main.settle": structural(
                "integration.main.settle",
                "oak merge returns after main is updated; nothing to poll",
            ),
        }
        return None, handlers, cleanup

    return run_race_operations(
        scenario, meta, run_index, "cli", race_branch_count(scenario),
        make_branch, budgeter, clock,
    )


def run_race_with_credentials(
    args: argparse.Namespace,
    scenario: dict[str, Any],
    meta: dict[str, Any],
    run_index: int,
    run_root: Path,
    settings: Settings,
) -> list[dict[str, Any]]:
    if args.platform == "github" and args.driver == "harness-api":
        return run_github_api_race(args, scenario, meta, run_index, settings)
    if args.platform == "github":
        # gh-cli race orchestration is deferred to Phase 4 (see
        # RACE_CLI_DEFER_REASON): per-op skip rows, a recorded gap, never
        # silently different semantics.
        scenario_name = str(scenario["name"])
        return [
            skip_row(meta, scenario_name, str(operation), run_index, RACE_CLI_DEFER_REASON)
            for operation in scenario.get("operations") or []
        ]
    return run_oak_cli_race(args, scenario, meta, run_index, run_root, settings)


# --------------------------------------------------------------------------
# branch_fleet_nN: end-to-end branch fleet triage and landing.
#
# N open branches are seeded; roughly 30% are constructed as conflicts and the
# rest are clean. The timed workflow is: classify -> plan -> merge clean
# branches -> leave conflicts open -> sync/observe final main -> oracle. This
# is the agent-scale workflow Git/GitHub makes agents perform with many round
# trips, while Oak should collapse classification into one checkout-free batch.
# --------------------------------------------------------------------------


def is_branch_fleet_scenario(scenario: dict[str, Any]) -> bool:
    return BRANCH_FLEET_SCENARIO_RE.match(str(scenario.get("name", ""))) is not None


def branch_fleet_count(scenario: dict[str, Any]) -> int:
    if scenario.get("branch_count") is not None:
        return int(scenario["branch_count"])
    match = BRANCH_FLEET_SCENARIO_RE.match(str(scenario.get("name", "")))
    return int(match.group(1)) if match else 10


def branch_fleet_live_skip_reason(
    scenario: dict[str, Any],
    driver: str,
    environ: Optional[dict[str, str]] = None,
) -> Optional[str]:
    env = os.environ if environ is None else environ
    if driver == BRANCH_TRIAGE_FAKE_DRIVER:
        return None
    if branch_fleet_count(scenario) <= BRANCH_FLEET_LIVE_LARGE_LIMIT:
        return None
    if str(env.get(BRANCH_FLEET_LIVE_LARGE_ENV) or "").strip() == "1":
        return None
    return (
        f"branch_fleet_large_live_requires_opt_in:"
        f"{BRANCH_FLEET_LIVE_LARGE_ENV}=1"
    )


def branch_fleet_worker_count(
    branch_count: int,
    *,
    env_name: str,
    default: int,
    environ: Optional[dict[str, str]] = None,
) -> int:
    env = os.environ if environ is None else environ
    raw = str(env.get(env_name) or "").strip()
    if raw:
        try:
            configured = int(raw)
        except ValueError:
            configured = default
    else:
        configured = default
    bounded = max(1, min(configured, BRANCH_FLEET_WORKERS_MAX))
    return max(1, min(bounded, max(1, branch_count)))


def branch_fleet_seed_worker_count(
    branch_count: int, environ: Optional[dict[str, str]] = None
) -> int:
    return branch_fleet_worker_count(
        branch_count,
        env_name=BRANCH_FLEET_SEED_WORKERS_ENV,
        default=BRANCH_FLEET_SEED_WORKERS_DEFAULT,
        environ=environ,
    )


def branch_fleet_cleanup_worker_count(
    branch_count: int, environ: Optional[dict[str, str]] = None
) -> int:
    return branch_fleet_worker_count(
        branch_count,
        env_name=BRANCH_FLEET_CLEANUP_WORKERS_ENV,
        default=BRANCH_FLEET_CLEANUP_WORKERS_DEFAULT,
        environ=environ,
    )


def branch_fleet_conflict_count(branch_count: int) -> int:
    # The launch/demo shape is "10 branches, 3 conflicts, 7 clean"; scale that
    # ratio for larger fleets while keeping at least one conflict when N > 1.
    if branch_count <= 1:
        return 0
    return max(1, round(branch_count * 0.3))


def branch_fleet_specs(
    *, bench_id: str, scenario_name: str, run_index: int, branch_count: int
) -> list[dict[str, Any]]:
    conflict_count = branch_fleet_conflict_count(branch_count)
    clean_count = branch_count - conflict_count
    prefix = disposable_branch(bench_id, scenario_name, "fleet", run_index)
    specs: list[dict[str, Any]] = []
    for index in range(branch_count):
        kind = "conflicting" if index >= clean_count else "clean"
        specs.append(
            {
                "index": index,
                "name": f"{prefix}-{kind[:4]}-{index:04d}",
                "kind": kind,
                "expected_action": "leave_conflict_open" if kind == "conflicting" else "merge",
                "path": (
                    f".oak-bench/fleet/conflict-{index:04d}.txt"
                    if kind == "conflicting"
                    else f".oak-bench/fleet/clean-{index:04d}.txt"
                ),
            }
        )
    return specs


def branch_fleet_plan(reviewed: list[dict[str, Any]]) -> dict[str, Any]:
    actions: list[dict[str, str]] = []
    merge_order: list[str] = []
    clean_detected = 0
    conflict_detected = 0
    unknown_detected = 0
    for branch in reviewed:
        name = str(branch["name"])
        mergeable = branch.get("mergeable")
        if mergeable is True:
            classification = "clean"
            action = "merge"
            clean_detected += 1
            merge_order.append(name)
        elif mergeable is False:
            classification = "conflicting"
            action = "leave_conflict_open"
            conflict_detected += 1
        else:
            classification = "unknown"
            action = "review"
            unknown_detected += 1
        actions.append({"branch": name, "classification": classification, "action": action})
    return {
        "actions": actions,
        "merge_order": merge_order,
        "detected_counts": {
            "clean": clean_detected,
            "conflicting": conflict_detected,
            "unknown": unknown_detected,
        },
    }


def branch_fleet_oracle(
    specs: list[dict[str, Any]],
    plan: dict[str, Any],
    apply_result: dict[str, Any],
) -> dict[str, Any]:
    expected_actions = [
        {
            "branch": str(spec["name"]),
            "classification": str(spec["kind"]),
            "action": str(spec["expected_action"]),
        }
        for spec in specs
    ]
    expected_merge_order = [
        str(spec["name"]) for spec in specs if spec["expected_action"] == "merge"
    ]
    expected_survivors = [
        str(spec["name"])
        for spec in specs
        if spec["expected_action"] == "leave_conflict_open"
    ]
    apply_succeeded = apply_result.get("apply_succeeded") is not False
    final_survivors = sorted(str(name) for name in apply_result.get("final_survivors") or [])
    return {
        "apply_succeeded": apply_succeeded,
        "action_plan_correct": plan.get("actions") == expected_actions,
        "merge_order_correct": plan.get("merge_order") == expected_merge_order,
        "final_survivors_correct": apply_succeeded and final_survivors == sorted(expected_survivors),
        "expected_action_plan": expected_actions,
        "expected_merge_order": expected_merge_order,
        "expected_final_survivors": sorted(expected_survivors),
        "final_survivors": final_survivors,
    }


def branch_fleet_survivors_from_sync_stdout(
    stdout: str,
    specs: list[dict[str, Any]],
) -> list[str]:
    payload = json.loads(stdout)
    if isinstance(payload, list):
        rows = payload
    elif isinstance(payload, dict):
        if "branches" in payload:
            rows = payload["branches"]
        elif "rows" in payload:
            rows = payload["rows"]
        else:
            raise ValueError("branch_list_missing_rows")
        if not isinstance(rows, list):
            raise ValueError("branch_list_rows_not_list")
    else:
        raise ValueError("branch_list_not_array_or_object")
    expected_names = {str(spec["name"]) for spec in specs}
    names = set()
    for item in rows:
        if not isinstance(item, dict):
            raise ValueError("branch_list_row_not_object")
        name = item.get("name") or item.get("branch")
        if not name:
            raise ValueError("branch_list_row_missing_name")
        names.add(str(name))
    return sorted(name for name in names if name in expected_names)


BRANCH_FLEET_WORKFLOW_OPERATIONS = (
    "fleet.classify",
    "fleet.plan",
    "fleet.apply",
    "fleet.sync",
    "fleet.oracle",
)


def branch_fleet_workflow_summary_fields(rows: list[dict[str, Any]]) -> dict[str, Any]:
    setup_rows = [row for row in rows if row.get("operation") == "fleet.seed"]
    workflow_rows = [
        row for row in rows
        if row.get("operation") in BRANCH_FLEET_WORKFLOW_OPERATIONS
    ]
    workflow_failures = [
        row for row in workflow_rows
        if row_returncode(row) not in (0, SKIP_RETURNCODE)
    ]
    workflow_skips = [
        row for row in workflow_rows
        if row_returncode(row) == SKIP_RETURNCODE
    ]
    workflow_returncode: Optional[int]
    if not workflow_rows:
        workflow_returncode = None
    elif workflow_failures:
        workflow_returncode = 1
    elif len(workflow_skips) == len(workflow_rows):
        workflow_returncode = SKIP_RETURNCODE
    else:
        workflow_returncode = 0
    tool_counts = [
        int(row["tool_call_count"])
        for row in workflow_rows
        if row.get("tool_call_count") is not None
    ]
    return {
        "branch_fleet_workflow_operations": [row["operation"] for row in workflow_rows],
        "branch_fleet_setup_operations": [row["operation"] for row in setup_rows],
        "branch_fleet_setup_excluded_from_workflow": True,
        "branch_fleet_workflow_elapsed_ms": round(
            sum(float(row.get("elapsed_ms") or 0.0) for row in workflow_rows),
            3,
        ),
        "branch_fleet_setup_elapsed_ms": round(
            sum(float(row.get("elapsed_ms") or 0.0) for row in setup_rows),
            3,
        ),
        "branch_fleet_workflow_failure_count": len(workflow_failures),
        "branch_fleet_workflow_skip_count": len(workflow_skips),
        "branch_fleet_workflow_returncode": workflow_returncode,
        "branch_fleet_workflow_tool_call_count": sum(tool_counts) if tool_counts else None,
    }


def branch_fleet_metric_fields(
    *,
    branch_count: int,
    reviewed: Optional[list[dict[str, Any]]] = None,
    plan: Optional[dict[str, Any]] = None,
    apply_result: Optional[dict[str, Any]] = None,
    oracle: Optional[dict[str, Any]] = None,
) -> dict[str, Any]:
    reviewed = reviewed or []
    counts = ((plan or {}).get("detected_counts") or {})
    if not counts and reviewed:
        counts = {"clean": 0, "conflicting": 0, "unknown": 0}
        for branch in reviewed:
            if branch.get("mergeable") is True:
                counts["clean"] += 1
            elif branch.get("mergeable") is False:
                counts["conflicting"] += 1
            else:
                counts["unknown"] += 1
    merged = list((apply_result or {}).get("merged_branches") or [])
    survivors = list((apply_result or {}).get("final_survivors") or [])
    return {
        "branch_count": branch_count,
        "branches_seeded": branch_count,
        "branches_reviewed": len(reviewed),
        "branches_merged": len(merged),
        "branches_left_open": len(survivors),
        "branches_closed": len(merged),
        "clean_detected_count": int(counts.get("clean", 0)),
        "conflict_detected_count": int(counts.get("conflicting", 0)),
        "unknown_detected_count": int(counts.get("unknown", 0)),
        "merge_order_correct": (oracle or {}).get("merge_order_correct"),
    }


def branch_fleet_row(
    meta: dict[str, Any],
    scenario_name: str,
    operation: str,
    run_index: int,
    elapsed_ms: float,
    returncode: int = 0,
    *,
    provider: str,
    command: Optional[list[Any]] = None,
    **fields: Any,
) -> dict[str, Any]:
    row = {
        **meta,
        "scenario": scenario_name,
        "run": run_index,
        "operation": operation,
        "elapsed_ms": round(elapsed_ms, 3),
        "returncode": returncode,
        "command": command or [],
        "branch_fleet_provider": provider,
        **fields,
    }
    if provider in (BRANCH_FLEET_PROVIDER_GITHUB_API, BRANCH_FLEET_PROVIDER_FAKE):
        row.update(null_token_fields() if provider == BRANCH_FLEET_PROVIDER_GITHUB_API else fake_provider_token_fields())
        row.setdefault("tool_call_count", None if provider == BRANCH_FLEET_PROVIDER_GITHUB_API else 0)
    return row


def run_branch_fleet_fake_scenario(
    scenario: dict[str, Any],
    meta: dict[str, Any],
    run_index: int,
) -> list[dict[str, Any]]:
    scenario_name = str(scenario["name"])
    operations = [str(op) for op in scenario.get("operations") or []]
    count = branch_fleet_count(scenario)
    specs = branch_fleet_specs(
        bench_id=str(meta.get("bench_id") or "bench"),
        scenario_name=scenario_name,
        run_index=run_index,
        branch_count=count,
    )
    rows: list[dict[str, Any]] = []
    reviewed: list[dict[str, Any]] = []
    plan: dict[str, Any] = {}
    apply_result: dict[str, Any] = {
        "apply_succeeded": False,
        "merged_branches": [],
        "final_survivors": [],
    }
    oracle: dict[str, Any] = {}
    total_start = platform_clock.monotonic_ms()

    for operation in operations:
        start = platform_clock.monotonic_ms()
        if operation == "fleet.seed":
            rows.append(
                branch_fleet_row(
                    meta, scenario_name, operation, run_index,
                    platform_clock.monotonic_ms() - start,
                    provider=BRANCH_FLEET_PROVIDER_FAKE,
                    seeded_branch_classes=[spec["kind"] for spec in specs],
                    **branch_fleet_metric_fields(branch_count=count),
                )
            )
        elif operation == "fleet.classify":
            reviewed = [
                {"name": spec["name"], "mergeable": spec["kind"] == "clean"}
                for spec in specs
            ]
            rows.append(
                branch_fleet_row(
                    meta, scenario_name, operation, run_index,
                    platform_clock.monotonic_ms() - start,
                    provider=BRANCH_FLEET_PROVIDER_FAKE,
                    reviewed_branches=[row["name"] for row in reviewed],
                    **branch_fleet_metric_fields(branch_count=count, reviewed=reviewed),
                )
            )
        elif operation == "fleet.plan":
            plan = branch_fleet_plan(reviewed)
            rows.append(
                branch_fleet_row(
                    meta, scenario_name, operation, run_index,
                    platform_clock.monotonic_ms() - start,
                    provider=BRANCH_FLEET_PROVIDER_FAKE,
                    action_plan=plan["actions"],
                    merge_order=plan["merge_order"],
                    **branch_fleet_metric_fields(branch_count=count, reviewed=reviewed, plan=plan),
                )
            )
        elif operation == "fleet.apply":
            apply_result = {
                "apply_succeeded": True,
                "merged_branches": list(plan.get("merge_order") or []),
                "final_survivors": [
                    action["branch"]
                    for action in plan.get("actions") or []
                    if action["action"] == "leave_conflict_open"
                ],
            }
            rows.append(
                branch_fleet_row(
                    meta, scenario_name, operation, run_index,
                    platform_clock.monotonic_ms() - start,
                    provider=BRANCH_FLEET_PROVIDER_FAKE,
                    **apply_result,
                    **branch_fleet_metric_fields(
                        branch_count=count, reviewed=reviewed, plan=plan, apply_result=apply_result
                    ),
                )
            )
        elif operation == "fleet.sync":
            rows.append(
                branch_fleet_row(
                    meta, scenario_name, operation, run_index,
                    platform_clock.monotonic_ms() - start,
                    provider=BRANCH_FLEET_PROVIDER_FAKE,
                    note="fake provider has no local checkout; final state already materialized in memory",
                    **branch_fleet_metric_fields(
                        branch_count=count, reviewed=reviewed, plan=plan, apply_result=apply_result
                    ),
                )
            )
        elif operation == "fleet.oracle":
            oracle = branch_fleet_oracle(specs, plan, apply_result)
            ok = all(
                bool(oracle[key])
                for key in (
                    "apply_succeeded",
                    "action_plan_correct",
                    "merge_order_correct",
                    "final_survivors_correct",
                )
            )
            fields = {
                **oracle,
                **branch_fleet_metric_fields(
                    branch_count=count,
                    reviewed=reviewed,
                    plan=plan,
                    apply_result=apply_result,
                    oracle=oracle,
                ),
            }
            rows.append(
                branch_fleet_row(
                    meta, scenario_name, operation, run_index,
                    platform_clock.monotonic_ms() - start,
                    returncode=0 if ok else 1,
                    provider=BRANCH_FLEET_PROVIDER_FAKE,
                    **fields,
                )
            )
        elif operation == "fleet.total":
            failures = [
                row for row in rows
                if row_returncode(row) not in (0, SKIP_RETURNCODE)
            ]
            rows.append(
                branch_fleet_row(
                    meta, scenario_name, operation, run_index,
                    platform_clock.monotonic_ms() - total_start,
                    returncode=1 if failures else 0,
                    provider=BRANCH_FLEET_PROVIDER_FAKE,
                    summarized_operations=[row["operation"] for row in rows],
                    **branch_fleet_workflow_summary_fields(rows),
                    **branch_fleet_metric_fields(
                        branch_count=count,
                        reviewed=reviewed,
                        plan=plan,
                        apply_result=apply_result,
                        oracle=oracle,
                    ),
                )
            )
        else:
            rows.append(skip_row(meta, scenario_name, operation, run_index, "operation_not_implemented"))
    return rows


def run_github_api_branch_fleet_scenario(
    args: argparse.Namespace,
    scenario: dict[str, Any],
    meta: dict[str, Any],
    run_index: int,
    settings: Settings,
    *,
    client: Optional[GitHubClient] = None,
    clock: Callable[[], float] = time.monotonic,
    sleep: Callable[[float], None] = time.sleep,
) -> list[dict[str, Any]]:
    scenario_name = str(scenario["name"])
    operations = [str(op) for op in scenario.get("operations") or []]
    count = branch_fleet_count(scenario)
    specs = branch_fleet_specs(
        bench_id=str(meta.get("bench_id") or "bench"),
        scenario_name=scenario_name,
        run_index=run_index,
        branch_count=count,
    )
    repo = args.repo
    if client is None:
        client = GitHubClient(token=os.environ.get(args.token_env))

    state: dict[str, Any] = {"prs": {}, "heads": {}}
    reviewed: list[dict[str, Any]] = []
    plan: dict[str, Any] = {}
    apply_result: dict[str, Any] = {
        "apply_succeeded": False,
        "merged_branches": [],
        "final_survivors": [],
    }
    oracle: dict[str, Any] = {}
    total_start = platform_clock.monotonic_ms()

    def seed() -> dict[str, Any]:
        def do() -> None:
            base_sha = client.get_main_sha(repo)
            state["base_sha"] = base_sha
            for spec in specs:
                branch = str(spec["name"])
                path = str(spec["path"])
                client.create_ref(repo, f"refs/heads/{branch}", base_sha)
                put = client.put_file(
                    repo,
                    path=path,
                    message=f"bench branch fleet {meta['bench_id']} {branch}",
                    content=(
                        f"branch fleet {meta['bench_id']} run={run_index} "
                        f"branch={branch} kind={spec['kind']}\n"
                    ).encode("utf-8"),
                    branch=branch,
                )
                commit = put.get("commit") if isinstance(put, dict) else None
                if isinstance(commit, dict) and commit.get("sha"):
                    state["heads"][branch] = str(commit["sha"])
                pr = client.create_pr(
                    repo,
                    title=f"bench branch fleet {scenario_name} {branch}",
                    head=branch,
                    base="main",
                    body="Disposable branch-fleet benchmark PR.",
                )
                state["prs"][branch] = int(pr["number"])
            # Make conflict branches genuinely conflict with main: each conflict
            # branch added the same path from the old base; now main adds that
            # path with different content, producing add/add conflicts.
            for spec in specs:
                if spec["kind"] != "conflicting":
                    continue
                client.put_file(
                    repo,
                    path=str(spec["path"]),
                    message=f"bench branch fleet conflicting main side {meta['bench_id']}",
                    content=(
                        f"main-side conflict {meta['bench_id']} run={run_index} "
                        f"branch={spec['name']}\n"
                    ).encode("utf-8"),
                    branch="main",
                )

        row = api_step_row(
            meta,
            scenario_name,
            "fleet.seed",
            run_index,
            client,
            do,
            [
                f"api:GET /repos/{repo}/git/ref/heads/main",
                f"api:POST /repos/{repo}/git/refs x{count}",
                f"api:PUT /repos/{repo}/contents/... x{count + branch_fleet_conflict_count(count)}",
                f"api:POST /repos/{repo}/pulls x{count}",
            ],
        )
        row["branch_fleet_provider"] = BRANCH_FLEET_PROVIDER_GITHUB_API
        row.update(branch_fleet_metric_fields(branch_count=count))
        row["seeded_branch_classes"] = [spec["kind"] for spec in specs]
        return row

    def classify() -> dict[str, Any]:
        nonlocal reviewed
        before = metrics_snapshot(client)
        start = platform_clock.monotonic_ms()
        polls = 0
        settled = False
        last_reviewed: list[dict[str, Any]] = []
        error_text: Optional[str] = None
        deadline = clock() + settings.settle_timeout_s
        while True:
            polls += 1
            try:
                current: list[dict[str, Any]] = []
                all_known = True
                for spec in specs:
                    branch = str(spec["name"])
                    pr = client.get_pr(repo, int(state["prs"][branch]))
                    mergeable = pr.get("mergeable")
                    if mergeable is None:
                        all_known = False
                    current.append(
                        {
                            "name": branch,
                            "mergeable": mergeable,
                            "pr_number": int(state["prs"][branch]),
                            "expected_kind": spec["kind"],
                        }
                    )
                last_reviewed = current
                if all_known:
                    settled = True
                    break
            except (GitHubApiError, KeyError, json.JSONDecodeError) as exc:
                error_text = str(exc)[:4000]
                break
            if clock() >= deadline:
                break
            sleep(settings.interval_s)
        reviewed = last_reviewed
        elapsed_ms = platform_clock.monotonic_ms() - start
        row = branch_fleet_row(
            meta,
            scenario_name,
            "fleet.classify",
            run_index,
            elapsed_ms,
            returncode=0 if settled and error_text is None else 1,
            provider=BRANCH_FLEET_PROVIDER_GITHUB_API,
            command=[f"api:GET /repos/{repo}/pulls/<n> mergeable x{count} (poll)"],
            reviewed_branches=[branch["name"] for branch in reviewed],
            settled=settled,
            settle_elapsed_ms=round(elapsed_ms, 3) if settled else None,
            observed_wait_ms=round(elapsed_ms, 3),
            poll_iterations=polls,
            poll_interval_s=settings.interval_s,
            poll_quantization_ms=settings.interval_s * 1000.0,
            api_metrics=metrics_delta(before, metrics_snapshot(client)),
            **branch_fleet_metric_fields(branch_count=count, reviewed=reviewed),
        )
        if error_text is not None:
            row["stderr"] = error_text
        return row

    def plan_row() -> dict[str, Any]:
        nonlocal plan
        start = platform_clock.monotonic_ms()
        plan = branch_fleet_plan(reviewed)
        return branch_fleet_row(
            meta,
            scenario_name,
            "fleet.plan",
            run_index,
            platform_clock.monotonic_ms() - start,
            provider=BRANCH_FLEET_PROVIDER_GITHUB_API,
            action_plan=plan["actions"],
            merge_order=plan["merge_order"],
            **branch_fleet_metric_fields(branch_count=count, reviewed=reviewed, plan=plan),
        )

    def apply() -> dict[str, Any]:
        nonlocal apply_result

        def do() -> None:
            merged: list[str] = []
            survivors: list[str] = []
            for action in plan.get("actions") or []:
                branch = str(action["branch"])
                if action["action"] == "merge":
                    client.merge_pr(repo, int(state["prs"][branch]))
                    merged.append(branch)
                elif action["action"] == "leave_conflict_open":
                    survivors.append(branch)
            apply_result = {
                "apply_succeeded": True,
                "merged_branches": merged,
                "final_survivors": survivors,
            }
            state["apply_result"] = apply_result

        row = api_step_row(
            meta,
            scenario_name,
            "fleet.apply",
            run_index,
            client,
            do,
            [f"api:PUT /repos/{repo}/pulls/<n>/merge x{len(plan.get('merge_order') or [])}"],
        )
        # api_step_row cannot update this local binding through nested scope on
        # older Python without nonlocal assignment inside do(); read the state
        # value populated above.
        apply_result = dict(
            state.get("apply_result")
            or {"apply_succeeded": False, "merged_branches": [], "final_survivors": []}
        )
        row["branch_fleet_provider"] = BRANCH_FLEET_PROVIDER_GITHUB_API
        row.update(apply_result)
        row.update(
            branch_fleet_metric_fields(
                branch_count=count, reviewed=reviewed, plan=plan, apply_result=apply_result
            )
        )
        return row

    def sync() -> dict[str, Any]:
        row = api_step_row(
            meta,
            scenario_name,
            "fleet.sync",
            run_index,
            client,
            lambda: client.get_main_sha(repo),
            [f"api:GET /repos/{repo}/git/ref/heads/main"],
        )
        row["branch_fleet_provider"] = BRANCH_FLEET_PROVIDER_GITHUB_API
        row["note"] = (
            "harness-api driver has no local checkout; sync observes the final main ref."
        )
        row.update(
            branch_fleet_metric_fields(
                branch_count=count, reviewed=reviewed, plan=plan, apply_result=apply_result
            )
        )
        return row

    def oracle_row() -> dict[str, Any]:
        nonlocal oracle
        start = platform_clock.monotonic_ms()
        oracle = branch_fleet_oracle(specs, plan, apply_result)
        ok = all(
            bool(oracle[key])
            for key in (
                "apply_succeeded",
                "action_plan_correct",
                "merge_order_correct",
                "final_survivors_correct",
            )
        )
        fields = {
            **oracle,
            **branch_fleet_metric_fields(
                branch_count=count,
                reviewed=reviewed,
                plan=plan,
                apply_result=apply_result,
                oracle=oracle,
            ),
        }
        return branch_fleet_row(
            meta,
            scenario_name,
            "fleet.oracle",
            run_index,
            platform_clock.monotonic_ms() - start,
            returncode=0 if ok else 1,
            provider=BRANCH_FLEET_PROVIDER_GITHUB_API,
            **fields,
        )

    def total() -> dict[str, Any]:
        failures = [
            row for row in rows
            if row_returncode(row) not in (0, SKIP_RETURNCODE)
        ]
        row = branch_fleet_row(
            meta,
            scenario_name,
            "fleet.total",
            run_index,
            platform_clock.monotonic_ms() - total_start,
            returncode=1 if failures else 0,
            provider=BRANCH_FLEET_PROVIDER_GITHUB_API,
            summarized_operations=[row["operation"] for row in rows],
            **branch_fleet_workflow_summary_fields(rows),
            **branch_fleet_metric_fields(
                branch_count=count,
                reviewed=reviewed,
                plan=plan,
                apply_result=apply_result,
                oracle=oracle,
            ),
        )
        deltas = [
            r["api_metrics"] for r in rows if isinstance(r.get("api_metrics"), dict)
        ]
        row["api_metrics"] = {
            "request_count": sum(delta["request_count"] for delta in deltas),
            "request_bytes": sum(delta["request_bytes"] for delta in deltas),
            "response_bytes": sum(delta["response_bytes"] for delta in deltas),
            "rate_limit_remaining": deltas[-1]["rate_limit_remaining"] if deltas else None,
        }
        return row

    handlers: dict[str, Callable[[], dict[str, Any]]] = {
        "fleet.seed": seed,
        "fleet.classify": classify,
        "fleet.plan": plan_row,
        "fleet.apply": apply,
        "fleet.sync": sync,
        "fleet.oracle": oracle_row,
        "fleet.total": total,
    }

    rows: list[dict[str, Any]] = []
    aborted_at: Optional[str] = None
    for operation in operations:
        if aborted_at is not None:
            rows.append(skip_row(meta, scenario_name, operation, run_index, f"prior_step_failed:{aborted_at}"))
            continue
        handler = handlers.get(operation)
        if handler is None:
            rows.append(skip_row(meta, scenario_name, operation, run_index, "operation_not_implemented"))
            continue
        row = handler()
        rows.append(row)
        if row_returncode(row) not in (0, SKIP_RETURNCODE):
            aborted_at = operation

    for spec in specs:
        try:
            client.delete_ref(repo, f"heads/{spec['name']}")
        except GitHubApiError:
            pass
    return rows


def oak_branch_fleet_mergeable(row: dict[str, Any]) -> Optional[bool]:
    if row.get("merge_allowed") is True or row.get("recommended_action") in {
        "merge",
        "validate_then_merge",
    }:
        return True
    if row.get("vcs_merge_safe") is False or row.get("recommended_action") in {
        "resolve",
        "keep_open_manual_resolution",
    }:
        return False
    mergeability = str(row.get("mergeability") or "").lower()
    if mergeability in {"clean", "mergeable"}:
        return True
    if mergeability in {"conflict", "conflicts", "conflicting"}:
        return False
    return None


def oak_branch_fleet_cleanup_commands(
    specs: list[dict[str, Any]], advancers: list[str]
) -> list[list[str]]:
    branches = [str(spec["name"]) for spec in specs] + list(advancers)
    return oak_branch_fleet_cleanup_commands_for_branches(branches)


def oak_branch_fleet_cleanup_commands_for_branches(branches: list[str]) -> list[list[str]]:
    return [
        ["oak", "close", "--remote", "--reason", "stale", "--json", branch]
        for branch in branches
    ]


def oak_branch_fleet_apply_commands(merge_order: list[str]) -> list[list[str]]:
    return [["oak", "merge", branch] for branch in merge_order]


def oak_branch_fleet_cleanup_fields(failures: list[dict[str, Any]]) -> dict[str, Any]:
    fields: dict[str, Any] = {"cleanup_failure_count": len(failures)}
    if failures:
        fields["cleanup_failures"] = failures
    return fields


@dataclass(frozen=True)
class OakBranchFleetSeedStep:
    branch: str
    phase: str
    phase_index: int
    phase_total: int
    commands: list[list[str]]
    cleanup_after_command_index: int


def oak_branch_fleet_write_command(rel_path: Path, content: str) -> list[str]:
    return [
        "/bin/zsh",
        "-c",
        "mkdir -p "
        + shlex.quote(str(rel_path.parent))
        + " && printf %s "
        + shlex.quote(content)
        + " > "
        + shlex.quote(str(rel_path)),
    ]


def oak_branch_fleet_seed_steps(
    specs: list[dict[str, Any]],
    meta: dict[str, Any],
    run_index: int,
) -> tuple[list[OakBranchFleetSeedStep], list[str]]:
    steps: list[OakBranchFleetSeedStep] = []
    branch_total = len(specs)
    for index, spec in enumerate(specs, start=1):
        branch = str(spec["name"])
        rel_path = Path(str(spec["path"]))
        content = (
            f"branch fleet {meta['bench_id']} run={run_index} "
            f"branch={branch} kind={spec['kind']}\n"
        )
        steps.append(
            OakBranchFleetSeedStep(
                branch=branch,
                phase="branch",
                phase_index=index,
                phase_total=branch_total,
                commands=[
                    ["oak", "switch", "--clean", "-c", branch],
                    oak_branch_fleet_write_command(rel_path, content),
                    ["oak", "commit", "--no-verify", "--push"],
                ],
                cleanup_after_command_index=2,
            )
        )

    conflict_specs = [spec for spec in specs if spec["kind"] == "conflicting"]
    advancers: list[str] = []
    conflict_total = len(conflict_specs)
    for index, spec in enumerate(conflict_specs, start=1):
        advancer = f"{spec['name']}-main-side"
        advancers.append(advancer)
        rel_path = Path(str(spec["path"]))
        content = (
            f"main-side conflict {meta['bench_id']} run={run_index} "
            f"branch={spec['name']}\n"
        )
        steps.append(
            OakBranchFleetSeedStep(
                branch=advancer,
                phase="conflict",
                phase_index=index,
                phase_total=conflict_total,
                commands=[
                    ["oak", "switch", "--clean", "-c", advancer],
                    oak_branch_fleet_write_command(rel_path, content),
                    ["oak", "commit", "--no-verify", "--push"],
                    ["oak", "merge"],
                ],
                cleanup_after_command_index=2,
            )
        )
    return steps, advancers


def oak_branch_fleet_seed_diagnostics(
    *,
    branch_count: int,
    conflict_count: int,
    command_count: int,
    completed_count: int,
    completed_branch_count: int,
    completed_conflict_count: int,
    completed_command_count: int,
    last_completed_branch: Optional[str],
    failed_command_index: Optional[int],
    failed_branch: Optional[str],
) -> dict[str, Any]:
    return {
        "seed_branch_count": branch_count,
        "seed_conflict_count": conflict_count,
        "seed_command_count": command_count,
        "seed_completed_count": completed_count,
        "seed_completed_branch_count": completed_branch_count,
        "seed_completed_conflict_count": completed_conflict_count,
        "seed_completed_command_count": completed_command_count,
        "last_completed_branch": last_completed_branch,
        "failed_command_index": failed_command_index,
        "failed_branch": failed_branch,
    }


def oak_branch_fleet_seed_text(value: Any) -> str:
    if value is None:
        return ""
    if isinstance(value, bytes):
        return value.decode("utf-8", errors="replace")
    return str(value)


def oak_branch_fleet_seed_exception_result(
    command: list[str], exc: BaseException
) -> subprocess.CompletedProcess[str]:
    if isinstance(exc, subprocess.TimeoutExpired):
        stdout = oak_branch_fleet_seed_text(exc.stdout)
        stderr = oak_branch_fleet_seed_text(exc.stderr)
        timeout = "" if exc.timeout is None else f":{exc.timeout}s"
        message = f"seed_command_timeout{timeout}"
        return subprocess.CompletedProcess(
            command,
            BRANCH_FLEET_SEED_TIMEOUT_RETCODE,
            stdout,
            "\n".join(part for part in (message, stderr) if part),
        )
    return subprocess.CompletedProcess(
        command,
        1,
        "",
        f"seed_command_exception:{type(exc).__name__}:{exc}",
    )


def oak_branch_fleet_seed_transient_failure(proc: subprocess.CompletedProcess[str]) -> bool:
    text = f"{proc.stderr}\n{proc.stdout}".lower()
    return any(
        marker in text
        for marker in (
            "http 502",
            "502 bad gateway",
            "http 503",
            "503 service unavailable",
            "http 504",
            "504 gateway timeout",
        )
    )


def oak_branch_fleet_seed_progress(step: OakBranchFleetSeedStep, done: bool) -> str:
    if step.phase == "conflict":
        verb = "advanced conflicts" if done else "advancing conflict"
        return f"[branch-fleet seed] {verb} {step.phase_index}/{step.phase_total} branch={step.branch}"
    verb = "seeded" if done else "seeding"
    return f"[branch-fleet seed] {verb} {step.phase_index}/{step.phase_total} branch={step.branch}"


def oak_branch_fleet_seed_row(
    meta: dict[str, Any],
    scenario_name: str,
    seed_steps: list[OakBranchFleetSeedStep],
    cwd: Path,
    run_index: int,
    env: dict[str, str],
    admitted: int,
    branch_count: int,
    conflict_count: int,
    *,
    progress: Optional[Callable[[str], None]] = None,
    cleanup_branches: Optional[list[str]] = None,
    runner: Optional[
        Callable[[list[str], Optional[Path], dict[str, str]], subprocess.CompletedProcess[str]]
    ] = None,
    retry_limit: int = BRANCH_FLEET_SEED_TRANSIENT_RETRIES,
    retry_delay_s: float = BRANCH_FLEET_SEED_RETRY_DELAY_S,
    retry_sleep: Callable[[float], None] = time.sleep,
    command_index_offset: int = 0,
) -> dict[str, Any]:
    if runner is None:
        runner = run_untimed
    emit = progress or (lambda message: print(message, file=sys.stderr, flush=True))
    planned_commands = [command for step in seed_steps for command in step.commands]
    command_count = len(planned_commands)
    start = platform_clock.monotonic_ms()
    stdout_parts: list[str] = []
    stderr_parts: list[str] = []
    attempted_commands: list[list[str]] = []
    completed_count = 0
    last_completed_branch: Optional[str] = None
    failed_command_index: Optional[int] = None
    failed_branch: Optional[str] = None
    failed_command: Optional[list[str]] = None
    failed_proc: Optional[subprocess.CompletedProcess[str]] = None
    completed_branch_count = 0
    completed_conflict_count = 0
    retry_count = 0
    retry_records: list[dict[str, Any]] = []
    planned_command_index = 0

    for step in seed_steps:
        emit(oak_branch_fleet_seed_progress(step, done=False))
        step_failed = False
        for step_command_index, command in enumerate(step.commands):
            command_index = command_index_offset + planned_command_index
            attempt = 0
            while True:
                try:
                    proc = runner(command, cwd, env)
                except Exception as exc:
                    proc = oak_branch_fleet_seed_exception_result(command, exc)
                attempted_commands.append(command)
                stdout_parts.append(proc.stdout)
                stderr_parts.append(proc.stderr)
                if proc.returncode == 0:
                    break
                if attempt < retry_limit and oak_branch_fleet_seed_transient_failure(proc):
                    attempt += 1
                    retry_count += 1
                    retry_records.append(
                        {
                            "branch": step.branch,
                            "command_index": command_index,
                            "attempt": attempt,
                            "returncode": proc.returncode,
                            "stderr": proc.stderr[-1000:],
                        }
                    )
                    emit(
                        "[branch-fleet seed] retrying transient failure "
                        f"{attempt}/{retry_limit} branch={step.branch} "
                        f"command_index={command_index}"
                    )
                    retry_sleep(retry_delay_s)
                    continue
                failed_command_index = command_index
                failed_branch = step.branch
                failed_command = command
                failed_proc = proc
                step_failed = True
                break
            planned_command_index += 1
            if step_failed:
                break
            if (
                cleanup_branches is not None
                and step_command_index == step.cleanup_after_command_index
                and step.branch not in cleanup_branches
            ):
                cleanup_branches.append(step.branch)
        if step_failed:
            break
        completed_count += 1
        if step.phase == "conflict":
            completed_conflict_count += 1
        else:
            completed_branch_count += 1
        last_completed_branch = step.branch
        emit(oak_branch_fleet_seed_progress(step, done=True))

    elapsed_ms = platform_clock.monotonic_ms() - start
    stdout = "\n".join(part for part in stdout_parts if part)
    stderr = "\n".join(part for part in stderr_parts if part)
    command_text = "\n".join(command_display(command) for command in planned_commands)
    row = {
        **meta,
        "scenario": scenario_name,
        "run": run_index,
        "operation": "fleet.seed",
        "elapsed_ms": round(elapsed_ms, 3),
        "returncode": 0 if failed_proc is None else int(failed_proc.returncode),
        "command": planned_commands,
        "tool_call_count": len(attempted_commands),
        "terminal_tool_call_count": len(attempted_commands),
        "branch_fleet_provider": BRANCH_FLEET_PROVIDER_OAK_CLI,
        "seed_retry_count": retry_count,
        **oakbench_tokens.interaction_token_fields(
            command_text,
            stdout[:admitted],
            stderr[:admitted],
            len(stdout.encode("utf-8")),
            len(stderr.encode("utf-8")),
            len(stdout) > admitted,
            len(stderr) > admitted,
        ),
        "_stdout": stdout,
        **oak_branch_fleet_seed_diagnostics(
            branch_count=branch_count,
            conflict_count=conflict_count,
            command_count=command_count,
            completed_count=completed_count,
            completed_branch_count=completed_branch_count,
            completed_conflict_count=completed_conflict_count,
            completed_command_count=len(attempted_commands),
            last_completed_branch=last_completed_branch,
            failed_command_index=failed_command_index,
            failed_branch=failed_branch,
        ),
    }
    if retry_records:
        row["seed_retries"] = retry_records
    if failed_proc is not None:
        row["failure_reason"] = "seed_command_failed"
        row["failed_command"] = failed_command
        row["failed_command_stdout"] = failed_proc.stdout[-1000:]
        row["failed_command_stderr"] = failed_proc.stderr[-1000:]
        row["stderr"] = stderr[-4000:] or failed_proc.stdout[-4000:]
    return row


def oak_branch_fleet_seed_step_chunks(
    seed_steps: list[OakBranchFleetSeedStep],
    worker_count: int,
) -> list[tuple[int, int, list[OakBranchFleetSeedStep]]]:
    indexed_branch_steps = [
        (index, step) for index, step in enumerate(seed_steps) if step.phase == "branch"
    ]
    if not indexed_branch_steps:
        return []
    bounded_workers = max(1, min(worker_count, len(indexed_branch_steps)))
    base_size, remainder = divmod(len(indexed_branch_steps), bounded_workers)
    chunks: list[tuple[int, int, list[OakBranchFleetSeedStep]]] = []
    start = 0
    for worker_id in range(bounded_workers):
        size = base_size + (1 if worker_id < remainder else 0)
        pairs = indexed_branch_steps[start:start + size]
        if not pairs:
            continue
        first_step_index = pairs[0][0]
        command_offset = sum(len(step.commands) for step in seed_steps[:first_step_index])
        chunks.append((worker_id, command_offset, [step for _index, step in pairs]))
        start += size
    return chunks


def oak_branch_fleet_seed_clone_failure_row(
    meta: dict[str, Any],
    scenario_name: str,
    run_index: int,
    *,
    worker_id: int,
    command: list[str],
    proc: subprocess.CompletedProcess[str],
    elapsed_ms: float,
    admitted: int,
    command_index_offset: int,
    failed_branch: Optional[str],
    branch_count: int,
    command_count: int,
) -> dict[str, Any]:
    stdout = proc.stdout or ""
    stderr = proc.stderr or ""
    return {
        **meta,
        "scenario": scenario_name,
        "run": run_index,
        "operation": "fleet.seed",
        "elapsed_ms": round(elapsed_ms, 3),
        "returncode": int(proc.returncode),
        "command": [command],
        "tool_call_count": 1,
        "terminal_tool_call_count": 1,
        "branch_fleet_provider": BRANCH_FLEET_PROVIDER_OAK_CLI,
        "seed_worker_id": worker_id,
        "seed_retry_count": 0,
        "failure_reason": "seed_worker_clone_failed",
        "failed_command": command,
        "failed_command_stdout": stdout[-1000:],
        "failed_command_stderr": stderr[-1000:],
        "stderr": stderr[-4000:] or stdout[-4000:],
        **oakbench_tokens.interaction_token_fields(
            command_display(command),
            stdout[:admitted],
            stderr[:admitted],
            len(stdout.encode("utf-8")),
            len(stderr.encode("utf-8")),
            len(stdout) > admitted,
            len(stderr) > admitted,
        ),
        **oak_branch_fleet_seed_diagnostics(
            branch_count=branch_count,
            conflict_count=0,
            command_count=command_count,
            completed_count=0,
            completed_branch_count=0,
            completed_conflict_count=0,
            completed_command_count=0,
            last_completed_branch=None,
            failed_command_index=command_index_offset,
            failed_branch=failed_branch,
        ),
    }


def oak_branch_fleet_aggregate_seed_rows(
    meta: dict[str, Any],
    scenario_name: str,
    run_index: int,
    *,
    seed_rows: list[dict[str, Any]],
    elapsed_ms: float,
    planned_commands: list[list[str]],
    branch_count: int,
    conflict_count: int,
    worker_count: int,
    worker_clone_count: int,
) -> dict[str, Any]:
    failed_rows = [
        row for row in seed_rows
        if row_returncode(row) not in (0, SKIP_RETURNCODE)
    ]
    worker_rows = [
        row for row in seed_rows
        if isinstance(row.get("seed_worker_id"), int)
    ]
    failed_worker_rows = [
        row for row in worker_rows
        if row_returncode(row) not in (0, SKIP_RETURNCODE)
    ]
    failed_rows.sort(
        key=lambda row: (
            row.get("failed_command_index")
            if row.get("failed_command_index") is not None
            else 1_000_000_000
        )
    )
    failed = failed_rows[0] if failed_rows else None
    retry_records: list[dict[str, Any]] = []
    for row in seed_rows:
        retry_records.extend(row.get("seed_retries") or [])
    last_completed_branch: Optional[str] = None
    for row in seed_rows:
        if row.get("last_completed_branch") is not None:
            last_completed_branch = str(row["last_completed_branch"])
    command_count = len(planned_commands)
    row = {
        **meta,
        "scenario": scenario_name,
        "run": run_index,
        "operation": "fleet.seed",
        "elapsed_ms": round(elapsed_ms, 3),
        "returncode": 0 if failed is None else row_returncode(failed),
        "command": planned_commands,
        "tool_call_count": sum(int(child.get("tool_call_count") or 0) for child in seed_rows),
        "terminal_tool_call_count": sum(
            int(child.get("terminal_tool_call_count") or 0) for child in seed_rows
        ),
        "branch_fleet_provider": BRANCH_FLEET_PROVIDER_OAK_CLI,
        "seed_parallel": worker_count > 1,
        "seed_worker_count": worker_count,
        "seed_worker_clone_count": worker_clone_count,
        "seed_worker_success_count": len(worker_rows) - len(failed_worker_rows),
        "seed_conflict_serial": True,
        "seed_retry_count": sum(int(child.get("seed_retry_count") or 0) for child in seed_rows),
        **oakbench_tokens.summed_token_fields(
            seed_rows,
            "sum_of_branch_fleet_seed_worker_rows_command_plus_admitted_output_chars_div_4",
        ),
        **oak_branch_fleet_seed_diagnostics(
            branch_count=branch_count,
            conflict_count=conflict_count,
            command_count=command_count,
            completed_count=sum(int(child.get("seed_completed_count") or 0) for child in seed_rows),
            completed_branch_count=sum(
                int(child.get("seed_completed_branch_count") or 0) for child in seed_rows
            ),
            completed_conflict_count=sum(
                int(child.get("seed_completed_conflict_count") or 0) for child in seed_rows
            ),
            completed_command_count=sum(
                int(child.get("seed_completed_command_count") or 0) for child in seed_rows
            ),
            last_completed_branch=last_completed_branch,
            failed_command_index=None if failed is None else failed.get("failed_command_index"),
            failed_branch=None if failed is None else failed.get("failed_branch"),
        ),
        "_stdout": "\n".join(str(child.get("_stdout") or "") for child in seed_rows),
    }
    if retry_records:
        row["seed_retries"] = retry_records
    if failed is not None:
        row["failure_reason"] = failed.get("failure_reason") or "seed_command_failed"
        row["failed_command"] = failed.get("failed_command")
        row["failed_command_stdout"] = failed.get("failed_command_stdout", "")
        row["failed_command_stderr"] = failed.get("failed_command_stderr", "")
        row["stderr"] = failed.get("stderr") or row.get("failed_command_stderr", "")
    return row


def oak_branch_fleet_run_cleanup(
    cleanup_commands: list[list[str]],
    workdirs: list[Path],
    env: dict[str, str],
    worker_count: int,
    runner: Optional[
        Callable[[list[str], Optional[Path], dict[str, str]], subprocess.CompletedProcess[str]]
    ] = None,
) -> list[dict[str, Any]]:
    if runner is None:
        runner = run_untimed
    if not cleanup_commands:
        return []
    existing_workdirs = [workdir for workdir in workdirs if workdir.exists()]
    if not existing_workdirs:
        return []
    bounded_workers = max(1, min(worker_count, len(existing_workdirs), len(cleanup_commands)))
    base_size, remainder = divmod(len(cleanup_commands), bounded_workers)
    command_chunks: list[tuple[int, Path, list[tuple[int, list[str]]]]] = []
    start = 0
    for worker_id in range(bounded_workers):
        size = base_size + (1 if worker_id < remainder else 0)
        indexed = list(enumerate(cleanup_commands[start:start + size], start=start))
        command_chunks.append((worker_id, existing_workdirs[worker_id], indexed))
        start += size

    def run_chunk(
        worker_id: int, cwd: Path, indexed_commands: list[tuple[int, list[str]]]
    ) -> list[dict[str, Any]]:
        failures: list[dict[str, Any]] = []
        for command_index, command in indexed_commands:
            try:
                proc = runner(command, cwd, env)
            except subprocess.TimeoutExpired as exc:
                failures.append(
                    {
                        "branch": command[-1],
                        "command": command,
                        "command_index": command_index,
                        "worker_id": worker_id,
                        "cwd": str(cwd),
                        "returncode": 124,
                        "stdout": oak_branch_fleet_seed_text(exc.stdout)[-1000:],
                        "stderr": f"cleanup_command_timeout:{exc.timeout}s",
                        "failure_reason": "cleanup_command_failed",
                    }
                )
                continue
            except Exception as exc:
                failures.append(
                    {
                        "branch": command[-1],
                        "command": command,
                        "command_index": command_index,
                        "worker_id": worker_id,
                        "cwd": str(cwd),
                        "returncode": 1,
                        "stdout": "",
                        "stderr": f"cleanup_command_exception:{type(exc).__name__}:{exc}",
                        "failure_reason": "cleanup_command_failed",
                    }
                )
                continue
            if proc.returncode != 0:
                failures.append(
                    {
                        "branch": command[-1],
                        "command": command,
                        "command_index": command_index,
                        "worker_id": worker_id,
                        "cwd": str(cwd),
                        "returncode": proc.returncode,
                        "stdout": proc.stdout[-1000:],
                        "stderr": proc.stderr[-1000:],
                    }
                )
        return failures

    if bounded_workers == 1:
        return run_chunk(command_chunks[0][0], command_chunks[0][1], command_chunks[0][2])

    failures: list[dict[str, Any]] = []
    with concurrent.futures.ThreadPoolExecutor(max_workers=bounded_workers) as pool:
        futures = [
            pool.submit(run_chunk, worker_id, cwd, indexed_commands)
            for worker_id, cwd, indexed_commands in command_chunks
        ]
        for future in concurrent.futures.as_completed(futures):
            failures.extend(future.result())
    failures.sort(key=lambda failure: int(failure.get("command_index") or 0))
    return failures


def run_oak_cli_branch_fleet_scenario(
    args: argparse.Namespace,
    scenario: dict[str, Any],
    meta: dict[str, Any],
    run_index: int,
    run_root: Path,
    settings: Settings,
) -> list[dict[str, Any]]:
    scenario_name = str(scenario["name"])
    operations = [str(op) for op in scenario.get("operations") or []]
    count = branch_fleet_count(scenario)
    specs = branch_fleet_specs(
        bench_id=str(meta.get("bench_id") or "bench"),
        scenario_name=scenario_name,
        run_index=run_index,
        branch_count=count,
    )
    env = cli_env(args)
    workdir = run_root / f"oak-branch-fleet-r{run_index}"
    reviewed: list[dict[str, Any]] = []
    plan: dict[str, Any] = {}
    apply_result: dict[str, Any] = {
        "apply_succeeded": False,
        "merged_branches": [],
        "final_survivors": [],
    }
    oracle: dict[str, Any] = {}
    total_start = platform_clock.monotonic_ms()
    cleanup_branches: list[str] = []
    seed_workdirs: list[Path] = []

    def seed() -> dict[str, Any]:
        nonlocal seed_workdirs
        start = platform_clock.monotonic_ms()
        seed_steps, _advancers = oak_branch_fleet_seed_steps(specs, meta, run_index)
        planned_commands = [command for step in seed_steps for command in step.commands]
        seed_command_count = sum(len(step.commands) for step in seed_steps)
        conflict_count = len(_advancers)
        clone = run_untimed(["oak", "clone", args.repo, str(workdir)], run_root, env)
        if clone.returncode != 0:
            row = branch_fleet_row(
                meta,
                scenario_name,
                "fleet.seed",
                run_index,
                platform_clock.monotonic_ms() - start,
                returncode=clone.returncode,
                provider=BRANCH_FLEET_PROVIDER_OAK_CLI,
                command=[["oak", "clone", args.repo, str(workdir)]],
                stderr=clone.stderr[-4000:],
                tool_call_count=1,
                terminal_tool_call_count=1,
                **oak_branch_fleet_seed_diagnostics(
                    branch_count=count,
                    conflict_count=conflict_count,
                    command_count=seed_command_count,
                    completed_count=0,
                    completed_branch_count=0,
                    completed_conflict_count=0,
                    completed_command_count=0,
                    last_completed_branch=None,
                    failed_command_index=None,
                    failed_branch=None,
                ),
                seed_parallel=False,
                seed_worker_count=0,
                seed_worker_clone_count=1,
                seed_conflict_serial=True,
                **branch_fleet_metric_fields(branch_count=count),
            )
            row["branches_seeded"] = 0
            return row

        branch_steps = [step for step in seed_steps if step.phase == "branch"]
        conflict_steps = [step for step in seed_steps if step.phase == "conflict"]
        worker_count = branch_fleet_seed_worker_count(len(branch_steps), env)
        branch_chunks = oak_branch_fleet_seed_step_chunks(seed_steps, worker_count)
        progress_lock = threading.Lock()

        def emit_progress(message: str) -> None:
            with progress_lock:
                print(message, file=sys.stderr, flush=True)

        def worker_seed(
            worker_id: int,
            command_index_offset: int,
            worker_steps: list[OakBranchFleetSeedStep],
        ) -> tuple[int, dict[str, Any], list[str], Path]:
            worker_dir = run_root / f"oak-branch-fleet-r{run_index}-seed-w{worker_id}"
            worker_start = platform_clock.monotonic_ms()
            worker_clone = run_untimed(["oak", "clone", args.repo, str(worker_dir)], run_root, env)
            if worker_clone.returncode != 0:
                return (
                    worker_id,
                    oak_branch_fleet_seed_clone_failure_row(
                        meta,
                        scenario_name,
                        run_index,
                        worker_id=worker_id,
                        command=["oak", "clone", args.repo, str(worker_dir)],
                        proc=worker_clone,
                        elapsed_ms=platform_clock.monotonic_ms() - worker_start,
                        admitted=settings.admitted,
                        command_index_offset=command_index_offset,
                        failed_branch=worker_steps[0].branch if worker_steps else None,
                        branch_count=len(worker_steps),
                        command_count=sum(len(step.commands) for step in worker_steps),
                    ),
                    [],
                    worker_dir,
                )
            local_cleanup: list[str] = []
            row = oak_branch_fleet_seed_row(
                meta,
                scenario_name,
                worker_steps,
                worker_dir,
                run_index,
                env,
                settings.admitted,
                len(worker_steps),
                0,
                progress=emit_progress,
                cleanup_branches=local_cleanup,
                command_index_offset=command_index_offset,
            )
            row["seed_worker_id"] = worker_id
            row["seed_worker_clone_count"] = 1
            row["elapsed_ms"] = round(platform_clock.monotonic_ms() - worker_start, 3)
            return worker_id, row, local_cleanup, worker_dir

        branch_seed_results: list[tuple[int, dict[str, Any], list[str], Path]] = []
        if branch_chunks:
            with concurrent.futures.ThreadPoolExecutor(max_workers=len(branch_chunks)) as pool:
                futures = [
                    pool.submit(worker_seed, worker_id, command_offset, worker_steps)
                    for worker_id, command_offset, worker_steps in branch_chunks
                ]
                for future in concurrent.futures.as_completed(futures):
                    branch_seed_results.append(future.result())
            branch_seed_results.sort(key=lambda result: result[0])

        seed_rows = [result[1] for result in branch_seed_results]
        for _worker_id, _row, worker_cleanup, worker_dir in branch_seed_results:
            cleanup_branches.extend(worker_cleanup)
            seed_workdirs.append(worker_dir)

        branch_seed_failed = any(row_returncode(row) not in (0, SKIP_RETURNCODE) for row in seed_rows)
        if not branch_seed_failed and conflict_steps:
            conflict_cleanup: list[str] = []
            conflict_row = oak_branch_fleet_seed_row(
                meta,
                scenario_name,
                conflict_steps,
                workdir,
                run_index,
                env,
                settings.admitted,
                0,
                conflict_count,
                progress=emit_progress,
                cleanup_branches=conflict_cleanup,
                command_index_offset=sum(len(step.commands) for step in branch_steps),
            )
            conflict_row["seed_worker_id"] = "conflict-serial"
            conflict_row["seed_worker_clone_count"] = 0
            seed_rows.append(conflict_row)
            cleanup_branches.extend(conflict_cleanup)

        row = oak_branch_fleet_aggregate_seed_rows(
            meta,
            scenario_name,
            run_index,
            seed_rows=seed_rows,
            elapsed_ms=platform_clock.monotonic_ms() - start,
            planned_commands=planned_commands,
            branch_count=count,
            conflict_count=conflict_count,
            worker_count=len(branch_chunks),
            worker_clone_count=len(branch_seed_results) + 1,
        )
        row["seeded_branch_classes"] = [spec["kind"] for spec in specs]
        row.update(branch_fleet_metric_fields(branch_count=count))
        row["branches_seeded"] = int(row.get("seed_completed_branch_count") or 0)
        return row

    def classify() -> dict[str, Any]:
        nonlocal reviewed
        row = cli_step_row(
            meta,
            scenario_name,
            "fleet.classify",
            [
                "oak",
                "branch",
                "triage",
                "--remote",
                "--against",
                "main",
                "--status",
                "open",
                "--analysis-depth",
                "manifest",
                "--json",
            ],
            workdir,
            run_index,
            env,
            settings.admitted,
            full_stdout_bytes=BRANCH_FLEET_CLASSIFY_PARSE_STDOUT_BYTES,
        )
        row["branch_fleet_provider"] = BRANCH_FLEET_PROVIDER_OAK_CLI
        rows_payload: list[Any] = []
        if row["returncode"] == 0:
            if int(row.get("stdout_bytes") or 0) > BRANCH_FLEET_CLASSIFY_PARSE_STDOUT_BYTES:
                row["returncode"] = 1
                row["stderr"] = (
                    "oak_branch_triage_stdout_exceeded_parse_limit:"
                    f"{BRANCH_FLEET_CLASSIFY_PARSE_STDOUT_BYTES}"
                )
            else:
                try:
                    payload = json.loads(row.get("_stdout") or "{}")
                    rows_payload = payload.get("branches") or payload.get("rows") or []
                except json.JSONDecodeError as exc:
                    row["returncode"] = 1
                    row["stderr"] = f"oak_branch_triage_invalid_json:{exc}"
        by_name = {
            str(item.get("branch") or item.get("name")): item
            for item in rows_payload
            if isinstance(item, dict)
        }
        reviewed = []
        for spec in specs:
            actual = by_name.get(str(spec["name"])) or {}
            reviewed.append(
                {
                    "name": str(spec["name"]),
                    "mergeable": oak_branch_fleet_mergeable(actual),
                    "expected_kind": spec["kind"],
                    "raw": actual,
                }
            )
        row["reviewed_branches"] = [branch["name"] for branch in reviewed]
        row.update(branch_fleet_metric_fields(branch_count=count, reviewed=reviewed))
        return row

    def plan_row() -> dict[str, Any]:
        nonlocal plan
        start = platform_clock.monotonic_ms()
        plan = branch_fleet_plan(reviewed)
        return branch_fleet_row(
            meta,
            scenario_name,
            "fleet.plan",
            run_index,
            platform_clock.monotonic_ms() - start,
            provider=BRANCH_FLEET_PROVIDER_OAK_CLI,
            action_plan=plan["actions"],
            merge_order=plan["merge_order"],
            tool_call_count=0,
            terminal_tool_call_count=0,
            **branch_fleet_metric_fields(branch_count=count, reviewed=reviewed, plan=plan),
        )

    def apply() -> dict[str, Any]:
        nonlocal apply_result
        merge_order = list(plan.get("merge_order") or [])
        commands = oak_branch_fleet_apply_commands([str(branch) for branch in merge_order])
        row = cli_sequence_row(
            meta, scenario_name, "fleet.apply", commands, workdir, run_index, env, settings.admitted
        )
        survivors = [
            action["branch"]
            for action in plan.get("actions") or []
            if action["action"] == "leave_conflict_open"
        ]
        apply_result = {
            "apply_succeeded": row["returncode"] == 0,
            "merged_branches": merge_order if row["returncode"] == 0 else [],
            "final_survivors": survivors if row["returncode"] == 0 else [],
        }
        row["branch_fleet_provider"] = BRANCH_FLEET_PROVIDER_OAK_CLI
        row.update(apply_result)
        row.update(
            branch_fleet_metric_fields(
                branch_count=count, reviewed=reviewed, plan=plan, apply_result=apply_result
            )
        )
        return row

    def sync() -> dict[str, Any]:
        nonlocal apply_result
        row = cli_sequence_row(
            meta,
            scenario_name,
            "fleet.sync",
            [["oak", "branch", "list", "--remote", "--status", "open", "--json"]],
            workdir,
            run_index,
            env,
            settings.admitted,
        )
        row["branch_fleet_provider"] = BRANCH_FLEET_PROVIDER_OAK_CLI
        if row["returncode"] == 0:
            try:
                final_survivors = branch_fleet_survivors_from_sync_stdout(
                    str(row.get("_stdout") or ""), specs
                )
            except json.JSONDecodeError as exc:
                row["returncode"] = 1
                row["stderr"] = f"oak_branch_list_invalid_json:{exc}"
            except ValueError as exc:
                row["returncode"] = 1
                row["stderr"] = f"oak_branch_list_schema_error:{exc}"
            else:
                apply_result = {
                    **apply_result,
                    "final_survivors": final_survivors,
                }
                row["final_survivors"] = final_survivors
        row.update(
            branch_fleet_metric_fields(
                branch_count=count, reviewed=reviewed, plan=plan, apply_result=apply_result
            )
        )
        return row

    def oracle_row() -> dict[str, Any]:
        nonlocal oracle
        start = platform_clock.monotonic_ms()
        oracle = branch_fleet_oracle(specs, plan, apply_result)
        ok = all(
            bool(oracle[key])
            for key in (
                "apply_succeeded",
                "action_plan_correct",
                "merge_order_correct",
                "final_survivors_correct",
            )
        )
        fields = {
            **oracle,
            **branch_fleet_metric_fields(
                branch_count=count,
                reviewed=reviewed,
                plan=plan,
                apply_result=apply_result,
                oracle=oracle,
            ),
        }
        return branch_fleet_row(
            meta,
            scenario_name,
            "fleet.oracle",
            run_index,
            platform_clock.monotonic_ms() - start,
            returncode=0 if ok else 1,
            provider=BRANCH_FLEET_PROVIDER_OAK_CLI,
            tool_call_count=0,
            terminal_tool_call_count=0,
            **fields,
        )

    def total() -> dict[str, Any]:
        failures = [
            row for row in rows
            if row_returncode(row) not in (0, SKIP_RETURNCODE)
        ]
        row = branch_fleet_row(
            meta,
            scenario_name,
            "fleet.total",
            run_index,
            platform_clock.monotonic_ms() - total_start,
            returncode=1 if failures else 0,
            provider=BRANCH_FLEET_PROVIDER_OAK_CLI,
            summarized_operations=[row["operation"] for row in rows],
            **branch_fleet_workflow_summary_fields(rows),
            **branch_fleet_metric_fields(
                branch_count=count,
                reviewed=reviewed,
                plan=plan,
                apply_result=apply_result,
                oracle=oracle,
            ),
        )
        row["tool_call_count"] = sum(int(r.get("tool_call_count") or 0) for r in rows)
        row.update(
            oakbench_tokens.summed_token_fields(
                rows, "sum_of_branch_fleet_steps_command_plus_admitted_output_chars_div_4"
            )
        )
        return row

    handlers: dict[str, Callable[[], dict[str, Any]]] = {
        "fleet.seed": seed,
        "fleet.classify": classify,
        "fleet.plan": plan_row,
        "fleet.apply": apply,
        "fleet.sync": sync,
        "fleet.oracle": oracle_row,
        "fleet.total": total,
    }
    rows: list[dict[str, Any]] = []
    aborted_at: Optional[str] = None
    for operation in operations:
        if operation == "fleet.total":
            row = total()
            row.pop("_stdout", None)
            rows.append(row)
            continue
        if aborted_at is not None:
            rows.append(skip_row(meta, scenario_name, operation, run_index, f"prior_step_failed:{aborted_at}"))
            continue
        handler = handlers.get(operation)
        if handler is None:
            rows.append(skip_row(meta, scenario_name, operation, run_index, "operation_not_implemented"))
            continue
        row = handler()
        row.pop("_stdout", None)
        rows.append(row)
        if row_returncode(row) not in (0, SKIP_RETURNCODE):
            aborted_at = operation

    cleanup_commands = oak_branch_fleet_cleanup_commands_for_branches(cleanup_branches)
    cleanup_failures = oak_branch_fleet_run_cleanup(
        cleanup_commands,
        [workdir] + seed_workdirs,
        env,
        branch_fleet_cleanup_worker_count(len(cleanup_commands), env),
    )
    cleanup_fields = oak_branch_fleet_cleanup_fields(cleanup_failures)
    for row in rows:
        if row.get("operation") == "fleet.total":
            row.update(cleanup_fields)
            break
    return rows


# --------------------------------------------------------------------------
# branch_triage_nN: seeded remote branch/PR review classification.
#
# The real provider path is intentionally skip-only for now. The fake provider
# exercises the decision/oracle logic deterministically: N seeded branches cycle
# through clean/stale/duplicate/conflicting, the plan decides merge/refresh/
# close/leave-open actions, and the oracle checks both the action plan and the
# final survivors.
# --------------------------------------------------------------------------


def is_branch_triage_scenario(scenario: dict[str, Any]) -> bool:
    return BRANCH_TRIAGE_SCENARIO_RE.match(str(scenario.get("name", ""))) is not None


def branch_triage_count(scenario: dict[str, Any]) -> int:
    if scenario.get("branch_count") is not None:
        return int(scenario["branch_count"])
    match = BRANCH_TRIAGE_SCENARIO_RE.match(str(scenario.get("name", "")))
    return int(match.group(1)) if match else 4


def seeded_branch_triage_branches(branch_count: int) -> list[dict[str, Any]]:
    branches: list[dict[str, Any]] = []
    clean_patch_id = "patch-clean-0"
    for index in range(branch_count):
        expected_class = BRANCH_TRIAGE_CLASSES[index % len(BRANCH_TRIAGE_CLASSES)]
        name = f"triage-{expected_class}-{index}"
        branch: dict[str, Any] = {
            "name": name,
            "expected_class": expected_class,
            "expected_action": BRANCH_TRIAGE_ACTIONS[expected_class],
            "base_sha": "main-sha-0",
            "head_sha": f"head-{index}",
            "behind_by": 0,
            "mergeable": True,
            "conflict_paths": [],
            "patch_id": f"patch-{expected_class}-{index}",
        }
        if expected_class == "clean":
            branch["patch_id"] = clean_patch_id if index == 0 else f"patch-clean-{index}"
        elif expected_class == "stale":
            branch["behind_by"] = 2
            branch["base_sha"] = "old-main-sha"
        elif expected_class == "duplicate":
            branch["patch_id"] = clean_patch_id
            branch["duplicate_of"] = "triage-clean-0"
        elif expected_class == "conflicting":
            branch["mergeable"] = False
            branch["conflict_paths"] = [f"src/conflict-{index}.txt"]
        branches.append(branch)
    return branches


def classify_triage_branch(branch: dict[str, Any], seen_patch_ids: set[str]) -> str:
    if branch.get("conflict_paths") or branch.get("mergeable") is False:
        return "conflicting"
    patch_id = str(branch.get("patch_id") or "")
    if patch_id and patch_id in seen_patch_ids:
        return "duplicate"
    if int(branch.get("behind_by") or 0) > 0:
        if patch_id:
            seen_patch_ids.add(patch_id)
        return "stale"
    if patch_id:
        seen_patch_ids.add(patch_id)
    return "clean"


def branch_triage_plan(reviewed: list[dict[str, Any]]) -> dict[str, Any]:
    seen_patch_ids: set[str] = set()
    actions: list[dict[str, str]] = []
    merge_order: list[str] = []
    detected_counts = {key: 0 for key in BRANCH_TRIAGE_CLASSES}
    for branch in reviewed:
        classification = classify_triage_branch(branch, seen_patch_ids)
        detected_counts[classification] += 1
        action = BRANCH_TRIAGE_ACTIONS[classification]
        name = str(branch["name"])
        actions.append({"branch": name, "classification": classification, "action": action})
        if action in ("merge", "refresh_then_merge"):
            merge_order.append(name)
    return {
        "actions": actions,
        "merge_order": merge_order,
        "detected_counts": detected_counts,
    }


def expected_triage_plan(branches: list[dict[str, Any]]) -> dict[str, Any]:
    counts = {key: 0 for key in BRANCH_TRIAGE_CLASSES}
    actions: list[dict[str, str]] = []
    merge_order: list[str] = []
    for branch in branches:
        classification = str(branch["expected_class"])
        action = str(branch["expected_action"])
        counts[classification] += 1
        name = str(branch["name"])
        actions.append({"branch": name, "classification": classification, "action": action})
        if action in ("merge", "refresh_then_merge"):
            merge_order.append(name)
    return {"actions": actions, "merge_order": merge_order, "detected_counts": counts}


class FakeBranchTriageProvider:
    def __init__(self, branch_count: int) -> None:
        self._branches = seeded_branch_triage_branches(branch_count)
        self._open = {str(branch["name"]) for branch in self._branches}
        self.closed: list[str] = []

    def seed(self) -> list[dict[str, Any]]:
        return [dict(branch) for branch in self._branches]

    def review(self) -> list[dict[str, Any]]:
        return [dict(branch) for branch in self._branches if str(branch["name"]) in self._open]

    def apply(self, plan: dict[str, Any]) -> dict[str, Any]:
        for action in plan.get("actions") or []:
            branch = str(action["branch"])
            decision = str(action["action"])
            if decision in ("merge", "refresh_then_merge", "close_duplicate") and branch in self._open:
                self._open.remove(branch)
                self.closed.append(branch)
        return {"branches_closed": len(self.closed), "final_survivors": sorted(self._open)}


def branch_triage_metric_fields(
    *,
    branches_reviewed: int,
    branches_closed: int,
    plan: Optional[dict[str, Any]] = None,
    oracle: Optional[dict[str, Any]] = None,
) -> dict[str, Any]:
    counts = ((plan or {}).get("detected_counts") or {})
    return {
        "branches_reviewed": branches_reviewed,
        "branches_closed": branches_closed,
        "merge_order_correct": (oracle or {}).get("merge_order_correct"),
        "clean_detected_count": int(counts.get("clean", 0)),
        "stale_detected_count": int(counts.get("stale", 0)),
        "duplicate_detected_count": int(counts.get("duplicate", 0)),
        "conflict_detected_count": int(counts.get("conflicting", 0)),
    }


def branch_triage_oracle(
    seeded: list[dict[str, Any]],
    plan: dict[str, Any],
    apply_result: dict[str, Any],
) -> dict[str, Any]:
    expected = expected_triage_plan(seeded)
    expected_survivors = sorted(
        str(branch["name"])
        for branch in seeded
        if branch["expected_action"] == "keep_open_manual_resolution"
    )
    final_survivors = sorted(str(name) for name in apply_result.get("final_survivors") or [])
    return {
        "action_plan_correct": plan.get("actions") == expected["actions"],
        "merge_order_correct": plan.get("merge_order") == expected["merge_order"],
        "final_survivors_correct": final_survivors == expected_survivors,
        "expected_action_plan": expected["actions"],
        "expected_merge_order": expected["merge_order"],
        "expected_final_survivors": expected_survivors,
        "final_survivors": final_survivors,
    }


def fake_branch_triage_row(
    meta: dict[str, Any],
    scenario_name: str,
    operation: str,
    run_index: int,
    elapsed_ms: float,
    returncode: int = 0,
    **fields: Any,
) -> dict[str, Any]:
    return {
        **meta,
        "scenario": scenario_name,
        "run": run_index,
        "operation": operation,
        "elapsed_ms": round(elapsed_ms, 3),
        "returncode": returncode,
        "command": [],
        "tool_call_count": 0,
        "branch_triage_provider": "fake",
        **fake_provider_token_fields(),
        **fields,
    }


def run_branch_triage_fake_scenario(
    scenario: dict[str, Any],
    meta: dict[str, Any],
    run_index: int,
) -> list[dict[str, Any]]:
    scenario_name = str(scenario["name"])
    operations = [str(op) for op in scenario.get("operations") or []]
    provider = FakeBranchTriageProvider(branch_triage_count(scenario))
    rows: list[dict[str, Any]] = []
    seeded: list[dict[str, Any]] = []
    reviewed: list[dict[str, Any]] = []
    plan: dict[str, Any] = {}
    apply_result: dict[str, Any] = {"branches_closed": 0, "final_survivors": []}
    oracle: dict[str, Any] = {}
    total_start = platform_clock.monotonic_ms()

    for operation in operations:
        start = platform_clock.monotonic_ms()
        if operation == "triage.seed":
            seeded = provider.seed()
            rows.append(
                fake_branch_triage_row(
                    meta, scenario_name, operation, run_index,
                    platform_clock.monotonic_ms() - start,
                    branches_seeded=len(seeded),
                    seeded_branch_classes=[branch["expected_class"] for branch in seeded],
                    **branch_triage_metric_fields(branches_reviewed=0, branches_closed=0),
                )
            )
        elif operation == "triage.review":
            reviewed = provider.review()
            rows.append(
                fake_branch_triage_row(
                    meta, scenario_name, operation, run_index,
                    platform_clock.monotonic_ms() - start,
                    branches_reviewed=len(reviewed),
                    branches_closed=0,
                    reviewed_branches=[branch["name"] for branch in reviewed],
                )
            )
        elif operation == "triage.plan":
            plan = branch_triage_plan(reviewed)
            rows.append(
                fake_branch_triage_row(
                    meta, scenario_name, operation, run_index,
                    platform_clock.monotonic_ms() - start,
                    action_plan=plan["actions"],
                    merge_order=plan["merge_order"],
                    **branch_triage_metric_fields(
                        branches_reviewed=len(reviewed), branches_closed=0, plan=plan
                    ),
                )
            )
        elif operation == "triage.apply":
            apply_result = provider.apply(plan)
            rows.append(
                fake_branch_triage_row(
                    meta, scenario_name, operation, run_index,
                    platform_clock.monotonic_ms() - start,
                    final_survivors=apply_result["final_survivors"],
                    **branch_triage_metric_fields(
                        branches_reviewed=len(reviewed),
                        branches_closed=int(apply_result["branches_closed"]),
                        plan=plan,
                    ),
                )
            )
        elif operation == "triage.oracle":
            oracle = branch_triage_oracle(seeded, plan, apply_result)
            ok = all(
                bool(oracle[key])
                for key in ("action_plan_correct", "merge_order_correct", "final_survivors_correct")
            )
            rows.append(
                fake_branch_triage_row(
                    meta, scenario_name, operation, run_index,
                    platform_clock.monotonic_ms() - start,
                    returncode=0 if ok else 1,
                    **{
                        **oracle,
                        **branch_triage_metric_fields(
                            branches_reviewed=len(reviewed),
                            branches_closed=int(apply_result["branches_closed"]),
                            plan=plan,
                            oracle=oracle,
                        ),
                    },
                )
            )
        elif operation == "triage.total":
            failures = [
                row for row in rows
                if row_returncode(row) not in (0, SKIP_RETURNCODE)
            ]
            rows.append(
                fake_branch_triage_row(
                    meta, scenario_name, operation, run_index,
                    platform_clock.monotonic_ms() - total_start,
                    returncode=1 if failures else 0,
                    summarized_operations=[row["operation"] for row in rows],
                    final_survivors=apply_result.get("final_survivors", []),
                    **branch_triage_metric_fields(
                        branches_reviewed=len(reviewed),
                        branches_closed=int(apply_result["branches_closed"]),
                        plan=plan,
                        oracle=oracle,
                    ),
                )
            )
        else:
            rows.append(skip_row(meta, scenario_name, operation, run_index, "operation_not_implemented"))
    return rows


def run_branch_triage_real_skip(
    args: argparse.Namespace,
    scenario: dict[str, Any],
    meta: dict[str, Any],
    run_index: int,
) -> list[dict[str, Any]]:
    scenario_name = str(scenario["name"])
    reason = f"branch_triage_real_provider_not_implemented:{args.platform}:{args.driver}"
    return [
        skip_row(meta, scenario_name, str(operation), run_index, reason)
        for operation in scenario.get("operations") or []
    ]


def run_scenario_with_credentials(
    args: argparse.Namespace,
    scenario: dict[str, Any],
    meta: dict[str, Any],
    run_index: int,
    run_root: Path,
    settings: Settings,
) -> list[dict[str, Any]]:
    if scenario.get("name") == "platform_capability_probe":
        return run_capability_probe(args, meta, run_index, settings)
    if is_branch_fleet_scenario(scenario):
        live_skip_reason = branch_fleet_live_skip_reason(scenario, args.driver)
        if live_skip_reason is not None:
            scenario_name = str(scenario["name"])
            return [
                skip_row(meta, scenario_name, str(operation), run_index, live_skip_reason)
                for operation in scenario.get("operations") or []
            ]
        if args.driver == BRANCH_TRIAGE_FAKE_DRIVER:
            return run_branch_fleet_fake_scenario(scenario, meta, run_index)
        if args.platform == "github" and args.driver == "harness-api":
            return run_github_api_branch_fleet_scenario(args, scenario, meta, run_index, settings)
        if args.platform == "oak" and args.driver == "cli":
            capability_skip = oak_cli_commit_push_skip_reason(cli_env(args))
            if capability_skip is not None:
                scenario_name = str(scenario["name"])
                return [
                    skip_row(meta, scenario_name, str(operation), run_index, capability_skip)
                    for operation in scenario.get("operations") or []
                ]
            return run_oak_cli_branch_fleet_scenario(args, scenario, meta, run_index, run_root, settings)
        scenario_name = str(scenario["name"])
        reason = f"branch_fleet_provider_not_implemented:{args.platform}:{args.driver}"
        return [
            skip_row(meta, scenario_name, str(operation), run_index, reason)
            for operation in scenario.get("operations") or []
        ]
    if is_branch_triage_scenario(scenario):
        if args.driver == BRANCH_TRIAGE_FAKE_DRIVER:
            return run_branch_triage_fake_scenario(scenario, meta, run_index)
        return run_branch_triage_real_skip(args, scenario, meta, run_index)
    if is_race_scenario(scenario):
        return run_race_with_credentials(args, scenario, meta, run_index, run_root, settings)
    if args.platform == "github" and args.driver == "harness-api":
        return run_github_api_scenario(args, scenario, meta, run_index, settings)
    if args.platform == "github":
        return run_github_cli_scenario(args, scenario, meta, run_index, run_root, settings)
    return run_oak_cli_scenario(args, scenario, meta, run_index, run_root, settings)


# --------------------------------------------------------------------------
# Summary + main
# --------------------------------------------------------------------------


def summary_text(rows: list[dict[str, Any]]) -> str:
    lines = [
        "# Platform Lane Summary",
        "",
        "Hosted integration workflow: task branch exists locally -> change merged",
        "in main, present in local checkout. Round trips, bytes, polls, and tokens",
        "are always comparable across platforms; wall clock only with per-platform",
        "RTT context and an explicit caveat. Settle durations are bounded by their",
        "poll quantization. This lane is excluded from devloop.",
        "",
        "| Platform | Driver | Scenario | Run | Operation | Wall ms | Polls | rc |",
        "| --- | --- | --- | ---: | --- | ---: | ---: | ---: |",
    ]
    for row in rows:
        if row.get("skipped"):
            continue
        lines.append(
            "| {} | {} | {} | {} | `{}` | {} | {} | {} |".format(
                row.get("platform"), row.get("driver"), row.get("scenario"), row.get("run"),
                row.get("operation"), row.get("elapsed_ms"),
                row.get("poll_iterations", ""), row.get("returncode"),
            )
        )
    skips = [row for row in rows if row.get("skipped")]
    if skips:
        lines += ["", "## Skips", ""]
        reasons: dict[str, int] = {}
        for row in skips:
            reasons[str(row.get("skip_reason"))] = reasons.get(str(row.get("skip_reason")), 0) + 1
        for reason, count in sorted(reasons.items()):
            lines.append(f"- `{reason}`: {count} row(s)")
    return "\n".join(lines) + "\n"


def main(argv: Optional[list[str]] = None) -> int:
    args = parse_args(argv)
    spec = load_platform_spec(args.spec)
    semantics = load_semantics(args.semantics_config)
    scenario = scenario_by_name(spec, args.scenario)
    if scenario is None:
        names = ", ".join(str(s.get("name")) for s in spec.get("scenarios") or [] if isinstance(s, dict))
        print(f"unknown scenario {args.scenario!r}; known: {names}", file=sys.stderr)
        return 2
    declared_variants = semantics.get("protection_variants") or []
    if args.protection_variant not in declared_variants:
        print(
            f"unknown protection variant {args.protection_variant!r}; declared: {', '.join(declared_variants)}",
            file=sys.stderr,
        )
        return 2

    timestamp = datetime.now(timezone.utc).strftime("%Y%m%dT%H%M%SZ")
    settings = scenario_settings(spec, scenario, args.admitted_output_chars)
    runner = runner_fields()
    meta: dict[str, Any] = {
        "bench_id": timestamp,
        "profile": "platform",
        "benchmark_track": "platform-hosted",
        "timestamp_utc": timestamp,
        "host": platform_module.node(),
        "os_platform": platform_module.platform(),
        "machine": platform_module.machine(),
        "python": platform_module.python_version(),
        "env_isolation_version": ENV_ISOLATION_VERSION,
        "platform": args.platform,
        "driver": args.driver,
        "platform_semantics_class": args.semantics_class,
        "protection_variant": args.protection_variant,
        "platform_semantics_version": semantics.get("semantics_version"),
        "platform_comparison_key": platform_comparison_key(
            semantics.get("semantics_version"),
            args.semantics_class,
            args.protection_variant,
        ),
        # The contract only matches when the variant is declared+implemented
        # AND this driver actually executes it (check-instant needs the
        # status API, which the cli driver does not have).
        "semantic_contract_match": (
            compute_semantic_contract_match(
                semantics, args.semantics_class, args.protection_variant
            )
            and protection_variant_executable(
                args.platform, args.driver, args.protection_variant
            )
        ),
        "semantics_version": semantics.get("semantics_version"),
        "semantics_schema_version": semantics.get("schema_version"),
        "repo": args.repo or None,
        "token_env": args.token_env,
        "excluded_from_devloop": True,
        "clock": "time.monotonic (single local monotonic clock for all headline durations)",
    }

    skip_reason = readiness_skip_reason(
        args.platform,
        args.driver,
        args.repo,
        args.token_env,
        requires_remote=bool(scenario.get("requires_remote")),
    )
    if skip_reason is None and args.protection_variant not in (
        semantics.get("implemented_protection_variants") or []
    ):
        skip_reason = f"protection_variant_not_implemented:{args.protection_variant}"

    operations = [str(op) for op in scenario.get("operations") or []]
    scenario_name = str(scenario.get("name"))
    rows: list[dict[str, Any]] = []
    run_root = args.workdir / "runs" / timestamp

    with measurement_lock("platform_lifecycle") as lock_info:
        meta["measurement_lock_wait_ms"] = lock_info.wait_ms
        meta["measurement_lock"] = "held" if lock_info.enabled else "disabled"
        for run_index in range(max(1, args.runs)):
            if skip_reason is not None:
                rows.extend(
                    skip_row(
                        meta,
                        scenario_name,
                        operation,
                        run_index,
                        skip_reason,
                        settings=settings,
                    )
                    for operation in operations
                )
                continue
            run_root.mkdir(parents=True, exist_ok=True)
            print(f"[run] {scenario_name} run={run_index} platform={args.platform} driver={args.driver}", flush=True)
            rows.extend(
                run_scenario_with_credentials(args, scenario, meta, run_index, run_root, settings)
            )

    for row in rows:
        row.pop("_stdout", None)
    stamped = [stamp_row(row, runner) for row in rows]
    store = ResultsStore(args.results, lane="platform", filename_suffix="platform")
    raw_path, summary_path = store.write(timestamp, stamped, summary_text(stamped))
    print(f"[result] {raw_path}")
    print(f"[summary] {summary_path}")
    if skip_reason is None and not args.keep_workdirs:
        shutil.rmtree(run_root, ignore_errors=True)

    if stamped and all(row.get("skipped") for row in stamped):
        printed_skip_reason = skip_reason
        if printed_skip_reason is None:
            row_reasons = sorted(
                {
                    str(row.get("skip_reason"))
                    for row in stamped
                    if row.get("skip_reason")
                }
            )
            printed_skip_reason = ", ".join(row_reasons) if row_reasons else None
        print(f"[skip] all rows skipped: {printed_skip_reason}", file=sys.stderr)
        return EXIT_ALL_SKIPPED
    failures = [
        row for row in stamped if row_returncode(row) not in (0, SKIP_RETURNCODE)
    ]
    return 1 if failures else 0


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

Reversible clean-root candidate rebuilt from oak/benchmarks main@009dead264ca; original repository retained.
Download

RunnersDetails

No runs on main yet.
Open branchesHistory

mrmrs-516758agent_loop_hosted_v1: hosted agent-loop gap map lane (Oak vs GitHub vs ideal)checking…6 behind1mo ago
mrmrs-b26e28Add stale_branch_land_v1: baseline for landing branches that are behind mainchecking…6 behind1mo ago
mrmrs-fad256task_loop: shallow_worker_branch_refresh_v1 (fresh shallow worker -> task branch -> first parent refresh, Oak vs Git, HTTP request counts)checking…6 behind1mo ago
oakmark-throughput-benchmark-transplantAdd oakmark: per-repo push/clone throughput benchmark for oak (ForgeMark analogue)checking…11 behind3mo ago
All branches →
Activity12 commits to main

May 2026now
1	`#!/usr/bin/env python3`
2	`"""The platform lane: hosted integration workflows, end to end.`
3
4	`Measures GitHub-the-platform vs oak hosted workflows from "task branch exists`
5	`locally" to "change merged in main, present in local checkout" — including`
6	`the async waits agents actually experience (GitHub computes PR mergeability`
7	`asynchronously; that settling is a measured operation, observed by polling on`
8	`one local monotonic clock).`
9
10	`Drivers per platform:`
11	`cli shells out to gh/git/oak via run_timed — full token/output`
12	`accounting from captured output, like every other CLI lane.`
13	`harness-api stdlib urllib REST against the GitHub API — exact api_metrics`
14	`(request_count/bytes, rate-limit header, request ids); token`
15	`fields stay null because API JSON is not an agent transcript`
16	`and token costs would be fiction (ADR-0002).`
17	`fake-provider`
18	`deterministic in-process branch triage provider for oracle`
19	`coverage when no hosted platform is available; not comparable`
20	`to real platform timings.`
21
22	`This lane runs against real hosted services and is EXCLUDED from devloop:`
23	`external variance must never poison changeset verdicts.`
24
25	`Without credentials/configured repos (the only path CI exercises) every`
26	`operation becomes a skip row with an explicit skip_reason and the process`
27	`exits 3 (all-skipped), never silent absence.`
28
29	`Setup:`
30	`GITHUB_TOKEN=... python3 scripts/platform_lifecycle.py \\`
31	`--platform github --driver harness-api --scenario pr_single_anatomy \\`
32	`--repo your-org/disposable-bench-repo --runs 3`
33	`"""`
34
35	`from __future__ import annotations`
36
37	`import argparse`
38	`import concurrent.futures`
39	`import json`
40	`import os`