#!/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)

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