#!/usr/bin/env python3

"""One command, one verdict: is this Oak changeset better, same, or worse?

The inner loop for agents developing Oak against this suite:

1. Resolve (or `cargo build --release`) the local Oak binary from the Oak

   source checkout.

2. Run the comparison lanes (core, workflow, contention) with git, the

   installed Oak baseline, an A/A control (the SAME baseline binary under the

   subject name ``oak_installed_aa``) and the local changeset, all interleaved

   in randomized order within every run.

3. Judge every (scenario, operation) cell on its own evidence

   (oakbench.cellstats): the cell's A/A floor, a bootstrap CI of the relative

   median delta, an exact permutation p-value and Benjamini-Hochberg across

   the comparison family. A latency change is claimed only when BH rejects,

   the CI excludes zero and lies wholly beyond the cell's A/A floor. There is

   no global threshold imported from another profile.

4. Print a one-screen verdict and exit 0 (PASS), 1 (REGRESSED), or 2 (UNMEASURED).

Examples:

    python3 scripts/devloop.py                       # build ../oak, full quick loop

    python3 scripts/devloop.py --oak-local-bin ../oak/target/release/oak --skip-build

    python3 scripts/devloop.py --lanes core          # fastest signal only

Thresholds follow the suite's regression policy: efficiency gates (exact

metrics: tool calls, output bytes) apply to the Oak-baseline comparison only

(ADR-0004); the Git comparison is a latency guardrail.

"""

from __future__ import annotations

import argparse

import dataclasses

import json

import shutil

import statistics

import subprocess

import sys

import time

from datetime import datetime, timezone

from pathlib import Path

from typing import Any, Optional

from oakbench import cellstats

from oakbench import remotes as oakbench_remotes

from oakbench.reporting import percentile_nearest, slower_pct, symmetric_ratio_spread_pct

from oakbench.rows import output_bytes as row_output_bytes

from oakbench.rows import row_returncode

from oakbench.rows import token_total as row_token_total

from oakbench.rows import tool_calls as row_tool_calls

from oakbench.subjects import DEFAULT_OAK_REPO

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

SCRIPTS = ROOT / "scripts"

LANES = ("core", "workflow", "contention")

DEFAULT_CONTENTION_WORKERS = "2,8"

DEFAULT_CONTENTION_COMMITS_PER_WORKER = 3

DEFAULT_CONTENTION_MODES = "shared_checkout,workspace_per_task,clone_push_storm"

SKIP_RETURNCODE = 77

AA_SUBJECT = "oak_installed_aa"

# Operations excluded from the noise floor: process spawn and sub-steps have

# the largest relative jitter and are not the deltas the verdict gates on.

NOISE_FLOOR_EXCLUDED = ("proc.spawn",)

NOISE_FLOOR_MIN_PCT = 3.0

def parse_args() -> argparse.Namespace:

    parser = argparse.ArgumentParser(description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter)

    parser.add_argument("--oak-repo", type=Path, default=DEFAULT_OAK_REPO)

    parser.add_argument("--oak-local-bin", type=Path, help="Use this binary instead of building --oak-repo.")

    parser.add_argument(

        "--oak-baseline-bin",

        type=Path,

        help="Oak baseline binary (default: `oak` on PATH = released install).",

    )

    parser.add_argument("--skip-build", action="store_true", help="Never run cargo; require --oak-local-bin.")

    parser.add_argument("--lanes", default="core,workflow,contention", help=f"Comma list from: {', '.join(LANES)}")

    parser.add_argument(

        "--contention-workers",

        default=DEFAULT_CONTENTION_WORKERS,

        help="Comma-separated worker tiers for the contention lane; launch gates use 16,32,64,128.",

    )

    parser.add_argument(

        "--contention-commits-per-worker",

        type=int,

        default=DEFAULT_CONTENTION_COMMITS_PER_WORKER,

    )

    parser.add_argument(

        "--contention-modes",

        default=DEFAULT_CONTENTION_MODES,

        help="Comma-separated parallel_contention modes to run in devloop.",

    )

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

                        help="Repetitions per comparison lane (default 5; per-cell statistics need >= 3).")

    parser.add_argument("--noise-floor-runs", type=int, default=5,

                        help="Repetitions for the optional legacy global micro A/A (--global-noise-floor).")

    parser.add_argument("--skip-noise-floor", action="store_true",

                        help="Omit the in-lane A/A control subject; every latency cell is then UNMEASURED.")

    parser.add_argument("--global-noise-floor", action="store_true",

                        help="Also run the legacy micro-profile A/A (informational; never gates a cell).")

    parser.add_argument("--core-profiles", default="smoke",

                        help="Comma list of bench.py profiles for the core lane (e.g. micro,smoke).")

    parser.add_argument("--fdr", type=float, default=cellstats.DEFAULT_FDR,

                        help="Benjamini-Hochberg false discovery rate across each comparison family.")

    parser.add_argument("--results", type=Path, default=ROOT / "results" / "devloop")

    parser.add_argument("--oak-threshold-pct", type=float, default=10.0)

    parser.add_argument("--git-threshold-pct", type=float, default=25.0)

    parser.add_argument("--min-delta-ms", type=float, default=5.0)

    parser.add_argument("--output-bytes-threshold-pct", type=float, default=25.0)

    parser.add_argument(

        "--scorecard",

        action="store_true",

        help="Append a target scorecard vs the CURRENT Git Baseline Book (informational; never changes the verdict exit code).",

    )

    return parser.parse_args()

def scorecard_lines(rows: list[dict[str, Any]], root: Path = ROOT, noise_floor_pct: float | None = None,

                    cell_floors: dict[tuple[str, str], float] | None = None) -> list[str]:

    """Target-TDD lines for the devloop report; regression-TDD owns the exit code."""

    targets_path = root / "config" / "targets.json"

    if not targets_path.exists():

        return ["- no targets config (config/targets.json missing)"]

    current = root / "baselines" / "git" / "CURRENT"

    if not current.exists():

        return ["- no baseline book (baselines/git/CURRENT missing); run scripts/baseline_campaign.py run"]

    book_id = current.read_text().strip()

    book_path = root / "baselines" / "git" / book_id / "book.json"

    if not book_path.exists():

        return [f"- baseline book {book_id!r} is named by CURRENT but has no book.json"]

    from oakbench.baseline_book import load_baseline_book, load_targets

    from oakbench.scorecard import evaluate_targets

    baseline = load_baseline_book(book_path)

    if noise_floor_pct is not None:

        baseline = {

            key: dataclasses.replace(

                cell,

                noise_floor_pct=max(float(cell.noise_floor_pct or 0.0), float(noise_floor_pct)),

            )

            for key, cell in baseline.items()

        }

    if cell_floors:

        # Each scorecard cell's noise floor is at least that cell's own A/A floor.

        baseline = {

            key: dataclasses.replace(

                cell,

                noise_floor_pct=max(float(cell.noise_floor_pct or 0.0),

                                    cell_floors.get((str(getattr(cell, "scenario", "")), str(getattr(cell, "operation", ""))), 0.0)),

            )

            for key, cell in baseline.items()

        }

    results = evaluate_targets(rows, load_targets(targets_path), baseline)

    lines = []

    for item in results:

        ratio = item.get("ratio")

        ratio_text = "" if ratio is None else f" ratio {ratio:.3f} vs goal {item['goal_value']:.3f}"

        notes = "; ".join(str(note) for note in item.get("messages") or [])

        lines.append(f"- {item['verdict']} `{item['target_id']}`{ratio_text}" + (f" — {notes}" if notes else ""))

    return lines or ["- no targets defined"]

def run_lane(command: list[str], label: str) -> None:

    print(f"[devloop] {label}: {' '.join(str(part) for part in command)}", flush=True)

    proc = subprocess.run(command, cwd=ROOT, check=False)

    # Lane exit codes signal benchmark-operation failures; the verdict reads

    # those from the rows, so a nonzero lane exit is not fatal here.

    if proc.returncode not in (0, 1):

        raise SystemExit(f"{label} failed to run (exit {proc.returncode})")

def _csv(value: str) -> list[str]:

    return [item.strip() for item in value.split(",") if item.strip()]

def remote_preflight_requirements(

    lanes: list[str],

    *,

    contention_modes: list[str] | None = None,

    contention_workers: list[str] | None = None,

) -> list[oakbench_remotes.RemoteRequirement]:

    requirements: list[oakbench_remotes.RemoteRequirement] = []

    if "core" in lanes:

        requirements.append(

            oakbench_remotes.RemoteRequirement(

                purpose="default",

                lane="devloop/core",

                operations=(

                    "remote.net.push.first",

                    "remote.net.clone.cold",

                    "remote.net.fetch.uptodate",

                ),

            )

        )

    if "workflow" in lanes:

        requirements.append(

            oakbench_remotes.RemoteRequirement(

                purpose="sync",

                lane="devloop/workflow",

                operations=(

                    "sync_push_divergence/workflow.skipped",

                    "sync_pull_upstream/workflow.skipped",

                    "sync_pull_dirty/workflow.skipped",

                    "sync_non_ff_after_amend/workflow.skipped",

                ),

            )

        )

    if "contention" in lanes:

        modes = contention_modes or _csv(DEFAULT_CONTENTION_MODES)

        workers = contention_workers or _csv(DEFAULT_CONTENTION_WORKERS)

        operations: list[str] = []

        if "workspace_per_task" in modes:

            operations.extend(

                f"contention_workspace_per_task_w{workers_level}/parallel.total"

                for workers_level in workers

            )

        if "clone_push_storm" in modes:

            operations.extend(

                f"contention_clone_push_storm_w{workers_level}/parallel.total"

                for workers_level in workers

            )

        if operations:

            requirements.append(

                oakbench_remotes.RemoteRequirement(

                    purpose="fleet",

                    lane="devloop/contention",

                    operations=tuple(operations),

                )

            )

    return requirements

def build_oak_local(oak_repo: Path) -> Path:

    cargo_toml = oak_repo / "Cargo.toml"

    if not cargo_toml.exists():

        raise SystemExit(

            f"No Cargo.toml at {oak_repo}; pass --oak-local-bin or --oak-repo. "

            "devloop never mutates the Oak source checkout, it only builds it."

        )

    if shutil.which("cargo") is None:

        raise SystemExit("cargo is not installed; pass --oak-local-bin instead of building.")

    print(f"[devloop] building oak from {oak_repo} (cargo build --release)", flush=True)

    start = time.perf_counter()

    proc = subprocess.run(

        ["cargo", "build", "--release"],

        cwd=oak_repo,

        text=True,

        stdout=subprocess.PIPE,

        stderr=subprocess.STDOUT,

        check=False,

    )

    elapsed = time.perf_counter() - start

    if proc.returncode != 0:

        tail = "\n".join(proc.stdout.splitlines()[-25:])

        raise SystemExit(f"cargo build failed (exit {proc.returncode}) after {elapsed:.0f}s:\n{tail}")

    binary = oak_repo / "target" / "release" / "oak"

    if not binary.exists():

        raise SystemExit(f"build succeeded but {binary} does not exist")

    print(f"[devloop] built {binary} in {elapsed:.0f}s", flush=True)

    return binary

def load_jsonl(path: Path) -> list[dict[str, Any]]:

    try:

        text = path.read_text()

    except FileNotFoundError:

        raise SystemExit(f"[devloop] result JSONL missing: {path}") from None

    except OSError as exc:

        raise SystemExit(f"[devloop] result JSONL unreadable: {path}: {exc}") from None

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

    for line_number, line in enumerate(text.splitlines(), start=1):

        stripped = line.strip()

        if not stripped:

            continue

        try:

            row = json.loads(stripped)

        except json.JSONDecodeError as exc:

            raise SystemExit(

                f"[devloop] malformed result JSONL: {path}:{line_number}: {exc.msg}"

            ) from None

        if not isinstance(row, dict):

            raise SystemExit(f"[devloop] result JSONL row is not an object: {path}:{line_number}") from None

        rows.append(row)

    if not rows:

        raise SystemExit(f"[devloop] result JSONL contains no rows: {path}")

    return rows

def median_by_key(

    rows: list[dict[str, Any]], value_of, *, successful_only: bool = True

) -> dict[tuple[str, str, str], float]:

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

    for row in rows:

        operation = str(row.get("operation") or "")

        if operation.endswith((".add", ".commit")):

            continue

        if successful_only and row_returncode(row) != 0:

            continue

        value = value_of(row)

        if value is None:

            continue

        key = (str(row.get("subject")), str(row.get("scenario")), operation)

        grouped.setdefault(key, []).append(float(value))

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

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

    counts: dict[tuple[str, str, str], dict[str, int]] = {}

    for row in rows:

        operation = str(row.get("operation") or "")

        if operation.endswith((".add", ".commit")):

            continue

        key = (str(row.get("subject")), str(row.get("scenario")), operation)

        item = counts.setdefault(key, {"success": 0, "failure": 0, "skip": 0})

        returncode = row_returncode(row)

        if returncode == 0 and not row.get("skipped"):

            item["success"] += 1

        elif returncode == SKIP_RETURNCODE or row.get("skipped"):

            item["skip"] += 1

        else:

            item["failure"] += 1

    return counts

def diagnostic_operation(operation: str) -> bool:

    return operation.endswith((".determinism", ".inforecall"))

def measure_noise_floor(aa_dir: Path, baseline_bin: Path, runs: int) -> tuple[float, int]:

    """A/A null test: same binary as both subjects. Returns (p95 symmetric ratio spread, ops)."""

    run_lane(

        [

            sys.executable,

            str(SCRIPTS / "bench.py"),

            "--profile",

            "micro",

            "--runs",

            str(runs),

            "--subjects",

            "oak_installed,oak_local",

            "--oak-installed-bin",

            str(baseline_bin),

            "--oak-local-bin",

            str(baseline_bin),

            "--skip-determinism-probe",

            "--skip-remote",

            "--randomize-subject-order",

            "--results",

            str(aa_dir),

        ],

        "noise floor (A/A)",

    )

    rows = load_jsonl(aa_dir / "latest.jsonl")

    medians = median_by_key(rows, lambda row: row.get("elapsed_ms"))

    spreads: list[float] = []

    for (subject, scenario, operation), value in medians.items():

        if subject != "oak_installed" or operation in NOISE_FLOOR_EXCLUDED or operation.startswith("mode.setup"):

            continue

        other = medians.get(("oak_local", scenario, operation))

        pct = symmetric_ratio_spread_pct(value, other)

        if pct is not None:

            spreads.append(pct)

    if not spreads:

        return NOISE_FLOOR_MIN_PCT, 0

    p95 = percentile_nearest(spreads, 95.0) or NOISE_FLOOR_MIN_PCT

    return max(p95, NOISE_FLOOR_MIN_PCT), len(spreads)

def vcs_operations(rows: list[dict[str, Any]]) -> set[tuple[str, str]]:

    """(scenario, operation) pairs that actually invoke the subject's VCS.

    Latency gates apply only to these: search/read/edit/test steps run

    identical commands for every subject, so a delta there is host noise, not

    the changeset.

    """

    ops: set[tuple[str, str]] = set()

    for row in rows:

        if int(row.get("vcs_tool_call_count", 0) or 0) >= 1:

            ops.add((str(row.get("scenario")), str(row.get("operation"))))

    return ops

def embedded_null_p95(rows: list[dict[str, Any]], target: str, baseline: str) -> tuple[float, int]:

    """Noise estimate from identical-command steps: the spread between subjects

    on operations with zero VCS calls is what zero looks like in this lane."""

    latency = median_by_key(rows, lambda row: row.get("elapsed_ms"))

    vcs_ops = vcs_operations(rows)

    outcomes = outcome_counts_by_key(rows)

    spreads: list[float] = []

    for (subject, scenario, operation), value in latency.items():

        if subject != baseline or (scenario, operation) in vcs_ops or diagnostic_operation(operation):

            continue

        baseline_outcome = outcomes.get((baseline, scenario, operation), {})

        target_outcome = outcomes.get((target, scenario, operation), {})

        if baseline_outcome.get("failure", 0) or baseline_outcome.get("skip", 0):

            continue

        if target_outcome.get("failure", 0) or target_outcome.get("skip", 0):

            continue

        pct = symmetric_ratio_spread_pct(value, latency.get((target, scenario, operation)))

        if pct is not None:

            spreads.append(pct)

    if not spreads:

        return 0.0, 0

    return percentile_nearest(spreads, 95.0) or 0.0, len(spreads)

def same_patch_failure(left: dict[str, Any], right: dict[str, Any]) -> bool:

    if left.get("patch_failure_source") == "subject_tree_manifest_comparison":

        fields = ("patch_failure_source", "patch_failure_reason", "patch_oracle_version",

                  "patch_input_sha256", "patch_expected_tree_sha256", "patch_subject_tree_after_sha256")

        return (left.get("patch_apply_ok") is False and right.get("patch_apply_ok") is False

                and left.get("patch_failure_reason") == "diff_mutated_subject_tree"

                and all(isinstance(left.get(field), str) and len(left[field]) == 64 for field in fields[3:])

                and all(left.get(field) == right.get(field) for field in fields))

    fields = ("patch_input_sha256", "patch_expected_tree_sha256", "patch_apply_returncode",

              "patch_failure_reason", "patch_oracle_version", "patch_instrument_sha256",

              "patch_instrument_version")

    fingerprints = all(isinstance(left.get(field), str) and len(left[field]) == 64

                       for field in fields[:2])

    return (fingerprints and left.get("patch_apply_ok") is False and right.get("patch_apply_ok") is False

            and type(left.get("patch_apply_returncode")) is int

            and bool(left.get("patch_failure_reason")) and bool(left.get("patch_oracle_version"))

            and isinstance(left.get("patch_instrument_sha256"), str) and len(left["patch_instrument_sha256"]) == 64

            and bool(left.get("patch_instrument_version"))

            and all(left.get(field) == right.get(field) for field in fields))

def preexisting_patch_breach(breach: str, rows: list[dict[str, Any]]) -> bool:

    target_rows = [row for row in rows if row.get("subject") == "oak_local"

                   and breach.startswith(f"{breach.split(' ', 1)[0]} `{row.get('scenario')}/{row.get('operation')}`:")]

    if not target_rows:

        return False

    for target_row in target_rows:

        baseline_rows = [row for row in rows if row.get("subject") == "oak_installed"

                         and (row.get("scenario"), row.get("operation")) ==

                         (target_row.get("scenario"), target_row.get("operation"))]

        baseline_rows = [base for base in baseline_rows if base.get("run") == target_row.get("run")]

        if not baseline_rows or not all(same_patch_failure(target_row, base) for base in baseline_rows):

            return False

    return True

def compare_subjects(

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

    target: str,

    baseline: str,

    label: str,

    latency_threshold_pct: float,

    min_delta_ms: float,

    *,

    efficiency_gates: bool,

    output_bytes_threshold_pct: float,

    diagnostics: list[str] | None = None,

    cell_verdicts: dict[tuple[str, str], dict[str, Any]] | None = None,

    unconfirmed_prefix: str = "UNCONFIRMED",

) -> tuple[list[str], list[str]]:

    """Returns (regressions, improvements) lines for one baseline comparison.

    With ``cell_verdicts`` (oakbench.cellstats), latency is judged per cell;

    the legacy single threshold is used only when none are supplied.

    """

    diagnostics = diagnostics if diagnostics is not None else []

    gate_ops = vcs_operations(rows)

    latency = median_by_key(rows, lambda row: row.get("elapsed_ms"))

    tools = median_by_key(rows, row_tool_calls)

    bytes_out = median_by_key(rows, row_output_bytes)

    tokens = median_by_key(rows, row_token_total)

    outcomes = outcome_counts_by_key(rows)

    regressions: list[str] = []

    improvements: list[str] = []

    scenario_ops = sorted(

        {

            (scenario, op)

            for subject, scenario, op in set(latency) | set(outcomes)

            if subject in {target, baseline}

        }

    )

    target_kinds = {row.get("subject_kind") for row in rows if row.get("subject") == target} - {None}

    baseline_kinds = {row.get("subject_kind") for row in rows if row.get("subject") == baseline} - {None}

    incompatible_remote = (target_kinds == {"oak"} and baseline_kinds == {"git"}) or (target_kinds == {"git"} and baseline_kinds == {"oak"})

    for scenario, operation in scenario_ops:

        if incompatible_remote and operation.startswith("remote."):

            diagnostics.append(f"INFO TRANSPORT `{scenario}/{operation}`: local Git and network Oak operations are not comparable")

            continue

        if (scenario, operation) not in gate_ops or diagnostic_operation(operation):

            continue

        target_key = (target, scenario, operation)

        baseline_key = (baseline, scenario, operation)

        target_outcome = outcomes.get(target_key, {"success": 0, "failure": 0, "skip": 0})

        baseline_outcome = outcomes.get(baseline_key, {"success": 0, "failure": 0, "skip": 0})

        target_successes = target_outcome["success"]

        target_failures = target_outcome["failure"]

        target_skips = target_outcome["skip"]

        baseline_successes = baseline_outcome["success"]

        baseline_failures = baseline_outcome["failure"]

        if baseline_successes and target_successes == 0 and target_failures == 0 and target_skips > 0:

            diagnostics.append(f"UNMEASURED `{scenario}/{operation}` vs {label}: "

                               f"target has no measured attempts ({target_skips} skipped)")

        elif baseline_successes and target_successes == 0:

            regressions.append(

                f"NO SUCCESS `{scenario}/{operation}` vs {label}: target had 0 successful run(s) "

                f"({target_failures} failed, {target_skips} skipped), baseline had {baseline_successes}"

            )

        else:

            target_attempts = target_successes + target_failures

            baseline_attempts = baseline_successes + baseline_failures

            if target_attempts and baseline_attempts:

                target_failure_rate = target_failures / target_attempts

                baseline_failure_rate = baseline_failures / baseline_attempts

                target_less_reliable = target_failures > baseline_failures or target_successes < baseline_successes

                if target_failure_rate > baseline_failure_rate and target_less_reliable:

                    if baseline_failures == 0:

                        regressions.append(

                            f"FAIL `{scenario}/{operation}` vs {label}: {target_failures} failed run(s), "

                            "baseline clean"

                        )

                    else:

                        regressions.append(

                            f"FAIL-RATE `{scenario}/{operation}` vs {label}: "

                            f"{baseline_failures}/{baseline_attempts} -> {target_failures}/{target_attempts} "

                            f"failed run(s); successes {baseline_successes} -> {target_successes}"

                        )

        if baseline_key not in latency:

            continue

        pct = slower_pct(latency[baseline_key], latency.get(target_key))

        delta = (latency.get(target_key) or 0) - latency[baseline_key]

        if cell_verdicts is not None:

            item = cell_verdicts.get((scenario, operation))

            verdict = item.get("verdict") if item else "unmeasured"

            if verdict == "regression":

                regressions.append("SLOWER " + cellstats.describe(item, label))

            elif verdict == "improvement":

                improvements.append("faster " + cellstats.describe(item, label))

            elif verdict == "unconfirmed_regression":

                diagnostics.append(unconfirmed_prefix + " slower " + cellstats.describe(item, label) +

                                   f"; not claimed; ~{item.get('required_n')} runs/arm to confirm")

            elif verdict == "unconfirmed_improvement":

                diagnostics.append("INFO UNCONFIRMED faster " + cellstats.describe(item, label) + "; not claimed")

            pct = None

        if pct is not None:

            if pct >= latency_threshold_pct and delta >= min_delta_ms:

                regressions.append(

                    f"SLOWER `{scenario}/{operation}` vs {label}: {pct:+.1f}% "

                    f"({latency[baseline_key]:.1f} -> {latency[target_key]:.1f} ms)"

                )

            elif pct <= -latency_threshold_pct and -delta >= min_delta_ms:

                improvements.append(

                    f"faster `{scenario}/{operation}` vs {label}: {pct:+.1f}% "

                    f"({latency[baseline_key]:.1f} -> {latency[target_key]:.1f} ms)"

                )

        if efficiency_gates:

            tool_pct = slower_pct(tools.get(baseline_key), tools.get(target_key))

            if tool_pct is not None and tool_pct > 0:

                regressions.append(

                    f"TOOLS `{scenario}/{operation}` vs {label}: {tool_pct:+.1f}% "

                    f"({tools[baseline_key]:.0f} -> {tools[target_key]:.0f} calls)"

                )

            bytes_pct = slower_pct(bytes_out.get(baseline_key), bytes_out.get(target_key))

            if bytes_pct is not None and bytes_pct > output_bytes_threshold_pct:

                regressions.append(

                    f"OUTPUT `{scenario}/{operation}` vs {label}: {bytes_pct:+.1f}% "

                    f"({bytes_out[baseline_key]:.0f} -> {bytes_out[target_key]:.0f} bytes)"

                )

            elif bytes_pct is not None and bytes_pct < -output_bytes_threshold_pct:

                improvements.append(

                    f"leaner `{scenario}/{operation}` vs {label}: {bytes_pct:+.1f}% output bytes"

                )

            token_pct = slower_pct(tokens.get(baseline_key), tokens.get(target_key))

            if token_pct is not None and token_pct < -output_bytes_threshold_pct:

                improvements.append(f"fewer tokens `{scenario}/{operation}` vs {label}: {token_pct:+.1f}%")

    # Output-sufficiency gates: exact semantic metrics from the inforecall

    # probe rows, always on. A changeset can win every token/latency gate by

    # dropping changed-file names from output or breaking piped-diff

    # structure; that is a regression, not an optimization.

    oracle_versions: dict[tuple[str, str, str], set[Any]] = {}

    for row in rows:

        if "information_recall" in row or "information_precision" in row:

            key = (str(row.get("subject")), str(row.get("scenario")), str(row.get("operation")))

            oracle_versions.setdefault(key, set()).add(row.get("output_oracle_version"))

    incompatible_output = set()

    for (subject, scenario, operation), versions in oracle_versions.items():

        if subject != target:

            continue

        baseline_versions = oracle_versions.get((baseline, scenario, operation))

        if baseline_versions is not None and (len(versions) != 1 or versions != baseline_versions):

            incompatible_output.add((scenario, operation))

            regressions.append(f"OUTPUT-ORACLE `{scenario}/{operation}`: incompatible output oracle versions")

    output_rows = [row for row in rows if (str(row.get("scenario")), str(row.get("operation"))) not in incompatible_output]

    recall = median_by_key(output_rows, lambda row: row.get("information_recall"))

    precision = median_by_key(output_rows, lambda row: row.get("information_precision"))

    pipe_compat = median_by_key(

        rows,

        lambda row: (

            1.0

            if row.get("pipe_compatible_unified") is True

            else (0.0 if row.get("pipe_compatible_unified") is False else None)

        ),

    )

    for (subject, scenario, operation), value in sorted(recall.items()):

        if subject != target:

            continue

        base_value = recall.get((baseline, scenario, operation))

        if base_value is not None and value < base_value - 1e-6:

            regressions.append(

                f"RECALL `{scenario}/{operation}` vs {label}: changed-file recall "

                f"{base_value:.3f} -> {value:.3f} (output got cheaper by losing information)"

            )

    for (subject, scenario, operation), value in sorted(pipe_compat.items()):

        if subject != target:

            continue

        base_value = pipe_compat.get((baseline, scenario, operation))

        if base_value is not None and value < base_value - 1e-6:

            regressions.append(

                f"PIPE-COMPAT `{scenario}/{operation}` vs {label}: unified-diff structure lost "

                f"(rate {base_value:.2f} -> {value:.2f}); this legacy metric checks structure only"

            )

    for (subject, scenario, operation), value in sorted(precision.items()):

        if subject == target:

            base_value = precision.get((baseline, scenario, operation))

            if base_value is not None and value < base_value - 1e-6:

                regressions.append(f"PRECISION `{scenario}/{operation}` vs {label}: {base_value:.3f} -> {value:.3f}")

    patch_cells = {}

    for item in rows:

        if item.get("patch_application_required") is True or "patch_apply_ok" in item:

            key = (str(item.get("subject")), str(item.get("scenario")), str(item.get("operation")))

            patch_cells.setdefault(key, []).append(item)

    for scenario, operation in sorted({(key[1], key[2]) for key in patch_cells if key[0] in {target, baseline}}):

        base_rows = patch_cells.get((baseline, scenario, operation), [])

        target_rows = patch_cells.get((target, scenario, operation), [])

        proven_baseline = any(item.get("patch_apply_ok") is True and item.get("patch_application_required") is True

                              for item in base_rows)

        prefix = f"PATCH-APPLY `{scenario}/{operation}`"

        if not target_rows:

            if proven_baseline:

                regressions.append(prefix + ": target omitted previously measured patch evidence")

            elif any(base.get("patch_apply_ok") is True for base in base_rows):

                diagnostics.append("INFO " + prefix + ": target omitted previously measured optional patch evidence")

            continue

        for item in target_rows:

            if item.get("patch_apply_ok") is None:

                severity = "UNMEASURED " if item.get("patch_application_required") is True else "INFO "

                diagnostics.append(severity + prefix + ": " + str(item.get("patch_skip_reason") or "patch evidence unavailable"))

                continue

            instrument_rows = [probe for probe in [item, *[base for base in base_rows if base.get("run") == item.get("run")]]

                               if probe.get("patch_apply_ok") is not None

                               and probe.get("patch_failure_source") != "subject_tree_manifest_comparison"]

            if any(not isinstance(probe.get("patch_instrument_sha256"), str)

                   or len(probe["patch_instrument_sha256"]) != 64

                   or not probe.get("patch_instrument_version") for probe in instrument_rows):

                severity = "UNMEASURED " if item.get("patch_application_required") is True else "INFO "

                diagnostics.append(severity + prefix + ": patch instrument identity unavailable")

                continue

            versions = {base.get("patch_oracle_version") for base in base_rows if base.get("patch_apply_ok") is not None}

            if versions and versions != {item.get("patch_oracle_version")}:

                message = f"PATCH-ORACLE `{scenario}/{operation}`: incompatible or missing patch oracle version"

                if item.get("patch_application_required") is True:

                    regressions.append(message)

                else:

                    diagnostics.append("INFO " + message)

            elif item.get("patch_apply_ok") is False and (item.get("patch_application_required") is True

                                                        or item.get("patch_failure_source") == "subject_tree_manifest_comparison"):

                matched_rows = [base for base in base_rows if base.get("run") == item.get("run")]

                identical_failure = bool(matched_rows) and all(same_patch_failure(item, base) for base in matched_rows)

                if identical_failure:

                    diagnostics.append("PREEXISTING " + prefix + ": same measured defect in baseline and target")

                else:

                    regressions.append(prefix + ": exact patch application failed")

            elif item.get("patch_apply_ok") is False:

                diagnostics.append("INFO " + prefix + ": optional exact patch application failed")

    return regressions, improvements

def contention_checks(rows: list[dict[str, Any]], target: str, baseline: str) -> tuple[list[str], list[str]]:

    regressions: list[str] = []

    improvements: list[str] = []

    totals: dict[tuple[str, str], dict[str, Any]] = {}

    for row in rows:

        if row.get("operation") != "parallel.total" or row.get("skipped"):

            continue

        totals[(str(row.get("subject")), str(row.get("scenario")))] = row

    for (subject, scenario), row in sorted(totals.items()):

        if subject != target:

            continue

        metrics = row.get("parallel_metrics", {})

        if metrics.get("integrity_check_passed") is False:

            regressions.append(f"INTEGRITY `{scenario}`: post-contention check failed for {target}")

        if (metrics.get("lost_updates_detected") or 0) > 0:

            regressions.append(

                f"LOST UPDATES `{scenario}`: {metrics['lost_updates_detected']} payload(s) lost for {target}"

            )

        base = totals.get((baseline, scenario))

        if base:

            target_tp = metrics.get("commit_throughput_per_s")

            base_tp = base.get("parallel_metrics", {}).get("commit_throughput_per_s")

            if target_tp and base_tp:

                pct = ((target_tp - base_tp) / base_tp) * 100.0

                if pct < -15.0:

                    regressions.append(

                        f"THROUGHPUT `{scenario}` vs Oak baseline: {pct:+.1f}% "

                        f"({base_tp:.1f} -> {target_tp:.1f} snapshots/s)"

                    )

                elif pct > 15.0:

                    improvements.append(f"throughput `{scenario}` vs Oak baseline: {pct:+.1f}%")

    return regressions, improvements

def main() -> int:

    args = parse_args()

    lanes = [lane.strip() for lane in args.lanes.split(",") if lane.strip()]

    unknown = [lane for lane in lanes if lane not in LANES]

    if unknown:

        raise SystemExit(f"unknown lanes: {', '.join(unknown)}; valid: {', '.join(LANES)}")

    contention_modes = _csv(args.contention_modes)

    contention_workers = _csv(args.contention_workers)

    oakbench_remotes.print_remote_preflight_warnings(

        oakbench_remotes.oak_remote_preflight_warnings(

            remote_preflight_requirements(

                lanes,

                contention_modes=contention_modes,

                contention_workers=contention_workers,

            )

        )

    )

    baseline_bin = args.oak_baseline_bin or (Path(shutil.which("oak")) if shutil.which("oak") else None)

    if baseline_bin is None:

        raise SystemExit("No Oak baseline: install oak or pass --oak-baseline-bin")

    if args.oak_local_bin:

        local_bin = args.oak_local_bin.resolve()

        if not local_bin.exists():

            raise SystemExit(f"--oak-local-bin does not exist: {local_bin}")

    elif args.skip_build:

        raise SystemExit("--skip-build requires --oak-local-bin")

    else:

        local_bin = build_oak_local(args.oak_repo)

    bench_id = datetime.now(timezone.utc).strftime("%Y%m%dT%H%M%SZ")

    out = args.results / bench_id

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

    started = time.perf_counter()

    noise_pct = NOISE_FLOOR_MIN_PCT

    noise_ops = 0

    if args.global_noise_floor:

        noise_pct, noise_ops = measure_noise_floor(out / "aa", baseline_bin, args.noise_floor_runs)

    effective_oak_threshold = max(args.oak_threshold_pct, 2.0 * noise_pct)

    effective_git_threshold = max(args.git_threshold_pct, 2.0 * noise_pct)

    if args.runs < cellstats.MIN_SAMPLES:

        print(f"[devloop] WARNING --runs {args.runs} < {cellstats.MIN_SAMPLES}: every latency cell will be UNMEASURED",

              flush=True)

    aa_names = [] if args.skip_noise_floor else [AA_SUBJECT]

    subject_flags = [

        "--subjects",

        ",".join(["git", "oak_installed", *aa_names, "oak_local"]),

        "--oak-installed-bin",

        str(baseline_bin),

        *([] if args.skip_noise_floor else ["--oak-aa-bin", str(baseline_bin)]),

        "--oak-local-bin",

        str(local_bin),

    ]

    core_profiles = _csv(args.core_profiles)

    lane_dirs: list[tuple[str, Path]] = []

    if "core" in lanes:

        for profile in core_profiles:

            directory = out / ("core" if core_profiles == ["smoke"] else f"core-{profile}")

            run_lane(

                [

                    sys.executable,

                    str(SCRIPTS / "bench.py"),

                    "--profile",

                    profile,

                    "--runs",

                    str(args.runs),

                    "--randomize-subject-order",

                    *subject_flags,

                    "--results",

                    str(directory),

                ],

                f"core lane ({profile})",

            )

            lane_dirs.append(("core" if core_profiles == ["smoke"] else f"core:{profile}", directory))

    if "workflow" in lanes:

        run_lane(

            [

                sys.executable,

                str(SCRIPTS / "workflow_ab.py"),

                "--workflows",

                "all",

                "--runs",

                str(args.runs),

                "--randomize-subject-order",

                *subject_flags,

                "--results",

                str(out / "workflow"),

            ],

            "workflow lane",

        )

        lane_dirs.append(("workflow", out / "workflow"))

    if "contention" in lanes:

        run_lane(

            [

                sys.executable,

                str(SCRIPTS / "parallel_contention.py"),

                "--workers",

                args.contention_workers,

                "--commits-per-worker",

                str(args.contention_commits_per_worker),

                "--modes",

                args.contention_modes,

                *subject_flags,

                "--results",

                str(out / "contention"),

            ],

            "contention lane",

        )

    regressions: list[str] = []

    improvements: list[str] = []

    guardrails: list[str] = []

    known_gaps: list[str] = []

    null_notes: list[str] = []

    measurement_notes: list[str] = []

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

    cell_notes: list[str] = []

    cell_floor_map: dict[tuple[str, str], float] = {}

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

    for lane, directory in lane_dirs:

        rows = load_jsonl(directory / "latest.jsonl")

        scorecard_rows += rows

        latency_of = lambda row: row.get("elapsed_ms")

        gated = {key for key in vcs_operations(rows) if not diagnostic_operation(key[1])

                 and not key[1].endswith((".add", ".commit"))}

        floors = cellstats.cell_floors(rows, "oak_installed", AA_SUBJECT, latency_of) if aa_names else {}

        for key, value in floors.items():

            cell_floor_map[key] = max(value, cell_floor_map.get(key, 0.0))

        def verdicts(target: str, baseline: str, min_delta: float) -> dict[tuple[str, str], dict[str, Any]]:

            return cellstats.classify_cells(rows, target, baseline, floors, value_of=latency_of,

                                            cells=gated, min_delta_ms=min_delta, fdr=args.fdr)

        oak_cells = verdicts("oak_local", "oak_installed", args.min_delta_ms)

        git_min_delta = max(args.min_delta_ms, 10.0)

        local_git_cells = verdicts("oak_local", "git", git_min_delta)

        baseline_git_cells = verdicts("oak_installed", "git", git_min_delta)

        aa_cells = verdicts(AA_SUBJECT, "oak_installed", args.min_delta_ms) if aa_names else {}

        tested = [item for item in oak_cells.values() if item["p_value"] is not None]

        unmeasured = [item for item in oak_cells.values() if item["verdict"] == "unmeasured"]

        if floors:

            ordered = sorted(floors.values())

            cell_notes.append(

                f"[{lane}] per-cell A/A floors ({len(floors)} cells): median {statistics.median(ordered):.1f}%, "

                f"max {ordered[-1]:.1f}%; {len(tested)} cells tested, BH q={args.fdr}; "

                f"{len(unmeasured)} cell(s) unmeasured")

        else:

            cell_notes.append(f"[{lane}] no A/A control rows: every latency cell is UNMEASURED")

        cell_records.extend({"lane": lane, "comparison": name, **item}

                            for name, family in (("oak_local_vs_oak_installed", oak_cells),

                                                 ("oak_local_vs_git", local_git_cells),

                                                 ("oak_installed_vs_git", baseline_git_cells),

                                                 (AA_SUBJECT + "_vs_oak_installed", aa_cells))

                            for item in family.values())

        aa_changes = [item for item in aa_cells.values() if item["verdict"] in ("regression", "improvement")]

        if aa_changes:

            measurement_notes.append(f"UNMEASURED [{lane}] A/A control shows {len(aa_changes)} change(s) "

                                     "between identical binaries; host too noisy for latency verdicts")

        # Per-lane embedded null control: identical-command steps (search,

        # read, edit, test) cannot differ between subjects, so their spread is

        # this lane's live noise measurement; thresholds rise to clear it.

        null_pct, null_ops = embedded_null_p95(rows, "oak_local", "oak_installed")

        lane_oak_threshold = max(effective_oak_threshold, 2.0 * null_pct)

        lane_git_threshold = max(effective_git_threshold, 2.0 * null_pct)

        if null_ops:

            null_notes.append(

                f"[{lane}] embedded null ({null_ops} identical-command ops): p95 spread {null_pct:.1f}%; "

                f"lane threshold {lane_oak_threshold:.1f}%"

            )

        lane_regressions, lane_improvements = compare_subjects(

            rows,

            "oak_local",

            "oak_installed",

            "Oak baseline",

            lane_oak_threshold,

            args.min_delta_ms,

            efficiency_gates=True,

            output_bytes_threshold_pct=args.output_bytes_threshold_pct,

            diagnostics=measurement_notes,

            cell_verdicts=oak_cells,

        )

        regressions += [f"[{lane}] {item}" for item in lane_regressions]

        improvements += [f"[{lane}] {item}" for item in lane_improvements]

        # Git guardrail: only breaches the CHANGESET introduced fail the

        # verdict. Pre-existing Oak-vs-Git gaps (also breached by the installed

        # baseline) are listed as known gaps — a verdict that fails every run

        # on inherited gaps teaches agents to ignore it (ADR-0004's lesson).

        local_breaches, git_improvements = compare_subjects(

            rows,

            "oak_local",

            "git",

            "Git guardrail",

            lane_git_threshold,

            max(args.min_delta_ms, 10.0),

            efficiency_gates=False,

            output_bytes_threshold_pct=args.output_bytes_threshold_pct,

            diagnostics=measurement_notes,

            cell_verdicts=local_git_cells,

            # Git-guardrail candidates inform; only Oak-baseline candidates make the verdict inconclusive.

            unconfirmed_prefix="INFO UNCONFIRMED",

        )

        baseline_breaches, _ = compare_subjects(

            rows,

            "oak_installed",

            "git",

            "Git guardrail",

            lane_git_threshold,

            max(args.min_delta_ms, 10.0),

            efficiency_gates=False,

            output_bytes_threshold_pct=args.output_bytes_threshold_pct,

            cell_verdicts=baseline_git_cells,

        )

        preexisting = {breach.split(":")[0] for breach in baseline_breaches}

        for breach in local_breaches:

            is_patch = breach.startswith(("PATCH-APPLY ", "PATCH-ORACLE "))

            if breach.split(":")[0] in preexisting and (not is_patch or preexisting_patch_breach(breach, rows)):

                known_gaps.append(f"[{lane}] {breach}")

            else:

                guardrails.append(f"[{lane}] {breach}")

        improvements += [f"[{lane}] {item}" for item in git_improvements]

    if "contention" in lanes:

        rows = load_jsonl(out / "contention" / "latest.jsonl")

        scorecard_rows += rows

        lane_regressions, lane_improvements = contention_checks(rows, "oak_local", "oak_installed")

        regressions += [f"[contention] {item}" for item in lane_regressions]

        improvements += [f"[contention] {item}" for item in lane_improvements]

    elapsed_s = time.perf_counter() - started

    verdict = ("REGRESSED" if regressions or guardrails else

               "UNMEASURED" if any(note.startswith(("UNMEASURED", "UNCONFIRMED")) for note in measurement_notes)

               else "PASS")

    lines = [

        "# Devloop Verdict",

        "",

        f"**{verdict}** — oak_local ({local_bin}) vs Oak baseline ({baseline_bin}) and Git, "

        f"lanes: {', '.join(lanes)}, runs={args.runs}, total {elapsed_s:.0f}s.",

        "",

        f"- Latency verdicts are per cell (cellstats {cellstats.CELL_STATS_VERSION}): change only when "

        f"BH-FDR q={args.fdr} rejects AND the 95% bootstrap CI excludes 0 AND lies wholly beyond that "

        f"cell's A/A floor ({AA_SUBJECT} = baseline binary, interleaved). UNCONFIRMED candidates are never claimed.",

        *[f"- {note}" for note in cell_notes],

        *([f"- Legacy global micro A/A (informational only, {noise_ops} ops): p95 spread {noise_pct:.1f}%."]

          if args.global_noise_floor else []),

        f"- Efficiency gates (exact metrics) vs Oak baseline only (ADR-0004): any tool-call increase, "

        f">{args.output_bytes_threshold_pct:.0f}% output bytes.",

        *[f"- {note}" for note in null_notes],

        "",

        f"## Regressions vs Oak baseline ({len(regressions)})",

        "",

        *([f"- {item}" for item in regressions] or ["- none"]),

        "",

        f"## New Git-guardrail breaches introduced by this changeset ({len(guardrails)})",

        "",

        *([f"- {item}" for item in guardrails] or ["- none"]),

        "",

        f"## Known Git gaps, pre-existing in the baseline ({len(known_gaps)}) — tracked, not failing",

        "",

        *([f"- {item}" for item in known_gaps] or ["- none"]),

        "",

        "## Correctness measurement notes",

        "",

        *([f"- {item}" for item in dict.fromkeys(measurement_notes)] or ["- none"]),

        "",

        f"## Improvements ({len(improvements)})",

        "",

        *([f"- {item}" for item in improvements[:20]] or ["- none"]),

        "",

        f"Raw rows: {out}/ (per-cell decisions: cells.json)",

    ]

    if args.scorecard:

        lines += [

            "",

            "## Scorecard vs Git Baseline Book (informational)",

            "",

            *scorecard_lines(scorecard_rows, noise_floor_pct=noise_pct if args.global_noise_floor else None,

                             cell_floors=cell_floor_map),

        ]

    report = "\n".join(lines) + "\n"

    (out / "verdict.md").write_text(report)

    # Every per-cell decision (floor, CI, p, BH) for audit; derived, never committed.

    (out / "cells.json").write_text(json.dumps(cell_records, indent=1, sort_keys=True) + "\n")

    print()

    print(report)

    return 0 if verdict == "PASS" else (2 if verdict == "UNMEASURED" else 1)

if __name__ == "__main__":

    raise SystemExit(main())