scripts/task_loop.py · oak/benchmarks

Viewing files at commit 186ae84b74bdCommit details View latest
Browse files
scripts/task_loop.py 3177 lines · 128.5 KB · python
#!/usr/bin/env python3
"""The full task loop, end to end — the product's signature motion.

oak:  mount -> edit -> commit -> push -> desc -> end, then REMOUNT for a
      follow-up task (the warm-cache story) and repeat edit/commit/push/end.
git:  worktree add -> branch edit -> add/commit -> push -> worktree remove,
      then a second worktree for the follow-up. The task description lives in
      the commit message git already paid for; oak pays a separate `desc`
      call. That asymmetry is the measurement, not a harness gap
      (docs/command-semantics.md): this is agent-default-track, whole-task
      evidence, never a step-by-step equivalence claim.

Transport honesty: oak rows always cross the network to a real oak server.
git runs against GitHub (--git-remote github, transport-comparable) or a
local bare repo (--git-remote local, the core-equivalent track). Rows carry
remote_transport/remote_server; latency comparisons across transports are
forbidden — cross-subject evidence at unlike transports is tool calls and
tokens only.

Setup:
  OAK_BENCH_SYNC_REPO=oak/bench-sync-tmp \\
  [email protected]:oakdotspace/bench-task-loop-tmp.git \\
  python3 scripts/task_loop.py --runs 3

Unconfigured remotes emit skip rows (returncode 77), never silent absence.
"""

from __future__ import annotations

import argparse
import hashlib
import json
import os
import platform
import random
import re
import shutil
import socket
import sqlite3
import stat
import subprocess
import tempfile
import threading
import time
import urllib.error
import urllib.parse
import urllib.request
from datetime import datetime, timezone
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
from pathlib import Path
from typing import Any

from oakbench import remotes as oakbench_remotes
from oakbench import tokens as oakbench_tokens
from oakbench import merge_history_fixture, shallow_refresh
from oakbench.byteproxy import CountingProxy
from oakbench.gitserver import GitHTTPServer
from oakbench.httpcount import RequestCountingProxy
from oakbench.diskprobe import bounded_tree_usage
from oakbench.environment import ENV_ISOLATION_VERSION, base_env as oakbench_base_env
from oakbench.environment import command_display
from oakbench.execution import run_timed, stop_owned_process
from oakbench.results import ResultsStore
from oakbench.rows import row_returncode
from oakbench.runlock import measurement_lock
from oakbench.subjects import (
    DEFAULT_OAK_REPO,
    Subject,
    binary_sha256,
    load_subjects,
    source_metadata,
    subject_details,
    subject_version,
    subject_versions,
)

ROOT = Path(__file__).resolve().parents[1]
DEFAULT_WORKDIR = Path(tempfile.gettempdir()) / "oak-task-loop"
SKIP_RETURNCODE = 77
VIRTUAL_BRANCH_RE = re.compile(r"virtual branch ([^\s)]+)")
PHASES = ("task", "followup")
TASK_LOOP_SCENARIO = "task_loop"
AGENT_TASK_SCENARIO = "agent_task_baseline_v1"
AGENT_TASK_OAK_LOCAL_SCENARIO = "agent_task_oak_local_v1"
INTERRUPTED_PUBLICATION_SCENARIO = "interrupted_publication_after_request_v1"
LEGACY_SELECTED_MATERIALIZATION_SCENARIO = "legacy_selected_clone_materialization_v1"
SHALLOW_REFRESH_SCENARIO = "shallow_worker_branch_refresh_v1"
SHALLOW_REFRESH_DEFAULT_FIXTURES = "1000:1,2000:2"
SHALLOW_REFRESH_TASK_BRANCH = "task"
SHALLOW_REFRESH_STEP_TIMEOUT_S = 900.0
COMMITS_INFO_SUFFIX = "/commits/info"
INTERRUPTED_PUBLICATION_REQUEST_TIMEOUT_S = 2.0
INTERRUPTED_PUBLICATION_PROCESS_TIMEOUT_S = 30.0
INTERRUPTED_PUBLICATION_MAX_REQUEST_BYTES = 64 * 1024 * 1024
LOCAL_OAK_SERVER_READY_TIMEOUT_S = 10.0


def legacy_selected_receipt_evidence(receipt: dict[str, Any]) -> dict[str, Any]:
    result = receipt.get("result") if isinstance(receipt.get("result"), dict) else {}
    observed = receipt.get("observed") if isinstance(receipt.get("observed"), dict) else {}
    phases = observed.get("pull_phases") if isinstance(observed.get("pull_phases"), list) else []
    phase_names = [phase.get("name") for phase in phases if isinstance(phase, dict)]
    materialization = (
        observed.get("materialization")
        if isinstance(observed.get("materialization"), dict)
        else {}
    )
    return {
        "legacy_route_observed": phase_names == ["initial", "legacy_selected_branch"],
        "pull_phase_names": phase_names,
        "pull_phases": phases,
        "materialization_passes_observed": materialization.get("passes_observed"),
        "written_paths": materialization.get("written_paths"),
        "snapshot_bound": result.get("snapshot_bound"),
        "head": result.get("head"),
        "manifest": result.get("manifest"),
    }


def legacy_selected_receipt_contract(evidence: dict[str, Any]) -> dict[str, Any]:
    phases = []
    for phase in evidence.get("pull_phases", []):
        if not isinstance(phase, dict):
            continue
        phases.append(
            {
                key: value
                for key, value in phase.items()
                if key not in {"materialization_passes"}
            }
        )
    return {
        "legacy_route_observed": evidence.get("legacy_route_observed"),
        "pull_phases": phases,
        "snapshot_bound": evidence.get("snapshot_bound"),
        "head": evidence.get("head"),
        "manifest": evidence.get("manifest"),
        "written_paths": evidence.get("written_paths"),
    }


def normalized_cache_contract(snapshot: dict[str, Any] | None) -> list[dict[str, Any]] | None:
    if not isinstance(snapshot, dict) or not isinstance(snapshot.get("rows"), list):
        return None
    return [
        {key: row.get(key) for key in ("path", "size", "blob_hash")}
        for row in snapshot["rows"]
        if isinstance(row, dict)
    ]


def normalized_status(value: dict[str, Any] | list[Any] | None) -> dict[str, Any] | None:
    if not isinstance(value, dict):
        return None
    return {
        key: value.get(key)
        for key in ("branch", "parent", "branch_status", "head", "working_changes", "branch_changes")
    }


def materialization_timing_controls_met(
    metadata: dict[str, Any], runs: int, randomize_order: bool, controls_requested: bool
) -> bool:
    return bool(
        controls_requested
        and metadata.get("measurement_lock") == "held"
        and randomize_order
        and runs >= 10
    )


def isolated_oak_fixture_env(task_config_root: Path) -> dict[str, str]:
    task_config_root.mkdir(mode=0o700, parents=True, exist_ok=True)
    retained = {
        "PATH",
        "TMPDIR",
        "TMP",
        "TEMP",
        "HOME",
        "USER",
        "USERNAME",
        "SHELL",
        "SYSTEMROOT",
        "WINDIR",
        "COMSPEC",
        "PATHEXT",
        "SSL_CERT_FILE",
        "SSL_CERT_DIR",
    }
    env = {key: value for key, value in os.environ.items() if key in retained}
    env.update(
        {
            "GIT_AUTHOR_NAME": "Oak Materialization Bench",
            "GIT_AUTHOR_EMAIL": "[email protected]",
            "GIT_COMMITTER_NAME": "Oak Materialization Bench",
            "GIT_COMMITTER_EMAIL": "[email protected]",
            "GIT_CONFIG_GLOBAL": os.devnull,
            "GIT_CONFIG_NOSYSTEM": "1",
            "OAK_AUTHOR": "oak-materialization-bench",
            "OAK_NO_UPDATE_CHECK": "1",
            "OAK_BENCH_ENV_ISOLATION_VERSION": ENV_ISOLATION_VERSION,
            "NO_COLOR": "1",
            "CLICOLOR": "0",
            "LC_ALL": "C",
            "LANG": "C",
            "TZ": "UTC",
            "GIT_AUTHOR_DATE": "2026-01-01T00:00:00Z",
            "GIT_COMMITTER_DATE": "2026-01-01T00:00:00Z",
        }
    )
    return env


def normalized_worktree_snapshot(root: Path) -> list[dict[str, Any]]:
    rows: list[dict[str, Any]] = []
    for path in sorted(root.rglob("*")):
        relative = path.relative_to(root)
        if relative.parts and relative.parts[0] in {".oak", ".git"}:
            continue
        info = path.lstat()
        if stat.S_ISDIR(info.st_mode):
            continue
        row: dict[str, Any] = {
            "path": relative.as_posix(),
            "mode": stat.S_IMODE(info.st_mode),
        }
        if stat.S_ISLNK(info.st_mode):
            row.update({"kind": "symlink", "target": os.readlink(path)})
        else:
            row.update({"kind": "file", "sha256": hashlib.sha256(path.read_bytes()).hexdigest()})
        rows.append(row)
    return rows


def json_command(
    command: list[str], cwd: Path, env: dict[str, str]
) -> dict[str, Any] | list[Any] | None:
    result = run_bounded_harness_command(command, cwd, env=env)
    if result.returncode != 0:
        return None
    try:
        value = json.loads(result.stdout)
    except (TypeError, ValueError):
        return None
    return value if isinstance(value, (dict, list)) else None


def normalized_history(value: dict[str, Any] | list[Any] | None) -> list[dict[str, Any]] | None:
    if not isinstance(value, list):
        return None
    rows: list[dict[str, Any]] = []
    for item in value:
        if not isinstance(item, dict):
            return None
        rows.append(
            {
                key: item.get(key)
                for key in (
                    "hash",
                    "timestamp",
                    "branch",
                    "description_or_subject",
                    "files_changed",
                )
            }
        )
    return rows


def normalized_oak_artifacts(root: Path) -> list[str] | None:
    oak_dir = root / ".oak"
    if not oak_dir.is_dir():
        return None
    return sorted(
        path.relative_to(oak_dir).as_posix() + ("/" if path.is_dir() else "")
        for path in oak_dir.rglob("*")
    )


def stat_cache_snapshot(root: Path) -> dict[str, Any] | None:
    db = root / ".oak/oak.db"
    if not db.is_file():
        return {"rows": None, "all_metadata_valid": False, "error": "oak_db_missing"}
    try:
        connection = sqlite3.connect(str(db))
        connection.execute("PRAGMA query_only = ON")
        raw_rows = connection.execute(
            "SELECT path, mtime_ns, ctime_ns, size, blob_hash FROM stat_cache ORDER BY path"
        ).fetchall()
    except sqlite3.Error as error:
        return {"rows": None, "all_metadata_valid": False, "error": str(error)}
    finally:
        if "connection" in locals():
            connection.close()
    rows: list[dict[str, Any]] = []
    valid = True
    for path, mtime_ns, ctime_ns, size, blob_hash in raw_rows:
        file_path = root / path
        try:
            info = file_path.stat()
        except OSError:
            valid = False
            continue
        expected_ctime = info.st_ctime_ns if os.name == "posix" else 0
        metadata_valid = (
            int(size) == info.st_size
            and int(mtime_ns) == info.st_mtime_ns
            and int(ctime_ns) == expected_ctime
        )
        valid = valid and metadata_valid
        rows.append(
            {
                "path": str(path),
                "size": int(size),
                "blob_hash": str(blob_hash),
                "metadata_valid": metadata_valid,
            }
        )
    return {"rows": rows, "all_metadata_valid": valid}


def base_env() -> dict[str, str]:
    return oakbench_base_env(
        author_name="Oak Task Loop",
        author_email="[email protected]",
        oak_author="oak-task-loop",
    )


def parse_args() -> argparse.Namespace:
    parser = argparse.ArgumentParser(description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter)
    parser.add_argument("--subjects", help="Comma-separated subject names from config/subjects.toml")
    parser.add_argument("--config", type=Path, default=ROOT / "config" / "subjects.toml")
    parser.add_argument("--git-bin", type=Path)
    parser.add_argument("--oak-installed-bin", type=Path)
    parser.add_argument("--oak-local-bin", type=Path)
    parser.add_argument(
        "--oak-serve-bin",
        type=Path,
        help="Immutable Oak binary used only as the fixed loopback server and fixture writer.",
    )
    parser.add_argument(
        "--oak-serve-source",
        type=Path,
        help="Read-only source checkout that produced --oak-serve-bin.",
    )
    parser.add_argument(
        "--oak-serve-source-head",
        help="Exact 64-hex Oak source head that produced --oak-serve-bin.",
    )
    parser.add_argument(
        "--materialization-baseline-subject",
        help="Named control subject required by the legacy materialization scenario.",
    )
    parser.add_argument(
        "--materialization-timing-controls-met",
        action="store_true",
        help=(
            "Record that harness-verifiable controls are met only with a held measurement lock, "
            "randomized subject order, and at least 10 runs. This is not a speedup claim: "
            "separate A/A noise and externally verified quiet-host evidence remain required."
        ),
    )
    parser.add_argument(
        "--shallow-refresh-fixtures",
        default=SHALLOW_REFRESH_DEFAULT_FIXTURES,
        help=(
            "shallow_worker_branch_refresh_v1 merge-history fixtures as COMMITS:SEED[,...]; "
            "the first is primary, the rest are held-out scaling cases."
        ),
    )
    parser.add_argument(
        "--injected-request-delay-ms",
        type=float,
        default=0.0,
        help=(
            "shallow_worker_branch_refresh_v1: fixed sleep the loopback proxy adds before each "
            "HTTP request (labeled injected_request_delay_ms; not a shaped network)."
        ),
    )
    parser.add_argument("--runs", type=int, default=3)
    parser.add_argument(
        "--scenarios",
        default=TASK_LOOP_SCENARIO,
        help=(
            "Comma-separated scenarios: task_loop, agent_task_baseline_v1, "
            "agent_task_oak_local_v1, legacy_selected_clone_materialization_v1, "
            "shallow_worker_branch_refresh_v1, "
            "and/or interrupted_publication_after_request_v1. "
            "The interruption "
            "scenario is a loopback-only Oak fixture and never contacts a remote service."
        ),
    )
    parser.add_argument(
        "--git-remote",
        choices=("github", "local"),
        default="github",
        help=(
            "github: agent-default track, real push to GIT_BENCH_REMOTE "
            "(transport-comparable to oak). local: core-equivalent track, "
            "local bare remote (no network in git's loop)."
        ),
    )
    parser.add_argument("--workdir", type=Path, default=DEFAULT_WORKDIR)
    parser.add_argument("--results", type=Path, default=ROOT / "results" / "task-loop")
    parser.add_argument("--oak-repo", type=Path, default=DEFAULT_OAK_REPO)
    parser.add_argument("--admitted-output-chars", type=int, default=20_000)
    parser.add_argument("--keep-workdirs", action="store_true")
    parser.add_argument(
        "--randomize-subject-order",
        action="store_true",
        help="Shuffle subject order independently in each run using --random-seed.",
    )
    parser.add_argument("--random-seed", type=int, default=20260908)
    return parser.parse_args()


def run_untimed(command: list[str], cwd: Path | None, timeout: float = 300.0) -> subprocess.CompletedProcess[str]:
    return subprocess.run(
        command,
        cwd=cwd,
        env=base_env(),
        text=True,
        stdout=subprocess.PIPE,
        stderr=subprocess.PIPE,
        timeout=timeout,
        check=False,
    )


def step_row(
    subject: Subject,
    scenario: str,
    operation: str,
    command: list[str],
    cwd: Path,
    run_index: int,
    metadata: dict[str, Any],
    admitted_output_chars: int,
    *,
    kind: str = "vcs",
    expected_returncodes: tuple[int, ...] = (0,),
    preserve_stdout: bool = False,
    preserve_stderr: bool = False,
    env: dict[str, str] | None = None,
    timeout_seconds: float | None = None,
) -> dict[str, Any]:
    capture = run_timed(
        command,
        cwd,
        env or base_env(),
        admitted_output_chars,
        timeout_seconds=timeout_seconds,
    )
    expected = not capture.timed_out and capture.returncode in expected_returncodes
    is_vcs = 1 if kind == "vcs" else 0
    row = {
        **metadata,
        "subject": subject.name,
        "subject_kind": subject.kind,
        "subject_label": subject.label,
        "scenario": scenario,
        "run": run_index,
        "operation": operation,
        "step_kind": kind,
        "elapsed_ms": round(capture.elapsed_ms, 3),
        "returncode": 0 if expected else (capture.returncode or 1),
        "process_returncode": capture.returncode,
        "timed_out": capture.timed_out,
        "expected_returncodes": list(expected_returncodes),
        "command": command,
        "tool_call_count": 1,
        "terminal_tool_call_count": 1,
        "vcs_tool_call_count": is_vcs,
        **oakbench_tokens.interaction_token_fields(
            command_display(command),
            capture.stdout_text,
            capture.stderr_text,
            capture.stdout_bytes,
            capture.stderr_bytes,
            capture.stdout_truncated,
            capture.stderr_truncated,
        ),
    }
    if not expected:
        row["stderr"] = capture.stderr_text[-4000:]
    if preserve_stdout:
        row["stdout"] = capture.stdout_text
    if preserve_stderr:
        row["stderr"] = capture.stderr_text
    return row


def skip_row(
    subject: Subject,
    scenario: str,
    run_index: int,
    metadata: dict[str, Any],
    reason: str,
) -> dict[str, Any]:
    return {
        **metadata,
        "subject": subject.name,
        "subject_kind": subject.kind,
        "subject_label": subject.label,
        "scenario": scenario,
        "run": run_index,
        "operation": "task_loop.skipped",
        "step_kind": "summary",
        "elapsed_ms": 0.0,
        "returncode": SKIP_RETURNCODE,
        "command": [],
        "skipped": True,
        "skip_reason": reason,
        "tool_call_count": 0,
        "terminal_tool_call_count": 0,
        "vcs_tool_call_count": 0,
    }


def unmeasured_step_row(
    subject: Subject,
    scenario: str,
    operation: str,
    run_index: int,
    metadata: dict[str, Any],
    reason: str,
) -> dict[str, Any]:
    row = skip_row(subject, scenario, run_index, metadata, reason)
    row["operation"] = operation
    return row


def harness_step_row(
    subject: Subject,
    scenario: str,
    operation: str,
    run_index: int,
    metadata: dict[str, Any],
    elapsed_ms: float,
    *,
    returncode: int = 0,
    harness_command_count: int = 0,
) -> dict[str, Any]:
    return {
        **metadata,
        "subject": subject.name,
        "subject_kind": subject.kind,
        "subject_label": subject.label,
        "scenario": scenario,
        "run": run_index,
        "operation": operation,
        "step_kind": "harness",
        "elapsed_ms": round(elapsed_ms, 3),
        "returncode": returncode,
        "command": [],
        "tool_call_count": 0,
        "terminal_tool_call_count": 0,
        "vcs_tool_call_count": 0,
        "harness_command_count": harness_command_count,
        "stdout_bytes": 0,
        "stderr_bytes": 0,
        "retry_count": 0,
        "human_action_count": 0,
    }


def phase_total_row(
    rows: list[dict[str, Any]],
    subject: Subject,
    scenario: str,
    run_index: int,
    metadata: dict[str, Any],
    operation: str,
) -> dict[str, Any]:
    failures = [row for row in rows if row_returncode(row) not in (0, SKIP_RETURNCODE)]
    unmeasured = [
        str(row["operation"])
        for row in rows
        if row_returncode(row) == SKIP_RETURNCODE
    ]
    phase_elapsed_ms: dict[str, float] = {}
    for row in rows:
        phase = str(row.get("operation", "unknown")).split(".", 1)[0]
        phase_elapsed_ms[phase] = phase_elapsed_ms.get(phase, 0.0) + float(
            row.get("elapsed_ms", 0.0)
        )
    return {
        **metadata,
        "subject": subject.name,
        "subject_kind": subject.kind,
        "subject_label": subject.label,
        "scenario": scenario,
        "run": run_index,
        "operation": operation,
        "step_kind": "summary",
        "elapsed_ms": round(sum(float(row.get("elapsed_ms", 0)) for row in rows), 3),
        "returncode": 1 if failures else 0,
        "coverage_complete": not unmeasured,
        "unmeasured_operations": unmeasured,
        "command": [row["command"] for row in rows],
        "tool_call_count": sum(oakbench_tokens.int_or_zero(row.get("tool_call_count")) for row in rows),
        "terminal_tool_call_count": sum(
            oakbench_tokens.int_or_zero(row.get("terminal_tool_call_count")) for row in rows
        ),
        "vcs_tool_call_count": sum(oakbench_tokens.int_or_zero(row.get("vcs_tool_call_count")) for row in rows),
        "agent_tool_call_count": sum(
            oakbench_tokens.int_or_zero(row.get("tool_call_count")) for row in rows
        ),
        "harness_command_count": sum(
            oakbench_tokens.int_or_zero(row.get("harness_command_count")) for row in rows
        ),
        "retry_count": sum(oakbench_tokens.int_or_zero(row.get("retry_count")) for row in rows),
        "human_action_count": sum(
            oakbench_tokens.int_or_zero(row.get("human_action_count")) for row in rows
        ),
        "phase_elapsed_ms": {
            phase: round(elapsed, 3) for phase, elapsed in sorted(phase_elapsed_ms.items())
        },
        **oakbench_tokens.summed_token_fields(
            rows, "sum_of_task_loop_steps_command_plus_admitted_output_chars_div_4"
        ),
        "summarized_operations": [row["operation"] for row in rows],
    }


def has_lane_failures(rows: list[dict[str, Any]]) -> bool:
    return any(row_returncode(row) not in (0, SKIP_RETURNCODE) for row in rows)


def enforce_oracle_result(total: dict[str, Any], oracle_ok: bool | None) -> dict[str, Any]:
    if oracle_ok is False:
        total["returncode"] = 1
        total["stderr"] = "oracle_payload_missing_from_remote_branch"
    return total


def interrupted_publication_fields(
    *,
    request_count: int,
    request_bytes: int | None,
    declared_request_bytes: int | None,
    request_framing_valid: bool,
    request_body_timed_out: bool,
    local_head_before: str | None,
    local_head_after: str | None,
    status_after: dict[str, Any] | None,
    worktree_payload_before: str | None,
    worktree_payload_after: str | None,
    cli_observation: dict[str, str],
) -> dict[str, Any]:
    """Evidence fields for a single interrupted mutation attempt.

    The fixture's receipt of a full request is ground truth for the harness,
    not a server durability receipt available to the CLI.  A disconnect after
    that point is therefore unknown to the client and must never imply replay.
    """
    request_complete = bool(
        request_count == 1
        and request_framing_valid
        and declared_request_bytes is not None
        and request_bytes == declared_request_bytes
    )
    if request_count == 0:
        delivery_state = "no_request_observed"
        fixture_outcome = "not_sent_observed"
    elif not request_framing_valid:
        delivery_state = "invalid_request_framing"
        fixture_outcome = "indeterminate_after_invalid_request_disconnect"
    elif request_complete:
        delivery_state = "complete_request_observed"
        fixture_outcome = "unknown_after_complete_request_disconnect"
    else:
        delivery_state = "partial_request_observed"
        fixture_outcome = "indeterminate_after_partial_request_disconnect"
    payload_preserved = bool(
        worktree_payload_before
        and worktree_payload_before == worktree_payload_after
    )
    unpublished_state_preserved = bool(status_after and status_after.get("needs_push"))
    return {
        "publication_outcome": cli_observation["reported_outcome"],
        "cli_observation": cli_observation["observation"],
        "fixture_delivery_state": delivery_state,
        "fixture_outcome_knowledge": fixture_outcome,
        "fixture_received_complete_request_before_disconnect": request_complete,
        "fixture_request_framing_valid": request_framing_valid,
        "fixture_request_body_timed_out": request_body_timed_out,
        "fixture_declared_request_bytes": declared_request_bytes,
        "mutation_requests_observed": request_count,
        "blind_replay_requests": max(0, request_count - 1),
        "network_request_bytes_observed": request_bytes,
        "local_head_preserved": bool(local_head_before) and local_head_before == local_head_after,
        "worktree_payload_preserved": payload_preserved,
        "unpublished_state_preserved": unpublished_state_preserved,
        "local_work_preserved": payload_preserved and unpublished_state_preserved,
        "working_tree_dirty_after": None if status_after is None else status_after.get("dirty"),
    }


def cli_publication_observation(row: dict[str, Any], *, json_requested: bool) -> dict[str, str]:
    """Classify only what the CLI reported, never what the fault fixture knows."""
    if row.get("timed_out"):
        return {"reported_outcome": "unreported", "observation": "process_timeout"}
    stdout = str(row.get("stdout") or "")
    stderr = str(row.get("stderr") or "")
    value = _parse_json_object(stdout) if json_requested else None
    if value is not None and row.get("process_returncode") == 0 and value.get("published") is True:
        return {"reported_outcome": "published", "observation": "json_success_receipt"}
    if value is not None and isinstance(value.get("error"), dict):
        message = str(value["error"].get("message") or "").lower()
        if any(marker in message for marker in ("unconfirmed", "unknown", "may have")):
            return {"reported_outcome": "unknown", "observation": "json_error_unknown"}
        return {
            "reported_outcome": "unreported",
            "observation": "json_error_without_publication_outcome",
        }
    message = f"{stdout}\n{stderr}".lower()
    if any(marker in message for marker in ("unconfirmed", "unknown", "may have")):
        return {"reported_outcome": "unknown", "observation": "text_error_unknown"}
    return {"reported_outcome": "unreported", "observation": "text_error_without_publication_outcome"}


def file_sha256(path: Path) -> str | None:
    try:
        return hashlib.sha256(path.read_bytes()).hexdigest()
    except OSError:
        return None


class _DisconnectAfterPushHandler(BaseHTTPRequestHandler):
    """Minimal loopback fixture for a complete request with a lost reply."""

    server_version = "oakbench-interruption/1"

    def log_message(self, _format: str, *_args: object) -> None:
        return

    def _json(self, status: int, payload: dict[str, Any]) -> None:
        encoded = json.dumps(payload).encode()
        self.send_response(status)
        self.send_header("Content-Type", "application/json")
        self.send_header("Content-Length", str(len(encoded)))
        self.end_headers()
        self.wfile.write(encoded)

    def do_GET(self) -> None:  # noqa: N802 - BaseHTTPRequestHandler API
        self._json(404, {"error": "absent synthetic repo"})

    def do_POST(self) -> None:  # noqa: N802 - BaseHTTPRequestHandler API
        declared_size: int | None = None
        framing_valid = False
        body = b""
        body_timed_out = False
        raw_sizes = self.headers.get_all("Content-Length", failobj=[])
        try:
            raw_size = raw_sizes[0].strip(" \t") if len(raw_sizes) == 1 else ""
            parsed_size = int(raw_size) if raw_size.isascii() and raw_size.isdecimal() else -1
            framing_valid = (
                0 <= parsed_size <= INTERRUPTED_PUBLICATION_MAX_REQUEST_BYTES
                and self.headers.get("Transfer-Encoding") is None
            )
            if framing_valid:
                declared_size = parsed_size
                self.connection.settimeout(INTERRUPTED_PUBLICATION_REQUEST_TIMEOUT_S)
                while len(body) < parsed_size:
                    try:
                        chunk = self.rfile.read1(parsed_size - len(body))
                    except (socket.timeout, TimeoutError):
                        body_timed_out = True
                        break
                    except OSError:
                        break
                    if not chunk:
                        break
                    body += chunk
        except ValueError:
            framing_valid = False
        server = self.server
        if self.path == "/api/repos":
            self._json(200, {"name": "repo"})
            return
        if self.path == "/api/oak/repo/push":
            with server.receipt_lock:  # type: ignore[attr-defined]
                server.push_request_count += 1  # type: ignore[attr-defined]
                server.push_request_bytes += len(body)  # type: ignore[attr-defined]
                server.push_declared_request_bytes = declared_size  # type: ignore[attr-defined]
                server.push_request_framing_valid = framing_valid  # type: ignore[attr-defined]
                server.push_request_body_timed_out = body_timed_out  # type: ignore[attr-defined]
                server.request_complete_before_disconnect = (  # type: ignore[attr-defined]
                    framing_valid and len(body) == declared_size
                )
            self.close_connection = True
            try:
                self.connection.shutdown(socket.SHUT_RDWR)
            except OSError:
                pass
            self.connection.close()
            return
        self._json(404, {"error": "unexpected synthetic path"})


def start_interrupted_publication_server() -> tuple[ThreadingHTTPServer, threading.Thread]:
    server = ThreadingHTTPServer(("127.0.0.1", 0), _DisconnectAfterPushHandler)
    server.receipt_lock = threading.Lock()  # type: ignore[attr-defined]
    server.push_request_count = 0  # type: ignore[attr-defined]
    server.push_request_bytes = 0  # type: ignore[attr-defined]
    server.push_declared_request_bytes = None  # type: ignore[attr-defined]
    server.push_request_framing_valid = False  # type: ignore[attr-defined]
    server.push_request_body_timed_out = False  # type: ignore[attr-defined]
    server.request_complete_before_disconnect = False  # type: ignore[attr-defined]
    thread = threading.Thread(target=server.serve_forever, name="oakbench-interrupted-push", daemon=True)
    thread.start()
    return server, thread


def _parse_json_object(text: str | None) -> dict[str, Any] | None:
    try:
        value = json.loads(text or "")
    except (TypeError, json.JSONDecodeError):
        return None
    return value if isinstance(value, dict) else None


def run_interrupted_publication(
    subject: Subject,
    run_root: Path,
    run_index: int,
    metadata: dict[str, Any],
    admitted: int,
) -> list[dict[str, Any]]:
    """Run one real Oak push against a loopback server that drops its reply.

    The fixture observes request delivery but deliberately has no durable
    remote state. That makes the client's outcome unknown and lets the lane
    prove work preservation and absence of blind replay without production
    mutations or a fake success claim.
    """
    scenario = INTERRUPTED_PUBLICATION_SCENARIO
    metadata = {
        **metadata,
        "benchmark_track": "failure-evidence",
        "remote_transport": "loopback_fault_fixture",
        "remote_server": "127.0.0.1",
    }
    if subject.kind != "oak":
        return [
            skip_row(
                subject,
                scenario,
                run_index,
                metadata,
                "interrupted publication fixture currently exercises the actual Oak HTTP client only",
            )
        ]

    workspace = run_root / f"interrupted-{subject.name}-r{run_index}"
    workspace.mkdir(parents=True)
    rows: list[dict[str, Any]] = []
    vcs = str(subject.bin)
    rows.append(
        step_row(subject, scenario, "setup.init", [vcs, "init", "."], workspace, run_index, metadata, admitted)
    )
    if row_returncode(rows[-1]) != 0:
        return rows
    rows.append(
        edit_step(
            subject,
            scenario,
            "setup.edit",
            workspace,
            f"interrupted publication run={run_index}",
            run_index,
            metadata,
            admitted,
        )
    )
    rows.append(
        step_row(subject, scenario, "setup.checkpoint", [vcs, "commit"], workspace, run_index, metadata, admitted)
    )
    if row_returncode(rows[-1]) != 0:
        return rows

    before = run_untimed([vcs, "hash"], workspace)
    head_before = before.stdout.strip() if before.returncode == 0 else None
    payload_path = workspace / "task-notes.md"
    payload_before = file_sha256(payload_path)
    before_usage = bounded_tree_usage(workspace)
    push_help = run_untimed([vcs, "push", "--help"], workspace)
    push_supports_json = "--json" in (push_help.stdout + push_help.stderr)
    server, thread = start_interrupted_publication_server()
    host, port = server.server_address[:2]
    remote = f"http://{host}:{port}"
    fixture_env = base_env()
    fixture_env.pop("OAK_API_KEY", None)
    fixture_env.pop("OAK_REMOTE", None)
    fixture_env["OAK_NO_UPDATE_CHECK"] = "1"
    try:
        push_command = [vcs, "push", "--repo", "oak/repo", "--remote", remote]
        if push_supports_json:
            push_command.append("--json")
        publication_attempt = step_row(
            subject,
            scenario,
            "publication.attempt",
            push_command,
            workspace,
            run_index,
            metadata,
            admitted,
            expected_returncodes=tuple(range(1, 128)),
            preserve_stdout=True,
            preserve_stderr=True,
            env=fixture_env,
            timeout_seconds=INTERRUPTED_PUBLICATION_PROCESS_TIMEOUT_S,
        )
        rows.append(publication_attempt)
    finally:
        server.shutdown()
        server.server_close()
        thread.join(timeout=2.0)

    state_row = step_row(
        subject,
        scenario,
        "reconcile.local_state",
        [vcs, "agent", "state", "--json", "--compact"],
        workspace,
        run_index,
        metadata,
        admitted,
        preserve_stdout=True,
        env=fixture_env,
    )
    rows.append(state_row)
    after = run_untimed([vcs, "hash"], workspace)
    head_after = after.stdout.strip() if after.returncode == 0 else None
    payload_after = file_sha256(payload_path)
    status_after = _parse_json_object(state_row.get("stdout"))
    with server.receipt_lock:  # type: ignore[attr-defined]
        request_count = int(server.push_request_count)  # type: ignore[attr-defined]
        request_bytes = int(server.push_request_bytes)  # type: ignore[attr-defined]
        declared_request_bytes = server.push_declared_request_bytes  # type: ignore[attr-defined]
        request_framing_valid = bool(server.push_request_framing_valid)  # type: ignore[attr-defined]
        request_body_timed_out = bool(server.push_request_body_timed_out)  # type: ignore[attr-defined]
    cli_observation = cli_publication_observation(
        publication_attempt,
        json_requested=push_supports_json,
    )
    evidence = interrupted_publication_fields(
        request_count=request_count,
        request_bytes=request_bytes if request_count else None,
        declared_request_bytes=declared_request_bytes,
        request_framing_valid=request_framing_valid,
        request_body_timed_out=request_body_timed_out,
        local_head_before=head_before,
        local_head_after=head_after,
        status_after=status_after,
        worktree_payload_before=payload_before,
        worktree_payload_after=payload_after,
        cli_observation=cli_observation,
    )
    evidence["workspace_usage_before"] = before_usage
    evidence["workspace_usage_after"] = bounded_tree_usage(workspace)
    evidence["remote_transport"] = "loopback_fault_fixture"
    evidence["remote_server"] = str(host)
    evidence["human_action_count"] = 0
    evidence["compatibility_probe_calls"] = 1
    evidence["harness_probe_calls"] = 3
    evidence["push_json_supported"] = push_supports_json
    total = phase_total_row(rows, subject, scenario, run_index, metadata, "interrupted_publication.total")
    evidence["harness_command_count"] = total["harness_command_count"] + 3
    total.update(evidence)
    if not (
        evidence["fixture_received_complete_request_before_disconnect"]
        and request_count == 1
        and evidence["local_head_preserved"]
        and evidence["local_work_preserved"]
        and not publication_attempt["timed_out"]
        and row_returncode(rows[-1]) == 0
    ):
        total["returncode"] = 1
        total["stderr"] = "interrupted_publication_evidence_incomplete"
    rows.append(total)
    return rows


def edit_step(
    subject: Subject,
    scenario: str,
    operation: str,
    workspace: Path,
    payload: str,
    run_index: int,
    metadata: dict[str, Any],
    admitted: int,
) -> dict[str, Any]:
    command = ["/bin/zsh", "-c", f"echo {payload!r} >> task-notes.md"]
    return step_row(
        subject, scenario, operation, command, workspace, run_index, metadata, admitted, kind="edit"
    )


def oak_virtual_branch(subject: Subject, mount_dir: Path) -> str | None:
    proc = run_untimed([str(subject.bin), "status"], mount_dir)
    match = VIRTUAL_BRANCH_RE.search(proc.stdout + proc.stderr)
    return match.group(1) if match else None


def run_oak_loop(
    subject: Subject,
    remote: oakbench_remotes.RemoteResolution,
    run_root: Path,
    run_index: int,
    metadata: dict[str, Any],
    admitted: int,
) -> list[dict[str, Any]]:
    scenario = "task_loop"
    meta = {**metadata, **remote.row_fields()}
    rows: list[dict[str, Any]] = []
    branches: dict[str, str | None] = {}
    payloads: dict[str, str] = {}
    mounts: list[Path] = []
    vcs = str(subject.bin)
    try:
        for phase in PHASES:
            workspace = run_root / f"tl-{subject.name}-r{run_index}-{phase}"
            payload = f"payload {phase} run={run_index} bench={metadata['bench_id']}"
            payloads[phase] = payload
            acquire = step_row(
                subject, scenario, f"{phase}.acquire", [vcs, "mount", remote.repo or "", str(workspace)],
                run_root, run_index, meta, admitted,
            )
            rows.append(acquire)
            if int(acquire["returncode"]) != 0:
                return rows
            mounts.append(workspace)
            branches[phase] = oak_virtual_branch(subject, workspace)
            rows.append(edit_step(subject, scenario, f"{phase}.edit", workspace, payload, run_index, meta, admitted))
            rows.append(step_row(subject, scenario, f"{phase}.snapshot", [vcs, "commit", "--no-verify"], workspace, run_index, meta, admitted))
            rows.append(step_row(subject, scenario, f"{phase}.publish", [vcs, "push"], workspace, run_index, meta, admitted))
            if phase == "task":
                rows.append(
                    step_row(
                        subject, scenario, "task.describe",
                        [vcs, "desc", f"bench task loop {metadata['bench_id']} r{run_index}"],
                        workspace, run_index, meta, admitted,
                    )
                )
            end = step_row(
                subject, scenario, f"{phase}.end", [vcs, "mount", "end", str(workspace)],
                run_root, run_index, meta, admitted,
            )
            rows.append(end)
            if int(end["returncode"]) == 0 and workspace in mounts:
                mounts.remove(workspace)

        # Untimed oracle: each phase's payload must be on its pushed branch.
        verifier = run_root / f"tl-verify-{subject.name}-r{run_index}"
        clone = run_untimed([vcs, "clone", remote.repo or "", str(verifier)], run_root)
        oracle_ok: bool | None = None
        if clone.returncode == 0:
            oracle_ok = True
            for phase in PHASES:
                branch = branches.get(phase)
                if not branch:
                    oracle_ok = None
                    break
                switch = run_untimed([vcs, "switch", branch], verifier)
                content = (verifier / "task-notes.md")
                seen = content.read_text() if content.exists() else ""
                if switch.returncode != 0 or payloads[phase] not in seen:
                    oracle_ok = False
        total = {
            **phase_total_row(rows, subject, scenario, run_index, meta, "task_loop.total"),
            "oracle_payload_on_remote_branch": oracle_ok,
            "workspace_branches": branches,
            "measurement_source": "verifier_clone_switch_grep; null oracle means branch name unparsed",
        }
        rows.append(enforce_oracle_result(total, oracle_ok))
        return rows
    finally:
        for mount_dir in mounts:
            run_untimed([vcs, "mount", "end", str(mount_dir), "-f"], run_root)


def ensure_git_remote_seeded(subject: Subject, remote: oakbench_remotes.RemoteResolution, scratch: Path) -> str | None:
    """Untimed: make sure the git remote has a main with a seed commit."""
    vcs = str(subject.bin)
    probe = run_untimed([vcs, "ls-remote", str(remote.repo), "refs/heads/main"], None)
    if probe.returncode == 0 and probe.stdout.strip():
        if remote.transport == oakbench_remotes.TRANSPORT_LOCAL_FILE:
            head = run_untimed([vcs, "symbolic-ref", "HEAD", "refs/heads/main"], Path(str(remote.repo)))
            if head.returncode != 0:
                return f"git remote HEAD setup failed: {head.stderr.strip()[:200]}"
        return None
    seed = scratch / "git-seed"
    if seed.exists():
        shutil.rmtree(seed)
    seed.mkdir(parents=True)
    (seed / "README.md").write_text("Task-loop benchmark repo (disposable).\n")
    (seed / "task-notes.md").write_text("# task notes\n")
    for command in (
        [vcs, "init", "-q", "-b", "main"],
        [vcs, "add", "."],
        [vcs, "commit", "-q", "-m", "seed"],
        [vcs, "push", "-q", str(remote.repo), "main:main"],
    ):
        proc = run_untimed(command, seed)
        if proc.returncode != 0:
            return f"git remote seed `{' '.join(command[1:3])}` failed: {proc.stderr.strip()[:200]}"
    if remote.transport == oakbench_remotes.TRANSPORT_LOCAL_FILE:
        head = run_untimed([vcs, "symbolic-ref", "HEAD", "refs/heads/main"], Path(str(remote.repo)))
        if head.returncode != 0:
            return f"git remote HEAD setup failed: {head.stderr.strip()[:200]}"
    return None


def run_git_agent_task_baseline(
    subject: Subject,
    remote: oakbench_remotes.RemoteResolution,
    run_root: Path,
    run_index: int,
    metadata: dict[str, Any],
    admitted: int,
    remote_create_ms: float = 0.0,
) -> list[dict[str, Any]]:
    """Measure a cold clone and a warm isolated worktree on one local remote.

    This is the core-equivalent, local-file cell. Hosted CI is deliberately a
    skip row: a local commit oracle is not relabelled as provider CI.
    """
    scenario = AGENT_TASK_SCENARIO
    meta = {
        **metadata,
        **remote.row_fields(),
        "benchmark_track": (
            "core-equivalent"
            if remote.transport == oakbench_remotes.TRANSPORT_LOCAL_FILE
            else "agent-default"
        ),
    }
    vcs = str(subject.bin)
    rows: list[dict[str, Any]] = [
        harness_step_row(
            subject,
            scenario,
            "setup.remote_container",
            run_index,
            meta,
            remote_create_ms,
            harness_command_count=1,
        )
    ]
    setup_start = time.perf_counter()
    seed_error = ensure_git_remote_seeded(subject, remote, run_root)
    rows.append(
        harness_step_row(
            subject,
            scenario,
            "setup.remote_seed",
            run_index,
            meta,
            (time.perf_counter() - setup_start) * 1000,
            returncode=1 if seed_error else 0,
            harness_command_count=6,
        )
    )
    if seed_error:
        rows[-1]["stderr"] = seed_error
        return rows

    base = run_root / f"agent-task-{subject.name}-r{run_index}-cold"
    cold_branch = oakbench_remotes.disposable_branch(
        str(metadata["bench_id"]), "agent-task-cold", subject.name, run_index
    )
    warm_branch = oakbench_remotes.disposable_branch(
        str(metadata["bench_id"]), "agent-task-warm", subject.name, run_index
    )
    cold_payload = f"cold task run={run_index} bench={metadata['bench_id']}"
    warm_payload = f"warm task run={run_index} bench={metadata['bench_id']}"

    rows.append(
        step_row(
            subject,
            scenario,
            "cold.acquire",
            [vcs, "clone", "-q", str(remote.repo), str(base)],
            run_root,
            run_index,
            meta,
            admitted,
        )
    )
    if row_returncode(rows[-1]) != 0:
        return rows
    rows.append(
        step_row(
            subject, scenario, "cold.branch", [vcs, "switch", "-q", "-c", cold_branch],
            base, run_index, meta, admitted,
        )
    )
    rows.append(
        step_row(
            subject, scenario, "cold.inspect", [vcs, "status", "--porcelain=v1"],
            base, run_index, meta, admitted,
        )
    )
    rows.append(edit_step(subject, scenario, "cold.edit", base, cold_payload, run_index, meta, admitted))
    rows.append(step_row(subject, scenario, "cold.checkpoint.add", [vcs, "add", "."], base, run_index, meta, admitted))
    rows.append(
        step_row(
            subject, scenario, "cold.checkpoint.commit",
            [vcs, "commit", "-q", "-m", f"agent task cold {metadata['bench_id']} r{run_index}"],
            base, run_index, meta, admitted,
        )
    )
    rows.append(
        step_row(
            subject, scenario, "cold.publish", [vcs, "push", "-q", "-u", "origin", cold_branch],
            base, run_index, meta, admitted,
        )
    )
    rows.append(
        unmeasured_step_row(
            subject, scenario, "cold.ci.observe", run_index, meta,
            "local-file transport has no hosted CI provider; exact CI observation is unmeasured",
        )
    )
    rows.append(
        step_row(
            subject, scenario, "cold.review_handoff",
            [vcs, "show", "--format=%H%n%s", "--stat", "--oneline", "HEAD"],
            base, run_index, meta, admitted,
        )
    )

    warm = run_root / f"agent-task-{subject.name}-r{run_index}-warm"
    rows.append(
        step_row(
            subject, scenario, "warm.acquire",
            [vcs, "worktree", "add", "-q", str(warm), "-b", warm_branch, "origin/main"],
            base, run_index, meta, admitted,
        )
    )
    if row_returncode(rows[-1]) == 0:
        rows.append(
            step_row(
                subject, scenario, "warm.inspect", [vcs, "status", "--porcelain=v1"],
                warm, run_index, meta, admitted,
            )
        )
        rows.append(edit_step(subject, scenario, "warm.edit", warm, warm_payload, run_index, meta, admitted))
        rows.append(step_row(subject, scenario, "warm.checkpoint.add", [vcs, "add", "."], warm, run_index, meta, admitted))
        rows.append(
            step_row(
                subject, scenario, "warm.checkpoint.commit",
                [vcs, "commit", "-q", "-m", f"agent task warm {metadata['bench_id']} r{run_index}"],
                warm, run_index, meta, admitted,
            )
        )
        rows.append(
            step_row(
                subject, scenario, "warm.publish", [vcs, "push", "-q", "-u", "origin", warm_branch],
                warm, run_index, meta, admitted,
            )
        )
        rows.append(
            unmeasured_step_row(
                subject, scenario, "warm.ci.observe", run_index, meta,
                "local-file transport has no hosted CI provider; exact CI observation is unmeasured",
            )
        )
        rows.append(
            step_row(
                subject, scenario, "warm.review_handoff",
                [vcs, "show", "--format=%H%n%s", "--stat", "--oneline", "HEAD"],
                warm, run_index, meta, admitted,
            )
        )

    usage_before_cleanup = {
        "cold": bounded_tree_usage(base),
        "warm": bounded_tree_usage(warm),
    }
    oracle_ok = True
    for branch, payload in ((cold_branch, cold_payload), (warm_branch, warm_payload)):
        show = run_untimed([vcs, "--git-dir", str(remote.repo), "show", f"{branch}:task-notes.md"], run_root)
        if show.returncode != 0 or payload not in show.stdout:
            oracle_ok = False

    if warm.exists():
        rows.append(
            step_row(
                subject, scenario, "cleanup.warm",
                [vcs, "worktree", "remove", str(warm)], base, run_index, meta, admitted,
            )
        )
    cleanup_start = time.perf_counter()
    shutil.rmtree(base, ignore_errors=True)
    rows.append(
        harness_step_row(
            subject, scenario, "cleanup.cold", run_index, meta,
            (time.perf_counter() - cleanup_start) * 1000,
        )
    )

    total = phase_total_row(rows, subject, scenario, run_index, meta, "agent_task.total")
    total.update(
        {
            "oracle_payload_on_remote_branch": oracle_ok,
            "acquisition_models": {
                "cold": "fresh_full_clone",
                "warm": "shared_object_worktree",
            },
            "repository_cache_state": {
                "cold": "fresh_destination_no_shared_object_store",
                "warm": "explicit_shared_object_store",
            },
            "os_cache_state": "unknown",
            "os_cache_state_reason": "page_cache_not_purged_or_instrumented",
            "workspace_usage_before_cleanup": usage_before_cleanup,
            "network_bytes": None,
            "network_bytes_reason": "local_file_transport_not_observed_by_byte_proxy",
            "partial_clone_comparator": None,
            "partial_clone_reason": "local file transport does not establish a network partial-clone comparison",
            "human_action_count": 0,
            "parallel_evidence_lane": "parallel_contention.py/workspace_per_task",
            "oracle_probe_calls": 2,
            "harness_command_count": total["harness_command_count"] + 2,
        }
    )
    if not oracle_ok:
        total["returncode"] = 1
        total["stderr"] = "agent_task_remote_payload_oracle_failed"
    rows.append(total)
    return rows


def reserve_loopback_port() -> int:
    with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as listener:
        listener.bind(("127.0.0.1", 0))
        return int(listener.getsockname()[1])


def wait_for_loopback_port(
    process: subprocess.Popen[Any], port: int, timeout_s: float = LOCAL_OAK_SERVER_READY_TIMEOUT_S
) -> bool:
    deadline = time.monotonic() + timeout_s
    while time.monotonic() < deadline:
        if process.poll() is not None:
            return False
        try:
            with socket.create_connection(("127.0.0.1", port), timeout=0.2):
                return True
        except OSError:
            time.sleep(0.05)
    return False


def harness_command_row(
    subject: Subject,
    scenario: str,
    operation: str,
    command: list[str],
    cwd: Path,
    run_index: int,
    metadata: dict[str, Any],
) -> tuple[dict[str, Any], subprocess.CompletedProcess[str]]:
    started = time.perf_counter()
    result = run_bounded_harness_command(command, cwd)
    row = harness_step_row(
        subject,
        scenario,
        operation,
        run_index,
        metadata,
        (time.perf_counter() - started) * 1000,
        returncode=0 if result.returncode == 0 else result.returncode,
        harness_command_count=1,
    )
    if result.returncode != 0:
        row["stderr"] = str(result.stderr)[-4000:]
    return row, result


def run_bounded_harness_command(
    command: list[str], cwd: Path, *, env: dict[str, str] | None = None
) -> subprocess.CompletedProcess[str]:
    capture = run_timed(
        command,
        cwd,
        env or base_env(),
        admitted_output_chars=4_000,
        timeout_seconds=INTERRUPTED_PUBLICATION_PROCESS_TIMEOUT_S,
    )
    return subprocess.CompletedProcess(
        command,
        capture.returncode,
        capture.stdout_text,
        capture.stderr_text,
    )


def remote_branch_head(remote: str, repo_spec: str, branch: str) -> str | None:
    owner, repo = repo_spec.split("/", 1)
    branch_segment = urllib.parse.quote(branch, safe="")
    url = f"{remote}/api/{owner}/{repo}/branches/{branch_segment}"
    try:
        with urllib.request.urlopen(url, timeout=2.0) as response:
            value = json.loads(response.read())
    except (OSError, ValueError, urllib.error.URLError):
        return None
    head = value.get("head") if isinstance(value, dict) else None
    return head if isinstance(head, str) else None


def run_oak_local_agent_task_baseline(
    subject: Subject,
    run_root: Path,
    run_index: int,
    metadata: dict[str, Any],
    admitted: int,
) -> list[dict[str, Any]]:
    """Exercise two fresh Oak checkouts against an owned loopback Oak server."""
    run_root = run_root.resolve()
    scenario = AGENT_TASK_OAK_LOCAL_SCENARIO
    meta = {
        **metadata,
        "benchmark_track": "oak-local-loopback",
        "remote_transport": "loopback_http",
        "remote_server": "127.0.0.1",
    }
    rows: list[dict[str, Any]] = []
    vcs = str(subject.bin)
    server_root = run_root / f"oak-local-server-{subject.name}-r{run_index}"
    seed = run_root / f"oak-local-seed-{subject.name}-r{run_index}"
    cold = run_root / f"oak-local-cold-{subject.name}-r{run_index}"
    followup = run_root / f"oak-local-followup-{subject.name}-r{run_index}"
    verifier = run_root / f"oak-local-verifier-{subject.name}-r{run_index}"
    repo_spec = f"qa/task-loop-{subject.name}-r{run_index}"
    port = reserve_loopback_port()
    remote = f"http://127.0.0.1:{port}"
    server_root.mkdir(parents=True)
    server_started = time.perf_counter()
    server: subprocess.Popen[Any] | None = None
    server_start_error: str | None = None
    try:
        server = subprocess.Popen(
            [vcs, "serve", "--dir", str(server_root), "--host", "127.0.0.1", "--port", str(port)],
            cwd=run_root,
            env=base_env(),
            stdout=subprocess.DEVNULL,
            stderr=subprocess.DEVNULL,
            start_new_session=os.name == "posix",
        )
    except OSError as error:
        server_start_error = type(error).__name__
    ready = server is not None and wait_for_loopback_port(server, port)
    rows.append(
        harness_step_row(
            subject,
            scenario,
            "setup.server",
            run_index,
            meta,
            (time.perf_counter() - server_started) * 1000,
            returncode=0 if ready else 1,
            harness_command_count=1,
        )
    )
    branch: str | None = None
    expected_head: str | None = None
    expected_payload = "# task notes\n"
    failure: str | None = None if ready else (
        f"local_oak_server_start_failed:{server_start_error}"
        if server_start_error else "local_oak_server_not_ready"
    )
    try:
        if failure is None:
            seed.mkdir(parents=True)
            (seed / "task-notes.md").write_text(expected_payload)
            for operation, command in (
                ("setup.seed_init", [vcs, "init", "."]),
                ("setup.seed_commit", [vcs, "commit"]),
                ("setup.seed_push", [vcs, "push", "--repo", repo_spec, "--remote", remote]),
            ):
                row, result = harness_command_row(
                    subject, scenario, operation, command, seed, run_index, meta
                )
                rows.append(row)
                if result.returncode != 0:
                    failure = f"{operation}_failed"
                    break

        def agent_step(operation: str, command: list[str], cwd: Path, *, preserve: bool = False) -> dict[str, Any]:
            row = step_row(
                subject,
                scenario,
                operation,
                command,
                cwd,
                run_index,
                meta,
                admitted,
                preserve_stdout=preserve,
                timeout_seconds=INTERRUPTED_PUBLICATION_PROCESS_TIMEOUT_S,
            )
            rows.append(row)
            return row

        if failure is None:
            row = agent_step(
                "cold.acquire",
                [vcs, "clone", repo_spec, str(cold), "--remote", remote, "--allow-legacy-scope"],
                run_root,
            )
            if row_returncode(row) != 0:
                failure = "cold.acquire_failed"
        if failure is None:
            status = agent_step("cold.inspect", [vcs, "status", "--json"], cold, preserve=True)
            status_value = _parse_json_object(status.get("stdout"))
            branch_value = status_value.get("branch") if status_value else None
            branch = branch_value if isinstance(branch_value, str) else None
            if not branch:
                failure = "cold.branch_identity_missing"
        if failure is None:
            cold_payload = f"cold task run={run_index} bench={metadata['bench_id']}\n"
            expected_payload += cold_payload
            rows.append(edit_step(subject, scenario, "cold.edit", cold, cold_payload.strip(), run_index, meta, admitted))
            if row_returncode(agent_step("cold.checkpoint", [vcs, "commit"], cold)) != 0:
                failure = "cold.checkpoint_failed"
            elif row_returncode(
                agent_step("cold.publish", [vcs, "push", "--repo", repo_spec, "--remote", remote], cold)
            ) != 0:
                failure = "cold.publish_failed"
            rows.append(
                unmeasured_step_row(
                    subject, scenario, "cold.ci.observe", run_index, meta,
                    "owned local oak serve has no configured CI provider",
                )
            )
            rows.append(
                unmeasured_step_row(
                    subject, scenario, "cold.review_handoff", run_index, meta,
                    "owned local oak serve does not provide hosted review semantics",
                )
            )
        if failure is None and branch is not None:
            row = agent_step(
                "followup.acquire",
                [
                    vcs, "clone", repo_spec, str(followup), "--remote", remote,
                    "--branch", branch, "--allow-legacy-scope",
                ],
                run_root,
            )
            if row_returncode(row) != 0:
                failure = "followup.acquire_failed"
        if failure is None:
            agent_step("followup.inspect", [vcs, "status", "--json"], followup)
            followup_payload = f"followup task run={run_index} bench={metadata['bench_id']}\n"
            expected_payload += followup_payload
            rows.append(
                edit_step(
                    subject, scenario, "followup.edit", followup,
                    followup_payload.strip(), run_index, meta, admitted,
                )
            )
            if row_returncode(agent_step("followup.checkpoint", [vcs, "commit"], followup)) != 0:
                failure = "followup.checkpoint_failed"
            else:
                head_result = run_bounded_harness_command([vcs, "hash"], followup)
                expected_head = head_result.stdout.strip() if head_result.returncode == 0 else None
                if row_returncode(
                    agent_step(
                        "followup.publish",
                        [vcs, "push", "--repo", repo_spec, "--remote", remote],
                        followup,
                    )
                ) != 0:
                    failure = "followup.publish_failed"
            rows.append(
                unmeasured_step_row(
                    subject, scenario, "followup.ci.observe", run_index, meta,
                    "owned local oak serve has no configured CI provider",
                )
            )
            rows.append(
                unmeasured_step_row(
                    subject, scenario, "followup.review_handoff", run_index, meta,
                    "owned local oak serve does not provide hosted review semantics",
                )
            )

        payload_ok = False
        verifier_head: str | None = None
        advertised_head: str | None = None
        if failure is None and branch is not None:
            clone = run_bounded_harness_command(
                [
                    vcs, "clone", repo_spec, str(verifier), "--remote", remote,
                    "--branch", branch, "--allow-legacy-scope",
                ],
                run_root,
            )
            if clone.returncode == 0:
                payload_path = verifier / "task-notes.md"
                payload_ok = payload_path.exists() and payload_path.read_text() == expected_payload
                verifier_result = run_bounded_harness_command([vcs, "hash"], verifier)
                verifier_head = verifier_result.stdout.strip() if verifier_result.returncode == 0 else None
                advertised_head = remote_branch_head(remote, repo_spec, branch)
            else:
                failure = "verifier_clone_failed"
        head_ok = bool(expected_head and expected_head == verifier_head == advertised_head)
        if not payload_ok:
            failure = failure or "local_oak_payload_oracle_failed"
        if not head_ok:
            failure = failure or "local_oak_head_oracle_failed"
    finally:
        if server is not None:
            stop_owned_process(server)
        server_stopped = server is None or server.poll() is not None
        cleanup_started = time.perf_counter()
        for path in (seed, cold, followup, verifier, server_root):
            shutil.rmtree(path, ignore_errors=True)
        rows.append(
            harness_step_row(
                subject,
                scenario,
                "cleanup.local_server_and_checkouts",
                run_index,
                meta,
                (time.perf_counter() - cleanup_started) * 1000,
            )
        )

    total = phase_total_row(rows, subject, scenario, run_index, meta, "agent_task_oak_local.total")
    total.update(
        {
            "oracle_payload_on_remote_branch": payload_ok and head_ok,
            "oracle_payload_matches": payload_ok,
            "oracle_branch_head_matches": head_ok,
            "server_process_stopped": server_stopped,
            "acquisition_models": {
                "cold": "fresh_full_clone",
                "followup": "second_fresh_full_clone",
            },
            "repository_cache_state": {
                "cold": "fresh_destination_no_shared_object_store",
                "followup": "no_shared_object_store",
            },
            "os_cache_state": "unknown",
            "os_cache_state_reason": "page_cache_not_purged_or_instrumented",
            "network_bytes": None,
            "network_bytes_reason": "loopback transport not observed by byte proxy",
            "hosted_ci_configured": False,
            "oracle_probe_calls": 4,
            # The expected-head hash plus verifier clone/hash/HTTP head probe
            # are oracle/setup commands, not agent-facing tool calls.
            "harness_command_count": total["harness_command_count"] + 4,
        }
    )
    if failure or not server_stopped:
        total["returncode"] = 1
        total["stderr"] = failure or "local_oak_server_cleanup_failed"
    rows.append(total)
    return rows


def run_legacy_selected_materialization_campaign(
    subjects: list[Subject],
    server_bin: Path | None,
    server_source: Path | None,
    server_source_head: str | None,
    baseline_subject_name: str | None,
    run_root: Path,
    runs: int,
    metadata: dict[str, Any],
    admitted: int,
    randomize_order: bool,
    random_seed: int,
    timing_controls_requested: bool,
) -> list[dict[str, Any]]:
    """Measure one legacy selected-branch clone against one fixed Serve fixture."""
    scenario = LEGACY_SELECTED_MATERIALIZATION_SCENARIO
    oak_subjects = [subject for subject in subjects if subject.kind == "oak"]
    if not oak_subjects:
        return []
    if (
        server_bin is None
        or not server_bin.is_absolute()
        or not server_bin.is_file()
        or not os.access(server_bin, os.X_OK)
    ):
        return [
            unmeasured_step_row(
                subject,
                scenario,
                "scenario.skip",
                0,
                metadata,
                "--oak-serve-bin must name an absolute executable used only as the fixed Serve fixture",
            )
            for subject in oak_subjects
        ]
    if server_source_head is None or re.fullmatch(r"[0-9a-f]{64}", server_source_head) is None:
        return [
            unmeasured_step_row(
                subject,
                scenario,
                "scenario.skip",
                0,
                metadata,
                "the fixed Serve binary's exact 64-hex --oak-serve-source-head is required",
            )
            for subject in oak_subjects
        ]
    baseline_subject = next(
        (subject for subject in oak_subjects if subject.name == baseline_subject_name), None
    )
    if baseline_subject is None:
        return [
            unmeasured_step_row(
                subject,
                scenario,
                "scenario.skip",
                0,
                metadata,
                "--materialization-baseline-subject must name one enabled Oak subject",
            )
            for subject in oak_subjects
        ]
    git_bin = shutil.which("git")
    if git_bin is None:
        return [
            unmeasured_step_row(
                subject,
                scenario,
                "scenario.skip",
                0,
                metadata,
                "git is required only to create the deterministic main-history fixture",
            )
            for subject in oak_subjects
        ]

    server_bin = server_bin.resolve()
    resolved_server_source: Path | None = None
    if server_source is not None:
        if not server_source.is_absolute() or not server_source.is_dir():
            return [
                unmeasured_step_row(
                    subject,
                    scenario,
                    "scenario.skip",
                    0,
                    metadata,
                    "--oak-serve-source must name an absolute read-only source checkout",
                )
                for subject in oak_subjects
            ]
        resolved_server_source = server_source.resolve()
        source_head_result = run_bounded_harness_command(
            [str(server_bin), "hash"],
            resolved_server_source,
            env=isolated_oak_fixture_env(run_root / "source-check"),
        )
        if source_head_result.returncode != 0 or source_head_result.stdout.strip() != server_source_head:
            return [
                unmeasured_step_row(
                    subject,
                    scenario,
                    "scenario.skip",
                    0,
                    metadata,
                    "fixed Serve source checkout does not match --oak-serve-source-head",
                )
                for subject in oak_subjects
            ]
    baseline_binary_sha256 = binary_sha256(baseline_subject.bin)
    timing_controls_met = materialization_timing_controls_met(
        metadata, runs, randomize_order, timing_controls_requested
    )
    fixture_root = run_root / "legacy-selected-materialization-fixture"
    server_root = fixture_root / "server"
    git_seed = fixture_root / "git-seed.git"
    seed = fixture_root / "seed"
    task_config_root = fixture_root / "config"
    for path in (server_root, git_seed, task_config_root):
        path.mkdir(parents=True, exist_ok=True)
    fixture_env = isolated_oak_fixture_env(task_config_root)
    fixture_meta = {
        **metadata,
        "benchmark_track": "oak-local-loopback-materialization",
        "remote_transport": "loopback_http",
        "remote_server": "127.0.0.1",
        "serve_binary": str(server_bin),
        "serve_binary_sha256": binary_sha256(server_bin),
        "serve_source": str(resolved_server_source) if resolved_server_source else None,
        "serve_source_checkout_verification": (
            "exact_head_verified" if resolved_server_source else "unavailable_binary_hash_and_head_supplied"
        ),
        "serve_source_head": server_source_head,
        "serve_version": subject_version(Subject("serve_fixture", "oak", "Serve fixture", server_bin)),
        "baseline_subject": baseline_subject.name,
        "baseline_binary_sha256": baseline_binary_sha256,
        "fixture_id": "legacy-selected-materialization-192x16k-v1",
        "fixture_regular_files": 192,
        "fixture_regular_file_bytes": 16 * 1024,
        "fixture_empty_files": 1,
        "fixture_executable_files": 8,
        "fixture_selected_only_files": 1,
        "os_cache_state": "warm_or_unknown_after_required_fixture_probe",
        "materialization_timing_controls_met": timing_controls_met,
        "materialization_timing_controls_reason": (
            "harness_controls_met_aa_and_quiet_host_evidence_still_required"
            if timing_controls_met
            else "correctness_only_or_harness_controls_not_met"
        ),
    }
    rows: list[dict[str, Any]] = []
    setup_subject = baseline_subject

    def setup_command(operation: str, command: list[str], cwd: Path) -> bool:
        started = time.perf_counter()
        result = run_bounded_harness_command(command, cwd, env=fixture_env)
        rows.append(
            harness_step_row(
                setup_subject,
                scenario,
                operation,
                0,
                fixture_meta,
                (time.perf_counter() - started) * 1000,
                returncode=0 if result.returncode == 0 else result.returncode,
                harness_command_count=1,
            )
        )
        if result.returncode != 0:
            rows[-1]["stderr"] = result.stderr[-4000:]
        return result.returncode == 0

    server: subprocess.Popen[Any] | None = None
    proxy: CountingProxy | None = None
    failure: str | None = None
    expected_artifacts: list[str] | None = None
    baseline_cache: dict[str, Any] | None = None
    baseline_receipt_contract: dict[str, Any] | None = None
    baseline_network_contract: dict[str, Any] | None = None
    expected_status_contract: dict[str, Any] | None = None
    selected_branch = "selected-materialization"
    repo_spec = "qa/materialization-bench"
    try:
        for index in range(192):
            directory = git_seed / f"group-{index % 12:02d}"
            directory.mkdir(parents=True, exist_ok=True)
            digest = hashlib.sha256(f"oak-materialization-{index}".encode()).digest()
            content = (digest * ((16 * 1024 + len(digest) - 1) // len(digest)))[: 16 * 1024]
            path = directory / f"file-{index:03d}.bin"
            path.write_bytes(content)
            if index < 8:
                path.chmod(0o755)
        (git_seed / "empty.txt").write_bytes(b"")
        for operation, command in (
            ("setup.git_init", [git_bin, "init", "-q", "-b", "main"]),
            ("setup.git_add", [git_bin, "add", "."]),
            ("setup.git_commit", [git_bin, "commit", "-q", "-m", "fixture main"]),
        ):
            if not setup_command(operation, command, git_seed):
                failure = f"{operation}_failed"
                break

        port = reserve_loopback_port()
        direct_remote = f"http://127.0.0.1:{port}"
        if failure is None:
            server = subprocess.Popen(
                [
                    str(server_bin),
                    "serve",
                    "--dir",
                    str(server_root),
                    "--host",
                    "127.0.0.1",
                    "--port",
                    str(port),
                ],
                cwd=fixture_root,
                env=fixture_env,
                stdout=subprocess.DEVNULL,
                stderr=subprocess.DEVNULL,
                start_new_session=os.name == "posix",
            )
            if not wait_for_loopback_port(server, port):
                failure = "setup.server_not_ready"
        if failure is None:
            for operation, command, cwd in (
                ("setup.oak_import", [str(server_bin), "clone", str(git_seed), str(seed)], fixture_root),
                ("setup.branch", [str(server_bin), "switch", "-c", selected_branch], seed),
            ):
                if not setup_command(operation, command, cwd):
                    failure = f"{operation}_failed"
                    break
        if failure is None:
            (seed / "selected.txt").write_text("selected branch\n")
            for operation, command in (
                ("setup.oak_commit", [str(server_bin), "commit"]),
                (
                    "setup.oak_push",
                    [str(server_bin), "push", "--repo", repo_spec, "--remote", direct_remote],
                ),
            ):
                if not setup_command(operation, command, seed):
                    failure = f"{operation}_failed"
                    break

        expected_head_result = (
            run_bounded_harness_command([str(server_bin), "hash"], seed, env=fixture_env)
            if failure is None
            else None
        )
        expected_head = (
            expected_head_result.stdout.strip()
            if expected_head_result is not None and expected_head_result.returncode == 0
            else None
        )
        expected_history = (
            normalized_history(json_command([str(server_bin), "log", "--json"], seed, fixture_env))
            if failure is None
            else None
        )
        expected_worktree = normalized_worktree_snapshot(seed) if failure is None else None
        if not expected_head or expected_history is None or expected_worktree is None:
            failure = failure or "setup.fixture_oracle_failed"

        if failure is None:
            proxy = CountingProxy("127.0.0.1", port).start()
            measured_remote = f"http://127.0.0.1:{proxy.address[1]}"
            probe = fixture_root / "baseline-probe"
            probe_env = isolated_oak_fixture_env(fixture_root / "baseline-probe-config")
            proxy.reset_counts()
            result = run_bounded_harness_command(
                [
                    str(baseline_subject.bin),
                    "clone",
                    "--json",
                    "--allow-legacy-scope",
                    "--remote",
                    measured_remote,
                    "--branch",
                    selected_branch,
                    repo_spec,
                    str(probe),
                ],
                fixture_root,
                env=probe_env,
            )
            receipt = _parse_json_object(result.stdout)
            evidence = legacy_selected_receipt_evidence(receipt or {})
            baseline_proxy_stats = proxy.snapshot()
            baseline_cache = stat_cache_snapshot(probe)
            expected_artifacts = normalized_oak_artifacts(probe)
            baseline_receipt_contract = legacy_selected_receipt_contract(evidence)
            baseline_network_contract = {
                "connections": baseline_proxy_stats.connections,
                "receipt": baseline_receipt_contract,
            }
            expected_status_contract = normalized_status(
                json_command([str(baseline_subject.bin), "status", "--json"], probe, probe_env)
            )
            probe_checks = {
                "returncode": result.returncode == 0,
                "legacy_route": evidence["legacy_route_observed"],
                "passes_two": evidence["materialization_passes_observed"] == 2,
                "head": evidence["head"] == expected_head,
                "manifest": bool(evidence["manifest"]),
                "artifacts": expected_artifacts is not None,
                "cache": normalized_cache_contract(baseline_cache) is not None,
                "status": expected_status_contract is not None,
                "network": baseline_proxy_stats.connection_errors == 0,
            }
            if not all(probe_checks.values()):
                failed_checks = ",".join(key for key, passed in probe_checks.items() if not passed)
                failure = f"baseline_probe_failed:{failed_checks}"
                if isinstance(baseline_cache, dict) and baseline_cache.get("error"):
                    failure += f":{baseline_cache['error']}"
            shutil.rmtree(probe, ignore_errors=True)

        if failure is None and proxy is not None:
            measured_remote = f"http://127.0.0.1:{proxy.address[1]}"
            order_rng = random.Random(random_seed)
            for run_index in range(runs):
                run_subjects = list(oak_subjects)
                if randomize_order:
                    order_rng.shuffle(run_subjects)
                for subject in run_subjects:
                    destination = run_root / f"materialization-{subject.name}-r{run_index}"
                    subject_env = isolated_oak_fixture_env(
                        run_root / f"materialization-config-{subject.name}-r{run_index}"
                    )
                    proxy.reset_counts()
                    row = step_row(
                        subject,
                        scenario,
                        "legacy_selected.acquire",
                        [
                            str(subject.bin),
                            "clone",
                            "--json",
                            "--allow-legacy-scope",
                            "--remote",
                            measured_remote,
                            "--branch",
                            selected_branch,
                            repo_spec,
                            str(destination),
                        ],
                        fixture_root,
                        run_index,
                        {
                            **fixture_meta,
                            "run_subject_order": [item.name for item in run_subjects],
                        },
                        admitted,
                        preserve_stdout=True,
                        env=subject_env,
                        timeout_seconds=INTERRUPTED_PUBLICATION_PROCESS_TIMEOUT_S,
                    )
                    receipt = _parse_json_object(row.get("stdout"))
                    evidence = legacy_selected_receipt_evidence(receipt or {})
                    proxy_stats = proxy.snapshot()
                    cache_before_status = stat_cache_snapshot(destination)
                    artifacts = normalized_oak_artifacts(destination)
                    trial_meta = {
                        **fixture_meta,
                        "run_subject_order": [item.name for item in run_subjects],
                    }
                    status_row = step_row(
                        subject,
                        scenario,
                        "legacy_selected.first_status",
                        [str(subject.bin), "status", "--json"],
                        destination,
                        run_index,
                        trial_meta,
                        admitted,
                        preserve_stdout=True,
                        env=subject_env,
                        timeout_seconds=INTERRUPTED_PUBLICATION_PROCESS_TIMEOUT_S,
                    )
                    status = normalized_status(_parse_json_object(status_row.get("stdout")) or {})
                    commit_row = step_row(
                        subject,
                        scenario,
                        "legacy_selected.first_noop_commit",
                        [str(subject.bin), "commit"],
                        destination,
                        run_index,
                        trial_meta,
                        admitted,
                        env=subject_env,
                        timeout_seconds=INTERRUPTED_PUBLICATION_PROCESS_TIMEOUT_S,
                    )
                    history = normalized_history(
                        json_command([str(subject.bin), "log", "--json"], destination, subject_env)
                    )
                    head_result = run_bounded_harness_command(
                        [str(subject.bin), "hash"], destination, env=subject_env
                    )
                    local_head = head_result.stdout.strip() if head_result.returncode == 0 else None
                    worktree = normalized_worktree_snapshot(destination) if destination.exists() else None
                    advertised_head = remote_branch_head(direct_remote, repo_spec, selected_branch)
                    subject_sha256 = binary_sha256(subject.bin)
                    expected_passes = 2 if subject_sha256 == baseline_binary_sha256 else 1
                    receipt_contract = legacy_selected_receipt_contract(evidence)
                    network_contract = {
                        "connections": proxy_stats.connections,
                        "receipt": receipt_contract,
                    }
                    cache_contract = normalized_cache_contract(cache_before_status)
                    baseline_cache_contract = normalized_cache_contract(baseline_cache)
                    expected_cache_valid = subject_sha256 != baseline_binary_sha256
                    cache_valid = bool(
                        isinstance(cache_before_status, dict)
                        and cache_before_status.get("all_metadata_valid") is expected_cache_valid
                    )
                    status_ok = status == expected_status_contract
                    artifacts_ok = artifacts == expected_artifacts
                    network_ok = network_contract == baseline_network_contract
                    oracle_ok = bool(
                        row_returncode(row) == 0
                        and row_returncode(status_row) == 0
                        and row_returncode(commit_row) == 0
                        and evidence["legacy_route_observed"]
                        and evidence["snapshot_bound"] is False
                        and evidence["materialization_passes_observed"] == expected_passes
                        and evidence["head"] == expected_head == local_head == advertised_head
                        and evidence["manifest"] == baseline_receipt_contract["manifest"]
                        and evidence["written_paths"] == len(expected_worktree)
                        and receipt_contract == baseline_receipt_contract
                        and history == expected_history
                        and worktree == expected_worktree
                        and status_ok
                        and artifacts_ok
                        and cache_contract == baseline_cache_contract
                        and cache_valid
                        and network_ok
                        and proxy_stats.connection_errors == 0
                    )
                    row.update(
                        {
                            **evidence,
                            "network_connections_observed": proxy_stats.connections,
                            "network_client_to_server_bytes_observed": proxy_stats.client_to_upstream_bytes,
                            "network_server_to_client_bytes_observed": proxy_stats.upstream_to_client_bytes,
                            "network_connection_errors": proxy_stats.connection_errors,
                            "normalized_network_work": network_contract,
                            "oracle_network_work_matches_baseline": network_ok,
                            "oracle_head_matches": expected_head == local_head == advertised_head,
                            "oracle_manifest_matches": evidence["manifest"]
                            == baseline_receipt_contract["manifest"],
                            "oracle_receipt_contract_matches": receipt_contract
                            == baseline_receipt_contract,
                            "oracle_history_matches": history == expected_history,
                            "oracle_worktree_bytes_and_modes_match": worktree == expected_worktree,
                            "oracle_clean_status": status_ok,
                            "oracle_local_artifact_names_match": artifacts_ok,
                            "oracle_stat_cache_contract_matches": cache_contract
                            == baseline_cache_contract,
                            "stat_cache_metadata_valid_before_status": (
                                cache_before_status.get("all_metadata_valid")
                                if isinstance(cache_before_status, dict)
                                else None
                            ),
                            "stat_cache_rows_before_status": (
                                len(cache_before_status.get("rows", []))
                                if isinstance(cache_before_status, dict)
                                and isinstance(cache_before_status.get("rows"), list)
                                else None
                            ),
                            "stat_cache_expected_valid_before_status": expected_cache_valid,
                            "oracle_clone_residue_absent": artifacts_ok,
                            "oracle_passed": oracle_ok,
                            "repository_cache_state": "fresh_destination_no_shared_object_store",
                        }
                    )
                    for followup_row in (status_row, commit_row):
                        followup_row.update(
                            {
                                "materialization_timing_controls_met": timing_controls_met,
                                "stat_cache_expected_valid_before_first_status": expected_cache_valid,
                                "oracle_passed": oracle_ok,
                            }
                        )
                    if not oracle_ok:
                        row["returncode"] = 1
                        row["stderr"] = "legacy_selected_materialization_oracle_failed"
                        status_row["returncode"] = 1
                        commit_row["returncode"] = 1
                    total = phase_total_row(
                        [row, status_row, commit_row],
                        subject,
                        scenario,
                        run_index,
                        trial_meta,
                        "legacy_selected_clone_materialization.total",
                    )
                    total.update(
                        {
                            key: row[key]
                            for key in (
                                "materialization_passes_observed",
                                "network_connections_observed",
                                "network_client_to_server_bytes_observed",
                                "network_server_to_client_bytes_observed",
                                "oracle_passed",
                                "oracle_head_matches",
                                "oracle_manifest_matches",
                                "oracle_receipt_contract_matches",
                                "oracle_history_matches",
                                "oracle_worktree_bytes_and_modes_match",
                                "oracle_clean_status",
                                "oracle_local_artifact_names_match",
                                "oracle_stat_cache_contract_matches",
                                "stat_cache_metadata_valid_before_status",
                                "stat_cache_rows_before_status",
                                "oracle_clone_residue_absent",
                                "oracle_network_work_matches_baseline",
                                "serve_binary_sha256",
                                "serve_source_head",
                                "fixture_id",
                                "os_cache_state",
                                "materialization_timing_controls_met",
                            )
                        }
                    )
                    total.update(
                        {
                            "clone_elapsed_ms": row.get("elapsed_ms"),
                            "first_status_elapsed_ms": status_row.get("elapsed_ms"),
                            "first_noop_commit_elapsed_ms": commit_row.get("elapsed_ms"),
                        }
                    )
                    rows.extend([row, status_row, commit_row, total])
                    shutil.rmtree(destination, ignore_errors=True)
    finally:
        if proxy is not None:
            proxy.stop()
        if server is not None:
            stop_owned_process(server)
        shutil.rmtree(fixture_root, ignore_errors=True)

    if failure is not None:
        rows.append(
            harness_step_row(
                setup_subject,
                scenario,
                "scenario.fixture_failure",
                0,
                fixture_meta,
                0.0,
                returncode=1,
            )
        )
        rows[-1]["stderr"] = failure
    return rows


def parse_shallow_refresh_fixtures(text: str) -> list[tuple[int, int]]:
    """Parse ``COMMITS:SEED[,COMMITS:SEED...]``; the first is primary, the rest held out."""
    specs: list[tuple[int, int]] = []
    for item in text.split(","):
        item = item.strip()
        if not item:
            continue
        commits, sep, seed = item.partition(":")
        if not sep or not commits.isdigit() or not seed.isdigit() or int(commits) < 1:
            raise ValueError(f"invalid fixture spec {item!r}; expected COMMITS:SEED")
        specs.append((int(commits), int(seed)))
    if not specs:
        raise ValueError("at least one COMMITS:SEED fixture is required")
    return specs


def _summed_or_none(rows: list[dict[str, Any]], key: str) -> int | None:
    values = [row.get(key) for row in rows]
    if any(value is None for value in values):
        return None
    return sum(int(value) for value in values)


def run_shallow_refresh_campaign(
    subjects: list[Subject],
    server_bin: Path | None,
    server_source_head: str | None,
    fixture_specs: list[tuple[int, int]],
    run_root: Path,
    runs: int,
    metadata: dict[str, Any],
    admitted: int,
    randomize_order: bool,
    random_seed: int,
    injected_request_delay_ms: float,
) -> list[dict[str, Any]]:
    """Fresh shallow worker -> named task branch at main -> first parent refresh.

    One deterministic merge-history fixture per spec is imported into an owned
    loopback ``oak serve`` (fixed ``--oak-serve-bin``) and a bare Git repo served
    by ``GitHTTPServer``, both behind request-framing proxies, during untimed
    setup. Each subject then runs its cells in (optionally shuffled) order.
    """
    scenario = SHALLOW_REFRESH_SCENARIO
    rows: list[dict[str, Any]] = []
    if not subjects:
        return rows

    def skip_all(reason: str) -> list[dict[str, Any]]:
        return [
            unmeasured_step_row(subject, scenario, "scenario.skip", 0, metadata, reason)
            for subject in subjects
        ]

    if (
        server_bin is None
        or not server_bin.is_absolute()
        or not server_bin.is_file()
        or not os.access(server_bin, os.X_OK)
    ):
        return skip_all("--oak-serve-bin must name an absolute executable used only as the fixed Serve fixture")
    git_bin = shutil.which("git")
    if git_bin is None:
        return skip_all("git is required to create the deterministic merge-history fixture")
    server_bin = server_bin.resolve()
    serve_meta = {
        "benchmark_track": "loopback-shallow-refresh",
        "remote_transport": "loopback_http",
        "remote_server": "127.0.0.1",
        "serve_binary": str(server_bin),
        "serve_binary_sha256": binary_sha256(server_bin),
        "serve_version": subject_version(Subject("serve_fixture", "oak", "Serve fixture", server_bin)),
        "serve_source_head": server_source_head,
        "injected_request_delay_ms": injected_request_delay_ms,
        "injected_request_delay_kind": "fixed_per_request_sleep_in_proxy_not_netem",
        "hosted_equivalent": False,
        "hosted_equivalence_reason": (
            "loopback oak serve has no bounded integrity preflight (clone needs --allow-legacy-scope) "
            "and is not the hosted server; see serve_depth_behavior for whether --shallow was honored"
        ),
        "os_cache_state": "warm_or_unknown",
        "repository_cache_state": "fresh_destination_per_trial",
    }
    for commits, seed in fixture_specs:
        rows.extend(
            _run_shallow_refresh_fixture(
                subjects, server_bin, git_bin, commits, seed, run_root, runs,
                {**metadata, **serve_meta}, admitted, randomize_order, random_seed,
                injected_request_delay_ms,
            )
        )
    return rows


def _run_shallow_refresh_fixture(
    subjects: list[Subject],
    server_bin: Path,
    git_bin: str,
    commits: int,
    seed: int,
    run_root: Path,
    runs: int,
    metadata: dict[str, Any],
    admitted: int,
    randomize_order: bool,
    random_seed: int,
    injected_request_delay_ms: float,
) -> list[dict[str, Any]]:
    scenario = SHALLOW_REFRESH_SCENARIO
    fixture_root = (run_root / f"shallow-refresh-c{commits}-s{seed}").resolve()
    served_git = fixture_root / "git-served"
    for path in (fixture_root, served_git):
        path.mkdir(parents=True, exist_ok=True)
    fixture_env = isolated_oak_fixture_env(fixture_root / "config")
    rows: list[dict[str, Any]] = []
    setup_subject = subjects[0]
    fixture_meta: dict[str, Any] = {**metadata, "fixture_commits": commits, "fixture_seed": str(seed)}
    repo_spec = "qa/shallow-refresh"
    task_branch = SHALLOW_REFRESH_TASK_BRANCH
    server: subprocess.Popen[Any] | None = None
    oak_proxy: RequestCountingProxy | None = None
    git_proxy: RequestCountingProxy | None = None
    git_server: GitHTTPServer | None = None
    failure: str | None = None

    def setup_command(operation: str, command: list[str], cwd: Path) -> bool:
        started = time.perf_counter()
        result = run_bounded_harness_command(command, cwd, env=fixture_env)
        rows.append(
            harness_step_row(
                setup_subject, scenario, operation, 0, fixture_meta,
                (time.perf_counter() - started) * 1000,
                returncode=0 if result.returncode == 0 else result.returncode,
                harness_command_count=1,
            )
        )
        if result.returncode != 0:
            rows[-1]["stderr"] = result.stderr[-4000:]
        return result.returncode == 0

    try:
        manifest = merge_history_fixture.generate(
            fixture_root / "src.git", commits, seed, git_bin=git_bin, env=fixture_env
        )
        fixture_meta.update(manifest.as_dict())
        expected_tree = manifest.tree_content_hash
        bare = served_git / "fixture.git"
        if not setup_command(
            "setup.git_bare", [git_bin, "clone", "-q", "--bare", str(fixture_root / "src.git"), str(bare)],
            fixture_root,
        ):
            failure = "setup.git_bare_failed"
        port = reserve_loopback_port()
        direct_remote = f"http://127.0.0.1:{port}"
        if failure is None:
            server = subprocess.Popen(
                [str(server_bin), "serve", "--dir", str(fixture_root / "oak-served"),
                 "--host", "127.0.0.1", "--port", str(port)],
                cwd=fixture_root, env=fixture_env,
                stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL,
                start_new_session=os.name == "posix",
            )
            if not wait_for_loopback_port(server, port):
                failure = "setup.server_not_ready"
        seed_dir = fixture_root / "oak-seed"
        if failure is None:
            for operation, command, cwd in (
                ("setup.oak_import", [str(server_bin), "clone", str(fixture_root / "src.git"), str(seed_dir)], fixture_root),
                ("setup.oak_push", [str(server_bin), "push", "--repo", repo_spec, "--remote", direct_remote], seed_dir),
            ):
                if not setup_command(operation, command, cwd):
                    failure = f"{operation}_failed"
                    break
        oak_main_head = remote_branch_head(direct_remote, repo_spec, "main") if failure is None else None
        git_main = run_bounded_harness_command([git_bin, "rev-parse", "refs/heads/main"], bare, env=fixture_env)
        git_main_head = git_main.stdout.strip() if git_main.returncode == 0 else None
        import_tree = shallow_refresh.worktree_tree_hash(seed_dir) if failure is None else None
        fixture_meta.update(
            {
                "server_oak_main_head": oak_main_head,
                "server_git_main_head": git_main_head,
                "oracle_import_tree_matches_fixture": import_tree == expected_tree,
            }
        )
        if failure is None and not oak_main_head:
            failure = "setup.oak_server_main_head_unobserved"
        if failure is None and git_main_head != manifest.head:
            failure = "setup.git_server_head_mismatch"
        if failure is None and import_tree != expected_tree:
            failure = "setup.oak_import_tree_mismatch"
        shutil.rmtree(seed_dir, ignore_errors=True)

        if failure is None:
            git_server = GitHTTPServer(served_git, git_bin=git_bin).start()
            oak_proxy = RequestCountingProxy(
                "127.0.0.1", port, path_suffixes=(COMMITS_INFO_SUFFIX,),
                injected_request_delay_ms=injected_request_delay_ms,
            ).start()
            git_proxy = RequestCountingProxy(
                "127.0.0.1", git_server.server_port, path_suffixes=(COMMITS_INFO_SUFFIX,),
                injected_request_delay_ms=injected_request_delay_ms,
            ).start()
            oak_remote = f"http://127.0.0.1:{oak_proxy.address[1]}"
            git_url = f"http://127.0.0.1:{git_proxy.address[1]}/fixture.git"
            order_rng = random.Random(f"{random_seed}:{commits}:{seed}")
            for run_index in range(runs):
                run_subjects = list(subjects)
                if randomize_order:
                    order_rng.shuffle(run_subjects)
                trial_meta = {**fixture_meta, "run_subject_order": [s.name for s in run_subjects]}
                for subject in run_subjects:
                    print(
                        f"[run] {scenario} {manifest.fixture_id} run={run_index} subject={subject.name}",
                        flush=True,
                    )
                    tag = f"{subject.name}-c{commits}-s{seed}-r{run_index}"
                    subject_env = isolated_oak_fixture_env(run_root / f"sr-config-{tag}")
                    if subject.kind == "oak":
                        rows.extend(
                            _shallow_refresh_oak_trial(
                                subject, oak_proxy, oak_remote, repo_spec, oak_main_head,
                                expected_tree, run_root / f"sr-oak-{tag}", run_index, trial_meta,
                                admitted, subject_env,
                            )
                        )
                    elif subject.kind == "git":
                        rows.extend(
                            _shallow_refresh_git_trials(
                                subject, git_proxy, git_url, git_main_head, expected_tree,
                                run_root, tag, run_index, trial_meta, admitted, subject_env,
                            )
                        )
                    else:
                        rows.append(
                            unmeasured_step_row(
                                subject, scenario, "scenario.skip", run_index, trial_meta,
                                f"subject kind {subject.kind!r} has no shallow-refresh cell",
                            )
                        )
    except (OSError, RuntimeError, ValueError) as error:
        failure = failure or f"setup_exception:{type(error).__name__}:{error}"
    finally:
        for proxy in (oak_proxy, git_proxy):
            if proxy is not None:
                proxy.stop()
        if git_server is not None:
            git_server.stop()
        if server is not None:
            stop_owned_process(server)
        shutil.rmtree(fixture_root, ignore_errors=True)

    if failure is not None:
        rows.append(
            harness_step_row(setup_subject, scenario, "scenario.fixture_failure", 0, fixture_meta, 0.0, returncode=1)
        )
        rows[-1]["stderr"] = failure
    return rows


def _shallow_refresh_step(
    subject: Subject,
    proxy: RequestCountingProxy,
    operation: str,
    command: list[str],
    cwd: Path,
    run_index: int,
    meta: dict[str, Any],
    admitted: int,
    env: dict[str, str],
) -> dict[str, Any]:
    proxy.reset_counts()
    row = step_row(
        subject, SHALLOW_REFRESH_SCENARIO, operation, command, cwd, run_index, meta, admitted,
        env=env, timeout_seconds=SHALLOW_REFRESH_STEP_TIMEOUT_S,
    )
    row.update(
        shallow_refresh.request_metric_fields(proxy.request_stats(), proxy.snapshot(), COMMITS_INFO_SUFFIX)
    )
    return row


def _shallow_refresh_total(
    steps: list[dict[str, Any]],
    subject: Subject,
    run_index: int,
    meta: dict[str, Any],
    cell: str,
    oracle: dict[str, Any],
    extra: dict[str, Any],
) -> dict[str, Any]:
    total = phase_total_row(steps, subject, SHALLOW_REFRESH_SCENARIO, run_index, meta, f"{cell}.total")
    total.update(oracle)
    total.update(extra)
    for key in (
        "http_request_count",
        "http_commits_info_request_count",
        "network_wire_bytes_observed",
        "network_connections_observed",
    ):
        total[key] = _summed_or_none(steps, key)
    total["step_elapsed_ms"] = {row["operation"]: row.get("elapsed_ms") for row in steps}
    if not oracle.get("oracle_passed"):
        total["returncode"] = 1
        total["stderr"] = total.get("stderr") or f"{cell}_final_state_oracle_failed"
    for row in steps:
        row["oracle_passed"] = oracle.get("oracle_passed")
    return total


def _shallow_refresh_oak_trial(
    subject: Subject,
    proxy: RequestCountingProxy,
    remote: str,
    repo_spec: str,
    expected_head: str | None,
    expected_tree: str,
    worker: Path,
    run_index: int,
    meta: dict[str, Any],
    admitted: int,
    env: dict[str, str],
) -> list[dict[str, Any]]:
    vcs = str(subject.bin)
    cell_meta = {
        **meta,
        "cell": "shallow_clone",
        "starting_state": "no_local_repository_fresh_shallow_clone",
    }
    steps: list[dict[str, Any]] = []
    extra: dict[str, Any] = {}
    try:
        acquire = _shallow_refresh_step(
            subject, proxy, "shallow.acquire",
            [vcs, "clone", "--shallow", "--allow-legacy-scope", "--remote", remote, repo_spec, str(worker)],
            worker.parent, run_index, cell_meta, admitted, env,
        )
        steps.append(acquire)
        after_clone = shallow_refresh.oak_disk_state(worker, SHALLOW_REFRESH_TASK_BRANCH)
        depth_behavior = shallow_refresh.serve_depth_behavior(after_clone["commit_rows"])
        extra.update(
            {
                "oak_commit_rows_after_clone": after_clone["commit_rows"],
                "serve_depth_behavior": depth_behavior,
            }
        )
        if row_returncode(acquire) == 0:
            for operation, command in (
                ("shallow.branch_create", [vcs, "switch", "-c", SHALLOW_REFRESH_TASK_BRANCH]),
                ("shallow.parent_refresh", [vcs, "fetch"]),
            ):
                row = _shallow_refresh_step(
                    subject, proxy, operation, command, worker, run_index, cell_meta, admitted, env
                )
                steps.append(row)
                if row_returncode(row) != 0:
                    break
        disk = shallow_refresh.oak_disk_state(worker, SHALLOW_REFRESH_TASK_BRANCH)
        oracle = shallow_refresh.evaluate_oracle(
            expected_branch=SHALLOW_REFRESH_TASK_BRANCH,
            expected_head=expected_head,
            expected_tree_hash=expected_tree,
            observed_branch=disk["current_branch"],
            observed_branch_head=disk["branch_head"],
            observed_head=disk["head"],
            worktree_tree_hash=shallow_refresh.worktree_tree_hash(worker),
        )
        extra.update(
            {
                "oracle_source": "oak_db_query_only_plus_server_branch_api_plus_fixture_model",
                "oak_commit_rows_after_refresh": disk["commit_rows"],
                "oak_ancestry_missing_edges_after_refresh": disk["ancestry_missing_edges"],
                "oak_disk_state_error": disk["error"],
            }
        )
        for row in steps:
            row["serve_depth_behavior"] = depth_behavior
        total = _shallow_refresh_total(steps, subject, run_index, cell_meta, "shallow", oracle, extra)
        rows = [*steps, total]
        if oracle["oracle_passed"]:
            # Separate measurement: a second branch creation in the now-refreshed
            # worker. Isolates local switch cost from the first-refresh network work.
            repeat = _shallow_refresh_step(
                subject, proxy, "shallow.branch_create_repeat_local",
                [vcs, "switch", "-c", f"{SHALLOW_REFRESH_TASK_BRANCH}-repeat"],
                worker, run_index,
                {**cell_meta, "starting_state": "refreshed_worker_after_first_parent_refresh"},
                admitted, env,
            )
            repeat_disk = shallow_refresh.oak_disk_state(worker, f"{SHALLOW_REFRESH_TASK_BRANCH}-repeat")
            repeat_ok = bool(
                row_returncode(repeat) == 0
                and repeat_disk["current_branch"] == f"{SHALLOW_REFRESH_TASK_BRANCH}-repeat"
                and repeat_disk["branch_head"] == expected_head
            )
            repeat.update({"serve_depth_behavior": depth_behavior, "oracle_passed": repeat_ok})
            if not repeat_ok:
                repeat["returncode"] = 1
                repeat["stderr"] = repeat.get("stderr") or "branch_create_repeat_oracle_failed"
            rows.append(repeat)
        return rows
    finally:
        shutil.rmtree(worker, ignore_errors=True)


def _shallow_refresh_git_trials(
    subject: Subject,
    proxy: RequestCountingProxy,
    url: str,
    expected_head: str | None,
    expected_tree: str,
    run_root: Path,
    tag: str,
    run_index: int,
    meta: dict[str, Any],
    admitted: int,
    env: dict[str, str],
) -> list[dict[str, Any]]:
    git = str(subject.bin)
    rows: list[dict[str, Any]] = []

    def finish(cell: str, cell_meta: dict[str, Any], steps: list[dict[str, Any]], worktree: Path) -> None:
        disk = shallow_refresh.git_disk_state(worktree, git, env)
        oracle = shallow_refresh.evaluate_oracle(
            expected_branch=SHALLOW_REFRESH_TASK_BRANCH,
            expected_head=expected_head,
            expected_tree_hash=expected_tree,
            observed_branch=disk["current_branch"],
            observed_branch_head=disk["head"] if disk["current_branch"] == SHALLOW_REFRESH_TASK_BRANCH else None,
            observed_head=disk["head"],
            worktree_tree_hash=shallow_refresh.worktree_tree_hash(worktree),
        )
        extra = {
            "oracle_source": "git_rev_parse_plus_server_bare_repo_plus_fixture_model",
            "git_commit_rows_after_refresh": disk["commit_rows"],
            "git_is_shallow_after_refresh": disk["shallow"],
            "serve_depth_behavior": None,
        }
        rows.extend([*steps, _shallow_refresh_total(steps, subject, run_index, cell_meta, cell, oracle, extra)])

    shallow = run_root / f"sr-git-shallow-{tag}"
    shallow_meta = {**meta, "cell": "shallow_clone", "starting_state": "no_local_repository_fresh_depth1_clone"}
    try:
        steps = [
            _shallow_refresh_step(
                subject, proxy, "shallow.acquire", [git, "clone", "-q", "--depth", "1", url, str(shallow)],
                run_root, run_index, shallow_meta, admitted, env,
            )
        ]
        if row_returncode(steps[0]) == 0:
            for operation, command in (
                ("shallow.branch_create", [git, "switch", "-q", "-c", SHALLOW_REFRESH_TASK_BRANCH]),
                ("shallow.parent_refresh", [git, "fetch", "-q"]),
            ):
                steps.append(
                    _shallow_refresh_step(subject, proxy, operation, command, shallow, run_index, shallow_meta, admitted, env)
                )
                if row_returncode(steps[-1]) != 0:
                    break
        finish("shallow", shallow_meta, steps, shallow)
    finally:
        shutil.rmtree(shallow, ignore_errors=True)

    full = run_root / f"sr-git-full-{tag}"
    worktree = run_root / f"sr-git-worktree-{tag}"
    warm_meta = {
        **meta,
        "cell": "warm_full_clone_worktree",
        "starting_state": "warm_full_clone_created_untimed_not_equivalent_to_fresh_shallow_worker",
    }
    try:
        setup_started = time.perf_counter()
        clone = run_bounded_harness_command([git, "clone", "-q", url, str(full)], run_root, env=env)
        setup_row = harness_step_row(
            subject, SHALLOW_REFRESH_SCENARIO, "warm_worktree.setup_full_clone", run_index, warm_meta,
            (time.perf_counter() - setup_started) * 1000,
            returncode=0 if clone.returncode == 0 else clone.returncode, harness_command_count=1,
        )
        rows.append(setup_row)
        steps = []
        if clone.returncode == 0:
            for operation, command, cwd in (
                ("warm_worktree.branch_create",
                 [git, "worktree", "add", "-q", "-b", SHALLOW_REFRESH_TASK_BRANCH, str(worktree), "origin/main"], full),
                ("warm_worktree.parent_refresh", [git, "fetch", "-q"], worktree),
            ):
                steps.append(
                    _shallow_refresh_step(subject, proxy, operation, command, cwd, run_index, warm_meta, admitted, env)
                )
                if row_returncode(steps[-1]) != 0:
                    break
            finish("warm_worktree", warm_meta, steps, worktree)
    finally:
        shutil.rmtree(worktree, ignore_errors=True)
        shutil.rmtree(full, ignore_errors=True)
    return rows


def run_git_loop(
    subject: Subject,
    remote: oakbench_remotes.RemoteResolution,
    run_root: Path,
    run_index: int,
    metadata: dict[str, Any],
    admitted: int,
) -> list[dict[str, Any]]:
    scenario = "task_loop"
    meta = {
        **metadata,
        **remote.row_fields(),
        "benchmark_track": (
            "core-equivalent"
            if remote.transport == oakbench_remotes.TRANSPORT_LOCAL_FILE
            else "agent-default"
        ),
    }
    vcs = str(subject.bin)
    rows: list[dict[str, Any]] = []

    seed_error = ensure_git_remote_seeded(subject, remote, run_root)
    if seed_error:
        return [skip_row(subject, scenario, run_index, meta, seed_error)]

    # Untimed: the base clone is the long-lived checkout worktrees hang off;
    # an agent fleet pays it once, not per task.
    base = run_root / f"tl-git-base-r{run_index}"
    clone = run_untimed([vcs, "clone", "-q", str(remote.repo), str(base)], run_root)
    if clone.returncode != 0:
        return [skip_row(subject, scenario, run_index, meta, f"base clone failed: {clone.stderr.strip()[:200]}")]

    branches: dict[str, str] = {}
    payloads: dict[str, str] = {}
    for phase in PHASES:
        workspace = run_root / f"tl-{subject.name}-r{run_index}-{phase}"
        branch = oakbench_remotes.disposable_branch(
            str(metadata["bench_id"]), f"taskloop-{phase}", subject.name, run_index
        )
        branches[phase] = branch
        payload = f"payload {phase} run={run_index} bench={metadata['bench_id']}"
        payloads[phase] = payload
        acquire = step_row(
            subject, scenario, f"{phase}.acquire",
            [vcs, "worktree", "add", str(workspace), "-b", branch], base, run_index, meta, admitted,
        )
        rows.append(acquire)
        if int(acquire["returncode"]) != 0:
            return rows
        rows.append(edit_step(subject, scenario, f"{phase}.edit", workspace, payload, run_index, meta, admitted))
        # Agent-default snapshot: git pays stage+commit as two calls.
        rows.append(step_row(subject, scenario, f"{phase}.snapshot.add", [vcs, "add", "."], workspace, run_index, meta, admitted))
        rows.append(
            step_row(
                subject, scenario, f"{phase}.snapshot.commit",
                [vcs, "commit", "-q", "-m", f"bench task loop {metadata['bench_id']} r{run_index} {phase}"],
                workspace, run_index, meta, admitted,
            )
        )
        rows.append(
            step_row(
                subject, scenario, f"{phase}.publish",
                [vcs, "push", "-q", "-u", "origin", branch], workspace, run_index, meta, admitted,
            )
        )
        # No task.describe: git's description is the commit message already
        # paid in {phase}.snapshot.commit (asymmetry documented in the lane doc).
        rows.append(
            step_row(
                subject, scenario, f"{phase}.end",
                [vcs, "worktree", "remove", str(workspace)], base, run_index, meta, admitted,
            )
        )

    # Untimed oracle: payloads visible on the remote branches.
    oracle_ok: bool | None = True
    for phase in PHASES:
        fetch = run_untimed([vcs, "fetch", "-q", "origin", branches[phase]], base)
        show = run_untimed([vcs, "show", f"FETCH_HEAD:task-notes.md"], base)
        if fetch.returncode != 0 or payloads[phase] not in show.stdout:
            oracle_ok = False
    rows.append(
        {
            **phase_total_row(rows, subject, scenario, run_index, meta, "task_loop.total"),
            "oracle_payload_on_remote_branch": oracle_ok,
            "workspace_branches": branches,
            "measurement_source": "fetch_plus_show_grep",
        }
    )
    return rows


def loop_summary_fields(rows: list[dict[str, Any]]) -> dict[str, Any]:
    """followup-vs-task warm deltas computed from this run's step rows."""
    def op_ms(operation: str) -> float | None:
        for row in rows:
            if row.get("operation") == operation:
                return float(row.get("elapsed_ms", 0.0))
        return None

    def phase_ms(prefix: str) -> float:
        return sum(
            float(row.get("elapsed_ms", 0.0))
            for row in rows
            if str(row.get("operation", "")).startswith(prefix + ".")
        )

    task_acquire = op_ms("task.acquire")
    followup_acquire = op_ms("followup.acquire")
    fields: dict[str, Any] = {
        "task_phase_ms": round(phase_ms("task"), 3),
        "followup_phase_ms": round(phase_ms("followup"), 3),
    }
    if task_acquire is not None and followup_acquire is not None:
        fields["followup_warm_delta_ms"] = round(followup_acquire - task_acquire, 3)
        fields["acquire_ms_task"] = round(task_acquire, 3)
        fields["acquire_ms_followup"] = round(followup_acquire, 3)
    else:
        fields["followup_warm_delta_ms"] = None
    return fields


def summary_text(rows: list[dict[str, Any]]) -> str:
    totals = [row for row in rows if row.get("operation") == "task_loop.total"]
    lines = [
        "# Task Loop Summary",
        "",
        "Whole-task cost of the signature motion: acquire workspace -> edit -> snapshot ->",
        "publish -> describe -> end, then a follow-up task in a fresh workspace.",
        "Command sequences are NOT step-equivalent across subjects",
        "(oak pays desc; git's description is its commit message). Latency across unlike",
        "transports is never comparable; scripted call equivalents and estimated token proxies are.",
        "",
        "| Subject | Run | Transport | Wall ms | Scripted call equivalents | Cost-wt tokens (env) | Warm acquire delta ms | Oracle |",
        "| --- | ---: | --- | ---: | ---: | ---: | ---: | --- |",
    ]
    for row in totals:
        oracle = row.get("oracle_payload_on_remote_branch")
        lines.append(
            "| `{}` | {} | {} | {} | {} | {} | {} | {} |".format(
                row["subject"],
                row["run"],
                row.get("remote_transport", "unmeasured"),
                fmt_ms(row.get("elapsed_ms")),
                row.get("tool_call_count", "unmeasured"),
                row.get("estimated_cost_weighted_tokens_with_envelope", "unmeasured"),
                row.get("followup_warm_delta_ms", "unmeasured"),
                "pass" if oracle else ("FAIL" if oracle is False else "unmeasured"),
            )
        )
    skips = [
        row for row in rows
        if row.get("skipped") and row.get("scenario") == TASK_LOOP_SCENARIO
    ]
    for row in skips:
        lines.append(f"| `{row['subject']}` | {row['run']} | skipped | | | | | {row['skip_reason'][:70]} |")

    agent_totals = [
        row
        for row in rows
        if row.get("operation") in ("agent_task.total", "agent_task_oak_local.total")
    ]
    if agent_totals:
        lines.extend(
            [
                "",
                "## Cold clone and warm isolated task",
                "",
                "Local-file Git rows compare a fresh full clone with a worktree sharing",
                "the first clone's object store. Oak local rows use two independent fresh",
                "clones against an owned loopback server; they do not claim a shared cache.",
                "OS cache and network bytes remain unknown. Hosted CI/review are skips.",
                "",
                "| Subject | Run | Wall ms | Scripted call equivalents | Output bytes | Coverage | Oracle |",
                "| --- | ---: | ---: | ---: | ---: | --- | --- |",
            ]
        )
        for row in agent_totals:
            lines.append(
                "| `{}` | {} | {} | {} | {} | {} | {} |".format(
                    row["subject"],
                    row["run"],
                    fmt_ms(row.get("elapsed_ms")),
                    row.get("tool_call_count", "unmeasured"),
                    row.get("raw_output_bytes", "unmeasured"),
                    "complete" if row.get("coverage_complete") else "partial",
                    "pass" if row.get("oracle_payload_on_remote_branch") else "FAIL",
                )
            )

    materialization_totals = [
        row
        for row in rows
        if row.get("operation") == "legacy_selected_clone_materialization.total"
    ]
    if materialization_totals:
        lines.extend(
            [
                "",
                "## Legacy selected-branch materialization",
                "",
                "One fixed loopback Serve and one immutable fixture are shared across subjects.",
                "Fresh destinations do not imply a cold OS page cache. Timing claims require",
                "separate interleaved A/A noise and A/B campaigns.",
                "",
                "| Subject | Run | Clone ms | First status ms | First no-op commit ms | Passes | Harness controls met | Oracle |",
                "| --- | ---: | ---: | ---: | ---: | ---: | --- | --- |",
            ]
        )
        for row in materialization_totals:
            lines.append(
                "| `{}` | {} | {} | {} | {} | {} | {} | {} |".format(
                    row["subject"],
                    row["run"],
                    fmt_ms(row.get("clone_elapsed_ms")),
                    fmt_ms(row.get("first_status_elapsed_ms")),
                    fmt_ms(row.get("first_noop_commit_elapsed_ms")),
                    row.get("materialization_passes_observed", "unmeasured"),
                    "yes" if row.get("materialization_timing_controls_met") else "no",
                    "pass" if row.get("oracle_passed") else "FAIL",
                )
            )

    interrupted_totals = [
        row for row in rows if row.get("operation") == "interrupted_publication.total"
    ]
    if interrupted_totals:
        lines.extend(
            [
                "",
                "## Interrupted publication",
                "",
                "A loopback fixture validates the request framing and drops the response.",
                "Fixture delivery knowledge is reported separately from the CLI's own outcome",
                "claim. Request receipt is not durability proof; the harness never retries.",
                "",
                "| Subject | Run | Wall ms | Scripted call equivalents | Output bytes | CLI outcome | Fixture delivery | POSTs | Request bytes | Head/work preserved |",
                "| --- | ---: | ---: | ---: | ---: | --- | --- | ---: | ---: | --- |",
            ]
        )
        for row in interrupted_totals:
            lines.append(
                "| `{}` | {} | {} | {} | {} | `{}` | `{}` | {} | {} | {}/{} |".format(
                    row["subject"],
                    row["run"],
                    fmt_ms(row.get("elapsed_ms")),
                    row.get("tool_call_count", "unmeasured"),
                    row.get("raw_output_bytes", "unmeasured"),
                    row.get("publication_outcome", "unmeasured"),
                    row.get("fixture_delivery_state", "unmeasured"),
                    row.get("mutation_requests_observed", "unmeasured"),
                    row.get("network_request_bytes_observed", "unmeasured"),
                    "yes" if row.get("local_head_preserved") else "no",
                    "yes" if row.get("local_work_preserved") else "no",
                )
            )
    return "\n".join(lines) + "\n"


def fmt_ms(value: Any) -> str:
    try:
        return f"{float(value):.0f}"
    except (TypeError, ValueError):
        return "unmeasured"


def main() -> int:
    args = parse_args()
    scenarios = [item.strip() for item in args.scenarios.split(",") if item.strip()]
    known_scenarios = {
        TASK_LOOP_SCENARIO,
        AGENT_TASK_SCENARIO,
        AGENT_TASK_OAK_LOCAL_SCENARIO,
        LEGACY_SELECTED_MATERIALIZATION_SCENARIO,
        SHALLOW_REFRESH_SCENARIO,
        INTERRUPTED_PUBLICATION_SCENARIO,
    }
    unknown_scenarios = sorted(set(scenarios) - known_scenarios)
    if unknown_scenarios:
        raise SystemExit(
            "Unknown scenarios: " + ", ".join(unknown_scenarios)
            + "; valid: " + ", ".join(sorted(known_scenarios))
        )
    subjects = load_subjects(args)
    timestamp = datetime.now(timezone.utc).strftime("%Y%m%dT%H%M%SZ")
    args.results.mkdir(parents=True, exist_ok=True)
    args.workdir.mkdir(parents=True, exist_ok=True)
    run_root = args.workdir / "runs" / timestamp
    run_root.mkdir(parents=True, exist_ok=True)

    metadata = {
        "bench_id": timestamp,
        "profile": "task-loop",
        "benchmark_track": "agent-default",
        "timestamp_utc": timestamp,
        "host": platform.node(),
        "platform": platform.platform(),
        "machine": platform.machine(),
        "python": platform.python_version(),
        "env_isolation_version": ENV_ISOLATION_VERSION,
        "subject_versions": subject_versions(subjects),
        "subject_details": subject_details(subjects),
        "source": source_metadata(args.oak_repo),
        "git_remote_mode": args.git_remote,
        "trial_order_randomized": args.randomize_subject_order,
        "trial_order_seed": args.random_seed if args.randomize_subject_order else None,
    }

    rows: list[dict[str, Any]] = []
    with measurement_lock("task_loop") as lock_info:
        metadata["measurement_lock_wait_ms"] = lock_info.wait_ms
        metadata["measurement_lock"] = "held" if lock_info.enabled else "disabled"
        order_rng = random.Random(args.random_seed)
        if LEGACY_SELECTED_MATERIALIZATION_SCENARIO in scenarios:
            rows.extend(
                run_legacy_selected_materialization_campaign(
                    subjects,
                    args.oak_serve_bin,
                    args.oak_serve_source,
                    args.oak_serve_source_head,
                    args.materialization_baseline_subject,
                    run_root,
                    args.runs,
                    metadata,
                    args.admitted_output_chars,
                    args.randomize_subject_order,
                    args.random_seed,
                    args.materialization_timing_controls_met,
                )
            )
        if SHALLOW_REFRESH_SCENARIO in scenarios:
            rows.extend(
                run_shallow_refresh_campaign(
                    subjects,
                    args.oak_serve_bin,
                    args.oak_serve_source_head,
                    parse_shallow_refresh_fixtures(args.shallow_refresh_fixtures),
                    run_root,
                    args.runs,
                    metadata,
                    args.admitted_output_chars,
                    args.randomize_subject_order,
                    args.random_seed,
                    args.injected_request_delay_ms,
                )
            )
        campaign_scenarios = {LEGACY_SELECTED_MATERIALIZATION_SCENARIO, SHALLOW_REFRESH_SCENARIO}
        ordinary_scenarios = [scenario for scenario in scenarios if scenario not in campaign_scenarios]
        for run_index in range(args.runs):
            run_subjects = list(subjects)
            if args.randomize_subject_order:
                order_rng.shuffle(run_subjects)
            run_metadata = {
                **metadata,
                "run_subject_order": [subject.name for subject in run_subjects],
            }
            for subject in run_subjects:
                for scenario in ordinary_scenarios:
                    print(f"[run] {scenario} run={run_index} subject={subject.name}", flush=True)
                    if scenario == INTERRUPTED_PUBLICATION_SCENARIO:
                        rows.extend(
                            run_interrupted_publication(
                                subject, run_root, run_index, run_metadata, args.admitted_output_chars
                            )
                        )
                        continue
                    if scenario == AGENT_TASK_SCENARIO:
                        if subject.kind != "git":
                            rows.append(
                                unmeasured_step_row(
                                    subject,
                                    scenario,
                                    "agent_task.skipped",
                                    run_index,
                                    run_metadata,
                                    "Oak end-to-end publication and exact CI require an explicitly configured disposable staging lane; production mutation is forbidden",
                                )
                            )
                            continue
                        remote_create_start = time.perf_counter()
                        remote = oakbench_remotes.make_git_bare_remote(
                            subject.bin, run_root, f"agent-task-remote-r{run_index}.git"
                        )
                        remote_create_ms = (time.perf_counter() - remote_create_start) * 1000
                        rows.extend(
                            run_git_agent_task_baseline(
                                subject,
                                remote,
                                run_root,
                                run_index,
                                run_metadata,
                                args.admitted_output_chars,
                                remote_create_ms,
                            )
                        )
                        continue
                    if scenario == AGENT_TASK_OAK_LOCAL_SCENARIO:
                        if subject.kind != "oak":
                            rows.append(
                                unmeasured_step_row(
                                    subject,
                                    scenario,
                                    "agent_task_oak_local.skipped",
                                    run_index,
                                    run_metadata,
                                    "owned local oak serve scenario applies only to Oak subjects",
                                )
                            )
                            continue
                        rows.extend(
                            run_oak_local_agent_task_baseline(
                                subject,
                                run_root,
                                run_index,
                                run_metadata,
                                args.admitted_output_chars,
                            )
                        )
                        continue
                    if subject.kind == "git":
                        if args.git_remote == "github":
                            remote = oakbench_remotes.resolve_git_github_remote()
                        else:
                            remote = oakbench_remotes.make_git_bare_remote(
                                subject.bin, run_root, f"task-loop-remote-r{run_index}.git"
                            )
                        if not remote.resolved:
                            rows.append(skip_row(subject, TASK_LOOP_SCENARIO, run_index, run_metadata, remote.skip_reason or ""))
                            continue
                        subject_rows = run_git_loop(subject, remote, run_root, run_index, run_metadata, args.admitted_output_chars)
                    else:
                        remote = oakbench_remotes.resolve_oak_remote("sync")
                        if not remote.resolved:
                            rows.append(skip_row(subject, TASK_LOOP_SCENARIO, run_index, run_metadata, remote.skip_reason or ""))
                            continue
                        subject_rows = run_oak_loop(subject, remote, run_root, run_index, run_metadata, args.admitted_output_chars)
                    total = next((row for row in subject_rows if row.get("operation") == "task_loop.total"), None)
                    if total is not None:
                        total.update(loop_summary_fields(subject_rows))
                    rows.extend(subject_rows)

    store = ResultsStore(args.results, lane="task-loop")
    raw_path, summary_path = store.write(timestamp, rows, summary_text(rows))
    if not args.keep_workdirs:
        shutil.rmtree(run_root, ignore_errors=True)
    print(f"[result] {raw_path}")
    print(f"[summary] {summary_path}")
    return 1 if has_lane_failures(rows) else 0


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

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

RunnersDetails

No runs on main yet.
Open branchesHistory

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

May 2026now
1	`#!/usr/bin/env python3`
2	`"""The full task loop, end to end — the product's signature motion.`
3
4	`oak: mount -> edit -> commit -> push -> desc -> end, then REMOUNT for a`
5	`follow-up task (the warm-cache story) and repeat edit/commit/push/end.`
6	`git: worktree add -> branch edit -> add/commit -> push -> worktree remove,`
7	`then a second worktree for the follow-up. The task description lives in`
8	the commit message git already paid for; oak pays a separate `desc`
9	`call. That asymmetry is the measurement, not a harness gap`
10	`(docs/command-semantics.md): this is agent-default-track, whole-task`
11	`evidence, never a step-by-step equivalence claim.`
12
13	`Transport honesty: oak rows always cross the network to a real oak server.`
14	`git runs against GitHub (--git-remote github, transport-comparable) or a`
15	`local bare repo (--git-remote local, the core-equivalent track). Rows carry`
16	`remote_transport/remote_server; latency comparisons across transports are`
17	`forbidden — cross-subject evidence at unlike transports is tool calls and`
18	`tokens only.`
19
20	`Setup:`
21	`OAK_BENCH_SYNC_REPO=oak/bench-sync-tmp \\`
22	`[email protected]:oakdotspace/bench-task-loop-tmp.git \\`
23	`python3 scripts/task_loop.py --runs 3`
24
25	`Unconfigured remotes emit skip rows (returncode 77), never silent absence.`
26	`"""`
27
28	`from __future__ import annotations`
29
30	`import argparse`
31	`import hashlib`
32	`import json`
33	`import os`
34	`import platform`
35	`import random`
36	`import re`
37	`import shutil`
38	`import socket`
39	`import sqlite3`
40	`import stat`