scripts/agent_workflow.py · oak/benchmarks

Viewing files at commit 1a441edafc71Commit details View latest
Browse files
scripts/agent_workflow.py 2148 lines · 90.2 KB · python
#!/usr/bin/env python3
"""Run real local coding-agent workflows against Git and Oak subjects.

This runner is intentionally separate from workflow_ab.py. workflow_ab.py runs
deterministic agent-shaped command recipes; this file runs actual CLI coding
agents such as Codex, Claude Code, and Cursor Agent, then normalizes their
transcripts into scripts/agent_metrics_schema.py rows.
"""

from __future__ import annotations

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

import workflow_ab
from agent_metrics_schema import SCHEMA_VERSION, validate_row
from oakbench import workflow_oracles
from oakbench.agent_execution import DEFAULT_OUTPUT_LIMIT_BYTES, EXECUTION_VERSION, run_captured
from oakbench.environment import ENV_ISOLATION_VERSION, base_env as shared_base_env
from oakbench.runner import runner_fields
from oakbench.cachectl import cache_fields
from oakbench.agent_environment import agent_environment as filtered_agent_environment
from oakbench.classify import classify_command, unwrap_shell_command, vcs_invocations
from oakbench.command_semantics import semantics
from oakbench.configio import load_toml
from oakbench.results import ResultsStore
from oakbench.rows import token_total as shared_token_total
from oakbench.stream_adapters import adapter_for, analyze_stream, first_number, walk_json
from oakbench.thrash import agent_blocked_on_vcs_ms, thrash_events, vcs_share_of_task_wall
from oakbench.vcs_shim import create_shim_dir, measure_shim_overhead, read_sidecar_rows
from oakbench.tokens import (
    CACHE_READ_TOKEN_WEIGHT,
    COST_WEIGHTS_NOTE,
    EMITTED_TOKEN_WEIGHT,
    INGESTED_TOKEN_WEIGHT,
)


ROOT = Path(__file__).resolve().parents[1]
DEFAULT_WORKDIR = Path(tempfile.gettempdir()) / "oak-agent-workflow"
PROMPT_VERSION = "2026-06-09.2"
ATTRIBUTION_VERSION = "2026-09-18.1"
ORACLE_VERSION = workflow_oracles.VERSION
COMMAND_SEMANTICS_VERSION = semantics().version
TRACKS = ("agent-default", "core-equivalent")
INSTRUCTION_LEVELS = ("zero-shot", "cheat-sheet", "full-docs")
VCS_DIR_NAMES = {".git", ".oak", ".hg", ".svn"}
GENERATED_ARTIFACT_DIR_NAMES = {"__pycache__"}
GENERATED_ARTIFACT_SUFFIXES = (".pyc", ".pyo")


@dataclass(frozen=True)
class AgentConfig:
    name: str
    adapter: str
    provider: str | None
    bin: Path | None
    model: str | None
    version: str | None


@dataclass(frozen=True)
class AgentExecution:
    returncode: int
    timed_out: bool
    elapsed_ms: float
    stdout_path: Path
    stderr_path: Path
    prompt_path: Path
    output_limited: bool = False
    capture_complete: bool = True
    capture_metadata: dict[str, Any] | None = None


@dataclass(frozen=True)
class OracleResult:
    elapsed_ms: float
    test_ms: float
    vcs_ms: float
    success: bool
    outcome: str
    failure_reason: str | None
    checks: dict[str, bool | None]
    test_returncode: int
    test_output_bytes: int
    status_output: str
    dirty_files_at_end: int | None
    changed_files: list[str]
    bytes_written: int
    commit_delta: int
    artifact_paths: list[Path]
    evidence: dict[str, Any] | None = None


def parse_args() -> argparse.Namespace:
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument(
        "--workflows",
        default="bugfix_test_loop",
        help="Comma-separated workflow_ab workflow names, or 'all'. Default keeps real-agent runs cheap.",
    )
    parser.add_argument("--subjects", help="Comma-separated subject names from config/subjects.toml")
    parser.add_argument(
        "--agents",
        default="mock",
        help="Comma-separated agent names, 'available', or 'all'. Default is mock to avoid accidental model spend.",
    )
    parser.add_argument("--runs", type=int, default=1)
    parser.add_argument("--config", type=Path, default=ROOT / "config" / "subjects.toml")
    parser.add_argument("--agents-config", type=Path, default=ROOT / "config" / "agents.toml")
    parser.add_argument("--workdir", type=Path, default=DEFAULT_WORKDIR)
    parser.add_argument("--results", type=Path, default=ROOT / "results" / "agent-workflow")
    parser.add_argument(
        "--oak-repo",
        type=Path,
        default=workflow_ab.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("--timeout-seconds", type=float, default=900.0)
    parser.add_argument("--output-limit-bytes", type=int, default=DEFAULT_OUTPUT_LIMIT_BYTES,
                        help="Maximum combined real-agent stdout/stderr bytes retained per trial.")
    parser.add_argument(
        "--agent-environment",
        choices=("minimal", "local-default"),
        default="minimal",
        help=(
            "minimal asks CLIs to avoid user/project customizations when supported; "
            "local-default measures the normal local agent environment."
        ),
    )
    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="Command semantics prompt track. Real agent runs normally use agent-default.",
    )
    parser.add_argument(
        "--instruction-level",
        choices=INSTRUCTION_LEVELS,
        default="full-docs",
        help=(
            "How much VCS-specific instruction the prompt and AGENTS.md carry. "
            "zero-shot names the VCS but gives no commands, measuring the model-familiarity tax; "
            "cheat-sheet gives a one-line command list; full-docs is the historical default."
        ),
    )
    parser.add_argument(
        "--vcs-shim",
        action="store_true",
        help=(
            "Wrap git/oak with PATH shim executables that record per-call VCS timing "
            "to a sidecar JSONL. Shim overhead is measured and recorded per run, never subtracted."
        ),
    )
    parser.add_argument("--randomize-subject-order", action="store_true")
    parser.add_argument("--randomize-agent-order", action="store_true")
    parser.add_argument("--list-agents", action="store_true", help="Print configured/discovered agents and exit.")
    return parser.parse_args()


def resolve_bin(raw: str | None) -> Path | None:
    if not raw:
        return None
    path = Path(raw).expanduser()
    if path.is_absolute() and path.exists():
        return path
    found = shutil.which(raw)
    return Path(found) if found else path


def sanitized_agent_env(env: dict[str, str]) -> dict[str, str]:
    return filtered_agent_environment(env)


def trial_agent_env(artifacts: Path, home_root: Path | None = None) -> tuple[dict[str, str], dict[str, Any]]:
    if home_root is not None:
        home_root.mkdir(parents=True, exist_ok=True)
    home = Path(tempfile.mkdtemp(prefix="agent-home-", dir=home_root))
    env = sanitized_agent_env(shared_base_env(home=home))
    override_names = ("CODEX_HOME", "CLAUDE_CONFIG_DIR", "CURSOR_CONFIG_DIR", "SSH_AUTH_SOCK")
    return env, {
        "fresh_agent_home": str(home), "environment_policy": "fresh-home-explicit-agent-allowlist-v2",
        "preserved_config_override_names": [name for name in override_names if name in env],
        "provider_environment_preserved": True,
        "os_sandbox_enforced": False,
        "limitations": "Provider/config overrides remain usable; HOME is not filesystem or network containment. "
                       "POSIX descendants creating new sessions can escape group cleanup; other platforms own only the direct child.",
    }


def command_output(command: list[str], cwd: Path | None = None, timeout: float = 10.0) -> str | None:
    try:
        return subprocess.check_output(
            command,
            cwd=cwd,
            env=sanitized_agent_env(workflow_ab.base_env()),
            text=True,
            stderr=subprocess.STDOUT,
            timeout=timeout,
        ).strip()
    except Exception:
        return None


def agent_version(agent: AgentConfig) -> str | None:
    if agent.adapter == "deterministic_mock":
        return "deterministic-mock-" + PROMPT_VERSION
    if agent.bin is None or not agent.bin.exists():
        return None
    return command_output([str(agent.bin), "--version"])


def load_agents(path: Path, selector: str) -> list[AgentConfig]:
    data = load_toml(path)
    raw_agents = data.get("agents", {})
    configured: dict[str, AgentConfig] = {}
    for key, raw in raw_agents.items():
        if not raw.get("enabled", False):
            continue
        name = str(raw.get("name") or key)
        adapter = str(raw["adapter"])
        resolved = resolve_bin(raw.get("bin"))
        if adapter != "deterministic_mock" and (resolved is None or not resolved.exists()):
            configured[name] = AgentConfig(
                name=name,
                adapter=adapter,
                provider=raw.get("provider"),
                bin=resolved,
                model=raw.get("model") or None,
                version=None,
            )
            continue
        agent = AgentConfig(
            name=name,
            adapter=adapter,
            provider=raw.get("provider"),
            bin=resolved,
            model=raw.get("model") or None,
            version=None,
        )
        configured[name] = AgentConfig(
            name=agent.name,
            adapter=agent.adapter,
            provider=agent.provider,
            bin=agent.bin,
            model=agent.model,
            version=agent_version(agent),
        )

    if selector == "all":
        selected = list(configured.values())
    elif selector == "available":
        selected = [
            agent
            for agent in configured.values()
            if agent.adapter != "deterministic_mock" and agent.bin is not None and agent.bin.exists()
        ]
    else:
        wanted = [item.strip() for item in selector.split(",") if item.strip()]
        missing = [name for name in wanted if name not in configured]
        if missing:
            raise SystemExit("Unknown agents: " + ", ".join(missing))
        selected = [configured[name] for name in wanted]

    unavailable = [
        agent.name
        for agent in selected
        if agent.adapter != "deterministic_mock" and (agent.bin is None or not agent.bin.exists())
    ]
    if unavailable:
        raise SystemExit("Selected agents are not installed or not executable: " + ", ".join(unavailable))
    if not selected:
        raise SystemExit("No agents selected")
    return selected


def print_agents(agents: list[AgentConfig]) -> None:
    print("| Agent | Adapter | Provider | Binary | Version |")
    print("| --- | --- | --- | --- | --- |")
    for agent in agents:
        print(
            f"| `{agent.name}` | `{agent.adapter}` | `{agent.provider or ''}` | "
            f"`{agent.bin or ''}` | `{agent.version or ''}` |"
        )


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


def vcs_mode(subject: workflow_ab.Subject) -> str:
    if subject.kind == "git":
        return "git_single_checkout"
    if subject.kind == "oak":
        return "oak_single_mount"
    return "git_single_checkout"


def strip_ansi(text: str) -> str:
    return re.sub(r"\x1b\[[0-9;]*m", "", text)


def is_workspace_file(path: Path, root: Path) -> bool:
    try:
        rel = path.relative_to(root)
    except ValueError:
        return False
    rel_posix = rel.as_posix()
    return not any(part in VCS_DIR_NAMES for part in rel.parts) and not is_generated_artifact(rel_posix)


def is_generated_artifact(path: str) -> bool:
    rel = PurePosixPath(path)
    return any(part in GENERATED_ARTIFACT_DIR_NAMES for part in rel.parts) or rel.name.endswith(
        GENERATED_ARTIFACT_SUFFIXES
    )


def workspace_manifest(root: Path) -> dict[str, dict[str, Any]]:
    manifest: dict[str, dict[str, Any]] = {}
    for path in sorted(root.rglob("*")):
        if not (path.is_file() or path.is_symlink()) or not is_workspace_file(path, root):
            continue
        rel = path.relative_to(root).as_posix()
        if path.is_symlink():
            target = os.readlink(path)
            target_bytes = os.fsencode(target)
            manifest[rel] = {
                "sha256": hashlib.sha256(target_bytes).hexdigest(),
                "bytes": len(target_bytes),
                "kind": "symlink",
                "target": target,
            }
            continue
        digest = hashlib.sha256()
        size = 0
        with path.open("rb") as fh:
            for chunk in iter(lambda: fh.read(1024 * 1024), b""):
                size += len(chunk)
                digest.update(chunk)
        manifest[rel] = {
            "sha256": digest.hexdigest(),
            "bytes": size,
            "kind": "file",
            "executable": bool(path.stat().st_mode & 0o111),
        }
    return manifest


def changed_files(before: dict[str, dict[str, Any]], after: dict[str, dict[str, Any]]) -> list[str]:
    names = sorted(set(before) | set(after))
    return [name for name in names if before.get(name) != after.get(name)]


def allowed_change(workflow: workflow_ab.Workflow, path: str) -> bool:
    if workflow.name == "bugfix_test_loop":
        return path == "app/pricing.py"
    if workflow.name == "wide_config_refactor":
        return path.startswith("services/") and path.endswith("/config.toml")
    if workflow.name == "large_asset_manifest":
        return path == "manifest.json"
    if workflow.name == "history_archaeology":
        return False
    if workflow.name == "vcs_error_recovery":
        return path == "README.md"
    return True


def expected_change_ok(workflow: workflow_ab.Workflow, changed: list[str]) -> bool:
    changed = [path for path in changed if not is_generated_artifact(path)]
    # Read-only scenarios must leave the tree untouched; everything else must
    # change something, and only within the allowed shape.
    if workflow.name == "history_archaeology":
        return not changed
    return bool(changed) and all(allowed_change(workflow, path) for path in changed)


def write_workspace_instructions(
    repo: Path, subject: workflow_ab.Subject, track: str, instruction_level: str
) -> None:
    if instruction_level == "zero-shot":
        vcs_name = "Git" if subject.kind == "git" else "Oak (a version control system, not Git)"
        text = "\n".join(
            [
                "# AGENTS.md",
                "",
                f"This benchmark workspace uses {vcs_name}.",
                "Use the current checkout; do not create another workspace.",
                "Snapshot your final work with the workspace's VCS and leave the tree clean.",
                "",
            ]
        )
    elif instruction_level == "cheat-sheet":
        if subject.kind == "git":
            cheat = "`git status --short`, `git diff --stat`, `git add . && git commit -m \"agent task\"`"
            header = "This benchmark workspace uses Git."
        else:
            cheat = "`oak status`, `oak diff`, `oak commit --no-verify`"
            header = "This benchmark workspace uses Oak, not Git. Do not run `git` commands."
        text = "\n".join(
            [
                "# AGENTS.md",
                "",
                header,
                f"VCS cheat sheet: {cheat}.",
                "Use the current checkout; do not create another workspace.",
                "Leave the working tree clean after the final snapshot.",
                "",
            ]
        )
    elif subject.kind == "git":
        sem = semantics()
        status_command = sem.status_display("git", track)
        diff_command = sem.diff_display("git", track)
        text = "\n".join(
            [
                "# AGENTS.md",
                "",
                "This benchmark workspace uses Git.",
                "Use the current checkout; do not create another clone or worktree.",
                f"Before snapshotting, inspect `{status_command}` and `{diff_command}`.",
                "Snapshot final work with `git add .` and `git commit -m \"agent task\"`.",
                "Leave the working tree clean.",
                "",
            ]
        )
    else:
        text = "\n".join(
            [
                "# AGENTS.md",
                "",
                "This benchmark workspace uses Oak, not Git.",
                "Do not run `git` commands.",
                "Use the current Oak checkout; do not create another mount or space.",
                "Before snapshotting, inspect `oak status` and `oak diff`.",
                "Snapshot final work with `oak commit --no-verify`.",
                "Leave the working tree clean.",
                "",
            ]
        )
    (repo / "AGENTS.md").write_text(text)


def run_command(
    command: list[str],
    cwd: Path,
    timeout: float = 120.0,
    env: dict[str, str] | None = None,
) -> subprocess.CompletedProcess[str]:
    return subprocess.run(
        command,
        cwd=cwd,
        env=sanitized_agent_env(env if env is not None else workflow_ab.base_env()),
        text=True,
        stdout=subprocess.PIPE,
        stderr=subprocess.PIPE,
        timeout=timeout,
        check=False,
    )


def initialize_repo(subject: workflow_ab.Subject, repo: Path) -> tuple[float, str, int]:
    start = time.perf_counter()
    output: list[str] = []
    commands: list[list[str]]
    if subject.kind == "git":
        vcs = str(subject.bin)
        commands = [
            [vcs, "init"],
            [vcs, "add", "."],
            [vcs, "commit", "-m", "initial"],
        ]
    else:
        vcs = str(subject.bin)
        commands = [
            [vcs, "init", "."],
            [vcs, "commit", "--no-verify"],
        ]

    returncode = 0
    for command in commands:
        proc = run_command(command, repo)
        output.append("$ " + workflow_ab.command_display(command))
        if proc.stdout:
            output.append(proc.stdout)
        if proc.stderr:
            output.append(proc.stderr)
        if proc.returncode != 0:
            returncode = proc.returncode
            break
    return (time.perf_counter() - start) * 1000.0, "\n".join(output), returncode


def commit_count(subject: workflow_ab.Subject, repo: Path) -> int | None:
    if subject.kind == "git":
        text = command_output([str(subject.bin), "rev-list", "--count", "HEAD"], repo)
        if text and text.isdigit():
            return int(text)
        return None
    text = command_output([str(subject.bin), "log", "--json"], repo)
    if text is None:
        return None
    try:
        commits = json.loads(text)
    except (ValueError, TypeError):
        return None
    if not isinstance(commits, list) or any(
        not isinstance(commit, dict)
        or not isinstance(commit.get("hash"), str)
        or not re.fullmatch(r"[0-9a-f]{64}", commit["hash"])
        for commit in commits
    ):
        return None
    return len(commits)


def vcs_status(subject: workflow_ab.Subject, repo: Path) -> tuple[str, int | None, float]:
    start = time.perf_counter()
    if subject.kind == "git":
        proc = run_command([str(subject.bin), "status", "--porcelain=v1"], repo)
        text = proc.stdout + proc.stderr
        dirty = len([line for line in text.splitlines() if line.strip()]) if proc.returncode == 0 else None
    else:
        proc = run_command([str(subject.bin), "status", "--json"], repo)
        text = proc.stdout + proc.stderr
        dirty = None
        if proc.returncode == 0:
            try:
                state = json.loads(proc.stdout)
            except (ValueError, TypeError):
                state = None
            changes = state.get("changes") if isinstance(state, dict) else None
            if isinstance(changes, list) and all(
                isinstance(change, dict)
                and isinstance(change.get("path"), str)
                and isinstance(change.get("status"), str)
                for change in changes
            ):
                dirty = len(changes)
    return text, dirty, (time.perf_counter() - start) * 1000.0


def test_command_for(workflow: workflow_ab.Workflow, subject: workflow_ab.Subject | None = None) -> list[str]:
    if workflow.name in workflow_oracles.SCENARIOS:
        return [sys.executable, "-I", str(Path(workflow_oracles.__file__).resolve()), workflow.name, ".",
                "--evidence-json", *(["--git-head", "--binary", str(subject.bin)] if subject is not None and subject.kind == "git"
                else ["--oak-head", "--binary", str(subject.bin)] if subject is not None and subject.kind == "oak" else []),
                *(["--git-instrument", os.environ["OAK_BENCH_GIT_INSTRUMENT"]] if os.environ.get("OAK_BENCH_GIT_INSTRUMENT") else []),
                *(["--oak-reader-python", os.environ["OAK_BENCH_OAK_READER_PYTHON"]] if os.environ.get("OAK_BENCH_OAK_READER_PYTHON") else [])]
    return ["python3", "-c", "raise SystemExit(0)"]


def build_prompt(
    subject: workflow_ab.Subject,
    workflow: workflow_ab.Workflow,
    track: str,
    instruction_level: str,
) -> str:
    if instruction_level == "zero-shot":
        if subject.kind == "git":
            vcs_lines = [
                "This repository uses Git.",
                "Do not create another workspace.",
                "Snapshot your final work and leave the working tree clean.",
            ]
        else:
            vcs_lines = [
                "This repository uses Oak (a version control system), not Git.",
                "Do not run `git` commands and do not create another workspace.",
                "Snapshot your final work with Oak and leave the working tree clean.",
            ]
    elif instruction_level == "cheat-sheet":
        if subject.kind == "git":
            vcs_lines = [
                "This repository uses Git. Do not create another workspace.",
                "Cheat sheet: `git status --short`, `git diff --stat`, "
                "`git add . && git commit -m \"agent task\"`.",
            ]
        else:
            vcs_lines = [
                "This repository uses Oak, not Git. Do not run `git` commands.",
                "Cheat sheet: `oak status`, `oak diff`, `oak commit --no-verify`.",
            ]
    elif subject.kind == "git":
        sem = semantics()
        status_command = sem.status_display("git", track)
        diff_command = sem.diff_display("git", track)
        vcs_lines = [
            "This repository uses Git.",
            "Do not create another clone or worktree.",
            f"Before the final commit, inspect `{status_command}` and `{diff_command}`.",
            "Commit with `git add .` and `git commit -m \"agent task\"`.",
        ]
    else:
        vcs_lines = [
            "This repository uses Oak, not Git.",
            "Do not run `git` commands.",
            "Do not create another mount or space.",
            "Before the final snapshot, inspect `oak status` and `oak diff`.",
            "Snapshot with `oak commit --no-verify`.",
        ]

    if workflow.name == "bugfix_test_loop":
        task = (
            "Fix the failing pricing unit test by making the smallest reasonable code change. "
            "Run the test before changing code if possible, make the fix, then rerun the test."
        )
        success = [
            "`python3 -m unittest discover -s tests` passes.",
            "Only the necessary source file changes.",
        ]
    elif workflow.name == "wide_config_refactor":
        task = (
            "Update all service configs using the deprecated timeout value so validation passes. "
            "Avoid unrelated file changes."
        )
        success = ["`python3 validate_configs.py` passes.", "Only matching service config files change."]
    elif workflow.name == "large_asset_manifest":
        task = (
            "Inspect the manifest and mark every listed asset reviewed without rewriting binary assets. "
            "Avoid dumping large binary files into context."
        )
        success = ["`python3 validate_manifest.py` passes.", "Only manifest.json changes."]
    else:
        task = workflow.description
        success = ["The fixture validator passes.", "Unrelated files are not changed."]

    lines = [
        "You are running inside an isolated benchmark workspace.",
        "Use local files and local commands only. Do not use network access.",
        "Keep the final answer concise; the benchmark oracle checks the repository state.",
        "",
        "Task:",
        task,
        "",
        "Version-control rules:",
        *vcs_lines,
        "",
        "Success criteria:",
        *["- " + item for item in success],
        "- The final VCS state is clean after the commit/snapshot.",
        "",
        "Do the work now.",
    ]
    return "\n".join(lines) + "\n"


def build_agent_command(
    agent: AgentConfig,
    repo: Path,
    prompt: str,
    agent_environment: str,
) -> tuple[list[str], str | None]:
    if agent.adapter == "deterministic_mock":
        return [], None
    if agent.bin is None:
        raise RuntimeError(f"Agent {agent.name} has no binary")

    model_args: list[str] = []
    if agent.model:
        model_args = ["--model", agent.model]
    minimal = agent_environment == "minimal"

    if agent.adapter == "codex_exec_json":
        command = [
            str(agent.bin),
            "exec",
            "--json",
            "--color",
            "never",
            "-C",
            str(repo),
            "--skip-git-repo-check",
            "--ephemeral",
            "--sandbox",
            "danger-full-access",
            "--dangerously-bypass-approvals-and-sandbox",
            *(["--ignore-user-config", "--ignore-rules"] if minimal else []),
            *model_args,
            "-",
        ]
        return command, prompt

    if agent.adapter == "claude_print_stream_json":
        command = [
            str(agent.bin),
            "-p",
            "--output-format",
            "stream-json",
            # Required by the CLI for stream-json with --print; verified
            # against claude 2.1.170 (without it the run exits immediately).
            "--verbose",
            "--permission-mode",
            "bypassPermissions",
            "--dangerously-skip-permissions",
            "--no-session-persistence",
            *(["--safe-mode", "--disable-slash-commands"] if minimal else []),
            *model_args,
            prompt,
        ]
        return command, None

    if agent.adapter == "cursor_agent_stream_json":
        command = [
            str(agent.bin),
            "--print",
            "--output-format",
            "stream-json",
            "--force",
            "--trust",
            "--sandbox",
            "disabled",
            "--workspace",
            str(repo),
            *model_args,
            prompt,
        ]
        return command, None

    raise RuntimeError(f"Unsupported adapter: {agent.adapter}")


def run_mock_agent(
    subject: workflow_ab.Subject,
    workflow: workflow_ab.Workflow,
    repo: Path,
    track: str,
    artifacts: Path,
    prompt_path: Path,
    env: dict[str, str] | None = None,
) -> AgentExecution:
    stdout_path = artifacts / "agent.stdout.log"
    stderr_path = artifacts / "agent.stderr.log"
    start = time.perf_counter()
    stdout_parts: list[str] = []
    stderr_parts: list[str] = []
    returncode = 0
    for step in workflow.steps(subject, track):
        proc = run_command(step.command, repo, env=workflow_ab.step_env(step, env))
        stdout_parts.append("$ " + workflow_ab.command_display(step.command))
        if proc.stdout:
            stdout_parts.append(proc.stdout)
        if proc.stderr:
            stderr_parts.append(proc.stderr)
        if proc.returncode not in step.expected_returncodes:
            returncode = proc.returncode or 1
            break
    elapsed_ms = (time.perf_counter() - start) * 1000.0
    stdout_path.write_text("\n".join(stdout_parts))
    stderr_path.write_text("\n".join(stderr_parts))
    return AgentExecution(returncode, False, elapsed_ms, stdout_path, stderr_path, prompt_path)


def run_real_agent(
    agent: AgentConfig,
    repo: Path,
    prompt: str,
    artifacts: Path,
    timeout_seconds: float,
    agent_environment: str,
    env: dict[str, str] | None = None,
    output_limit_bytes: int = DEFAULT_OUTPUT_LIMIT_BYTES,
) -> AgentExecution:
    artifacts.mkdir(parents=True, exist_ok=True)
    prompt_path = artifacts / "prompt.txt"
    stdout_path = artifacts / "agent.stdout.log"
    stderr_path = artifacts / "agent.stderr.log"
    prompt_path.write_text(prompt)

    if agent.adapter == "deterministic_mock":
        raise RuntimeError("mock agent is handled separately")

    command, stdin_text = build_agent_command(agent, repo, prompt, agent_environment)
    if env is None:
        env, environment_metadata = trial_agent_env(artifacts)
    else:
        env = sanitized_agent_env(env)
        environment_metadata = {"fresh_agent_home": env.get("HOME"),
                                "environment_policy": "caller-supplied-sanitized-git",
                                "os_sandbox_enforced": False}
    result = run_captured(command, repo, env, stdin_text, stdout_path, stderr_path,
                          timeout_seconds, output_limit_bytes)
    capture_metadata = {
        "execution_version": EXECUTION_VERSION, "timeout_seconds": timeout_seconds,
        "output_limit_bytes": output_limit_bytes, "command": command,
        "stdout_observed_bytes": result.stdout_observed_bytes,
        "stderr_observed_bytes": result.stderr_observed_bytes,
        "stdout_retained_bytes": result.stdout_retained_bytes,
        "stderr_retained_bytes": result.stderr_retained_bytes,
        "capture_complete": result.capture_complete,
        "output_limited": result.output_limited,
        "byte_measurement": "observed_pipe_bytes; lower_bound_if_capture_incomplete",
        "cleanup_scope": "owned_posix_process_group" if os.name == "posix" else "direct_child_only",
        **environment_metadata,
    }
    (artifacts / "execution.json").write_text(json.dumps(capture_metadata, sort_keys=True) + "\n")
    return AgentExecution(result.returncode, result.timed_out, result.elapsed_ms,
                          stdout_path, stderr_path, prompt_path, result.output_limited,
                          result.capture_complete, capture_metadata)


def read_text(path: Path, byte_limit: int = DEFAULT_OUTPUT_LIMIT_BYTES) -> str:
    try:
        with path.open("rb") as handle:
            return handle.read(byte_limit).decode("utf-8", "replace")
    except FileNotFoundError:
        return ""


def parse_json_lines(text: str) -> list[Any]:
    events: list[Any] = []
    for line in text.splitlines():
        stripped = line.strip()
        if not stripped or not stripped.startswith(("{", "[")):
            continue
        try:
            events.append(json.loads(stripped))
        except json.JSONDecodeError:
            continue
    return events


def collect_usage(events: list[Any]) -> dict[str, int | None]:
    """Provider-reported usage, snake_case first.

    camelCase keys (cursor-agent 2026.06 reports {inputTokens, outputTokens,
    cacheReadTokens, cacheWriteTokens}) are a FALLBACK pass, never merged with
    snake_case: claude's result event carries both `usage` (snake_case) and a
    per-model `modelUsage` (camelCase) breakdown of the same tokens, and
    summing both would double-count.
    """
    snake = _collect_usage_with_keys(
        events,
        input_keys=("input_tokens", "prompt_tokens", "inputTokenCount", "promptTokens"),
        output_keys=("output_tokens", "completion_tokens", "outputTokenCount", "completionTokens"),
        reasoning_keys=("reasoning_tokens", "reasoning_output_tokens", "reasoningTokenCount"),
        cache_read_keys=("cache_read_input_tokens", "cache_read_tokens", "cached_tokens", "cached_input_tokens"),
        cache_write_keys=("cache_creation_input_tokens", "cache_write_tokens"),
        total_keys=("total_tokens", "totalTokens", "tokens"),
    )
    if any(value is not None for value in snake.values()):
        return snake
    return _collect_usage_with_keys(
        events,
        input_keys=("inputTokens",),
        output_keys=("outputTokens",),
        reasoning_keys=("reasoningTokens",),
        cache_read_keys=("cacheReadTokens", "cacheReadInputTokens"),
        cache_write_keys=("cacheWriteTokens", "cacheCreationInputTokens"),
        total_keys=(),
    )


def _collect_usage_with_keys(
    events: list[Any],
    *,
    input_keys: tuple[str, ...],
    output_keys: tuple[str, ...],
    reasoning_keys: tuple[str, ...],
    cache_read_keys: tuple[str, ...],
    cache_write_keys: tuple[str, ...],
    total_keys: tuple[str, ...],
) -> dict[str, int | None]:

    # Per-field None until that field is actually observed: a stream that
    # exposes only total_tokens must not report input/output/reasoning/cache
    # as 0 — that is a fabricated zero (ADR-0002) and silently corrupts
    # cost_weighted_total.
    totals: dict[str, int | None] = {
        "input_tokens_reported": None,
        "output_tokens_reported": None,
        "reasoning_tokens_reported": None,
        "cache_read_tokens_reported": None,
        "cache_write_tokens_reported": None,
        "total_tokens_reported": None,
    }

    result_totals = _collect_result_usage_with_keys(
        events,
        input_keys=input_keys,
        output_keys=output_keys,
        reasoning_keys=reasoning_keys,
        cache_read_keys=cache_read_keys,
        cache_write_keys=cache_write_keys,
        total_keys=total_keys,
    )
    if result_totals is not None:
        return result_totals

    for event in events:
        for item in walk_json(event):
            if not isinstance(item, dict):
                continue
            direct_keys = set(item)
            known = set(input_keys + output_keys + reasoning_keys + cache_read_keys + cache_write_keys + total_keys)
            if not direct_keys.intersection(known):
                continue
            values = {
                "input_tokens_reported": first_number(item, input_keys),
                "output_tokens_reported": first_number(item, output_keys),
                "reasoning_tokens_reported": first_number(item, reasoning_keys),
                "cache_read_tokens_reported": first_number(item, cache_read_keys),
                "cache_write_tokens_reported": first_number(item, cache_write_keys),
                "total_tokens_reported": first_number(item, total_keys),
            }
            for key, value in values.items():
                if value is not None:
                    totals[key] = (totals[key] or 0) + value

    if totals["total_tokens_reported"] is None:
        components = [
            totals["input_tokens_reported"],
            totals["output_tokens_reported"],
            totals["reasoning_tokens_reported"],
        ]
        if any(component is not None for component in components):
                totals["total_tokens_reported"] = sum(component or 0 for component in components)
    return totals


def _collect_result_usage_with_keys(
    events: list[Any],
    *,
    input_keys: tuple[str, ...],
    output_keys: tuple[str, ...],
    reasoning_keys: tuple[str, ...],
    cache_read_keys: tuple[str, ...],
    cache_write_keys: tuple[str, ...],
    total_keys: tuple[str, ...],
) -> dict[str, int | None] | None:
    """Return cumulative provider totals from top-level result.usage.

    Claude repeats the same message.usage across assistant content-block
    events, then emits the cumulative run total in result.usage. That result
    usage may also contain per-iteration usage beneath it, so this intentionally
    reads only direct keys from the top-level usage object.
    """

    for event in reversed(events):
        if not isinstance(event, dict) or str(event.get("type") or "").lower() != "result":
            continue
        usage = event.get("usage")
        if not isinstance(usage, dict):
            continue
        values = {
            "input_tokens_reported": first_number(usage, input_keys),
            "output_tokens_reported": first_number(usage, output_keys),
            "reasoning_tokens_reported": first_number(usage, reasoning_keys),
            "cache_read_tokens_reported": first_number(usage, cache_read_keys),
            "cache_write_tokens_reported": first_number(usage, cache_write_keys),
            "total_tokens_reported": first_number(usage, total_keys),
        }
        if not any(value is not None for value in values.values()):
            continue
        if values["total_tokens_reported"] is None:
            components = [
                values["input_tokens_reported"],
                values["output_tokens_reported"],
                values["reasoning_tokens_reported"],
            ]
            if any(component is not None for component in components):
                values["total_tokens_reported"] = sum(component or 0 for component in components)
        return values
    return None


def extract_text_chars(events: list[Any], fallback: str) -> int:
    texts: list[str] = []
    text_keys = {"text", "message", "delta", "summary", "result", "final_response"}
    for event in events:
        for item in walk_json(event):
            if not isinstance(item, dict):
                continue
            for key, value in item.items():
                if key in text_keys and isinstance(value, str):
                    texts.append(value)
                elif key == "content" and isinstance(value, str):
                    texts.append(value)
    if texts:
        return sum(len(text) for text in texts)
    return len(fallback)


def cost_weighted_total_from_usage(
    usage: dict[str, int | None],
    *,
    cache_read_included_in_input: bool,
) -> float | None:
    has_direction_components = any(
        usage.get(key) is not None
        for key in ("input_tokens_reported", "output_tokens_reported", "reasoning_tokens_reported")
    )
    if not has_direction_components:
        return None

    emitted = (usage.get("output_tokens_reported") or 0) + (usage.get("reasoning_tokens_reported") or 0)
    ingested = usage.get("input_tokens_reported") or 0
    cache_read = usage.get("cache_read_tokens_reported") or 0
    cache_write = usage.get("cache_write_tokens_reported") or 0
    if cache_read_included_in_input:
        ingested = max(0, ingested - cache_read)
    return round(
        emitted * EMITTED_TOKEN_WEIGHT
        + ingested * INGESTED_TOKEN_WEIGHT
        + cache_read * CACHE_READ_TOKEN_WEIGHT
        + cache_write * INGESTED_TOKEN_WEIGHT,
        1,
    )


def cursor_tool_call_parts(event: Any) -> tuple[str, str | None] | None:
    if not isinstance(event, dict) or str(event.get("type") or "").lower() != "tool_call":
        return None
    tool_call = event.get("tool_call")
    if not isinstance(tool_call, dict):
        return None
    tool_name = next((key for key in tool_call if key.endswith("ToolCall")), None)
    if tool_name is None:
        return None
    payload = tool_call.get(tool_name)
    if not isinstance(payload, dict):
        return tool_name, None
    args = payload.get("args")
    if isinstance(args, dict) and isinstance(args.get("command"), str):
        return tool_name, unwrap_shell_command(args["command"])
    result = payload.get("result")
    if isinstance(result, dict):
        shell_result = result.get("success") or result.get("failure")
        if isinstance(shell_result, dict) and isinstance(shell_result.get("command"), str):
            return tool_name, unwrap_shell_command(shell_result["command"])
    return tool_name, None


def collect_tool_metrics(stdout_text: str, stderr_text: str, events: list[Any]) -> tuple[dict[str, int], dict[str, int]]:
    """One tool call counts exactly once.

    A Claude-style tool_use block carries name + input.command in one item;
    the old walker counted the name, the input command, and the inner input
    dict's command again — 3 tool_calls_total per actual call. Commands that
    belong to a counted tool_use go to attached_commands: classified for
    kind/VCS/help counts but never re-counted as separate calls.
    """
    tool_names: list[str] = []
    commands: list[str] = []
    attached_commands: list[str] = []
    consumed_input_ids: set[int] = set()

    for event in events:
        cursor_call = cursor_tool_call_parts(event)
        if cursor_call is not None:
            if str(event.get("subtype") or "").lower() == "started":
                tool_name, command = cursor_call
                tool_names.append(tool_name)
                if command is not None:
                    attached_commands.append(command)
            continue

        for item in walk_json(event):
            if not isinstance(item, dict):
                continue
            item_type = str(item.get("type") or item.get("event") or "").lower()
            name = item.get("name") or item.get("tool_name") or item.get("tool")
            if item_type == "file_change" and str(item.get("status") or "").lower() == "completed":
                tool_names.append("file_change")
            is_tool_call_item = isinstance(name, str) and (
                "tool" in item_type or item_type in {"function_call", "function"}
            )
            if is_tool_call_item:
                tool_names.append(name)
            input_value = item.get("input")
            if isinstance(input_value, dict):
                inner_command = input_value.get("command") or input_value.get("cmd")
                if isinstance(inner_command, str):
                    # The walker will visit this input dict again; mark it so
                    # its command is not re-counted as a standalone call.
                    consumed_input_ids.add(id(input_value))
                    if is_tool_call_item:
                        attached_commands.append(unwrap_shell_command(inner_command))
                    else:
                        commands.append(unwrap_shell_command(inner_command))
            if id(item) in consumed_input_ids:
                continue
            command = item.get("command") or item.get("cmd")
            if isinstance(command, str):
                status = str(item.get("status") or "").lower()
                exit_code = item.get("exit_code")
                if item.get("type") != "command_execution" or status not in {"in_progress", "queued"} or exit_code is not None:
                    if is_tool_call_item:
                        attached_commands.append(unwrap_shell_command(command))
                    else:
                        commands.append(unwrap_shell_command(command))

    for line in (stdout_text + "\n" + stderr_text).splitlines():
        if line.startswith("$ "):
            commands.append(unwrap_shell_command(line[2:]))

    counts = {
        "tool_calls_total": 0,
        "terminal_commands_total": 0,
        "file_read_calls": 0,
        "file_write_calls": 0,
        "search_calls": 0,
        "edit_calls": 0,
        "vcs_commands_total": 0,
        "test_commands_total": 0,
        "network_calls_total": 0,
        "privileged_calls_total": 0,
        "help_calls_total": 0,
    }
    by_kind: dict[str, int] = {}

    for name in tool_names:
        lowered = name.lower()
        by_kind[lowered] = by_kind.get(lowered, 0) + 1
        counts["tool_calls_total"] += 1
        if lowered in {"bash", "shell", "terminal", "exec_command", "run_terminal_cmd"} or "shell" in lowered:
            counts["terminal_commands_total"] += 1
        if "read" in lowered:
            counts["file_read_calls"] += 1
        if "edit" in lowered or "write" in lowered or lowered == "file_change":
            counts["edit_calls"] += 1
            counts["file_write_calls"] += 1
        if "search" in lowered or "grep" in lowered or "glob" in lowered:
            counts["search_calls"] += 1

    def classify_into_counts(command: str, counts_as_call: bool) -> None:
        kinds = classify_command(command)
        if counts_as_call:
            counts["tool_calls_total"] += 1
            for kind in kinds:
                by_kind[kind] = by_kind.get(kind, 0) + 1
            if "terminal" in kinds:
                counts["terminal_commands_total"] += 1
        if "read" in kinds:
            counts["file_read_calls"] += 1
        if "file_write" in kinds:
            counts["file_write_calls"] += 1
        if "search" in kinds:
            counts["search_calls"] += 1
        if "edit" in kinds:
            counts["edit_calls"] += 1
        if "vcs" in kinds:
            counts["vcs_commands_total"] += 1
        if "test" in kinds:
            counts["test_commands_total"] += 1
        if "help" in kinds:
            counts["help_calls_total"] += 1
        if re.search(r"\b(curl|wget|gh api|npm install|pip install)\b", command):
            counts["network_calls_total"] += 1
        if re.search(r"\b(sudo|su\s+-)\b", command):
            counts["privileged_calls_total"] += 1

    for command in commands:
        classify_into_counts(command, counts_as_call=True)
    for command in attached_commands:
        # The call was already counted via its tool_use name; only the
        # command's nature (VCS/test/help/...) still needs attribution.
        classify_into_counts(command, counts_as_call=False)

    return counts, by_kind


# PATH-shim VCS attribution fields, merged into vcs_metrics. Metric nulls carry
# explicit reasons (ADR-0002: unmeasured is null, never zero). Shim overhead is
# recorded, never subtracted.
SHIM_REASON_FIELDS = (
    "shim_sidecar_path_unavailable_reason",
    "shim_overhead_unavailable_reason",
    "vcs_ms_unavailable_reason",
    "vcs_call_count_shim_unavailable_reason",
    "agent_blocked_on_vcs_ms_unavailable_reason",
    "vcs_share_of_task_wall_unavailable_reason",
    "thrash_events_count_unavailable_reason",
    "polling_loop_count_unavailable_reason",
)

SHIM_NULL_VALUES: dict[str, Any] = {
    "shim_sidecar_path": None,
    "shim_overhead": None,
    "shim_overhead_ms": None,
    "vcs_ms": None,
    "vcs_call_count_shim": None,
    "agent_blocked_on_vcs_ms": None,
    "vcs_share_of_task_wall": None,
    "thrash_events_count": None,
    "polling_loop_count": None,
}


def shim_metrics_unavailable(reason: str) -> dict[str, Any]:
    fields = dict(SHIM_NULL_VALUES)
    for key in SHIM_REASON_FIELDS:
        fields[key] = reason
    return fields


def shim_real_binaries(subject: workflow_ab.Subject) -> dict[str, str]:
    """Real binaries to wrap: the subject's VCS, plus git if discoverable."""
    real: dict[str, str] = {}
    from oakbench.subjects import which_measured_git
    git_found = which_measured_git()
    if git_found:
        real["git"] = git_found
    if subject.kind in ("git", "oak") and subject.bin is not None and Path(subject.bin).exists():
        real[subject.kind] = str(subject.bin)
    return real


def shim_metrics_from_sidecar(
    sidecar: Path,
    shim_overhead_ms: float | None,
    task_wall_ms: float,
) -> dict[str, Any]:
    """Shim-attributed VCS fields from the sidecar JSONL, with null reasons."""
    fields = shim_metrics_unavailable("sidecar_empty")
    fields["shim_sidecar_path"] = str(sidecar)
    fields["shim_sidecar_path_unavailable_reason"] = None
    fields["shim_overhead"] = shim_overhead_ms
    fields["shim_overhead_ms"] = shim_overhead_ms
    fields["shim_overhead_unavailable_reason"] = (
        None if shim_overhead_ms is not None else "shim_overhead_unmeasured"
    )
    rows = read_sidecar_rows(sidecar)
    if not rows:
        return fields

    fields["vcs_call_count_shim"] = len(rows)
    fields["vcs_call_count_shim_unavailable_reason"] = None

    elapsed = [row.get("elapsed_ms") for row in rows]
    measured = [value for value in elapsed if isinstance(value, (int, float)) and not isinstance(value, bool)]
    if measured:
        fields["vcs_ms"] = round(sum(measured), 3)
        fields["vcs_ms_unavailable_reason"] = None
    else:
        fields["vcs_ms_unavailable_reason"] = "sidecar_rows_missing_elapsed_ms"

    blocked = agent_blocked_on_vcs_ms(rows)
    fields["agent_blocked_on_vcs_ms"] = blocked
    fields["agent_blocked_on_vcs_ms_unavailable_reason"] = (
        None if blocked is not None else "sidecar_rows_missing_vcs_timing_or_command"
    )
    fields["vcs_share_of_task_wall"] = vcs_share_of_task_wall(blocked, task_wall_ms)
    if fields["vcs_share_of_task_wall"] is not None:
        fields["vcs_share_of_task_wall_unavailable_reason"] = None
    elif blocked is None:
        fields["vcs_share_of_task_wall_unavailable_reason"] = "agent_blocked_on_vcs_ms_unavailable"
    else:
        fields["vcs_share_of_task_wall_unavailable_reason"] = "task_wall_ms_unavailable"

    events = thrash_events(rows)
    if events is None:
        fields["thrash_events_count_unavailable_reason"] = "sidecar_rows_missing_vcs_timestamp_or_command"
        fields["polling_loop_count_unavailable_reason"] = "sidecar_rows_missing_vcs_timestamp_or_command"
    else:
        fields["thrash_events_count"] = len(events)
        fields["polling_loop_count"] = sum(1 for event in events if event.get("type") == "vcs_polling_loop")
        fields["thrash_events_count_unavailable_reason"] = None
        fields["polling_loop_count_unavailable_reason"] = None
    return fields


def vcs_metrics_from_tools(
    subject: workflow_ab.Subject,
    stdout_text: str,
    stderr_text: str,
    dirty_files: int | None,
    commit_delta: int,
) -> dict[str, Any]:
    text = stdout_text + "\n" + stderr_text
    commands = [unwrap_shell_command(line[2:]) for line in text.splitlines() if line.startswith("$ ")]
    events = parse_json_lines(text)
    for event in events:
        cursor_call = cursor_tool_call_parts(event)
        if cursor_call is not None:
            if str(event.get("subtype") or "").lower() == "started":
                _, command = cursor_call
                if command is not None:
                    commands.append(command)
            continue

        for item in walk_json(event):
            if not isinstance(item, dict):
                continue
            if str(item.get("type") or "").lower() == "tool_use":
                name = str(item.get("name") or item.get("tool_name") or item.get("tool") or "").lower()
                input_value = item.get("input")
                command = input_value.get("command") if isinstance(input_value, dict) else None
                if isinstance(command, str) and name in {"bash", "shell", "terminal"}:
                    commands.append(unwrap_shell_command(command))
                continue
            if item.get("type") != "command_execution":
                continue
            status = str(item.get("status") or "").lower()
            if status in {"in_progress", "queued"} and item.get("exit_code") is None:
                continue
            command = item.get("command")
            if isinstance(command, str):
                commands.append(unwrap_shell_command(command))
    commands_available = bool(commands)
    # One classification authority: every count below derives from the SAME
    # command list through oakbench.classify.vcs_invocations, so sub-counts can
    # never disagree with the total (the reviewer-found vcs_commands_total=0
    # alongside status_commands=1). Sub-counts are a family breakdown, not a
    # partition — add/log/etc. count in the total only.
    invocations = [
        (segment, subcommand)
        for command in commands
        for segment, subcommand in vcs_invocations(command)
    ]
    subcommands = [subcommand for _, subcommand in invocations]
    commands_total = sum(1 for subcommand in subcommands if subcommand is not None) if commands_available else None
    status_commands = sum(1 for subcommand in subcommands if subcommand == "status") if commands_available else None
    diff_commands = sum(1 for subcommand in subcommands if subcommand == "diff") if commands_available else None
    snapshot_commands = sum(1 for subcommand in subcommands if subcommand == "commit") if commands_available else None
    branch_commands = sum(
        1
        for command, subcommand in invocations
        if subcommand in {"checkout", "switch"} and re.search(r"\s-[bc]\b", command)
    ) if commands_available else None
    merge_commands = sum(1 for subcommand in subcommands if subcommand == "merge") if commands_available else None
    cleanup_commands = sum(
        1
        for command, subcommand in invocations
        if subcommand is not None and re.search(r"\b(clean|end|remove)\b", command)
    ) if commands_available else None
    return {
        "commands_total": commands_total,
        "status_commands": status_commands,
        "diff_commands": diff_commands,
        "snapshot_commands": snapshot_commands,
        "branch_commands": branch_commands,
        "merge_commands": merge_commands,
        "cleanup_commands": cleanup_commands,
        "commits_created": max(0, commit_delta),
        "branches_created": branch_commands or 0,
        "conflicts_total": None,
        "conflicts_resolved": None,
        "dirty_files_at_end": dirty_files,
        "cleanup_success": dirty_files == 0,
        "measurement_source": (
            "transcript_command_classification_plus_oracle_state"
            if commands_available
            else "oracle_state_only; transcript_vcs_commands_unavailable"
        ),
    }


def run_oracle(
    subject: workflow_ab.Subject,
    workflow: workflow_ab.Workflow,
    repo: Path,
    before_manifest: dict[str, dict[str, Any]],
    initial_commit_count: int | None,
    artifacts: Path,
) -> OracleResult:
    start = time.perf_counter()
    artifact_paths: list[Path] = []

    test_start = time.perf_counter()
    test_proc = run_command(
        test_command_for(workflow, subject),
        repo,
        timeout=300.0,
        env=workflow_ab.python_no_bytecode_env(),
    )
    test_ms = (time.perf_counter() - test_start) * 1000.0
    test_log = artifacts / "oracle.test.log"
    test_log.write_text(test_proc.stdout + test_proc.stderr)
    artifact_paths.append(test_log)
    evidence = None
    if workflow.name in workflow_oracles.SCENARIOS:
        try:
            evidence = json.loads(test_proc.stdout)
            passed = evidence.get("workflow_integrity_passed") if isinstance(evidence, dict) else "invalid"
            valid = (isinstance(evidence, dict) and evidence.get("oracle_version") == ORACLE_VERSION
                     and isinstance(evidence.get("workflow_integrity_source"), str)
                     and (passed is True or passed is False or passed is None)
                     and ((passed is None and bool(evidence.get("workflow_integrity_unmeasured_reason")))
                          or (passed is False and bool(evidence.get("workflow_integrity_failure_reason")))
                          or passed is True)
                     and test_proc.returncode == (77 if passed is None else 0 if passed else 1))
            if not valid:
                evidence = None
        except ValueError:
            evidence = None
        if evidence is None:
            evidence = {"oracle_version": ORACLE_VERSION,
                        "workflow_integrity_source": "trusted_oracle_instrument_unavailable",
                        "workflow_integrity_passed": None,
                        "workflow_integrity_failure_reason": None,
                        "workflow_integrity_unmeasured_reason": "trusted_oracle_evidence_invalid_or_missing"}

    if test_proc.returncode == 77 and evidence is None:
        evidence = {"oracle_version": ORACLE_VERSION,
                    "workflow_integrity_source": "subject_status;required_check_unavailable",
                    "workflow_integrity_evidence_class": "subject_status_only",
                    "workflow_integrity_passed": None,
                    "workflow_integrity_failure_reason": None,
                    "workflow_integrity_unmeasured_reason": "required_check_instrument_unavailable"}

    behavior_passed = None
    if workflow.name == "bugfix_test_loop":
        behavior_start = time.perf_counter()
        # This command runs agent-edited code, so its exit status is
        # subject-controlled: any nonzero (including 77) is a measured failure.
        # "Unmeasured" is only ever decided by the harness itself, here when
        # the interpreter could not be launched at all.
        try:
            behavior = run_command(["python3", "-m", "unittest", "discover", "-s", "tests"], repo,
                                   timeout=300.0, env=workflow_ab.python_no_bytecode_env())
            behavior_passed = behavior.returncode == 0
            behavior_text = behavior.stdout + behavior.stderr
        except OSError as exc:
            behavior_passed = None
            behavior_text = f"harness could not launch behavior tests: {exc}\n"
            evidence = {**(evidence or {}), "workflow_integrity_passed": None,
                        "workflow_integrity_failure_reason": None,
                        "workflow_integrity_unmeasured_reason": "behavior_test_instrument_unavailable"}
        except subprocess.TimeoutExpired as exc:
            behavior_passed = False
            behavior_text = f"behavior tests exceeded {exc.timeout}s\n"
        test_ms += (time.perf_counter() - behavior_start) * 1000
        behavior_log = artifacts / "oracle.behavior.log"
        behavior_log.write_text(behavior_text)
        artifact_paths.append(behavior_log)

    status_text, dirty_count, status_ms = vcs_status(subject, repo)
    status_log = artifacts / "oracle.status.log"
    status_log.write_text(status_text)
    artifact_paths.append(status_log)

    after_manifest = workspace_manifest(repo)
    changed = changed_files(before_manifest, after_manifest)
    expected_change_shape = expected_change_ok(workflow, changed)
    bytes_written = sum(int(after_manifest.get(path, {}).get("bytes", 0)) for path in changed)

    final_commit_count = commit_count(subject, repo)
    commit_delta = 0
    if initial_commit_count is not None and final_commit_count is not None:
        commit_delta = final_commit_count - initial_commit_count

    checks = {
        "required_checks_passed": None if (evidence is not None and evidence.get("workflow_integrity_passed") is None) else test_proc.returncode == 0,
        "expected_change_shape": expected_change_shape,
        "final_vcs_clean": dirty_count == 0,
        "snapshot_created": workflow.name == "history_archaeology" or commit_delta > 0,
    }
    if workflow.name == "bugfix_test_loop":
        checks["behavior_tests_passed"] = behavior_passed
    success = all(checks.values())
    failure_reason = None
    for key, passed in checks.items():
        if passed is False:
            failure_reason = key
            break

    elapsed_ms = (time.perf_counter() - start) * 1000.0
    return OracleResult(
        elapsed_ms=elapsed_ms,
        test_ms=test_ms,
        vcs_ms=status_ms,
        success=success,
        outcome="pass" if success else "fail",
        failure_reason=failure_reason,
        checks=checks,
        test_returncode=test_proc.returncode,
        test_output_bytes=len((test_proc.stdout + test_proc.stderr).encode("utf-8", "replace")),
        status_output=status_text,
        dirty_files_at_end=dirty_count,
        changed_files=changed,
        bytes_written=bytes_written,
        commit_delta=commit_delta,
        artifact_paths=artifact_paths,
        evidence=evidence,
    )


def fixture_shape(workflow: workflow_ab.Workflow) -> str:
    if workflow.name == "bugfix_test_loop":
        return "small Python app with one failing unit test"
    if workflow.name == "wide_config_refactor":
        return "600 small service config files plus validator"
    if workflow.name == "large_asset_manifest":
        return "binary-heavy repo with 64 MiB assets plus JSON manifest"
    return workflow.description


def make_row(
    bench_id: str,
    timestamp_utc: str,
    profile: str,
    agent: AgentConfig,
    subject: workflow_ab.Subject,
    workflow: workflow_ab.Workflow,
    track: str,
    run_index: int,
    repo: Path,
    artifacts: Path,
    setup_ms: float,
    setup_output_bytes: int,
    execution: AgentExecution,
    oracle: OracleResult,
    prompt: str,
    before_manifest: dict[str, dict[str, Any]],
    source: dict[str, Any],
    host_meta: dict[str, str],
    agent_environment: str,
    instruction_level: str,
    shim_info: dict[str, Any] | None = None,
    agent_launched: bool = True,
    setup_returncode: int = 0,
) -> dict[str, Any]:
    capture_metadata = dict(execution.capture_metadata or {})
    parsing_limit = capture_metadata.get("output_limit_bytes", DEFAULT_OUTPUT_LIMIT_BYTES)
    if type(parsing_limit) is not int or parsing_limit < 1:
        raise ValueError("invalid configured output limit")
    stdout_text = read_text(execution.stdout_path, parsing_limit)
    stderr_text = read_text(execution.stderr_path, parsing_limit)
    stored_stdout_bytes = execution.stdout_path.stat().st_size if execution.stdout_path.exists() else 0
    stored_stderr_bytes = execution.stderr_path.stat().st_size if execution.stderr_path.exists() else 0
    stdout_bytes = capture_metadata.get("stdout_observed_bytes", stored_stdout_bytes)
    stderr_bytes = capture_metadata.get("stderr_observed_bytes", stored_stderr_bytes)
    output_truncated = (execution.output_limited or not execution.capture_complete
                        or stored_stdout_bytes + stored_stderr_bytes > parsing_limit)
    usage_complete = agent_launched and not (output_truncated or execution.timed_out or execution.returncode != 0)
    events = parse_json_lines(stdout_text) + parse_json_lines(stderr_text)
    usage = collect_usage(events)
    if not usage_complete:
        # A terminal or cumulative event in a partial transcript is not proof
        # of the complete run's usage, including any in-flight provider work.
        usage = {key: None for key in usage}
    output_chars = extract_text_chars(events, stdout_text + stderr_text)
    transcript_chars = len(prompt) + len(stdout_text) + len(stderr_text)
    token_source = "provider_reported" if usage["total_tokens_reported"] is not None else "char_only"
    if not usage_complete:
        token_source = "unknown"

    analysis = analyze_stream(events, agent.adapter)
    turn_timeline_path: str | None = None
    if analysis.turns:
        timeline_file = artifacts / "turn_timeline.jsonl"
        with timeline_file.open("w") as fh:
            for turn in analysis.turns:
                fh.write(json.dumps(turn, sort_keys=True) + "\n")
        turn_timeline_path = str(timeline_file)

    # Billing-direction cost. Provider-reported when available, otherwise a
    # char/4 estimate split by direction: assistant output is model-emitted
    # (expensive), prompt plus tool results are model-ingested.
    # Direction components must actually be reported: a stream exposing only
    # total_tokens has no direction split, and weighting Nones as zeros would
    # produce cost_weighted_total = 0 for a run that cost real money.
    has_direction_components = any(
        usage[key] is not None
        for key in ("input_tokens_reported", "output_tokens_reported", "reasoning_tokens_reported")
    )
    if not usage_complete:
        cost_weighted_total = None
    elif has_direction_components:
        cost_weighted_total = cost_weighted_total_from_usage(
            usage,
            cache_read_included_in_input=adapter_for(agent.adapter).cache_read_included_in_input,
        )
    else:
        emitted_est = output_chars / 4.0
        ingested_est = max(0, transcript_chars - output_chars) / 4.0
        cost_weighted_total = round(
            emitted_est * EMITTED_TOKEN_WEIGHT + ingested_est * INGESTED_TOKEN_WEIGHT, 1
        )

    tool_counts, by_kind = collect_tool_metrics(stdout_text, stderr_text, events)
    total_ms = setup_ms + execution.elapsed_ms + oracle.elapsed_ms
    logical_bytes = sum(int(item["bytes"]) for item in before_manifest.values())
    logical_file_count = len(before_manifest)
    output_bytes = stdout_bytes + stderr_bytes
    process_ok = (execution.returncode == 0 and not execution.timed_out
                  and not execution.output_limited and execution.capture_complete)
    success = process_ok and oracle.success
    evidence = getattr(oracle, "evidence", None) or {
        "oracle_version": ORACLE_VERSION, "workflow_integrity_source": "fixture_tests_and_subject_status",
        "workflow_integrity_passed": oracle.success,
        "workflow_integrity_failure_reason": oracle.failure_reason,
        "workflow_integrity_unmeasured_reason": None,
    }
    if workflow.name in workflow_oracles.SCENARIOS and getattr(oracle, "evidence", None) is None:
        evidence = {"oracle_version": ORACLE_VERSION,
                    "workflow_integrity_source": "trusted_oracle_instrument_unavailable",
                    "workflow_integrity_passed": None, "workflow_integrity_failure_reason": None,
                    "workflow_integrity_unmeasured_reason": "trusted_oracle_evidence_invalid_or_missing"}
    unmeasured = evidence.get("workflow_integrity_passed") is None
    measured_other_failure = any(value is False for key, value in oracle.checks.items()
                                 if key != "required_checks_passed")
    skipped_instrument = unmeasured and process_ok and not measured_other_failure
    harness_capture_failure = execution.output_limited or (output_truncated and not execution.timed_out)
    skipped_measurement = skipped_instrument or harness_capture_failure or not agent_launched
    outcome = "timeout" if execution.timed_out else (
        "error" if execution.output_limited or not execution.capture_complete else ("pass" if success else "fail"))
    failure_reason = None
    if execution.timed_out:
        failure_reason = "agent_timeout"
    elif execution.output_limited:
        failure_reason = "agent_output_limit"
    elif not execution.capture_complete:
        failure_reason = "agent_capture_incomplete"
    elif execution.returncode != 0:
        failure_reason = "agent_exit_nonzero"
    elif not oracle.success:
        failure_reason = oracle.failure_reason
    if skipped_measurement:
        outcome = "error"
        success = False
        failure_reason = ("agent_setup_failed" if not agent_launched else
                          "agent_output_limit" if execution.output_limited else
                          "agent_capture_incomplete" if harness_capture_failure else evidence.get("workflow_integrity_unmeasured_reason"))

    row = {
        "schema_version": SCHEMA_VERSION,
        "bench_id": bench_id,
        "timestamp_utc": timestamp_utc,
        "profile": profile,
        "scenario": workflow.name,
        "env_isolation_version": ENV_ISOLATION_VERSION,
        **runner_fields(),
        **cache_fields(None, None),
        "calibration_version": None,
        "calibration_unmeasured_reason": "no_token_calibration_artifact_bound_to_this_trial",
        "scenario_version": f"{PROMPT_VERSION}/oracle-{ORACLE_VERSION}/execution-{EXECUTION_VERSION}/outcomes-{ATTRIBUTION_VERSION}",
        "operation": "agent.workflow" if agent_launched else "agent.setup",
        "agent_launched": agent_launched,
        "outcome_attribution": "harness" if skipped_measurement else "agent",
        "setup_returncode": setup_returncode,
        "benchmark_track": track,
        "command_semantics_version": COMMAND_SEMANTICS_VERSION,
        "elapsed_ms": round(total_ms, 3),
        "returncode": 77 if skipped_measurement else 0 if success else 1,
        "skipped": skipped_measurement,
        "skip_reason": failure_reason if skipped_measurement else None,
        **evidence,
        "workflow_integrity_evidence_class": evidence.get("workflow_integrity_evidence_class", "tests_and_subject_status" if workflow.name not in workflow_oracles.SCENARIOS else None),
        "raw_output_bytes": output_bytes,
        "stdout_bytes": stdout_bytes,
        "stderr_bytes": stderr_bytes,
        "output_truncated": output_truncated,
        "task_id": None,
        "task_prompt_id": f"real-agent-{workflow.name}-{track}-{instruction_level}-{PROMPT_VERSION}",
        "run": run_index,
        "subject": subject.name,
        "subject_kind": subject.kind,
        "subject_label": subject.label,
        "vcs_mode": vcs_mode(subject),
        "agent": {
            "name": agent.name,
            "adapter": agent.adapter,
            "provider": agent.provider,
            "model": agent.model,
            "version": agent.version,
            "environment": agent_environment,
            "instruction_level": instruction_level,
            "temperature": None,
            "max_output_tokens": None,
        },
        "fixture": {
            "id": workflow.name,
            "repo_shape": fixture_shape(workflow),
            "fixture_version": workflow_ab.FIXTURE_VERSION,
            "logical_file_count": logical_file_count,
            "logical_bytes": logical_bytes,
        },
        "source": {**source, "agent_execution": {
            **capture_metadata, "execution_version": EXECUTION_VERSION,
            "usage_complete": usage_complete,
            "transcript_metrics_complete": usage_complete,
            "counters_coverage": "complete" if usage_complete else "observed_lower_bounds",
            "stdout_retained_bytes": stored_stdout_bytes,
            "stderr_retained_bytes": stored_stderr_bytes,
            "parsing_limit_bytes_per_stream": parsing_limit,
        }},
        "outcome": outcome,
        "success": success,
        "failure_reason": failure_reason,
        "success_criteria": {
            "passed": success,
            "checks": {
                "agent_exit_zero": process_ok if agent_launched else None,
                **oracle.checks,
            },
            "notes": None if success else f"changed_files={oracle.changed_files[:20]}",
        },
        "token_metrics": {
            "token_source": token_source,
            "tokenizer": None,
            **usage,
            "normalized_input_chars": len(prompt),
            "normalized_output_chars": output_chars if usage_complete else None,
            "transcript_chars": transcript_chars if usage_complete else None,
            "context_truncation_events": analysis.context_truncation_events if usage_complete else None,
            "token_budget": None,
            "cost_weighted_total": cost_weighted_total,
            "cost_weights": f"{COST_WEIGHTS_NOTE},cache_write={INGESTED_TOKEN_WEIGHT}",
        },
        "turn_metrics": {
            "turn_source": analysis.turn_source if usage_complete else "unmeasured",
            "assistant_turns_total": analysis.assistant_turns_total if usage_complete else None,
            "tool_calls_per_turn_avg": analysis.tool_calls_per_turn_avg if usage_complete else None,
            "max_tool_calls_per_turn": analysis.max_tool_calls_per_turn if usage_complete else None,
            "cumulative_input_tokens_reported": analysis.cumulative_input_tokens if usage_complete else None,
            "peak_input_tokens_reported": analysis.peak_input_tokens if usage_complete else None,
            "turn_timeline_path": turn_timeline_path,
        },
        "tool_call_metrics": {
            **tool_counts,
            "tool_calls_by_kind": by_kind,
            "unknown_command_failures_total": analysis.unknown_command_failures_total,
            "vcs_info_followup_calls_total": analysis.vcs_info_followup_calls_total,
            "parallel_batches_total": analysis.parallel_batches_total,
            "max_concurrency": analysis.max_concurrency,
            "measurement_source": ("adapter_stream_events_and_transcript_lines" if usage_complete
                                   else "partial_adapter_events_observed_lower_bounds"),
        },
        "failed_command_retry_metrics": {
            "failed_tool_calls_total": analysis.failed_tool_calls_total,
            "failed_commands_total": analysis.failed_tool_calls_total,
            "failed_vcs_commands_total": analysis.failed_vcs_commands_total,
            "failed_test_commands_total": analysis.failed_test_commands_total,
            "retry_attempts_total": analysis.same_command_retries_total,
            "retry_chains_total": None,
            "same_command_retries_total": analysis.same_command_retries_total,
            "retry_after_failure_ms": None,
            "recoverable_failures_total": None,
            "unrecoverable_failures_total": None,
            "turns_to_recovery": analysis.turns_to_recovery,
            "tokens_to_recovery": analysis.tokens_to_recovery,
            "failed_command_examples": analysis.failed_command_examples or [],
            "measurement_source": (
                "adapter_stream_exit_codes" if analysis.failed_tool_calls_total is not None else "not_adapter_visible"
            ),
        },
        "wall_clock_metrics": {
            "total_ms": round(total_ms, 3),
            "setup_ms": round(setup_ms, 3),
            "agent_active_ms": round(execution.elapsed_ms, 3) if agent_launched else None,
            "tool_wait_ms": None,
            "vcs_ms": round(oracle.vcs_ms, 3),
            "test_ms": round(oracle.test_ms, 3),
            "merge_ms": None,
            "cleanup_ms": None,
            "idle_ms": None,
            "timeout_ms": (capture_metadata["timeout_seconds"] * 1000
                           if capture_metadata.get("timeout_seconds") is not None else None),
            "measurement_source": "outer_cli_wall_clock_plus_oracle_phases",
        },
        "bytes_metrics": {
            "workspace_logical_bytes": logical_bytes,
            "workspace_materialized_bytes": logical_bytes,
            # Read-tool payload bytes are not adapter-visible in this lane;
            # the transcript output bytes some dashboards want are already in
            # command_output_bytes. A proxy here would corrupt hydration and
            # read-efficiency comparisons (ADR-0002: null, never a stand-in).
            "bytes_read_by_agent": None,
            "bytes_written_by_agent": oracle.bytes_written,
            "bytes_hydrated": None,
            "files_hydrated": None,
            "hydration_events": None,
            "large_file_bytes_scanned": None,
            "binary_bytes_touched": None,
            "command_output_bytes": output_bytes + oracle.test_output_bytes + setup_output_bytes,
            "measurement_source": "agent_cli_transcript_and_workspace_manifest; read-tool bytes and hydration not adapter-visible in this lane",
        },
        "vcs_metrics": {
            **vcs_metrics_from_tools(
                subject,
                stdout_text,
                stderr_text,
                oracle.dirty_files_at_end,
                oracle.commit_delta,
            ),
            # Shim attribution is additive: null fields when --vcs-shim is off
            # (unmeasured, ADR-0002), populated from the sidecar when on.
            **(shim_info if shim_info is not None else shim_metrics_unavailable("vcs_shim_disabled")),
        },
        "test_metrics": {
            # Harness oracle commands: the trusted payload check, plus the
            # behavior test run for bugfix_test_loop.
            "commands_total": tool_counts["test_commands_total"] + (2 if workflow.name == "bugfix_test_loop" else 1),
            "passed_commands": None if unmeasured else int(oracle.test_returncode == 0),
            "failed_commands": None if unmeasured else int(oracle.test_returncode != 0),
            "assertions_total": None,
            "assertions_failed": None,
            "required_checks_passed": None if unmeasured else oracle.test_returncode == 0,
        },
        "parallel_metrics": {
            "tasks_total": 1,
            "tasks_completed": 1 if success else 0,
            "max_parallel_tasks": 1,
            "overlapping_files_count": None,
            "lost_updates_detected": None,
            "lock_wait_ms": None,
            "commit_throughput_per_s": None,
            "integrity_check_passed": None,
            "measurement_source": "single_task_run; see parallel_contention.py for the contention lane",
        },
        "artifacts": {
            "transcript_path": str(execution.stdout_path),
            "command_log_path": str(execution.stderr_path),
            "diff_path": None,
            "workspace_path": str(repo),
            "extra_paths": [str(execution.prompt_path), *[str(path) for path in oracle.artifact_paths],
                            *([str(artifacts / "execution.json")] if (artifacts / "execution.json").exists() else [])],
        },
        "notes": (
            "agent_active_ms is the outer CLI wall time and may include model time plus hidden tool wait; "
            "tool counts are limited to adapter-visible transcript events. Null metric values mean "
            "unmeasured, never zero. tokens_to_recovery attribution is approximate for adapters that "
            "emit turn boundaries after tool results (codex-style streams)."
        ),
        **host_meta,
    }
    if not usage_complete:
        for key in row["tool_call_metrics"]:
            if key != "measurement_source":
                row["tool_call_metrics"][key] = None
        row["tool_call_metrics"]["measurement_source"] = "incomplete_transcript; aggregates_unmeasured"
        for key in row["failed_command_retry_metrics"]:
            if key not in {"measurement_source", "failed_command_examples"}:
                row["failed_command_retry_metrics"][key] = None
        row["failed_command_retry_metrics"]["measurement_source"] = "incomplete_transcript; examples_are_observed_only"
        for key in ("commands_total", "status_commands", "diff_commands", "snapshot_commands",
                    "branch_commands", "merge_commands", "cleanup_commands", "branches_created"):
            row["vcs_metrics"][key] = None
        row["vcs_metrics"]["measurement_source"] = "transcript_commands_unmeasured; final_state_from_oracle; shim_separate"
        row["test_metrics"]["commands_total"] = None
    return row


def copy_fixture(src: Path, dest: Path) -> None:
    if dest.exists():
        shutil.rmtree(dest)
    shutil.copytree(src, dest)


def run_one(
    agent: AgentConfig,
    subject: workflow_ab.Subject,
    workflow: workflow_ab.Workflow,
    fixture: Path,
    run_index: int,
    run_root: Path,
    artifacts_root: Path,
    timeout_seconds: float,
    track: str,
    agent_environment: str,
    instruction_level: str,
    bench_id: str,
    timestamp_utc: str,
    source: dict[str, Any],
    host_meta: dict[str, str],
    vcs_shim: bool = False,
    output_limit_bytes: int = DEFAULT_OUTPUT_LIMIT_BYTES,
    fixture_copy_validator: Callable[[Path], None] | None = None,
    fixture_prepared: bool = False,
) -> dict[str, Any]:
    repo = run_root / workflow.name / agent.name / subject.name / f"run-{run_index}"
    artifacts = artifacts_root / workflow.name / agent.name / subject.name / f"run-{run_index}"
    artifacts.mkdir(parents=True, exist_ok=True)
    if not fixture_prepared:
        copy_fixture(fixture, repo)
    if fixture_copy_validator is not None:
        fixture_copy_validator(repo)
    write_workspace_instructions(repo, subject, track, instruction_level)
    setup_ms, setup_output, setup_returncode = initialize_repo(subject, repo)

    # Workflows with prepare steps (seeded history, pre-broken state) need them
    # in the agent lane too, and untimed: they are setup, not agent work. The
    # baseline manifest is taken AFTER prepare so prepare edits are not
    # miscounted as agent changes.
    if setup_returncode == 0 and workflow.prepare is not None:
        prepare_start = time.perf_counter()
        prepare_lines: list[str] = []
        for step in workflow.prepare(subject, track):
            proc = run_command(step.command, repo)
            prepare_lines.append("$ " + workflow_ab.command_display(step.command))
            if proc.stdout:
                prepare_lines.append(proc.stdout)
            if proc.stderr:
                prepare_lines.append(proc.stderr)
            if proc.returncode not in step.expected_returncodes:
                setup_returncode = proc.returncode or 1
                break
        setup_ms += (time.perf_counter() - prepare_start) * 1000.0
        setup_output = setup_output + "\n" + "\n".join(prepare_lines)

    before_manifest = workspace_manifest(repo)
    setup_log = artifacts / "setup.log"
    setup_log.write_text(setup_output)
    initial_commit_count = commit_count(subject, repo) if setup_returncode == 0 else None
    prompt = build_prompt(subject, workflow, track, instruction_level)
    prompt_path = artifacts / "prompt.txt"
    prompt_path.write_text(prompt)

    # VCS PATH shim: wrapper git/oak executables prepended to the agent's
    # PATH record per-call timing to a sidecar JSONL. The overhead calibration
    # runs first against the same sidecar, which is then reset so calibration
    # rows are never attributed to the agent. Overhead is recorded, never
    # subtracted (ADR-0002).
    agent_env, environment_metadata = trial_agent_env(artifacts, home_root=run_root.parent / "agent-homes")
    sidecar: Path | None = None
    shim_overhead_ms: float | None = None
    if vcs_shim and setup_returncode == 0:
        real_binaries = shim_real_binaries(subject)
        sidecar = artifacts / "vcs_shim_sidecar.jsonl"
        shim_dir = create_shim_dir(
            artifacts / "vcs-shim", real_binaries=real_binaries, sidecar_path=sidecar
        )
        real_git = real_binaries.get("git")
        if real_git:
            shim_overhead_ms = measure_shim_overhead(shim_dir, sidecar, real_git)
        sidecar.write_text("")
        agent_env["PATH"] = str(shim_dir) + os.pathsep + agent_env.get("PATH", "")

    if setup_returncode != 0:
        stdout_path = artifacts / "agent.stdout.log"
        stderr_path = artifacts / "agent.stderr.log"
        stdout_path.write_text("")
        stderr_path.write_text("setup failed\n")
        execution = AgentExecution(setup_returncode, False, 0.0, stdout_path, stderr_path, prompt_path)
    elif agent.adapter == "deterministic_mock":
        execution = run_mock_agent(subject, workflow, repo, track, artifacts, prompt_path, env=agent_env)
    else:
        execution = run_real_agent(
            agent, repo, prompt, artifacts, timeout_seconds, agent_environment, env=agent_env,
            output_limit_bytes=output_limit_bytes,
        )
    if execution.capture_metadata is not None:
        execution.capture_metadata.update(environment_metadata)
        (artifacts / "execution.json").write_text(json.dumps(execution.capture_metadata, sort_keys=True) + "\n")

    shim_info: dict[str, Any] | None = None
    if vcs_shim and sidecar is not None:
        shim_info = shim_metrics_from_sidecar(sidecar, shim_overhead_ms, execution.elapsed_ms)

    if setup_returncode != 0:
        oracle = OracleResult(0, 0, 0, False, "error", "agent_setup_failed", {}, 77, 0, "", None, [], 0, 0, [],
            evidence={"oracle_version": ORACLE_VERSION, "workflow_integrity_source": "setup_failed_no_oracle",
                      "workflow_integrity_passed": None, "workflow_integrity_failure_reason": None,
                      "workflow_integrity_evidence_class": None, "workflow_integrity_unmeasured_reason": "agent_setup_failed"})
    else:
        oracle = run_oracle(subject, workflow, repo, before_manifest, initial_commit_count, artifacts)
    return make_row(
        bench_id=bench_id,
        timestamp_utc=timestamp_utc,
        profile="agent-workflow",
        agent=agent,
        subject=subject,
        workflow=workflow,
        track=track,
        run_index=run_index,
        repo=repo,
        artifacts=artifacts,
        setup_ms=setup_ms,
        setup_output_bytes=len(setup_output.encode("utf-8", "replace")),
        execution=execution,
        oracle=oracle,
        prompt=prompt,
        before_manifest=before_manifest,
        source=source,
        host_meta=host_meta,
        agent_environment=agent_environment,
        instruction_level=instruction_level,
        shim_info=shim_info,
        agent_launched=setup_returncode == 0, setup_returncode=setup_returncode,
    )


def metric_average(rows: list[dict[str, Any]], metric_getter: Any) -> dict[tuple[str, str, str], float]:
    grouped: dict[tuple[str, str, str], list[float]] = {}
    for row in rows:
        # Harness-aborted/unmeasured rows are not agent measurements; averaging
        # them in would contradict the success cell that reads unmeasured.
        if row.get("skipped") or row.get("outcome_attribution") in {"harness", "subject"}:
            continue
        value = metric_getter(row)
        if value is None:
            continue
        key = (str(row["agent"]["name"]), str(row["subject"]), str(row["scenario"]))
        grouped.setdefault(key, []).append(float(value))
    return {key: statistics.mean(values) for key, values in grouped.items()}


def token_total(row: dict[str, Any]) -> float | None:
    return shared_token_total(row)


def summary_text(rows: list[dict[str, Any]]) -> str:
    avg_ms = metric_average(rows, lambda row: row.get("elapsed_ms"))
    avg_tokens = metric_average(rows, token_total)
    avg_tools = metric_average(rows, lambda row: row["tool_call_metrics"]["tool_calls_total"])
    avg_output = metric_average(rows, lambda row: row.get("raw_output_bytes"))
    avg_turns = metric_average(rows, lambda row: row["turn_metrics"]["assistant_turns_total"])
    avg_cost = metric_average(rows, lambda row: row["token_metrics"]["cost_weighted_total"])
    avg_help = metric_average(rows, lambda row: row["tool_call_metrics"]["help_calls_total"])
    success_rate = metric_average(rows, lambda row: (1.0 if row.get("success") else 0.0)
        if row.get("agent_launched") is True and row.get("outcome_attribution") == "agent" and not row.get("skipped") else None)
    keys = sorted(set(avg_ms) | set(avg_tokens) | set(avg_tools) | set(avg_output) | set(success_rate))

    def cell(values: dict[tuple[str, str, str], float], key: tuple[str, str, str], precision: int = 1) -> str:
        value = values.get(key)
        if value is None:
            return "unmeasured"
        return f"{value:.{precision}f}"

    instruction_levels = sorted({str(row["agent"].get("instruction_level")) for row in rows})
    lines = [
        "# Oak Real Agent Workflow Summary",
        "",
        "Rows include agent attempts and setup/instrument skips. Success rates use measured agent-attributed outcomes only. Token estimates require a complete transcript.",
        f"Instruction levels in this run: {', '.join(f'`{level}`' for level in instruction_levels)}.",
        "Cost-weighted tokens use billing direction: " + COST_WEIGHTS_NOTE + ".",
        "`unmeasured` means the adapter did not expose the signal; it is not zero.",
        "",
        "| Agent | Scenario | Subject | Success rate | Avg ms | Avg tokens | Avg cost-weighted | Avg turns | Avg tool calls | Avg help calls | Avg output bytes |",
        "| --- | --- | --- | ---: | ---: | ---: | ---: | ---: | ---: | ---: | ---: |",
    ]
    for agent_name, subject, scenario in keys:
        key = (agent_name, subject, scenario)
        lines.append(
            f"| `{agent_name}` | `{scenario}` | `{subject}` | "
            f"{format(success_rate[key] * 100, '.1f') + '%' if key in success_rate else 'unmeasured'} | "
            f"{cell(avg_ms, key)} | "
            f"{cell(avg_tokens, key)} | "
            f"{cell(avg_cost, key)} | "
            f"{cell(avg_turns, key)} | "
            f"{cell(avg_tools, key)} | "
            f"{cell(avg_help, key)} | "
            f"{cell(avg_output, key, 0)} |"
        )
    return "\n".join(lines) + "\n"


def main() -> int:
    args = parse_args()
    agents = load_agents(args.agents_config, args.agents if not args.list_agents else "all")
    if args.list_agents:
        print_agents(agents)
        return 0

    subjects = workflow_ab.load_subjects(args)
    workflows = select_workflows(args.workflows)

    timestamp = datetime.now(timezone.utc).strftime("%Y%m%dT%H%M%SZ")
    timestamp_iso = datetime.now(timezone.utc).isoformat().replace("+00:00", "Z")
    args.results.mkdir(parents=True, exist_ok=True)
    args.workdir.mkdir(parents=True, exist_ok=True)
    run_root = args.workdir / "runs" / timestamp
    artifacts_root = args.results / "artifacts" / timestamp
    source = workflow_ab.source_metadata(args.oak_repo)
    host_meta = {
        "host": platform.node(),
        "platform": platform.platform(),
        "machine": platform.machine(),
    }

    rows: list[dict[str, Any]] = []
    for workflow in workflows:
        fixture = workflow_ab.prepare_fixture(args.workdir, workflow)
        print(f"[fixture] {workflow.name}: {workflow.description}", flush=True)
        for run_index in range(args.runs):
            run_agents = list(agents)
            run_subjects = list(subjects)
            if args.randomize_agent_order:
                random.Random(f"{timestamp}:{workflow.name}:{run_index}:agents").shuffle(run_agents)
            if args.randomize_subject_order:
                random.Random(f"{timestamp}:{workflow.name}:{run_index}:subjects").shuffle(run_subjects)
            for agent in run_agents:
                for subject in run_subjects:
                    print(
                        f"[run] {workflow.name} run={run_index} agent={agent.name} subject={subject.name}",
                        flush=True,
                    )
                    row = run_one(
                        agent=agent,
                        subject=subject,
                        workflow=workflow,
                        fixture=fixture,
                        run_index=run_index,
                        run_root=run_root,
                        artifacts_root=artifacts_root,
                        timeout_seconds=args.timeout_seconds,
                        track=args.track,
                        agent_environment=args.agent_environment,
                        instruction_level=args.instruction_level,
                        bench_id=timestamp,
                        timestamp_utc=timestamp_iso,
                        source=source,
                        host_meta=host_meta,
                        vcs_shim=args.vcs_shim,
                        output_limit_bytes=args.output_limit_bytes,
                    )
                    row["randomized_subject_order"] = args.randomize_subject_order
                    row["subject_order"] = [item.name for item in run_subjects]
                    rows.append(row)

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

    # The agent lane's full contract is the versioned schema, validated on
    # every run so emitter drift fails the producing run, not the dashboard.
    schema_errors: list[str] = []
    for index, row in enumerate(rows, 1):
        for error in validate_row(row):
            schema_errors.append(f"row {index}: {error}")

    if not args.keep_workdirs and run_root.exists():
        shutil.rmtree(run_root)

    print(f"[result] {raw_path}")
    print(f"[summary] {summary_path}")
    print(f"[artifacts] {artifacts_root}")
    if schema_errors:
        preview = "\n  ".join(schema_errors[:20])
        print(f"[schema] FAILED agent row schema validation:\n  {preview}")
        return 1
    return 0


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

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

RunnersDetails

No runs on main yet.
Open branchesHistory

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

May 2026now
1	`#!/usr/bin/env python3`
2	`"""Run real local coding-agent workflows against Git and Oak subjects.`
3
4	`This runner is intentionally separate from workflow_ab.py. workflow_ab.py runs`
5	`deterministic agent-shaped command recipes; this file runs actual CLI coding`
6	`agents such as Codex, Claude Code, and Cursor Agent, then normalizes their`
7	`transcripts into scripts/agent_metrics_schema.py rows.`
8	`"""`
9
10	`from __future__ import annotations`
11
12	`import argparse`
13	`import hashlib`
14	`import json`
15	`import math`
16	`import os`
17	`import platform`
18	`import random`
19	`import re`
20	`import shutil`
21	`import statistics`
22	`import subprocess`
23	`import sys`
24	`import tempfile`
25	`import time`
26	`from dataclasses import dataclass`
27	`from datetime import datetime, timezone`
28	`from pathlib import Path, PurePosixPath`
29	`from typing import Callable, Any`
30
31	`import workflow_ab`
32	`from agent_metrics_schema import SCHEMA_VERSION, validate_row`
33	`from oakbench import workflow_oracles`
34	`from oakbench.agent_execution import DEFAULT_OUTPUT_LIMIT_BYTES, EXECUTION_VERSION, run_captured`
35	`from oakbench.environment import ENV_ISOLATION_VERSION, base_env as shared_base_env`
36	`from oakbench.runner import runner_fields`
37	`from oakbench.cachectl import cache_fields`
38	`from oakbench.agent_environment import agent_environment as filtered_agent_environment`
39	`from oakbench.classify import classify_command, unwrap_shell_command, vcs_invocations`
40	`from oakbench.command_semantics import semantics`