#!/usr/bin/env python3

"""netshape lane: git vs oak network operations through the same shaped pipe.

Both subjects are served from local disk through one Linux netns + veth +

tc/netem pipe (profiles from config/netshapes.json), which is what makes a

cross-subject network comparison legitimate. Before any campaign rows, a

self-test row asserts the pipe's achieved RTT is within +/-15% of the

configured profile — the pipe is an instrument, and it is tested like one.

Skip-row honesty (ADR-0002):

- On hosts that cannot shape packets (macOS, missing ip/tc, no

  CAP_NET_ADMIN), every requested run emits a returncode-77 skip row with

  reason "netshape_unavailable:<reason>" and the process exits 3. Never an

  error: an unshaped host is a coverage gap, not a crash.

- The oak subject needs OAK_BENCH_LOCAL_SERVER_CMD (a self-hostable oak

  server is a flagged product dependency). Absent, oak runs emit skip rows

  with reason "oak_local_server_missing". The skip is the pressure: the row

  stream itself records that oak cannot yet be benchmarked on a local shaped

  pipe.

Transport honesty: every row carries remote_transport "network_shaped" plus

netshape_profile; shaped rows are comparable only same-profile/same-server

(oakbench.remotes.shaped_comparison_legal).

Bytes on wire: the gitserver/oak server sit behind oakbench.byteproxy's

CountingProxy bound to the shaped pipe's host address, so measured rows carry

wire byte counts. Null means unmeasured.

"""

from __future__ import annotations

import argparse

import json

import os

import platform as platform_module

import shlex

import shutil

import socket

import subprocess

import sys

import tempfile

import time

from datetime import datetime, timezone

from pathlib import Path

from typing import Any, Callable, Optional

from oakbench import netshape as oakbench_netshape

from oakbench import remotes as oakbench_remotes

from oakbench.byteproxy import CountingProxy

from oakbench.environment import ENV_ISOLATION_VERSION, base_env as oakbench_base_env

from oakbench.execution import run_timed

from oakbench.gitserver import GitHTTPServer

from oakbench.rows import RowContractViolation, validate_rows

ROOT = Path(__file__).resolve().parents[1]

EXIT_OK = 0

EXIT_FAILURES = 1

EXIT_USAGE = 2

EXIT_ALL_SKIPPED = 3

SKIP_RETURNCODE = 77

SCENARIOS = (

    "netshape_clone_cold",

    "netshape_pull_uptodate",

    "netshape_push_small_delta",

    "netshape_fetch_negotiation_refs100k",

    # _cold is part of the identity (ADR-0005): the measured unit is a cold

    # network acquire plus the first file-history query, like

    # netshape_clone_cold encodes its cache state in the name.

    "netshape_log_follow_cold",

    "netshape_mount_cold",

    "netshape_task_loop",

)

SUBJECT_KINDS = {"git": "git", "oak": "oak"}

SERVER_NAME = "netshape-local"

OAK_LOCAL_SERVER_ENV = "OAK_BENCH_LOCAL_SERVER_CMD"

OAK_SERVED_REPO = "bench"

TASK_LOOP_ITERATIONS = 3

EPILOG = f"""\

oak subject: a self-hostable oak server is a flagged product dependency. Set

{OAK_LOCAL_SERVER_ENV} to an argv command template ({{host}} {{port}} {{root}}

placeholders) that serves a repo named '{OAK_SERVED_REPO}' from {{root}} at

http://{{host}}:{{port}}. When it is absent, oak runs emit skip rows with

reason "oak_local_server_missing" — the skip is the pressure: the row stream

records that oak cannot yet be benchmarked on a local shaped pipe.

This lane needs Linux + ip/tc + CAP_NET_ADMIN (run as root). Anywhere else it

emits skip rows (reason "netshape_unavailable:<reason>") and exits 3.

"""

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

    return oakbench_base_env(

        author_name="Oak Netshape",

        author_email="[email protected]",

        oak_author="oak-netshape",

    )

def local_server_argv(template: str, *, host: str, port: int, root: Path) -> list[str]:

    rendered = template.format(host=host, port=port, root=str(root))

    argv = shlex.split(rendered)

    if not argv:

        raise ValueError(f"{OAK_LOCAL_SERVER_ENV} rendered to an empty command")

    return argv

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

    parser = argparse.ArgumentParser(

        description=__doc__,

        epilog=EPILOG,

        formatter_class=argparse.RawDescriptionHelpFormatter,

    )

    parser.add_argument("--profile", default="lan-1ms", help="netshape profile id from config/netshapes.json")

    parser.add_argument("--scenario", choices=SCENARIOS, default="netshape_clone_cold")

    parser.add_argument("--subjects", default="git,oak", help="comma-separated: git,oak")

    parser.add_argument("--results", type=Path, default=ROOT / "results" / "netshape",

                        help="results directory, or an explicit .jsonl file path")

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

    parser.add_argument("--netshapes", type=Path, default=oakbench_netshape.DEFAULT_NETSHAPES)

    parser.add_argument("--refs", type=int, default=100_000,

                        help="ref count seeded for the fetch-negotiation scenario")

    parser.add_argument("--git-bin", default="git")

    parser.add_argument("--oak-bin", default="oak")

    parser.add_argument("--admitted-output-chars", type=int, default=20_000)

    return parser.parse_args(argv)

def metadata(bench_id: str, args: argparse.Namespace) -> dict[str, Any]:

    return {

        "bench_id": bench_id,

        "profile": "netshape",

        "timestamp_utc": bench_id,

        "host": platform_module.node(),

        "platform": platform_module.platform(),

        "machine": platform_module.machine(),

        "python": platform_module.python_version(),

        "env_isolation_version": ENV_ISOLATION_VERSION,

        "netshape_profile": args.profile,

        "remote_transport": oakbench_remotes.TRANSPORT_NETWORK_SHAPED,

        "remote_server": SERVER_NAME,

    }

def skip_row(

    meta: dict[str, Any],

    scenario: str,

    subject: str,

    run_index: int,

    reason: str,

    operation: str = "scenario.skip",

) -> dict[str, Any]:

    return {

        **meta,

        "scenario": scenario,

        "operation": operation,

        "run": run_index,

        "subject": subject,

        "subject_kind": SUBJECT_KINDS.get(subject, subject),

        "elapsed_ms": 0.0,

        "returncode": SKIP_RETURNCODE,

        "command": [],

        "skipped": True,

        "skip_reason": reason,

    }

def write_rows(results_arg: Path, bench_id: str, rows: list[dict[str, Any]]) -> Path:

    """Write JSONL, then validate against the shared row contract.

    rows.py owns no 'netshape' lane yet and this slice may not modify it, so

    netshape rows opt into the core key contract (a strict superset of what

    they carry) to keep skip-row coherence machine-checked.

    """

    if results_arg.suffix == ".jsonl":

        path = results_arg

    else:

        path = results_arg / f"netshape-{bench_id}.jsonl"

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

    with path.open("w") as fh:

        for row in rows:

            fh.write(json.dumps(row, sort_keys=True) + "\n")

    errors = validate_rows(rows, "core")

    if errors:

        preview = "\n  ".join(errors[:20])

        raise RowContractViolation(

            f"Row contract violation (netshape lane). Rows were written to {path} "

            f"but the emitter has a bug:\n  {preview}"

        )

    return path

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

# Linux-with-privileges measurement path. Untestable on this macOS dev host

# by design (skip-row honesty above); kept simple and obviously correct.

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

def run_untimed(command: list[str], cwd: Optional[Path] = None, timeout: float = 600.0) -> subprocess.CompletedProcess:

    return subprocess.run(

        command,

        cwd=cwd,

        env=base_env(),

        stdout=subprocess.PIPE,

        stderr=subprocess.PIPE,

        text=True,

        timeout=timeout,

        check=False,

    )

def timed_row(

    meta: dict[str, Any],

    args: argparse.Namespace,

    subject: str,

    run_index: int,

    operation: str,

    command: list[str],

    cwd: Path,

    proxy: Optional[CountingProxy],

    extra: Optional[dict[str, Any]] = None,

) -> dict[str, Any]:

    if proxy is not None:

        proxy.reset_counts()

    capture = run_timed(command, cwd, base_env(), args.admitted_output_chars)

    row: dict[str, Any] = {

        **meta,

        "scenario": args.scenario,

        "operation": operation,

        "run": run_index,

        "subject": subject,

        "subject_kind": SUBJECT_KINDS.get(subject, subject),

        "elapsed_ms": round(capture.elapsed_ms, 3),

        "returncode": capture.returncode,

        "command": command,

        "stdout_bytes": capture.stdout_bytes,

        "stderr_bytes": capture.stderr_bytes,

        # Null means unmeasured (ADR-0002): byte counts and first-payload

        # timing exist only when the operation went through the counting

        # proxy, and the keys are always present so a missing measurement is

        # an explicit null, never an absent key.

        "wire_bytes_client_to_server": None,

        "wire_bytes_server_to_client": None,

        "wire_bytes_total": None,

        "wire_connections": None,

        "ttfd_ms": None,

        "ttfd_source": None,

        "first_payload_bytes_server_to_client": None,

    }

    if capture.returncode != 0 and capture.stderr_text:

        row["stderr"] = capture.stderr_text[-4000:]

    if proxy is not None:

        stats = proxy.snapshot()

        row.update(

            {

                "wire_bytes_client_to_server": stats.client_to_upstream_bytes,

                "wire_bytes_server_to_client": stats.upstream_to_client_bytes,

                "wire_bytes_total": stats.total_bytes,

                "wire_connections": stats.connections,

                "ttfd_ms": stats.first_upstream_to_client_ms,

                "ttfd_source": (

                    "byteproxy_first_upstream_to_client_byte"

                    if stats.first_upstream_to_client_ms is not None

                    else None

                ),

                "first_payload_bytes_server_to_client": stats.first_payload_bytes_server_to_client,

            }

        )

    if extra:

        row.update(extra)

    return row

def selftest_row(meta: dict[str, Any], args: argparse.Namespace, selftest: dict[str, Any]) -> dict[str, Any]:

    return {

        **meta,

        **selftest,

        "scenario": args.scenario,

        "run": 0,

        "subject": "harness",

        "subject_kind": "harness",

        "elapsed_ms": float(selftest["measured_rtt_ms"]),

        "returncode": 0 if selftest["within_tolerance"] else 1,

        "command": ["netshape:selftest", args.profile],

    }

def make_served_git_repo(

    git_bin: str,

    workdir: Path,

    served_root: Path,

    refs_count: int,

    history_commits: int = 1,

) -> Path:

    """Untimed precondition: a small source repo, served as a bare repo with

    http.receivepack enabled; optionally seeded with many refs."""

    src = workdir / "git-src"

    run_untimed([git_bin, "-c", "init.defaultBranch=main", "init", str(src)])

    (src / "README.md").write_text("# netshape fixture\n")

    (src / "src").mkdir(exist_ok=True)

    (src / "src" / "main.py").write_text("print('netshape')\n")

    run_untimed([git_bin, "add", "-A"], cwd=src)

    run_untimed([git_bin, "commit", "-m", "netshape fixture"], cwd=src)

    for index in range(1, max(1, history_commits)):

        with (src / "src" / "main.py").open("a") as fh:

            fh.write(f"print('netshape history {index}')\n")

        run_untimed([git_bin, "add", "src/main.py"], cwd=src)

        run_untimed([git_bin, "commit", "-m", f"history {index}"], cwd=src)

    bare = served_root / "repo.git"

    run_untimed([git_bin, "clone", "--bare", str(src), str(bare)])

    run_untimed([git_bin, "-C", str(bare), "config", "http.receivepack", "true"])

    if refs_count > 0:

        head = run_untimed([git_bin, "-C", str(bare), "rev-parse", "HEAD"]).stdout.strip()

        lines = "".join(f"create refs/bench/r{i} {head}\n" for i in range(refs_count))

        subprocess.run(

            [git_bin, "-C", str(bare), "update-ref", "--stdin"],

            input=lines,

            env=base_env(),

            text=True,

            stdout=subprocess.DEVNULL,

            stderr=subprocess.DEVNULL,

            check=False,

        )

    return bare

def timed_sequence_row(

    meta: dict[str, Any],

    args: argparse.Namespace,

    subject: str,

    run_index: int,

    operation: str,

    commands: list[list[str]],

    cwd: Path,

    proxy: Optional[CountingProxy],

    extra: Optional[dict[str, Any]] = None,

) -> dict[str, Any]:

    """Time a small command sequence as one semantic operation.

    Used when the measured unit is "cold network acquire until X is available";

    splitting the clone into an untimed precondition would hide the network bytes

    that make the operation meaningful.

    """

    if proxy is not None:

        proxy.reset_counts()

    stdout_bytes = 0

    stderr_bytes = 0

    stderr_tail = ""

    returncode = 0

    start = time.perf_counter()

    for command in commands:

        proc = subprocess.run(

            command,

            cwd=cwd,

            env=base_env(),

            stdout=subprocess.PIPE,

            stderr=subprocess.PIPE,

            timeout=1800.0,

            check=False,

        )

        stdout_bytes += len(proc.stdout or b"")

        stderr = proc.stderr or b""

        stderr_bytes += len(stderr)

        if stderr:

            stderr_tail = (stderr_tail + stderr.decode("utf-8", errors="replace"))[-4000:]

        if proc.returncode != 0:

            returncode = proc.returncode

            break

    elapsed_ms = (time.perf_counter() - start) * 1000

    row: dict[str, Any] = {

        **meta,

        "scenario": args.scenario,

        "operation": operation,

        "run": run_index,

        "subject": subject,

        "subject_kind": SUBJECT_KINDS.get(subject, subject),

        "elapsed_ms": round(elapsed_ms, 3),

        "returncode": returncode,

        "command": commands,

        "stdout_bytes": stdout_bytes,

        "stderr_bytes": stderr_bytes,

        # Null means unmeasured (ADR-0002): the ttfd/first-payload keys are

        # always present; null records that the proxy never observed an

        # upstream-to-client payload byte, never a zero.

        "wire_bytes_client_to_server": None,

        "wire_bytes_server_to_client": None,

        "wire_bytes_total": None,

        "wire_connections": None,

        "ttfd_ms": None,

        "ttfd_source": None,

        "first_payload_bytes_server_to_client": None,

    }

    if stderr_tail:

        row["stderr"] = stderr_tail

    if proxy is not None:

        stats = proxy.snapshot()

        row.update(

            {

                "wire_bytes_client_to_server": stats.client_to_upstream_bytes,

                "wire_bytes_server_to_client": stats.upstream_to_client_bytes,

                "wire_bytes_total": stats.total_bytes,

                "wire_connections": stats.connections,

                "ttfd_ms": stats.first_upstream_to_client_ms,

                "ttfd_source": (

                    "byteproxy_first_upstream_to_client_byte"

                    if stats.first_upstream_to_client_ms is not None

                    else None

                ),

                "first_payload_bytes_server_to_client": stats.first_payload_bytes_server_to_client,

            }

        )

    if extra:

        row.update(extra)

    return row

def free_port() -> int:

    with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as sock:

        sock.bind(("127.0.0.1", 0))

        return int(sock.getsockname()[1])

def wait_for_port(host: str, port: int, timeout_s: float = 30.0) -> bool:

    deadline = time.monotonic() + timeout_s

    while time.monotonic() < deadline:

        try:

            with socket.create_connection((host, port), timeout=1.0):

                return True

        except OSError:

            time.sleep(0.2)

    return False

def seed_oak_served_repo(

    oak_bin: str,

    server_url: str,

    workdir: Path,

    runner: Callable[..., subprocess.CompletedProcess] = run_untimed,

    history_commits: int = 1,

) -> tuple[bool, str]:

    """Untimed precondition: OAK_BENCH_LOCAL_SERVER_CMD serves a freshly

    created EMPTY root, so the served repo must be seeded before any measured

    clone/pull/push has something to talk to (the task_loop lane's

    ensure_git_remote_seeded idiom, oak-shaped: init a tiny workdir, commit

    one file, push it to the server URL).

    Returns (ok, reason). On failure the reason names the failed stage —

    "oak_server_seed_failed:<stage>" — and the caller must emit per-run skip

    rows instead of measured rows (ADR-0002: an unseedable server is a

    coverage gap, never fabricated measurements). ``runner`` is injectable so

    the helper is testable on hosts that cannot run the Linux path.

    """

    seed = workdir / "oak-seed-src"

    seed.mkdir(parents=True, exist_ok=True)

    (seed / "README.md").write_text("# netshape oak fixture (server seed)\n")

    (seed / "src").mkdir(exist_ok=True)

    (seed / "src" / "main.py").write_text("print('netshape oak')\n")

    stages = (

        ("init", [oak_bin, "init", "."]),

        ("commit", [oak_bin, "commit", "--no-verify"]),

        ("push", [oak_bin, "push", server_url]),

    )

    for stage, command in stages:

        proc = runner(command, cwd=seed)

        if proc.returncode != 0:

            return False, f"oak_server_seed_failed:{stage}"

    for index in range(1, max(1, history_commits)):

        with (seed / "src" / "main.py").open("a") as fh:

            fh.write(f"print('netshape oak history {index}')\n")

        proc = runner([oak_bin, "commit", "--no-verify"], cwd=seed)

        if proc.returncode != 0:

            return False, "oak_server_seed_failed:history_commit"

        proc = runner([oak_bin, "push", server_url], cwd=seed)

        if proc.returncode != 0:

            return False, "oak_server_seed_failed:history_push"

    return True, ""

def precondition_clone_failed_rows(

    meta: dict[str, Any],

    scenario: str,

    subject: str,

    run_index: int,

    command: list[str],

    proc: subprocess.CompletedProcess,

    dest: Path,

    dependent_operations: list[str],

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

    """Guard for dest-dependent steps behind an untimed precondition clone.

    Returns None when the clone produced a usable dest. Otherwise: one

    failure row recording the clone, then a skip row per dependent step

    (reason "prior_step_failed:clone") — a failed clone is recorded data and

    the rest of that run is an honest gap, never a crash into a missing dest.

    """

    if proc.returncode == 0 and dest.is_dir():

        return None

    failure: dict[str, Any] = {

        **meta,

        "scenario": scenario,

        "operation": f"{subject}.clone",

        "run": run_index,

        "subject": subject,

        "subject_kind": SUBJECT_KINDS.get(subject, subject),

        # The precondition clone is untimed; 0.0 records the failure, not a

        # latency measurement.

        "elapsed_ms": 0.0,

        "returncode": proc.returncode if proc.returncode != 0 else 1,

        "command": command,

        "note": "untimed precondition clone failed; elapsed_ms is not a measurement",

    }

    stderr_text = (proc.stderr or "").strip() if isinstance(proc.stderr, str) else ""

    if not stderr_text and proc.returncode == 0:

        stderr_text = f"clone exited 0 but dest missing: {dest}"

    if stderr_text:

        failure["stderr"] = stderr_text[-4000:]

    rows = [failure]

    for operation in dependent_operations:

        rows.append(

            skip_row(meta, scenario, subject, run_index, "prior_step_failed:clone", operation=operation)

        )

    return rows

def run_git_subject(

    args: argparse.Namespace,

    meta: dict[str, Any],

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

    shaper: oakbench_netshape.NetnsShaper,

    workdir: Path,

) -> None:

    git_bin = shutil.which(args.git_bin)

    if git_bin is None:

        for run_index in range(args.runs):

            rows.append(skip_row(meta, args.scenario, "git", run_index, f"missing_binary:{args.git_bin}"))

        return

    if args.scenario == "netshape_mount_cold":

        for run_index in range(args.runs):

            rows.append(skip_row(meta, args.scenario, "git", run_index, "mount_scenario_oak_only"))

        return

    served_root = workdir / "git-served"

    served_root.mkdir(parents=True, exist_ok=True)

    refs_count = args.refs if args.scenario == "netshape_fetch_negotiation_refs100k" else 0

    history_commits = 8 if args.scenario == "netshape_log_follow_cold" else 1

    make_served_git_repo(git_bin, workdir, served_root, refs_count, history_commits=history_commits)

    server = GitHTTPServer(served_root).start()

    proxy = CountingProxy("127.0.0.1", server.server_port, listen_host=oakbench_netshape.HOST_ADDR).start()

    url = f"http://{oakbench_netshape.HOST_ADDR}:{proxy.address[1]}/repo.git"

    ns = shaper.name

    try:

        for run_index in range(args.runs):

            dest = workdir / f"git-run-{run_index}"

            def in_ns(command: list[str]) -> list[str]:

                return oakbench_netshape.netns_exec(ns, command)

            def precondition_clone(dependent_operations: list[str], timeout: float = 600.0) -> bool:

                command = in_ns([git_bin, "clone", url, str(dest)])

                proc = run_untimed(command, timeout=timeout)

                failed = precondition_clone_failed_rows(

                    meta, args.scenario, "git", run_index, command, proc, dest, dependent_operations,

                )

                if failed is not None:

                    rows.extend(failed)

                    return False

                return True

            if args.scenario == "netshape_clone_cold":

                rows.append(timed_row(meta, args, "git", run_index, "git.clone",

                                      in_ns([git_bin, "clone", url, str(dest)]), workdir, proxy))

            elif args.scenario == "netshape_pull_uptodate":

                if not precondition_clone(["git.pull"]):

                    continue

                rows.append(timed_row(meta, args, "git", run_index, "git.pull",

                                      in_ns([git_bin, "-C", str(dest), "pull"]), workdir, proxy))

            elif args.scenario == "netshape_push_small_delta":

                if not precondition_clone(["git.push"]):

                    continue

                (dest / "delta.txt").write_text(f"small delta run {run_index}\n")

                run_untimed([git_bin, "-C", str(dest), "add", "-A"])

                run_untimed([git_bin, "-C", str(dest), "commit", "-m", f"delta {run_index}"])

                rows.append(timed_row(meta, args, "git", run_index, "git.push",

                                      in_ns([git_bin, "-C", str(dest), "push", "origin", "HEAD"]), workdir, proxy))

            elif args.scenario == "netshape_fetch_negotiation_refs100k":

                if not precondition_clone(["git.fetch"], timeout=1800.0):

                    continue

                rows.append(timed_row(meta, args, "git", run_index, "git.fetch",

                                      in_ns([git_bin, "-C", str(dest), "fetch", "origin"]), workdir, proxy,

                                      extra={"refs_seeded": refs_count}))

            elif args.scenario == "netshape_log_follow_cold":

                clone_cmd = in_ns([git_bin, "clone", "--filter=blob:none", url, str(dest)])

                log_cmd = in_ns([git_bin, "-C", str(dest), "log", "--follow", "--", "src/main.py"])

                rows.append(timed_sequence_row(

                    meta,

                    args,

                    "git",

                    run_index,

                    "git.log.follow",

                    [clone_cmd, log_cmd],

                    workdir,

                    proxy,

                    extra={

                        "history_path": "src/main.py",

                        "history_commits_seeded": history_commits,

                        "ttfd_semantics": "cold_network_acquire_then_file_history_query",

                    },

                ))

            elif args.scenario == "netshape_task_loop":

                if not precondition_clone(["loop.push"] * TASK_LOOP_ITERATIONS):

                    continue

                for iteration in range(TASK_LOOP_ITERATIONS):

                    (dest / "loop.txt").write_text(f"iteration {iteration} run {run_index}\n")

                    run_untimed([git_bin, "-C", str(dest), "add", "-A"])

                    run_untimed([git_bin, "-C", str(dest), "commit", "-m", f"loop {iteration}"])

                    rows.append(timed_row(meta, args, "git", run_index, "loop.push",

                                          in_ns([git_bin, "-C", str(dest), "push", "origin", "HEAD"]),

                                          workdir, proxy, extra={"iteration": iteration}))

    finally:

        proxy.stop()

        server.stop()

def run_oak_subject(

    args: argparse.Namespace,

    meta: dict[str, Any],

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

    shaper: oakbench_netshape.NetnsShaper,

    workdir: Path,

) -> None:

    server_cmd = os.environ.get(OAK_LOCAL_SERVER_ENV, "").strip()

    if not server_cmd:

        for run_index in range(args.runs):

            rows.append(skip_row(

                meta, args.scenario, "oak", run_index,

                f"oak_local_server_missing: set {OAK_LOCAL_SERVER_ENV} to a self-hostable "

                "oak server command ({host} {port} {root} placeholders)",

            ))

        return

    oak_bin = shutil.which(args.oak_bin)

    if oak_bin is None:

        for run_index in range(args.runs):

            rows.append(skip_row(meta, args.scenario, "oak", run_index, f"missing_binary:{args.oak_bin}"))

        return

    served_root = workdir / "oak-served"

    served_root.mkdir(parents=True, exist_ok=True)

    port = free_port()

    try:

        server_argv = local_server_argv(

            server_cmd,

            host="127.0.0.1",

            port=port,

            root=served_root,

        )

        server_proc = subprocess.Popen(

            server_argv,

            env=base_env(),

            stdout=subprocess.DEVNULL,

            stderr=subprocess.DEVNULL,

        )

    except (OSError, ValueError) as exc:

        for run_index in range(args.runs):

            rows.append(skip_row(meta, args.scenario, "oak", run_index, f"oak_local_server_failed_to_start:{exc}"))

        return

    proxy: Optional[CountingProxy] = None

    try:

        if not wait_for_port("127.0.0.1", port):

            for run_index in range(args.runs):

                rows.append(skip_row(meta, args.scenario, "oak", run_index, "oak_local_server_failed_to_start"))

            return

        # Untimed precondition: the server starts over an EMPTY served_root, so

        # the repo the measured runs clone/pull/push must be seeded first

        # (directly against the server, off the shaped pipe — setup is never

        # the measurement). A failed seed is a per-run skip, never a crash.

        history_commits = 8 if args.scenario == "netshape_log_follow_cold" else 1

        seeded, seed_reason = seed_oak_served_repo(

            oak_bin,

            f"http://127.0.0.1:{port}/{OAK_SERVED_REPO}",

            workdir,

            history_commits=history_commits,

        )

        if not seeded:

            for run_index in range(args.runs):

                rows.append(skip_row(meta, args.scenario, "oak", run_index, seed_reason))

            return

        proxy = CountingProxy("127.0.0.1", port, listen_host=oakbench_netshape.HOST_ADDR).start()

        url = f"http://{oakbench_netshape.HOST_ADDR}:{proxy.address[1]}/{OAK_SERVED_REPO}"

        ns = shaper.name

        for run_index in range(args.runs):

            dest = workdir / f"oak-run-{run_index}"

            def in_ns(command: list[str]) -> list[str]:

                return oakbench_netshape.netns_exec(ns, command)

            def precondition_clone(dependent_operations: list[str]) -> bool:

                command = in_ns([oak_bin, "clone", url, str(dest)])

                proc = run_untimed(command)

                failed = precondition_clone_failed_rows(

                    meta, args.scenario, "oak", run_index, command, proc, dest, dependent_operations,

                )

                if failed is not None:

                    rows.extend(failed)

                    return False

                return True

            if args.scenario == "netshape_clone_cold":

                rows.append(timed_row(meta, args, "oak", run_index, "oak.clone",

                                      in_ns([oak_bin, "clone", url, str(dest)]), workdir, proxy))

            elif args.scenario == "netshape_pull_uptodate":

                if not precondition_clone(["oak.pull"]):

                    continue

                rows.append(timed_row(meta, args, "oak", run_index, "oak.pull",

                                      in_ns([oak_bin, "pull"]), dest, proxy))

            elif args.scenario == "netshape_push_small_delta":

                if not precondition_clone(["oak.push"]):

                    continue

                (dest / "delta.txt").write_text(f"small delta run {run_index}\n")

                run_untimed([oak_bin, "commit", "--no-verify"], cwd=dest)

                rows.append(timed_row(meta, args, "oak", run_index, "oak.push",

                                      in_ns([oak_bin, "push"]), dest, proxy))

            elif args.scenario == "netshape_fetch_negotiation_refs100k":

                if not precondition_clone(["oak.fetch"]):

                    continue

                rows.append(timed_row(meta, args, "oak", run_index, "oak.fetch",

                                      in_ns([oak_bin, "fetch"]), dest, proxy))

            elif args.scenario == "netshape_log_follow_cold":

                clone_cmd = in_ns([oak_bin, "clone", url, str(dest)])

                log_cmd = in_ns([oak_bin, "log", "src/main.py"])

                rows.append(timed_sequence_row(

                    meta,

                    args,

                    "oak",

                    run_index,

                    "oak.log.follow",

                    [clone_cmd, log_cmd],

                    workdir,

                    proxy,

                    extra={

                        "history_path": "src/main.py",

                        "history_commits_seeded": history_commits,

                        "ttfd_semantics": "cold_network_acquire_then_file_history_query",

                    },

                ))

            elif args.scenario == "netshape_mount_cold":

                rows.append(timed_row(meta, args, "oak", run_index, "oak.mount",

                                      in_ns([oak_bin, "mount", url, str(dest)]), workdir, proxy))

                if dest.exists():

                    run_untimed([oak_bin, "mount", "end", str(dest)])

            elif args.scenario == "netshape_task_loop":

                if not precondition_clone(["loop.push"] * TASK_LOOP_ITERATIONS):

                    continue

                for iteration in range(TASK_LOOP_ITERATIONS):

                    (dest / "loop.txt").write_text(f"iteration {iteration} run {run_index}\n")

                    run_untimed([oak_bin, "commit", "--no-verify"], cwd=dest)

                    rows.append(timed_row(meta, args, "oak", run_index, "loop.push",

                                          in_ns([oak_bin, "push"]), dest, proxy,

                                          extra={"iteration": iteration}))

    finally:

        if proxy is not None:

            proxy.stop()

        server_proc.terminate()

        try:

            server_proc.wait(timeout=10.0)

        except subprocess.TimeoutExpired:

            server_proc.kill()

def run_shaped_campaign(

    args: argparse.Namespace,

    profile: oakbench_netshape.NetshapeProfile,

    subjects: list[str],

    meta: dict[str, Any],

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

) -> None:

    workdir = Path(tempfile.mkdtemp(prefix="oak-netshape-"))

    shaper = oakbench_netshape.NetnsShaper()

    shaper.create(profile)

    try:

        selftest = shaper.self_test(profile)

        rows.append(selftest_row(meta, args, selftest))

        if not selftest["within_tolerance"]:

            # The instrument failed its own calibration; nothing measured

            # through it is evidence.

            for subject in subjects:

                for run_index in range(args.runs):

                    rows.append(skip_row(

                        meta, args.scenario, subject, run_index,

                        f"netshape_selftest_out_of_tolerance: configured "

                        f"{selftest['configured_rtt_ms']}ms measured {selftest['measured_rtt_ms']}ms",

                    ))

            return

        for subject in subjects:

            if subject == "git":

                run_git_subject(args, meta, rows, shaper, workdir)

            else:

                run_oak_subject(args, meta, rows, shaper, workdir)

    finally:

        shaper.teardown()

        shutil.rmtree(workdir, ignore_errors=True)

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

    """Exit decision over the emitted row stream (pure, unit-testable).

    Failures (any returncode other than 0/77, on ANY row including the

    self-test) dominate. The all-skipped verdict is then computed over

    SUBJECT MEASUREMENT rows only: diagnostic rows (operation

    "netshape.selftest", subject_kind "harness") prove the instrument worked,

    not that anything was measured, so a green self-test plus all-skipped

    subject rows — or a self-test with no subject rows at all — is still

    EXIT_ALL_SKIPPED.

    """

    failures = [row for row in rows if row.get("returncode") not in (0, SKIP_RETURNCODE)]

    if failures:

        return EXIT_FAILURES

    subject_rows = [

        row for row in rows

        if row.get("operation") != "netshape.selftest" and row.get("subject_kind") != "harness"

    ]

    if rows and not subject_rows:

        return EXIT_ALL_SKIPPED  # nothing was measured, only diagnostics

    if subject_rows and all(row.get("skipped") for row in subject_rows):

        return EXIT_ALL_SKIPPED

    return EXIT_OK

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

    args = parse_args(argv)

    try:

        profiles = oakbench_netshape.load_netshapes(args.netshapes)

    except oakbench_netshape.NetshapeConfigError as exc:

        print(f"error: {exc}", file=sys.stderr)

        return EXIT_USAGE

    if args.profile not in profiles:

        known = ", ".join(sorted(profiles))

        print(f"error: unknown netshape profile {args.profile!r}; known: {known}", file=sys.stderr)

        return EXIT_USAGE

    subjects = [item.strip() for item in args.subjects.split(",") if item.strip()]

    unknown_subjects = [item for item in subjects if item not in SUBJECT_KINDS]

    if not subjects or unknown_subjects:

        print(f"error: --subjects must be a non-empty subset of git,oak; got {args.subjects!r}", file=sys.stderr)

        return EXIT_USAGE

    if args.runs < 1:

        print("error: --runs must be >= 1", file=sys.stderr)

        return EXIT_USAGE

    bench_id = datetime.now(timezone.utc).strftime("%Y%m%dT%H%M%SZ")

    meta = metadata(bench_id, args)

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

    available, reason = oakbench_netshape.netshape_available()

    if not available:

        for subject in subjects:

            for run_index in range(args.runs):

                rows.append(skip_row(meta, args.scenario, subject, run_index, f"netshape_unavailable:{reason}"))

    else:

        run_shaped_campaign(args, profiles[args.profile], subjects, meta, rows)

    path = write_rows(args.results, bench_id, rows)

    print(f"[result] {path}")

    return exit_code_for_rows(rows)

if __name__ == "__main__":

    raise SystemExit(main())