#!/usr/bin/env python3

"""Measure VCS behavior under N concurrent agents.

This is the contention lane the parallel-agents thesis needs evidence from.

N workers concurrently edit disjoint files and snapshot in a loop, against:

- shared_checkout: every worker commits in the same checkout. Measures lock

  contention (git index.lock and Oak's equivalent), commit serialization

  throughput, and lost updates.

- workspace_per_task: one isolated workspace per worker on its own branch,

  plus a merge phase. Git: worktrees (local disk) merged locally. Oak: one

  `oak mount` per worker against the disposable fleet remote (set

  OAK_BENCH_FLEET_REPO), each on its auto-created virtual branch, pushed and

  merged back into main via an integrator clone. Measures isolation cost,

  time-to-Nth-workspace, marginal disk per workspace (visible vs allocated),

  and merge throughput. TRANSPORT ASYMMETRY: oak workspaces cross the network,

  git worktrees do not; rows carry workspace_transport and reports must state

  the asymmetry, never delta across it.

After the storm, an integrity pass runs: git fsck (or the closest Oak sanity

check) plus payload survival — every payload whose snapshot reported success

must be intact in the final tree. A "fast" VCS that loses updates under

contention is worse than a slow one; this lane makes that visible.

Fleet-scale modes (additive; existing modes unchanged, ADR-0005):

- merge_train: N branches built concurrently, landed serially through merge.

  Rows record per-worker queue_wait_ms (land requested -> merge started),

  merge_ms, and total merges/min.

- rebase_storm: workers rebase onto a moving main and race to push; rows

  record the attempts-to-land distribution.

- mixed_fleet: readers poll status/log on an interval while writers commit;

  reader rows carry raw poll-latency samples (zero-thrash evidence).

- conflict_storm: workers append sentinels to OVERLAPPING files; after the

  storm every sentinel must survive in HEAD — a missing sentinel is a

  recorded lost update.

- clone_push_storm: N concurrent clone -> commit -> push cycles against one

  remote (server write-path stress).

- long_divergence: one branch accumulates K commits then merges; scenario

  contention_long_divergence_k<K> (K from --divergence-k, not the tier loop).

Worker tiers come from --workers (any CSV; fleet campaigns use 16,32,64,128).

Completed rows are flushed incrementally to <results>/<ts>.partial.jsonl

(ResultsStore flush_partial) so a long campaign killed mid-run keeps its rows;

the canonical file finalizes the partial at completion.

Rows are bench.py-style JSONL plus real parallel metrics: lock_wait_ms,

commit_throughput_per_s, lost_updates_detected, integrity_check_passed,

workspace disk and inode counts. Unmeasurable values are null, never zero.

Examples:

    python3 scripts/parallel_contention.py --subjects git,oak_installed

    python3 scripts/parallel_contention.py --subjects git --workers 2,8,32 \

        --commits-per-worker 5 --modes shared_checkout,workspace_per_task

"""

from __future__ import annotations

import argparse

import platform

import re

import shutil

import subprocess

import tempfile

import threading

import time

from dataclasses import dataclass, field

from datetime import datetime, timezone

from pathlib import Path

from typing import Any

from oakbench.command_semantics import semantics

from oakbench.diskprobe import bounded_tree_usage

from oakbench.environment import ENV_ISOLATION_VERSION, base_env

from oakbench.remotes import (

    RemoteRequirement,

    attach_git_origin,

    disposable_branch,

    make_git_bare_remote,

    oak_remote_preflight_warnings,

    point_git_bare_head,

    print_remote_preflight_warnings,

    resolve_git_github_remote,

    resolve_oak_remote,

)

from oakbench.reporting import tail_latency_summary

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-parallel-contention"

# Fleet-scale modes are additive (ADR-0005): the two original modes keep their

# exact semantics and row shapes; new modes get new scenario names.

FLEET_MODES = ("merge_train", "rebase_storm", "mixed_fleet", "conflict_storm", "clone_push_storm")

MODES = ("shared_checkout", "workspace_per_task", *FLEET_MODES, "long_divergence")

SKIP_RETURNCODE = 77

MAX_SNAPSHOT_ATTEMPTS = 8

MAX_LAND_ATTEMPTS = 16

RETRY_SLEEP_S = 0.02

READER_POLL_INTERVAL_S = 0.05

CONFLICT_STORM_FILES = 3

LONG_DIVERGENCE_MAIN_COMMITS = 5

LOCK_HINT_PATTERN = ("index.lock", "another git process", "lock", "Resource temporarily unavailable")

@dataclass

class WorkerResult:

    worker: int

    elapsed_ms: float = 0.0

    commits_attempted: int = 0

    commits_succeeded: int = 0

    # Payloads recorded by ANOTHER worker's `git add .` sweep ("nothing to

    # commit" on our own attempt). The payload survived, but this worker

    # performed no snapshot — throughput must distinguish the two.

    commits_swept: int = 0

    retries: int = 0

    lock_failures: int = 0

    lock_wait_ms: float = 0.0

    # Per-commit end-to-end latency (including retries and backoff sleeps)

    # and the attempt count each successful commit needed. Retained raw so

    # tail percentiles and retry-depth histograms come from samples, not from

    # already-aggregated means.

    commit_latency_ms: list[float] = field(default_factory=list)

    attempts_to_commit: list[int] = field(default_factory=list)

    failed_payloads: list[str] = field(default_factory=list)

    succeeded_payloads: list[tuple[str, str]] = field(default_factory=list)

    errors: list[str] = field(default_factory=list)

    failure_contexts: list[dict[str, Any]] = field(default_factory=list)

def parse_args() -> argparse.Namespace:

    parser = argparse.ArgumentParser(description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter)

    parser.add_argument("--subjects", help="Comma-separated subject names from config/subjects.toml")

    parser.add_argument("--config", type=Path, default=ROOT / "config" / "subjects.toml")

    parser.add_argument("--git-bin", type=Path)

    parser.add_argument("--oak-installed-bin", type=Path)

    parser.add_argument("--oak-local-bin", type=Path)

    parser.add_argument("--oak-aa-bin", type=Path, help="A/A control: the baseline Oak binary under subject oak_installed_aa.")

    parser.add_argument(

        "--workers",

        default="2,8",

        help="Comma-separated concurrency levels (any CSV; fleet tiers: 16,32,64,128).",

    )

    parser.add_argument("--commits-per-worker", type=int, default=5)

    parser.add_argument("--modes", default="shared_checkout,workspace_per_task")

    parser.add_argument(

        "--divergence-k",

        default="10,100",

        help=(

            "Comma-separated divergence depths for the long_divergence mode "

            "(scenario contention_long_divergence_k<K>, ADR-0005). Canonical "

            "tiers: 10,100,1000; 1000 is accepted but expensive."

        ),

    )

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

    parser.add_argument("--workdir", type=Path, default=DEFAULT_WORKDIR)

    parser.add_argument("--results", type=Path, default=ROOT / "results" / "parallel-contention")

    parser.add_argument(

        "--oak-repo",

        type=Path,

        default=DEFAULT_OAK_REPO,

        help="Optional Oak source checkout for provenance metadata.",

    )

    parser.add_argument("--keep-workdirs", action="store_true")

    return parser.parse_args()

def run(command: list[str], cwd: Path, timeout: float = 120.0) -> subprocess.CompletedProcess[str]:

    # A step exceeding its timeout is a (terrible) measurement, not a harness

    # crash: convert to a synthetic failed process so the row records it and

    # the campaign survives. Killing the whole run an hour in loses every row

    # (rows flush only at write time).

    try:

        return subprocess.run(

            command,

            cwd=cwd,

            env=base_env(),

            text=True,

            stdout=subprocess.PIPE,

            stderr=subprocess.PIPE,

            timeout=timeout,

            check=False,

        )

    except subprocess.TimeoutExpired as exc:

        stdout = exc.stdout.decode("utf-8", "replace") if isinstance(exc.stdout, bytes) else (exc.stdout or "")

        stderr = exc.stderr.decode("utf-8", "replace") if isinstance(exc.stderr, bytes) else (exc.stderr or "")

        return subprocess.CompletedProcess(

            command, 124, stdout, stderr + f"\n[harness] step timed out after {timeout:.0f}s"

        )

def make_fixture(path: Path) -> None:

    if path.exists():

        shutil.rmtree(path)

    (path / "src").mkdir(parents=True)

    for index in range(100):

        (path / "src" / f"module-{index:03d}.txt").write_text(f"module {index}\n" * 8)

    (path / "README.md").write_text("Parallel contention fixture.\n")

def init_repo(subject: Subject, repo: Path) -> bool:

    vcs = str(subject.bin)

    if subject.kind == "git":

        commands = [

            [vcs, "init", "-b", "main"],

            [vcs, "add", "."],

            [vcs, "commit", "-m", "initial"],

        ]

    else:

        commands = [[vcs, "init", "."], [vcs, "commit", "--no-verify"]]

    for command in commands:

        if run(command, repo).returncode != 0:

            return False

    return True

def snapshot_commands(subject: Subject, message: str) -> list[list[str]]:

    return semantics().snapshot_commands(str(subject.bin), subject.kind, message)

def looks_like_lock_failure(stderr: str) -> bool:

    lowered = stderr.lower()

    return any(hint.lower() in lowered for hint in LOCK_HINT_PATTERN)

def excerpt(text: str, limit: int = 400) -> str:

    return text.strip()[:limit]

def record_worker_failure(

    result: WorkerResult,

    reason: str,

    *,

    command: list[str] | None = None,

    proc: subprocess.CompletedProcess[str] | None = None,

    workspace: Path | None = None,

    affected_path: str | None = None,

) -> None:

    if len(result.failure_contexts) < 5:

        context: dict[str, Any] = {

            "worker": result.worker,

            "failure_reason": reason,

        }

        if command is not None:

            context["command"] = command

        if proc is not None:

            context["returncode"] = proc.returncode

            context["stdout_excerpt"] = excerpt(proc.stdout)

            context["stderr_excerpt"] = excerpt(proc.stderr)

        if workspace is not None:

            context["workspace"] = str(workspace)

        if affected_path is not None:

            context["affected_path"] = affected_path

        result.failure_contexts.append(context)

    message = reason

    if proc is not None:

        detail = excerpt(proc.stderr) or excerpt(proc.stdout)

        if detail:

            message = f"{reason}: {detail}"

    if len(result.errors) < 5 and message not in result.errors:

        result.errors.append(message[:200])

TERMINAL_WORKER_FAILURE_REASONS = {

    "snapshot_attempts_exhausted",

    "workspace_push_failed",

    "workspace_switch_failed",

    "workspace_merge_failed",

    "worktree_merge_failed",

    "train_merge_failed",

}

def worker_result_failed(result: WorkerResult) -> bool:

    if result.commits_succeeded != result.commits_attempted or result.failed_payloads:

        return True

    for context in result.failure_contexts:

        reason = str(context.get("failure_reason") or "")

        if reason in TERMINAL_WORKER_FAILURE_REASONS or reason.startswith("worker_exception:"):

            return True

    return False

def any_worker_result_failed(results: list[WorkerResult]) -> bool:

    return any(worker_result_failed(result) for result in results)

def worker_failure_fields(result: WorkerResult, workspace: Path | None = None) -> dict[str, Any]:

    fields: dict[str, Any] = {

        "worker_id": result.worker,

        "error_examples": result.errors,

        "failure_contexts": result.failure_contexts,

        "failed_payloads": result.failed_payloads,

    }

    if worker_result_failed(result):

        first = result.failure_contexts[0] if result.failure_contexts else {}

        reason = str(first.get("failure_reason") or "worker_incomplete")

        fields["failure_reason"] = reason

        if workspace is not None:

            fields["worker_workspace"] = str(workspace)

        if result.failed_payloads:

            fields["affected_paths"] = result.failed_payloads

        if "command" in first:

            fields["command"] = first["command"]

        stderr = str(first.get("stderr_excerpt") or "")

        stdout = str(first.get("stdout_excerpt") or "")

        fields["stderr"] = stderr

        fields["stdout"] = stdout

    return fields

def aggregate_worker_failure_fields(

    results: list[WorkerResult],

    *,

    workspace: Path | None = None,

) -> dict[str, Any]:

    contexts = [context for result in results for context in result.failure_contexts]

    failed_payloads = [path for result in results for path in result.failed_payloads]

    fields: dict[str, Any] = {

        "worker_failure_contexts": contexts[:10],

        "failed_payloads": failed_payloads,

    }

    if contexts or failed_payloads:

        first = contexts[0] if contexts else {}

        fields["failure_reason"] = str(first.get("failure_reason") or "worker_incomplete")

        if workspace is not None:

            fields["worker_workspace"] = str(workspace)

        fields["stderr"] = str(first.get("stderr_excerpt") or "")

        fields["stdout"] = str(first.get("stdout_excerpt") or "")

        if "command" in first:

            fields["command"] = first["command"]

    return fields

def worker_loop(

    subject: Subject,

    workspace: Path,

    worker_index: int,

    commits: int,

    result: WorkerResult,

    rel_prefix: str = "workers",

) -> None:

    # Exceptions must not escape the thread: a silently dead worker would leave

    # partially-zero metrics that look like a fast clean run.

    start = time.perf_counter()

    try:

        _worker_loop_inner(subject, workspace, worker_index, commits, result, rel_prefix)

    except Exception as exc:  # noqa: BLE001 - recorded, not hidden

        record_worker_failure(result, f"worker_exception:{type(exc).__name__}", workspace=workspace)

        if result.failure_contexts:

            result.failure_contexts[-1]["exception"] = str(exc)[:400]

    finally:

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

def _worker_loop_inner(

    subject: Subject,

    workspace: Path,

    worker_index: int,

    commits: int,

    result: WorkerResult,

    rel_prefix: str = "workers",

) -> None:

    for k in range(commits):

        rel_path = f"{rel_prefix}/w{worker_index:03d}/f{k:03d}.txt"

        payload = f"payload worker={worker_index} commit={k} nonce={worker_index * 100003 + k}\n"

        target = workspace / rel_path

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

        target.write_text(payload)

        result.commits_attempted += 1

        committed = False

        swept = False

        commit_start = time.perf_counter()

        for attempt in range(MAX_SNAPSHOT_ATTEMPTS):

            attempt_start = time.perf_counter()

            failed = False

            stderr_text = ""

            last_command: list[str] | None = None

            last_proc: subprocess.CompletedProcess[str] | None = None

            for command in snapshot_commands(subject, f"w{worker_index} k{k}"):

                last_command = command

                proc = run(command, workspace)

                last_proc = proc

                stderr_text = proc.stderr

                if proc.returncode != 0:

                    # Concurrent `git add .` can commit this worker's file from

                    # another worker's snapshot; "nothing to commit" then means

                    # the change is already recorded — but by someone else.

                    # Payload survival: yes. Snapshot performed by this

                    # worker: no. Count it as swept, never as a snapshot.

                    if subject.kind == "git" and "nothing to commit" in (proc.stdout + proc.stderr):

                        swept = True

                        break

                    failed = True

                    if not looks_like_lock_failure(proc.stderr):

                        record_worker_failure(

                            result,

                            "snapshot_command_failed",

                            command=command,

                            proc=proc,

                            workspace=workspace,

                            affected_path=rel_path,

                        )

                    break

            attempt_ms = (time.perf_counter() - attempt_start) * 1000

            if not failed:

                committed = True

                break

            result.retries += 1

            result.lock_wait_ms += attempt_ms + RETRY_SLEEP_S * 1000

            if looks_like_lock_failure(stderr_text):

                result.lock_failures += 1

            elif len(result.errors) < 5:

                result.errors.append(stderr_text.strip()[:200])

            time.sleep(RETRY_SLEEP_S)

        if committed:

            result.commits_succeeded += 1

            result.succeeded_payloads.append((rel_path, payload))

            if swept:

                result.commits_swept += 1

            else:

                # Latency/attempt samples describe snapshots this worker

                # actually performed; a swept "success" took no commit.

                result.commit_latency_ms.append(round((time.perf_counter() - commit_start) * 1000, 3))

                result.attempts_to_commit.append(attempt + 1)

        else:

            if not result.failure_contexts or result.failure_contexts[-1].get("affected_path") != rel_path:

                record_worker_failure(

                    result,

                    "snapshot_attempts_exhausted",

                    command=last_command,

                    proc=last_proc,

                    workspace=workspace,

                    affected_path=rel_path,

                )

            result.failed_payloads.append(rel_path)

def tree_payload(subject: Subject, repo: Path, rel_path: str) -> str | None:

    if subject.kind == "git":

        proc = run([str(subject.bin), "show", f"HEAD:{rel_path}"], repo)

        return proc.stdout if proc.returncode == 0 else None

    if subject.kind == "oak":

        # Committed HEAD through the independent reader, never the worktree.

        from oakbench import oak_sqlite_reader as reader

        try:

            payloads, _ = reader.committed_payloads(repo, [rel_path])

        except reader.Invalid:

            return None

        except (reader.Unavailable, OSError, ValueError):

            return _worktree_payload(repo, rel_path)

        data = payloads.get(rel_path)

        return data.decode("utf-8", "replace") if data is not None else None

    return _worktree_payload(repo, rel_path)

def _worktree_payload(repo: Path, rel_path: str) -> str | None:

    try:

        return (repo / rel_path).read_text()

    except OSError:

        return None

def oak_committed_integrity(repo: Path) -> tuple[bool | None, str]:

    """Deep independent read: every committed object and ancestor re-hashed."""

    from oakbench import oak_sqlite_reader as reader

    try:

        reader.read_committed_tree(repo, None, deep=True)

        return True, "independent_oak_reader_deep_verify"

    except reader.Invalid as exc:

        return False, f"independent_oak_reader_invalid:{exc}"

    except (reader.Unavailable, OSError, ValueError) as exc:

        return None, f"independent_oak_reader_unavailable:{exc}"

def integrity_check(

    subject: Subject,

    repo: Path,

    results: list[WorkerResult],

) -> tuple[bool | None, int | None, str]:

    """Returns (integrity_passed, lost_updates, method)."""

    # Sweep any straggler files into a final snapshot so HEAD reflects the run.

    for command in snapshot_commands(subject, "contention sweep"):

        run(command, repo)

    lost = 0

    for result in results:

        for rel_path, payload in result.succeeded_payloads:

            recorded = tree_payload(subject, repo, rel_path)

            if recorded is None or payload not in recorded:

                lost += 1

    if subject.kind == "git":

        fsck = run([str(subject.bin), "fsck", "--no-progress"], repo)

        passed = fsck.returncode == 0 and lost == 0

        return passed, lost, "git_fsck_plus_payload_survival"

    sanity = run([str(subject.bin), "log", "-n", "1"], repo)

    status = run([str(subject.bin), "status"], repo)

    passed = sanity.returncode == 0 and status.returncode == 0 and lost == 0

    if subject.kind == "oak":

        verified, how = oak_committed_integrity(repo)

        if verified is False:

            passed = False

        return passed, lost, ("oak_log_status_sanity_plus_committed_payload_survival(independent_reader); "

                              + how)

    return passed, lost, "oak_log_status_sanity_plus_payload_survival; oak has no fsck equivalent yet"

def disk_usage(path: Path) -> tuple[int | None, int | None]:

    try:

        du = subprocess.run(["du", "-sk", str(path)], text=True, stdout=subprocess.PIPE, check=False)

        kb = int(du.stdout.split()[0]) if du.returncode == 0 and du.stdout.strip() else None

    except Exception:

        kb = None

    try:

        entries = sum(1 for _ in path.rglob("*"))

    except OSError:

        entries = None

    return kb, entries

def base_row(metadata: dict[str, Any], subject: Subject, mode: str, workers: int, run_index: int) -> dict[str, Any]:

    return {

        **metadata,

        "subject": subject.name,

        "subject_kind": subject.kind,

        "subject_label": subject.label,

        "scenario": f"contention_{mode}_w{workers}",

        "contention_mode": mode,

        "workers": workers,

        "run": run_index,

    }

def contention_payload_prefix(bench_id: str, subject_name: str, workers: int, run_index: int) -> str:

    """File namespace for payloads merged into a long-lived disposable repo."""

    return f"workers/{bench_id}/{subject_name}/w{workers}/r{run_index}"

def run_shared_checkout(

    subject: Subject,

    fixture: Path,

    workers: int,

    commits: int,

    run_root: Path,

    run_index: int,

    metadata: dict[str, Any],

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

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

    repo = run_root / f"shared-{subject.name}-w{workers}-r{run_index}"

    shutil.copytree(fixture, repo)

    if not init_repo(subject, repo):

        row = base_row(metadata, subject, "shared_checkout", workers, run_index)

        row.update({"operation": "parallel.total", "elapsed_ms": 0.0, "returncode": 1, "stderr": "repo init failed"})

        return [row]

    results = [WorkerResult(worker=index) for index in range(workers)]

    threads = [

        threading.Thread(target=worker_loop, args=(subject, repo, index, commits, results[index]))

        for index in range(workers)

    ]

    wall_start = time.perf_counter()

    for thread in threads:

        thread.start()

    for thread in threads:

        thread.join()

    wall_ms = (time.perf_counter() - wall_start) * 1000

    for result in results:

        row = base_row(metadata, subject, "shared_checkout", workers, run_index)

        row.update(

            {

                "operation": "parallel.worker",

                "worker": result.worker,

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

                "returncode": 1 if worker_result_failed(result) else 0,

                "commits_attempted": result.commits_attempted,

                "commits_succeeded": result.commits_succeeded,

                "commits_swept": result.commits_swept,

                "snapshots_performed": result.commits_succeeded - result.commits_swept,

                "retries": result.retries,

                "lock_failures": result.lock_failures,

                "lock_wait_ms": round(result.lock_wait_ms, 3),

                "commit_latency_samples_ms": result.commit_latency_ms,

                "attempts_to_commit": result.attempts_to_commit,

                **worker_failure_fields(result, repo),

            }

        )

        rows.append(row)

    integrity_passed, lost_updates, method = integrity_check(subject, repo, results)

    disk_kb, entries = disk_usage(repo)

    total_succeeded = sum(result.commits_succeeded for result in results)

    total_attempted = sum(result.commits_attempted for result in results)

    all_latencies = [sample for result in results for sample in result.commit_latency_ms]

    retry_depth_histogram: dict[str, int] = {}

    for result in results:

        for attempts in result.attempts_to_commit:

            key = str(attempts)

            retry_depth_histogram[key] = retry_depth_histogram.get(key, 0) + 1

    row = base_row(metadata, subject, "shared_checkout", workers, run_index)

    row.update(

        {

            "operation": "parallel.total",

            "elapsed_ms": round(wall_ms, 3),

            "returncode": 0 if integrity_passed and total_succeeded == total_attempted and not any_worker_result_failed(results) else 1,

            "parallel_metrics": {

                "tasks_total": total_attempted,

                "tasks_completed": total_succeeded,

                "max_parallel_tasks": workers,

                "overlapping_files_count": 0,

                "lost_updates_detected": lost_updates,

                "lock_wait_ms": round(sum(result.lock_wait_ms for result in results), 3),

                # commit_throughput_per_s is PAYLOAD throughput (its historical

                # meaning, ADR-0005): payloads safely recorded per second,

                # including those swept into another worker's snapshot.

                # snapshot_throughput_per_s counts only snapshots a worker

                # actually performed — the honest "snapshots/s" number.

                "commit_throughput_per_s": round(total_succeeded / (wall_ms / 1000.0), 3) if wall_ms else None,

                "commits_swept_total": sum(result.commits_swept for result in results),

                "snapshot_throughput_per_s": (

                    round(

                        sum(result.commits_succeeded - result.commits_swept for result in results)

                        / (wall_ms / 1000.0),

                        3,

                    )

                    if wall_ms

                    else None

                ),

                "integrity_check_passed": integrity_passed,

                # Tail of per-commit latency under contention: an agent fleet

                # feels the p99 commit (a stuck tool call), not the median.

                # Percentiles below their honest sample minimum are null.

                "commit_latency_tail": tail_latency_summary(all_latencies),

                # attempts -> commits that needed that many attempts; depth >1

                # is retry churn an agent pays as extra tool calls and waiting.

                "retry_depth_histogram": retry_depth_histogram,

                "measurement_source": method,

            },

            "retries_total": sum(result.retries for result in results),

            "lock_failures_total": sum(result.lock_failures for result in results),

            "workspace_disk_kb": disk_kb,

            "workspace_entries": entries,

        }

    )

    if row["returncode"] != 0:

        row.update(aggregate_worker_failure_fields(results, workspace=repo))

    rows.append(row)

    return rows

VIRTUAL_BRANCH_RE = re.compile(r"virtual branch ([^\s)]+)")

def oak_virtual_branch(subject: Subject, mount_dir: Path) -> str | None:

    proc = run([str(subject.bin), "status"], mount_dir)

    match = VIRTUAL_BRANCH_RE.search(proc.stdout + proc.stderr)

    return match.group(1) if match else None

def ensure_oak_fleet_seed(

    subject: Subject, fixture: Path, integrator: Path, remote_repo: str

) -> str | None:

    """Untimed: clone the disposable fleet repo and make sure main carries the

    contention fixture. Returns an error string on failure, None on success.

    Seeding is idempotent: identical fixture bytes produce a no-op commit on

    re-runs, and the disposable repo accumulates bench branches by design.

    """

    vcs = str(subject.bin)

    proc = run([vcs, "clone", remote_repo, str(integrator)], integrator.parent, timeout=300.0)

    if proc.returncode != 0:

        return f"oak clone failed: {proc.stderr.strip()[:200]}"

    marker = integrator / "CONTENTION_FIXTURE.md"

    if marker.exists():

        return None

    for source in fixture.rglob("*"):

        if not source.is_file():

            continue

        target = integrator / source.relative_to(fixture)

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

        shutil.copy2(source, target)

    marker.write_text("Parallel contention fixture (seeded by parallel_contention.py).\n")

    for command in ([vcs, "commit", "--no-verify"], [vcs, "push"], [vcs, "merge"]):

        proc = run(command, integrator, timeout=300.0)

        if proc.returncode != 0:

            return f"fleet seed `{' '.join(command[1:])}` failed: {proc.stderr.strip()[:200]}"

    return None

def error_total_row(

    metadata: dict[str, Any],

    subject: Subject,

    workers: int,

    run_index: int,

    stderr: str,

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

) -> dict[str, Any]:

    row = base_row(metadata, subject, "workspace_per_task", workers, run_index)

    row.update({"operation": "parallel.total", "elapsed_ms": 0.0, "returncode": 1, "stderr": stderr})

    if extra:

        row.update(extra)

    return row

def run_workspace_per_task_oak(

    subject: Subject,

    fixture: Path,

    workers: int,

    commits: int,

    run_root: Path,

    run_index: int,

    metadata: dict[str, Any],

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

    """N oak mounts against the disposable fleet remote — oak's worktree analog.

    Transport honesty: every mount, push, and merge here crosses the network

    to a real oak server, while the git comparator's worktrees are pure local

    disk. Rows carry workspace_transport/remote_* so reports state the

    asymmetry instead of burying it; deltas across transports are forbidden.

    """

    remote = resolve_oak_remote("fleet")

    if not remote.resolved:

        row = base_row(metadata, subject, "workspace_per_task", workers, run_index)

        row.update(

            {

                "operation": "parallel.total",

                "elapsed_ms": 0.0,

                "returncode": SKIP_RETURNCODE,

                "skipped": True,

                "skip_reason": remote.skip_reason,

            }

        )

        return [row]

    vcs = str(subject.bin)

    bench_id = str(metadata.get("bench_id", "bench"))

    lane_root = run_root / f"wpt-{subject.name}-w{workers}-r{run_index}"

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

    integrator = lane_root / "integrator"

    remote_fields = {**remote.row_fields(), "workspace_transport": remote.transport}

    mounts: list[Path] = []

    try:

        seed_error = ensure_oak_fleet_seed(subject, fixture, integrator, remote.repo)

        if seed_error:

            return [error_total_row(metadata, subject, workers, run_index, seed_error, remote_fields)]

        # Timed, sequential: workspace i usable = `oak mount` returned. The

        # cumulative curve is the fleet-spinup headline number.

        setup_samples: list[float] = []

        time_to_nth: list[float] = []

        branches: list[str] = []

        setup_start = time.perf_counter()

        for index in range(workers):

            mount_dir = lane_root / f"wt-{index:03d}"

            mount_start = time.perf_counter()

            proc = run([vcs, "mount", remote.repo, str(mount_dir)], lane_root, timeout=300.0)

            setup_samples.append(round((time.perf_counter() - mount_start) * 1000, 3))

            time_to_nth.append(round((time.perf_counter() - setup_start) * 1000, 3))

            if proc.returncode != 0:

                return [

                    error_total_row(

                        metadata,

                        subject,

                        workers,

                        run_index,

                        f"oak mount failed for workspace {index}: {proc.stderr.strip()[:200]}",

                        remote_fields,

                    )

                ]

            mounts.append(mount_dir)

        setup_ms = (time.perf_counter() - setup_start) * 1000

        # Untimed probe: each mount's auto-created virtual branch is the

        # workspace branch the merge phase folds back into main.

        for mount_dir in mounts:

            branch = oak_virtual_branch(subject, mount_dir)

            if branch is None:

                return [

                    error_total_row(

                        metadata,

                        subject,

                        workers,

                        run_index,

                        f"could not parse virtual branch for {mount_dir.name}",

                        remote_fields,

                    )

                ]

            branches.append(branch)

        # Payload paths carry the bench id: merged results accumulate on the

        # disposable repo's main, so runs must stay file-disjoint.

        rel_prefix = contention_payload_prefix(bench_id, subject.name, workers, run_index)

        results = [WorkerResult(worker=index) for index in range(workers)]

        threads = [

            threading.Thread(

                target=worker_loop,

                args=(subject, mounts[index], index, commits, results[index], rel_prefix),

            )

            for index in range(workers)

        ]

        wall_start = time.perf_counter()

        for thread in threads:

            thread.start()

        for thread in threads:

            thread.join()

        wall_ms = (time.perf_counter() - wall_start) * 1000

        # Timed per workspace: publishing the branch is part of oak's task

        # cycle (network), with no git analog in this mode.

        publish_samples: list[float] = []

        publish_failures = 0

        for index, mount_dir in enumerate(mounts):

            push_start = time.perf_counter()

            proc = run([vcs, "push"], mount_dir, timeout=300.0)

            publish_samples.append(round((time.perf_counter() - push_start) * 1000, 3))

            if proc.returncode != 0:

                publish_failures += 1

                record_worker_failure(

                    results[index],

                    "workspace_push_failed",

                    command=[vcs, "push"],

                    proc=proc,

                    workspace=mount_dir,

                )

        # Disk before teardown: visible vs allocated is the lazy-hydration

        # claim — a mount can show the whole tree while allocating ~nothing.

        usage_samples = [bounded_tree_usage(mount_dir) for mount_dir in mounts]

        marginal_alloc_kb: int | None = None

        marginal_visible_kb: int | None = None

        if usage_samples:

            marginal_alloc_kb = round(

                sum(sample["allocated_tree_bytes"] for sample in usage_samples) / len(usage_samples) / 1024

            )

            marginal_visible_kb = round(

                sum(sample["visible_tree_bytes"] for sample in usage_samples) / len(usage_samples) / 1024

            )

        # Timed merge phase: pull each workspace branch into the integrator

        # and fold it into main (server-side), oak's merge-throughput analog.

        merge_samples: list[float] = []

        conflicts = 0

        merged = 0

        merge_start = time.perf_counter()

        for index, branch in enumerate(branches):

            branch_start = time.perf_counter()

            switch = run([vcs, "switch", branch], integrator, timeout=300.0)

            if switch.returncode != 0:

                conflicts += 1

                record_worker_failure(

                    results[index],

                    "workspace_switch_failed",

                    command=[vcs, "switch", branch],

                    proc=switch,

                    workspace=integrator,

                )

                merge_samples.append(round((time.perf_counter() - branch_start) * 1000, 3))

                continue

            proc = run([vcs, "merge"], integrator, timeout=300.0)

            merge_samples.append(round((time.perf_counter() - branch_start) * 1000, 3))

            if proc.returncode != 0:

                conflicts += 1

                run([vcs, "merge", "--abort"], integrator)

                record_worker_failure(

                    results[index],

                    "workspace_merge_failed",

                    command=[vcs, "merge"],

                    proc=proc,

                    workspace=integrator,

                )

            else:

                merged += 1

        merge_ms = (time.perf_counter() - merge_start) * 1000

        # Integrity: payload survival in the merged tree plus oak's closest

        # fsck analog (log/status sanity); the integrator sits on a fresh

        # branch parented onto the merged main, so files are on disk.

        lost = 0

        for result in results:

            for rel_path, payload in result.succeeded_payloads:

                recorded = tree_payload(subject, integrator, rel_path)

                if recorded is None or payload not in recorded:

                    lost += 1

        sanity = run([vcs, "log", "-n", "1"], integrator)

        status = run([vcs, "status"], integrator)

        integrity_passed = (

            sanity.returncode == 0

            and status.returncode == 0

            and lost == 0

            and conflicts == 0

            and publish_failures == 0

        )

        base_kb, base_entries = disk_usage(integrator)

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

        for result in results:

            row = base_row(metadata, subject, "workspace_per_task", workers, run_index)

            row.update(

                {

                    "operation": "parallel.worker",

                    "worker": result.worker,

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

                    "returncode": 1 if worker_result_failed(result) else 0,

                    "commits_attempted": result.commits_attempted,

                    "commits_succeeded": result.commits_succeeded,

                    "commits_swept": result.commits_swept,

                    "snapshots_performed": result.commits_succeeded - result.commits_swept,

                    "retries": result.retries,

                    "lock_failures": result.lock_failures,

                    "lock_wait_ms": round(result.lock_wait_ms, 3),

                    **worker_failure_fields(result, mounts[result.worker] if result.worker < len(mounts) else None),

                    **remote_fields,

                }

            )

            rows.append(row)

        total_succeeded = sum(result.commits_succeeded for result in results)

        total_attempted = sum(result.commits_attempted for result in results)

        row = base_row(metadata, subject, "workspace_per_task", workers, run_index)

        row.update(

            {

                "operation": "parallel.total",

                "elapsed_ms": round(wall_ms, 3),

                "setup_ms": round(setup_ms, 3),

                "workspace_setup_ms_samples": setup_samples,

                "time_to_nth_workspace_ms": time_to_nth,

                "workspace_publish_ms_samples": publish_samples,

                "publish_failures": publish_failures,

                "merge_ms": round(merge_ms, 3),

                "merge_branch_ms_samples": merge_samples,

                "merges_clean": merged,

                "merge_conflicts": conflicts,

                "returncode": 0 if integrity_passed and total_succeeded == total_attempted and not any_worker_result_failed(results) else 1,

                "parallel_metrics": {

                    "tasks_total": total_attempted,

                    "tasks_completed": total_succeeded,

                    "max_parallel_tasks": workers,

                    "overlapping_files_count": 0,

                    "lost_updates_detected": lost,

                    "lock_wait_ms": round(sum(result.lock_wait_ms for result in results), 3),

                    "commit_throughput_per_s": round(total_succeeded / (wall_ms / 1000.0), 3) if wall_ms else None,

                    "integrity_check_passed": integrity_passed,

                    "measurement_source": (

                        "oak_log_status_sanity_plus_payload_survival_after_merge; oak has no fsck equivalent yet"

                    ),

                },

                "workspace_disk_kb": base_kb,

                "workspace_entries": base_entries,

                "marginal_workspace_disk_kb": marginal_alloc_kb,

                "marginal_workspace_visible_kb": marginal_visible_kb,

                "workspace_disk_measurement": "allocated_bytes_mean_across_mounts (visible vs allocated: lazy hydration)",

                **remote_fields,

            }

        )

        if row["returncode"] != 0:

            row.update(aggregate_worker_failure_fields(results, workspace=integrator))

        rows.append(row)

        return rows

    finally:

        for mount_dir in mounts:

            run([vcs, "mount", "end", str(mount_dir), "-f"], lane_root, timeout=120.0)

def run_workspace_per_task(

    subject: Subject,

    fixture: Path,

    workers: int,

    commits: int,

    run_root: Path,

    run_index: int,

    metadata: dict[str, Any],

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

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

    if subject.kind != "git":

        return run_workspace_per_task_oak(

            subject, fixture, workers, commits, run_root, run_index, metadata

        )

    vcs = str(subject.bin)

    base = run_root / f"wpt-{subject.name}-w{workers}-r{run_index}" / "base"

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

    shutil.copytree(fixture, base)

    if not init_repo(subject, base):

        row = base_row(metadata, subject, "workspace_per_task", workers, run_index)

        row.update({"operation": "parallel.total", "elapsed_ms": 0.0, "returncode": 1, "stderr": "repo init failed"})

        return [row]

    base_kb, base_entries = disk_usage(base)

    setup_start = time.perf_counter()

    worktrees: list[Path] = []

    setup_samples: list[float] = []

    time_to_nth: list[float] = []

    for index in range(workers):

        worktree = base.parent / f"wt-{index:03d}"

        worktree_start = time.perf_counter()

        proc = run([vcs, "worktree", "add", str(worktree), "-b", f"task-{index:03d}"], base)

        setup_samples.append(round((time.perf_counter() - worktree_start) * 1000, 3))

        time_to_nth.append(round((time.perf_counter() - setup_start) * 1000, 3))

        if proc.returncode != 0:

            row = base_row(metadata, subject, "workspace_per_task", workers, run_index)

            row.update(

                {

                    "operation": "parallel.total",

                    "elapsed_ms": 0.0,

                    "returncode": 1,

                    "stderr": f"worktree add failed: {proc.stderr.strip()[:200]}",

                }

            )

            return [row]

        worktrees.append(worktree)

    setup_ms = (time.perf_counter() - setup_start) * 1000

    results = [WorkerResult(worker=index) for index in range(workers)]

    threads = [

        threading.Thread(target=worker_loop, args=(subject, worktrees[index], index, commits, results[index]))

        for index in range(workers)

    ]

    wall_start = time.perf_counter()

    for thread in threads:

        thread.start()

    for thread in threads:

        thread.join()

    wall_ms = (time.perf_counter() - wall_start) * 1000

    merge_start = time.perf_counter()

    conflicts = 0

    merged = 0

    for index in range(workers):

        proc = run([vcs, "merge", "--no-edit", f"task-{index:03d}"], base)

        if proc.returncode != 0:

            conflicts += 1

            run([vcs, "merge", "--abort"], base)

            record_worker_failure(

                results[index],

                "worktree_merge_failed",

                command=[vcs, "merge", "--no-edit", f"task-{index:03d}"],

                proc=proc,

                workspace=base,

            )

        else:

            merged += 1

    merge_ms = (time.perf_counter() - merge_start) * 1000

    lost = 0

    for result in results:

        for rel_path, payload in result.succeeded_payloads:

            recorded = tree_payload(subject, base, rel_path)

            if recorded is None or payload not in recorded:

                lost += 1

    fsck = run([vcs, "fsck", "--no-progress"], base)

    integrity_passed = fsck.returncode == 0 and lost == 0 and conflicts == 0

    marginal_kb: int | None = None

    if worktrees and base_kb is not None:

        worktree_kb = [disk_usage(worktree)[0] for worktree in worktrees]

        known = [kb for kb in worktree_kb if kb is not None]

        if known:

            marginal_kb = round(sum(known) / len(known))

    for result in results:

        row = base_row(metadata, subject, "workspace_per_task", workers, run_index)

        row.update(

            {

                "operation": "parallel.worker",

                "worker": result.worker,

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

                "returncode": 1 if worker_result_failed(result) else 0,

                "commits_attempted": result.commits_attempted,

                "commits_succeeded": result.commits_succeeded,

                "commits_swept": result.commits_swept,

                "snapshots_performed": result.commits_succeeded - result.commits_swept,

                "retries": result.retries,

                "lock_failures": result.lock_failures,

                "lock_wait_ms": round(result.lock_wait_ms, 3),

                **worker_failure_fields(result, worktrees[result.worker] if result.worker < len(worktrees) else None),

            }

        )

        rows.append(row)

    total_succeeded = sum(result.commits_succeeded for result in results)

    total_attempted = sum(result.commits_attempted for result in results)

    row = base_row(metadata, subject, "workspace_per_task", workers, run_index)

    row.update(

        {

            "operation": "parallel.total",

            "elapsed_ms": round(wall_ms, 3),

            "setup_ms": round(setup_ms, 3),

            "workspace_setup_ms_samples": setup_samples,

            "time_to_nth_workspace_ms": time_to_nth,

            "workspace_transport": "local_file",

            "merge_ms": round(merge_ms, 3),

            "merges_clean": merged,

            "merge_conflicts": conflicts,

            "returncode": 0 if integrity_passed and total_succeeded == total_attempted and not any_worker_result_failed(results) else 1,

            "parallel_metrics": {

                "tasks_total": total_attempted,

                "tasks_completed": total_succeeded,

                "max_parallel_tasks": workers,

                "overlapping_files_count": 0,

                "lost_updates_detected": lost,

                "lock_wait_ms": round(sum(result.lock_wait_ms for result in results), 3),

                "commit_throughput_per_s": round(total_succeeded / (wall_ms / 1000.0), 3) if wall_ms else None,

                "integrity_check_passed": integrity_passed,

                "measurement_source": "git_fsck_plus_payload_survival_after_merge",

            },

            "workspace_disk_kb": base_kb,

            "workspace_entries": base_entries,

            "marginal_workspace_disk_kb": marginal_kb,

        }

    )

    if row["returncode"] != 0:

        row.update(aggregate_worker_failure_fields(results, workspace=base))

    rows.append(row)

    return rows

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

# Fleet-scale modes (additive; see FLEET_MODES). Scenario names follow the

# established contention_<mode>_w<N> shape via base_row, except long_divergence

# which encodes its variable as contention_long_divergence_k<K> (ADR-0005).

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

def mode_skip_row(

    metadata: dict[str, Any],

    subject: Subject,

    mode: str,

    workers: int,

    run_index: int,

    reason: str,

    scenario: str | None = None,

) -> dict[str, Any]:

    """Structured skip: a recorded coverage gap, never a silent absence (ADR-0002)."""

    row = base_row(metadata, subject, mode, workers, run_index)

    if scenario is not None:

        row["scenario"] = scenario

    row.update(

        {

            "operation": "parallel.total",

            "elapsed_ms": 0.0,

            "returncode": SKIP_RETURNCODE,

            "skipped": True,

            "skip_reason": reason,

        }

    )

    return row

def fleet_mode_subject_skip(

    metadata: dict[str, Any],

    subject: Subject,

    mode: str,

    workers: int,

    run_index: int,

    scenario: str | None = None,

) -> dict[str, Any]:

    return mode_skip_row(

        metadata,

        subject,

        mode,

        workers,

        run_index,

        reason=(

            f"{mode}_not_implemented_for_kind_{subject.kind}: this fleet mode currently "

            "drives git binaries only; an oak runner (mount/clone orchestration against "

            "OAK_BENCH_FLEET_REPO, see workspace_per_task) is a work item, not an absence"

        ),

        scenario=scenario,

    )

def mode_error_row(

    metadata: dict[str, Any],

    subject: Subject,

    mode: str,

    workers: int,

    run_index: int,

    stderr: str,

) -> dict[str, Any]:

    row = base_row(metadata, subject, mode, workers, run_index)

    row.update({"operation": "parallel.total", "elapsed_ms": 0.0, "returncode": 1, "stderr": stderr})

    return row

def payload_survival(subject: Subject, repo: Path, payloads: list[tuple[str, str]]) -> int:

    """Count payloads a worker recorded as landed that are missing from HEAD."""

    lost = 0

    for rel_path, payload in payloads:

        recorded = tree_payload(subject, repo, rel_path)

        if recorded is None or payload not in recorded:

            lost += 1

    return lost

def run_merge_train(

    subject: Subject,

    fixture: Path,

    workers: int,

    commits: int,

    run_root: Path,

    run_index: int,

    metadata: dict[str, Any],

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

    """N branches built concurrently, landed serially through merge — the

    merge-train pattern. Measures per-worker queue_wait_ms (land requested →

    merge started), merge_ms, and total merges/min through the train."""

    mode = "merge_train"

    if subject.kind != "git":

        return [fleet_mode_subject_skip(metadata, subject, mode, workers, run_index)]

    vcs = str(subject.bin)

    base = run_root / f"mt-{subject.name}-w{workers}-r{run_index}" / "base"

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

    shutil.copytree(fixture, base)

    if not init_repo(subject, base):

        return [mode_error_row(metadata, subject, mode, workers, run_index, "repo init failed")]

    worktrees: list[Path] = []

    for index in range(workers):

        worktree = base.parent / f"wt-{index:03d}"

        proc = run([vcs, "worktree", "add", str(worktree), "-b", f"train-{index:03d}"], base, timeout=300.0)

        if proc.returncode != 0:

            return [

                mode_error_row(

                    metadata, subject, mode, workers, run_index,

                    f"worktree add failed: {proc.stderr.strip()[:200]}",

                )

            ]

        worktrees.append(worktree)

    results = [WorkerResult(worker=index) for index in range(workers)]

    ready_at: list[float | None] = [None] * workers

    def build(index: int) -> None:

        worker_loop(subject, worktrees[index], index, commits, results[index])

        # The land request: the moment this branch is ready for the train.

        ready_at[index] = time.perf_counter()

    threads = [threading.Thread(target=build, args=(index,)) for index in range(workers)]

    wall_start = time.perf_counter()

    for thread in threads:

        thread.start()

    for thread in threads:

        thread.join()

    build_ms = (time.perf_counter() - wall_start) * 1000

    # Serial landing in arrival order: queue_wait is request->merge-start.

    order = sorted(range(workers), key=lambda index: ready_at[index] if ready_at[index] is not None else float("inf"))

    queue_wait_ms: list[float | None] = [None] * workers

    merge_ms_per_worker: list[float | None] = [None] * workers

    landed: list[bool] = [False] * workers

    land_start = time.perf_counter()

    merged = 0

    conflicts = 0

    for index in order:

        merge_start = time.perf_counter()

        if ready_at[index] is not None:

            queue_wait_ms[index] = round((merge_start - ready_at[index]) * 1000, 3)

        proc = run([vcs, "merge", "--no-edit", f"train-{index:03d}"], base, timeout=300.0)

        merge_ms_per_worker[index] = round((time.perf_counter() - merge_start) * 1000, 3)

        if proc.returncode != 0:

            conflicts += 1

            run([vcs, "merge", "--abort"], base)

            record_worker_failure(

                results[index],

                "train_merge_failed",

                command=[vcs, "merge", "--no-edit", f"train-{index:03d}"],

                proc=proc,

                workspace=base,

            )

        else:

            merged += 1

            landed[index] = True

    land_ms = (time.perf_counter() - land_start) * 1000

    wall_ms = (time.perf_counter() - wall_start) * 1000

    merges_per_min = round(merged / (land_ms / 60000.0), 3) if land_ms > 0 else None

    lost = sum(

        payload_survival(subject, base, result.succeeded_payloads)

        for index, result in enumerate(results)

        if landed[index]

    )

    fsck = run([vcs, "fsck", "--no-progress"], base)

    integrity_passed = fsck.returncode == 0 and lost == 0 and conflicts == 0

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

    for index, result in enumerate(results):

        row = base_row(metadata, subject, mode, workers, run_index)

        row.update(

            {

                "operation": "parallel.worker",

                "worker": result.worker,

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

                "returncode": 0 if not worker_result_failed(result) and landed[index] else 1,

                "commits_attempted": result.commits_attempted,

                "commits_succeeded": result.commits_succeeded,

                "retries": result.retries,

                "lock_failures": result.lock_failures,

                "queue_wait_ms": queue_wait_ms[index],

                "merge_ms": merge_ms_per_worker[index],

                "landed": landed[index],

                **worker_failure_fields(result, worktrees[index] if index < len(worktrees) else None),

            }

        )

        rows.append(row)

    queue_samples = [value for value in queue_wait_ms if value is not None]

    merge_samples = [value for value in merge_ms_per_worker if value is not None]

    total = base_row(metadata, subject, mode, workers, run_index)

    total.update(

        {

            "operation": "parallel.total",

            "elapsed_ms": round(wall_ms, 3),

            "build_ms": round(build_ms, 3),

            "land_ms": round(land_ms, 3),

            "merges_per_min": merges_per_min,

            "merges_clean": merged,

            "merge_conflicts": conflicts,

            "queue_wait_tail": tail_latency_summary(queue_samples),

            "merge_latency_tail": tail_latency_summary(merge_samples),

            "returncode": 0 if integrity_passed and merged == workers and not any_worker_result_failed(results) else 1,

            "parallel_metrics": {

                "tasks_total": sum(result.commits_attempted for result in results),

                "tasks_completed": sum(result.commits_succeeded for result in results),

                "max_parallel_tasks": workers,

                "overlapping_files_count": 0,

                "lost_updates_detected": lost,

                "lock_wait_ms": round(sum(result.lock_wait_ms for result in results), 3),

                "commit_throughput_per_s": (

                    round(sum(result.commits_succeeded for result in results) / (wall_ms / 1000.0), 3)

                    if wall_ms

                    else None

                ),

                "integrity_check_passed": integrity_passed,

                "measurement_source": "git_fsck_plus_payload_survival_after_merge_train",

            },

            "retries_total": sum(result.retries for result in results),

            "lock_failures_total": sum(result.lock_failures for result in results),

        }

    )

    if total["returncode"] != 0:

        total.update(aggregate_worker_failure_fields(results, workspace=base))

    rows.append(total)

    return rows

def seed_git_bare_remote(

    subject: Subject, fixture: Path, lane_root: Path

) -> tuple[Any, Path] | str:

    """Untimed precondition: bare remote + seeded main. Returns error text on failure."""

    vcs = str(subject.bin)

    seed = lane_root / "seed"

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

    shutil.copytree(fixture, seed)

    if not init_repo(subject, seed):

        return "seed repo init failed"

    remote = make_git_bare_remote(vcs, lane_root)

    attach_git_origin(vcs, seed, remote)

    proc = run([vcs, "push", "origin", "main"], seed, timeout=300.0)

    if proc.returncode != 0:

        return f"seed push failed: {proc.stderr.strip()[:200]}"

    point_git_bare_head(vcs, remote, "main")

    return remote, seed

def run_rebase_storm(

    subject: Subject,

    fixture: Path,

    workers: int,

    commits: int,

    run_root: Path,

    run_index: int,

    metadata: dict[str, Any],

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

    """Workers race to land on a moving main: rebase onto origin/main, push,

    and on non-fast-forward rejection rebase again. Rows record the

    attempts-to-land distribution — the cost of a moving target."""

    mode = "rebase_storm"

    if subject.kind != "git":

        return [fleet_mode_subject_skip(metadata, subject, mode, workers, run_index)]

    vcs = str(subject.bin)

    lane_root = run_root / f"rs-{subject.name}-w{workers}-r{run_index}"

    seeded = seed_git_bare_remote(subject, fixture, lane_root)

    if isinstance(seeded, str):

        return [mode_error_row(metadata, subject, mode, workers, run_index, seeded)]

    remote, _seed = seeded

    remote_fields = remote.row_fields()

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

        {

            "worker": index,

            "elapsed_ms": 0.0,

            "lands_attempted": 0,

            "lands_succeeded": 0,

            "attempts_to_land": [],

            "land_latency_ms": [],

            "payloads": [],

            "errors": [],

        }

        for index in range(workers)

    ]

    def storm(index: int) -> None:

        state = states[index]

        start = time.perf_counter()

        try:

            wdir = lane_root / f"w{index:03d}"

            proc = run([vcs, "clone", remote.repo, str(wdir)], lane_root, timeout=300.0)

            if proc.returncode != 0:

                state["errors"].append(f"clone failed: {proc.stderr.strip()[:200]}")

                return

            for k in range(commits):

                rel_path = f"storm/w{index:03d}/f{k:03d}.txt"

                payload = f"payload worker={index} commit={k} nonce={index * 100003 + k}\n"

                target = wdir / rel_path

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

                target.write_text(payload)

                state["lands_attempted"] += 1

                committed = True

                for command in snapshot_commands(subject, f"storm w{index} k{k}"):

                    proc = run(command, wdir)

                    if proc.returncode != 0:

                        state["errors"].append(f"local commit failed: {proc.stderr.strip()[:200]}")

                        committed = False

                        break

                if not committed:

                    continue

                land_start = time.perf_counter()

                attempts = 0

                landed = False

                while attempts < MAX_LAND_ATTEMPTS:

                    attempts += 1

                    run([vcs, "fetch", "origin", "main"], wdir, timeout=300.0)

                    rebase = run([vcs, "rebase", "origin/main"], wdir, timeout=300.0)

                    if rebase.returncode != 0:

                        run([vcs, "rebase", "--abort"], wdir)

                        state["errors"].append(f"rebase failed: {rebase.stderr.strip()[:200]}")

                        break

                    push = run([vcs, "push", "origin", "HEAD:main"], wdir, timeout=300.0)

                    if push.returncode == 0:

                        landed = True

                        break

                state["attempts_to_land"].append(attempts)

                state["land_latency_ms"].append(round((time.perf_counter() - land_start) * 1000, 3))

                if landed:

                    state["lands_succeeded"] += 1

                    state["payloads"].append((rel_path, payload))

        except Exception as exc:  # noqa: BLE001 - recorded, not hidden

            state["errors"].append(f"worker exception: {type(exc).__name__}: {exc}"[:200])

        finally:

            state["elapsed_ms"] = (time.perf_counter() - start) * 1000

    threads = [threading.Thread(target=storm, args=(index,)) for index in range(workers)]

    wall_start = time.perf_counter()

    for thread in threads:

        thread.start()

    for thread in threads:

        thread.join()

    wall_ms = (time.perf_counter() - wall_start) * 1000

    # Integrity: a fresh clone of the remote must contain every landed payload.

    verify = lane_root / "verify"

    clone = run([vcs, "clone", remote.repo, str(verify)], lane_root, timeout=300.0)

    lost: int | None = None

    fsck_ok: bool | None = None

    if clone.returncode == 0:

        lost = sum(payload_survival(subject, verify, state["payloads"]) for state in states)

        fsck_ok = run([vcs, "fsck", "--no-progress"], verify).returncode == 0

    total_landed = sum(state["lands_succeeded"] for state in states)

    total_attempted = sum(state["lands_attempted"] for state in states)

    integrity_passed = fsck_ok is True and lost == 0

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

    for state in states:

        row = base_row(metadata, subject, mode, workers, run_index)

        row.update(

            {

                "operation": "parallel.worker",

                "worker": state["worker"],

                "elapsed_ms": round(state["elapsed_ms"], 3),

                "returncode": 0 if state["lands_succeeded"] == state["lands_attempted"] else 1,

                "lands_attempted": state["lands_attempted"],

                "lands_succeeded": state["lands_succeeded"],

                "attempts_to_land": state["attempts_to_land"],

                "land_latency_samples_ms": state["land_latency_ms"],

                "error_examples": state["errors"][:5],

                **remote_fields,

            }

        )

        rows.append(row)

    attempts_histogram: dict[str, int] = {}

    for state in states:

        for attempts in state["attempts_to_land"]:

            key = str(attempts)

            attempts_histogram[key] = attempts_histogram.get(key, 0) + 1

    all_land_latencies = [sample for state in states for sample in state["land_latency_ms"]]

    total = base_row(metadata, subject, mode, workers, run_index)

    total.update(

        {

            "operation": "parallel.total",

            "elapsed_ms": round(wall_ms, 3),

            "returncode": 0 if integrity_passed and total_landed == total_attempted else 1,

            "attempts_to_land_histogram": attempts_histogram,

            "land_latency_tail": tail_latency_summary(all_land_latencies),

            "lands_per_min": round(total_landed / (wall_ms / 60000.0), 3) if wall_ms > 0 else None,

            "parallel_metrics": {

                "tasks_total": total_attempted,

                "tasks_completed": total_landed,

                "max_parallel_tasks": workers,

                "overlapping_files_count": 0,

                "lost_updates_detected": lost,

                "lock_wait_ms": None,

                "commit_throughput_per_s": round(total_landed / (wall_ms / 1000.0), 3) if wall_ms else None,

                "integrity_check_passed": integrity_passed,

                "measurement_source": "git_fsck_plus_payload_survival_in_fresh_clone",

            },

            **remote_fields,

        }

    )

    rows.append(total)

    return rows

def run_mixed_fleet(

    subject: Subject,

    fixture: Path,

    workers: int,

    commits: int,

    run_root: Path,

    run_index: int,

    metadata: dict[str, Any],

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

    """Readers poll status/log while writers commit in the same checkout —

    direct evidence for (or against) read-path thrash under write load.

    Reader rows carry raw poll-latency samples; zero failed polls plus a flat

    tail is the zero-thrash claim."""

    mode = "mixed_fleet"

    if subject.kind != "git":

        return [fleet_mode_subject_skip(metadata, subject, mode, workers, run_index)]

    vcs = str(subject.bin)

    repo = run_root / f"mf-{subject.name}-w{workers}-r{run_index}"

    shutil.copytree(fixture, repo)

    if not init_repo(subject, repo):

        return [mode_error_row(metadata, subject, mode, workers, run_index, "repo init failed")]

    writer_count = max(1, workers // 2)

    reader_count = workers - writer_count

    writer_results = [WorkerResult(worker=index) for index in range(writer_count)]

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

        {

            "worker": writer_count + index,

            "elapsed_ms": 0.0,

            "polls_total": 0,

            "poll_failures": 0,

            "poll_latency_ms": [],

            "errors": [],

        }

        for index in range(reader_count)

    ]

    stop = threading.Event()

    def reader_loop(state: dict[str, Any]) -> None:

        start = time.perf_counter()

        try:

            while not stop.is_set():

                poll_start = time.perf_counter()

                status = run([vcs, "status", "--porcelain"], repo)

                log = run([vcs, "log", "-n", "5", "--oneline"], repo)

                state["poll_latency_ms"].append(round((time.perf_counter() - poll_start) * 1000, 3))

                state["polls_total"] += 1

                if status.returncode != 0 or log.returncode != 0:

                    state["poll_failures"] += 1

                    if len(state["errors"]) < 5: