scripts/task_loop.py · oak/benchmarks

Viewing files at commit dc75e12cb299Commit details View latest
Browse files
scripts/task_loop.py 1264 lines · 48.8 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 json
import platform
import random
import re
import shutil
import socket
import subprocess
import tempfile
import threading
import time
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
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"
INTERRUPTED_PUBLICATION_SCENARIO = "interrupted_publication_after_request_v1"


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, 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,
    env: dict[str, str] | None = None,
) -> dict[str, Any]:
    capture = run_timed(command, cwd, env or base_env(), admitted_output_chars)
    expected = 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,
        "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
    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,
    local_head_before: str | None,
    local_head_after: str | None,
    status_after: dict[str, Any] | None,
    request_complete_before_disconnect: bool,
) -> 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.
    """
    return {
        "publication_outcome": "unknown" if request_complete_before_disconnect else "not_sent",
        "fixture_received_complete_request_before_disconnect": request_complete_before_disconnect,
        "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,
        "local_work_preserved": bool(status_after and status_after.get("needs_push")),
        "working_tree_dirty_after": None if status_after is None else status_after.get("dirty"),
    }


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
        size = int(self.headers.get("Content-Length", "0"))
        body = self.rfile.read(size)
        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.request_complete_before_disconnect = True  # type: ignore[attr-defined]
            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.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
    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")
        rows.append(
            step_row(
                subject,
                scenario,
                "publication.attempt",
                push_command,
                workspace,
                run_index,
                metadata,
                admitted,
                expected_returncodes=tuple(range(1, 128)),
                env=fixture_env,
            )
        )
    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
    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]
        request_complete = bool(server.request_complete_before_disconnect)  # type: ignore[attr-defined]
    evidence = interrupted_publication_fields(
        request_count=request_count,
        request_bytes=request_bytes if request_count else None,
        local_head_before=head_before,
        local_head_after=head_after,
        status_after=status_after,
        request_complete_before_disconnect=request_complete,
    )
    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 (
        request_complete
        and request_count == 1
        and evidence["local_head_preserved"]
        and evidence["local_work_preserved"]
        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 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; tool calls and tokens are.",
        "",
        "| Subject | Run | Transport | Wall ms | Tool calls | 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") == "agent_task.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. OS cache state and network bytes remain",
                "unknown. Hosted CI is a skip, not a local-oracle success.",
                "",
                "| Subject | Run | Wall ms | Calls | 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 reads one complete ordinary Oak push request and drops",
                "the response. `unknown` is the client-side outcome; fixture request receipt",
                "is not a durability receipt and the harness never retries the mutation.",
                "",
                "| Subject | Run | Wall ms | Calls | Output bytes | Outcome | 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("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, 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 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 json`
32	`import platform`
33	`import random`
34	`import re`
35	`import shutil`
36	`import socket`
37	`import subprocess`
38	`import tempfile`
39	`import threading`
40	`import time`