#!/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.byteproxy import CountingProxy

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"

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("--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, "

            "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,

                )