#!/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.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"

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(

        "--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) -> tuple[dict[str, str], dict[str, Any]]:

    home = Path(tempfile.mkdtemp(prefix="agent-home-", dir=artifacts))

    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 == "bugfix_test_loop":

        return ["python3", "-m", "unittest", "discover", "-s", "tests"]

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

    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] = {}

    git_found = shutil.which("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