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

    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"

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

    "lfs": [

        ("lfs_install", ["lfs", "install", "--local"]),

        ("lfs_track", ["lfs", "track", *LFS_TRACK_PATTERNS]),

    ],

}

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

    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

    return {"binary_sha256": digest or binary_sha256(subject.bin)}

@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("--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("--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(

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

        )

        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.

    """

    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

    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": round(elapsed[-1], 3),

        "returncode": returncode,

        "command": command,

        "output_stable": hashes[0] == hashes[1],

        "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

    return row

# Cap on probe output read into memory for recall scoring. Recall is computed

# on what fits; the truncation flag marks rows where the cap bit.

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.

    """

    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 = tree_manifest(cwd)

    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": round(capture.elapsed_ms, 3),

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

            ))

            row = oakbench_semantics.finalize_patch_evidence(

                row, expected_manifest, tree_manifest(cwd), application_required=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["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)

        if not content_fields["content_integrity_check_passed"] 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

    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,

        **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),

        "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,

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

    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)

    return [

        run_agent_native_json_probe(

            subject, scenario, op, cwd, spec, run_index, metadata, admitted_output_chars

        )

        for op, spec in probes.items()

    ]

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 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]:

    capture = run_timed(command, cwd, base_env(), admitted_output_chars)

    result = {

        **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": round(capture.elapsed_ms, 3),

        "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,

        "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 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.

    """

    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)

        try:

            for base_op in base_ops:

                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)

                rows.append(row)

        finally:

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

    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)

        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

    )

    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_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 [