#!/usr/bin/env python3

"""stale_branch_land_v1: land N agent branches that are k commits behind main.

Baseline instrument for "landing a stale branch" before any server-side land

primitive exists. Each trial creates a fresh repository whose main has moved k

commits past the point where N ordinary agent branches (disjoint files) plus

one seeded semantic-conflict pair (green alone, red combined) were forked, then

releases every agent at once and measures how each arm lands them.

Arms (``contention_mode`` on every row):

- ``oak_safe``   documented safe flow against a local oakspace server: shallow

                 worker -> ``oak pull`` converge (fresh shallow worker + reapply

                 on failure) -> local check on the combined tree -> push ->

                 exact-commit CI on the branch tip -> main-unchanged recheck ->

                 hosted merge -> verify.

- ``oak_naive``  ``oak merge`` on the stale branch as-is (server 3-way; the CI

                 gate only knows the branch tip).

- ``git_competent`` bare origin + clone per agent over file transport; fetch ->

                 rebase -> CI on the combined tree -> push; non-fast-forward

                 rejection retries the loop.

- ``git_fifo``   the same flow serialized through one FIFO lock (a merge queue

                 of depth 1); no push races.

CI is a real check script (``python3 ci/check.py``) plus a fixed, labeled

``--ci-added-seconds`` delay. Oak arms run it through oakspace's real dispatch

+ merge gate with ``oakbench.ci_stub`` as the loopback execution backend; Git

arms run the same script as a local subprocess with the same delay.

Independent oracles (never the agent's own report):

- every main head created during the trial is materialized (Oak: the server's

  token-scoped git export of that exact commit; Git: ``git archive`` from the

  bare origin) and checked -> ``red_main_events``;

- the final main tree must equal base + k mover files + all N agent patches +

  exactly one member of the pair -> ``intended_changes_exact``;

- ``combined_tree_tested_before_land``: the landed main commit's tree digest

  was among the tree digests this agent checked (locally or in CI) before its

  landing call.

Rows use the ``contention`` lane contract. Null means unmeasured (ADR-0002).

Scenario/operation names are immutable identity (ADR-0005).

"""

from __future__ import annotations

import argparse

import hashlib

import hmac

import json

import os

import platform

import random

import re

import shutil

import statistics

import string

import subprocess

import sys

import tempfile

import threading

import time

from dataclasses import dataclass, field

from pathlib import Path

from typing import Any, Callable, Optional

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

from oakbench.ci_stub import CiStub, read_ledger, tree_digest  # noqa: E402

from oakbench.results import ResultsStore  # noqa: E402

from oakbench.runlock import measurement_lock  # noqa: E402

from oakbench.subjects import binary_sha256  # noqa: E402

SCENARIO = "stale_branch_land_v1"

SCENARIO_VERSION = 1

FIXTURE_VERSION = "stale_land_fixture_v1"

PROFILE = "stale-land"

ARMS = ("oak_safe", "oak_naive", "git_competent", "git_fifo")

OAK_ARMS = ("oak_safe", "oak_naive")

MAX_ATTEMPTS = 8

CI_WAIT_TIMEOUT_S = 300

PAIR = ("pair_rename", "pair_consumer")

CI_TRIGGER_NOTE = (

    "explicit exact-commit `oak ci trigger` after each push: the local server's publication-"

    "triggered path reads workflows with local_chunks=None (oakspace ci.rs:2687), so push/"

    "branch.created runs conclude 'no readable blob' on a local-chunk server; hosted (R2) "

    "auto-triggers. tool_call_count_hosted_equivalent subtracts these trigger calls."

)

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

# Fixture (pure; unit tested)

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

CHECK_PY = '''import importlib

import pathlib

import sys

sys.dont_write_bytecode = True

ROOT = pathlib.Path(__file__).resolve().parent.parent

sys.path.insert(0, str(ROOT))

failures = 0

for test in sorted((ROOT / "tests").glob("test_*.py")):

    module = importlib.import_module("tests." + test.stem)

    for name in sorted(dir(module)):

        if name.startswith("test_"):

            try:

                getattr(module, name)()

            except Exception as exc:  # noqa: BLE001

                failures += 1

                print("FAIL %s::%s: %r" % (test.name, name, exc))

print("ok" if not failures else "%d failed" % failures)

sys.exit(1 if failures else 0)

'''

WORKFLOW_YML = """on: [push, branch.created, merge, manual]

jobs:

  check:

    steps:

      - name: check

        run: PYTHONDONTWRITEBYTECODE=1 python3 ci/check.py

"""

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

    return {

        ".oak/workflows/ci.yml": WORKFLOW_YML,

        "ci/check.py": CHECK_PY,

        "pkg/__init__.py": "",

        "pkg/config.py": "LIMIT = 10\n",

        "pkg/core.py": "from pkg import config\n\n\ndef scale(x):\n    return x * config.LIMIT\n",

        "tests/__init__.py": "",

        "tests/test_core.py": "from pkg.core import scale\n\n\ndef test_scale():\n    assert scale(2) == 20\n",

        "README.md": f"fixture {FIXTURE_VERSION}\n",

    }

def agent_names(n: int) -> list[str]:

    return [f"agent{i:02d}" for i in range(n)] + list(PAIR)

def agent_patch(name: str) -> dict[str, str]:

    """Files each agent writes. Ordinary agents and the pair touch disjoint paths."""

    if name == "pair_rename":

        return {

            "pkg/config.py": "MAX_LIMIT = 10\n",

            "pkg/core.py": "from pkg import config\n\n\ndef scale(x):\n    return x * config.MAX_LIMIT\n",

        }

    if name == "pair_consumer":

        return {

            "pkg/feature_b.py": "from pkg import config\n\n\ndef limit():\n    return config.LIMIT\n",

            "tests/test_feature_b.py": "from pkg.feature_b import limit\n\n\ndef test_limit():\n    assert limit() == 10\n",

        }

    return {

        f"pkg/{name}.py": f"def value():\n    return {name!r}\n",

        f"tests/test_{name}.py": f"from pkg.{name} import value\n\n\ndef test_value():\n    assert value() == {name!r}\n",

    }

def mover_file(j: int) -> tuple[str, str]:

    return f"hist/h{j:03d}.txt", f"main moved {j}\n"

def expected_final_files(n: int, k: int, pair_winner: str) -> dict[str, str]:

    files = dict(base_files())

    for j in range(k):

        path, content = mover_file(j)

        files[path] = content

    for name in agent_names(n)[:n]:

        files.update(agent_patch(name))

    files.update(agent_patch(pair_winner))

    return files

def write_files(root: Path, files: dict[str, str]) -> None:

    for rel, content in files.items():

        path = root / rel

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

        path.write_text(content)

def files_digest(files: dict[str, str]) -> str:

    """tree_digest() of a tree that holds exactly ``files``."""

    with tempfile.TemporaryDirectory(prefix="sl-digest-") as tmp:

        write_files(Path(tmp), files)

        return tree_digest(Path(tmp))

def read_tree(root: Path) -> dict[str, str]:

    out: dict[str, str] = {}

    for path in sorted(root.rglob("*")):

        rel = path.relative_to(root).as_posix()

        if not path.is_file() or rel.startswith(".git/") or "__pycache__" in rel:

            continue

        if rel.startswith(".oak/") and not rel.startswith(".oak/workflows/"):

            continue

        out[rel] = path.read_text(errors="replace")

    return out

def run_check(root: Path) -> tuple[bool, str]:

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

    proc = subprocess.run(

        [sys.executable, "ci/check.py"], cwd=root, env=env, capture_output=True, text=True, timeout=120

    )

    return proc.returncode == 0, (proc.stdout + proc.stderr)[-400:]

def classify_final(files: dict[str, str], n: int, k: int) -> dict[str, Any]:

    """Final-state oracle: exactly the intended changes, one pair member."""

    rename_in = files.get("pkg/config.py") == agent_patch("pair_rename")["pkg/config.py"]

    consumer_in = "pkg/feature_b.py" in files

    winners = [name for name, present in (("pair_rename", rename_in), ("pair_consumer", consumer_in)) if present]

    missing_agents = [

        name for name in agent_names(n)[:n] if any(files.get(p) != c for p, c in agent_patch(name).items())

    ]

    exact = False

    if len(winners) == 1:

        exact = files == expected_final_files(n, k, winners[0])

    return {

        "pair_landed": winners,

        "both_pair_landed": len(winners) == 2,

        "missing_agents": missing_agents,

        "intended_changes_exact": exact,

    }

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

# Agent tool-call accounting

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

@dataclass

class AgentRun:

    name: str

    arm: str

    tool_calls: int = 0

    ci_trigger_calls: int = 0

    output_bytes: int = 0

    attempts: int = 0

    ci_runs_local: int = 0

    outcome: str = "pending"

    error: Optional[str] = None

    landed_commit: Optional[str] = None

    landed_at: Optional[float] = None

    finished_at: Optional[float] = None

    tested_digests: list[str] = field(default_factory=list)

    local_checks: int = 0

    converge_path: list[str] = field(default_factory=list)

    trace: list[dict[str, Any]] = field(default_factory=list)

    branch: Optional[str] = None

    abandoned_branches: list[str] = field(default_factory=list)

    rate_limited: int = 0

    conflict_retries: int = 0

    def call(self, argv: list[str], cwd: Path, env: dict[str, str], timeout: float = 600) -> subprocess.CompletedProcess[str]:

        start = time.perf_counter()

        try:

            proc = subprocess.run(argv, cwd=cwd, env=env, capture_output=True, text=True, timeout=timeout)

        except subprocess.TimeoutExpired as exc:

            proc = subprocess.CompletedProcess(argv, 124, exc.stdout or "", (exc.stderr or "") + "\n[timeout]")

        self.tool_calls += 1

        out = len((proc.stdout or "").encode()) + len((proc.stderr or "").encode())

        self.output_bytes += out

        self.trace.append(

            {

                "argv": [Path(argv[0]).name] + argv[1:],

                "rc": proc.returncode,

                "ms": round((time.perf_counter() - start) * 1000, 1),

                "bytes": out,

                "tail": ((proc.stdout or "") + (proc.stderr or ""))[-300:],

            }

        )

        return proc

    def call_retrying(

        self, argv: list[str], cwd: Path, env: dict[str, str], timeout: float = 600, retry_conflicts: int = 0

    ) -> subprocess.CompletedProcess[str]:

        """A competent agent honours HTTP 429 retry-after and, where the caller

        allows it, explicitly retryable 409s. Every retry is a tool call."""

        conflicts = 0

        for _ in range(30):

            proc = self.call(argv, cwd, env, timeout)

            wait = retry_after(proc)

            if wait is not None:

                self.rate_limited += 1

                time.sleep(wait)

                continue

            if proc.returncode != 0 and conflicts < retry_conflicts and retryable_conflict(proc):

                conflicts += 1

                self.conflict_retries += 1

                time.sleep(0.5)

                continue

            return proc

        return proc

    def local_check(self, root: Path, added_seconds: float = 0.0) -> bool:

        """One tool call: run the check script on the working tree."""

        start = time.perf_counter()

        self.tested_digests.append(tree_digest(root))

        ok, tail = run_check(root)

        if added_seconds > 0:

            time.sleep(added_seconds)

        self.tool_calls += 1

        self.local_checks += 1

        self.output_bytes += len(tail.encode())

        self.trace.append({"argv": ["check"], "rc": 0 if ok else 1, "ms": round((time.perf_counter() - start) * 1000, 1), "bytes": len(tail.encode()), "tail": tail})

        return ok

    def apply_patch(self, root: Path, files: dict[str, str]) -> None:

        """One tool call: the agent re-applies its intended edit (fallback path)."""

        write_files(root, files)

        self.tool_calls += 1

        self.trace.append({"argv": ["apply_patch"], "rc": 0, "ms": 0.0, "bytes": 0, "tail": ""})

def last_json(text: str) -> Optional[dict[str, Any]]:

    """Last JSON object in stdout (some clients print human lines first, fb-479)."""

    for line in reversed((text or "").strip().splitlines()):

        line = line.strip()

        if line.startswith("{"):

            try:

                value = json.loads(line)

            except ValueError:

                continue

            if isinstance(value, dict):

                return value

    try:

        value = json.loads(text)

        return value if isinstance(value, dict) else None

    except ValueError:

        return None

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

# Oak arms (local oakspace server)

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

@dataclass

class OakServer:

    url: str

    token: str

    secret: str

    owner: str

    psql: list[str]

    home: Path

    def env(self) -> dict[str, str]:

        env = dict(os.environ)

        env.update(

            {

                "HOME": str(self.home),

                "OAK_API_KEY": self.token,

                "OAK_URL": self.url,

                "OAK_REMOTE": self.url,

                "OAK_AUTHOR": self.owner,

                "OAK_TRUSTED_REMOTES": self.url,

                "PYTHONDONTWRITEBYTECODE": "1",

                "NO_COLOR": "1",

            }

        )

        return env

    def sql(self, query: str) -> list[list[str]]:

        proc = subprocess.run(self.psql + ["-qAt", "-F", "\t", "-c", query], capture_output=True, text=True, timeout=60)

        if proc.returncode != 0:

            raise RuntimeError(f"psql failed: {proc.stderr.strip()[:300]}")

        return [line.split("\t") for line in proc.stdout.splitlines() if line.strip()]

    def repo_id(self, repo: str) -> int:

        rows = self.sql(f"select id from repos where name='{repo}'")

        return int(rows[0][0])

    def main_heads(self, repo_id: int) -> list[tuple[int, str]]:

        rows = self.sql(

            f"select transition_generation, head_hash from publication_events where repo_id={repo_id} "

            "and branch_name='main' order by transition_generation"

        )

        return [(int(gen), head) for gen, head in rows]

    def fetch_token(self, repo_id: int, commit: str) -> str:

        payload = f"{repo_id}.{commit}.{int(time.time()) + 900}"

        return payload + "." + hmac.new(self.secret.encode(), payload.encode(), hashlib.sha256).hexdigest()

    def materialize(self, repo_id: int, commit: str, dest: Path) -> Path:

        env = dict(os.environ)

        env.update(

            {

                "GIT_TERMINAL_PROMPT": "0",

                "GIT_CONFIG_COUNT": "1",

                "GIT_CONFIG_KEY_0": "http.extraHeader",

                "GIT_CONFIG_VALUE_0": f"x-oak-ci-fetch-token: {self.fetch_token(repo_id, commit)}",

            }

        )

        shutil.rmtree(dest, ignore_errors=True)

        proc = subprocess.run(

            ["git", "clone", "-q", "--depth", "1", "--no-tags", f"{self.url}/api/ci/fetch/{repo_id}", str(dest)],

            env=env, capture_output=True, text=True, timeout=120,

        )

        if proc.returncode != 0:

            raise RuntimeError(f"materialize {commit[:12]} failed: {proc.stderr.strip()[:300]}")

        return dest

RETRY_AFTER_RE = re.compile(r"429 Too Many Requests.*?retry after (\d+) second")

def retry_after(proc: subprocess.CompletedProcess[str]) -> Optional[float]:

    """Seconds the server asked us to wait (HTTP 429), else None."""

    match = RETRY_AFTER_RE.search((proc.stderr or "") + (proc.stdout or ""))

    return float(match.group(1)) if match else None

def retryable_conflict(proc: subprocess.CompletedProcess[str]) -> bool:

    """A conclusive server refusal that explicitly says it is safe to retry.

    Oak prints e.g. ``HTTP 409 Conflict: a verified blob mapping changed before

    branch publication; retry`` (concurrent pushes) or a ``main_moved`` /

    ``branch_moved`` code with ``retryable: true`` (concurrent merges). Nothing

    was written, so retrying is not replaying an ambiguous mutation.

    """

    text = (proc.stderr or "") + (proc.stdout or "")

    return proc.returncode != 0 and "409" in text and ("; retry" in text or '"retryable":true' in text or "retryable: true" in text)

def oak_run(oak: Path, args: list[str], cwd: Path, env: dict[str, str], timeout: float = 300) -> subprocess.CompletedProcess[str]:

    """Untimed setup call; honours the server's 429 retry-after hint."""

    for _ in range(30):

        proc = subprocess.run([str(oak)] + args, cwd=cwd, env=env, capture_output=True, text=True, timeout=timeout)

        wait = retry_after(proc)

        if wait is None and retryable_conflict(proc):

            wait = 0.5

        if proc.returncode == 0 or wait is None:

            return proc

        time.sleep(wait)

    return proc

def must(proc: subprocess.CompletedProcess[str], what: str) -> subprocess.CompletedProcess[str]:

    if proc.returncode != 0:

        raise RuntimeError(f"setup step failed ({what}): rc={proc.returncode} {(proc.stderr or proc.stdout).strip()[-300:]}")

    return proc

def oak_head(oak: Path, cwd: Path, env: dict[str, str]) -> str:

    return must(oak_run(oak, ["hash"], cwd, env), "hash").stdout.strip()

def oak_branch(oak: Path, cwd: Path, env: dict[str, str]) -> str:

    return json.loads(must(oak_run(oak, ["status", "--json"], cwd, env), "status").stdout)["branch"]

def oak_trigger_and_wait(agent: Optional[AgentRun], oak: Path, cwd: Path, env: dict[str, str], key: str) -> tuple[bool, Optional[str]]:

    """Exact-commit CI: trigger on the current head and wait for its conclusion."""

    call = agent.call_retrying if agent else (lambda argv, cwd, env, timeout=600: oak_run(Path(argv[0]), argv[1:], cwd, env, timeout))

    head = oak_head(oak, cwd, env) if agent is None else None

    if agent is not None:

        proc = call([str(oak), "hash"], cwd, env)

        head = proc.stdout.strip()

    trig = call([str(oak), "ci", "trigger", "--expected-commit", head, "--idempotency-key", key, "--json"], cwd, env)

    if agent is not None:

        agent.ci_trigger_calls += 1

    doc = last_json(trig.stdout) or {}

    run_ids = [str(r) for r in doc.get("run_ids") or []]

    if trig.returncode != 0 or not run_ids:

        return False, f"ci trigger failed: {(trig.stderr or trig.stdout).strip()[-200:]}"

    wait = call([str(oak), "ci", "wait", *run_ids, "--commit", head, "--timeout", str(CI_WAIT_TIMEOUT_S), "--json"], cwd, env, timeout=CI_WAIT_TIMEOUT_S + 30)

    wdoc = last_json(wait.stdout) or {}

    text = json.dumps(wdoc)

    ok = wait.returncode == 0 and '"failure"' not in text and '"cancelled"' not in text and '"timed_out"' not in text

    return ok, None if ok else f"ci not green: rc={wait.returncode} {text[:200]}"

@dataclass

class OakTrial:

    server: OakServer

    oak: Path

    setup_oak: Path

    root: Path

    repo: str

    n: int

    k: int

    arm: str

    stub: CiStub

    repo_id: int = 0

    workers: dict[str, Path] = field(default_factory=dict)

    branches: dict[str, str] = field(default_factory=dict)

    setup_ledger: Path = Path()

    trial_ledger: Path = Path()

    base_generation: int = 0

    def setup(self) -> dict[str, Any]:

        env = self.server.env()

        t0 = time.perf_counter()

        seed = self.root / "seed"

        seed.mkdir(parents=True)

        must(oak_run(self.setup_oak, ["init"], seed, env), "init")

        write_files(seed, base_files())

        must(oak_run(self.setup_oak, ["commit"], seed, env), "seed commit")

        must(oak_run(self.setup_oak, ["push", "-r", self.server.url, "--repo", f"{self.server.owner}/{self.repo}"], seed, env), "seed push")

        # Setup-only: bypass the gate (the auto-triggered run cannot read the workflow locally).

        must(oak_run(self.setup_oak, ["merge", "--force", "--json"], seed, env), "seed merge")

        self.repo_id = self.server.repo_id(self.repo)

        self.stub.ledger = self.setup_ledger

        errors: list[str] = []

        def prep(name: str) -> None:

            try:

                wdir = self.root / f"w-{name}"

                must(oak_run(self.oak, ["clone", "--shallow", f"{self.server.owner}/{self.repo}", str(wdir)], self.root, env), f"clone {name}")

                write_files(wdir, agent_patch(name))

                must(oak_run(self.oak, ["commit"], wdir, env), f"commit {name}")

                must(oak_run(self.oak, ["push"], wdir, env), f"push {name}")

                ok, err = oak_trigger_and_wait(None, self.oak, wdir, env, f"setup-{name}")

                if not ok:

                    raise RuntimeError(f"setup CI for {name}: {err}")

                self.workers[name] = wdir

                self.branches[name] = oak_branch(self.oak, wdir, env)

            except Exception as exc:  # noqa: BLE001

                errors.append(f"{name}: {exc}")

        threads = [threading.Thread(target=prep, args=(name,)) for name in agent_names(self.n)]

        for thread in threads:

            thread.start()

        for thread in threads:

            thread.join()

        if errors:

            raise RuntimeError("; ".join(errors))

        mover = self.root / "mover"

        must(oak_run(self.setup_oak, ["clone", "--shallow", f"{self.server.owner}/{self.repo}", str(mover)], self.root, env), "mover clone")

        for j in range(self.k):

            path, content = mover_file(j)

            write_files(mover, {path: content})

            must(oak_run(self.setup_oak, ["commit"], mover, env), "mover commit")

            must(oak_run(self.setup_oak, ["push"], mover, env), "mover push")

            must(oak_run(self.setup_oak, ["merge", "--force", "--json"], mover, env), "mover merge")

        heads = self.server.main_heads(self.repo_id)

        self.base_generation = heads[-1][0]

        self.stub.ledger = self.trial_ledger

        return {"setup_ms": (time.perf_counter() - t0) * 1000, "main_head_before": heads[-1][1], "main_commits_before": len(heads)}

    # -- timed flows -----------------------------------------------------------

    def main_head(self, agent: AgentRun, cwd: Path, env: dict[str, str]) -> Optional[str]:

        proc = agent.call_retrying([str(self.oak), "branch", "show", "main", "--remote", "--json"], cwd, env)

        doc = last_json(proc.stdout) or {}

        return doc.get("head")

    def flow_naive(self, agent: AgentRun) -> None:

        env = self.server.env()

        wdir = self.workers[agent.name]

        agent.attempts = 1

        proc = agent.call_retrying([str(self.oak), "merge", "--json"], wdir, env, retry_conflicts=3)

        doc = last_json(proc.stdout) or {}

        if proc.returncode == 0 and doc.get("merged"):

            agent.landed_commit = doc.get("commit_hash")

            agent.landed_at = time.perf_counter()

            agent.outcome = "landed"

        else:

            agent.outcome = "failed"

            agent.error = f"merge rc={proc.returncode} {(proc.stderr or proc.stdout).strip()[-200:]}"

    def flow_safe(self, agent: AgentRun) -> None:

        env = self.server.env()

        wdir = self.workers[agent.name]

        while agent.attempts < MAX_ATTEMPTS:

            agent.attempts += 1

            seen_main = self.main_head(agent, wdir, env)

            pull = agent.call_retrying([str(self.oak), "pull"], wdir, env)

            if pull.returncode != 0:

                # Documented fallback: fresh shallow worker from current main + reapply.

                agent.converge_path.append("fresh_reapply")

                if agent.branch:

                    agent.abandoned_branches.append(agent.branch)

                fresh = self.root / f"w-{agent.name}-fresh{agent.attempts}"

                clone = agent.call_retrying([str(self.oak), "clone", "--shallow", f"{self.server.owner}/{self.repo}", str(fresh)], self.root, env)

                if clone.returncode != 0:

                    agent.outcome, agent.error = "failed", f"fresh clone failed: {clone.stderr.strip()[-200:]}"

                    return

                seen_main = self.main_head(agent, fresh, env)

                wdir = fresh

                agent.apply_patch(wdir, agent_patch(agent.name))

                commit = agent.call_retrying([str(self.oak), "commit"], wdir, env)

                if commit.returncode != 0:

                    agent.outcome, agent.error = "failed", f"commit failed: {commit.stderr.strip()[-200:]}"

                    return

            else:

                agent.converge_path.append("pull")

            if not agent.local_check(wdir):

                agent.outcome = "refused_red_combined"

                return

            push = agent.call_retrying([str(self.oak), "push", "--json"], wdir, env, retry_conflicts=3)

            if push.returncode != 0:

                agent.outcome, agent.error = "failed", f"push failed: {(push.stderr or push.stdout).strip()[-200:]}"

                return

            agent.branch = (last_json(push.stdout) or {}).get("branch") or agent.branch

            ok, err = oak_trigger_and_wait(agent, self.oak, wdir, env, f"{agent.name}-a{agent.attempts}")

            if not ok:

                agent.outcome, agent.error = "refused_ci_red", err

                return

            now_main = self.main_head(agent, wdir, env)

            if seen_main is None or now_main != seen_main:

                continue  # main moved since converge: the tested tree is stale; re-converge

            merge = agent.call_retrying([str(self.oak), "merge", "--json"], wdir, env)

            doc = last_json(merge.stdout) or {}

            if merge.returncode == 0 and doc.get("merged"):

                agent.landed_commit = doc.get("commit_hash")

                agent.landed_at = time.perf_counter()

                verify = self.main_head(agent, wdir, env)

                agent.outcome = "landed"

                if verify is None:

                    agent.error = "post-merge verify could not read main"

                return

            text = (merge.stderr or "") + (merge.stdout or "")

            if "main_moved" in text or "branch_moved" in text or "409" in text:

                continue

            agent.outcome, agent.error = "failed", f"merge rc={merge.returncode} {text.strip()[-200:]}"

            return

        agent.outcome, agent.error = "failed", "attempt budget exhausted"

    # -- oracles ---------------------------------------------------------------

    def oracle(self, agents: list[AgentRun]) -> dict[str, Any]:

        heads = [(g, h) for g, h in self.server.main_heads(self.repo_id) if g > self.base_generation]

        red = 0

        digests: dict[str, str] = {}

        final_files: dict[str, str] = {}

        for index, (_, head) in enumerate(heads):

            tree = self.server.materialize(self.repo_id, head, self.root / f"oracle-{index}")

            ok, _ = run_check(tree)

            red += 0 if ok else 1

            digests[head] = tree_digest(tree)

            final_files = read_tree(tree)

        if not heads:

            tree = self.server.materialize(self.repo_id, self.server.main_heads(self.repo_id)[-1][1], self.root / "oracle-final")

            final_files = read_tree(tree)

        ci_trial = read_ledger(self.trial_ledger)

        ci_setup = read_ledger(self.setup_ledger)

        return {"heads": heads, "red": red, "digests": digests, "final_files": final_files, "ci_trial": ci_trial, "ci_setup": ci_setup}

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

# Git arms (bare origin, file transport)

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

def git(args: list[str], cwd: Path, env: Optional[dict[str, str]] = None) -> subprocess.CompletedProcess[str]:

    return subprocess.run(["git"] + args, cwd=cwd, env=env, capture_output=True, text=True, timeout=300)

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

    env = dict(os.environ)

    env.update(

        {

            "GIT_AUTHOR_NAME": "bench", "GIT_AUTHOR_EMAIL": "bench@localhost", "GIT_COMMITTER_NAME": "bench",

            "GIT_COMMITTER_EMAIL": "bench@localhost", "GIT_CONFIG_NOSYSTEM": "1", "GIT_TERMINAL_PROMPT": "0",

            "PYTHONDONTWRITEBYTECODE": "1", "HOME": env.get("OAK_BENCH_GIT_HOME", tempfile.gettempdir()),

        }

    )

    return env

@dataclass

class GitTrial:

    root: Path

    n: int

    k: int

    arm: str

    ci_added_seconds: float

    workers: dict[str, Path] = field(default_factory=dict)

    base_main: str = ""

    queue: threading.Lock = field(default_factory=threading.Lock)

    @property

    def origin(self) -> Path:

        return self.root / "origin.git"

    def setup(self) -> dict[str, Any]:

        env = git_env()

        t0 = time.perf_counter()

        must(git(["init", "-q", "--bare", "-b", "main", str(self.origin)], self.root, env), "bare init")

        seed = self.root / "seed"

        must(git(["clone", "-q", str(self.origin), str(seed)], self.root, env), "seed clone")

        write_files(seed, base_files())

        must(git(["checkout", "-q", "-b", "main"], seed, env), "seed branch")

        must(git(["add", "-A"], seed, env), "seed add")

        must(git(["commit", "-q", "-m", "base"], seed, env), "seed commit")

        must(git(["push", "-q", "origin", "main"], seed, env), "seed push")

        for name in agent_names(self.n):

            wdir = self.root / f"w-{name}"

            must(git(["clone", "-q", str(self.origin), str(wdir)], self.root, env), f"clone {name}")

            must(git(["checkout", "-q", "-b", name], wdir, env), f"branch {name}")

            write_files(wdir, agent_patch(name))

            must(git(["add", "-A"], wdir, env), f"add {name}")

            must(git(["commit", "-q", "-m", name], wdir, env), f"commit {name}")

            self.workers[name] = wdir

        for j in range(self.k):

            path, content = mover_file(j)

            write_files(seed, {path: content})

            must(git(["add", "-A"], seed, env), "mover add")

            must(git(["commit", "-q", "-m", f"move {j}"], seed, env), "mover commit")

        must(git(["push", "-q", "origin", "main"], seed, env), "mover push")

        self.base_main = must(git(["rev-parse", "main"], self.origin, env), "rev-parse").stdout.strip()

        return {"setup_ms": (time.perf_counter() - t0) * 1000, "main_head_before": self.base_main, "main_commits_before": self.k + 1}

    def flow(self, agent: AgentRun) -> None:

        env = git_env()

        wdir = self.workers[agent.name]

        g = lambda args: agent.call(["git"] + args, wdir, env)  # noqa: E731

        while agent.attempts < MAX_ATTEMPTS * 3:

            ctx = self.queue if self.arm == "git_fifo" else _NullLock()

            with ctx:

                agent.attempts += 1

                if g(["fetch", "-q", "origin"]).returncode != 0:

                    continue

                if g(["rebase", "-q", "origin/main"]).returncode != 0:

                    g(["rebase", "--abort"])

                    agent.outcome, agent.error = "failed", "rebase conflict"

                    return

                agent.ci_runs_local += 1

                if not agent.local_check(wdir, added_seconds=self.ci_added_seconds):

                    agent.outcome = "refused_red_combined"

                    return

                push = g(["push", "-q", "origin", "HEAD:main"])

                if push.returncode == 0:

                    agent.landed_commit = must(git(["rev-parse", "HEAD"], wdir, env), "rev-parse").stdout.strip()

                    agent.landed_at = time.perf_counter()

                    agent.outcome = "landed"

                    return

                if "rejected" not in (push.stderr or "") and "fetch first" not in (push.stderr or ""):

                    agent.outcome, agent.error = "failed", f"push failed: {push.stderr.strip()[-200:]}"

                    return

        agent.outcome, agent.error = "failed", "attempt budget exhausted"

    def oracle(self, agents: list[AgentRun]) -> dict[str, Any]:

        env = git_env()

        revs = must(git(["rev-list", "--reverse", f"{self.base_main}..main"], self.origin, env), "rev-list").stdout.split()

        red = 0

        digests: dict[str, str] = {}

        final_files: dict[str, str] = {}

        targets = revs or [self.base_main]

        for index, rev in enumerate(targets):

            dest = self.root / f"oracle-{index}"

            dest.mkdir()

            archive = subprocess.run(["git", "archive", rev], cwd=self.origin, capture_output=True, timeout=120)

            subprocess.run(["tar", "-x", "-C", str(dest)], input=archive.stdout, check=True, timeout=120)

            if revs:

                ok, _ = run_check(dest)

                red += 0 if ok else 1

                digests[rev] = tree_digest(dest)

            final_files = read_tree(dest)

        return {"heads": [(i, r) for i, r in enumerate(revs)], "red": red, "digests": digests, "final_files": final_files, "ci_trial": [], "ci_setup": []}

class _NullLock:

    def __enter__(self) -> None:

        return None

    def __exit__(self, *exc: Any) -> None:

        return None

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

# Trial driver + rows

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

def percentile(values: list[float], p: float) -> Optional[float]:

    if not values:

        return None

    ordered = sorted(values)

    index = min(len(ordered) - 1, max(0, int(round(p / 100 * (len(ordered) - 1)))))

    return ordered[index]

def run_agents(agents: list[AgentRun], flow: Callable[[AgentRun], None]) -> tuple[float, float]:

    barrier = threading.Barrier(len(agents) + 1)

    def body(agent: AgentRun) -> None:

        barrier.wait()

        try:

            flow(agent)

        except Exception as exc:  # noqa: BLE001 - recorded as a failed agent, never hidden

            agent.outcome, agent.error = "failed", f"harness exception: {exc}"

        agent.finished_at = time.perf_counter()

    threads = [threading.Thread(target=body, args=(agent,)) for agent in agents]

    for thread in threads:

        thread.start()

    barrier.wait()

    start = time.perf_counter()

    for thread in threads:

        thread.join()

    return start, time.perf_counter()

def build_rows(

    *,

    meta: dict[str, Any],

    arm: str,

    n: int,

    k: int,

    run_index: int,

    agents: list[AgentRun],

    start: float,

    end: float,

    setup: dict[str, Any],

    oracle: dict[str, Any],

    subject: str,

    subject_kind: str,

    binary_hash: Optional[str],

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

    is_oak = arm in OAK_ARMS

    ci_by_commit: dict[str, list[dict[str, Any]]] = {}

    for record in oracle["ci_trial"]:

        ci_by_commit.setdefault(record.get("ref") or "", []).append(record)

    setup_by_ref: dict[str, list[dict[str, Any]]] = {}

    for record in oracle["ci_setup"]:

        setup_by_ref.setdefault(record.get("ref") or "", []).append(record)

    common = {

        **meta,

        "profile": PROFILE,

        "scenario": SCENARIO,

        "scenario_version": SCENARIO_VERSION,

        "fixture_version": FIXTURE_VERSION,

        "contention_mode": arm,

        "workers": n,

        "agents_total": n + len(PAIR),

        "behind_k": k,

        "run": run_index,

        "subject": subject,

        "subject_kind": subject_kind,

        "binary_sha256": binary_hash,

        "transport": "loopback" if is_oak else "local-file",

        "hosted_equivalence": "hosted-equivalent-local" if is_oak else "not-applicable-local-git",

        "ci_executor": "oakspace_dispatch+oakbench.ci_stub(loopback)" if is_oak else "local_subprocess",

        "ci_trigger_mode": CI_TRIGGER_NOTE if is_oak else "local_pre_push",

    }

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

    classification = classify_final(oracle["final_files"], n, k)

    for agent in agents:

        refs = [agent.branch or ""] + list(agent.abandoned_branches)

        ci_runs_trial = [r for ref in refs for r in ci_by_commit.get(ref, [])]

        ci_digests = [r.get("tree_digest") for r in ci_runs_trial if r.get("conclusion") == "success"]

        if arm == "oak_naive":

            ci_digests += [r.get("tree_digest") for r in setup_by_ref.get(agent.branch or "", []) if r.get("conclusion") == "success"]

        landed_digest = oracle["digests"].get(agent.landed_commit or "")

        tested = None

        local_tested = ci_tested = None

        if agent.outcome == "landed":

            local_tested = landed_digest is not None and landed_digest in agent.tested_digests

            ci_tested = landed_digest is not None and landed_digest in ci_digests

            tested = bool(local_tested or ci_tested)

        ci_runs = len(ci_runs_trial) if is_oak else agent.ci_runs_local

        rows.append(

            {

                **common,

                "operation": "land.agent",

                "agent": agent.name,

                "agent_role": "pair" if agent.name in PAIR else "ordinary",

                "elapsed_ms": round(((agent.landed_at or agent.finished_at or end) - start) * 1000, 1),

                "time_to_landed_ms": round((agent.landed_at - start) * 1000, 1) if agent.landed_at else None,

                "returncode": 0 if agent.outcome in ("landed", "refused_red_combined", "refused_ci_red") else 1,

                "outcome": agent.outcome,

                "error": agent.error,

                "tool_call_count": agent.tool_calls,

                "tool_call_count_hosted_equivalent": agent.tool_calls - agent.ci_trigger_calls if is_oak else agent.tool_calls,

                "ci_trigger_calls": agent.ci_trigger_calls if is_oak else None,

                "output_bytes": agent.output_bytes,

                "attempts": agent.attempts,

                "rate_limited_retries": agent.rate_limited if is_oak else None,

                "conflict_retries": agent.conflict_retries if is_oak else None,

                "retries": max(0, agent.attempts - 1),

                "ci_runs": ci_runs,

                "local_checks": agent.local_checks,

                "converge_path": agent.converge_path or None,

                "landed_commit": agent.landed_commit,

                "combined_tree_tested_before_land": tested,

                "combined_tree_ci_tested_before_land": ci_tested,

                "combined_tree_locally_tested_before_land": local_tested,

                "command": " ; ".join(" ".join(t["argv"][:3]) for t in agent.trace)[:2000],

                "trace": agent.trace,

            }

        )

    landed = [a for a in agents if a.outcome == "landed"]

    ordinary = [a for a in agents if a.name not in PAIR]

    times = [(a.landed_at - start) * 1000 for a in ordinary if a.landed_at]

    tested_flags = [r["combined_tree_tested_before_land"] for r in rows if r["outcome"] == "landed"]

    total_ci = len(oracle["ci_trial"]) if is_oak else sum(a.ci_runs_local for a in agents)

    final_green = None

    if oracle["final_files"]:

        with tempfile.TemporaryDirectory(prefix="sl-final-") as tmp:

            write_files(Path(tmp), oracle["final_files"])

            final_green = run_check(Path(tmp))[0]

    lost = [a.name for a in ordinary if a.outcome != "landed"]

    rows.append(

        {

            **common,

            "operation": "land.trial",

            "elapsed_ms": round((end - start) * 1000, 1),

            "returncode": 0 if (classification["intended_changes_exact"] and oracle["red"] == 0 and not lost) else 1,

            "setup_ms": round(setup["setup_ms"], 1),

            "main_commits_before": setup["main_commits_before"],

            "main_heads_created": len(oracle["heads"]),

            "red_main_events": oracle["red"],

            "final_main_green": final_green,

            **classification,

            "ordinary_landed": len(times),

            "ordinary_lost": lost,

            "failed_agents": [a.name for a in agents if a.outcome == "failed"],

            "retries_total": sum(max(0, a.attempts - 1) for a in agents),

            "rate_limited_retries_total": sum(a.rate_limited for a in agents) if is_oak else None,

            "conflict_retries_total": sum(a.conflict_retries for a in agents) if is_oak else None,

            "tool_call_count": sum(a.tool_calls for a in agents),

            "output_bytes": sum(a.output_bytes for a in agents),

            "ci_runs_total": total_ci,

            "time_to_landed_ms_p50": percentile(times, 50),

            "time_to_landed_ms_max": max(times) if times else None,

            "combined_tested_fraction": (sum(1 for f in tested_flags if f) / len(tested_flags)) if tested_flags else None,

            "landed_total": len(landed),

            "human_interventions": 0,

            "human_interventions_source": "scripted_harness_has_no_human_step",

            "command": arm,

        }

    )

    return rows

def rand_suffix() -> str:

    return "".join(random.choice(string.ascii_lowercase + string.digits) for _ in range(6))

def run_trial(args: argparse.Namespace, arm: str, n: int, k: int, run_index: int, stub: Optional[CiStub], meta: dict[str, Any], subject: tuple[str, str, Optional[Path]]) -> list[dict[str, Any]]:

    subject_name, subject_kind, oak_bin = subject

    root = Path(tempfile.mkdtemp(prefix=f"sl-{arm}-n{n}-k{k}-r{run_index}-", dir=args.scratch))

    agents = [AgentRun(name=name, arm=arm) for name in agent_names(n)]

    if arm in OAK_ARMS:

        assert stub is not None and oak_bin is not None

        server = OakServer(url=args.oak_url, token=Path(args.token_file).read_text().strip(), secret=os.environ[args.secret_env],

                           owner=args.owner, psql=args.psql.split(), home=Path(args.oak_home))

        trial = OakTrial(server=server, oak=oak_bin, setup_oak=Path(args.setup_oak), root=root,

                         repo=f"sl-{arm.replace('_', '')}-{subject_name.replace('_', '')}-n{n}-k{k}-r{run_index}-{rand_suffix()}",

                         n=n, k=k, arm=arm, stub=stub, setup_ledger=root / "ci-setup.jsonl", trial_ledger=root / "ci-trial.jsonl")

        setup = trial.setup()

        for agent in agents:

            agent.branch = trial.branches.get(agent.name)

        flow = trial.flow_naive if arm == "oak_naive" else trial.flow_safe

        with measurement_lock(f"stale_land:{arm}:n{n}:k{k}:r{run_index}") as lock:

            start, end = run_agents(agents, flow)

            stub.join_runs(timeout=60)

        oracle = trial.oracle(agents)

        binary_hash = binary_sha256(oak_bin)

    else:

        trial_g = GitTrial(root=root, n=n, k=k, arm=arm, ci_added_seconds=args.ci_added_seconds)

        setup = trial_g.setup()

        with measurement_lock(f"stale_land:{arm}:n{n}:k{k}:r{run_index}") as lock:

            start, end = run_agents(agents, trial_g.flow)

        oracle = trial_g.oracle(agents)

        binary_hash = binary_sha256(Path(shutil.which("git") or "git"))

    rows = build_rows(meta={**meta, "lock_wait_ms": lock.wait_ms, "lock_enabled": lock.enabled, "trial_root": str(root)}, arm=arm, n=n, k=k,

                      run_index=run_index, agents=agents, start=start, end=end, setup=setup, oracle=oracle,

                      subject=subject_name, subject_kind=subject_kind, binary_hash=binary_hash)

    if not args.keep_trials:

        shutil.rmtree(root, ignore_errors=True)

    return rows

def summary_text(rows: list[dict[str, Any]]) -> str:

    trials = [r for r in rows if r["operation"] == "land.trial"]

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

    for row in trials:

        groups.setdefault((row["contention_mode"], row["subject"], row["workers"], row["behind_k"], row.get("aa_group")), []).append(row)

    lines = [

        f"# {SCENARIO} baseline",

        "",

        f"CI added delay: {trials[0].get('ci_added_seconds') if trials else None} s (labeled). Oak rows: loopback, hosted-equivalent-local. Git rows: local-file.",

        "",

        "| arm | subject | N | k | trials | median p50 time-to-landed s | median makespan s | CI runs (median) | tool calls (median) | combined-tested % | red-main events | exact final | failures/lost | retries (median) |",

        "|---|---|---|---|---|---|---|---|---|---|---|---|---|---|",

    ]

    def med(values: list[Any]) -> Optional[float]:

        clean = [v for v in values if v is not None]

        return statistics.median(clean) if clean else None

    def fmt(value: Optional[float], scale: float = 1.0, digits: int = 1) -> str:

        return "n/a" if value is None else f"{value / scale:.{digits}f}"

    for (arm, subject, n, k, aa), group in sorted(groups.items(), key=lambda kv: [str(x) for x in kv[0]]):

        tested = [r["combined_tested_fraction"] for r in group if r["combined_tested_fraction"] is not None]

        lines.append(

            "| {arm}{aa} | {subject} | {n} | {k} | {t} | {p50} | {mk} | {ci} | {tc} | {tested} | {red} | {exact}/{t} | {fail} | {retries} |".format(

                arm=arm, aa=f" ({aa})" if aa else "", subject=subject, n=n, k=k, t=len(group),

                p50=fmt(med([r["time_to_landed_ms_p50"] for r in group]), 1000),

                mk=fmt(med([r["elapsed_ms"] for r in group]), 1000),

                ci=fmt(med([r["ci_runs_total"] for r in group]), 1, 0),

                tc=fmt(med([r["tool_call_count"] for r in group]), 1, 0),

                tested="n/a" if not tested else f"{100 * statistics.mean(tested):.0f}%",

                red=sum(r["red_main_events"] for r in group),

                exact=sum(1 for r in group if r["intended_changes_exact"]),

                fail=sum(len(r["failed_agents"]) + len(r["ordinary_lost"]) for r in group),

                retries=fmt(med([r["retries_total"] for r in group]), 1, 0),

            )

        )

    return "\n".join(lines) + "\n"

def parse_args(argv: Optional[list[str]] = None) -> argparse.Namespace:

    parser = argparse.ArgumentParser(description=__doc__.splitlines()[0])

    parser.add_argument("--arms", default=",".join(ARMS))

    parser.add_argument("--agents", default="1,4", help="comma list of N ordinary agents (the pair is added to each)")

    parser.add_argument("--behind", default="1,50", help="comma list of k commits main moved after the fork")

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

    parser.add_argument("--aa-group", default=None, help="label for A/A noise-floor batches")

    parser.add_argument("--ci-added-seconds", type=float, default=10.0)

    parser.add_argument("--oak-bin", action="append", default=[], help="NAME=PATH oak client subject (repeatable)")

    parser.add_argument("--setup-oak", default=None, help="oak binary for untimed seed/mover setup")

    parser.add_argument("--oak-url", default="http://127.0.0.1:3811")

    parser.add_argument("--owner", default="bench")

    parser.add_argument("--token-file", default=None)

    parser.add_argument("--oak-home", default=None, help="isolated HOME for oak clients")

    parser.add_argument("--secret-env", default="OAK_CI_SIGNING_SECRET")

    parser.add_argument("--ci-stub-port", type=int, default=3812, help="port the server's OAK_CI_DISPATCH_URL points at")

    parser.add_argument("--psql", default="psql -h 127.0.0.1 -p 55811 -U oak oak")

    parser.add_argument("--server-binary", default=None, help="oakspace server binary (provenance only)")

    parser.add_argument("--server-source", default=None, help="oakspace source identity (provenance only)")

    parser.add_argument("--scratch", default=None)

    parser.add_argument("--results", type=Path, default=Path("results/stale-land"))

    parser.add_argument("--keep-trials", action="store_true")

    parser.add_argument("--dry-run", action="store_true", help="print the trial schedule and exit")

    return parser.parse_args(argv)

def schedule(args: argparse.Namespace) -> list[tuple[str, tuple[str, str, Optional[Path]], int, int, int]]:

    arms = [a for a in args.arms.split(",") if a]

    for arm in arms:

        if arm not in ARMS:

            raise SystemExit(f"unknown arm {arm!r}; known: {', '.join(ARMS)}")

    oak_subjects: list[tuple[str, str, Optional[Path]]] = []

    for spec in args.oak_bin:

        name, _, path = spec.partition("=")

        oak_subjects.append((name, "oak", Path(path)))

    plan = []

    for n in [int(x) for x in args.agents.split(",")]:

        for k in [int(x) for x in args.behind.split(",")]:

            for run_index in range(1, args.runs + 1):

                for arm in arms:

                    subjects = oak_subjects if arm in OAK_ARMS else [("git", "git", None)]

                    for subject in subjects:

                        plan.append((arm, subject, n, k, run_index))

    return plan

def main(argv: Optional[list[str]] = None) -> int:

    args = parse_args(argv)

    plan = schedule(args)

    if args.dry_run:

        for item in plan:

            print(item[0], item[1][0], f"n={item[2]}", f"k={item[3]}", f"run={item[4]}")

        return 0

    needs_oak = any(item[0] in OAK_ARMS for item in plan)

    stub = None

    if needs_oak:

        if not (args.token_file and args.oak_home and args.setup_oak and os.environ.get(args.secret_env)):

            raise SystemExit("oak arms need --token-file, --oak-home, --setup-oak and $" + args.secret_env)

        stub = CiStub(api_url=args.oak_url, secret=os.environ[args.secret_env], ledger=Path(os.devnull),

                      added_seconds=args.ci_added_seconds, port=args.ci_stub_port).start()

    bench_id = time.strftime("%Y%m%dT%H%M%SZ", time.gmtime()) + f"-{SCENARIO}"

    meta = {

        "bench_id": bench_id,

        "ci_added_seconds": args.ci_added_seconds,

        "aa_group": args.aa_group,

        "host": platform.node(),

        "server_binary_sha256": binary_sha256(Path(args.server_binary)) if args.server_binary else None,

        "server_source": args.server_source,

    }

    store = ResultsStore(args.results, lane="contention", filename_suffix="stale-land", flush_partial=True)

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

    try:

        for arm, subject, n, k, run_index in plan:

            label = f"{arm} {subject[0]} n={n} k={k} run={run_index}"

            print(f"[stale_land] {label} ...", file=sys.stderr, flush=True)

            try:

                trial_rows = run_trial(args, arm, n, k, run_index, stub, meta, subject)

            except Exception as exc:  # noqa: BLE001 - a setup failure is a recorded row, not a silent gap

                trial_rows = [{

                    **meta, "profile": PROFILE, "scenario": SCENARIO, "scenario_version": SCENARIO_VERSION,

                    "contention_mode": arm, "workers": n, "behind_k": k, "operation": "land.trial", "run": run_index,

                    "subject": subject[0], "subject_kind": subject[1], "elapsed_ms": 0.0, "returncode": 1,

                    "error": f"trial setup/oracle failed: {exc}", "command": arm,

                }]

            trial = trial_rows[-1]

            print(

                f"[stale_land] {label}: rc={trial['returncode']} makespan={trial['elapsed_ms'] / 1000:.1f}s "

                f"p50={trial.get('time_to_landed_ms_p50')} ci={trial.get('ci_runs_total')} red={trial.get('red_main_events')} "

                f"exact={trial.get('intended_changes_exact')} tested={trial.get('combined_tested_fraction')} err={trial.get('error')}",

                file=sys.stderr, flush=True,

            )

            store.append_partial(bench_id, trial_rows)

            rows.extend(trial_rows)

    finally:

        if stub is not None:

            stub.stop()

    complete = [r for r in rows if "tool_call_count" in r or r["operation"] == "land.agent"]

    raw, summary = store.write(bench_id, rows, summary_text([r for r in complete if r["operation"] == "land.trial"] or []))

    print(f"[stale_land] rows: {raw}\n[stale_land] summary: {summary}", file=sys.stderr)

    return 0

if __name__ == "__main__":

    raise SystemExit(main())