scripts/bench.py · oak/benchmarks

Browse files
scripts/bench.py 2402 lines · 100.7 KB · python
#!/usr/bin/env python3
"""Benchmark Git and Oak on agent-shaped repository workloads.

The harness is intentionally dependency-free. It generates deterministic
fixtures, copies each fixture per subject, times only VCS commands, and writes
JSONL plus a Markdown summary.
"""

from __future__ import annotations

import argparse
import fnmatch
import functools
import hashlib
import json
import math
import os
import platform
import random
import re
import secrets
import shutil
import statistics
import subprocess
import tempfile
import time
from dataclasses import dataclass
from datetime import datetime, timezone
from pathlib import Path
from typing import Any

from oakbench import cachectl as oakbench_cachectl
from oakbench import environment as oakbench_environment
from oakbench import envwatch as oakbench_envwatch
from oakbench import fixtures as oakbench_fixtures
from oakbench import loadgen as oakbench_loadgen
from oakbench import integrity as oakbench_integrity
from oakbench import output_semantics as oakbench_semantics
from oakbench import remotes as oakbench_remotes
from oakbench import tokens as oakbench_tokens
from oakbench.tokens import COST_WEIGHTS_NOTE
from oakbench.command_semantics import semantics
from oakbench.environment import command_display
from oakbench.execution import PEAK_RSS_SOURCE, RUSAGE_SOURCE, run_timed
from oakbench.reporting import delta_pct, fmt_delta_lower_better, fmt_num
from oakbench.results import ResultsStore
from oakbench.rows import row_returncode
from oakbench.runner import runner_fields
from oakbench.runlock import measurement_lock
from oakbench.subjects import (
    DEFAULT_OAK_REPO,
    Subject,
    binary_sha256,
    git_config_profile,
    load_subjects,
    source_metadata,
    subject_details,
    subject_version,
)

ROOT = Path(__file__).resolve().parents[1]
DEFAULT_WORKDIR = Path(tempfile.gettempdir()) / "oak-bench-work"
DEFAULT_ADMITTED_OUTPUT_CHARS = 20_000
JSON_PROBE_MAX_BYTES = 64 * 1024 * 1024
COMMAND_SEMANTICS_VERSION = semantics().version
TRACKS = ("agent-default", "core-equivalent")
GIT_MODES_CONFIG = ROOT / "config" / "git_modes.json"
BENCH_TASK_BRANCH = "agent-task"
# Core-lane row contract additions (additive; ADR-0002/0005): page-cache
# labels on every timed row, opt-in cold mode, rusage CPU/IO, recorded skips
# for operations abandoned after a failed prerequisite, per-row provenance.
TIMING_CONTRACT_VERSION = "2026-09-27.1"
CACHE_MODES = ("warm", "cold")

# Tuned Git modes. Setup commands run after repo.init as recorded
# mode.setup.<step> operation rows (excluded from cross-op averages but never
# absorbed), so the timed operations measure the tuned behavior while the
# setup cost stays visible. LFS is fixture-aware: it only applies to scenarios
# listed in config/git_modes.json applies_to and tracks the generated *.bin
# fixture files before the first snapshot. Each entry is (operation suffix,
# git args): the suffix names the mode.setup.<suffix> row, so multi-step
# setups (lfs install + lfs track) record one row per step exactly like the
# single-step fsmonitor mode records mode.setup.fsmonitor.
LFS_TRACK_PATTERNS: tuple[str, ...] = ("*.bin",)
WIRED_GIT_MODES: dict[str, list[tuple[str, list[str]]]] = {
    "untracked_cache": [("untracked_cache", ["update-index", "--untracked-cache"])],
    "split_index": [("split_index", ["update-index", "--split-index"])],
    "fsmonitor": [("fsmonitor", ["config", "core.fsmonitor", "true"]),
                  ("fsmonitor_start", ["fsmonitor--daemon", "start"]),
                  ("fsmonitor_warm_status", ["status", "--porcelain=v1"])],
    # Scalar's recommended local-performance configuration (git scalar.c
    # set_recommended_config), restricted to keys that affect local
    # operations. Commit-graph / multi-pack-index are enabled for reads; the
    # files themselves are produced by Scalar's background maintenance, which
    # is not applicable to the core lane's 1-3 commit histories.
    "scalar": [
        ("scalar_fsmonitor", ["config", "core.fsmonitor", "true"]),
        ("scalar_untracked_cache", ["config", "core.untrackedCache", "true"]),
        ("scalar_preload_index", ["config", "core.preloadIndex", "true"]),
        ("scalar_index_threads", ["config", "index.threads", "true"]),
        ("scalar_index_version", ["config", "index.version", "4"]),
        ("scalar_multi_pack_index", ["config", "core.multiPackIndex", "true"]),
        ("scalar_commit_graph", ["config", "core.commitGraph", "true"]),
        ("scalar_status_ahead_behind", ["config", "status.aheadBehind", "false"]),
        ("scalar_gc_auto", ["config", "gc.auto", "0"]),
        # Start and warm the fsmonitor daemon and untracked cache OUTSIDE any
        # timed operation (recorded mode.setup rows, excluded from averages),
        # so timed status/diff see a warm tuned Git, not daemon startup.
        ("scalar_fsmonitor_start", ["fsmonitor--daemon", "start"]),
        ("scalar_warm_status", ["status", "--porcelain=v1", "--untracked-files=all"]),
    ],
    "lfs": [
        ("lfs_install", ["lfs", "install", "--local"]),
        ("lfs_track", ["lfs", "track", *LFS_TRACK_PATTERNS]),
    ],
}
FSMONITOR_MODES = frozenset({"fsmonitor", "scalar"})


def base_env() -> dict[str, str]:
    return oakbench_environment.base_env(
        author_name="Oak Bench",
        author_email="[email protected]",
        oak_author="oakbench",
    )


def probe_env() -> dict[str, str]:
    env = base_env()
    env.pop("NO_COLOR", None)
    env.pop("CLICOLOR", None)
    return env


def subject_provenance(subject: Subject, metadata: dict[str, Any]) -> dict[str, Any]:
    """Per-row binary identity for the subject that executed the command."""
    details = metadata.get("subject_details")
    digest: str | None = None
    version: str | None = None
    if isinstance(details, dict):
        item = details.get(subject.name)
        if isinstance(item, dict):
            value = item.get("binary_sha256")
            digest = value if isinstance(value, str) else None
            version = item.get("version") if isinstance(item.get("version"), str) else None
    fields = {"binary_sha256": digest or binary_sha256(subject.bin),
              "subject_bin": str(subject.bin), "subject_version": version}
    if subject.kind == "git":
        fields.update(git_shim_measured=subject.git_shim,
                      git_requested_bin=str(subject.requested_bin) if subject.requested_bin else None,
                      git_config_profile=git_config_profile(subject.git_mode))
    return fields


@dataclass(frozen=True)
class Scenario:
    name: str
    file_count: int
    file_size: int
    dirty_count: int
    binary: bool
    runs: int


PROFILES: dict[str, list[Scenario]] = {
    # micro: noise-floor probes (devloop A/A), integration tests, and the
    # fastest possible regression signal. Same tiny_text shape as smoke
    # (identical scenario name = identical meaning, ADR-0005), more reps.
    "micro": [
        Scenario("tiny_text", 24, 1024, 4, False, 3),
    ],
    "smoke": [
        Scenario("tiny_text", 24, 1024, 4, False, 1),
        Scenario("single_medium_binary", 1, 8 * 1024 * 1024, 1, True, 1),
        Scenario("few_large_binaries", 4, 4 * 1024 * 1024, 2, True, 1),
    ],
    "standard": [
        # NOT tiny_text: 100 files is a different shape, and a different shape
        # must be a different scenario name (ADR-0005).
        Scenario("tiny_text_100", 100, 1024, 12, False, 3),
        Scenario("many_small_files", 5_000, 512, 500, False, 3),
        Scenario("wide_dirty_tree", 2_000, 1024, 1_000, False, 3),
        Scenario("single_large_binary", 1, 128 * 1024 * 1024, 1, True, 3),
        Scenario("many_large_binaries", 32, 8 * 1024 * 1024, 8, True, 3),
    ],
    "large": [
        Scenario("many_small_files_50k", 50_000, 256, 5_000, False, 5),
        Scenario("single_huge_binary_1gb", 1, 1024 * 1024 * 1024, 1, True, 5),
        Scenario("many_large_binaries_multigb", 128, 32 * 1024 * 1024, 32, True, 5),
    ],
}
GIT_HUGE_BINARY_DIFF_LIMIT_BYTES = 1024 * 1024 * 1024
GIT_HUGE_BINARY_DIFF_OPS = {
    "diff.dirty",
    "diff.dirty.determinism",
    "diff.dirty.inforecall",
    "diff.full.inforecall",
}


def parse_args() -> argparse.Namespace:
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument("--profile", choices=sorted(PROFILES), default="smoke")
    parser.add_argument("--scenarios", default="",
                        help="Comma list restricting the profile to these scenario names (devloop confirmation runs).")
    parser.add_argument("--subjects", help="Comma-separated subject names")
    parser.add_argument("--config", type=Path, default=ROOT / "config" / "subjects.toml")
    parser.add_argument("--workdir", type=Path, default=DEFAULT_WORKDIR)
    parser.add_argument("--results", type=Path, default=ROOT / "results")
    parser.add_argument(
        "--oak-repo",
        type=Path,
        default=DEFAULT_OAK_REPO,
        help="Optional Oak source checkout for provenance metadata. Defaults to $OAK_REPO or ../oak.",
    )
    parser.add_argument("--keep-workdirs", action="store_true")
    parser.add_argument("--git-bin", type=Path)
    parser.add_argument("--oak-installed-bin", type=Path)
    parser.add_argument("--oak-local-bin", type=Path)
    parser.add_argument("--oak-aa-bin", type=Path, help="A/A control: the baseline Oak binary under subject oak_installed_aa.")
    parser.add_argument("--runs", type=int, help="Override scenario run count")
    parser.add_argument("--skip-diff", action="store_true", help="Skip dirty diff timing")
    parser.add_argument(
        "--track",
        choices=TRACKS,
        default="agent-default",
        help=(
            "agent-default measures commands an agent would likely call today; "
            "core-equivalent prefers matching semantic output levels across VCSs."
        ),
    )
    parser.add_argument(
        "--randomize-subject-order",
        action="store_true",
        help="Shuffle subject order per scenario run to reduce cache/order bias.",
    )
    parser.add_argument(
        "--admitted-output-chars",
        type=int,
        default=DEFAULT_ADMITTED_OUTPUT_CHARS,
        help=(
            "Per-command stdout/stderr characters counted as agent-visible output "
            "for token estimates. Full output is not stored in JSONL."
        ),
    )
    parser.add_argument(
        "--git-modes",
        default="",
        help=(
            "Comma-separated tuned Git modes to run as extra subjects: "
            + ", ".join(sorted(WIRED_GIT_MODES))
            + ". Mode setup runs untimed after repo.init. Required for credible "
            "Oak-vs-Git status/diff claims; stock Git alone overstates Oak wins."
        ),
    )
    parser.add_argument(
        "--allow-git-shim",
        action="store_true",
        help=(
            "Measure the macOS /usr/bin/git xcrun trampoline as-is (rows say git_shim_measured). "
            "By default the shim is resolved to the real Git binary, or the run is refused."
        ),
    )
    parser.add_argument(
        "--cache-mode",
        choices=CACHE_MODES,
        default="warm",
        help=(
            "warm (default): no purge; rows are labelled page_cache_state=warm. cold: purge the OS "
            "page cache immediately before every timed command (after harness reads). macOS needs "
            "root for purge(8) (sudo -n with a NOPASSWD entry); Linux needs write access to "
            "/proc/sys/vm/drop_caches. A failed preflight purge refuses the run."
        ),
    )
    parser.add_argument(
        "--skip-determinism-probe",
        action="store_true",
        help="Skip the output byte-stability and ANSI-pollution probe rows.",
    )
    parser.add_argument(
        "--skip-remote",
        action="store_true",
        help=(
            "Skip remote.push.first / remote.clone.cold / remote.pull.uptodate rows. "
            "Clone duplicates the repo on disk; consider skipping for the large profile."
        ),
    )
    parser.add_argument(
        "--load-tier",
        choices=sorted(oakbench_loadgen.LOAD_TIERS),
        default="none",
        help=(
            "Inject a machine-independent stress-ng background load tier while "
            "measuring (noisy-neighbor realism). Requires the optional stress-ng "
            "binary; when it is missing under a non-none tier every row becomes "
            "an explicit skip row — load that was not applied is never claimed."
        ),
    )
    parser.add_argument(
        "--dirty-spectrum",
        action="store_true",
        help=(
            "Additionally measure status.dirty.p01/.p10/.p50 (and diff.dirty.pNN unless "
            "--skip-diff): pNN dirties NN%% of tracked files (deterministic seeded "
            "selection, fixed appended line, untimed) before the timed command, then "
            "restores the tree byte-identically. Off by default; without this flag the "
            "operation mix and every existing row are unchanged."
        ),
    )
    return parser.parse_args()


def expand_git_modes(subjects: list[Subject], raw_modes: str) -> list[Subject]:
    requested = [item.strip() for item in raw_modes.split(",") if item.strip()]
    if not requested:
        return subjects
    normalized = [mode[len("git_") :] if mode.startswith("git_") else mode for mode in requested]
    unknown = [mode for mode in normalized if mode not in WIRED_GIT_MODES]
    if unknown:
        raise SystemExit(
            "Unknown or unwired git modes: "
            + ", ".join(unknown)
            + ". Wired modes: "
            + ", ".join(sorted(WIRED_GIT_MODES))
            + ". Parameterized modes (sparse-checkout, worktree) live in config/git_modes.json "
            "and need fixture-specific wiring."
        )
    base_git = next((subject for subject in subjects if subject.kind == "git" and subject.git_mode is None
                     and subject.name != "git_aa"), None)
    if base_git is None:
        raise SystemExit("--git-modes requires a stock git subject to derive tuned variants from")
    tuned = [
        Subject(
            name=f"{base_git.name}_{mode}",
            kind="git",
            label=f"{base_git.label} ({mode.replace('_', '-')})",
            bin=base_git.bin,
            git_mode=mode,
            requested_bin=base_git.requested_bin,
            git_shim=base_git.git_shim,
        )
        for mode in normalized
    ]
    return subjects + tuned


def git_lfs_available(git_bin: Path) -> bool:
    try:
        proc = subprocess.run(
            [str(git_bin), "lfs", "version"],
            env=base_env(),
            stdout=subprocess.DEVNULL,
            stderr=subprocess.DEVNULL,
            timeout=10,
            check=False,
        )
    except (OSError, subprocess.TimeoutExpired):
        return False
    return proc.returncode == 0


@functools.lru_cache(maxsize=1)
def git_modes_config() -> dict[str, dict[str, Any]]:
    """config/git_modes.json modes table — the single source of truth for
    where a tuned mode applies (and for its fairness notes)."""
    return json.loads(GIT_MODES_CONFIG.read_text())["modes"]


@functools.lru_cache(maxsize=1)
def known_scenario_names() -> frozenset[str]:
    return frozenset(scenario.name for profile in PROFILES.values() for scenario in profile)


def git_mode_applies_to_scenario(mode: str | None, scenario: Scenario) -> tuple[bool, str | None]:
    """Decide from config/git_modes.json whether a tuned mode measures this scenario.

    A mode's applies_to list is enforced when its entries name concrete
    scenarios (the git_lfs list does): scenarios outside the list emit
    scenario.mode_skipped rows instead of running under the tuned label.
    Some modes' applies_to entries are fixture-class tags ("large-tree",
    "huge-tree", ...) that name no scenario; until those tags are mapped to
    scenario names they keep the legacy applies-everywhere behavior. The
    binary-flag check stays as the fallback for an lfs mode entry without an
    applies_to list.
    """
    if not mode:
        return True, None
    spec = git_modes_config().get(f"git_{mode}", {})
    applies_to = spec.get("applies_to") or []
    if "all" in applies_to:
        return True, None
    if applies_to and (scenario.name in applies_to or known_scenario_names() & set(applies_to)):
        if scenario.name in applies_to:
            return True, None
        return False, (
            f"git_{mode}_not_applicable: scenario {scenario.name} is not in the "
            f"config/git_modes.json applies_to list for git_{mode}"
        )
    if mode == "lfs" and not scenario.binary:
        return False, "git_lfs_not_applicable: scenario has no binary fixture files"
    return True, None


def lfs_untracked_extensions(repo: Path, patterns: tuple[str, ...] = LFS_TRACK_PATTERNS) -> list[str]:
    """Extensions of fixture payload files no LFS track pattern covers.

    Called only for binary scenarios, where every generated payload file is
    binary content (ensure_fixture writes a single extension per scenario
    today, so with *.bin tracked this passes; the guard exists for future
    fixtures that add other extensions). Sorted so the skip reason is stable.
    """
    missing: set[str] = set()
    for path in repo.rglob("*"):
        if not path.is_file():
            continue
        rel = path.relative_to(repo)
        if rel.parts[0] == ".git" or rel.name == ".gitattributes":
            continue
        if any(fnmatch.fnmatch(rel.name, pattern) for pattern in patterns):
            continue
        missing.add(rel.suffix.lstrip(".") or rel.name)
    return sorted(missing)


def interaction_metrics(
    command: Any,
    stdout_text: str,
    stderr_text: str,
    stdout_bytes: int,
    stderr_bytes: int,
    stdout_truncated: bool,
    stderr_truncated: bool,
    tool_call_count: int,
) -> dict[str, Any]:
    return {
        "tool_call_count": tool_call_count,
        "vcs_tool_call_count": tool_call_count,
        "terminal_tool_call_count": tool_call_count,
        **oakbench_tokens.interaction_token_fields(
            command_display(command),
            stdout_text,
            stderr_text,
            stdout_bytes,
            stderr_bytes,
            stdout_truncated,
            stderr_truncated,
            tool_call_count=tool_call_count,
        ),
        "tool_calls": {
            "total": tool_call_count,
            "vcs": tool_call_count,
            "terminal": tool_call_count,
        },
    }


def merge_interaction_metrics(command: Any, rows: list[dict[str, Any]]) -> dict[str, Any]:
    tool_calls = sum(oakbench_tokens.int_or_zero(row.get("tool_call_count")) for row in rows)
    return {
        "tool_call_count": tool_calls,
        "vcs_tool_call_count": tool_calls,
        "terminal_tool_call_count": tool_calls,
        **oakbench_tokens.summed_token_fields(
            rows,
            "sum_of_subcommands_command_plus_admitted_output_chars_div_4",
            command_chars=len(command_display(command)),
        ),
        "tool_calls": {
            "total": tool_calls,
            "vcs": tool_calls,
            "terminal": tool_calls,
        },
    }


def hash_and_count_ansi(path_handle: Any) -> tuple[str, int, int]:
    path_handle.flush()
    path_handle.seek(0)
    digest = hashlib.sha256()
    ansi_bytes = 0
    total = 0
    for chunk in iter(lambda: path_handle.read(1024 * 1024), b""):
        digest.update(chunk)
        ansi_bytes += chunk.count(b"\x1b")
        total += len(chunk)
    return digest.hexdigest(), ansi_bytes, total


def tail_text_from_handle(path_handle: Any, max_bytes: int = 4000) -> str:
    path_handle.flush()
    path_handle.seek(0, os.SEEK_END)
    size = path_handle.tell()
    path_handle.seek(max(0, size - max_bytes))
    return path_handle.read(max_bytes).decode("utf-8", "replace")


def run_determinism_probe(
    subject: Subject,
    scenario: Scenario,
    base_op: str,
    cwd: Path,
    command: list[str],
    run_index: int,
    metadata: dict[str, Any],
) -> dict[str, Any]:
    """Run the same command twice on identical state.

    Byte-identical output for identical state is what makes provider prompt
    caches hit; instability (timestamps, ordering, progress output) silently
    raises token cost for every downstream agent.
    """
    from oakbench.conformance import tree_manifest
    before_manifest = tree_manifest(cwd) if base_op.startswith("diff.") else None
    mutation_fields = {}
    hashes: list[str] = []
    ansi_total = 0
    output_bytes = 0
    elapsed: list[float] = []
    returncode = 0
    stdout_excerpt = ""
    stderr_excerpt = ""
    for _ in range(2):
        start = time.perf_counter()
        with tempfile.TemporaryFile() as stdout_fh, tempfile.TemporaryFile() as stderr_fh:
            proc = subprocess.run(
                command,
                cwd=cwd,
                env=probe_env(),
                stdout=stdout_fh,
                stderr=stderr_fh,
                check=False,
            )
            elapsed.append((time.perf_counter() - start) * 1000)
            stdout_hash, stdout_ansi, stdout_total = hash_and_count_ansi(stdout_fh)
            stderr_hash, stderr_ansi, stderr_total = hash_and_count_ansi(stderr_fh)
            if proc.returncode != 0:
                stdout_excerpt = tail_text_from_handle(stdout_fh)
                stderr_excerpt = tail_text_from_handle(stderr_fh)
        hashes.append(stdout_hash + ":" + stderr_hash)
        ansi_total += stdout_ansi + stderr_ansi
        output_bytes += stdout_total + stderr_total
        returncode = proc.returncode
        if before_manifest is not None:
            mutation_fields = oakbench_semantics.finalize_patch_evidence(
                {"process_returncode": returncode}, before_manifest, tree_manifest(cwd), application_required=False)
            if mutation_fields.get("integrity_ok") is False:
                break
        if returncode != 0:
            break
    row = {
        **metadata,
        **subject_provenance(subject, metadata),
        "subject": subject.name,
        "subject_kind": subject.kind,
        "subject_label": subject.label,
        "git_mode": subject.git_mode,
        "scenario": scenario.name,
        "run": run_index,
        "operation": f"{base_op}.determinism",
        "elapsed_ms": None,
        "measurement_phase": "output_diagnostic",
        "cache_state": "oracle_warmed_diagnostic",
        "returncode": returncode,
        "command": command,
        "output_stable": hashes[0] == hashes[1] if len(hashes) == 2 else None,
        "ansi_escape_count_non_tty": ansi_total,
        "probe_output_bytes": output_bytes,
        "tool_call_count": 0,
        "vcs_tool_call_count": 0,
        "terminal_tool_call_count": 0,
        "notes": "diagnostic probe row; excluded from token/tool-call deltas (tool_call_count=0)",
    }
    if returncode != 0:
        row["stdout"] = stdout_excerpt
        row["stderr"] = stderr_excerpt
    row.update(mutation_fields)
    return row


# Cap on probe output read into memory. Truncated output has unmeasured
# recall/precision; a partial path set is never a complete output oracle.
INFO_PROBE_MAX_CHARS = 16 * 1024 * 1024
PATCH_PROBE_MAX_BYTES = 16 * 1024 * 1024


def run_information_probe(
    subject: Subject,
    scenario: Scenario,
    base_op: str,
    cwd: Path,
    command: list[str],
    run_index: int,
    metadata: dict[str, Any],
    changed_paths: list[str],
    pristine_fixture: Path | None = None,
) -> dict[str, Any]:
    """Score one status/diff output against the ground-truth changed set.

    Bytes saved are only a win if the file list is still recoverable from the
    output; this row makes "compact but lossy" visible as information_recall
    < 1.0 instead of looking like a token improvement. diff rows additionally
    record unified-diff structure for piped-tooling compatibility.
    """
    from oakbench.conformance import tree_manifest
    before_manifest = tree_manifest(cwd) if base_op.startswith("diff.") else None
    expected_manifest = None
    patch_skip_reason = None
    if base_op == "diff.full" and pristine_fixture is not None:
        if scenario.file_count * scenario.file_size > PATCH_PROBE_MAX_BYTES:
            patch_skip_reason = "patch_fixture_exceeds_16MiB_limit"
        elif subject.git_mode == "lfs":
            patch_skip_reason = "lfs_filters_require_separate_patch_contract"
        else:
            from oakbench.conformance import tree_manifest
            expected_manifest = before_manifest
    capture = run_timed(command, cwd, base_env(), INFO_PROBE_MAX_CHARS,
                        raw_output_bytes=PATCH_PROBE_MAX_BYTES if expected_manifest is not None else None)
    # stderr may contain arbitrary diagnostics mentioning a path; it is not a
    # path-bearing stdout protocol and must never inflate recall or precision.
    output_text = capture.stdout_text
    output_format = semantics().output_format(subject.kind, base_op,
                                             metadata.get("benchmark_track", "agent-default"))
    row = {
        **metadata,
        **subject_provenance(subject, metadata),
        "subject": subject.name,
        "subject_kind": subject.kind,
        "subject_label": subject.label,
        "git_mode": subject.git_mode,
        "scenario": scenario.name,
        "run": run_index,
        "operation": f"{base_op}.inforecall",
        "elapsed_ms": None,
        "measurement_phase": "output_diagnostic",
        "cache_state": "oracle_warmed_diagnostic",
        "returncode": capture.returncode,
        "command": command,
        "probe_output_bytes": capture.stdout_bytes + capture.stderr_bytes,
        "output_format": output_format,
        **oakbench_semantics.information_recall_fields(
            output_text,
            changed_paths,
            capture.stdout_bytes + capture.stderr_bytes,
            capture.stdout_truncated or capture.stderr_truncated,
            output_format,
        ),
        "tool_call_count": 0,
        "vcs_tool_call_count": 0,
        "terminal_tool_call_count": 0,
        "notes": "diagnostic probe row; excluded from token/tool-call deltas (tool_call_count=0)",
    }
    if base_op.startswith("diff."):
        row.update(oakbench_semantics.unified_diff_compat_fields(capture.stdout_text))
    if base_op == "diff.full":
        row.update(patch_oracle_version=oakbench_semantics.PATCH_ORACLE_VERSION,
                   patch_apply_ok=None, patch_tree_matches=None,
                   patch_application_required=not scenario.binary,
                   patch_apply_source="git_apply_exact_tree")
        if capture.returncode != 0:
            row["patch_skip_reason"] = "subject_diff_command_failed"
        elif expected_manifest is None:
            row["patch_skip_reason"] = patch_skip_reason or "pristine_fixture_unavailable"
        elif capture.stdout_raw is None:
            row["patch_skip_reason"] = "patch_output_exceeds_capture_limit"
        else:
            row.update(oakbench_semantics.patch_application_fields(
                capture.stdout_raw, pristine_fixture, expected_manifest, cwd.parent,
                pristine_excluded_root_paths=(".fixture-ready",),
            ))
    if before_manifest is not None:
        row["process_returncode"] = capture.returncode
        row = oakbench_semantics.finalize_patch_evidence(
            row, before_manifest, tree_manifest(cwd), application_required=base_op == "diff.full" and not scenario.binary)
    if capture.returncode != 0:
        row["stderr"] = capture.stderr_text[-4000:]
    return row


REMOTE_OPS = ("remote.push.first", "remote.clone.cold", "remote.pull.uptodate")
# Oak remote ops are network-transport (a real Oak server) while git's are
# local-file; different transports are different measurements, so the names
# differ (ADR-0005) and no report ever computes a git-vs-oak delta between
# them. The oak ops answer the release-blocking trend: did this changeset
# make oak's real push/clone/sync slower?
OAK_NET_REMOTE_OPS = ("remote.net.push.first", "remote.net.clone.cold", "remote.net.fetch.uptodate")
OAK_REMOTE_ENV = oakbench_remotes.OAK_REMOTE_ENV


def apply_state_change_attestation(
    row: dict[str, Any],
    repo: Path,
    before_fingerprint: str | None,
    op: str,
    *,
    expected_hashes: dict[str, str] | None = None,
    expected_payloads: dict[str, str | bytes] | None = None,
) -> dict[str, Any]:
    process_returncode = row_returncode(row)
    fields = oakbench_integrity.state_change_fields(
        repo,
        before_fingerprint,
        operation=op,
        process_returncode=process_returncode,
    )
    row.setdefault("process_returncode", process_returncode)
    row.update(fields)
    failure_reason = fields["integrity_failure_reason"] if not fields["integrity_check_passed"] else None
    if expected_hashes or expected_payloads:
        content_fields = oakbench_integrity.content_integrity_fields(
            repo,
            operation=op,
            process_returncode=process_returncode,
            expected_hashes=expected_hashes,
            expected_payloads=expected_payloads,
        )
        row.update(content_fields)
        # None is unmeasured (reader unavailable), never a failure or a pass.
        if content_fields["content_integrity_check_passed"] is False and failure_reason is None:
            failure_reason = content_fields["content_integrity_failure_reason"]
    if process_returncode == 0 and failure_reason is not None:
        row["returncode"] = 1
        row["stderr"] = str(failure_reason or "integrity_check_failed")
    return row


def remote_skip_row(
    subject: Subject,
    scenario: Scenario,
    op: str,
    run_index: int,
    metadata: dict[str, Any],
    reason: str,
) -> dict[str, Any]:
    return {
        **metadata,
        **subject_provenance(subject, metadata),
        "subject": subject.name,
        "subject_kind": subject.kind,
        "subject_label": subject.label,
        "git_mode": subject.git_mode,
        "scenario": scenario.name,
        "run": run_index,
        "operation": op,
        "elapsed_ms": 0.0,
        "returncode": 77,
        "command": [],
        "skipped": True,
        "skip_reason": reason,
        "tool_call_count": 0,
        "vcs_tool_call_count": 0,
        "terminal_tool_call_count": 0,
    }


def git_huge_binary_diff_limit_reason(subject: Subject, scenario: Scenario, op: str) -> str | None:
    if subject.kind != "git":
        return None
    if subject.git_mode == "lfs":
        return None
    if op not in GIT_HUGE_BINARY_DIFF_OPS:
        return None
    if not scenario.binary or scenario.file_size < GIT_HUGE_BINARY_DIFF_LIMIT_BYTES:
        return None
    return (
        "git_huge_binary_diff_fixture_limit: stock/tuned git diff rows are intentionally "
        "skipped for generated >=1GiB binary fixtures because git may fail while generating "
        "binary diffstat; use git_lfs or smaller binary fixtures for this capability"
    )


def git_huge_binary_diff_skip_row(
    subject: Subject,
    scenario: Scenario,
    op: str,
    command: list[str],
    run_index: int,
    metadata: dict[str, Any],
) -> dict[str, Any]:
    reason = git_huge_binary_diff_limit_reason(subject, scenario, op)
    if reason is None:
        raise ValueError(f"{op} is not a huge-binary git diff skip")
    row = remote_skip_row(subject, scenario, op, run_index, metadata, reason)
    row.update(
        {
            "command": command,
            "failure_classification": "git_huge_binary_diff_fixture_limit",
            "notes": (
                "preflighted fixture-limit skip; no subprocess was run for this row, "
                "so stderr is intentionally empty"
            ),
        }
    )
    return row


_OAK_JSON_SUPPORT_CACHE: dict[tuple[str, tuple[str, ...], tuple[str, ...]], str | None] = {}
_OAK_VERSION_CACHE: dict[str, tuple[int, int, int] | None] = {}


def parse_semver_triplet(text: str) -> tuple[int, int, int] | None:
    match = re.search(r"\b(\d+)\.(\d+)\.(\d+)\b", text)
    if match is None:
        return None
    return (int(match.group(1)), int(match.group(2)), int(match.group(3)))


def oak_cli_semver(vcs: str) -> tuple[int, int, int] | None:
    if vcs in _OAK_VERSION_CACHE:
        return _OAK_VERSION_CACHE[vcs]
    try:
        probe = subprocess.run(
            [vcs, "--version"],
            capture_output=True,
            text=True,
            env=base_env(),
            timeout=15,
            check=False,
        )
    except (FileNotFoundError, subprocess.TimeoutExpired):
        _OAK_VERSION_CACHE[vcs] = None
        return None
    version = parse_semver_triplet((probe.stdout or "") + "\n" + (probe.stderr or ""))
    _OAK_VERSION_CACHE[vcs] = version
    return version


def oak_json_probe_version_skip_reason(vcs: str, spec: dict[str, Any]) -> str | None:
    minimum = spec.get("minimum_oak_version")
    if not minimum:
        return None
    required = parse_semver_triplet(str(minimum))
    actual = oak_cli_semver(vcs)
    if required is None:
        return f"capability_gap:oak_json_probe_bad_minimum_version:{minimum}"
    if actual is None:
        return f"capability_gap:oak_json_probe_version_unknown:requires_oak>={minimum}"
    if actual < required:
        actual_text = ".".join(str(part) for part in actual)
        return f"capability_gap:oak_json_probe_requires_oak>={minimum}:found_{actual_text}"
    return None


def oak_json_probe_skip_reason(vcs: str, help_args: list[str], command_args: list[str]) -> str | None:
    """Return a capability skip reason, or None when the JSON command is present."""
    key = (vcs, tuple(help_args), tuple(command_args))
    if key in _OAK_JSON_SUPPORT_CACHE:
        return _OAK_JSON_SUPPORT_CACHE[key]
    if not help_args:
        reason = "capability_gap:oak_json_probe_help_missing"
        _OAK_JSON_SUPPORT_CACHE[key] = reason
        return reason
    try:
        probe = subprocess.run(
            [vcs, *help_args],
            capture_output=True,
            text=True,
            env=base_env(),
            timeout=15,
            check=False,
        )
    except FileNotFoundError:
        reason = f"missing_binary:{vcs}"
        _OAK_JSON_SUPPORT_CACHE[key] = reason
        return reason
    except subprocess.TimeoutExpired:
        reason = "capability_gap:oak_json_probe_help_timeout:" + " ".join(help_args)
        _OAK_JSON_SUPPORT_CACHE[key] = reason
        return reason
    help_text = (probe.stdout or "") + "\n" + (probe.stderr or "")
    if probe.returncode != 0:
        reason = "capability_gap:oak_json_probe_help_failed:" + " ".join(help_args)
    elif "--json" in command_args and "--json" not in help_text:
        reason = "capability_gap:oak_json_flag_missing:" + " ".join(command_args)
    else:
        reason = None
    _OAK_JSON_SUPPORT_CACHE[key] = reason
    return reason


def json_field_oracle_fields(stdout_text: str, required_fields: list[str]) -> dict[str, Any]:
    """JSON parse and top-level required-field completeness for CLI capability rows."""
    base: dict[str, Any] = {
        "json_required_fields": required_fields,
        "json_required_field_count": len(required_fields),
    }
    try:
        payload = json.loads(stdout_text)
    except json.JSONDecodeError as exc:
        return {
            **base,
            "json_parse_success": False,
            "json_payload_type": None,
            "json_present_required_field_count": 0,
            "json_missing_required_fields": required_fields,
            "json_field_completeness": 0.0 if required_fields else None,
            "json_oracle_passed": False,
            "json_error": f"{exc.msg} at line {exc.lineno} column {exc.colno}",
        }
    payload_type = type(payload).__name__
    if not isinstance(payload, dict):
        return {
            **base,
            "json_parse_success": True,
            "json_payload_type": payload_type,
            "json_present_required_field_count": 0,
            "json_missing_required_fields": required_fields,
            "json_field_completeness": 0.0 if required_fields else None,
            "json_oracle_passed": False,
            "json_error": "expected top-level JSON object",
        }
    missing = [field for field in required_fields if field not in payload]
    present = len(required_fields) - len(missing)
    return {
        **base,
        "json_parse_success": True,
        "json_payload_type": payload_type,
        "json_present_required_field_count": present,
        "json_missing_required_fields": missing,
        "json_field_completeness": round(present / len(required_fields), 4) if required_fields else None,
        "json_oracle_passed": not missing,
    }


def json_field_oracle_unmeasured_fields(required_fields: list[str], reason: str) -> dict[str, Any]:
    """Null JSON validation fields when the runner did not capture a complete payload."""
    return {
        "json_required_fields": required_fields,
        "json_required_field_count": len(required_fields),
        "json_parse_success": None,
        "json_payload_type": None,
        "json_present_required_field_count": None,
        "json_missing_required_fields": None,
        "json_field_completeness": None,
        "json_oracle_passed": None,
        "json_validation_unmeasured_reason": reason,
    }


def run_agent_native_json_probe(
    subject: Subject,
    scenario: Scenario,
    op: str,
    cwd: Path,
    spec: dict[str, Any],
    run_index: int,
    metadata: dict[str, Any],
    admitted_output_chars: int,
) -> dict[str, Any]:
    vcs = str(subject.bin)
    command_args = list(spec["args"])
    help_args = list(spec["help_args"])
    command = [vcs, *command_args]
    skip_reason = oak_json_probe_version_skip_reason(vcs, spec)
    if skip_reason is None:
        skip_reason = oak_json_probe_skip_reason(vcs, help_args, command_args)
    if skip_reason is not None:
        row = remote_skip_row(subject, scenario, op, run_index, metadata, skip_reason)
        row["intended_command"] = command
        row["capability_probe_command"] = [vcs, *help_args]
        if spec.get("minimum_oak_version") is not None:
            row["minimum_oak_version"] = str(spec["minimum_oak_version"])
        return row

    from oakbench.conformance import tree_manifest
    before_manifest = tree_manifest(cwd) if op == "oak.branch.diff.json" else None
    cache = page_cache_step(metadata)
    capture = run_timed(
        command,
        cwd,
        base_env(),
        admitted_output_chars,
        full_output_bytes=JSON_PROBE_MAX_BYTES,
    )
    required_fields = list(spec["required_fields"])
    if capture.stdout_full_text is not None and capture.stdout_full_text_truncated is False:
        json_validation_source = "full_stdout_capture"
        json_fields = json_field_oracle_fields(capture.stdout_full_text, required_fields)
    elif capture.stdout_full_text_truncated is True or (
        capture.stdout_full_text is None and capture.stdout_truncated
    ):
        json_validation_source = None
        json_fields = json_field_oracle_unmeasured_fields(
            required_fields,
            "stdout_exceeded_json_validation_capture_limit",
        )
    else:
        json_validation_source = "admitted_stdout_capture"
        json_fields = json_field_oracle_fields(capture.stdout_text, required_fields)
    result = {
        **metadata,
        **cache,
        **subject_provenance(subject, metadata),
        "subject": subject.name,
        "subject_kind": subject.kind,
        "subject_label": subject.label,
        "git_mode": subject.git_mode,
        "scenario": scenario.name,
        "run": run_index,
        "operation": op,
        "elapsed_ms": round(capture.elapsed_ms, 3),
        # 77 is the HARNESS's skip sentinel; a subject exiting 77 is a subject
        # failure (QA P1 A7), recorded with its real code in process_returncode.
        "returncode": subject_returncode(capture.returncode),
        "process_returncode": capture.returncode,
        "command": command,
        "peak_rss_bytes": capture.peak_rss_bytes,
        "peak_rss_source": PEAK_RSS_SOURCE if capture.peak_rss_bytes is not None else None,
        "cpu_user_ms": capture.cpu_user_ms,
        "cpu_sys_ms": capture.cpu_sys_ms,
        "io_read_blocks": capture.io_read_blocks,
        "io_write_blocks": capture.io_write_blocks,
        "major_page_faults": capture.major_page_faults,
        "rusage_source": RUSAGE_SOURCE if capture.cpu_user_ms is not None else None,
        "measurement_source": "direct_cli_timed_subprocess_json_capability_probe",
        "probe_output_bytes": capture.stdout_bytes + capture.stderr_bytes,
        "output_format": spec.get("output_format", "json"),
        **interaction_metrics(
            command,
            capture.stdout_text,
            capture.stderr_text,
            capture.stdout_bytes,
            capture.stderr_bytes,
            capture.stdout_truncated,
            capture.stderr_truncated,
            tool_call_count=1,
        ),
        "json_validation_source": json_validation_source,
        **json_fields,
    }
    if capture.returncode != 0:
        result["stderr"] = capture.stderr_text[-4000:]
    elif result["json_oracle_passed"] is False:
        result["returncode"] = 1
        result["stderr"] = str(result.get("json_error") or "json_required_fields_missing")
    if before_manifest is not None:
        result.update(process_returncode=capture.returncode, measurement_phase="agent_native_probe",
                      cache_state="hash_warmed_native_diff")
        result = oakbench_semantics.finalize_patch_evidence(result, before_manifest, tree_manifest(cwd), application_required=False)
    return result


def run_agent_native_json_probes(
    subject: Subject,
    scenario: Scenario,
    cwd: Path,
    run_index: int,
    metadata: dict[str, Any],
    admitted_output_chars: int,
) -> list[dict[str, Any]]:
    probes = semantics().agent_native_json_probes(subject.kind, BENCH_TASK_BRANCH)
    rows = []
    for op, spec in probes.items():
        row = run_agent_native_json_probe(subject, scenario, op, cwd, spec, run_index, metadata, admitted_output_chars)
        rows.append(row)
        if row.get("integrity_ok") is False:
            break
    return rows


def load_tier_skip_row(
    subject: Subject,
    scenario: Scenario,
    run_index: int,
    metadata: dict[str, Any],
    reason: str,
    load_tier: str,
) -> dict[str, Any]:
    """Explicit skip row for a load tier the host cannot apply.

    A row may never claim load that was not applied: environment_suspect is
    None (suspicion machinery did not run, ADR-0002) and load_tier records the
    REQUESTED tier so the gap is visible, not hidden.
    """
    row = remote_skip_row(subject, scenario, "scenario.load_tier_skipped", run_index, metadata, reason)
    row["load_tier"] = load_tier
    row["environment_suspect"] = None
    return row


def load_tier_skip_rows(
    subjects: list[Subject],
    scenarios: list[Scenario],
    metadata: dict[str, Any],
    reason: str,
    load_tier: str,
) -> list[dict[str, Any]]:
    """One skip row per scenario x run x subject for an unappliable load tier."""
    return [
        load_tier_skip_row(subject, scenario, run_index, metadata, reason, load_tier)
        for scenario in scenarios
        for run_index in range(scenario.runs)
        for subject in subjects
    ]


def oak_network_branch_name(
    bench_id: str,
    scenario_name: str,
    subject_name: str,
    run_index: int,
    *,
    pid: int | None = None,
    token: str | None = None,
) -> str:
    """Disposable remote branch name: readable prefix plus per-run uniquifier."""
    return oakbench_remotes.disposable_branch(
        bench_id, scenario_name, subject_name, run_index, pid=pid, token=token
    )


def run_oak_network_remote_phase(
    subject: Subject,
    scenario: Scenario,
    repo: Path,
    run_index: int,
    metadata: dict[str, Any],
    admitted_output_chars: int,
    remote_repo: str,
) -> list[dict[str, Any]]:
    """Push to, clone from, and sync with a real Oak server.

    Network transport: latency includes the server round trip, so these rows
    are oak-vs-previous-oak trend evidence, never a git-vs-oak comparison
    (git's remote phase is local-file by construction).

    Same-named branches with unrelated histories conflict on the server, so
    each run pushes a unique bench branch (untimed setup); the disposable
    remote accumulates branches by design.
    """
    rows: list[dict[str, Any]] = []
    vcs = str(subject.bin)
    bench_id = str(metadata.get("bench_id", "bench"))
    branch_name = oak_network_branch_name(bench_id, scenario.name, subject.name, run_index)
    clone_dir = repo.parent / f"netclone-{subject.name}-{run_index}"
    # Untimed: a unique branch is a precondition (same-named unrelated
    # histories are rejected by the server), not the measurement.
    subprocess.run(
        [vcs, "switch", "-c", branch_name],
        cwd=repo,
        env=base_env(),
        stdout=subprocess.DEVNULL,
        stderr=subprocess.DEVNULL,
        check=False,
    )
    push_row = run_command(
        subject,
        scenario,
        "remote.net.push.first",
        repo,
        [vcs, "push", "--repo", remote_repo],
        run_index,
        metadata,
        admitted_output_chars,
    )
    rows.append({**push_row, "remote_repo": remote_repo, "remote_transport": "network", "cache_state": "cold_remote_branch"})
    clone_row = run_command(
        subject,
        scenario,
        "remote.net.clone.cold",
        repo.parent,
        [vcs, "clone", remote_repo, str(clone_dir)],
        run_index,
        metadata,
        admitted_output_chars,
    )
    rows.append(
        {
            **clone_row,
            "remote_repo": remote_repo,
            "remote_transport": "network",
            "cache_state": "cold_objects_network",
            "notes": "clones the disposable repo's default content, not this scenario's fixture; comparable only to other remote.net.clone.cold rows",
        }
    )
    if clone_row["returncode"] == 0:
        fetch_row = run_command(
            subject,
            scenario,
            "remote.net.fetch.uptodate",
            clone_dir,
            [vcs, "fetch"],
            run_index,
            metadata,
            admitted_output_chars,
        )
        rows.append({**fetch_row, "remote_repo": remote_repo, "remote_transport": "network", "cache_state": "warm"})
    return rows


def run_remote_phase(
    subject: Subject,
    scenario: Scenario,
    repo: Path,
    run_index: int,
    metadata: dict[str, Any],
    admitted_output_chars: int,
) -> list[dict[str, Any]]:
    """Push to, clone from, and sync with a local remote.

    Agents pay clone/pull/push constantly: every fresh workspace is a cold
    clone, every task start is a pull, every task end is a push. A local
    filesystem remote isolates the VCS object-transfer and checkout cost from
    network jitter; "cold" means no local object store, while the OS page
    cache stays warm (recorded in cache_state so rows never overclaim).

    Oak's remote model needs a configured server remote; until one is wired,
    Oak rows are explicit skips (returncode 77), never silently absent.
    """
    if subject.kind != "git":
        resolution = oakbench_remotes.resolve_oak_remote()
        remote_repo = resolution.repo or ""
        if not remote_repo:
            reason = (
                f"oak_remote_not_configured: set {OAK_REMOTE_ENV} to a DISPOSABLE ORG/REPO "
                "(e.g. oak/oak-benchmarks-tmp) to measure real-server push/clone/fetch"
            )
            return [
                remote_skip_row(subject, scenario, op, run_index, metadata, reason)
                for op in OAK_NET_REMOTE_OPS
            ]
        return run_oak_network_remote_phase(
            subject, scenario, repo, run_index, metadata, admitted_output_chars, remote_repo
        )

    rows: list[dict[str, Any]] = []
    vcs = str(subject.bin)
    # Untimed setup: the remote existing is a precondition, not the measurement.
    bare = oakbench_remotes.make_git_bare_remote(
        vcs, repo.parent, name=f"remote-{subject.name}-{run_index}.git"
    )
    remote_dir = Path(bare.repo)
    clone_dir = repo.parent / f"clone-{subject.name}-{run_index}"
    oakbench_remotes.attach_git_origin(vcs, repo, bare)
    push_row = run_command(
        subject,
        scenario,
        "remote.push.first",
        repo,
        [vcs, "push", "origin", "HEAD"],
        run_index,
        metadata,
        admitted_output_chars,
    )
    rows.append({**push_row, "cache_state": "cold_remote_warm_page_cache", "remote_transport": "local_file"})
    # Untimed: point the bare remote's HEAD at the pushed branch so the clone
    # checks out a working tree (precondition, not measurement).
    branch_name = oakbench_remotes.current_git_branch(vcs, repo)
    oakbench_remotes.point_git_bare_head(vcs, bare, branch_name)
    clone_row = run_command(
        subject,
        scenario,
        "remote.clone.cold",
        repo.parent,
        [vcs, "clone", str(remote_dir), str(clone_dir)],
        run_index,
        metadata,
        admitted_output_chars,
    )
    rows.append({**clone_row, "cache_state": "cold_objects_warm_page_cache", "remote_transport": "local_file"})
    if clone_row["returncode"] == 0:
        pull_row = run_command(
            subject,
            scenario,
            "remote.pull.uptodate",
            clone_dir,
            [vcs, "pull"],
            run_index,
            metadata,
            admitted_output_chars,
        )
        rows.append({**pull_row, "cache_state": "warm", "remote_transport": "local_file"})
    return rows


def subject_returncode(code: int) -> int:
    """Row returncode for a subject process: never the harness skip sentinel."""
    return 1 if code == 77 else code


def page_cache_step(metadata: dict[str, Any]) -> dict[str, Any]:
    """Cache state for the command about to be timed; purges in cold mode.

    Runs AFTER every harness read for this op (attestation fingerprints and
    payload hashes), so a cold row starts from a purged page cache. Warm rows
    say what warmed them. ``cache_state`` set by the caller (remote/replay
    phases) is preserved; ``page_cache_state`` is the always-present label.
    """
    mode = metadata.get("cache_mode_requested") or "warm"
    fields: dict[str, Any] = {"cache_mode_requested": mode}
    if mode == "cold":
        purge = oakbench_cachectl.purge_fs_caches()
        cold = oakbench_cachectl.cache_fields("cold", purge)
        fields.update(page_cache_state="purged" if cold["cache_state"] == "cold" else "unknown",
                      page_cache_reason=cold["cache_state_reason"])
        if "cache_state" not in metadata:
            fields.update(cache_state=cold["cache_state"], cache_state_reason=cold["cache_state_reason"])
        return fields
    fields.update(page_cache_state="warm",
                  page_cache_reason="no_purge: fixture copied and earlier ops/harness reads ran in this checkout")
    if "cache_state" not in metadata:
        fields.update(cache_state="warm", cache_state_reason="sequence_warm_no_purge")
    return fields


def run_command(
    subject: Subject,
    scenario: Scenario,
    op: str,
    cwd: Path,
    command: list[str],
    run_index: int,
    metadata: dict[str, Any],
    admitted_output_chars: int,
) -> dict[str, Any]:
    cache = page_cache_step(metadata)
    capture = run_timed(command, cwd, base_env(), admitted_output_chars)
    result = {
        **metadata,
        **cache,
        **subject_provenance(subject, metadata),
        "subject": subject.name,
        "subject_kind": subject.kind,
        "subject_label": subject.label,
        "git_mode": subject.git_mode,
        "scenario": scenario.name,
        "run": run_index,
        "operation": op,
        "elapsed_ms": round(capture.elapsed_ms, 3),
        # 77 is the HARNESS's skip sentinel; a subject exiting 77 is a subject
        # failure (QA P1 A7), recorded with its real code in process_returncode.
        "returncode": subject_returncode(capture.returncode),
        "process_returncode": capture.returncode,
        "command": command,
        "peak_rss_bytes": capture.peak_rss_bytes,
        "peak_rss_source": PEAK_RSS_SOURCE if capture.peak_rss_bytes is not None else None,
        "cpu_user_ms": capture.cpu_user_ms,
        "cpu_sys_ms": capture.cpu_sys_ms,
        "io_read_blocks": capture.io_read_blocks,
        "io_write_blocks": capture.io_write_blocks,
        "major_page_faults": capture.major_page_faults,
        "rusage_source": RUSAGE_SOURCE if capture.cpu_user_ms is not None else None,
        "measurement_source": "direct_cli_timed_subprocess",
        **interaction_metrics(
            command,
            capture.stdout_text,
            capture.stderr_text,
            capture.stdout_bytes,
            capture.stderr_bytes,
            capture.stdout_truncated,
            capture.stderr_truncated,
            tool_call_count=1,
        ),
    }
    if capture.returncode != 0:
        result["stderr"] = capture.stderr_text[-4000:]
    return result


def fixture_fingerprint(root: Path) -> dict[str, Any]:
    """Content identity of a generated fixture (git-sizer-style counts plus a
    digest over every path, size and content hash). Untimed; computed once per
    invocation, outside the measurement lock."""
    digest = hashlib.sha256()
    files = 0
    total = 0
    largest = 0
    max_depth = 0
    for path in sorted(root.rglob("*")):
        if not path.is_file() or path.name == ".fixture-ready":
            continue
        rel = path.relative_to(root).as_posix()
        size = path.stat().st_size
        files += 1
        total += size
        largest = max(largest, size)
        max_depth = max(max_depth, rel.count("/") + 1)
        digest.update(rel.encode("utf-8", "surrogateescape") + b"\0" + str(size).encode() + b"\0")
        with path.open("rb") as handle:
            for chunk in iter(lambda: handle.read(1024 * 1024), b""):
                digest.update(chunk)
    return {"fixture_sha256": digest.hexdigest(),
            "fixture_stats": {"files": files, "bytes": total, "largest_file_bytes": largest,
                              "max_path_depth": max_depth}}


def write_pattern_file(path: Path, size: int, seed: str, binary: bool) -> None:
    oakbench_fixtures.write_pattern_file(path, size, seed, binary=binary, prefix="oakbench")


def scenario_shape_id(scenario: Scenario) -> str:
    """Fixture identity is the byte shape, not the scenario label.

    Keying the cache on shape (and recording it in the ready marker) makes a
    name collision across profiles a cache miss instead of silent corruption:
    a smoke run can never poison a later standard run sharing the workdir.
    """
    kind = "bin" if scenario.binary else "txt"
    return f"{scenario.file_count}x{scenario.file_size}-{kind}"


def fixture_path(root: Path, scenario: Scenario) -> Path:
    return root / "fixtures" / f"{scenario.name}-{scenario_shape_id(scenario)}"


def ensure_fixture(root: Path, scenario: Scenario) -> Path:
    dest = fixture_path(root, scenario)
    marker = dest / ".fixture-ready"
    expected_marker = f"{scenario_shape_id(scenario)}\n"
    if marker.exists() and marker.read_text() == expected_marker:
        return dest
    if dest.exists():
        if marker.exists() and marker.read_text() == expected_marker:
            return dest
        shutil.rmtree(dest)

    temp = dest.parent / f"{dest.name}.building-{os.getpid()}"
    if temp.exists():
        shutil.rmtree(temp)
    temp.mkdir(parents=True)
    try:
        ext = "bin" if scenario.binary else "txt"
        width = max(4, len(str(scenario.file_count)))
        for index in range(scenario.file_count):
            subdir = temp / f"group-{index // 1000:03d}"
            path = subdir / f"file-{index:0{width}d}.{ext}"
            write_pattern_file(path, scenario.file_size, f"{scenario.name}:{index}", scenario.binary)
        (temp / ".fixture-ready").write_text(expected_marker)
        try:
            os.rename(temp, dest)
        except OSError:
            if marker.exists() and marker.read_text() == expected_marker:
                shutil.rmtree(temp)
                return dest
            got = marker.read_text().strip() if marker.exists() else "missing"
            raise RuntimeError(
                f"fixture {dest} exists with wrong shape marker "
                f"(expected {expected_marker.strip()!r}, got {got!r})"
            ) from None
        return dest
    except Exception:
        if temp.exists():
            shutil.rmtree(temp)
        raise


def mutate_files(repo: Path, scenario: Scenario, subject: Subject, run_index: int) -> list[str]:
    """Mutate the scenario's dirty set; returns repo-relative changed paths.

    Mutation bytes are identical across subjects within each run.
    The returned list is the ground truth the information-recall probe scores
    status/diff output against.
    """
    ext = "bin" if scenario.binary else "txt"
    paths = sorted(repo.glob(f"group-*/*.{ext}"))[: scenario.dirty_count]
    if not paths and scenario.file_count == 1:
        paths = sorted(repo.glob(f"**/*.{ext}"))[:1]
    for index, path in enumerate(paths):
        marker = f"\nmutated {scenario.name} {run_index} {index}\n".encode()
        if scenario.binary:
            with path.open("r+b") as fh:
                size = path.stat().st_size
                fh.seek(max(0, size // 2 - len(marker) // 2))
                fh.write(marker[: min(len(marker), max(1, size))])
        else:
            with path.open("ab") as fh:
                fh.write(marker)
    return [path.relative_to(repo).as_posix() for path in paths]


def copy_fixture(src: Path, dest: Path) -> None:
    oakbench_fixtures.copy_fixture(src, dest, ignore_marker=True)


# Dirty-tree spectrum (opt-in via --dirty-spectrum). pNN means NN% of the
# fixture's tracked files are dirtied (ceil) before the timed command; the
# dirtying and the restore are untimed setup/teardown. Scale is encoded in the
# operation name per ADR-0005: status.dirty.p01 is a different measurement
# than status.dirty and never shares its trend line. snapshot.dirty.pNN is
# deliberately NOT implemented: a snapshot commits, so restoring cleanliness
# would require history surgery or a re-copied fixture per step — not cheap,
# and it would change the repository state every later operation sees.
DIRTY_SPECTRUM_FRACTIONS: tuple[tuple[str, float], ...] = (
    ("p01", 0.01),
    ("p10", 0.10),
    ("p50", 0.50),
)
DIRTY_SPECTRUM_BASE_OPS: tuple[str, ...] = ("status.dirty", "diff.dirty")


def dirty_spectrum_operations(enabled: bool, skip_diff: bool = False) -> list[str]:
    """Operation names the spectrum adds. Empty when the flag is off."""
    if not enabled:
        return []
    names: list[str] = []
    for label, _fraction in DIRTY_SPECTRUM_FRACTIONS:
        names.append(f"status.dirty.{label}")
        if not skip_diff:
            names.append(f"diff.dirty.{label}")
    return names


def dirty_spectrum_marker(seed: str) -> bytes:
    """Fixed edit content for one spectrum step.

    Deterministic (a pure function of the seed) so restore_dirty_fraction can
    verify and strip exactly the bytes dirty_fraction appended.
    """
    return f"\ndirty-spectrum {seed}\n".encode("ascii")


def tracked_fixture_files(repo_path: Path) -> list[Path]:
    """Sorted tracked fixture files: everything outside VCS metadata dirs.

    After snapshot.initial every fixture file is tracked, so a filesystem walk
    is the subject-neutral tracked set (no VCS command, identical for git and
    oak subjects).
    """
    repo = Path(repo_path)
    return sorted(
        path
        for path in repo.rglob("*")
        if path.is_file() and not {".git", ".oak"}.intersection(path.relative_to(repo).parts)
    )


def dirty_fraction(repo_path: Path, fraction: float, seed: str) -> list[str]:
    """Deterministically dirty ceil(fraction * N) tracked files.

    Selection is a seeded sample over the sorted tracked-file list; the edit
    is one fixed appended line (same bytes for every selected file). Returns
    the repo-relative POSIX paths of the dirtied files. Untimed by
    construction: callers dirty first, then run_timed the measured command.
    """
    repo = Path(repo_path)
    files = tracked_fixture_files(repo)
    if not files or fraction <= 0:
        return []
    count = min(len(files), math.ceil(fraction * len(files)))
    chosen = sorted(random.Random(seed).sample(files, count))
    marker = dirty_spectrum_marker(seed)
    for path in chosen:
        with path.open("ab") as fh:
            fh.write(marker)
    return [path.relative_to(repo).as_posix() for path in chosen]


def restore_dirty_fraction(repo_path: Path, changed_paths: list[str], seed: str) -> None:
    """Undo dirty_fraction by stripping the exact appended marker bytes.

    Content returns byte-identical to the committed state, so the tree is
    clean again before the next spectrum step (and before the pre-existing
    operation sequence continues). Verification before truncation means a
    mismatch fails loudly instead of silently corrupting a fixture copy.
    """
    repo = Path(repo_path)
    marker = dirty_spectrum_marker(seed)
    for rel in changed_paths:
        path = repo / rel
        size = path.stat().st_size
        with path.open("r+b") as fh:
            if size < len(marker):
                raise RuntimeError(f"dirty-spectrum restore: {rel} is smaller than the marker")
            fh.seek(size - len(marker))
            if fh.read(len(marker)) != marker:
                raise RuntimeError(f"dirty-spectrum restore: {rel} does not end with the expected marker")
            fh.truncate(size - len(marker))


def run_dirty_spectrum(
    subject: Subject,
    scenario: Scenario,
    repo: Path,
    ops: dict[str, list[str]],
    run_index: int,
    metadata: dict[str, Any],
    skip_diff: bool,
    admitted_output_chars: int,
) -> list[dict[str, Any]]:
    """Measure status (and diff) across deterministic dirty fractions.

    Runs on a clean committed tree. Per fraction: dirty (untimed), time the
    same status/diff commands the existing status.dirty/diff.dirty rows use,
    then restore (untimed) so each step starts from the identical clean state.
    The seed is per scenario+fraction, never per subject or run, so every
    subject measures the same dirtied file set.
    """
    from oakbench.conformance import tree_manifest
    rows: list[dict[str, Any]] = []
    base_ops = [op for op in DIRTY_SPECTRUM_BASE_OPS if not (skip_diff and op == "diff.dirty")]
    for label, fraction in DIRTY_SPECTRUM_FRACTIONS:
        seed = f"dirty-spectrum:{scenario.name}:{label}"
        changed = dirty_fraction(repo, fraction, seed)
        mutated = False
        try:
            for base_op in base_ops:
                before = tree_manifest(repo) if base_op.startswith("diff.") else None
                row = run_command(
                    subject,
                    scenario,
                    f"{base_op}.{label}",
                    repo,
                    ops[base_op],
                    run_index,
                    metadata,
                    admitted_output_chars,
                )
                row["dirty_fraction"] = fraction
                row["dirty_file_count"] = len(changed)
                if before is not None:
                    row.setdefault("process_returncode", row["returncode"])
                    row = oakbench_semantics.finalize_patch_evidence(row, before, tree_manifest(repo), application_required=False)
                    row["cache_state"] = "hash_warmed_isolated_dirty_replay"
                rows.append(row)
                mutated = row.get("integrity_ok") is False
                if mutated:
                    return rows
        finally:
            if not mutated:
                restore_dirty_fraction(repo, changed, seed)
    return rows


def ops_for(subject: Subject, track: str) -> dict[str, list[str]]:
    """Non-snapshot operation commands from the command-semantics contract.

    Snapshot operations go through snapshot_message/run_git_snapshot so git
    pays its stage+commit as two recorded calls. Where Oak lacks a compact
    equivalent, core-equivalent rows use the closest available command; the
    contract JSON records that gap explicitly.
    """
    sem = semantics()
    vcs = str(subject.bin)
    status = [vcs, *sem.status_args(subject.kind, track)]
    return {
        "repo.init": [vcs, *sem.init_args(subject.kind)],
        "status.clean": status,
        "status.dirty": list(status),
        "diff.dirty": [vcs, *sem.diff_args(subject.kind, track)],
        "branch.create": [vcs, *sem.branch_create_args(subject.kind)],
    }


def snapshot_message(op: str) -> str:
    return "initial" if op == "snapshot.initial" else op


def run_snapshot(
    subject: Subject,
    scenario: Scenario,
    op: str,
    cwd: Path,
    run_index: int,
    metadata: dict[str, Any],
    admitted_output_chars: int,
) -> list[dict[str, Any]]:
    """Snapshot via the command-semantics contract.

    Subjects that stage separately (git) record the stage and commit as
    sub-rows plus a combined canonical row; one-command subjects (oak) record
    a single row. Agents pay every recorded call.
    """
    sem = semantics()
    commands = sem.snapshot_commands(str(subject.bin), subject.kind, snapshot_message(op))
    reads = ["repository_state_fingerprint", "payload_sample_hashes"]
    if len(commands) == 1:
        before = oakbench_integrity.repository_state_fingerprint(cwd)
        expected_hashes = oakbench_integrity.sample_worktree_payload_hashes(cwd)
        row = run_command(subject, scenario, op, cwd, commands[0], run_index, metadata, admitted_output_chars)
        row["harness_reads_before_op"] = reads
        return [apply_state_change_attestation(row, cwd, before, op, expected_hashes=expected_hashes)]

    stage_command, commit_command = commands
    add_result = run_command(
        subject, scenario, f"{op}.add", cwd, stage_command, run_index, metadata, admitted_output_chars
    )
    if add_result["returncode"] != 0:
        return [add_result]
    before_commit = oakbench_integrity.repository_state_fingerprint(cwd)
    expected_hashes = oakbench_integrity.sample_worktree_payload_hashes(cwd)
    commit_result = run_command(
        subject, scenario, f"{op}.commit", cwd, commit_command, run_index, metadata, admitted_output_chars
    )
    commit_result["harness_reads_before_op"] = reads
    apply_state_change_attestation(commit_result, cwd, before_commit, op, expected_hashes=expected_hashes)
    subrows = [add_result, commit_result]
    combined = {
        **commit_result,
        "operation": op,
        "elapsed_ms": round(add_result["elapsed_ms"] + commit_result["elapsed_ms"], 3),
        "returncode": commit_result["returncode"],
        "command": [stage_command, commit_command],
        **merge_interaction_metrics([stage_command, commit_command], subrows),
    }
    if add_result["returncode"] != 0 or commit_result["returncode"] != 0:
        combined["stderr"] = (add_result.get("stderr", "") + "\n" + commit_result.get("stderr", ""))[-4000:]
    return [add_result, commit_result, combined]


def run_isolated_dirty_phase(
    subject: Subject, scenario: Scenario, fixture: Path, run_index: int,
    run_root: Path, metadata: dict[str, Any], skip_diff: bool,
    admitted_output_chars: int, track: str, skip_determinism_probe: bool,
    dirty_spectrum: bool = False,
) -> list[dict[str, Any]]:
    """Reconstruct dirty state after the main chain; never reuse it on mutation.

    Setup and exact pre/post reads warm this disposable repository. Its one
    timed diff is explicitly a hash-warmed replay, not an in-sequence or cold
    measurement. Every subsequent output probe is diagnostic with null timing.
    """
    from oakbench.conformance import tree_manifest
    suffix = "dirty-spectrum" if dirty_spectrum else "dirty-replay"
    repo = run_root / scenario.name / subject.name / f"run-{run_index}-{suffix}"
    copy_fixture(fixture, repo)
    ops = ops_for(subject, track)
    replay_metadata = {**metadata, "measurement_phase": "isolated_dirty_replay",
                       "cache_state": "warm_isolated_fixture_replay"}
    setup_commands = [ops["repo.init"]]
    setup_commands += [[str(subject.bin), *args] for _, args in WIRED_GIT_MODES.get(subject.git_mode, [])]
    setup_commands += semantics().snapshot_commands(str(subject.bin), subject.kind, snapshot_message("snapshot.initial"))
    rows: list[dict[str, Any]] = []
    # Every operation this phase would emit, in order, so an abort records
    # the abandoned tail as skip rows (never "fewer samples").
    if dirty_spectrum:
        phase_ops = dirty_spectrum_operations(True, skip_diff)
    else:
        phase_ops = ([] if skip_diff else ["diff.dirty"])
        if not skip_determinism_probe:
            phase_ops += ["status.dirty.determinism", "status.dirty.inforecall"]
            if not skip_diff:
                phase_ops += ["diff.dirty.determinism", "diff.dirty.inforecall", "diff.full.inforecall"]

    def abandon(reason: str, **extra: Any) -> list[dict[str, Any]]:
        recorded = {str(row.get("operation")) for row in rows}
        for op in phase_ops:
            if op not in recorded:
                row = remote_skip_row(subject, scenario, op, run_index, replay_metadata, reason)
                row.update(extra)
                rows.append(row)
        return rows

    try:
        for command in setup_commands:
            proc = subprocess.run(command, cwd=repo, env=base_env(), stdout=subprocess.DEVNULL, stderr=subprocess.PIPE, check=False)
            if proc.returncode:
                return abandon("isolated_dirty_replay_setup_failed", setup_returncode=proc.returncode,
                               stderr=proc.stderr.decode("utf-8", "replace")[-4000:])
        if dirty_spectrum:
            return run_dirty_spectrum(subject, scenario, repo, ops, run_index, replay_metadata, skip_diff, admitted_output_chars)
        changed_paths = mutate_files(repo, scenario, subject, run_index)
        if not skip_diff:
            reason = git_huge_binary_diff_limit_reason(subject, scenario, "diff.dirty")
            if reason is not None:
                for operation in (["diff.dirty"] if skip_determinism_probe else
                                  ["diff.dirty", "diff.dirty.determinism", "diff.dirty.inforecall", "diff.full.inforecall"]):
                    rows.append(git_huge_binary_diff_skip_row(subject, scenario, operation, ops["diff.dirty"], run_index, replay_metadata))
            else:
                before = tree_manifest(repo)
                row = run_command(subject, scenario, "diff.dirty", repo, ops["diff.dirty"], run_index,
                                  {**replay_metadata, "cache_state": "hash_warmed_isolated_dirty_replay"}, admitted_output_chars)
                row.setdefault("process_returncode", row["returncode"])
                row = oakbench_semantics.finalize_patch_evidence(row, before, tree_manifest(repo), application_required=False)
                rows.append(row)
                if row.get("integrity_ok") is False:
                    return abandon("aborted_after_failed_prerequisite:diff.dirty")
        if skip_determinism_probe:
            return rows
        for base_op in ("status.dirty", "diff.dirty", "diff.full"):
            if base_op.startswith("diff.") and (skip_diff or git_huge_binary_diff_limit_reason(subject, scenario, "diff.dirty") is not None):
                continue
            command = ([str(subject.bin), *semantics().diff_args(subject.kind, "core-equivalent")]
                       if base_op == "diff.full" else ops[base_op])
            if base_op != "diff.full":
                row = run_determinism_probe(subject, scenario, base_op, repo, command, run_index, replay_metadata)
                rows.append(row)
                if row.get("integrity_ok") is False:
                    return abandon(f"aborted_after_failed_prerequisite:{row.get('operation')}")
            row = run_information_probe(subject, scenario, base_op, repo, command, run_index, replay_metadata,
                                        changed_paths, pristine_fixture=fixture if base_op == "diff.full" else None)
            rows.append(row)
            if row.get("integrity_ok") is False:
                return abandon(f"aborted_after_failed_prerequisite:{row.get('operation')}")
        return rows
    finally:
        if subject.git_mode in FSMONITOR_MODES:
            subprocess.run([str(subject.bin), "fsmonitor--daemon", "stop"], cwd=repo, env=base_env(),
                           stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL, check=False)


MAIN_TIMED_OPS = ("proc.spawn", "repo.init", "snapshot.initial", "status.clean", "status.dirty",
                  "snapshot.dirty", "branch.create", "task.snapshot")


def planned_operations(subject: Subject, *, skip_remote: bool, skip_diff: bool,
                       skip_determinism_probe: bool) -> list[str]:
    """Every operation row a complete scenario run would emit, in order.

    Used only to name abandoned operations as skip rows after an abort.
    """
    planned = list(MAIN_TIMED_OPS)
    if not skip_remote:
        planned += list(REMOTE_OPS if subject.kind == "git" else OAK_NET_REMOTE_OPS)
    planned += list(semantics().agent_native_json_probes(subject.kind, BENCH_TASK_BRANCH))
    if not skip_diff:
        planned.append("diff.dirty")
    if not skip_determinism_probe:
        planned += ["status.dirty.determinism", "status.dirty.inforecall"]
        if not skip_diff:
            planned += ["diff.dirty.determinism", "diff.dirty.inforecall", "diff.full.inforecall"]
    return planned


def run_subject_scenario(
    subject: Subject,
    scenario: Scenario,
    fixture: Path,
    run_index: int,
    run_root: Path,
    metadata: dict[str, Any],
    skip_diff: bool,
    admitted_output_chars: int,
    track: str,
    skip_determinism_probe: bool = False,
    skip_remote: bool = False,
    dirty_spectrum: bool = False,
) -> list[dict[str, Any]]:
    metadata = {**metadata, "benchmark_track": track}
    applies, skip_reason = git_mode_applies_to_scenario(subject.git_mode, scenario)
    if not applies:
        return [
            remote_skip_row(
                subject,
                scenario,
                "scenario.mode_skipped",
                run_index,
                metadata,
                skip_reason or "git_mode_not_applicable",
            )
        ]
    if subject.git_mode == "lfs" and not git_lfs_available(subject.bin):
        return [
            remote_skip_row(
                subject,
                scenario,
                "scenario.mode_skipped",
                run_index,
                metadata,
                "git_lfs_unavailable: install git-lfs to measure the git_lfs tuned comparator",
            )
        ]

    repo = run_root / scenario.name / subject.name / f"run-{run_index}"
    copy_fixture(fixture, repo)
    ops = ops_for(subject, track)
    results: list[dict[str, Any]] = []
    planned = planned_operations(subject, skip_remote=skip_remote, skip_diff=skip_diff,
                                 skip_determinism_probe=skip_determinism_probe)

    def abort_after(op: str) -> list[dict[str, Any]]:
        """A failed or mutating op ends this cell's timing: every abandoned
        operation is a recorded skip row naming the failed prerequisite, so a
        consumer never sees merely "fewer samples" (skips are never hidden)."""
        recorded = {str(row.get("operation")) for row in results}
        tail = planned[planned.index(op) + 1:] if op in planned else planned
        results.extend(
            remote_skip_row(subject, scenario, name, run_index, metadata,
                            f"aborted_after_failed_prerequisite:{op}")
            for name in tail if name not in recorded
        )
        return results

    def timed(op: str) -> bool:
        start_index = len(results)
        # Harness attestation reads (metadata fingerprint, payload sample
        # hashes) happen BEFORE the cache step, so in cold mode the purge
        # removes whatever they warmed; rows name the reads they followed.
        if op in {"snapshot.initial", "snapshot.dirty", "task.snapshot"}:
            results.extend(run_snapshot(subject, scenario, op, repo, run_index, metadata, admitted_output_chars))
        else:
            before = oakbench_integrity.repository_state_fingerprint(repo) if op == "branch.create" else None
            row = run_command(subject, scenario, op, repo, ops[op], run_index, metadata, admitted_output_chars)
            if op == "branch.create":
                row["harness_reads_before_op"] = ["repository_state_fingerprint"]
                apply_state_change_attestation(row, repo, before, op)
            results.append(row)
        return bool(results[start_index:]) and all(
            row_returncode(row) == 0 and all(row.get(flag) is not False for flag in
                ("integrity_ok", "integrity_check_passed", "content_integrity_check_passed"))
            for row in results[start_index:])

    try:
        # Binary startup overhead in isolation. Agent fleets invoke the VCS
        # thousands of times a day; process spawn cost is paid on every call.
        results.append(
            run_command(
                subject,
                scenario,
                "proc.spawn",
                repo,
                [str(subject.bin), "--version"],
                run_index,
                metadata,
                admitted_output_chars,
            )
        )

        if row_returncode(results[-1]) != 0:
            return abort_after("proc.spawn")
        if not timed("repo.init"):
            return abort_after("repo.init")
        if subject.git_mode:
            for setup_op, setup_args in WIRED_GIT_MODES[subject.git_mode]:
                setup_row = run_command(
                    subject,
                    scenario,
                    f"mode.setup.{setup_op}",
                    repo,
                    [str(subject.bin), *setup_args],
                    run_index,
                    metadata,
                    admitted_output_chars,
                )
                results.append(setup_row)
                if setup_row["returncode"] != 0:
                    # A tuned subject whose setup failed would silently measure
                    # stock git under a tuned label. Stop here: the failed setup
                    # row is the evidence, and the run exits nonzero.
                    setup_row["notes"] = (
                        f"mode setup failed; remaining {subject.name} operations for this run "
                        "were aborted so stock-git numbers are never published under a tuned label"
                    )
                    return abort_after("repo.init")
        if subject.git_mode == "lfs":
            untracked = lfs_untracked_extensions(repo)
            if untracked:
                # config/git_modes.json git_lfs fairness_note: the mode "requires
                # ... fixture generation with LFS-tracked binaries". A fixture file
                # the track patterns miss would be committed as a plain git blob,
                # and an untracked comparator is an unfair comparator — so the
                # mode setup fails closed here and the scenario's rows become an
                # honest skip instead of git numbers under the git_lfs label.
                results.append(
                    remote_skip_row(
                        subject,
                        scenario,
                        "mode.setup.lfs_track_coverage",
                        run_index,
                        metadata,
                        "lfs_track_incomplete:" + ",".join(untracked),
                    )
                )
                return abort_after("repo.init")
        for op in ("snapshot.initial", "status.clean", "status.dirty", "snapshot.dirty", "branch.create", "task.snapshot"):
            if op == "status.dirty":
                mutate_files(repo, scenario, subject, run_index)
            elif op == "task.snapshot":
                mutate_files(repo, scenario, subject, run_index + 10_000)
            if not timed(op):
                return abort_after(op)
        if not skip_remote:
            results.extend(
                run_remote_phase(subject, scenario, repo, run_index, metadata, admitted_output_chars)
            )
        results.extend(
            run_agent_native_json_probes(
                subject, scenario, repo, run_index, metadata, admitted_output_chars
            )
        )
        if not skip_diff or not skip_determinism_probe:
            results.extend(run_isolated_dirty_phase(subject, scenario, fixture, run_index, run_root,
                           metadata, skip_diff, admitted_output_chars, track, skip_determinism_probe))
        if dirty_spectrum:
            results.extend(run_isolated_dirty_phase(subject, scenario, fixture, run_index, run_root,
                           metadata, skip_diff, admitted_output_chars, track, True, dirty_spectrum=True))
        return results
    finally:
        if subject.git_mode in FSMONITOR_MODES:
            # One fsmonitor daemon spawns per repo; stop it so repeated benchmark
            # runs do not accumulate daemons watching deleted workdirs. Untimed and
            # unrecorded: teardown, not a measured operation.
            subprocess.run(
                [str(subject.bin), "fsmonitor--daemon", "stop"],
                cwd=repo,
                env=base_env(),
                stdout=subprocess.DEVNULL,
                stderr=subprocess.DEVNULL,
                check=False,
            )


def remote_preflight_requirements(
    subjects: list[Subject],
    *,
    skip_remote: bool,
) -> list[oakbench_remotes.RemoteRequirement]:
    if skip_remote or not any(subject.kind == "oak" for subject in subjects):
        return []
    return [
        oakbench_remotes.RemoteRequirement(
            purpose="default",
            lane="core",
            operations=tuple(OAK_NET_REMOTE_OPS),
        )
    ]


def metric_medians(rows: list[dict[str, Any]], metric: str) -> dict[tuple[str, str], float]:
    grouped: dict[tuple[str, str], list[float]] = {}
    for row in rows:
        if row["returncode"] != 0:
            continue
        op = row["operation"]
        if op.endswith(".add") or op.endswith(".commit"):
            continue
        try:
            value = float(row[metric])
        except (KeyError, TypeError, ValueError):
            continue
        key = (row["subject"], row["scenario"] + "/" + op)
        grouped.setdefault(key, []).append(value)
    return {key: statistics.median(values) for key, values in grouped.items()}


def average_by_subject(rows: list[dict[str, Any]], metric: str) -> dict[str, float]:
    grouped: dict[str, list[float]] = {}
    for row in rows:
        if row["returncode"] != 0:
            continue
        op = row["operation"]
        if op.endswith(".add") or op.endswith(".commit"):
            continue
        # Diagnostic and setup rows stay out of cross-operation subject
        # averages so trend lines keep comparing the same op mix.
        # remote.* ops stay out too: subjects without a wired remote skip them,
        # and an asymmetric op mix would silently skew the cross-op average.
        if (
            op.endswith(".determinism")
            or op.endswith(".inforecall")
            or op.startswith("mode.setup")
            or op.startswith("remote.")
            or op == "proc.spawn"
        ):
            continue
        try:
            value = float(row[metric])
        except (KeyError, TypeError, ValueError):
            continue
        grouped.setdefault(str(row["subject"]), []).append(value)
    return {subject: statistics.mean(values) for subject, values in grouped.items()}


# Delta math and formatting live in oakbench.reporting; these aliases keep
# the historical local names used throughout the summary writer.
fmt_delta = fmt_delta_lower_better
fmt_float = fmt_num


def strongest_git_baseline(by_key: dict[tuple[str, str], float], subjects: list[Subject]) -> dict[str, tuple[str, float]]:
    """Per scenario/op, the FASTEST Git subject that measured it (stock or tuned).

    Oak is compared against the stronger baseline so a weaker Git mode (e.g.
    fsmonitor IPC on a tiny tree) can never flatter Oak (ADR-0011).
    """
    git_names = [subject.name for subject in subjects if subject.kind == "git" and subject.name != "git_aa"]
    best: dict[str, tuple[str, float]] = {}
    for (name, scenario_op), value in by_key.items():
        if "/mode.setup." in scenario_op or "/scenario." in scenario_op:
            continue
        if name in git_names and (scenario_op not in best or value < best[scenario_op][1]):
            best[scenario_op] = (name, value)
    return best


def strongest_git_lines(by_key: dict[tuple[str, str], float], subjects: list[Subject]) -> list[str]:
    git_subjects = [subject for subject in subjects if subject.kind == "git" and subject.name != "git_aa"]
    if len(git_subjects) < 2:
        return []
    best = strongest_git_baseline(by_key, subjects)
    oak_names = [subject.name for subject in subjects if subject.kind == "oak"]
    lines = ["", "## Oak vs the strongest Git baseline per cell", "",
             "| Scenario / operation | Strongest Git | Git median ms | " + " | ".join(f"`{n}` vs strongest" for n in oak_names) + " |",
             "| --- | --- | ---: | " + " | ".join("---:" for _ in oak_names) + " |"]
    for scenario_op, (name, value) in sorted(best.items()):
        cells = []
        for oak_name in oak_names:
            oak_value = by_key.get((oak_name, scenario_op))
            cells.append("" if oak_value is None else f"{delta_pct(value, oak_value):+.1f}%")
        lines.append(f"| `{scenario_op}` | `{name}` | {value:.1f} | " + " | ".join(cells) + " |")
    return lines


def summary_text(rows: list[dict[str, Any]], subjects: list[Subject]) -> str:
    by_key = metric_medians(rows, "elapsed_ms")
    token_by_key = metric_medians(rows, "estimated_tokens_total")
    tools_by_key = metric_medians(rows, "tool_call_count")
    output_by_key = metric_medians(rows, "raw_output_bytes")
    avg_ms = average_by_subject(rows, "elapsed_ms")
    avg_tokens = average_by_subject(rows, "estimated_tokens_total")
    avg_cost = average_by_subject(rows, "estimated_cost_weighted_tokens")
    avg_tools = average_by_subject(rows, "tool_call_count")
    avg_output = average_by_subject(rows, "raw_output_bytes")
    git_name = next((subject.name for subject in subjects if subject.kind == "git"), None)
    oak_baseline = next((subject.name for subject in subjects if subject.name in {"oak_installed", "oak_main"}), None)
    lines = [
        "# Oak Benchmark Summary",
        "",
        f"Track: `{rows[0].get('benchmark_track', 'unknown') if rows else 'unknown'}`.",
        "Positive deltas mean the row subject is lower/better than the baseline for that metric.",
        "Estimated tokens are a dependency-free proxy: command text plus admitted stdout/stderr characters divided by four.",
        "Tool calls are exact terminal/VCS command counts for this direct CLI harness.",
        "Output bytes are full stdout/stderr byte counts, even when admitted token text was capped.",
        f"Cost-weighted tokens use billing direction ({COST_WEIGHTS_NOTE}): commands an agent types are model output, stdout it reads is model input.",
        "",
        "## Subject Averages",
        "",
        "| Subject | Avg ms/op | Avg est tokens/op | Avg cost-weighted/op | Avg output bytes/op | Avg tool calls/op | vs Git tokens | vs Oak tokens | vs Git tools | vs Oak tools |",
        "| --- | ---: | ---: | ---: | ---: | ---: | ---: | ---: | ---: | ---: |",
    ]
    git_avg_tokens = avg_tokens.get(git_name) if git_name else None
    git_avg_tools = avg_tools.get(git_name) if git_name else None
    oak_avg_tokens = avg_tokens.get(oak_baseline) if oak_baseline else None
    oak_avg_tools = avg_tools.get(oak_baseline) if oak_baseline else None
    for subject in subjects:
        token_value = avg_tokens.get(subject.name)
        tool_value = avg_tools.get(subject.name)
        lines.append(
            "| "
            f"`{subject.name}` | "
            f"{fmt_float(avg_ms.get(subject.name))} | "
            f"{fmt_float(token_value)} | "
            f"{fmt_float(avg_cost.get(subject.name))} | "
            f"{fmt_float(avg_output.get(subject.name))} | "
            f"{fmt_float(tool_value, 2)} | "
            f"{fmt_delta(git_avg_tokens, token_value, subject.name == git_name)} | "
            f"{fmt_delta(oak_avg_tokens, token_value, subject.name == oak_baseline or subject.kind == 'git')} | "
            f"{fmt_delta(git_avg_tools, tool_value, subject.name == git_name)} | "
            f"{fmt_delta(oak_avg_tools, tool_value, subject.name == oak_baseline or subject.kind == 'git')} |"
        )

    info_rows = [row for row in rows if str(row.get("operation", "")).endswith(".inforecall")]
    if info_rows:
        lines.extend(
            [
                "",
                "## Output Information Content",
                "",
                "Bytes saved are only a win if the changed-file list survives. Recall is the",
                "fraction of ground-truth changed paths recoverable from the command output;",
                "pipe-compat means diff output kept unified hunk/file structure (or an explicit",
                "binary notice) that scripts and `patch`-style tooling can consume.",
                "",
                "| Subject | Operation | Mean recall | Mean bytes/file named | Pipe-compat rate |",
                "| --- | --- | ---: | ---: | ---: |",
            ]
        )
        grouped: dict[tuple[str, str], list[dict[str, Any]]] = {}
        for row in info_rows:
            grouped.setdefault((str(row["subject"]), str(row["operation"])), []).append(row)
        for (subject_name, op), op_rows in sorted(grouped.items()):
            recalls = [row["information_recall"] for row in op_rows if row.get("information_recall") is not None]
            byte_costs = [
                row["bytes_per_changed_file_named"]
                for row in op_rows
                if row.get("bytes_per_changed_file_named") is not None
            ]
            compat_flags = [row["pipe_compatible_unified"] for row in op_rows if "pipe_compatible_unified" in row]
            recall_cell = f"{statistics.mean(recalls):.3f}" if recalls else ""
            bytes_cell = f"{statistics.mean(byte_costs):.0f}" if byte_costs else ""
            compat_cell = f"{sum(1 for flag in compat_flags if flag) / len(compat_flags) * 100:.0f}%" if compat_flags else ""
            lines.append(f"| `{subject_name}` | `{op}` | {recall_cell} | {bytes_cell} | {compat_cell} |")

    lines.extend(
        [
            "",
            "## Operation Medians",
            "",
            "| Scenario / operation | Subject | Median ms | Median est tokens | Median output bytes | Median tool calls | vs Git time | vs Git tokens | vs Git tools | vs Oak time | vs Oak tokens | vs Oak tools |",
            "| --- | --- | ---: | ---: | ---: | ---: | ---: | ---: | ---: | ---: | ---: | ---: |",
        ]
    )
    all_ops = sorted({scenario_op for _, scenario_op in by_key})
    for scenario_op in all_ops:
        git_base = by_key.get((git_name, scenario_op)) if git_name else None
        git_token_base = token_by_key.get((git_name, scenario_op)) if git_name else None
        git_tool_base = tools_by_key.get((git_name, scenario_op)) if git_name else None
        oak_base = by_key.get((oak_baseline, scenario_op)) if oak_baseline else None
        oak_token_base = token_by_key.get((oak_baseline, scenario_op)) if oak_baseline else None
        oak_tool_base = tools_by_key.get((oak_baseline, scenario_op)) if oak_baseline else None
        for subject in subjects:
            value = by_key.get((subject.name, scenario_op))
            if value is None:
                continue
            token_value = token_by_key.get((subject.name, scenario_op))
            tool_value = tools_by_key.get((subject.name, scenario_op))
            output_value = output_by_key.get((subject.name, scenario_op))
            vs_git_time = "" if git_base is None or subject.name == git_name else f"{delta_pct(git_base, value):+.1f}%"
            vs_git_tokens = (
                ""
                if git_token_base is None or token_value is None or subject.name == git_name
                else f"{delta_pct(git_token_base, token_value):+.1f}%"
            )
            vs_git_tools = (
                ""
                if git_tool_base is None or tool_value is None or subject.name == git_name
                else f"{delta_pct(git_tool_base, tool_value):+.1f}%"
            )
            vs_oak_time = (
                ""
                if oak_base is None or subject.name == oak_baseline or subject.kind == "git"
                else f"{delta_pct(oak_base, value):+.1f}%"
            )
            vs_oak_tokens = (
                ""
                if oak_token_base is None or token_value is None or subject.name == oak_baseline or subject.kind == "git"
                else f"{delta_pct(oak_token_base, token_value):+.1f}%"
            )
            vs_oak_tools = (
                ""
                if oak_tool_base is None or tool_value is None or subject.name == oak_baseline or subject.kind == "git"
                else f"{delta_pct(oak_tool_base, tool_value):+.1f}%"
            )
            # Unmeasured renders blank, never 0 (ADR-0002): diagnostic rows
            # carry no token/tool fields and must not look like free output.
            lines.append(
                f"| `{scenario_op}` | `{subject.name}` | {value:.1f} | "
                f"{fmt_float(token_value)} | "
                f"{fmt_float(output_value, 0)} | "
                f"{fmt_float(tool_value)} | "
                f"{vs_git_time} | {vs_git_tokens} | {vs_git_tools} | "
                f"{vs_oak_time} | {vs_oak_tokens} | {vs_oak_tools} |"
            )
    lines.extend(strongest_git_lines(by_key, subjects))
    skips = [row for row in rows if row.get("skipped")]
    if skips:
        lines.extend(["", "## Skipped (unmeasured, never hidden)", ""])
        seen_reasons: dict[tuple[str, str], int] = {}
        for row in skips:
            seen_reasons[(str(row["subject"]), str(row.get("skip_reason", "")))] = (
                seen_reasons.get((str(row["subject"]), str(row.get("skip_reason", ""))), 0) + 1
            )
        for (subject_name, reason), count in sorted(seen_reasons.items()):
            lines.append(f"- `{subject_name}`: {count} rows skipped — {reason}")
    failures = [row for row in rows if row["returncode"] != 0 and not row.get("skipped")]
    if failures:
        lines.extend(["", "## Failures", ""])
        for row in failures[:20]:
            reason = row.get("integrity_failure_reason") or row.get("content_integrity_failure_reason") or row.get(
                "stderr", ""
            )
            lines.append(
                f"- `{row['subject']}` `{row['scenario']}/{row['operation']}` "
                f"exit {row['returncode']}: `{str(reason).splitlines()[-1:]}`"
            )
    return "\n".join(lines) + "\n"


def main() -> int:
    args = parse_args()
    subjects = expand_git_modes(load_subjects(args), args.git_modes)
    oakbench_remotes.print_remote_preflight_warnings(
        oakbench_remotes.oak_remote_preflight_warnings(
            remote_preflight_requirements(subjects, skip_remote=args.skip_remote)
        )
    )
    scenarios = PROFILES[args.profile]
    if args.scenarios:
        wanted = {name.strip() for name in args.scenarios.split(",") if name.strip()}
        unknown = wanted - {scenario.name for scenario in scenarios}
        if unknown:
            raise SystemExit(f"--scenarios not in profile {args.profile}: {', '.join(sorted(unknown))}")
        scenarios = [scenario for scenario in scenarios if scenario.name in wanted]
    if args.runs:
        scenarios = [
            Scenario(s.name, s.file_count, s.file_size, s.dirty_count, s.binary, args.runs)
            for s in scenarios
        ]

    timestamp = datetime.now(timezone.utc).strftime("%Y%m%dT%H%M%SZ")
    args.results.mkdir(parents=True, exist_ok=True)
    args.workdir.mkdir(parents=True, exist_ok=True)
    run_root = args.workdir / "runs" / timestamp
    fixture_root = args.workdir

    metadata = {
        "bench_id": timestamp,
        "profile": args.profile,
        "benchmark_track": args.track,
        "command_semantics_version": COMMAND_SEMANTICS_VERSION,
        "timestamp_utc": timestamp,
        "host": platform.node(),
        "platform": platform.platform(),
        "machine": platform.machine(),
        "python": platform.python_version(),
        "env_isolation_version": oakbench_environment.ENV_ISOLATION_VERSION,
        "output_oracle_version": oakbench_semantics.OUTPUT_ORACLE_VERSION,
        "patch_oracle_version": oakbench_semantics.PATCH_ORACLE_VERSION,
        "subject_versions": {subject.name: subject_version(subject) for subject in subjects},
        "subject_details": subject_details(subjects),
        "source": source_metadata(args.oak_repo),
        "randomized_subject_order": bool(args.randomize_subject_order),
        "timing_contract_version": TIMING_CONTRACT_VERSION,
        "cache_mode_requested": args.cache_mode,
        **runner_fields(),
    }
    if args.cache_mode == "cold":
        preflight = oakbench_cachectl.purge_fs_caches()
        if not preflight.get("ok"):
            raise SystemExit(
                "--cache-mode cold refused: page-cache purge is not permitted on this host "
                f"({preflight.get('method')}: {preflight.get('error')}). On macOS add a NOPASSWD "
                "sudoers entry for /usr/sbin/purge; on Linux run with write access to "
                "/proc/sys/vm/drop_caches. Cold rows are never claimed without a verified purge.")

    # Fixture generation runs OUTSIDE the measurement lock: building file
    # trees parallelizes safely (atomic shape-keyed builds), only timed
    # measurement must serialize across processes.
    fixtures: dict[str, Path] = {}
    for scenario in scenarios:
        print(
            f"[fixture] {scenario.name}: {scenario.file_count} files x {scenario.file_size} bytes",
            flush=True,
        )
        fixtures[scenario.name] = ensure_fixture(fixture_root, scenario)
    fixture_fingerprints = {name: fixture_fingerprint(path) for name, path in fixtures.items()}

    metadata["load_tier"] = args.load_tier

    rows: list[dict[str, Any]] = []
    with measurement_lock("bench") as lock_info:
        metadata["measurement_lock_wait_ms"] = lock_info.wait_ms
        metadata["measurement_lock"] = "held" if lock_info.enabled else "disabled"
        # Background load (untimed start/stop). Failing to start under a
        # non-none tier converts the whole invocation to skip rows: load that
        # was not applied is never claimed.
        load_generator: oakbench_loadgen.LoadGenerator | None = None
        if args.load_tier != "none":
            load_generator = oakbench_loadgen.LoadGenerator(
                args.load_tier, args.workdir / "loadgen-temp"
            )
            load_start = load_generator.start()
            if not load_start["started"]:
                reason = oakbench_loadgen.load_tier_unavailable_reason(str(load_start["reason"]))
                rows = load_tier_skip_rows(subjects, scenarios, metadata, reason, args.load_tier)
                load_generator = None
        try:
            if not rows:
                for scenario in scenarios:
                    fixture = fixtures[scenario.name]
                    for run_index in range(scenario.runs):
                        run_subjects = list(subjects)
                        if args.randomize_subject_order:
                            random.Random(f"{timestamp}:{scenario.name}:{run_index}").shuffle(run_subjects)
                        run_metadata = {
                            **metadata,
                            "subject_order": [subject.name for subject in run_subjects],
                            **fixture_fingerprints[scenario.name],
                        }
                        for subject in run_subjects:
                            print(f"[run] {scenario.name} run={run_index} subject={subject.name}", flush=True)
                            env_before = oakbench_envwatch.sample_environment()
                            subject_rows = run_subject_scenario(
                                subject,
                                scenario,
                                fixture,
                                run_index,
                                run_root,
                                run_metadata,
                                args.skip_diff,
                                args.admitted_output_chars,
                                args.track,
                                args.skip_determinism_probe,
                                args.skip_remote,
                                args.dirty_spectrum,
                            )
                            env_after = oakbench_envwatch.sample_environment()
                            boundary = oakbench_loadgen.environment_boundary_fields(
                                args.load_tier, env_before, env_after
                            )
                            for row in subject_rows:
                                row.update(boundary)
                            rows.extend(subject_rows)
        finally:
            if load_generator is not None:
                load_stop = load_generator.stop()
                # bogo-ops parsed ONLY as did-the-load-run verification,
                # never reported as a benchmark number.
                for row in rows:
                    row["load_verified"] = load_stop.get("verified")

    store = ResultsStore(args.results, lane="core")
    raw_path, summary_path = store.write(timestamp, rows, summary_text(rows, subjects))

    if not args.keep_workdirs:
        shutil.rmtree(run_root, ignore_errors=True)

    print(f"[result] {raw_path}")
    print(f"[summary] {summary_path}")
    return 1 if any(row["returncode"] != 0 and not row.get("skipped") for row in rows) else 0


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

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

Download ZIP
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	`"""Benchmark Git and Oak on agent-shaped repository workloads.`
3
4	`The harness is intentionally dependency-free. It generates deterministic`
5	`fixtures, copies each fixture per subject, times only VCS commands, and writes`
6	`JSONL plus a Markdown summary.`
7	`"""`
8
9	`from __future__ import annotations`
10
11	`import argparse`
12	`import fnmatch`
13	`import functools`
14	`import hashlib`
15	`import json`
16	`import math`
17	`import os`
18	`import platform`
19	`import random`
20	`import re`
21	`import secrets`
22	`import shutil`
23	`import statistics`
24	`import subprocess`
25	`import tempfile`
26	`import time`
27	`from dataclasses import dataclass`
28	`from datetime import datetime, timezone`
29	`from pathlib import Path`
30	`from typing import Any`
31
32	`from oakbench import cachectl as oakbench_cachectl`
33	`from oakbench import environment as oakbench_environment`
34	`from oakbench import envwatch as oakbench_envwatch`
35	`from oakbench import fixtures as oakbench_fixtures`
36	`from oakbench import loadgen as oakbench_loadgen`
37	`from oakbench import integrity as oakbench_integrity`
38	`from oakbench import output_semantics as oakbench_semantics`
39	`from oakbench import remotes as oakbench_remotes`
40	`from oakbench import tokens as oakbench_tokens`