scripts/task_loop.py · oak/benchmarks

Viewing files at commit 13de5e5480b3Commit details View latest
Browse files
scripts/task_loop.py 1768 lines · 69.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 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.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,
    load_subjects,
    source_metadata,
    subject_details,
    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"
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 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("--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, 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
) -> subprocess.CompletedProcess[str]:
    capture = run_timed(
        command,
        cwd,
        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_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",
                )
            )

    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,
        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)
        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 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 subprocess`
40	`import tempfile`