#!/usr/bin/env python3

"""Run deterministic agent-shaped workflow A/B tests across Git and Oak.

This complements scripts/bench.py. The core benchmark times isolated VCS

commands; this script runs whole task recipes that look more like common agent

work: search, read, edit, test/validate, status, diff, and snapshot. The

fixtures and edits are deterministic so different subjects only vary by VCS.

"""

from __future__ import annotations

import argparse

import json

import platform

import random

import shlex

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

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 workflow_oracles

from oakbench import remotes as oakbench_remotes

from oakbench import tokens as oakbench_tokens

from oakbench.command_semantics import semantics

from oakbench.environment import command_display

from oakbench.execution import run_timed

from oakbench.reporting import fmt_delta_lower_better, fmt_num

from oakbench.results import ResultsStore

from oakbench.rows import SKIP_RETURNCODE, row_returncode

from oakbench.runlock import measurement_lock

from oakbench.subjects import (

    DEFAULT_OAK_REPO,

    Subject,

    load_subjects,

    source_metadata,

    subject_details,

    subject_version,

)

ROOT = Path(__file__).resolve().parents[1]

DEFAULT_WORKDIR = Path(tempfile.gettempdir()) / "oak-workflow-ab"

DEFAULT_ADMITTED_OUTPUT_CHARS = 20_000

FIXTURE_VERSION = "2026-06-09.2"

TRACKS = ("agent-default", "core-equivalent")

# Compatibility re-exports: agent_workflow.py and tests reference these

# through this module. The definitions live in oakbench (the measurement core).

COST_WEIGHTS_NOTE = oakbench_tokens.COST_WEIGHTS_NOTE

COMMAND_SEMANTICS_VERSION = semantics().version

_OAK_COMMIT_PUSH_SUPPORT_CACHE: dict[str, str | None] = {}

def base_env() -> dict[str, str]:

    return oakbench_environment.base_env(

        author_name="Oak Workflow Bench",

        author_email="[email protected]",

        oak_author="oak-workflow-bench",

    )

def python_no_bytecode_env(base: dict[str, str] | None = None) -> dict[str, str]:

    env = dict(base) if base is not None else base_env()

    env["PYTHONDONTWRITEBYTECODE"] = "1"

    return env

def step_env(step: "Step", base: dict[str, str] | None = None) -> dict[str, str]:

    if step.kind == "test":

        return python_no_bytecode_env(base)

    return dict(base) if base is not None else base_env()

def oak_commit_push_skip_reason(subject: Subject) -> str | None:

    if subject.kind != "oak":

        return None

    key = str(subject.bin)

    if key in _OAK_COMMIT_PUSH_SUPPORT_CACHE:

        return _OAK_COMMIT_PUSH_SUPPORT_CACHE[key]

    try:

        probe = subprocess.run(

            [key, "commit", "--help"],

            capture_output=True,

            text=True,

            env=base_env(),

            timeout=15,

            check=False,

        )

    except FileNotFoundError:

        reason = f"missing_binary:{key}"

        _OAK_COMMIT_PUSH_SUPPORT_CACHE[key] = reason

        return reason

    except subprocess.TimeoutExpired:

        reason = "oak_commit_push_capability_timeout"

        _OAK_COMMIT_PUSH_SUPPORT_CACHE[key] = reason

        return reason

    help_text = (probe.stdout or "") + "\n" + (probe.stderr or "")

    if probe.returncode != 0:

        reason = "oak_commit_push_capability_failed"

    elif "--push" not in help_text:

        reason = "oak_commit_push_not_supported:requires_current_oak"

    else:

        reason = None

    _OAK_COMMIT_PUSH_SUPPORT_CACHE[key] = reason

    return reason

@dataclass(frozen=True)

class Workflow:

    name: str

    description: str

    make_fixture: Callable[[Path], None]

    steps: Callable[..., list["Step"]]

    # Optional untimed-phase steps (run in the setup phase, after repo init)

    # for workflows that need pre-built history or pre-broken state.

    prepare: Callable[..., list["Step"]] | None = None

    # Workflows that exercise a remote set the oakbench.remotes purpose here.

    # Their steps/prepare callables take (subject, track, ctx); ctx carries the

    # per-run disposable branch and remote/clone paths. Oak subjects without

    # the purpose env var configured emit a skip row (returncode 77).

    remote_purpose: str | None = None

@dataclass(frozen=True)

class Step:

    operation: str

    kind: str

    command: list[str]

    expected_returncodes: tuple[int, ...] = (0,)

    # For expected-failure steps: the literal recovery command the workflow

    # runs next. The row records whether the error text actually mentions it —

    # the "error output is a prompt" actionability measurement. A misleading

    # hint (error suggests something other than the real fix) is a finding,

    # not a harness bug.

    recovery_hint: str | None = None

def parse_args() -> argparse.Namespace:

    parser = argparse.ArgumentParser(description=__doc__)

    parser.add_argument(

        "--workflows",

        default="bugfix_test_loop,wide_config_refactor,large_asset_manifest",

        help="Comma-separated workflow names, or 'all'.",

    )

    parser.add_argument("--subjects", help="Comma-separated subject names")

    parser.add_argument("--runs", type=int, default=1)

    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" / "workflow-ab")

    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(

        "--track",

        choices=TRACKS,

        default="agent-default",

        help=(

            "agent-default measures natural current commands; core-equivalent "

            "uses the closest matching semantic output level across VCSs."

        ),

    )

    parser.add_argument(

        "--randomize-subject-order",

        action="store_true",

        help="Shuffle subject order per workflow 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.",

    )

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

        ),

    )

    return parser.parse_args()

def shell(script: str) -> list[str]:

    return ["/bin/zsh", "-c", script]

def search_command(pattern: str, target: str) -> str:

    # rg is not guaranteed on benchmark runners; grep -rnE has the same exit

    # contract (0 match, 1 no match). The row's command field records which ran.

    # rg skips hidden directories by default; grep must be told to skip VCS

    # internals or its match scope (and token counts) diverge.

    if shutil.which("rg"):

        return f"rg -n {shlex.quote(pattern)} {target}"

    return f"grep -rnE --exclude-dir=.git --exclude-dir=.oak {shlex.quote(pattern)} {target}"

def snapshot_steps(subject: Subject, operation_prefix: str, message: str) -> list[Step]:

    """Snapshot via the command-semantics contract: git pays stage+commit as two

    tool calls, oak one. Operation names follow the historical .add/.commit split."""

    sem = semantics()

    vcs = str(subject.bin)

    stage = sem.stage_args(subject.kind)

    if stage is not None:

        return [

            Step(f"{operation_prefix}.add", "vcs", [vcs, *stage]),

            Step(f"{operation_prefix}.commit", "vcs", [vcs, *sem.commit_args(subject.kind, message)]),

        ]

    return [Step(operation_prefix, "vcs", [vcs, *sem.commit_args(subject.kind, message)])]

def vcs_init_steps(subject: Subject) -> list[Step]:

    sem = semantics()

    vcs = str(subject.bin)

    return [

        Step("setup.repo_init", "vcs", [vcs, *sem.init_args(subject.kind)]),

        *snapshot_steps(subject, "setup.snapshot", "initial"),

    ]

def vcs_workflow_steps(subject: Subject, track: str) -> list[Step]:

    sem = semantics()

    vcs = str(subject.bin)

    return [

        Step("vcs.branch", "vcs", [vcs, *sem.branch_create_args(subject.kind)]),

        Step("vcs.status", "vcs", [vcs, *sem.status_args(subject.kind, track)]),

        Step("vcs.diff", "vcs", [vcs, *sem.diff_args(subject.kind, track)]),

        *snapshot_steps(subject, "vcs.snapshot", "agent task"),

    ]

def make_bugfix_fixture(path: Path) -> None:

    (path / "app").mkdir(parents=True)

    (path / "tests").mkdir()

    (path / "app" / "__init__.py").write_text("")

    (path / "app" / "pricing.py").write_text(

        "\n".join(

            [

                "def apply_discount(cents: int, percent: int) -> int:",

                "    \"\"\"Return discounted cents rounded down.\"\"\"",

                "    # BUG: this subtracts the percent value, not the percentage.",

                "    return cents - percent",

                "",

            ]

        )

    )

    (path / "tests" / "test_pricing.py").write_text(

        "\n".join(

            [

                "import unittest",

                "from app.pricing import apply_discount",

                "",

                "",

                "class PricingTest(unittest.TestCase):",

                "    def test_apply_discount(self):",

                "        self.assertEqual(apply_discount(1000, 25), 750)",

                "        self.assertEqual(apply_discount(999, 10), 900)",

                "",

                "",

                "if __name__ == '__main__':",

                "    unittest.main()",

                "",

            ]

        )

    )

    (path / "README.md").write_text("Pricing service fixture for agent workflow benchmarks.\n")

def bugfix_steps(subject: Subject, track: str) -> list[Step]:

    return [

        Step("search.locate_symbol", "search", shell(search_command("apply_discount|BUG|discount", "."))),

        Step("read.source", "read", shell("sed -n '1,120p' app/pricing.py")),

        Step("read.tests", "read", shell("sed -n '1,160p' tests/test_pricing.py")),

        Step("test.red", "test", shell("python3 -m unittest discover -s tests"), (1,)),

        Step(

            "edit.fix_bug",

            "edit",

            shell(

                "python3 - <<'PY'\n"

                "from pathlib import Path\n"

                "p = Path('app/pricing.py')\n"

                "text = p.read_text()\n"

                "text = text.replace('return cents - percent', 'return cents - ((cents * percent) // 100)')\n"

                "p.write_text(text)\n"

                "PY"

            ),

        ),

        Step("test.green", "test", shell("python3 -m unittest discover -s tests")),

        *vcs_workflow_steps(subject, track),

    ]

def make_wide_config_fixture(path: Path) -> None:

    root = path / "services"

    for index in range(600):

        service = root / f"service-{index:04d}"

        service.mkdir(parents=True, exist_ok=True)

        timeout = 2500 if index % 3 == 0 else 1800

        service.joinpath("config.toml").write_text(

            "\n".join(

                [

                    f'name = "service-{index:04d}"',

                    f"timeout_ms = {timeout}",

                    "retries = 2",

                    "",

                ]

            )

        )

    (path / "validate_configs.py").write_text(

        "\n".join(

            [

                "from pathlib import Path",

                "bad = []",

                "for p in Path('services').glob('*/config.toml'):",

                "    text = p.read_text()",

                "    if 'timeout_ms = 2500' in text:",

                "        bad.append(str(p))",

                "print(f'bad={len(bad)}')",

                "raise SystemExit(1 if bad else 0)",

                "",

            ]

        )

    )

def wide_config_steps(subject: Subject, track: str) -> list[Step]:

    return [

        Step(

            "search.deprecated_timeout",

            "search",

            shell(search_command("timeout_ms = 2500", "services") + " | head -40"),

        ),

        Step("read.sample_config", "read", shell("sed -n '1,80p' services/service-0000/config.toml")),

        Step(

            "edit.rewrite_configs",

            "edit",

            shell(

                "python3 - <<'PY'\n"

                "from pathlib import Path\n"

                "for p in Path('services').glob('*/config.toml'):\n"

                "    text = p.read_text()\n"

                "    if 'timeout_ms = 2500' in text:\n"

                "        p.write_text(text.replace('timeout_ms = 2500', 'timeout_ms = 3000'))\n"

                "PY"

            ),

        ),

        Step("search.verify_none_left", "search", shell("! " + search_command("timeout_ms = 2500", "services"))),

        Step("test.validate_configs", "test", shell("python3 validate_configs.py")),

        *vcs_workflow_steps(subject, track),

    ]

def write_pattern_file(path: Path, size: int, seed: str) -> None:

    oakbench_fixtures.write_pattern_file(path, size, seed, binary=True, prefix="oak-workflow")

def make_large_asset_fixture(path: Path) -> None:

    assets = path / "assets"

    assets.mkdir(parents=True)

    manifest: list[dict[str, Any]] = []

    for index in range(4):

        asset_path = assets / f"asset-{index:02d}.bin"

        write_pattern_file(asset_path, 16 * 1024 * 1024, f"asset-{index}")

        manifest.append({"path": str(asset_path.relative_to(path)), "bytes": asset_path.stat().st_size, "reviewed": False})

    (path / "manifest.json").write_text(json.dumps({"assets": manifest}, indent=2) + "\n")

    (path / "validate_manifest.py").write_text(

        "\n".join(

            [

                "import json",

                "from pathlib import Path",

                "data = json.loads(Path('manifest.json').read_text())",

                "for item in data['assets']:",

                "    p = Path(item['path'])",

                "    assert p.exists(), p",

                "    assert p.stat().st_size == item['bytes'], p",

                "    assert item['reviewed'] is True, p",

                "print('manifest ok')",

                "",

            ]

        )

    )

def large_asset_steps(subject: Subject, track: str) -> list[Step]:

    return [

        Step("search.assets", "search", shell("find assets -type f -maxdepth 1 | sort")),

        Step("read.manifest", "read", shell("sed -n '1,120p' manifest.json")),

        Step(

            "read.binary_prefix",

            "read",

            shell("python3 - <<'PY'\nfrom pathlib import Path\nprint(Path('assets/asset-00.bin').read_bytes()[:64].hex())\nPY"),

        ),

        Step(

            "edit.mark_reviewed",

            "edit",

            shell(

                "python3 - <<'PY'\n"

                "import json\n"

                "from pathlib import Path\n"

                "p = Path('manifest.json')\n"

                "data = json.loads(p.read_text())\n"

                "for item in data['assets']:\n"

                "    item['reviewed'] = True\n"

                "p.write_text(json.dumps(data, indent=2) + '\\n')\n"

                "PY"

            ),

        ),

        Step("test.validate_manifest", "test", shell("python3 validate_manifest.py")),

        *vcs_workflow_steps(subject, track),

    ]

def make_history_fixture(path: Path) -> None:

    (path / "app").mkdir(parents=True)

    (path / "app" / "pricing.py").write_text(

        "\n".join(

            [

                "def apply_discount(cents: int, percent: int) -> int:",

                "    return cents - ((cents * percent) // 100)",

                "",

            ]

        )

    )

    (path / "app" / "totals.py").write_text("def total(items):\n    return sum(items)\n")

    (path / "README.md").write_text("History archaeology fixture for agent workflow benchmarks.\n")

def history_prepare_steps(subject: Subject, track: str) -> list[Step]:

    vcs = shlex.quote(str(subject.bin))

    steps: list[Step] = []

    edits = [

        ("app/pricing.py", "support negative-percent surcharges", "# rev: surcharge support"),

        ("app/totals.py", "guard empty item lists", "# rev: empty-list guard"),

        ("app/pricing.py", "clamp discount to 100 percent", "# rev: discount clamp"),

    ]

    for index, (target, message, marker) in enumerate(edits):

        if subject.kind == "git":

            script = (

                f"echo {shlex.quote(marker)} >> {shlex.quote(target)} && "

                f"{vcs} add . && {vcs} commit -m {shlex.quote(message)}"

            )

        else:

            script = f"echo {shlex.quote(marker)} >> {shlex.quote(target)} && {vcs} commit --no-verify"

        steps.append(Step(f"setup.history.commit_{index}", "vcs", shell(script)))

    return steps

def history_archaeology_steps(subject: Subject, track: str) -> list[Step]:

    # Agents constantly ask "what changed recently, in this file, and why".

    # Oak lacks show/blame/pickaxe equivalents today, so its rows use the

    # closest available command (oak log); the token/capability gap is the

    # measurement, and the command field records what actually ran.

    vcs = str(subject.bin)

    if subject.kind == "git":

        return [

            Step("history.recent", "vcs", [vcs, "log", "--oneline", "-15"]),

            Step("history.show_last", "vcs", [vcs, "show", "--stat", "HEAD"]),

            Step("history.file_history", "vcs", [vcs, "log", "--oneline", "-5", "--", "app/pricing.py"]),

            Step("history.pickaxe", "vcs", [vcs, "log", "-S", "surcharge", "--oneline"], (0, 1, 2)),

        ]

    return [

        Step("history.recent", "vcs", [vcs, "log", "-n", "15"]),

        Step("history.show_last", "vcs", [vcs, "log", "-n", "1", "-v"]),

        Step("history.file_history", "vcs", [vcs, "log", "-v"], (0, 1, 2)),

        Step("history.pickaxe", "vcs", [vcs, "log", "-v"], (0, 1, 2)),

    ]

def error_recovery_steps(subject: Subject, track: str) -> list[Step]:

    # Error output is a prompt: its size and clarity set how many tokens an

    # agent burns recovering. Failure-path rows are tolerated (wide expected

    # returncodes) so each subject's error text cost is recorded, not skipped.

    sem = semantics()

    vcs = str(subject.bin)

    if subject.kind == "git":

        commit_nothing = [vcs, "commit", "-m", "empty"]

    else:

        commit_nothing = [vcs, *sem.commit_args(subject.kind, "empty")]

    return [

        Step("error.commit_nothing", "vcs", commit_nothing, (0, 1, 2)),

        Step("error.unknown_subcommand", "vcs", [vcs, "stage-all-now"], (0, 1, 2, 101)),

        Step("error.status_after", "vcs", [vcs, "status"]),

        Step(

            "edit.recovery_change",

            "edit",

            shell("echo '# recovered' >> README.md"),

        ),

        *snapshot_steps(subject, "vcs.snapshot", "recovered"),

    ]

def make_sync_fixture(path: Path) -> None:

    (path / "work.txt").write_text("base line\n")

    (path / "notes.txt").write_text("shared notes\n")

    (path / "README.md").write_text("Sync/divergence recovery fixture for agent workflow benchmarks.\n")

def sync_prepare_steps(subject: Subject, track: str, ctx: dict[str, str]) -> list[Step]:

    """Untimed topology: link the repo to a remote, stand up a second checkout

    B, and move upstream from B so the measured checkout is behind.

    Git uses a local bare remote (local_file transport); oak uses the real

    server (network transport). Latency rows across subjects in sync workflows

    are therefore never comparable — the cross-subject evidence is error-output

    tokens, tool calls, and recovery step counts, which transport cannot change.

    """

    vcs = shlex.quote(str(subject.bin))

    bdir = shlex.quote(ctx["clone_b"])

    if subject.kind == "git":

        rdir = shlex.quote(ctx["remote_dir"])

        return [

            Step("setup.remote.create", "vcs", shell(f"{vcs} init -q --bare {rdir}")),

            Step(

                "setup.remote.link",

                "vcs",

                shell(f"{vcs} remote add origin {rdir} && {vcs} push -q -u origin HEAD"),

            ),

            Step("setup.remote.clone_b", "vcs", shell(f"{vcs} clone -q {rdir} {bdir}")),

            Step(

                "setup.upstream.move",

                "vcs",

                shell(

                    f"cd {bdir} && echo 'upstream change' >> notes.txt && "

                    f"{vcs} add . && {vcs} commit -q -m upstream && {vcs} push -q origin HEAD"

                ),

            ),

        ]

    return [

        *oak_clone_a_steps(subject, ctx),

        Step("setup.remote.clone_b", "vcs", shell(f"{vcs} clone {shlex.quote(ctx['oak_repo'])} {bdir}")),

        # B joins the shared branch (it descends from seeded main, so switch

        # pulls it) and moves it. Commit is local-only in current Oak, so setup

        # explicitly uses --push to publish the upstream actor's change.

        Step(

            "setup.upstream.move",

            "vcs",

            shell(

                f"cd {bdir} && {vcs} switch {shlex.quote(ctx['branch'])} && "

                f"echo 'upstream change' >> notes.txt && "

                f"{vcs} commit --no-verify --push"

            ),

        ),

    ]

# Reads the auto-generated current branch name so it can be renamed to the

# run's collision-safe bench branch (rename syncs to the server).

OAK_CURRENT_BRANCH_SNIPPET = (

    "branch list --json | /usr/bin/env python3 -c "

    '\'import json,sys; print(next(b["name"] for b in json.load(sys.stdin) if b.get("current")))\''

)

def oak_clone_a_steps(subject: Subject, ctx: dict[str, str]) -> list[Step]:

    """Untimed oak topology: clone A, seed the disposable repo's main with the

    static fixture once (idempotent — identical bytes merge as a no-op), and

    publish the run's shared bench branch via rename + explicit commit --push."""

    vcs = shlex.quote(str(subject.bin))

    adir = shlex.quote(ctx["clone_a"])

    repo = shlex.quote(ctx["oak_repo"])

    branch = shlex.quote(ctx["branch"])

    fixture_dir = shlex.quote(f"../{ctx['repo_dir']}")

    return [

        Step("setup.remote.clone_a", "vcs", shell(f"{vcs} clone {repo} {adir}")),

        Step(

            "setup.remote.seed_main",

            "vcs",

            shell(

                f"cd {adir} && if [ ! -f work.txt ]; then "

                f"cp {fixture_dir}/work.txt {fixture_dir}/notes.txt {fixture_dir}/README.md . && "

                f"{vcs} commit --no-verify --push && {vcs} merge; fi"

            ),

        ),

        # Commit and publish the auto-named branch, then rename: rename only

        # resolves a branch that exists server-side.

        Step(

            "setup.remote.link",

            "vcs",

            shell(

                f"cd {adir} && echo 'link line' >> README.md && {vcs} commit --no-verify --push && "

                f"cur=$({vcs} {OAK_CURRENT_BRANCH_SNIPPET}) && "

                f"{vcs} branch rename \"$cur\" {branch}"

            ),

        ),

    ]

def sync_push_divergence_steps(subject: Subject, track: str, ctx: dict[str, str]) -> list[Step]:

    # The trap an agent hits when any other actor pushed first. Expected

    # returncodes encode each subject's observed remote-failure contract; a

    # step that "fails to fail" still converges at verify.payload. Error

    # output is a prompt: the rejected push's text is what the agent reasons

    # from, and its cost is the measurement.

    vcs = str(subject.bin)

    verify = Step(

        "verify.payload",

        "read",

        shell("grep -q 'local change' work.txt && grep -q 'upstream change' notes.txt"),

    )

    if subject.kind == "git":

        return [

            Step("edit.local_change", "edit", shell("echo 'local change' >> work.txt")),

            *snapshot_steps(subject, "vcs.snapshot", "local change"),

            # Rejected non-fast-forward push; stderr says to pull.

            Step("error.push_diverged", "vcs", [vcs, "push"], (1,), recovery_hint="git pull"),

            # Following the hint verbatim fails again until a strategy is

            # chosen: fatal "Need to specify how to reconcile divergent

            # branches" (exit 128).

            Step("error.pull_unconfigured", "vcs", [vcs, "pull"], (128,), recovery_hint="--rebase"),

            Step("recovery.pull_rebase", "vcs", [vcs, "pull", "--rebase"]),

            Step("recovery.push_after", "vcs", [vcs, "push"]),

            verify,

        ]

    return [

        Step("edit.local_change", "edit", shell("echo 'local change' >> work.txt")),

        # oak commit is a local checkpoint. The publish failure is measured in

        # the explicit push row below.

        Step("vcs.snapshot", "vcs", [vcs, "commit", "--no-verify"], (0,)),

        Step("error.push_diverged", "vcs", [vcs, "push"], (0, 1, 5), recovery_hint="pull"),

        # The push error says "Pull changes first" — but pull also exits 5 on

        # a diverged branch and suggests only the destructive pull --force.

        Step("error.pull_diverged", "vcs", [vcs, "pull"], (5,), recovery_hint="pull --force"),

        # pull --force discards the local commit (observed: committed content

        # reverts). Recovery therefore includes redoing the lost work.

        Step("recovery.pull_force", "vcs", [vcs, "pull", "--force"]),

        Step("edit.redo_change", "edit", shell("echo 'local change' >> work.txt")),

        *snapshot_steps(subject, "vcs.snapshot.redo", "local change redo"),

        Step("recovery.push_after", "vcs", [vcs, "push"]),

        verify,

    ]

def sync_pull_upstream_steps(subject: Subject, track: str, ctx: dict[str, str]) -> list[Step]:

    # The everyday task-start motion: upstream moved, local tree clean.

    vcs = str(subject.bin)

    sem = semantics()

    return [

        Step("vcs.status", "vcs", [vcs, *sem.status_args(subject.kind, track)]),

        Step("sync.pull_upstream", "vcs", [vcs, "pull"]),

        Step("verify.payload", "read", shell("grep -q 'upstream change' notes.txt")),

    ]

def sync_amend_prepare_steps(subject: Subject, track: str, ctx: dict[str, str]) -> list[Step]:

    """Untimed: link the repo to a remote and push — no second actor. The

    divergence in this workflow is self-inflicted (history rewrite after

    push), so upstream never moves."""

    vcs = shlex.quote(str(subject.bin))

    if subject.kind == "git":

        rdir = shlex.quote(ctx["remote_dir"])

        return [

            Step("setup.remote.create", "vcs", shell(f"{vcs} init -q --bare {rdir}")),

            Step(

                "setup.remote.link",

                "vcs",

                shell(f"{vcs} remote add origin {rdir} && {vcs} push -q -u origin HEAD"),

            ),

        ]

    return oak_clone_a_steps(subject, ctx)

def sync_non_ff_after_amend_steps(subject: Subject, track: str, ctx: dict[str, str]) -> list[Step]:

    # The self-inflicted trap: fix the last commit after it was pushed. Git

    # reproduces it exactly (amend rewrites the pushed tip; push is rejected;

    # the error text suggests pull — which would merge the old tip back — not

    # the --force-with-lease an agent actually needs, so the actionability

    # check is expected to FAIL and that failure is the finding). Oak cannot

    # rewrite history (commits carry no messages; `oak desc` edits the branch

    # description), so the same intent is a follow-up commit with no trap:

    # zero error rows and zero recovery steps IS oak's measurement here.

    vcs = str(subject.bin)

    if subject.kind == "git":

        return [

            Step("edit.fix_last_commit", "edit", shell("echo 'amended line' >> work.txt")),

            Step(

                "vcs.amend",

                "vcs",

                shell(f"{shlex.quote(vcs)} add . && {shlex.quote(vcs)} commit -q --amend --no-edit"),

            ),

            Step("error.push_non_ff", "vcs", [vcs, "push"], (1,), recovery_hint="--force-with-lease"),

            Step("recovery.push_force_lease", "vcs", [vcs, "push", "--force-with-lease"]),

            Step("verify.payload", "read", shell("grep -q 'amended line' work.txt")),

        ]

    return [

        Step("edit.fix_last_commit", "edit", shell("echo 'amended line' >> work.txt")),

        *snapshot_steps(subject, "vcs.snapshot", "amended line"),

        Step("recovery_free.desc_update", "vcs", [vcs, "desc", "amended: follow-up fix"]),

        Step("recovery_free.push_after", "vcs", [vcs, "push"], (0,)),

        Step("verify.payload", "read", shell("grep -q 'amended line' work.txt")),

    ]

def sync_pull_dirty_steps(subject: Subject, track: str, ctx: dict[str, str]) -> list[Step]:

    # Same motion with uncommitted local edits to a different file — the

    # working state agents are usually in when they remember to sync.

    # The oracle is split so a failure names what was lost. The pull row

    # accepts both refusal and success contracts; the follow-up payload checks

    # report whether upstream and local dirty content survived.

    vcs = str(subject.bin)

    sem = semantics()

    return [

        Step("edit.dirty_change", "edit", shell("echo 'dirty local edit' >> work.txt")),

        Step("vcs.status", "vcs", [vcs, *sem.status_args(subject.kind, track)]),

        Step("sync.pull_dirty", "vcs", [vcs, "pull"], (0, 1, 2, 4, 5)),

        Step("verify.upstream_payload", "read", shell("grep -q 'upstream change' notes.txt")),

        Step("verify.local_payload", "read", shell("grep -q 'dirty local edit' work.txt")),

    ]

WORKFLOWS: dict[str, Workflow] = {

    "bugfix_test_loop": Workflow(

        "bugfix_test_loop",

        "Search, reproduce failing test, patch one source file, rerun tests, inspect and snapshot.",

        make_bugfix_fixture,

        bugfix_steps,

    ),

    "wide_config_refactor": Workflow(

        "wide_config_refactor",

        "Search and rewrite hundreds of small config files, validate, inspect and snapshot.",

        make_wide_config_fixture,

        wide_config_steps,

    ),

    "large_asset_manifest": Workflow(

        "large_asset_manifest",

        "Inspect large binary asset repo, edit small manifest only, validate, inspect and snapshot.",

        make_large_asset_fixture,

        large_asset_steps,

    ),

    "history_archaeology": Workflow(

        "history_archaeology",

        "Answer common history questions: recent commits, last change detail, file history, term search.",

        make_history_fixture,

        history_archaeology_steps,

        prepare=history_prepare_steps,

    ),

    "vcs_error_recovery": Workflow(

        "vcs_error_recovery",

        "Hit common VCS errors (empty commit, unknown subcommand), then recover; measures error-output token cost.",

        make_history_fixture,

        error_recovery_steps,

    ),

    "sync_push_divergence": Workflow(

        "sync_push_divergence",

        "Push rejected because upstream moved, then the full recovery arc. "

        "Cross-subject evidence is tokens/calls/steps, never latency: git syncs a "

        "local bare remote, oak a real server.",

        make_sync_fixture,

        sync_push_divergence_steps,

        prepare=sync_prepare_steps,

        remote_purpose="sync",

    ),

    "sync_pull_upstream": Workflow(

        "sync_pull_upstream",

        "Clean-tree pull with upstream changes — the task-start sync motion. "

        "Latency is transport-dependent; compare tokens and calls only.",

        make_sync_fixture,

        sync_pull_upstream_steps,

        prepare=sync_prepare_steps,

        remote_purpose="sync",

    ),

    "sync_pull_dirty": Workflow(

        "sync_pull_dirty",

        "Pull with upstream changes while holding uncommitted local edits. "

        "Latency is transport-dependent; compare tokens and calls only.",

        make_sync_fixture,

        sync_pull_dirty_steps,

        prepare=sync_prepare_steps,

        remote_purpose="sync",

    ),

    "sync_non_ff_after_amend": Workflow(

        "sync_non_ff_after_amend",

        "Fix the last commit after pushing. Git: amend rewrites the pushed tip, "

        "push is rejected, recovery needs --force-with-lease (which the error "

        "text does not suggest). Oak: history rewrite does not exist, so the "

        "same intent is a trap-free follow-up commit — zero recovery steps is "

        "the measurement, not a missing row.",

        make_sync_fixture,

        sync_non_ff_after_amend_steps,

        prepare=sync_amend_prepare_steps,

        remote_purpose="sync",

    ),

}

def sync_context(

    subject: Subject,

    workflow: Workflow,

    run_index: int,

    metadata: dict[str, Any],

) -> tuple[dict[str, str] | None, str | None, dict[str, Any]]:

    """Per-run remote context for remote_purpose workflows.

    Returns (ctx, skip_reason, remote_row_fields). ctx is None when the

    subject's remote is not configured; the lane emits a skip row instead.

    """

    branch = oakbench_remotes.disposable_branch(

        str(metadata.get("bench_id", "bench")), workflow.name, subject.name, run_index

    )

    ctx: dict[str, str] = {

        "branch": branch,

        "remote_dir": f"../remote-run-{run_index}.git",

        "clone_b": f"../clone-b-run-{run_index}",

    }

    if subject.kind != "git":

        resolution = oakbench_remotes.resolve_oak_remote(workflow.remote_purpose or "default")

        if not resolution.resolved:

            return None, resolution.skip_reason, {}

        capability_skip = oak_commit_push_skip_reason(subject)

        if capability_skip is not None:

            return None, capability_skip, resolution.row_fields()

        ctx["oak_repo"] = resolution.repo or ""

        # Oak topology is clone-based: every branch must descend from the

        # disposable repo's seeded main (locally-init'ed history is invisible

        # to other clones). Measured steps run inside clone A, the checkout

        # that hits the trap; the local fixture copy only seeds main.

        ctx["clone_a"] = f"../clone-a-run-{run_index}"

        ctx["repo_dir"] = f"run-{run_index}"

        ctx["measure_dir"] = ctx["clone_a"]

        return ctx, None, resolution.row_fields()

    return ctx, None, {

        "remote_repo": ctx["remote_dir"],

        "remote_transport": oakbench_remotes.TRANSPORT_LOCAL_FILE,

        "remote_server": "local",

    }

def workflow_skip_row(

    subject: Subject,

    workflow: Workflow,

    run_index: int,

    metadata: dict[str, Any],

    reason: str,

) -> dict[str, Any]:

    return {

        **metadata,

        "subject": subject.name,

        "subject_kind": subject.kind,

        "subject_label": subject.label,

        "scenario": workflow.name,

        "workflow": workflow.name,

        "workflow_description": workflow.description,

        "run": run_index,

        "phase": "workflow",

        "operation": "workflow.skipped",

        "step_kind": "summary",

        "elapsed_ms": 0.0,

        "returncode": 77,

        "process_returncode": 77,

        "command": [],

        "skipped": True,

        "skip_reason": reason,

        "tool_call_count": 0,

        "terminal_tool_call_count": 0,

        "vcs_tool_call_count": 0,

        "test_tool_call_count": 0,

    }

def load_tier_skip_rows(

    subjects: list[Subject],

    workflows: list[Workflow],

    runs: int,

    metadata: dict[str, Any],

    reason: str,

    load_tier: str,

) -> list[dict[str, Any]]:

    """Explicit skip rows for a load tier the host cannot apply.

    One row per workflow x run x subject; environment_suspect is None because

    the suspicion machinery never ran (ADR-0002), and load_tier records the

    REQUESTED tier — load that was not applied is never claimed.

    """

    rows: list[dict[str, Any]] = []

    for workflow in workflows:

        for run_index in range(runs):

            for subject in subjects:

                row = workflow_skip_row(subject, workflow, run_index, metadata, reason)

                row["load_tier"] = load_tier

                row["environment_suspect"] = None

                rows.append(row)

    return rows

def interaction_metrics(

    command: list[str],

    stdout_text: str,

    stderr_text: str,

    stdout_bytes: int,

    stderr_bytes: int,

    stdout_truncated: bool,

    stderr_truncated: bool,

    step: Step,

) -> dict[str, Any]:

    is_vcs = 1 if step.kind == "vcs" else 0

    is_test = 1 if step.kind == "test" else 0

    return {

        "tool_call_count": 1,

        "terminal_tool_call_count": 1,

        "vcs_tool_call_count": is_vcs,

        "test_tool_call_count": is_test,

        **oakbench_tokens.interaction_token_fields(

            command_display(command),

            stdout_text,

            stderr_text,

            stdout_bytes,

            stderr_bytes,

            stdout_truncated,

            stderr_truncated,

        ),

        "tool_calls": {

            "total": 1,

            "terminal": 1,

            "vcs": is_vcs,

            "test": is_test,

            step.kind: 1,

        },

    }

def run_step(

    subject: Subject,

    workflow: Workflow,

    step: Step,

    cwd: Path,

    run_index: int,

    phase: str,

    metadata: dict[str, Any],

    admitted_output_chars: int,

) -> dict[str, Any]:

    capture = run_timed(step.command, cwd, step_env(step), admitted_output_chars)

    expected = capture.returncode in step.expected_returncodes

    row = {

        **metadata,

        "subject": subject.name,

        "subject_kind": subject.kind,

        "subject_label": subject.label,

        "scenario": workflow.name,

        "workflow": workflow.name,

        "workflow_description": workflow.description,

        "run": run_index,

        "phase": phase,

        "operation": step.operation,

        "step_kind": step.kind,

        "elapsed_ms": round(capture.elapsed_ms, 3),

        "returncode": 0 if expected else (capture.returncode or 1),

        "process_returncode": capture.returncode,

        "expected_returncodes": list(step.expected_returncodes),

        "command": step.command,

        **interaction_metrics(

            step.command,

            capture.stdout_text,

            capture.stderr_text,

            capture.stdout_bytes,

            capture.stderr_bytes,

            capture.stdout_truncated,

            capture.stderr_truncated,

            step,

        ),

    }

    if not expected:

        row["stderr"] = capture.stderr_text[-4000:]

    if step.recovery_hint is not None:

        combined = (capture.stdout_text + capture.stderr_text).lower()

        row["recovery_hint"] = step.recovery_hint

        row["error_mentions_recovery_command"] = step.recovery_hint.lower() in combined

        row["error_actionability_source"] = "output_contains_recovery_command"

    return row

def snapshot_step_requires_state_change(operation: str) -> bool:

    return operation in {"setup.snapshot", "vcs.snapshot", "vcs.snapshot.redo"} or (

        ".snapshot" in operation and operation.endswith(".commit")

    )

def branch_step_requires_state_change(operation: str) -> bool:

    return operation == "vcs.branch"

def apply_state_change_attestation(

    row: dict[str, Any],

    repo: Path,

    before_fingerprint: str | None,

    operation: str,

    *,

    expected_hashes: dict[str, str] | None = None,

    expected_payloads: dict[str, str | bytes] | None = None,

) -> dict[str, Any]:

    raw_process_returncode = row.get("process_returncode")

    if raw_process_returncode is None:

        process_returncode = row_returncode(row)

    else:

        try:

            process_returncode = int(raw_process_returncode)

        except (TypeError, ValueError):

            process_returncode = 1

    fields = oakbench_integrity.state_change_fields(

        repo,

        before_fingerprint,

        operation=operation,

        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=operation,

            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 workflow_step_payload_requirements(workflow: Workflow, operation: str) -> list[tuple[str, str]]:

    """Content payloads required at an intermediate step.

    Final workflow integrity can require the fully converged payload. Snapshot

    steps must only require payloads that are expected to exist in that phase.

    """

    if workflow.name == "sync_push_divergence" and operation in {

        "vcs.snapshot",

        "vcs.snapshot.commit",

        "vcs.snapshot.redo",

        "vcs.snapshot.redo.commit",

    }:

        return [("work.txt", "local change")]

    return []

def apply_content_payload_attestation(

    row: dict[str, Any],

    repo: Path,

    expected_payloads: list[tuple[str, str]],

) -> dict[str, Any]:

    if not expected_payloads:

        return row

    fields = oakbench_integrity.content_payload_fields(repo, expected_payloads)

    row.update(fields)

    if row_returncode(row) == 0 and not fields["content_payload_check_passed"]:

        reason = str(fields["content_payload_failure_reason"] or "content_payload_check_failed")

        row["returncode"] = 1

        row["process_returncode"] = 1

        row["stderr"] = reason

        row["integrity_check_passed"] = False

        row["integrity_failure_reason"] = reason

        row["integrity_check_source"] = "content_payload_attestation"

    return row

def run_attested_step(

    subject: Subject,

    workflow: Workflow,

    step: Step,

    cwd: Path,

    run_index: int,

    phase: str,

    metadata: dict[str, Any],

    admitted_output_chars: int,

) -> dict[str, Any]:

    requires_state_change = snapshot_step_requires_state_change(step.operation) or branch_step_requires_state_change(

        step.operation

    )

    before = oakbench_integrity.repository_state_fingerprint(cwd) if requires_state_change else None

    expected_hashes = None

    expected_payloads = None

    if snapshot_step_requires_state_change(step.operation):

        expected_hashes = oakbench_integrity.sample_worktree_payload_hashes(cwd)

        if phase == "workflow":

            expected_payloads = workflow_snapshot_payload_expectations(workflow)

    row = run_step(subject, workflow, step, cwd, run_index, phase, metadata, admitted_output_chars)

    if requires_state_change:

        apply_state_change_attestation(

            row,

            cwd,

            before,

            step.operation,

            expected_hashes=expected_hashes,

            expected_payloads=expected_payloads,

        )

    apply_content_payload_attestation(

        row,

        cwd,

        workflow_step_payload_requirements(workflow, step.operation),

    )

    return row

def _read_text(path: Path) -> str | None:

    try:

        return path.read_text(errors="replace")

    except OSError:

        return None

def workflow_snapshot_payload_expectations(workflow: Workflow) -> dict[str, str]:

    return workflow_oracles.COMMITTED_PAYLOADS.get(workflow.name, {

        "wide_config_refactor": {"services/service-0000/config.toml": "timeout_ms = 3000"},

        "large_asset_manifest": {"manifest.json": '\"reviewed\": true'},

    }.get(workflow.name, {}))

def workflow_worktree_payload_failure(workflow: Workflow, repo: Path) -> str | None:

    name = workflow.name

    if name == "bugfix_test_loop":

        text = _read_text(repo / "app" / "pricing.py")

        expected = "return cents - ((cents * percent) // 100)"

        return None if text is not None and expected in text else "missing bugfix payload in app/pricing.py"

    if name in workflow_oracles.SCENARIOS:

        return workflow_oracles.validate(name, repo)

    if name == "vcs_error_recovery":

        text = _read_text(repo / "README.md")

        return None if text is not None and "# recovered" in text else "recovery payload missing in README.md"

    if name == "sync_push_divergence":

        work = _read_text(repo / "work.txt") or ""

        notes = _read_text(repo / "notes.txt") or ""

        if "local change" not in work:

            return "local pushed payload missing from work.txt"

        if "upstream change" not in notes:

            return "upstream pulled payload missing from notes.txt"

        return None

    if name == "sync_pull_upstream":

        notes = _read_text(repo / "notes.txt") or ""

        return None if "upstream change" in notes else "upstream pulled payload missing from notes.txt"

    if name == "sync_pull_dirty":

        work = _read_text(repo / "work.txt") or ""

        notes = _read_text(repo / "notes.txt") or ""

        if "dirty local edit" not in work:

            return "dirty local payload missing from work.txt"

        if "upstream change" not in notes:

            return "upstream pulled payload missing from notes.txt"

        return None

    if name == "sync_non_ff_after_amend":

        work = _read_text(repo / "work.txt") or ""

        return None if "amended line" in work else "amended payload missing from work.txt"

    return None

def _git_head_text(repo: Path, path: str, git_binary: str | None = None, resolved_instrument=None) -> tuple[str | None, str | None]:

    return workflow_oracles.git_head_text(repo, path, git_binary, resolved_instrument=resolved_instrument)

def workflow_git_head_payload_failure(workflow: Workflow, repo: Path, git_binary: str | None = None, resolved_instrument=None) -> str | None:

    if workflow.name in workflow_oracles.SCENARIOS:

        return workflow_oracles.validate_git_head(workflow.name, repo, git_binary, resolved_instrument=resolved_instrument)

    checks = workflow_snapshot_payload_expectations(workflow)

    if not checks:

        return None

    for path, expected in checks.items():

        text, error = _git_head_text(repo, path, git_binary, resolved_instrument)

        if error is not None:

            return error

        if text is None or expected not in text:

            return f"committed_payload_missing:{path}"

    return None

def workflow_integrity_fields(

    subject: Subject,

    workflow: Workflow,

    repo: Path,

    workflow_rows: list[dict[str, Any]],

) -> dict[str, Any]:

    failures = [row for row in workflow_rows if row_returncode(row) != 0]

    snapshot_failures = [

        row

        for row in workflow_rows

        if row.get("integrity_check_passed") is False or row.get("content_integrity_check_passed") is False

    ]

    if workflow.name in workflow_oracles.SCENARIOS and not failures and not snapshot_failures:

        evidence = workflow_oracles.task_evidence(workflow.name, repo, subject.kind, str(subject.bin))

        return {**evidence, "correctness_passed": evidence["workflow_integrity_passed"],

                "correctness_failure_reason": evidence["workflow_integrity_failure_reason"]}

    reason: str | None = None

    source = "trusted_exact_task_worktree" if workflow.name in workflow_oracles.SCENARIOS else "worktree_payload_oracle"

    if failures:

        first = failures[0]

        reason = str(

            first.get("integrity_failure_reason")

            or first.get("content_integrity_failure_reason")

            or f"workflow_step_failed:{first.get('operation')}"

        )

    elif snapshot_failures:

        first = snapshot_failures[0]

        reason = str(

            first.get("integrity_failure_reason")

            or first.get("content_integrity_failure_reason")

            or f"snapshot_integrity_failed:{first.get('operation')}"

        )

    if reason is None:

        reason = workflow_worktree_payload_failure(workflow, repo)

    reader_fields = {}

    evidence_class = "worktree_payload"

    if reason is None and subject.kind == "git" and workflow_snapshot_payload_expectations(workflow):

        instrument, reader_fields = workflow_oracles.resolve_git_instrument()

        reason = "git_head_oracle_unavailable" if instrument is None else workflow_git_head_payload_failure(workflow, repo, resolved_instrument=(instrument, reader_fields))

        if instrument is not None and reason is not None:

            evidence_class = "committed_payload_attempt"

            source += ";git_head_payload_read_attempt"

        if reason is None:

            reader_fields.update(workflow_oracles.reader_evidence("git", str(subject.bin), reader_fields))

            evidence_class = reader_fields.pop("workflow_integrity_evidence_class").replace("_tree", "_payload")

            source += ";" + evidence_class

    elif reason is None and subject.kind == "oak" and workflow_snapshot_payload_expectations(workflow):

        # Same committed-payload check as the Git arm, read by the independent

        # stdlib Oak reader (no Oak code or binary).

        from oakbench import oak_sqlite_reader as oak_reader

        checks = workflow_snapshot_payload_expectations(workflow)

        try:

            payloads, meta = oak_reader.committed_payloads(repo, list(checks))

        except oak_reader.Invalid as exc:

            reason = f"oak_stored_content_invalid:{exc}"

            evidence_class, meta = "committed_payload_attempt", {}

        except (oak_reader.Unavailable, OSError, ValueError) as exc:

            reason, meta = f"oak_reader_unavailable:{exc}", {}

        else:

            for path, expected in checks.items():

                data = payloads.get(path)

                if data is None or expected not in data.decode("utf-8", "replace"):

                    reason = f"committed_payload_missing:{path}"

                    break

            evidence_class = "independent_committed_payload" if reason is None else "committed_payload_attempt"

        source += ";independent_oak_reader_payload_read"

        reader_fields = {"oak_reader_version": oak_reader.READER_VERSION, "oak_reader_head": meta.get("head"),

                         "oak_reader_backend": meta.get("reader_backend")}

    elif reason is None and any(row.get("integrity_check_passed") is True for row in workflow_rows):

        source += ";direct_repo_metadata_fingerprint"

        evidence_class = "repository_metadata"

    unavailable = (reason in {"git_head_oracle_unavailable", "known_good_git_unavailable_for_head_payload_check"}

                   or str(reason).startswith("oak_reader_unavailable:"))

    passed = None if unavailable else reason is None

    return {

        "oracle_version": workflow_oracles.VERSION,

        "workflow_integrity_passed": passed,

        "workflow_integrity_failure_reason": reason,

        "workflow_integrity_source": source,

        **reader_fields,

        "workflow_integrity_evidence_class": evidence_class,

        "workflow_integrity_unmeasured_reason": reason if unavailable else None,

        "correctness_passed": passed,

        "correctness_failure_reason": reason,

    }

def recovery_metrics_row(

    workflow_rows: list[dict[str, Any]],

    metadata: dict[str, Any],

    subject: Subject,

    workflow: Workflow,

    run_index: int,

) -> dict[str, Any] | None:

    """Derived recovery-cost row for workflows that hit expected failures.

    Error output is a prompt: the rejected command's text is what the agent

    reasons from. This row prices that text (model-input tokens) and the

    recovery arc that follows it. tool_call_count is 0 — the row performs no

    calls and must not perturb tool-call or token deltas (bench.py probe-row

    precedent).

    """

    error_rows = [r for r in workflow_rows if str(r.get("operation", "")).startswith("error.")]

    recovery_prefixes = ("recovery.", "edit.redo", "vcs.snapshot.redo")

    recovery_free_prefixes = ("recovery_free.",)

    recovery_rows = [

        r for r in workflow_rows if str(r.get("operation", "")).startswith(recovery_prefixes)

    ]

    recovery_free_rows = [

        r for r in workflow_rows if str(r.get("operation", "")).startswith(recovery_free_prefixes)

    ]

    if not error_rows and not recovery_rows and not recovery_free_rows:

        return None

    hinted = [r for r in error_rows if "error_mentions_recovery_command" in r]

    actionable = [r for r in hinted if r.get("error_mentions_recovery_command")]

    return {

        **metadata,

        "subject": subject.name,

        "subject_kind": subject.kind,

        "subject_label": subject.label,

        "scenario": workflow.name,

        "workflow": workflow.name,

        "workflow_description": workflow.description,

        "run": run_index,

        "phase": "workflow",

        "operation": "sync.recovery_metrics",

        "step_kind": "summary",

        "elapsed_ms": 0.0,

        "returncode": 0,

        "process_returncode": 0,

        "command": [],

        "tool_call_count": 0,

        "terminal_tool_call_count": 0,

        "vcs_tool_call_count": 0,

        "test_tool_call_count": 0,

        "error_steps_count": len(error_rows),

        # The failure text an agent must ingest before it can act, priced as

        # model input (output of the tool, input to the model).

        "error_output_tokens": sum(int(r.get("estimated_tokens_output", 0)) for r in error_rows),

        "recovery_steps_count": len(recovery_rows),

        "recovery_wall_ms": round(sum(float(r.get("elapsed_ms", 0)) for r in recovery_rows), 3),

        "recovery_tokens_total": round(

            sum(float(r.get("estimated_cost_weighted_tokens_with_envelope", 0)) for r in recovery_rows), 1

        ),

        "error_steps_with_hint": len(hinted),

        "error_steps_hint_actionable": len(actionable),

        # Null when no step declared a hint: unmeasured, never a fabricated pass.

        "error_actionability": (len(actionable) == len(hinted)) if hinted else None,

        "measurement_source": "derived_from_workflow_rows; hint check is output_contains_recovery_command",

        "summarized_operations": [r["operation"] for r in error_rows + recovery_rows + recovery_free_rows],

    }

def sum_rows(

    rows: list[dict[str, Any]],

    metadata: dict[str, Any],

    subject: Subject,

    workflow: Workflow,

    run_index: int,

    phase: str,

    operation: str,

    extra_fields: dict[str, Any] | None = None,

) -> dict[str, Any]:

    elapsed_ms = sum(float(row.get("elapsed_ms", 0)) for row in rows)

    tool_calls = sum(oakbench_tokens.int_or_zero(row.get("tool_call_count")) for row in rows)

    terminal_calls = sum(oakbench_tokens.int_or_zero(row.get("terminal_tool_call_count")) for row in rows)

    vcs_calls = sum(oakbench_tokens.int_or_zero(row.get("vcs_tool_call_count")) for row in rows)

    test_calls = sum(oakbench_tokens.int_or_zero(row.get("test_tool_call_count")) for row in rows)

    failures = [row for row in rows if row_returncode(row) != 0]

    steps_total = len(rows)

    steps_failed = len(failures)

    result = {

        **metadata,

        "subject": subject.name,

        "subject_kind": subject.kind,

        "subject_label": subject.label,

        "scenario": workflow.name,

        "workflow": workflow.name,

        "workflow_description": workflow.description,

        "run": run_index,

        "phase": phase,

        "operation": operation,

        "step_kind": "summary",

        "elapsed_ms": round(elapsed_ms, 3),

        "returncode": 1 if failures else 0,

        "process_returncode": 1 if failures else 0,

        "command": [row["command"] for row in rows],

        "tool_call_count": tool_calls,

        "terminal_tool_call_count": terminal_calls,

        "vcs_tool_call_count": vcs_calls,

        "test_tool_call_count": test_calls,

        **oakbench_tokens.summed_token_fields(

            rows, "sum_of_workflow_steps_command_plus_admitted_output_chars_div_4"

        ),

        "tool_calls": {

            "total": tool_calls,

            "terminal": terminal_calls,

            "vcs": vcs_calls,

            "test": test_calls,

        },

        "steps_total": steps_total,

        "steps_succeeded": steps_total - steps_failed,

        "steps_failed": steps_failed,

        "step_failure_rate": (steps_failed / steps_total) if steps_total else None,

        "summarized_operations": [row["operation"] for row in rows],

    }

    if extra_fields:

        result.update(extra_fields)

        if extra_fields.get("workflow_integrity_passed") is False:

            result["returncode"] = 1

            result["process_returncode"] = 1

        elif extra_fields.get("workflow_integrity_unmeasured_reason"):

            result["returncode"] = 77

            result["skipped"] = True

            result["skip_reason"] = extra_fields["workflow_integrity_unmeasured_reason"]

    return result

def copy_fixture(src: Path, dest: Path) -> None:

    oakbench_fixtures.copy_fixture(src, dest)

def prepare_fixture(root: Path, workflow: Workflow) -> Path:

    fixture = root / "fixtures" / workflow.name

    marker = fixture / ".fixture-ready"

    if marker.exists() and marker.read_text().strip() == FIXTURE_VERSION:

        return fixture

    if fixture.exists():

        shutil.rmtree(fixture)

    fixture.mkdir(parents=True)

    workflow.make_fixture(fixture)

    marker.write_text(FIXTURE_VERSION + "\n")

    return fixture

def run_subject_workflow(

    subject: Subject,

    workflow: Workflow,

    fixture: Path,

    run_index: int,

    run_root: Path,

    metadata: dict[str, Any],

    admitted_output_chars: int,

    track: str,

) -> list[dict[str, Any]]:

    repo = run_root / workflow.name / subject.name / f"run-{run_index}"

    copy_fixture(fixture, repo)

    ctx: dict[str, str] | None = None

    if workflow.remote_purpose is not None:

        ctx, skip_reason, remote_fields = sync_context(subject, workflow, run_index, metadata)

        if ctx is None:

            return [workflow_skip_row(subject, workflow, run_index, metadata, skip_reason or "")]

        metadata = {**metadata, **remote_fields}

    setup_steps = list(vcs_init_steps(subject))

    if workflow.prepare is not None:

        setup_steps.extend(

            workflow.prepare(subject, track) if ctx is None else workflow.prepare(subject, track, ctx)

        )

    setup_rows = [

        run_attested_step(subject, workflow, step, repo, run_index, "setup", metadata, admitted_output_chars)

        for step in setup_steps

    ]

    workflow_rows: list[dict[str, Any]] = []

    if any(int(row["returncode"]) != 0 for row in setup_rows):

        return setup_rows + [

            sum_rows(setup_rows, metadata, subject, workflow, run_index, "setup", "setup.total"),

        ]

    # Oak remote workflows measure inside clone A (the trap-hitting checkout);

    # everything else measures in the fixture repo itself.

    measure_dir = repo

    if ctx is not None and "measure_dir" in ctx:

        measure_dir = (repo / ctx["measure_dir"]).resolve()

    for step in workflow.steps(subject, track) if ctx is None else workflow.steps(subject, track, ctx):

        row = run_attested_step(

            subject, workflow, step, measure_dir, run_index, "workflow", metadata, admitted_output_chars

        )

        workflow_rows.append(row)

        if int(row["returncode"]) != 0:

            break

    derived = recovery_metrics_row(workflow_rows, metadata, subject, workflow, run_index)

    integrity = workflow_integrity_fields(subject, workflow, measure_dir, workflow_rows)

    return [

        *setup_rows,

        sum_rows(setup_rows, metadata, subject, workflow, run_index, "setup", "setup.total"),

        *workflow_rows,

        sum_rows(

            workflow_rows,

            metadata,

            subject,

            workflow,

            run_index,

            "workflow",

            "workflow.total",

            integrity,

        ),

        *([derived] if derived else []),

    ]

def select_workflows(raw: str) -> list[Workflow]:

    if raw == "all":

        return list(WORKFLOWS.values())

    names = [item.strip() for item in raw.split(",") if item.strip()]

    missing = [name for name in names if name not in WORKFLOWS]

    if missing:

        raise SystemExit("Unknown workflows: " + ", ".join(missing))

    return [WORKFLOWS[name] for name in names]

def remote_preflight_requirements(

    subjects: list[Subject],

    workflows: list[Workflow],

) -> list[oakbench_remotes.RemoteRequirement]:

    if not any(subject.kind == "oak" for subject in subjects):

        return []

    grouped: dict[str, list[str]] = {}

    for workflow in workflows:

        if workflow.remote_purpose is None:

            continue

        grouped.setdefault(workflow.remote_purpose, []).append(f"{workflow.name}/workflow.skipped")

    return [

        oakbench_remotes.RemoteRequirement(

            purpose=purpose,

            lane="workflow",

            operations=tuple(operations),

        )

        for purpose, operations in sorted(grouped.items())

    ]

def metric_average(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.get("operation") != "workflow.total" or row_returncode(row) != 0:

            continue

        try:

            value = float(row[metric])

        except (KeyError, TypeError, ValueError):

            continue

        grouped.setdefault((str(row["subject"]), str(row["workflow"])), []).append(value)

    return {key: statistics.mean(values) for key, values in grouped.items()}

def metric_transport(rows: list[dict[str, Any]]) -> dict[tuple[str, str], tuple[str | None, str | None]]:

    transports: dict[tuple[str, str], set[tuple[str | None, str | None]]] = {}

    for row in rows:

        if row.get("operation") != "workflow.total" or row_returncode(row) != 0:

            continue

        key = (str(row["subject"]), str(row["workflow"]))

        transport = row.get("remote_transport") or row.get("workspace_transport")

        server = row.get("remote_server") or row.get("workspace_server")

        transports.setdefault(key, set()).add(

            (

                str(transport) if transport is not None else None,

                str(server) if server is not None else None,

            )

        )

    result: dict[tuple[str, str], tuple[str | None, str | None]] = {}

    for key, values in transports.items():

        if len(values) == 1:

            result[key] = next(iter(values))

        else:

            result[key] = ("mixed", "mixed")

    return result

def workflow_total_stats(rows: list[dict[str, Any]]) -> dict[tuple[str, str], dict[str, int | float]]:

    grouped: dict[tuple[str, str], list[dict[str, Any]]] = {}

    for row in rows:

        if row.get("operation") != "workflow.total":

            continue

        grouped.setdefault((str(row["subject"]), str(row["workflow"])), []).append(row)

    stats: dict[tuple[str, str], dict[str, int | float]] = {}

    for key, total_rows in grouped.items():

        runs = len(total_rows)

        failures = sum(1 for row in total_rows if row_returncode(row) != 0)

        stats[key] = {

            "runs": runs,

            "successes": runs - failures,

            "failures": failures,

            "failure_rate": failures / runs if runs else 0.0,

        }

    return stats

def lower_is_better_pct(base: float | None, value: float | None) -> str:

    return fmt_delta_lower_better(base, value)

def comparable_time_delta(

    base: float | None,

    value: float | None,

    base_transport: tuple[str | None, str | None] | None,