#!/usr/bin/env python3

"""agent_loop_hosted_v1: one agent task loop against HOSTED Oak vs GitHub.

Measures each step of a realistic agent loop -- acquire, branch, edit+commit,

inspect, publish, review another branch, main advances (k=1, k=20) -> sync,

merge to main, start the next task warm -- against oak.space and github.com.

Per step it records wall time, commands (tool calls), output bytes, and on

proxied trials the HTTP request count / bytes / serial request rounds:

* Oak (``--mode proxied``): the clone points at a local plaintext->TLS

  forwarding proxy (``oakbench.httpcount``), so every API request to

  oak.space is counted with method/path/status/bytes/timing. Chunk traffic

  to ``cdn.oak.space`` goes through the same proxy as HTTPS CONNECT tunnels

  (``HTTPS_PROXY``): bytes and connection counts are measured, CDN request

  counts are not (null, ADR-0002). Auth comes from ``OAK_API_KEY`` read from

  ``~/.oak/credentials`` (never printed or logged).

* Git (``--mode proxied``): ``http.proxy`` routes github.com through the same

  proxy as CONNECT tunnels (bytes, connections), and ``GIT_TRACE_CURL``

  (headers only; Authorization is redacted by Git and the trace file is

  parsed then deleted) supplies the request count.

* ``--mode direct``: no instrumentation; timing only. Proxied timing is

  diagnostic (HTTP/1.1 through a local hop; direct Oak uses HTTP/2).

Oracles are independent of the step's success output: remote heads via the

Oak REST API / ``git ls-remote``; working-tree content via Git blob ids of the

files on disk compared to the fixture's expected tree.

Lane is excluded from devloop: it depends on a real network and hosted

services (external variance). Hosted CI is out of scope (fixture carries no CI

config). Setup (actor commits, fixture seeding) is never timed.

"""

from __future__ import annotations

import argparse

import hashlib

import json

import os

import platform

import re

import shutil

import statistics

import subprocess

import sys

import tempfile

import time

import urllib.request

from dataclasses import dataclass, field

from pathlib import Path

from typing import Optional

sys.path.insert(0, str(Path(__file__).resolve().parent))

from oakbench.httpcount import HttpCountingProxy  # noqa: E402

from oakbench.runlock import measurement_lock  # noqa: E402

SCENARIO = "agent_loop_hosted_v1"

SCENARIO_VERSION = 1

OAK_HOST = "oak.space"

GITHUB_HOST = "github.com"

# Immutable operation vocabulary (ADR-0005).

OPERATIONS = (

    "acquire.cold",

    "branch.create",

    "edit.commit",

    "inspect.status",

    "inspect.diff_main",

    "inspect.log",

    "publish.branch",

    "inspect.review_remote",

    "inspect.main_since_branch.k1",

    "sync.k1",

    "sync.k20",

    "merge.main",

    "next_task.warm",

)

# ---------------------------------------------------------------------------

# Fixed task patch: 3 files, ~50 added lines, disjoint from actor files.

# ---------------------------------------------------------------------------

def task_patch(tag: str) -> dict:

    lines_a = [f"// agent task {tag} line {i}: widen the retry budget for step {i}" for i in range(20)]

    lines_b = [f"- agent task {tag} note {i}" for i in range(15)]

    lines_c = [f"agent task {tag} doc {i}" for i in range(15)]

    return {

        "cli/src/agent_task_note.rs": ("new", "\n".join(lines_a) + "\n"),

        "README.md": ("append", "\n" + "\n".join(lines_b) + "\n"),

        "docs/agent-task-notes.md": ("new", "\n".join(lines_c) + "\n"),

    }

def apply_patch(root: Path, patch: dict) -> None:

    for rel, (kind, text) in patch.items():

        p = root / rel

        p.parent.mkdir(parents=True, exist_ok=True)

        if kind == "append":

            with p.open("a") as fh:

                fh.write(text)

        else:

            p.write_text(text)

def actor_edit(root: Path, tag: str, i: int) -> str:

    rel = f"bench-history/actor-{tag}.txt"

    p = root / rel

    p.parent.mkdir(parents=True, exist_ok=True)

    with p.open("a") as fh:

        fh.write(f"actor {tag} commit {i}\n")

    return rel

# ---------------------------------------------------------------------------

# Oracles

# ---------------------------------------------------------------------------

def git_blob_id(data: bytes) -> str:

    return hashlib.sha1(b"blob %d\0" % len(data) + data).hexdigest()

def workspace_digest(root: Path) -> dict:

    out = {}

    for dirpath, dirs, files in os.walk(root):

        rel = Path(dirpath).relative_to(root)

        if rel == Path("."):

            dirs[:] = [d for d in dirs if d not in (".git", ".oak")]

        for f in files:

            p = Path(dirpath) / f

            if rel == Path(".") and f == ".git":  # worktree gitfile

                continue

            if p.is_symlink():

                continue

            out[str(rel / f) if str(rel) != "." else f] = git_blob_id(p.read_bytes())

    return out

def fixture_tree(git_dir: Path, rev: str) -> dict:

    out = {}

    txt = subprocess.run(["git", "-C", str(git_dir), "ls-tree", "-r", rev], capture_output=True, text=True, check=True).stdout

    for line in txt.splitlines():

        meta, path = line.split("\t", 1)

        out[path] = meta.split()[2]

    return out

TASK_PATHS = {"README.md", "cli/src/agent_task_note.rs", "docs/agent-task-notes.md"}

def acquire_oracle(ws: Path, fixture_git: Path):

    """Every fixture path present with fixture content, except paths earlier

    loop trials legitimately change (task patch files, actor files)."""

    dig = workspace_digest(ws)

    exp = fixture_tree(fixture_git, "HEAD")

    missing = [p for p in exp if p not in dig]

    differ = [p for p in exp if p in dig and dig[p] != exp[p] and p not in TASK_PATHS]

    extra = [p for p in dig if p not in exp and p not in TASK_PATHS and not p.startswith("bench-history/actor-")]

    ok = not missing and not differ and not extra

    return ok, f"files={len(dig)} expected={len(exp)} missing={len(missing)} differ={len(differ)} extra={len(extra)}"

def oak_token() -> str:

    for c in json.load(open(os.path.expanduser("~/.oak/credentials"))):

        if c.get("server") == f"https://{OAK_HOST}":

            return c["token"]

    raise SystemExit("no oak.space credential")

def oak_api(path: str, token: str):

    req = urllib.request.Request(f"https://{OAK_HOST}{path}", headers={"Authorization": f"Bearer {token}", "User-Agent": "oakbench-oracle"})

    with urllib.request.urlopen(req, timeout=60) as r:

        return json.loads(r.read())

def oak_remote_heads(repo: str, token: str) -> dict:

    d = oak_api(f"/api/{repo}/branches", token)

    return {b["name"]: b["head"] for b in d.get("branches", [])}

def git_remote_heads(url: str, env: dict) -> dict:

    out = subprocess.run(["git", "ls-remote", url], capture_output=True, text=True, env=env).stdout

    heads = {}

    for line in out.splitlines():

        sha, ref = line.split("\t")

        if ref.startswith("refs/heads/"):

            heads[ref[len("refs/heads/"):]] = sha

    return heads

# ---------------------------------------------------------------------------

# Step execution

# ---------------------------------------------------------------------------

@dataclass

class StepResult:

    op: str

    commands: list

    wall_s: float

    per_command_s: list

    returncodes: list

    output_bytes: int

    requests: Optional[int] = None

    request_bytes_up: Optional[int] = None

    request_bytes_down: Optional[int] = None

    api_connections: Optional[int] = None

    tunnel_connections: Optional[int] = None

    tunnel_bytes: Optional[int] = None

    tunnel_hosts: Optional[dict] = None

    serial_rounds: Optional[int] = None

    routes: Optional[dict] = None

    timeline: Optional[list] = None

    oracle_ok: Optional[bool] = None

    oracle_detail: str = ""

    stdout_tail: str = ""

    last_stdout: str = ""

def serial_rounds(records) -> Optional[int]:

    """Number of request 'waves': a new wave starts when a request begins after

    every earlier request has completed (i.e., it had to wait a round trip)."""

    if not records:

        return 0

    recs = sorted(records, key=lambda r: r.start_ms)

    rounds, horizon = 0, -1.0

    for r in recs:

        end = r.end_ms if r.end_ms is not None else r.start_ms

        if r.start_ms >= horizon:

            rounds += 1

        horizon = max(horizon, end)

    return rounds

class Subject:

    name = "?"

    def __init__(self, args, mode: str, workdir: Path):

        self.args = args

        self.mode = mode

        self.workdir = workdir

        self.proxy: Optional[HttpCountingProxy] = None

        self.token = oak_token() if self.name == "oak" else None

    # environment used for timed commands

    def env(self) -> dict:

        raise NotImplementedError

    def start(self):

        if self.mode == "proxied":

            self.proxy = HttpCountingProxy(self.upstream).start()

    def stop(self):

        if self.proxy:

            self.proxy.stop()

    def run_step(self, op: str, cmds: list, cwd: Path) -> StepResult:

        if self.proxy:

            self.proxy.reset()

        trace_files = []

        per, rcs, outb, tail, last_stdout = [], [], 0, "", ""

        gh_requests = 0

        t_all = time.monotonic()

        for cmd in cmds:

            env = self.env()

            if self.name == "git" and self.mode == "proxied":

                tf = tempfile.NamedTemporaryFile(prefix="gtrace-", delete=False, dir=self.workdir)

                tf.close()

                env["GIT_TRACE_CURL"] = tf.name

                env["GIT_TRACE_CURL_NO_DATA"] = "1"

                trace_files.append(tf.name)

            is_gh = Path(str(cmd[0])).name == "gh"

            if is_gh and self.proxy:

                # gh speaks HTTPS to api.github.com: count its requests from

                # GH_DEBUG=api (stderr; output_bytes are inflated in this mode,

                # use direct-mode rows for gh output bytes).

                env["HTTPS_PROXY"] = env["https_proxy"] = self.proxy.url

                env["GH_DEBUG"] = "api"

            t = time.monotonic()

            r = subprocess.run(cmd, cwd=cwd, env=env, capture_output=True)

            per.append(time.monotonic() - t)

            if is_gh and self.proxy:

                gh_requests += sum(1 for ln in r.stderr.decode(errors="replace").splitlines() if ln.startswith("* Request to "))

            rcs.append(r.returncode)

            outb += len(r.stdout) + len(r.stderr)

            tail = (r.stdout[-400:] + b"|" + r.stderr[-600:]).decode(errors="replace")

            last_stdout = r.stdout.decode(errors="replace")

            if self.token:

                tail = tail.replace(self.token, "<redacted>")

                last_stdout = last_stdout.replace(self.token, "<redacted>")

            if r.returncode != 0:

                break

        wall = time.monotonic() - t_all

        res = StepResult(op, [" ".join(str(c) for c in cmd) for cmd in cmds], wall, per, rcs, outb, stdout_tail=tail, last_stdout=last_stdout)

        if self.proxy:

            time.sleep(0.05)

            s = self.proxy.snapshot()

            res.tunnel_connections = len(s.tunnels)

            res.tunnel_bytes = s.tunnel_bytes

            res.tunnel_hosts = s.tunnel_hosts()

            if self.name == "oak":

                res.requests = s.requests

                res.request_bytes_up = s.request_bytes

                res.request_bytes_down = s.response_bytes

                res.api_connections = s.connections - len(s.tunnels)

                res.serial_rounds = serial_rounds(s.records)

                res.routes = s.by_route()

                from oakbench.httpcount import route_shape

                recs = sorted(s.records, key=lambda r: r.start_ms)

                res.timeline = [[f"{r.method} {route_shape(r.path)}", r.status, round(r.start_ms, 1),

                                 None if r.ttfb_ms is None else round(r.ttfb_ms, 1),

                                 None if r.end_ms is None else round(r.end_ms - r.start_ms, 1), r.response_bytes]

                                for r in (recs[:40] + (recs[-10:] if len(recs) > 50 else recs[40:]))]

            else:

                n, routes = 0, {}

                for tfn in trace_files:

                    k, rt = parse_git_trace(tfn)

                    n += k

                    for key, v in rt.items():

                        routes[key] = routes.get(key, 0) + v

                    os.unlink(tfn)

                res.requests = n + gh_requests

                if gh_requests:

                    routes["gh api (GH_DEBUG)"] = gh_requests

                res.routes = routes

                res.serial_rounds = n  # git smart-HTTP requests are issued sequentially

        return res

def parse_git_trace(path: str):

    """Count HTTP requests in a GIT_TRACE_CURL file.

    libcurl prints one ``=> Send header: METHOD /path HTTP/x`` request line per

    request for both HTTP/1.1 and HTTP/2; HTTP/2 additionally prints

    ``[HTTP/2] [n] [:method: ...]`` info lines for the same request, which must

    not be counted again (2026-09-27 rows were double counted this way).

    """

    n, routes = 0, {}

    with open(path, errors="replace") as fh:

        for line in fh:

            m = re.search(r"=> Send header: (GET|POST|PUT|HEAD|PATCH|DELETE) (\S+)", line)

            if m:

                n += 1

                key = f"{m.group(1)} {re.sub(r'^/[^/]+/[^/]+', '', m.group(2).split('?')[0])}"

                routes[key] = routes.get(key, 0) + 1

    return n, routes

class OakSubject(Subject):

    name = "oak"

    upstream = OAK_HOST

    def env(self) -> dict:

        e = dict(os.environ, OAK_NO_UPDATE_CHECK="1", NO_COLOR="1")

        e.pop("OAK_REMOTE", None)

        if self.proxy:

            e["OAK_API_KEY"] = self.token

            e["OAK_REMOTE"] = self.proxy.url

            e["HTTPS_PROXY"] = e["https_proxy"] = self.proxy.url

            e["NO_PROXY"] = e["no_proxy"] = "127.0.0.1,localhost"

        return e

    def remote_arg(self) -> list:

        return ["-r", self.proxy.url] if self.proxy else []

class GitSubject(Subject):

    name = "git"

    upstream = GITHUB_HOST

    def env(self) -> dict:

        e = dict(os.environ, GIT_TERMINAL_PROMPT="0", NO_COLOR="1")

        cfg = [("credential.helper", ""), ("credential.helper", f"!{shutil.which('gh')} auth git-credential"),

               ("advice.detachedHead", "false"), ("user.name", "bench-agent"), ("user.email", "[email protected]")]

        if self.proxy:

            cfg.append(("http.proxy", self.proxy.url))

        e["GIT_CONFIG_COUNT"] = str(len(cfg))

        for i, (k, v) in enumerate(cfg):

            e[f"GIT_CONFIG_KEY_{i}"] = k

            e[f"GIT_CONFIG_VALUE_{i}"] = v

        return e

# ---------------------------------------------------------------------------

# Loop

# ---------------------------------------------------------------------------

def loop_trial(subj: Subject, a, trial: int, runid: str, rows: list, actor_ws: Path, fixture_git: Path) -> None:

    ws = subj.workdir / f"{subj.name}-{subj.mode}-t{trial}"

    if ws.exists():

        shutil.rmtree(ws)

    tag = f"{runid}-{subj.mode[0]}{trial}"

    branch = f"alh-{tag}"

    patch = task_patch(tag)

    oak_bin = a.oak_bin

    def rec(res: StepResult, extra: Optional[dict] = None):

        row = {

            "lane": "agent_loop_hosted", "scenario": SCENARIO, "scenario_version": SCENARIO_VERSION,

            "subject": subj.name, "mode": subj.mode, "trial": trial, "run_id": runid,

            "operation": res.op, "variant": (extra or {}).get("variant"),

            "remote_transport": "network", "remote_server": OAK_HOST if subj.name == "oak" else GITHUB_HOST,

            "remote_repo": a.oak_repo if subj.name == "oak" else a.git_url,

            "wall_s": res.wall_s, "elapsed_ms": res.wall_s * 1000.0,

            "returncode": next((rc for rc in res.returncodes if rc != 0), 0), "per_command_s": res.per_command_s, "tool_calls": len(res.commands),

            "commands": res.commands, "returncodes": res.returncodes, "ok": all(rc == 0 for rc in res.returncodes),

            "output_bytes": res.output_bytes, "http_requests": res.requests,

            "http_request_bytes": res.request_bytes_up, "http_response_bytes": res.request_bytes_down,

            "api_connections": res.api_connections, "tunnel_connections": res.tunnel_connections,

            "tunnel_bytes": res.tunnel_bytes, "tunnel_hosts": res.tunnel_hosts, "serial_rounds": res.serial_rounds,

            "routes": res.routes, "timeline": res.timeline, "oracle_ok": res.oracle_ok, "oracle_detail": res.oracle_detail,

            "measurement_source": "httpcount_proxy+git_trace_curl" if subj.mode == "proxied" else "wall_clock_only",

            "tail": res.stdout_tail if not all(rc == 0 for rc in res.returncodes) else "",

        }

        if extra:

            row.update(extra)

        rows.append(row)

        flag = "ok" if row["ok"] else "FAIL"

        print(f"  {subj.name:3} {subj.mode:7} t{trial} {res.op:30} {res.wall_s:7.3f}s req={res.requests} "

              f"up={res.request_bytes_up} down={res.request_bytes_down} tun={res.tunnel_bytes} oracle={res.oracle_ok} {flag}",

              file=sys.stderr, flush=True)

        if not row["ok"]:

            print("    tail:", row["tail"][-500:], file=sys.stderr)

        return row["ok"]

    is_oak = subj.name == "oak"

    ws.parent.mkdir(parents=True, exist_ok=True)

    # 1. acquire (cold for this repo on this machine: fresh directory, no shared object store)

    if is_oak:

        cmd = [[oak_bin, "clone", "--shallow", a.oak_repo, str(ws)] + subj.remote_arg()]

    else:

        cmd = [["git", "clone", "--filter=blob:none", "--single-branch", "--branch", "main", a.git_url, str(ws)]]

    res = subj.run_step("acquire.cold", cmd, subj.workdir)

    if all(rc == 0 for rc in res.returncodes):

        res.oracle_ok, res.oracle_detail = acquire_oracle(ws, fixture_git)

    if not rec(res, {"variant": "oak_clone_shallow" if is_oak else "git_clone_blobless"}):

        return

    # 2. branch

    cmd = [[oak_bin, "switch", "-c", branch]] if is_oak else [["git", "switch", "-c", branch]]

    res = subj.run_step("branch.create", cmd, ws)

    if not rec(res):

        return

    # 4. edit + commit (patch application is untimed; commit is timed)

    apply_patch(ws, patch)

    cmd = [[oak_bin, "commit", "--no-verify", "--quiet"]] if is_oak else [["git", "add", "-A"], ["git", "commit", "-q", "-m", f"agent task {tag}"]]

    res = subj.run_step("edit.commit", cmd, ws)

    if not rec(res):

        return

    # 3. inspect

    steps = {

        "inspect.status": ([[oak_bin, "status", "--json"]], [["git", "status", "--porcelain=v2", "--branch"]]),

        "inspect.diff_main": ([[oak_bin, "diff", "--branch", "--stat"]], [["git", "diff", "--stat", "origin/main...HEAD"]]),

        "inspect.log": ([[oak_bin, "log", "-n", "20", "--oneline"]], [["git", "log", "-n", "20", "--oneline"]]),

    }

    for op, (oc, gc) in steps.items():

        res = subj.run_step(op, oc if is_oak else gc, ws)

        rec(res)

    # 5. publish

    if is_oak:

        cmd = [[oak_bin, "push", "--json"]]

    else:

        cmd = [["git", "push", "-q", "-u", "origin", branch]]

    res = subj.run_step("publish.branch", cmd, ws)

    if all(rc == 0 for rc in res.returncodes):

        if is_oak:

            local = json.loads(subprocess.run([oak_bin, "status", "--json"], cwd=ws, capture_output=True, env=dict(os.environ, OAK_NO_UPDATE_CHECK="1")).stdout)["head"]

            remote = oak_remote_heads(a.oak_repo, subj.token).get(branch)

        else:

            local = subprocess.run(["git", "rev-parse", "HEAD"], cwd=ws, capture_output=True, text=True).stdout.strip()

            remote = git_remote_heads(a.git_url, subj.env()).get(branch)

        res.oracle_ok = bool(local) and local == remote

        res.oracle_detail = f"local={local} remote={remote}"

    if not rec(res):

        return

    # review another agent's branch (pre-created fixture branch)

    if is_oak:

        cmd = [[oak_bin, "branch", "review", a.review_branch, "--remote", "--json"]]

    else:

        cmd = [["git", "fetch", "-q", "origin", a.review_branch], ["git", "diff", "--stat", "origin/main...FETCH_HEAD"]]

    res = subj.run_step("inspect.review_remote", cmd, ws)

    if all(rc == 0 for rc in res.returncodes):

        # the reviewed branch changes exactly review/r1..r3.txt

        seen = [f"review/r{i}.txt" in res.last_stdout for i in (1, 2, 3)]

        res.oracle_ok = all(seen)

        res.oracle_detail = f"review_files_listed={seen}"

    rec(res)

    # 6. main advances by k (actor, untimed) -> main_since_branch (k=1) -> sync

    for k in (1, 20):

        actor_land(subj, a, actor_ws, f"{tag}-k{k}", k)

        if k == 1:

            if is_oak:

                cmd = [[oak_bin, "log", "--remote", "--branch", "main", "-n", "20", "--json"]]

            else:

                cmd = [["git", "fetch", "-q", "origin", "main"], ["git", "log", "--oneline", "HEAD..origin/main"]]

            res = subj.run_step("inspect.main_since_branch.k1", cmd, ws)

            rec(res)

        cmd = [[oak_bin, "pull"]] if is_oak else [["git", "pull", "-q", "--rebase", "origin", "main"]]

        res = subj.run_step(f"sync.k{k}", cmd, ws)

        if all(rc == 0 for rc in res.returncodes):

            actor_file = ws / f"bench-history/actor-{tag}-k{k}.txt"

            ok_actor = actor_file.exists() and actor_file.read_text().count("\n") == k

            ok_patch = (ws / "docs/agent-task-notes.md").exists()

            res.oracle_ok = ok_actor and ok_patch

            res.oracle_detail = f"actor_lines_ok={ok_actor} patch_present={ok_patch}"

        if not rec(res, {"k": k}):

            return

    # 7. merge to main

    if is_oak:

        desc = ws.parent / f"desc-{tag}.txt"

        desc.write_text(f"agent task {tag}\n")

        subprocess.run([oak_bin, "desc", "--file", str(desc)], cwd=ws, env=subj.env(), capture_output=True)

        cmd = [[oak_bin, "merge", "--json"]]

    else:

        if a.git_merge == "pr_squash":

            gh = shutil.which("gh")

            cmd = [["git", "push", "-q", "-f", "origin", branch],

                   [gh, "pr", "create", "--repo", a.git_slug, "--head", branch, "--base", "main", "--title", f"agent task {tag}", "--body", "bench"],

                   [gh, "pr", "merge", branch, "--repo", a.git_slug, "--squash", "--delete-branch"]]

        else:

            cmd = [["git", "push", "-q", "origin", "HEAD:main"]]

    res = subj.run_step("merge.main", cmd, ws)

    if all(rc == 0 for rc in res.returncodes):

        if is_oak:

            heads = oak_remote_heads(a.oak_repo, subj.token)

            try:

                mj = json.loads(res.last_stdout)

            except ValueError:

                mj = {}

            res.oracle_ok = heads.get("main") is not None and (not mj.get("commit_hash") or heads.get("main") == mj.get("commit_hash"))

            res.oracle_detail = f"remote_main={heads.get('main')} merge_json_commit={mj.get('commit_hash')}"

        else:

            if a.git_merge == "pr_squash":

                genv = subj.env()

                subprocess.run(["git", "fetch", "-q", a.git_url, "main"], cwd=ws, env=genv, capture_output=True)

                tree_remote = subprocess.run(["git", "rev-parse", "FETCH_HEAD^{tree}"], cwd=ws, capture_output=True, text=True).stdout.strip()

                tree_local = subprocess.run(["git", "rev-parse", "HEAD^{tree}"], cwd=ws, capture_output=True, text=True).stdout.strip()

                res.oracle_ok = bool(tree_local) and tree_local == tree_remote

                res.oracle_detail = f"squash_tree_equals_branch_tree={res.oracle_ok}"

            else:

                local = subprocess.run(["git", "rev-parse", "HEAD"], cwd=ws, capture_output=True, text=True).stdout.strip()

                remote = git_remote_heads(a.git_url, subj.env()).get("main")

                res.oracle_ok = local == remote

                res.oracle_detail = f"local={local} remote_main={remote}"

    if not rec(res, {"variant": "oak_merge_hosted_squash" if is_oak else ("git_gh_pr_squash" if a.git_merge == "pr_squash" else "git_push_ff_to_main")}):

        return

    # 8. next task, warm workspace: fresh main + new branch

    nb = f"{branch}-next"

    if is_oak:

        cmd = [[oak_bin, "switch", "-c", nb]]

    else:

        cmd = [["git", "fetch", "-q", "origin", "main"], ["git", "switch", "-q", "-c", nb, "origin/main"]]

    res = subj.run_step("next_task.warm", cmd, ws)

    if all(rc == 0 for rc in res.returncodes):

        dig = workspace_digest(ws)

        # the merged task patch must be present on the new task's base

        want = git_blob_id((ws / "docs/agent-task-notes.md").read_bytes()) if (ws / "docs/agent-task-notes.md").exists() else None

        k20 = ws / f"bench-history/actor-{tag}-k20.txt"

        res.oracle_ok = want is not None and k20.exists()

        res.oracle_detail = f"files={len(dig)} patch_present={want is not None} actor_k20_present={k20.exists()}"

    rec(res)

ACQUIRE_VARIANTS = {

    # name: (subject, builder(a, subj, dest, warm) -> cmds)

    "oak_clone_shallow": ("oak", lambda a, subj, dest, warm: [[a.oak_bin, "clone", "--shallow", a.oak_repo, str(dest)] + subj.remote_arg()]),

    "oak_clone_full": ("oak", lambda a, subj, dest, warm: [[a.oak_bin, "clone", a.oak_repo, str(dest)] + subj.remote_arg()]),

    "git_clone_depth1": ("git", lambda a, subj, dest, warm: [["git", "clone", "-q", "--depth", "1", a.git_url, str(dest)]]),

    "git_clone_blobless": ("git", lambda a, subj, dest, warm: [["git", "clone", "-q", "--filter=blob:none", "--single-branch", "--branch", "main", a.git_url, str(dest)]]),

    "git_clone_full": ("git", lambda a, subj, dest, warm: [["git", "clone", "-q", a.git_url, str(dest)]]),

    "git_worktree_from_warm": ("git", lambda a, subj, dest, warm: [["git", "-C", str(warm), "fetch", "-q", "origin", "main"], ["git", "-C", str(warm), "worktree", "add", "-q", "-b", dest.name, str(dest), "origin/main"]]),

}

def acquire_trials(a, runid: str, rows: list) -> None:

    """acquire.cold variants, each into a fresh directory; oracle = fixture tree."""

    for trial in range(a.runs):

        for mode in a.modes.split(","):

            for vname, (sname, build) in ACQUIRE_VARIANTS.items():

                if sname not in a.subjects.split(",") or (a.variants and vname not in a.variants.split(",")):

                    continue

                subj = OakSubject(a, mode, a.workdir) if sname == "oak" else GitSubject(a, mode, a.workdir)

                subj.start()

                try:

                    dest = a.workdir / f"acq-{vname}-{mode}-{runid}-{trial}"

                    warm = a.workdir / "git-warm"

                    if vname == "git_worktree_from_warm" and not warm.exists():

                        subprocess.run(["git", "clone", "-q", "--filter=blob:none", a.git_url, str(warm)], env=GitSubject(a, "direct", a.workdir).env(), check=True)

                    res = subj.run_step("acquire.cold", build(a, subj, dest, warm), a.workdir)

                    if all(rc == 0 for rc in res.returncodes):

                        res.oracle_ok, res.oracle_detail = acquire_oracle(dest, a.fixture_git)

                    rows.append(row_for(subj, a, trial, runid, res, {"variant": vname, "warm_source": vname == "git_worktree_from_warm"}))

                    _print_row(rows[-1])

                finally:

                    subj.stop()

                if vname == "git_worktree_from_warm":

                    subprocess.run(["git", "-C", str(a.workdir / "git-warm"), "worktree", "remove", "--force", str(dest)], capture_output=True)

                    subprocess.run(["git", "-C", str(a.workdir / "git-warm"), "branch", "-q", "-D", dest.name], capture_output=True)

                elif dest.exists():

                    shutil.rmtree(dest, ignore_errors=True)

def review_trials(a, runid: str, rows: list) -> None:

    """inspect.review_remote only, from existing warm workspaces (--review-oak-ws /

    --review-git-ws), against a freshly created review branch. Proxied Oak runs

    rebind the proxy to the port stored as the workspace's remote."""

    for trial in range(a.runs):

        for mode in a.modes.split(","):

            for sname in a.subjects.split(","):

                ws = Path(a.review_oak_ws if sname == "oak" else a.review_git_ws)

                subj = OakSubject(a, mode, a.workdir) if sname == "oak" else GitSubject(a, mode, a.workdir)

                if mode == "proxied":

                    port = 0

                    if sname == "oak":

                        info = json.loads(subprocess.run([a.oak_bin, "status", "--json"], cwd=ws, capture_output=True,

                                                         env=dict(os.environ, OAK_NO_UPDATE_CHECK="1")).stdout)

                        m = re.match(r"http://127\.0\.0\.1:(\d+)", info.get("remote_url") or "")

                        if not m:

                            print("skip: oak review workspace is not proxied", file=sys.stderr)

                            continue

                        port = int(m.group(1))

                    subj.proxy = HttpCountingProxy(subj.upstream, listen_port=port).start()

                try:

                    if sname == "oak":

                        cmd = [[a.oak_bin, "branch", "review", a.review_branch, "--remote", "--json"]]

                    else:

                        cmd = [["git", "fetch", "-q", "origin", a.review_branch], ["git", "diff", "--stat", "origin/main...FETCH_HEAD"]]

                    res = subj.run_step("inspect.review_remote", cmd, ws)

                    if all(rc == 0 for rc in res.returncodes):

                        seen = [f"/r{i}.txt" in res.last_stdout for i in (1, 2, 3)]

                        res.oracle_ok = all(seen)

                        res.oracle_detail = f"review_files_listed={seen}"

                    rows.append(row_for(subj, a, trial, runid, res, {"variant": "fresh_review_branch"}))

                    _print_row(rows[-1])

                finally:

                    subj.stop()

def row_for(subj, a, trial, runid, res, extra=None) -> dict:

    row = {

        "lane": "agent_loop_hosted", "scenario": SCENARIO, "scenario_version": SCENARIO_VERSION,

        "subject": subj.name, "mode": subj.mode, "trial": trial, "run_id": runid,

        "operation": res.op, "variant": None,

        "remote_transport": "network", "remote_server": OAK_HOST if subj.name == "oak" else GITHUB_HOST,

        "remote_repo": a.oak_repo if subj.name == "oak" else a.git_url,

        "wall_s": res.wall_s, "elapsed_ms": res.wall_s * 1000.0,

        "returncode": next((rc for rc in res.returncodes if rc != 0), 0), "per_command_s": res.per_command_s, "tool_calls": len(res.commands),

        "commands": res.commands, "returncodes": res.returncodes, "ok": all(rc == 0 for rc in res.returncodes),

        "output_bytes": res.output_bytes, "http_requests": res.requests,

        "http_request_bytes": res.request_bytes_up, "http_response_bytes": res.request_bytes_down,

        "api_connections": res.api_connections, "tunnel_connections": res.tunnel_connections,

        "tunnel_bytes": res.tunnel_bytes, "tunnel_hosts": res.tunnel_hosts, "serial_rounds": res.serial_rounds,

        "routes": res.routes, "timeline": res.timeline, "oracle_ok": res.oracle_ok, "oracle_detail": res.oracle_detail,

        "measurement_source": "httpcount_proxy+git_trace_curl" if subj.mode == "proxied" else "wall_clock_only",

        "tail": res.stdout_tail if not all(rc == 0 for rc in res.returncodes) else "",

    }

    if extra:

        row.update(extra)

    return row

def _print_row(row: dict) -> None:

    print(f"  {row['subject']:3} {row['mode']:7} t{row['trial']} {row['operation']:30} {str(row.get('variant') or ''):24} {row['wall_s']:7.3f}s "

          f"req={row['http_requests']} down={row['http_response_bytes']} tun={row['tunnel_bytes']} oracle={row['oracle_ok']} {'ok' if row['ok'] else 'FAIL'}",

          file=sys.stderr, flush=True)

    if not row["ok"]:

        print("    tail:", row["tail"][-500:], file=sys.stderr)

def actor_land(subj: Subject, a, actor_ws: Path, tag: str, k: int) -> None:

    """Another actor lands k commits on main touching only bench-history/actor-*. Untimed."""

    env = dict(os.environ, OAK_NO_UPDATE_CHECK="1")

    if subj.name == "oak":

        for i in range(k):

            subprocess.run([a.oak_bin, "switch", "-c", f"actor-{tag}-{i}"], cwd=actor_ws, env=env, capture_output=True, check=True)

            actor_edit(actor_ws, tag, i)

            subprocess.run([a.oak_bin, "commit", "--no-verify", "--quiet"], cwd=actor_ws, env=env, capture_output=True, check=True)

            subprocess.run([a.oak_bin, "desc", f"actor {tag} {i}"], cwd=actor_ws, env=env, capture_output=True, check=True)

            r = subprocess.run([a.oak_bin, "merge", "--json"], cwd=actor_ws, env=env, capture_output=True)

            if r.returncode != 0:

                raise SystemExit(f"actor merge failed: {r.stdout[-400:]!r} {r.stderr[-400:]!r}")

    else:

        genv = GitSubject.env(subj) if isinstance(subj, GitSubject) else env

        genv = dict(genv)

        genv.pop("GIT_TRACE_CURL", None)

        # actor uses no proxy

        cfgn = int(genv.get("GIT_CONFIG_COUNT", "0"))

        for i in range(cfgn):

            if genv.get(f"GIT_CONFIG_KEY_{i}") == "http.proxy":

                genv[f"GIT_CONFIG_VALUE_{i}"] = ""

        subprocess.run(["git", "pull", "-q", "--rebase", "origin", "main"], cwd=actor_ws, env=genv, capture_output=True, check=True)

        for i in range(k):

            actor_edit(actor_ws, tag, i)

            subprocess.run(["git", "add", "-A"], cwd=actor_ws, env=genv, check=True)

            subprocess.run(["git", "commit", "-q", "-m", f"actor {tag} {i}"], cwd=actor_ws, env=genv, check=True)

        subprocess.run(["git", "push", "-q", "origin", "HEAD:main"], cwd=actor_ws, env=genv, capture_output=True, check=True)

def summarize(rows: list) -> str:

    groups: dict = {}

    for r in rows:

        groups.setdefault((r["operation"], r["subject"], r.get("mode", "?"), r.get("variant") or ""), []).append(r)

    lines = ["| operation | subject | mode | variant | n | ok | median s | min-max s | req (median) | bytes down (median) | tunnel bytes | serial rounds | tool calls | out bytes |",

             "|---|---|---|---|---|---|---|---|---|---|---|---|---|---|"]

    def med(vals):

        vals = [v for v in vals if v is not None]

        return statistics.median(vals) if vals else None

    for (op, s, m, v), rs in sorted(groups.items(), key=lambda kv: (OPERATIONS.index(kv[0][0]) if kv[0][0] in OPERATIONS else 99, kv[0][1], kv[0][2], kv[0][3])):

        ok = [r for r in rs if r["ok"]]

        w = [r["wall_s"] for r in ok if r.get("wall_s") is not None]

        lines.append(

            f"| {op} | {s} | {m} | {v} | {len(rs)} | {len(ok)} | "

            f"{statistics.median(w):.3f} | {min(w):.3f}-{max(w):.3f} | " if w else f"| {op} | {s} | {m} | {v} | {len(rs)} | 0 | - | - | "

        )

        lines[-1] += f"{med([r['http_requests'] for r in ok])} | {med([r['http_response_bytes'] for r in ok])} | {med([r['tunnel_bytes'] for r in ok])} | {med([r['serial_rounds'] for r in ok])} | {med([r['tool_calls'] for r in ok])} | {med([r['output_bytes'] for r in ok])} |"

    return "\n".join(lines)

def main(argv=None) -> int:

    with measurement_lock("agent_loop_hosted") as lock_info:

        print(f"measurement lock: {'held' if lock_info.enabled else 'DISABLED'} (waited {lock_info.wait_ms or 0:.0f} ms)", file=sys.stderr)

        return _main(argv)

def _main(argv=None) -> int:

    ap = argparse.ArgumentParser(description=__doc__.split("\n\n")[0])

    ap.add_argument("--subjects", default="oak,git")

    ap.add_argument("--modes", default="proxied,direct")

    ap.add_argument("--runs", type=int, default=3)

    ap.add_argument("--oak-bin", required=True)

    ap.add_argument("--oak-repo", required=True, help="ORG/REPO on oak.space (write access required)")

    ap.add_argument("--git-url", required=True, help="https URL of the GitHub fixture repo")

    ap.add_argument("--fixture-git", required=True, type=Path, help="local Git fixture (expected tree oracle)")

    ap.add_argument("--review-branch", required=True)

    ap.add_argument("--workdir", required=True, type=Path)

    ap.add_argument("--results", required=True, type=Path)

    ap.add_argument("--part", choices=("loop", "acquire", "review"), default="loop")

    ap.add_argument("--review-oak-ws", help="--part review: existing Oak workspace")

    ap.add_argument("--review-git-ws", help="--part review: existing Git workspace")

    ap.add_argument("--variants", default="", help="--part acquire: comma list of variant names (default all)")

    ap.add_argument("--git-merge", choices=("push_ff", "pr_squash"), default="push_ff",

                    help="Git merge.main path: fast-forward push to main, or gh pr create + gh pr merge --squash")

    a = ap.parse_args(argv)

    a.oak_bin = str(Path(a.oak_bin).resolve())

    a.git_slug = re.sub(r"^https://github\.com/|\.git$", "", a.git_url)

    runid = time.strftime("%H%M%S", time.gmtime())

    a.workdir.mkdir(parents=True, exist_ok=True)

    a.results.mkdir(parents=True, exist_ok=True)

    out = a.results / f"agent_loop_hosted-{runid}.jsonl"

    rows: list = []

    env_info = {"load_avg": os.getloadavg(), "host": platform.node(), "oak_bin_sha256": hashlib.sha256(Path(a.oak_bin).read_bytes()).hexdigest(),

                "git_version": subprocess.run(["git", "--version"], capture_output=True, text=True).stdout.strip()}

    print("env:", json.dumps(env_info), file=sys.stderr)

    if a.part in ("acquire", "review"):

        (acquire_trials if a.part == "acquire" else review_trials)(a, runid, rows)

        with out.open("w") as fh:

            for r in rows:

                r.setdefault("env", env_info)

                fh.write(json.dumps(r) + "\n")

        print(summarize(rows))

        return 0

    # actor workspaces (warm, persistent within this run)

    actors = {}

    for s in a.subjects.split(","):

        aw = a.workdir / f"actor-{s}"

        if not aw.exists():

            if s == "oak":

                subprocess.run([a.oak_bin, "clone", "--shallow", a.oak_repo, str(aw)], env=dict(os.environ, OAK_NO_UPDATE_CHECK="1"), check=True, capture_output=True)

            else:

                g = GitSubject(a, "direct", a.workdir)

                subprocess.run(["git", "clone", "-q", a.git_url, str(aw)], env=g.env(), check=True)

        actors[s] = aw

    for trial in range(a.runs):

        for mode in a.modes.split(","):

            order = a.subjects.split(",")

            if trial % 2:

                order = list(reversed(order))

            for s in order:

                subj = OakSubject(a, mode, a.workdir) if s == "oak" else GitSubject(a, mode, a.workdir)

                subj.start()

                try:

                    loop_trial(subj, a, trial, runid, rows, actors[s], a.fixture_git)

                except SystemExit as e:

                    print("trial aborted:", e, file=sys.stderr)

                    rows.append({"lane": "agent_loop_hosted", "scenario": SCENARIO, "subject": s, "mode": mode, "trial": trial, "operation": "trial.aborted",

                                 "ok": False, "elapsed_ms": 0.0, "returncode": 1, "measurement_source": "wall_clock_only", "tail": str(e)[-500:]})

                finally:

                    subj.stop()

                with out.open("w") as fh:

                    for r in rows:

                        r.setdefault("env", env_info)

                        fh.write(json.dumps(r) + "\n")

    summary = summarize(rows)

    (a.results / f"agent_loop_hosted-{runid}.summary.md").write_text(summary + "\n")

    print(summary)

    print(f"rows: {out}", file=sys.stderr)

    return 0

if __name__ == "__main__":

    raise SystemExit(main())