run_experiment.py

#!/usr/bin/env python3
# /// script
# requires-python = ">=3.10"
# dependencies = ["pygal"]
# ///
"""
LFS Transfer Benchmark: compare git-lfs-transfer variants and storage backends.

Usage:
  python3 run_experiment.py [--dataset small|r0] [--scenarios 0,1,...,11]
                            [--skip-build] [--skip-setup] [--keep-containers]
                            [--data-dir /workspace/desync/data/R0.git]

Scenarios:
  0  plain git (no LFS)
  1  baseline git-lfs-transfer (server-side, local files, scutiger or charmbracelet)
  2  desync git-lfs-transfer        (server-side, local chunks)
  3  desync git-lfs-transfer        (server-side, MinIO S3)
  4  desync git-lfs-transfer        (server-side, Garage S3)
  5  git-lfs-desync client agent    (client-side, MinIO S3)
  6  git-lfs-desync client agent    (client-side, Garage S3)
  7  Stock Forgejo HTTP batch API   (server-side, MinIO S3, upstream v14)
  8  Forgejo HTTP batch API         (server-side, MinIO S3, custom)
  9  Forgejo HTTP batch API         (server-side, desync Garage, custom)
  10 Forgejo SSH pkt-line           (server-side, desync Garage, custom)
  11 Forgejo SSH pkt-line           (server-side, MinIO S3, custom)

Cross-compatibility tests (after all scenarios):
  3→client: clone scenario-3 data using client-side desync agent
  4→client: clone scenario-4 data using client-side desync agent
  5→server: clone scenario-5 data using server-side desync transfer
  6→server: clone scenario-6 data using server-side desync transfer
  9→client: clone scenario-9 data using client-side desync agent
  10→http:  clone scenario-10 SSH data using HTTP batch API
"""

import argparse
import hashlib
import hmac
import http.client
import json
import os
import random
import re
import shutil
import socket
import subprocess
import sys
import time
import threading
from dataclasses import dataclass, field
from pathlib import Path
from typing import Optional
import urllib.parse
import urllib.request

# ---------------------------------------------------------------------------
# Paths
# ---------------------------------------------------------------------------
EXPERIMENT_DIR = Path(__file__).parent.resolve()
BINARIES_DIR = EXPERIMENT_DIR / "binaries"

# Source directories (git submodules).
DESYNC_SRC = EXPERIMENT_DIR / "desync"
SCUTIGER_SRC = EXPERIMENT_DIR / "scutiger"
CHARM_SRC = EXPERIMENT_DIR / "charm-git-lfs-transfer"
GIT_LFS_SRC = EXPERIMENT_DIR / "git-lfs"
FORGEJO_SRC = EXPERIMENT_DIR / "forgejo"
KEYS_DIR = EXPERIMENT_DIR / "keys"
SSH_KEY = KEYS_DIR / "lfs-test-key"
SSH_PUBKEY = KEYS_DIR / "lfs-test-key.pub"
GARAGE_TOML = EXPERIMENT_DIR / "garage.toml"

# Service coordinates (from host perspective, for management operations)
SSH_PORT = 2222
MINIO_USER = "minioadmin"
MINIO_PASS = "minioadmin"

# Shared S3 services (one MinIO + one Garage, wiped between scenarios)
SHARED_MINIO_CONTAINER = "lfs-exp-shared-minio"
SHARED_MINIO_ENDPOINT = "http://localhost:9000"
SHARED_MINIO_INTERNAL = "http://shared-minio:9000"
SHARED_GARAGE_CONTAINER = "lfs-exp-shared-garage"
SHARED_GARAGE_ENDPOINT = "http://localhost:3900"
SHARED_GARAGE_INTERNAL = "http://shared-garage:3900"

GIT_SSH_INTERNAL = "git-ssh"  # docker service hostname

# Client-side desync config file path (inside client container).
# Written by setup_containers() with local-only settings (cache, concurrency).
# Uses the default config location so the agent loads it automatically.
# Server config (URLs, credentials, chunk-size) is provided separately via
# the SSH transfer negotiation protocol; MergeServerConfig() combines both.
CLIENT_DESYNC_CONFIG = "/home/dev/.config/desync/config.json"

# Client-side desync chunk cache path (inside client container).
# Must be on the ext4 /work bind mount, NOT on /tmp (overlayfs), because
# overlayfs copy-on-write is slow for many small file writes.
CLIENT_DESYNC_CACHE = "/work/.desync-cache"

# Docker container names (as set in docker-compose.yml)
GIT_SSH_CONTAINER = "lfs-exp-git-ssh"
CLIENT_CONTAINER = "lfs-exp-client"
S8_FORGEJO_CONTAINER = "lfs-exp-s8-forgejo"
S9_FORGEJO_CONTAINER = "lfs-exp-s9-forgejo"
FORGEJO_DB_CONTAINER = "lfs-exp-forgejo-db"
S10_FORGEJO_CONTAINER = "lfs-exp-s10-forgejo"
S7_FORGEJO_CONTAINER = "lfs-exp-s7-forgejo"
S8_FORGEJO_HTTP = "http://s8-forgejo:3000"     # inside docker network
S9_FORGEJO_HTTP = "http://s9-forgejo:3000"
S10_FORGEJO_HTTP = "http://s10-forgejo:3000"
S7_FORGEJO_HTTP = "http://s7-forgejo:3000"
S11_FORGEJO_CONTAINER = "lfs-exp-s11-forgejo"
S12_FORGEJO_CONTAINER = "lfs-exp-s12-forgejo"
S13_GITEA_CONTAINER = "lfs-exp-s13-gitea"
S14_GITEA_CONTAINER = "lfs-exp-s14-gitea"
FORGEJO_USER = "testadmin"
FORGEJO_PASS = "admin12345678"
FORGEJO_EMAIL = "admin@test.local"

# Docker Compose project name (directory name); volumes are prefixed with this.
# Docker Compose derives the project name from the directory name (lowercased,
# keeping alphanumerics, hyphens, and underscores).
COMPOSE_PROJECT = re.sub(r"[^a-z0-9_-]", "", EXPERIMENT_DIR.name.lower())

# Path inside the client container where the host work dir is mounted
WORK_DIR_IN_CLIENT = "/work"

# SSH command for host→server connections (repo creation, config, etc.)
SSH_CMD = (
    f"ssh -i {SSH_KEY} -o StrictHostKeyChecking=no"
    f" -o UserKnownHostsFile=/dev/null -o BatchMode=yes"
)

# Set in main() before running scenarios; used by client path helpers.
WORK_DIR: Path = None

# Concurrency tuning.  All settings can be overridden via environment variables.
# GIT_LFS_CONCURRENCY: how many LFS objects Git LFS transfers in parallel
#   (each opens a separate SSH connection with its own git-lfs-transfer process).
GIT_LFS_CONCURRENCY = int(os.environ.get("GIT_LFS_CONCURRENCY", 8))
# DESYNC_CONCURRENCY: chunk workers inside each git-lfs-transfer / git-lfs-desync process.
DESYNC_CONCURRENCY = int(os.environ.get("DESYNC_CONCURRENCY", 10))

# Chunk size: min:avg:max in KB.  Default is "64:256:1024".
CHUNK_SIZE = os.environ.get("CHUNK_SIZE", "64:256:1024")

# Maximum total bytes allowed in-flight across all concurrent transfer processes.
# Set to 0 to disable the limit.  This caps memory usage when processing large
# datasets with multiple concurrent transfers.
MAX_IN_FLIGHT = int(os.environ.get("MAX_IN_FLIGHT", 250 * 1024**2))  # 250 MB

# Maximum concurrent storage operations (GetChunk/StoreChunk/GetIndex/StoreIndex)
# across all concurrent transfer processes.  Set to 0 to disable.
MAX_STORAGE_OPS = int(os.environ.get("MAX_STORAGE_OPS", 20))

# Baseline (S1) git-lfs-transfer implementation: "scutiger" (Rust, default,
# used as reference by git-lfs test suite) or "charmbracelet" (Go).
S1_LFS_TRANSFER = os.environ.get("S1_LFS_TRANSFER", "scutiger")

# Verbose mode: set by --verbose flag, disables output capture so all
# subprocess output is visible for debugging.
VERBOSE = False

# CLI fragments for agent args (empty when disabled).
_MAX_IN_FLIGHT_ARG = f" --max-in-flight {MAX_IN_FLIGHT}" if MAX_IN_FLIGHT > 0 else ""
_MAX_STORAGE_OPS_ARG = f" --max-storage-ops {MAX_STORAGE_OPS}" if MAX_STORAGE_OPS > 0 else ""

# How often to reclaim page cache during push loops (every N refspec pushes).
# Uses cgroup v2 memory.reclaim to evict file-backed pages.  Set to 0 to disable.
RECLAIM_INTERVAL = 0  # disabled


# ---------------------------------------------------------------------------
# Data classes
# ---------------------------------------------------------------------------
@dataclass
class ScenarioResult:
    id: int
    name: str
    push_time_s: float = 0.0
    clone_time_s: float = 0.0
    storage_bytes: int = 0
    raw_lfs_bytes: int = 0
    # Network bytes [rx, tx] per container per operation.
    push_net_bytes: list = field(default_factory=lambda: [-1, -1])
    clone_net_bytes: list = field(default_factory=lambda: [-1, -1])
    push_ssh_net_bytes: list = field(default_factory=lambda: [-1, -1])
    clone_ssh_net_bytes: list = field(default_factory=lambda: [-1, -1])
    push_s3_net_bytes: list = field(default_factory=lambda: [-1, -1])
    clone_s3_net_bytes: list = field(default_factory=lambda: [-1, -1])
    # S3 container measured (for display / JSON)
    s3_container: str = ""
    # LFS object counts: how many unique objects exist in .git/lfs/objects/
    source_lfs_count: int = -1   # in source_repo (set per scenario)
    clone_lfs_count: int = -1    # in clone after git lfs fetch --all
    verified: bool = False
    error: Optional[str] = None
    notes: str = ""
    # Resource usage per container: {container_name: ResourceSummary}
    resources: dict = field(default_factory=dict)

    @property
    def storage_efficiency(self) -> float:
        if self.raw_lfs_bytes == 0:
            return 0.0
        return self.storage_bytes / self.raw_lfs_bytes

    @property
    def storage_ratio(self) -> float:
        """Inverse of efficiency: raw bytes per stored byte. Higher = better."""
        if self.storage_bytes <= 0:
            return 0.0
        return self.raw_lfs_bytes / self.storage_bytes


@dataclass
class CrossCompatResult:
    name: str
    clone_time_s: float = 0.0
    verified: bool = False
    error: Optional[str] = None


@dataclass
class ResourceSnapshot:
    cpu_percent: float
    mem_usage_bytes: int
    mem_limit_bytes: int
    mem_anon_bytes: int = 0   # RSS (process heap, stacks, mmap)
    mem_cache_bytes: int = 0  # page cache (file-backed)


@dataclass
class ResourceSummary:
    avg_cpu_percent: float = 0.0
    max_cpu_percent: float = 0.0
    avg_mem_bytes: int = 0
    max_mem_bytes: int = 0
    max_anon_bytes: int = 0
    max_cache_bytes: int = 0
    mem_limit_bytes: int = 0
    sample_count: int = 0


# ---------------------------------------------------------------------------
# Logging helpers
# ---------------------------------------------------------------------------
RED = "\033[91m"
GREEN = "\033[92m"
YELLOW = "\033[93m"
BLUE = "\033[94m"
CYAN = "\033[96m"
BOLD = "\033[1m"
RESET = "\033[0m"


def log(msg, color=RESET):
    print(f"{color}{msg}{RESET}", flush=True)


def log_step(msg):
    log(f"\n[*] {msg}", BOLD + CYAN)


def log_ok(msg):
    log(f"    [OK] {msg}", GREEN)


def log_warn(msg):
    log(f"    [WARN] {msg}", YELLOW)


def log_err(msg):
    log(f"    [ERR] {msg}", RED)


# ---------------------------------------------------------------------------
# Subprocess helpers
# ---------------------------------------------------------------------------
def run(cmd, cwd=None, env=None, check=True, capture=False, hide_output=False,
        input=None, timeout=600):
    """Run a command, return CompletedProcess.

    capture: capture stdout/stderr for the caller to read (result.stdout).
    hide_output: suppress stdout/stderr (overridden by --verbose).
    """
    do_capture = capture or (hide_output and not VERBOSE)
    extra_env = None
    if env:
        extra_env = os.environ.copy()
        extra_env.update(env)
    result = subprocess.run(
        cmd,
        cwd=str(cwd) if cwd else None,
        env=extra_env,
        capture_output=do_capture,
        text=True,
        input=input,
        timeout=timeout,
    )
    if check and result.returncode != 0:
        stderr = result.stderr if do_capture else "(not captured)"
        raise RuntimeError(
            f"Command failed (exit {result.returncode}): {' '.join(str(c) for c in cmd)}\n{stderr}"
        )
    return result


def run_ssh(cmd_str, timeout=120, check=True):
    """Run a command on the git-ssh container as the git user."""
    return run(
        ["ssh", "-i", str(SSH_KEY),
         "-o", "StrictHostKeyChecking=no",
         "-o", "UserKnownHostsFile=/dev/null",
         "-o", "BatchMode=yes",
         "-p", str(SSH_PORT),
         f"git@localhost", cmd_str],
        capture=True, check=check, timeout=timeout,
    )


def compose(args, hide_output=True, env=None, check=True, timeout=120):
    """Run docker compose command in the experiment directory.

    Output is hidden by default to keep the log clean (visible with --verbose).
    """
    return run(
        ["docker", "compose"] + args,
        cwd=EXPERIMENT_DIR,
        hide_output=hide_output,
        env=env,
        check=check,
        timeout=timeout,
    )


def compose_exec(service, cmd_str, hide_output=True, check=True, timeout=120):
    """Run a command in a running docker compose service container."""
    return compose(
        ["exec", "-T", service, "sh", "-c", cmd_str],
        hide_output=hide_output,
        check=check,
        timeout=timeout,
    )


# ---------------------------------------------------------------------------
# Client container helpers
# ---------------------------------------------------------------------------
def host_path_to_client(host_path: Path) -> str:
    """Convert a host path (inside WORK_DIR) to its path in the client container."""
    rel = host_path.relative_to(WORK_DIR)
    return f"{WORK_DIR_IN_CLIENT}/{rel}"


def client_exec(cmd: list, cwd: str = None, env: dict = None,
                capture=False, check=True, timeout=3600) -> subprocess.CompletedProcess:
    """Run a command inside the client container via docker exec."""
    dc = ["docker", "exec"]
    if cwd:
        dc += ["--workdir", cwd]
    if env:
        for k, v in env.items():
            dc += ["-e", f"{k}={v}"]
    dc += [CLIENT_CONTAINER] + cmd
    return run(dc, capture=capture, hide_output=not capture, check=check, timeout=timeout)


def _batches(lst: list, size: int):
    """Yield successive non-overlapping slices of `lst` with at most `size` items."""
    for i in range(0, len(lst), size):
        yield lst[i:i + size]


def _pick_preferred_branch(branches: list) -> str:
    """Pick the best branch, preferring master/main, then alphabetical."""
    for b in sorted(branches):
        if "master" in b or "main" in b:
            return b
    return sorted(branches)[0]


def get_ordered_push_refspecs(source_repo: Path) -> list:
    """Return grouped push refspecs simulating a realistic CI workflow.

    Returns a list of lists: each inner list is a group of refspecs to
    push together in one ``git push`` invocation.

    For tags: pushes the tag itself plus the preferred containing branch
    (preferring master/main).  For branches: pushes the branch update
    unless it was already pushed to the same hash by a tag push.
    """
    result = run(
        ["git", "for-each-ref", "refs/heads/*", "refs/tags/*",
         "--format=%(creatordate:unix) %(objectname) %(refname)",
         "--sort=creatordate"],
        cwd=source_repo, capture=True,
    )
    refs = []
    for line in result.stdout.strip().splitlines():
        if not line:
            continue
        ts, sha, refname = line.split(None, 2)
        refs.append((int(ts), sha, refname))
    # Stable-sort: tags before branches at the same timestamp.
    refs.sort(key=lambda r: (r[0], 0 if r[2].startswith("refs/tags/") else 1))

    branch_pushed = {}  # branch name -> last pushed sha
    push_groups = []

    for _ts, sha, refname in refs:
        if refname.startswith("refs/tags/"):
            tag_name = refname[len("refs/tags/"):]
            group = [f"refs/tags/{tag_name}:refs/tags/{tag_name}"]
            # Also update the preferred branch containing this commit.
            r = run(
                ["git", "branch", "--contains", sha,
                 "--format=%(refname:short)"],
                cwd=source_repo, capture=True,
            )
            containing = [b for b in r.stdout.strip().splitlines() if b]
            if containing:
                branch = _pick_preferred_branch(containing)
                if branch_pushed.get(branch) != sha:
                    group.append(f"{sha}:refs/heads/{branch}")
                    branch_pushed[branch] = sha
            push_groups.append(group)
        elif refname.startswith("refs/heads/"):
            branch = refname[len("refs/heads/"):]
            if branch_pushed.get(branch) != sha:
                push_groups.append([f"{sha}:refs/heads/{branch}"])
                branch_pushed[branch] = sha
            # else: already pushed to this hash via a tag push — skip

    return push_groups


def check_lfs_endpoint(repo_dir: Path, extra_git_config: list = None):
    """Log the LFS endpoint that git-lfs selected for a repo (for debugging)."""
    cwd = host_path_to_client(repo_dir)
    cmd = ["git"]
    if extra_git_config:
        for kv in extra_git_config:
            cmd += ["-c", kv]
    cmd += ["lfs", "env"]
    try:
        r = client_exec(cmd, cwd=cwd, capture=True, check=False, timeout=10)
        for line in r.stdout.splitlines():
            if "Endpoint" in line or "SSH" in line:
                log(f"    LFS env: {line.strip()}")
    except Exception:
        pass


def _default_reclaim():
    """Reclaim page cache from core containers."""
    reclaim_page_cache(CLIENT_CONTAINER)
    reclaim_page_cache(GIT_SSH_CONTAINER)


def git_push_client(remote_url: str, source_repo: Path = None,
                    extra_git_config: list = None, extra_env: dict = None,
                    push_refspecs: list = None,
                    timeout: int = 3600,
                    container_cwd: str = None,
                    reclaim_fn=_default_reclaim,
                    trace_log: str = None,
                    stop_after: int = None) -> float:
    """Push source_repo to remote_url from inside the client container.

    Strategy (when push_refspecs is provided):
      Push one ref at a time. Each invocation creates a fresh SSH pool so the
      pool-shutdown bug (pool dies after first batch in a multi-branch push)
      never triggers. gc.auto=0 on the server repo prevents pack-file
      accumulation from causing lock/temp-dir errors.

    When None, falls back to --all for simple cases (synthetic dataset).

    container_cwd: explicit path inside the container (e.g. /data-src for a
    directly-mounted bare repo). When set, source_repo is ignored for path
    resolution.

    trace_log: when set, the *first* push invocation is run with GIT_TRACE=1,
    GIT_TRANSFER_TRACE=1, and GIT_CURL_VERBOSE=1, and all output (stdout +
    stderr) is saved to this path (on the host).  Subsequent pushes (if using
    push_refspecs) run without tracing.
    """
    cwd = container_cwd if container_cwd else host_path_to_client(source_repo)
    base_cmd = ["git",
                # Suppress LFS lock-verify round-trip — server doesn't support it.
                "-c", "lfs.locksverify=false",
                "-c", f"lfs.concurrenttransfers={GIT_LFS_CONCURRENCY}"]
    if extra_git_config:
        for kv in extra_git_config:
            base_cmd += ["-c", kv]

    trace_env = None
    if trace_log:
        trace_env = {
            "GIT_TRACE": "1",
            "GIT_TRANSFER_TRACE": "1",
            "GIT_CURL_VERBOSE": "1",
        }

    start = time.perf_counter()
    if push_refspecs:
        total = min(len(push_refspecs), stop_after) if stop_after else len(push_refspecs)
        ndots = total // 10
        if ndots > 0:
            print("." * ndots, end="\r", flush=True)
        for i, ref_group in enumerate(push_refspecs, 1):
            if stop_after and i > stop_after:
                break
            # ref_group is a list of refspecs to push together.
            if isinstance(ref_group, str):
                ref_group = [ref_group]  # backward compat
            # Trace only the first push.
            push_env = dict(extra_env) if extra_env else {}
            if trace_env and i == 1:
                push_env.update(trace_env)
            r = client_exec(
                base_cmd + ["push", remote_url] + ref_group,
                cwd=cwd, env=push_env or None, timeout=timeout,
            )
            if trace_log and i == 1:
                Path(trace_log).write_text(
                    f"=== TRACE: first push (refs={ref_group}) ===\n"
                    f"--- stdout ---\n{r.stdout}\n"
                    f"--- stderr ---\n{r.stderr}\n",
                    encoding="utf-8",
                )
                log(f"    [trace] Saved first-push trace to {trace_log}")
            if i % 10 == 0:
                print("*", end="", flush=True)
            if reclaim_fn and RECLAIM_INTERVAL > 0 and i % RECLAIM_INTERVAL == 0:
                reclaim_fn()
        if ndots > 0:
            print(flush=True)  # newline after progress
    else:
        push_env = dict(extra_env) if extra_env else {}
        if trace_env:
            push_env.update(trace_env)
        r = client_exec(
            base_cmd + ["push", "--all", remote_url],
            cwd=cwd, env=push_env or None, timeout=timeout,
        )
        if trace_log:
            Path(trace_log).write_text(
                f"=== TRACE: push --all ===\n"
                f"--- stdout ---\n{r.stdout}\n"
                f"--- stderr ---\n{r.stderr}\n",
                encoding="utf-8",
            )
            log(f"    [trace] Saved push trace to {trace_log}")
    return time.perf_counter() - start


def git_clone_client(remote_url: str, dest: Path,
                     extra_git_config: list = None, extra_env: dict = None,
                     timeout: int = 3600) -> float:
    """Clone remote_url into dest from inside the client container.

    dest is a path on the host (inside WORK_DIR), accessible in the container
    via the shared mount. The directory must not exist before cloning.
    """
    if dest.exists():
        shutil.rmtree(dest)
    dest_client = host_path_to_client(dest)
    cmd = ["git",
           "-c", f"lfs.concurrenttransfers={GIT_LFS_CONCURRENCY}"]
    if extra_git_config:
        for kv in extra_git_config:
            cmd += ["-c", kv]
    cmd += ["clone", remote_url, dest_client]
    start = time.perf_counter()
    client_exec(cmd, cwd=WORK_DIR_IN_CLIENT, env=extra_env, timeout=timeout)
    return time.perf_counter() - start


def git_lfs_pull_client(repo: Path, extra_git_config: list = None,
                        extra_env: dict = None,
                        extra_fetch_refspecs: list = None,
                        use_ssh_pull: bool = False,
                        timeout: int = 3600) -> float:
    """Fetch all LFS objects (all refs) and check out the working tree.

    Fetches git refs/heads/* from origin so all branches are local, then
    fetches all LFS objects with --all (traverses full history, not just tips).

    use_ssh_pull: when True, use 'git lfs pull' per branch instead of
    'git lfs fetch --all'.  This forces the SSH transfer adapter which
    is needed when the LFS server only supports SSH pkt-line transfer
    (no HTTP batch API endpoint).
    """
    cwd = host_path_to_client(repo)
    base_cmd = ["git",
                "-c", f"lfs.concurrenttransfers={GIT_LFS_CONCURRENCY}"]
    if extra_git_config:
        for kv in extra_git_config:
            base_cmd += ["-c", kv]
    start = time.perf_counter()
    # Populate refs/heads/* so all branches are reachable (not just main).
    client_exec(
        base_cmd + ["fetch", "origin", "+refs/heads/*:refs/heads/*", "--update-head-ok"],
        cwd=cwd, env=extra_env, timeout=timeout,
    )
    # Switch to the 'main' branch if it exists (Forgejo repos may default to
    # 'master' which can be empty; the R0 content is on 'main').
    client_exec(
        ["git", "checkout", "main"],
        cwd=cwd, check=False, timeout=600,
    )
    if use_ssh_pull:
        # Use 'git lfs pull' which goes through the SSH transfer adapter.
        # Pull for the current branch (fetches LFS objects via pkt-line).
        client_exec(
            base_cmd + ["lfs", "pull"],
            cwd=cwd, env=extra_env, timeout=timeout,
        )
    else:
        # Fetch all LFS objects reachable from any ref (full history traversal).
        # Uses the HTTP batch API.
        client_exec(
            base_cmd + ["lfs", "fetch", "--all", "origin"],
            cwd=cwd, env=extra_env, timeout=timeout,
        )
        client_exec(base_cmd + ["lfs", "checkout"], cwd=cwd, env=extra_env, timeout=timeout)
    return time.perf_counter() - start


def client_remote_url(repo_name: str) -> str:
    """SSH URL for a bare repo on git-ssh, reachable from inside the client container."""
    return f"ssh://git@{GIT_SSH_INTERNAL}/home/git/{repo_name}.git"


# ---------------------------------------------------------------------------
# Binary building
# ---------------------------------------------------------------------------
def build_binaries():
    log_step("Building binaries")
    BINARIES_DIR.mkdir(exist_ok=True)

    # Build desync git-lfs-transfer (from source, linux/amd64 for server container)
    dest = BINARIES_DIR / "git-lfs-transfer"
    log(f"  Building desync git-lfs-transfer → {dest}")
    run(
        ["go", "build", "-o", str(dest), "./cmd/git-lfs-transfer"],
        cwd=DESYNC_SRC,
        env={"CGO_ENABLED": "0", "GOOS": "linux", "GOARCH": "amd64"},
    )
    os.chmod(dest, 0o755)
    log_ok("desync git-lfs-transfer built")

    # Build desync git-lfs-desync (linux/amd64 for client container)
    dest_agent = BINARIES_DIR / "git-lfs-desync"
    log(f"  Building desync git-lfs-desync → {dest_agent}")
    run(
        ["go", "build", "-o", str(dest_agent), "./cmd/git-lfs-desync"],
        cwd=DESYNC_SRC,
        env={"CGO_ENABLED": "0", "GOOS": "linux", "GOARCH": "amd64"},
    )
    os.chmod(dest_agent, 0o755)
    log_ok("desync git-lfs-desync built")

    # Build the baseline (S1) git-lfs-transfer implementation.
    if S1_LFS_TRANSFER == "scutiger":
        # Rust, needs musl target for Alpine containers.
        s1_dest = BINARIES_DIR / "s1-git-lfs-transfer"
        if SCUTIGER_SRC.exists():
            log(f"  Building scutiger git-lfs-transfer → {s1_dest}")
            run(
                ["cargo", "build", "--release", "--target", "x86_64-unknown-linux-musl", "-p", "scutiger-lfs"],
                cwd=SCUTIGER_SRC, hide_output=True,
            )
            built = SCUTIGER_SRC / "target" / "x86_64-unknown-linux-musl" / "release" / "git-lfs-transfer"
            if not built.exists():
                raise RuntimeError(f"scutiger binary not found at {built}")
            shutil.copy2(built, s1_dest)
            os.chmod(s1_dest, 0o755)
            log_ok("scutiger git-lfs-transfer built")
        else:
            log_warn(f"scutiger source not found at {SCUTIGER_SRC}, skipping build")
    elif S1_LFS_TRANSFER == "charmbracelet":
        s1_dest = BINARIES_DIR / "s1-git-lfs-transfer"
        if CHARM_SRC.exists() and (CHARM_SRC / "go.mod").exists():
            log(f"  Building charmbracelet git-lfs-transfer → {s1_dest}")
            run(
                ["go", "build", "-o", str(s1_dest), "."],
                cwd=CHARM_SRC,
                env={"CGO_ENABLED": "0", "GOOS": "linux", "GOARCH": "amd64"},
            )
            os.chmod(s1_dest, 0o755)
            log_ok("charmbracelet git-lfs-transfer built")
        else:
            log_warn(f"charmbracelet source not found at {CHARM_SRC}, skipping build")
    else:
        log_warn(f"Unknown S1_LFS_TRANSFER={S1_LFS_TRANSFER!r}, skipping S1 build")

    # Build custom git-lfs with transfer negotiation support.
    lfs_dest = BINARIES_DIR / "git-lfs"
    if GIT_LFS_SRC.exists() and (GIT_LFS_SRC / "go.mod").exists():
        log(f"  Building custom git-lfs → {lfs_dest}")
        run(
            ["go", "build", "-o", str(lfs_dest), "."],
            cwd=GIT_LFS_SRC,
            env={"CGO_ENABLED": "0", "GOOS": "linux", "GOARCH": "amd64"},
        )
        os.chmod(lfs_dest, 0o755)
        log_ok("custom git-lfs built")
    elif not lfs_dest.exists():
        log_warn(f"git-lfs source not found at {GIT_LFS_SRC} and no pre-built binary")

    # Build the custom Forgejo Docker image if not already present.
    r = run(["docker", "image", "inspect", "forgejo-desync:local"], hide_output=True, check=False)
    if r.returncode != 0:
        if FORGEJO_SRC.exists() and (FORGEJO_SRC / "Dockerfile").exists():
            log("  Building forgejo-desync:local Docker image (this may take several minutes)...")
            # Initialize Forgejo's nested desync submodule if needed.
            run(["git", "submodule", "update", "--init"], cwd=FORGEJO_SRC, hide_output=True, check=False)
            # The Makefile sets FORGEJO_VERSION from a VERSION file (git tags
            # aren't available inside the Docker build context, and
            # .dockerignore excludes /VERSION).  Write it temporarily and
            # remove the .dockerignore exclusion during the build.
            version_file = FORGEJO_SRC / "VERSION"
            dockerignore = FORGEJO_SRC / ".dockerignore"
            created_version = not version_file.exists()
            if created_version:
                version = "0.0.0-dev"
                r = run(["git", "describe", "--tags", "--always"], cwd=FORGEJO_SRC, capture=True, check=False)
                if r.returncode == 0 and r.stdout.strip():
                    version = r.stdout.strip().lstrip("v")
                version_file.write_text(version)
            di_backup = None
            if dockerignore.exists():
                di_backup = dockerignore.read_text()
                if "/VERSION" in di_backup:
                    dockerignore.write_text(di_backup.replace("/VERSION\n", "").replace("/VERSION", ""))
            try:
                run(
                    ["docker", "build", "-t", "forgejo-desync:local", str(FORGEJO_SRC)],
                    timeout=1200,
                )
            finally:
                if created_version:
                    version_file.unlink(missing_ok=True)
                if di_backup is not None:
                    dockerignore.write_text(di_backup)
            log_ok("forgejo-desync:local image built")
        else:
            log_warn(f"Forgejo source not found at {FORGEJO_SRC}, scenarios 8-12 will not work")
    else:
        log_ok("forgejo-desync:local image already exists")


# ---------------------------------------------------------------------------
# SSH key management
# ---------------------------------------------------------------------------
def setup_ssh_keys():
    log_step("Setting up SSH keys")
    KEYS_DIR.mkdir(exist_ok=True)
    if not SSH_KEY.exists():
        run(["ssh-keygen", "-t", "ed25519", "-f", str(SSH_KEY), "-N", "", "-q"])
        log_ok(f"Generated SSH key pair at {SSH_KEY}")
    else:
        log_ok("SSH key pair already exists")


# ---------------------------------------------------------------------------
# Garage config generation
# ---------------------------------------------------------------------------
RPC_SECRET = None  # generated once


def generate_garage_toml():
    """Generate a single garage.toml shared by all Garage instances."""
    rpc_secret = subprocess.check_output(
        ["openssl", "rand", "-hex", "32"], text=True
    ).strip()
    content = f"""metadata_dir = "/garage-meta"
data_dir     = "/garage-data"
replication_factor = 1
rpc_bind_addr   = "[::]:3901"
rpc_public_addr = "127.0.0.1:3901"
rpc_secret      = "{rpc_secret}"

compression_level = "none"
block_ram_buffer_max = "512MiB"
block_max_concurrent_reads = 32

[s3_api]
s3_region    = "garage"
api_bind_addr = "[::]:3900"
root_domain  = ".s3.garage"
"""
    GARAGE_TOML.write_text(content)
    log_ok(f"Generated Garage config: {GARAGE_TOML.name}")


# ---------------------------------------------------------------------------
# Docker compose lifecycle
# ---------------------------------------------------------------------------
def start_services(work_dir: Path, lfs_dir: Path = None):
    log_step("Starting Docker services")
    generate_garage_toml()
    BINARIES_DIR.mkdir(exist_ok=True)
    KEYS_DIR.mkdir(exist_ok=True)

    # Create placeholder binaries if they don't exist (for docker volume mounts)
    for name in ["git-lfs-transfer", "s1-git-lfs-transfer", "git-lfs-desync", "git-lfs"]:
        p = BINARIES_DIR / name
        if not p.exists():
            p.write_bytes(b"placeholder")

    # Create placeholder authorized_keys
    auth_keys = KEYS_DIR / "authorized_keys"
    if not auth_keys.exists():
        auth_keys.write_text("")

    # Write .env so that manual `docker compose` invocations also get LFS_WORK_DIR.
    env_lines = f"LFS_WORK_DIR={work_dir}\n"
    compose_env = {"LFS_WORK_DIR": str(work_dir)}
    if lfs_dir:
        env_lines += f"LFS_LFS_DIR={lfs_dir}\n"
        compose_env["LFS_LFS_DIR"] = str(lfs_dir)
    (EXPERIMENT_DIR / ".env").write_text(env_lines)
    compose(["up", "--build", "-d"], env=compose_env, timeout=300)
    log_ok("Core services started (git-ssh + client)")

    _wait_ssh_ready()


def _wait_ssh_ready():
    """Wait for SSH service to accept connections."""
    log("  Waiting for SSH service...")
    for _ in range(30):
        try:
            result = run(
                ["ssh", "-i", str(SSH_KEY),
                 "-o", "StrictHostKeyChecking=no",
                 "-o", "UserKnownHostsFile=/dev/null",
                 "-o", "ConnectTimeout=2",
                 "-o", "BatchMode=yes",
                 "-p", str(SSH_PORT),
                 "git@localhost", "echo ok"],
                hide_output=True, check=False, timeout=5,
            )
            if result.returncode == 0:
                log_ok("SSH ready")
                return
        except Exception:
            pass
        time.sleep(1)
    log_warn("SSH may not be ready yet, continuing anyway")


# S3 services are started on demand per scenario to limit resource usage.
# Each service has a profile "s3" in docker-compose.yml and must be started
# explicitly.  Track which services have been set up to avoid re-running
# idempotent-but-slow setup steps (Garage layout apply, bucket creation).
_s3_setup_done: set = set()


def _wait_s3_healthy(service: str, timeout_s: int = 30):
    """Wait until an S3 service responds to health/status checks."""
    if service == "shared-minio":
        for _ in range(timeout_s):
            try:
                urllib.request.urlopen(f"{SHARED_MINIO_ENDPOINT}/minio/health/live", timeout=2)
                return
            except Exception:
                time.sleep(1)
        log_warn(f"{service} health check timed out after {timeout_s}s")
        return

    if service == "shared-garage":
        for _ in range(timeout_s):
            try:
                r = _shared_garage(["/garage", "status"])
                if r.returncode == 0:
                    return
            except Exception:
                pass
            time.sleep(1)
        log_warn(f"{service} status check timed out after {timeout_s}s")


def start_s3_service(service: str, setup_fn=None):
    """Start an S3-profile service and run its setup function if needed."""
    log(f"  Starting {service}...")
    compose(["--profile", "s3", "up", "-d", service], timeout=600)
    _wait_s3_healthy(service)
    if service not in _s3_setup_done and setup_fn:
        setup_fn()
        _s3_setup_done.add(service)
    log_ok(f"{service} ready")


def stop_s3_service(service: str):
    """Stop an S3 service to free resources."""
    log(f"  Stopping {service}...")
    compose(["--profile", "s3", "stop", service], timeout=120)
    log_ok(f"{service} stopped")


def wipe_shared_minio():
    """Stop, remove, wipe volume, and restart the shared MinIO instance."""
    log("  Wiping shared-minio volume...")
    compose(["--profile", "s3", "stop", "shared-minio"], check=False, timeout=120)
    compose(["--profile", "s3", "rm", "-f", "shared-minio"], check=False, timeout=120)
    run(["docker", "volume", "rm", "-f", f"{COMPOSE_PROJECT}_shared-minio-data"],
        hide_output=True, check=False, timeout=30)
    _s3_setup_done.discard("shared-minio")
    _container_full_ids.pop(SHARED_MINIO_CONTAINER, None)
    log_ok("shared-minio volume wiped")


def wipe_shared_garage():
    """Stop, remove, wipe volumes, and restart the shared Garage instance."""
    global SHARED_GARAGE_KEY_ID, SHARED_GARAGE_SECRET
    log("  Wiping shared-garage volumes...")
    compose(["--profile", "s3", "stop", "shared-garage"], check=False, timeout=120)
    compose(["--profile", "s3", "rm", "-f", "shared-garage"], check=False, timeout=120)
    run(["docker", "volume", "rm", "-f", f"{COMPOSE_PROJECT}_shared-garage-data"],
        hide_output=True, check=False, timeout=30)
    run(["docker", "volume", "rm", "-f", f"{COMPOSE_PROJECT}_shared-garage-meta"],
        hide_output=True, check=False, timeout=30)
    _s3_setup_done.discard("shared-garage")
    _container_full_ids.pop(SHARED_GARAGE_CONTAINER, None)
    SHARED_GARAGE_KEY_ID = None
    SHARED_GARAGE_SECRET = None
    log_ok("shared-garage volumes wiped")


_container_full_ids: dict = {}


def _get_container_full_id(container_name: str) -> str:
    """Return cached full container ID, refreshing on first call or after restart."""
    if container_name not in _container_full_ids:
        r = run(["docker", "inspect", container_name, "--format", "{{.Id}}"],
                capture=True, check=False, timeout=10)
        if r.returncode == 0:
            _container_full_ids[container_name] = r.stdout.strip()
    return _container_full_ids.get(container_name, "")


def reclaim_all_page_cache(*extra):
    """Reclaim page cache from core containers and any extras."""
    for name in (CLIENT_CONTAINER, GIT_SSH_CONTAINER) + extra:
        reclaim_page_cache(name)


def reclaim_page_cache(container_name: str):
    """Force kernel page cache reclaim for a container via cgroup v2 memory.reclaim.

    Requires /sys/fs/cgroup mounted as /host-cgroup inside the container.
    The write returns EAGAIN if less memory was reclaimable than requested,
    which is expected and not an error.
    """
    full_id = _get_container_full_id(container_name)
    if not full_id:
        return
    run(["docker", "exec", "-u", "root", container_name,
         "sh", "-c", f'echo 100G > /host-cgroup/docker/{full_id}/memory.reclaim'],
        check=False, hide_output=True, timeout=10)


def start_core_containers():
    """Start (not recreate) the core containers and wait for SSH."""
    log("  Starting core containers …")
    compose(["start", "client", "git-ssh"], timeout=120)
    _wait_ssh_ready()


def stop_all_containers():
    """Stop all running containers (preserving state and volumes)."""
    compose(["--profile", "s3", "--profile", "forgejo", "stop"], check=False, timeout=120)
    log_ok("Containers stopped")


def stop_services():
    """Remove all containers and volumes (final cleanup)."""
    log_step("Stopping Docker services")
    compose(["--profile", "s3", "--profile", "forgejo", "down"], check=False, timeout=300)
    log_ok("Services stopped")


def write_client_desync_config():
    """Write a desync config file on the client with local-only settings.

    Written to the default config location ($HOME/.config/desync/config.json)
    so the agent picks it up automatically without --config flag.

    This config provides client-side settings (cache, concurrency, resource
    limits) that are NOT part of the server-advertised config.  The server
    provides store URLs, credentials, chunk-size, and digest via the SSH
    transfer negotiation protocol; MergeServerConfig() combines the two.
    """
    config = {
        "defaults": {
            "cache": CLIENT_DESYNC_CACHE,
            "concurrency": DESYNC_CONCURRENCY,
            "conn-pool-size": 100,
        },
    }
    if MAX_IN_FLIGHT > 0:
        config["defaults"]["max-in-flight"] = MAX_IN_FLIGHT
    if MAX_STORAGE_OPS > 0:
        config["defaults"]["max-storage-ops"] = MAX_STORAGE_OPS
    config_json = json.dumps(config, indent=2)
    config_path = CLIENT_DESYNC_CONFIG
    client_exec(["mkdir", "-p", str(Path(config_path).parent)])
    client_exec(
        ["python3", "-c",
         f"import sys; open('{config_path}', 'w').write(sys.argv[1])",
         config_json],
    )
    # Pre-create the cache directory (on ext4 bind mount, not overlayfs).
    client_exec(["mkdir", "-p", CLIENT_DESYNC_CACHE])
    log_ok(f"Client desync config: cache={CLIENT_DESYNC_CACHE}, N={DESYNC_CONCURRENCY}")


def setup_containers():
    """One-time container setup: SSH keys, git config, repo cleanup.

    Called once after start_services.  After this, containers are stopped
    and subsequently started/stopped per scenario without recreation.
    """
    # Install SSH authorized keys
    pubkey = SSH_PUBKEY.read_text().strip()
    auth_keys_file = KEYS_DIR / "authorized_keys"
    if os.environ.get("DESYNC_HEAP_PROFILE_DIR", ""):
        pubkey = f'environment="DESYNC_HEAP_PROFILE_DIR=/home/git/profiles" {pubkey}'
    if os.environ.get("LFS_TRANSFER_TRACE_DIR", ""):
        pubkey = f'environment="LFS_TRANSFER_TRACE_DIR=/home/git/profiles" {pubkey}'
    auth_keys_file.write_text(pubkey + "\n")
    # Restart git-ssh once to pick up the new authorized_keys
    compose(["restart", "git-ssh"], timeout=30)
    _wait_ssh_ready()
    log_ok("SSH authorized_keys installed")

    # Disable automatic git GC in both containers
    run_ssh("git config --global gc.auto 0")
    client_exec(["git", "config", "--global", "gc.auto", "0"])
    log_ok("Disabled git gc.auto in client and git-ssh")

    # Create TMPDIR for git-lfs-desync (same mount as /work to avoid cross-device rename)
    client_exec(["mkdir", "-p", WORK_DIR_IN_CLIENT + "/.tmp"])

    # Write client-side desync config with local-only settings (cache, concurrency).
    # The server provides URLs, credentials, and chunk-size via the SSH transfer
    # negotiation protocol; the client config handles everything else.
    write_client_desync_config()

    # Clean any leftover repos from previous runs
    run_ssh("rm -rf /home/git/repo*.git", check=False)
    log_ok("Container setup complete")


# ---------------------------------------------------------------------------
# Garage setup
# ---------------------------------------------------------------------------
SHARED_GARAGE_KEY_ID = None
SHARED_GARAGE_SECRET = None


def measure_volume_storage(volume_name: str) -> int:
    """Return bytes stored in a Docker named volume.

    Resolves the volume name to a container+path and runs `du` inside the
    container — much faster than /system/df which scans all volumes.
    Falls back to the Docker API on unknown volume names. Returns -1 on failure.
    """
    # Map volume name suffix → (container, path inside container)
    _vol_map = {
        "shared-minio-data":  (SHARED_MINIO_CONTAINER, "/data"),
        "shared-garage-data": (SHARED_GARAGE_CONTAINER, "/garage-data"),
        "shared-garage-meta": (SHARED_GARAGE_CONTAINER, "/garage-meta"),
    }
    # Strip compose project prefix (e.g. "lfs-exp_s3-minio-data" → "s3-minio-data")
    suffix = volume_name.split("_", 1)[-1] if "_" in volume_name else volume_name
    if suffix in _vol_map:
        container, path = _vol_map[suffix]
        try:
            r = subprocess.run(
                ["docker", "exec", container, "du", "-sk", path],
                capture_output=True, text=True, timeout=120,
            )
            if r.returncode == 0:
                return int(r.stdout.strip().split()[0]) * 1024
            # Container may lack du (e.g. Garage scratch image) — use Alpine sidecar
            r2 = subprocess.run(
                ["docker", "run", "--rm", "-v", f"{volume_name}:/mnt", "alpine",
                 "du", "-sk", "/mnt"],
                capture_output=True, text=True, timeout=120,
            )
            if r2.returncode == 0:
                return int(r2.stdout.strip().split()[0]) * 1024
            log_warn(f"du in {container}:{path} failed: {r.stderr.strip()}")
            return -1
        except Exception as e:
            log_warn(f"du in {container}:{path} unavailable: {e}")
            return -1
    log_warn(f"Unknown volume {volume_name!r}, cannot measure storage")
    return -1


def _shared_garage(cmd_args: list, check=False) -> subprocess.CompletedProcess:
    """Run a garage CLI command inside the shared Garage container."""
    return run(
        ["docker", "exec", SHARED_GARAGE_CONTAINER] + cmd_args,
        capture=True, check=check, timeout=30,
    )


def _garage_setup(garage_fn, label: str, layout_version: int):
    """Shared Garage bootstrap: layout assign + key create + bucket create.

    Returns (key_id, secret) or (None, None) on failure.
    """
    # Wait for the Garage daemon to be ready (retry up to 30s)
    node_id = None
    for _ in range(30):
        status = garage_fn(["/garage", "status"])
        for line in status.stdout.splitlines():
            parts = line.split()
            if parts and len(parts[0]) == 16 and all(c in "0123456789abcdef" for c in parts[0]):
                node_id = parts[0]
                break
        if node_id:
            break
        time.sleep(1)

    if not node_id:
        log_warn(f"Could not find {label} node ID after 30s; skipping layout setup\n{status.stdout}")
    else:
        garage_fn(["/garage", "layout", "assign", "-z", "dc1", "-c", "1G", node_id])
        time.sleep(1)
        garage_fn(["/garage", "layout", "apply", "--version", str(layout_version)])
        time.sleep(2)

    result = garage_fn(["/garage", "key", "create", "lfs-key"])
    if result.returncode != 0:
        result = garage_fn(["/garage", "key", "info", "lfs-key"])

    key_id, secret = None, None
    for line in result.stdout.splitlines():
        line = line.strip()
        if line.startswith("Key ID:"):
            key_id = line.split(":", 1)[1].strip()
        elif line.startswith("Secret key:"):
            secret = line.split(":", 1)[1].strip()

    if not key_id or not secret:
        log_warn(f"Could not parse {label} key credentials:\n{result.stdout}")
        return None, None

    log_ok(f"{label} API key: {key_id}")
    bucket = "lfs"
    garage_fn(["/garage", "bucket", "create", bucket])
    garage_fn(["/garage", "bucket", "allow", "--read", "--write", bucket, "--key", key_id])
    log_ok(f"{label} bucket created: {bucket}")
    return key_id, secret


def setup_shared_garage():
    """Set up the shared Garage instance (layout + key + bucket)."""
    log_step("Setting up shared Garage")
    global SHARED_GARAGE_KEY_ID, SHARED_GARAGE_SECRET
    SHARED_GARAGE_KEY_ID, SHARED_GARAGE_SECRET = _garage_setup(_shared_garage, "shared-garage", layout_version=1)


# ---------------------------------------------------------------------------
# MinIO setup
# ---------------------------------------------------------------------------
def _minio_create_bucket(bucket: str, endpoint: str, user: str, password: str):
    """Create a MinIO bucket using the S3 API (PUT bucket)."""
    import urllib.error
    from urllib.parse import urlparse
    parsed = urlparse(endpoint)
    host = parsed.netloc
    url = f"{endpoint}/{bucket}"
    t = time.gmtime()
    amz_date = time.strftime("%Y%m%dT%H%M%SZ", t)
    date_stamp = time.strftime("%Y%m%d", t)
    region = "us-east-1"

    payload_hash = hashlib.sha256(b"").hexdigest()
    canonical_headers = f"host:{host}\nx-amz-date:{amz_date}\n"
    signed_headers = "host;x-amz-date"
    canonical_request = (
        f"PUT\n/{bucket}\n\n{canonical_headers}\n{signed_headers}\n{payload_hash}"
    )
    credential_scope = f"{date_stamp}/{region}/s3/aws4_request"
    string_to_sign = (
        f"AWS4-HMAC-SHA256\n{amz_date}\n{credential_scope}\n"
        + hashlib.sha256(canonical_request.encode()).hexdigest()
    )
    signing_key = _aws_signing_key(password, date_stamp, region, "s3")
    signature = hmac.new(signing_key, string_to_sign.encode(), hashlib.sha256).hexdigest()
    auth_header = (
        f"AWS4-HMAC-SHA256 Credential={user}/{credential_scope}, "
        f"SignedHeaders={signed_headers}, Signature={signature}"
    )
    req = urllib.request.Request(url, data=b"", method="PUT")
    req.add_header("x-amz-date", amz_date)
    req.add_header("Authorization", auth_header)
    for attempt in range(5):
        try:
            urllib.request.urlopen(req, timeout=10)
            return
        except urllib.error.HTTPError as e:
            if e.code == 409:
                return  # bucket already exists
            raise
        except (ConnectionError, OSError):
            if attempt < 4:
                time.sleep(2)
                continue
            raise


def _wait_minio(endpoint: str, label: str):
    """Wait for a MinIO instance to be healthy."""
    for _ in range(30):
        try:
            urllib.request.urlopen(f"{endpoint}/minio/health/live", timeout=2)
            return
        except Exception:
            time.sleep(1)
    log_warn(f"{label} health check timed out")


def setup_shared_minio():
    """Create the 'lfs' bucket on the shared MinIO instance."""
    log_step("Setting up shared MinIO")
    _minio_create_bucket("lfs", SHARED_MINIO_ENDPOINT, MINIO_USER, MINIO_PASS)
    log_ok("MinIO bucket created: lfs")




def _generate_desync_lfs_json(filename: str, conn_pool_size: int = 100):
    """Generate a desync-lfs.json config file for a Forgejo instance using shared Garage."""
    defaults = {
        "stores": [f"s3+{SHARED_GARAGE_INTERNAL}/lfs/chunks/"],
        "index-store": f"s3+{SHARED_GARAGE_INTERNAL}/lfs/index/",
        "chunk-size": CHUNK_SIZE,
        "cache": "/data/gitea/desync-cache",
    }
    if MAX_IN_FLIGHT > 0:
        defaults["max-in-flight"] = MAX_IN_FLIGHT
    if MAX_STORAGE_OPS > 0:
        defaults["max-storage-ops"] = MAX_STORAGE_OPS
    if conn_pool_size > 0:
        defaults["conn-pool-size"] = conn_pool_size
    config = {
        "defaults": defaults,
        "s3-credentials": {
            SHARED_GARAGE_INTERNAL: {
                "access-key": SHARED_GARAGE_KEY_ID,
                "secret-key": SHARED_GARAGE_SECRET,
                "aws-region": "garage",
            }
        },
    }
    (EXPERIMENT_DIR / filename).write_text(json.dumps(config, indent=2))
    log_ok(f"Generated {filename}")


def setup_s9_garage():
    """Set up shared Garage for scenario 9 and generate its desync-lfs.json."""
    setup_shared_garage()
    if not SHARED_GARAGE_KEY_ID:
        log_warn("Garage setup failed — scenario 9 will not work")
        return
    _generate_desync_lfs_json("s9-desync-lfs.json")


def setup_s10_garage():
    """Set up shared Garage for scenario 10 and generate its desync-lfs.json."""
    setup_shared_garage()
    if not SHARED_GARAGE_KEY_ID:
        log_warn("Garage setup failed — scenario 10 will not work")
        return
    _generate_desync_lfs_json("s10-desync-lfs.json")


def setup_s12_garage():
    """Set up shared Garage for scenario 12 and generate its desync-lfs.json with conn-pool-size=100."""
    setup_shared_garage()
    if not SHARED_GARAGE_KEY_ID:
        log_warn("Garage setup failed — scenario 12 will not work")
        return
    _generate_desync_lfs_json("s12-desync-lfs.json", conn_pool_size=100)


# ---------------------------------------------------------------------------
# Forgejo helpers (scenarios 8-12)
# ---------------------------------------------------------------------------
_forgejo_setup_done = set()


def _forgejo_api(api_base: str, method: str, path: str, data=None):
    """Make an authenticated Forgejo API request (retries on connection errors)."""
    import base64
    auth = base64.b64encode(f"{FORGEJO_USER}:{FORGEJO_PASS}".encode()).decode()
    for attempt in range(5):
        req = urllib.request.Request(
            f"{api_base}/api/v1{path}",
            data=json.dumps(data).encode() if data else None,
            headers={
                "Authorization": f"Basic {auth}",
                "Content-Type": "application/json",
                "Accept": "application/json",
            },
            method=method,
        )
        try:
            with urllib.request.urlopen(req, timeout=30) as resp:
                return json.loads(resp.read().decode()) if resp.status in (200, 201) else {}
        except urllib.error.HTTPError as e:
            body = e.read().decode() if e.fp else ""
            log_warn(f"Forgejo API {method} {path}: {e.code} {body[:200]}")
            return {}
        except (ConnectionError, OSError):
            if attempt < 4:
                time.sleep(2)
                continue
            raise


def _wait_forgejo_healthy(api_base: str, label: str, timeout_s: int = 60):
    """Wait until Forgejo's API responds."""
    for _ in range(timeout_s):
        try:
            urllib.request.urlopen(f"{api_base}/api/v1/version", timeout=2)
            return
        except Exception:
            time.sleep(1)
    log_warn(f"{label} health check timed out after {timeout_s}s")


def setup_forgejo_instance(container: str, api_base: str, label: str):
    """One-time init for a Forgejo instance: create admin, add SSH key, create repos."""
    if container in _forgejo_setup_done:
        return
    log_step(f"Setting up {label}")

    # Create admin user via CLI (idempotent — errors if already exists)
    run(["docker", "exec", "-u", "git", container,
         "forgejo", "admin", "user", "create",
         "--admin", "--username", FORGEJO_USER,
         "--password", FORGEJO_PASS,
         "--email", FORGEJO_EMAIL],
        hide_output=True, check=False, timeout=30)

    # Add SSH public key via API
    pubkey = SSH_PUBKEY.read_text().strip()
    _forgejo_api(api_base, "POST", "/user/keys", {
        "title": "lfs-test-key",
        "key": pubkey,
    })
    log_ok(f"{label}: admin user + SSH key configured")
    _forgejo_setup_done.add(container)


def _forgejo_create_repo(api_base: str, repo_name: str):
    """Create a repository via Forgejo API."""
    _forgejo_api(api_base, "POST", "/user/repos", {
        "name": repo_name,
        "auto_init": False,
        "default_branch": "main",
    })


def _forgejo_remote_url(host: str, repo_name: str) -> str:
    """SSH remote URL for a Forgejo repo (from inside the client container)."""
    return f"ssh://git@{host}/{FORGEJO_USER}/{repo_name}.git"


# --- Gitea helpers (shared REST API, different admin CLI binary name) ---

_gitea_setup_done: set = set()


def setup_gitea_instance(container: str, api_base: str, label: str):
    """One-time init for a Gitea instance: create admin, add SSH key."""
    if container in _gitea_setup_done:
        return
    log_step(f"Setting up {label}")

    run(["docker", "exec", "-u", "git", container,
         "gitea", "admin", "user", "create",
         "--admin", "--username", FORGEJO_USER,
         "--password", FORGEJO_PASS,
         "--email", FORGEJO_EMAIL],
        hide_output=True, check=False, timeout=30)

    pubkey = SSH_PUBKEY.read_text().strip()
    _forgejo_api(api_base, "POST", "/user/keys", {
        "title": "lfs-test-key",
        "key": pubkey,
    })
    log_ok(f"{label}: admin user + SSH key configured")
    _gitea_setup_done.add(container)


def _aws_sign(key, msg):
    return hmac.new(key, msg.encode("utf-8"), hashlib.sha256).digest()


def _aws_signing_key(secret_key, date_stamp, region, service):
    k_date = _aws_sign(("AWS4" + secret_key).encode("utf-8"), date_stamp)
    k_region = _aws_sign(k_date, region)
    k_service = _aws_sign(k_region, service)
    k_signing = _aws_sign(k_service, "aws4_request")
    return k_signing


def get_net_io(container_name: str) -> tuple[int, int]:
    """Return cumulative (rx_bytes, tx_bytes) for a container via the Docker API.

    Uses the Unix socket at /var/run/docker.sock with stream=false, which returns
    a single stats snapshot without keeping the connection open. Returns (-1, -1)
    if the container cannot be queried (e.g. not running, socket unavailable).
    """
    class _UnixConn(http.client.HTTPConnection):
        def connect(self):
            self.sock = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
            self.sock.settimeout(30)
            self.sock.connect("/var/run/docker.sock")

    try:
        conn = _UnixConn("localhost")
        conn.request("GET", f"/containers/{container_name}/stats?stream=false")
        resp = conn.getresponse()
        if resp.status != 200:
            log_warn(f"Docker stats HTTP {resp.status} for {container_name}")
            return -1, -1
        data = json.loads(resp.read())
        networks = data.get("networks", {})
        if not networks:
            # Container may not have a network interface yet
            return -1, -1
        rx = sum(n.get("rx_bytes", 0) for n in networks.values())
        tx = sum(n.get("tx_bytes", 0) for n in networks.values())
        return rx, tx
    except Exception as e:
        log_warn(f"Docker net stats unavailable for {container_name}: {e}")
        return -1, -1


def _net_delta(before: tuple[int, int], after: tuple[int, int]) -> tuple[int, int]:
    """Compute (rx_delta, tx_delta), returning -1 for any unavailable component."""
    rx = after[0] - before[0] if before[0] >= 0 and after[0] >= 0 else -1
    tx = after[1] - before[1] if before[1] >= 0 and after[1] >= 0 else -1
    return rx, tx


def _net_total(before: tuple[int, int], after: tuple[int, int]) -> int:
    """Compute total (RX+TX) delta, returning -1 if unavailable."""
    rx, tx = _net_delta(before, after)
    if rx < 0 and tx < 0:
        return -1
    return (rx if rx >= 0 else 0) + (tx if tx >= 0 else 0)


def _net_rxtx(before: tuple[int, int], after: tuple[int, int]) -> list:
    """Compute [rx, tx] delta as a list."""
    rx, tx = _net_delta(before, after)
    return [rx, tx]


def _net_sum(rxtx: list) -> int:
    """Sum [rx, tx] to total bytes, returning -1 if unavailable."""
    if not rxtx or (rxtx[0] < 0 and rxtx[1] < 0):
        return -1
    return (rxtx[0] if rxtx[0] >= 0 else 0) + (rxtx[1] if rxtx[1] >= 0 else 0)


def clear_client_desync_cache():
    """Remove all files in the client-side desync cache.

    Called between push and clone for S5/S6/S12 so the clone measures a
    cold cache (dedup within fetch-all, but no warm-cache advantage from
    the push). This gives a more realistic "fresh clone" measurement.
    """
    run(["docker", "exec", CLIENT_CONTAINER, "sh", "-c",
         f"rm -rf {CLIENT_DESYNC_CACHE} && mkdir -p {CLIENT_DESYNC_CACHE}"],
        hide_output=True, check=False, timeout=30)


def check_desync_cache(container: str, cache_path: str, expect_empty: bool, label: str):
    """Verify that a desync chunk cache is empty or non-empty.

    expect_empty=True: warn if cache is non-empty (e.g. after push).
    expect_empty=False: warn if cache is empty (e.g. after clone — means
    the cache config is wrong and chunks were fetched but not cached).
    """
    r = run(["docker", "exec", container, "sh", "-c",
             f"find {cache_path} -type f 2>/dev/null | head -1"],
            capture=True, check=False, timeout=10)
    has_files = bool(r.stdout.strip())
    if expect_empty and has_files:
        log_warn(f"desync cache not empty {label} ({cache_path} in {container})")
    elif not expect_empty and not has_files:
        log_warn(f"desync cache empty {label} — config may be wrong ({cache_path} in {container})")


def get_container_stats_raw(container_name: str) -> Optional[dict]:
    """Return the full Docker stats JSON for a container, or None on failure.

    Uses the same Unix socket approach as get_net_io().  The response includes
    cumulative CPU counters and instantaneous memory usage.
    """
    class _UnixConn(http.client.HTTPConnection):
        def connect(self):
            self.sock = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
            self.sock.settimeout(30)
            self.sock.connect("/var/run/docker.sock")

    try:
        conn = _UnixConn("localhost")
        conn.request("GET", f"/containers/{container_name}/stats?stream=false")
        resp = conn.getresponse()
        if resp.status != 200:
            return None
        return json.loads(resp.read())
    except Exception:
        return None


class ResourceMonitor:
    """Background thread that periodically samples CPU and memory for containers.

    CPU usage is computed from the delta of consecutive cumulative counters
    (container cpu_usage.total_usage and system_cpu_usage).  The first sample
    per container establishes the baseline and produces no snapshot.
    """

    def __init__(self, containers: list, interval: float = 1.0):
        self.containers = containers
        self.interval = interval
        self.samples: dict[str, list] = {c: [] for c in containers}
        self._stop = threading.Event()
        self._thread: Optional[threading.Thread] = None
        self._prev_cpu: dict[str, tuple[int, int]] = {}

    def start(self):
        self._stop.clear()
        self._prev_cpu.clear()
        for c in self.containers:
            self.samples[c] = []
        self._thread = threading.Thread(target=self._run, daemon=True)
        self._thread.start()

    def stop(self, settle_s: float = 5.0) -> dict:
        """Wait for data to settle, then stop polling and return summaries."""
        time.sleep(settle_s)
        self._stop.set()
        if self._thread:
            self._thread.join(timeout=10)
        return self._summarize()

    def _run(self):
        while not self._stop.wait(self.interval):
            for c in self.containers:
                raw = get_container_stats_raw(c)
                if not raw:
                    continue
                mem_stats = raw.get("memory_stats", {})
                mem_usage = mem_stats.get("usage", 0)
                mem_limit = mem_stats.get("limit", 0)
                mem_detail = mem_stats.get("stats", {})
                mem_anon = mem_detail.get("anon", 0)
                mem_cache = mem_detail.get("file", 0)
                cpu_stats = raw.get("cpu_stats", {})
                total_usage = cpu_stats.get("cpu_usage", {}).get("total_usage", 0)
                system_usage = cpu_stats.get("system_cpu_usage", 0)
                num_cpus = cpu_stats.get("online_cpus") or 1

                cpu_percent = 0.0
                if c in self._prev_cpu:
                    prev_total, prev_system = self._prev_cpu[c]
                    cpu_delta = total_usage - prev_total
                    sys_delta = system_usage - prev_system
                    if sys_delta > 0:
                        cpu_percent = (cpu_delta / sys_delta) * num_cpus * 100.0
                    self.samples[c].append(ResourceSnapshot(
                        cpu_percent=cpu_percent,
                        mem_usage_bytes=mem_usage,
                        mem_limit_bytes=mem_limit,
                        mem_anon_bytes=mem_anon,
                        mem_cache_bytes=mem_cache,
                    ))

                self._prev_cpu[c] = (total_usage, system_usage)

    def _summarize(self) -> dict:
        result = {}
        for c in self.containers:
            snaps = self.samples[c]
            if not snaps:
                result[c] = ResourceSummary()
                continue
            cpus = [s.cpu_percent for s in snaps]
            mems = [s.mem_usage_bytes for s in snaps]
            anons = [s.mem_anon_bytes for s in snaps]
            caches = [s.mem_cache_bytes for s in snaps]
            result[c] = ResourceSummary(
                avg_cpu_percent=round(sum(cpus) / len(cpus), 1),
                max_cpu_percent=round(max(cpus), 1),
                avg_mem_bytes=int(sum(mems) / len(mems)),
                max_mem_bytes=max(mems),
                max_anon_bytes=max(anons),
                max_cache_bytes=max(caches),
                mem_limit_bytes=snaps[-1].mem_limit_bytes,
                sample_count=len(snaps),
            )
        return result


def measure_local_storage_ssh(path):
    """Measure bytes in a directory on the git-ssh container."""
    result = run_ssh(f"du -sk {path} 2>/dev/null | awk '{{print $1}}'")
    try:
        kb = int(result.stdout.strip())
        return kb * 1024
    except ValueError:
        return -1


# ---------------------------------------------------------------------------
# Test dataset creation
# ---------------------------------------------------------------------------

# ~200 common English words for generating random text content.
_WORDS = [
    "the", "be", "to", "of", "and", "a", "in", "that", "have", "it",
    "for", "not", "on", "with", "he", "as", "you", "do", "at", "this",
    "but", "his", "by", "from", "they", "we", "say", "her", "she", "or",
    "an", "will", "my", "one", "all", "would", "there", "their", "what",
    "so", "up", "out", "if", "about", "who", "get", "which", "go", "me",
    "when", "make", "can", "like", "time", "no", "just", "him", "know",
    "take", "people", "into", "year", "your", "good", "some", "could",
    "them", "see", "other", "than", "then", "now", "look", "only", "come",
    "its", "over", "think", "also", "back", "after", "use", "two", "how",
    "our", "work", "first", "well", "way", "even", "new", "want", "because",
    "any", "these", "give", "day", "most", "us", "great", "between", "need",
    "large", "often", "hand", "high", "place", "hold", "each", "under",
    "begin", "country", "keep", "point", "never", "next", "old", "while",
    "help", "talk", "turn", "start", "show", "part", "every", "move",
    "live", "found", "city", "world", "long", "still", "learn", "should",
    "house", "change", "close", "night", "real", "life", "few", "north",
    "open", "seem", "together", "state", "run", "school", "small", "end",
    "read", "port", "spell", "add", "land", "here", "must", "big", "draw",
    "left", "late", "such", "right", "mean", "might", "letter", "mother",
    "answer", "follow", "study", "book", "hear", "plant", "food", "cover",
    "number", "water", "call", "earth", "eye", "group", "always", "carry",
    "second", "tree", "cross", "river", "mark", "rock", "table", "build",
    "story", "record", "paper", "music", "field", "face", "door", "light",
    "voice", "power", "visit", "class", "clear", "drive", "stand", "fall",
]


def _generate_text_content(rng: random.Random, size: int) -> bytes:
    """Generate random text of approximately *size* bytes.

    Builds lines of 50-100 characters by picking random words from _WORDS.
    """
    parts: list[str] = []
    total = 0
    while total < size:
        line_len = rng.randint(50, 100)
        words = []
        cur = 0
        while cur < line_len:
            w = rng.choice(_WORDS)
            words.append(w)
            cur += len(w) + 1  # +1 for space
        line = " ".join(words) + "\n"
        parts.append(line)
        total += len(line)
    return "".join(parts).encode("utf-8")[:size]


def _edit_text_content(rng: random.Random, content: bytes, edit_fraction: float = 0.10) -> bytes:
    """Replace ~edit_fraction of the lines with new random text."""
    lines = content.decode("utf-8").splitlines(keepends=True)
    if not lines:
        return content
    n_edit = max(1, int(len(lines) * edit_fraction))
    indices = rng.sample(range(len(lines)), min(n_edit, len(lines)))
    for i in indices:
        line_len = rng.randint(50, 100)
        words = []
        cur = 0
        while cur < line_len:
            w = rng.choice(_WORDS)
            words.append(w)
            cur += len(w) + 1
        lines[i] = " ".join(words) + "\n"
    return "".join(lines).encode("utf-8")


def _append_text_content(rng: random.Random, content: bytes, append_fraction: float = 0.10) -> bytes:
    """Append ~append_fraction more lines to existing content (no existing lines modified)."""
    lines = content.decode("utf-8").splitlines(keepends=True)
    n_append = max(1, int(len(lines) * append_fraction))
    for _ in range(n_append):
        line_len = rng.randint(50, 100)
        words = []
        cur = 0
        while cur < line_len:
            w = rng.choice(_WORDS)
            words.append(w)
            cur += len(w) + 1
        lines.append(" ".join(words) + "\n")
    return "".join(lines).encode("utf-8")


def create_synthetic_repo(repo_dir: Path, num_commits: int = 50, file_size_mb: int = 2) -> tuple:
    """Create a git+LFS repo with a realistic multi-commit history.

    Each commit adds 2 binary files (~file_size_mb MB) and removes 1 existing binary
    (skipped on the first commit).  The first commit also adds 50 small text files
    (10-100 KB).  On each subsequent commit 2 new text files are added, 2 are removed,
    and 8 are edited (6 with ~10% lines replaced, 2 with ~10% lines appended),
    producing 12 new LFS objects per commit (2 binary + 2 new text + 8 edited text).
    The first commit produces 52 LFS objects (2 binary + 50 text).

    File content: binary files have a unique-random first half and a shared repeating
    phrase second half (for desync dedup/compression testing).  Text files contain
    random English words.

    Returns (total_lfs_bytes, lfs_object_count).
    """
    log_step(f"Creating synthetic repo: {num_commits} commits × 2 bin × {file_size_mb} MB + text files")
    if repo_dir.exists():
        shutil.rmtree(repo_dir)
    repo_dir.mkdir(parents=True)

    run(["git", "init", "-q", "-b", "main"], cwd=repo_dir)
    run(["git", "lfs", "install", "--local"], cwd=repo_dir, hide_output=True)
    run(["git", "lfs", "track", "*.bin", "*.txt"], cwd=repo_dir, hide_output=True)
    run(["git", "add", ".gitattributes"], cwd=repo_dir)

    # Use fixed timestamps so every run produces identical commit hashes.
    # Increment by 1 second per commit to guarantee unique ordering.
    epoch = 1700000000  # 2023-11-14T22:13:20Z
    date_env = {"GIT_AUTHOR_DATE": f"{epoch} +0000", "GIT_COMMITTER_DATE": f"{epoch} +0000"}
    run(["git", "commit", "-q", "-m", "init", "--author", "Test <test@example.com>"],
        cwd=repo_dir, env=date_env)

    file_counter = 0
    current_files: list = []
    rng = random.Random(0)
    text_counter = 0
    current_text_files: list = []

    for n in range(1, num_commits + 1):
        # --- Binary files ---
        existing_files = list(current_files)
        for _ in range(2):
            file_counter += 1
            fname = f"file_{file_counter:04d}.bin"
            half = file_size_mb * 1024 * 1024 // 2
            unique_part = bytes(rng.getrandbits(8) for _ in range(half))
            phrase = b"The quick brown fox jumps over the lazy dog\n"
            shared_part = (phrase * (half // len(phrase) + 1))[:half]
            (repo_dir / fname).write_bytes(unique_part + shared_part)
            current_files.append(fname)
        if n > 1 and existing_files:
            victim = rng.choice(existing_files)
            current_files.remove(victim)
            run(["git", "rm", "-q", victim], cwd=repo_dir)

        # --- Text files ---
        if n == 1:
            # First commit: add 50 initial text files.
            for _ in range(50):
                text_counter += 1
                fname = f"text_{text_counter:04d}.txt"
                size = rng.randint(10 * 1024, 100 * 1024)
                (repo_dir / fname).write_bytes(_generate_text_content(rng, size))
                current_text_files.append(fname)
        else:
            # Subsequent commits: add 2, remove 2, edit 8.
            existing_text = list(current_text_files)
            for _ in range(2):
                text_counter += 1
                fname = f"text_{text_counter:04d}.txt"
                size = rng.randint(10 * 1024, 100 * 1024)
                (repo_dir / fname).write_bytes(_generate_text_content(rng, size))
                current_text_files.append(fname)
            if len(existing_text) >= 2:
                victims = rng.sample(existing_text, 2)
                for v in victims:
                    current_text_files.remove(v)
                    existing_text.remove(v)
                    run(["git", "rm", "-q", v], cwd=repo_dir)
            # Edit 8 text files: 6 with ~10% lines replaced, 2 with ~10% lines appended
            if len(existing_text) >= 8:
                to_edit = rng.sample(existing_text, 8)
                for fname in to_edit[:6]:
                    fpath = repo_dir / fname
                    fpath.write_bytes(_edit_text_content(rng, fpath.read_bytes()))
                for fname in to_edit[6:]:
                    fpath = repo_dir / fname
                    fpath.write_bytes(_append_text_content(rng, fpath.read_bytes()))

        run(["git", "add", "--all"], cwd=repo_dir)
        commit_epoch = epoch + n
        date_env = {"GIT_AUTHOR_DATE": f"{commit_epoch} +0000",
                     "GIT_COMMITTER_DATE": f"{commit_epoch} +0000"}
        run(["git", "commit", "-q", "-m", f"commit {n}",
             "--author", "Test <test@example.com>"], cwd=repo_dir, env=date_env)
        run(["git", "tag", f"v{n}"], cwd=repo_dir)

    checksums = {
        f.name: hashlib.sha256(f.read_bytes()).hexdigest()
        for f in sorted(repo_dir.glob("*.bin")) + sorted(repo_dir.glob("*.txt"))
    }
    (repo_dir / "checksums.json").write_text(json.dumps(checksums, indent=2))

    lfs_obj_count = count_lfs_objects(repo_dir)
    r = run(["du", "-sk", str(repo_dir / ".git" / "lfs")], capture=True)
    try:
        total_bytes = int(r.stdout.strip().split()[0]) * 1024
    except (ValueError, IndexError):
        total_bytes = lfs_obj_count * file_size_mb * 1024 * 1024

    log_ok(f"Synthetic repo ready: {num_commits} commits, {lfs_obj_count} LFS objects, "
           f"{total_bytes // (1024*1024)} MB, {len(current_files)} bin + {len(current_text_files)} txt HEAD files")
    return total_bytes, lfs_obj_count


def setup_r0_direct(r0_git_path: Path, repo_dir: Path) -> tuple:
    """Set up a working repo from R0.git using clone + LFS symlink (fast).

    Instead of cloning + ``git lfs fetch --all`` (which copies ~23 GB), this:
    1. ``git clone`` from the local bare repo (uses hardlinks — fast)
    2. Populates local refs/heads/*
    3. Inside the client container, symlinks ``.git/lfs`` → ``/lfs-cache``
       (the bare repo's ``lfs/`` dir mounted read-only via docker-compose).
    4. Checks out the working tree inside the container (resolves LFS pointers
       via the symlink, so git-lfs finds objects at /lfs-cache/objects/).
    5. Computes checksums on the host for clone verification.

    Requires: start_services() was called with lfs_dir=r0_git_path/"lfs" so
    that /lfs-cache is mounted in the client container.

    Returns (total_lfs_bytes, lfs_object_count).
    """
    log_step(f"Setting up R0 working repo (fast) from {r0_git_path}")
    if repo_dir.exists():
        shutil.rmtree(repo_dir)

    # 1. Clone from bare repo on host — git uses hardlinks for local clones (fast)
    run(["git", "clone", "-q", str(r0_git_path), str(repo_dir)])
    # Populate local refs/heads/* for all remote branches
    run(["git", "fetch", "-q", "origin", "+refs/heads/*:refs/heads/*", "--update-head-ok"],
        cwd=repo_dir)

    # 2. Inside the container, symlink .git/lfs → /lfs-cache (the mounted bare lfs/ dir)
    repo_in_client = host_path_to_client(repo_dir)
    client_exec(["rm", "-rf", f"{repo_in_client}/.git/lfs"])
    client_exec(["ln", "-s", "/lfs-cache", f"{repo_in_client}/.git/lfs"])
    log_ok("Symlinked .git/lfs → /lfs-cache inside client container")

    # 3. Checkout working tree inside the container (LFS objects resolved via symlink)
    client_exec(["git", "lfs", "checkout"], cwd=repo_in_client, timeout=1200)

    # 4. Measure LFS stats from the bare repo's lfs/ directory.
    # We count from the bare repo directly because .git/lfs in repo_dir is a
    # symlink to /lfs-cache which only resolves inside the container.
    bare_lfs_dir = r0_git_path / "lfs"
    if bare_lfs_dir.exists():
        result = run(["du", "-sk", str(bare_lfs_dir)], capture=True)
        try:
            total_bytes = int(result.stdout.strip().split()[0]) * 1024
        except (ValueError, IndexError):
            total_bytes = 0
    else:
        total_bytes = 0

    lfs_obj_count = count_lfs_objects(r0_git_path)

    # 5. Compute checksums of LFS-tracked files in HEAD (now checked out on host)
    lfs_extensions = ("*.obj", "*.bin", "*.dll")
    lfs_files = []
    for pattern in lfs_extensions:
        lfs_files.extend(repo_dir.rglob(pattern))
    checksums = {}
    for f in sorted(f for f in lfs_files if f.is_file()):
        try:
            checksums[str(f.relative_to(repo_dir))] = hashlib.sha256(f.read_bytes()).hexdigest()
        except Exception:
            pass
    (repo_dir / "checksums.json").write_text(json.dumps(checksums, indent=2))
    log_ok(
        f"R0 repo ready (fast): {total_bytes // (1024*1024)} MB, "
        f"{lfs_obj_count} LFS objects, {len(checksums)} HEAD files checksummed"
    )
    return total_bytes, lfs_obj_count


def create_nolfs_repo(lfs_repo: Path, nolfs_repo: Path):
    """Clone an LFS repo and convert all LFS pointers to regular git blobs.

    Used for the small dataset to create a non-LFS baseline (scenario 0).
    """
    log_step(f"Creating non-LFS repo from {lfs_repo}")
    if nolfs_repo.exists():
        shutil.rmtree(nolfs_repo)
    run(["git", "clone", "-q", str(lfs_repo), str(nolfs_repo)])
    run(["git", "lfs", "migrate", "export", "--everything", "--include=*"],
        cwd=nolfs_repo, timeout=1200)
    run(["git", "gc", "-q", "--prune"], cwd=nolfs_repo, timeout=120)
    # Compute checksums for verification (same as LFS repo but files are now regular blobs)
    lfs_extensions = ("*.bin", "*.txt")
    checksums = {}
    for pattern in lfs_extensions:
        for f in sorted(nolfs_repo.rglob(pattern)):
            if f.is_file():
                checksums[str(f.relative_to(nolfs_repo))] = hashlib.sha256(f.read_bytes()).hexdigest()
    (nolfs_repo / "checksums.json").write_text(json.dumps(checksums, indent=2))
    log_ok(f"Non-LFS repo ready: {len(checksums)} files checksummed")


def setup_r0_nolfs(r0_nolfs_path: Path, repo_dir: Path):
    """Set up a working repo from a non-LFS R0 bare repo for scenario 0.

    Fast clone (hardlinks) + populate local branches + compute checksums.
    """
    log_step(f"Setting up non-LFS R0 repo from {r0_nolfs_path}")
    if repo_dir.exists():
        shutil.rmtree(repo_dir)
    run(["git", "clone", "-q", str(r0_nolfs_path), str(repo_dir)])
    run(["git", "fetch", "-q", "origin", "+refs/heads/*:refs/heads/*", "--update-head-ok"],
        cwd=repo_dir)
    # Compute checksums of binary files in HEAD
    lfs_extensions = ("*.obj", "*.bin", "*.dll")
    checksums = {}
    for pattern in lfs_extensions:
        for f in sorted(repo_dir.rglob(pattern)):
            if f.is_file():
                try:
                    checksums[str(f.relative_to(repo_dir))] = hashlib.sha256(f.read_bytes()).hexdigest()
                except Exception:
                    pass
    (repo_dir / "checksums.json").write_text(json.dumps(checksums, indent=2))
    log_ok(f"Non-LFS R0 repo ready: {len(checksums)} files checksummed")


# ---------------------------------------------------------------------------
# Per-scenario server repo setup
# ---------------------------------------------------------------------------
def create_bare_repo(name: str) -> None:
    """Create a bare git repo on the server with main as the default branch.

    Removes any existing repo with the same name first so stale refs from
    a previous interrupted run don't cause 'cannot lock ref' errors.
    """
    path = f"/home/git/{name}.git"
    run_ssh(f"rm -rf {path}")
    run_ssh(f"git init --bare {path}")
    run_ssh(f"git -C {path} symbolic-ref HEAD refs/heads/main")
    run_ssh(f"git -C {path} config gc.auto 0")


def remove_bare_repo(name: str):
    run_ssh(f"rm -rf /home/git/{name}.git", timeout=30)


def write_server_config(repo_name: str, config: dict):
    """Write a desync-lfs.json into a bare repo on the server (via stdin)."""
    config_json = json.dumps(config, indent=2)
    path = f"/home/git/{repo_name}.git/desync-lfs.json"
    result = subprocess.run(
        ["ssh", "-i", str(SSH_KEY),
         "-o", "StrictHostKeyChecking=no", "-o", "UserKnownHostsFile=/dev/null",
         "-o", "BatchMode=yes", "-p", str(SSH_PORT),
         "git@localhost",
         f"python3 -c \"import sys; open('{path}', 'w').write(sys.stdin.read())\""],
        input=config_json, text=True, capture_output=True, timeout=10,
    )
    if result.returncode != 0:
        raise RuntimeError(f"Failed to write server config: {result.stderr}")


def configure_s1_repo(repo_name: str):
    """Configure a repo to delegate to the baseline (S1) git-lfs-transfer.

    `git config desync-lfs false` is not valid (no section), so we write
    the disable flag via desync-lfs.json and set the delegate exec path in
    git config (desync-lfs.transfer.exec is a valid section.subsection.key).
    """
    path = f"/home/git/{repo_name}.git"
    write_server_config(repo_name, {"desync-lfs": False})
    run_ssh(f"git -C {path} config desync-lfs.transfer.exec /usr/local/bin/s1-git-lfs-transfer")


def configure_desync_local_repo(repo_name: str):
    """Write desync-lfs.json with chunk-size and explicit local store/index paths.

    When a config file exists, git-lfs-transfer skips the convention-based
    fallback (desync-lfs/{chunks,index}), so we must set stores explicitly.
    """
    repo_path = f"/home/git/{repo_name}.git"
    defaults = {
        "stores": [f"{repo_path}/desync-lfs/chunks"],
        "index-store": f"{repo_path}/desync-lfs/index",
        "chunk-size": CHUNK_SIZE,
    }
    if MAX_IN_FLIGHT > 0:
        defaults["max-in-flight"] = MAX_IN_FLIGHT
    if MAX_STORAGE_OPS > 0:
        defaults["max-storage-ops"] = MAX_STORAGE_OPS
    config = {"defaults": defaults}
    write_server_config(repo_name, config)


def configure_desync_s3_repo(repo_name: str, store_url: str, index_url: str,
                               s3_endpoint: str, s3_access: str, s3_secret: str,
                               s3_region: str = "us-east-1",
                               cache_path: str = "",
                               advertise: str = ""):
    """Write server-side desync-lfs.json for S3 backend.

    Contains server-side settings: store/index URLs, S3 credentials,
    chunk-size, cache path, concurrency, and resource limits for the
    server's git-lfs-transfer process.

    Client-local settings (cache, conn-pool-size) are in the client's
    own config file written by write_client_desync_config().

    When advertise is set, the server sends a filtered subset of this
    config to clients via the SSH transfer negotiation protocol.
    FilterServerConfig() keeps only URLs, credentials, chunk-size, and
    digest; local-only fields are stripped automatically.
    """
    endpoint_key = s3_endpoint.rstrip("/")
    defaults = {
        "stores": [store_url],
        "index-store": index_url,
        "chunk-size": CHUNK_SIZE,
        # Server-side concurrency and resource limits (for git-lfs-transfer).
        "concurrency": DESYNC_CONCURRENCY,
    }
    if cache_path:
        defaults["cache"] = cache_path
    if MAX_IN_FLIGHT > 0:
        defaults["max-in-flight"] = MAX_IN_FLIGHT
    if MAX_STORAGE_OPS > 0:
        defaults["max-storage-ops"] = MAX_STORAGE_OPS
    config = {
        "defaults": defaults,
        "s3-credentials": {
            endpoint_key: {
                "access-key": s3_access,
                "secret-key": s3_secret,
                "aws-region": s3_region,
            }
        },
    }
    config_json = json.dumps(config, indent=2)
    path = f"/home/git/{repo_name}.git/desync-lfs.json"
    # Write via python3 on the server to avoid shell quoting issues with JSON
    py_cmd = f"python3 -c \"import sys; open('{path}', 'w').write(sys.stdin.read())\""
    result = subprocess.run(
        ["ssh", "-i", str(SSH_KEY),
         "-o", "StrictHostKeyChecking=no",
         "-o", "UserKnownHostsFile=/dev/null",
         "-o", "BatchMode=yes",
         "-p", str(SSH_PORT),
         "git@localhost", py_cmd],
        input=config_json,
        text=True,
        capture_output=True,
        timeout=10,
    )
    if result.returncode != 0:
        raise RuntimeError(f"Failed to write server config: {result.stderr}")

    if advertise:
        run([*SSH_CMD.split(), "-p", str(SSH_PORT), "git@localhost",
             f"git -C /home/git/{repo_name}.git config desync-lfs.advertise {advertise}"],
            hide_output=True, timeout=10)


def count_lfs_objects(repo_dir: Path) -> int:
    """Count unique LFS objects stored in lfs/objects/ (SHA256-addressed blobs).

    Works for both bare repos (lfs/objects/) and normal repos (.git/lfs/objects/).
    """
    # Bare repo: lfs/objects/; normal repo: .git/lfs/objects/
    lfs_dir = repo_dir / "lfs" / "objects"
    if not lfs_dir.exists():
        lfs_dir = repo_dir / ".git" / "lfs" / "objects"
    if not lfs_dir.exists():
        return 0
    return sum(1 for f in lfs_dir.rglob("*") if f.is_file())


def verify_clone(source_checksums: dict, clone_dir: Path,
                 source_lfs_count: int = -1) -> bool:
    """Verify cloned files match original checksums and LFS object count matches.

    source_lfs_count: expected number of LFS objects in .git/lfs/objects/.
    Pass -1 to skip the count check (e.g. synthetic dataset).
    """
    for filename, expected_sha in source_checksums.items():
        clone_file = clone_dir / filename
        if not clone_file.exists():
            log_warn(f"Missing file in clone: {filename}")
            return False
        actual = hashlib.sha256(clone_file.read_bytes()).hexdigest()
        if actual != expected_sha:
            log_warn(f"Checksum mismatch for {filename}")
            return False
    if source_lfs_count >= 0:
        clone_count = count_lfs_objects(clone_dir)
        if clone_count != source_lfs_count:
            log_warn(f"LFS object count: expected {source_lfs_count}, got {clone_count}")
            return False
    return True


def load_checksums(data_dir: Path) -> dict:
    csum_file = data_dir / "checksums.json"
    if csum_file.exists():
        return json.loads(csum_file.read_text())
    # Compute on the fly
    files = sorted(data_dir.glob("*.bin")) + sorted(data_dir.glob("*.txt"))
    return {f.name: hashlib.sha256(f.read_bytes()).hexdigest() for f in files}


# ---------------------------------------------------------------------------
# Individual scenario runners
# ---------------------------------------------------------------------------
def run_scenario_0(source_repo: Path, checksums: dict, raw_lfs_bytes: int,
                   tmpdir: Path, push_refspecs=None,
                   clone_extra_refspecs=None, source_lfs_count=-1,
                   container_cwd: str = None,
                   stop_after: int = None) -> ScenarioResult:
    """Plain git push/clone (no LFS) — baseline for raw git storage cost."""
    result = ScenarioResult(
        id=0,
        name="plain git (no LFS)",
        raw_lfs_bytes=raw_lfs_bytes,
        source_lfs_count=-1,  # no LFS objects
    )
    repo_name = "repo0"
    clone_dir = tmpdir / "clone0"

    monitor = ResourceMonitor([CLIENT_CONTAINER, GIT_SSH_CONTAINER])
    try:
        create_bare_repo(repo_name)
        remote = client_remote_url(repo_name)

        monitor.start()

        log(f"  → S{result.id} Push")
        _c0 = get_net_io(CLIENT_CONTAINER); _s0 = get_net_io(GIT_SSH_CONTAINER)
        result.push_time_s = git_push_client(remote, source_repo,
                                              push_refspecs=push_refspecs,
                                              container_cwd=container_cwd,
                                              stop_after=stop_after)
        _c1 = get_net_io(CLIENT_CONTAINER); _s1 = get_net_io(GIT_SSH_CONTAINER)
        result.push_net_bytes = _net_rxtx(_c0, _c1)
        result.push_ssh_net_bytes = _net_rxtx(_s0, _s1)
        log_ok(f"Push: {result.push_time_s:.1f}s  wire: {fmt_bytes(result.push_net_bytes[1])}")

        result.storage_bytes = measure_local_storage_ssh(f"/home/git/{repo_name}.git")
        log_ok(f"Storage: {result.storage_bytes // (1024*1024)} MB")

        reclaim_all_page_cache()
        log(f"  → S{result.id} Clone")
        _c0 = get_net_io(CLIENT_CONTAINER); _s0 = get_net_io(GIT_SSH_CONTAINER)
        clone_time = git_clone_client(remote, clone_dir)
        # Fetch all branches (plain git, no LFS pull needed)
        cwd = host_path_to_client(clone_dir)
        client_exec(
            ["git", "fetch", "-q", "origin", "+refs/heads/*:refs/heads/*", "--update-head-ok"],
            cwd=cwd, timeout=3600,
        )
        _c1 = get_net_io(CLIENT_CONTAINER); _s1 = get_net_io(GIT_SSH_CONTAINER)
        result.clone_net_bytes = _net_rxtx(_c0, _c1)
        result.clone_ssh_net_bytes = _net_rxtx(_s0, _s1)
        result.clone_time_s = clone_time
        log_ok(f"Clone: {result.clone_time_s:.1f}s  wire: {fmt_bytes(result.clone_net_bytes[0])}")

        result.clone_lfs_count = 0
        result.verified = verify_clone(checksums, clone_dir, -1)
        log_ok(f"Verified: {result.verified}")

    except Exception as e:
        result.error = str(e)
        log_err(f"Scenario 0 error: {e}")
    finally:
        result.resources = monitor.stop()

    return result


#
# ---------------------------------------------------------------------------
def run_scenario_1(source_repo: Path, checksums: dict, raw_lfs_bytes: int,
                   tmpdir: Path, push_refspecs=None,
                   clone_extra_refspecs=None, source_lfs_count=-1,
                   container_cwd: str = None,
                   stop_after: int = None) -> ScenarioResult:
    """Baseline git-lfs-transfer (scutiger or charmbracelet), server local storage."""
    result = ScenarioResult(
        id=1,
        name=f"{S1_LFS_TRANSFER} → files",
        raw_lfs_bytes=raw_lfs_bytes,
        source_lfs_count=source_lfs_count,
    )
    repo_name = "repo1"
    clone_dir = tmpdir / "clone1"

    monitor = ResourceMonitor([CLIENT_CONTAINER, GIT_SSH_CONTAINER])
    try:
        create_bare_repo(repo_name)
        configure_s1_repo(repo_name)
        remote = client_remote_url(repo_name)

        monitor.start()

        log(f"  → S{result.id} Push")
        _c0 = get_net_io(CLIENT_CONTAINER); _s0 = get_net_io(GIT_SSH_CONTAINER)
        result.push_time_s = git_push_client(remote, source_repo, push_refspecs=push_refspecs,
                                              container_cwd=container_cwd,
                                              stop_after=stop_after)
        _c1 = get_net_io(CLIENT_CONTAINER); _s1 = get_net_io(GIT_SSH_CONTAINER)
        result.push_net_bytes = _net_rxtx(_c0, _c1)
        result.push_ssh_net_bytes = _net_rxtx(_s0, _s1)
        log_ok(f"Push: {result.push_time_s:.1f}s  wire: {fmt_bytes(result.push_net_bytes[1])}")

        # Storage: scutiger stores LFS objects in lfs/objects/ inside the repo
        result.storage_bytes = measure_local_storage_ssh(f"/home/git/{repo_name}.git")
        log_ok(f"Storage: {result.storage_bytes // (1024*1024)} MB")

        reclaim_all_page_cache()
        log(f"  → S{result.id} Clone")
        _c0 = get_net_io(CLIENT_CONTAINER); _s0 = get_net_io(GIT_SSH_CONTAINER)
        clone_time = git_clone_client(remote, clone_dir)
        pull_time = git_lfs_pull_client(clone_dir, extra_fetch_refspecs=clone_extra_refspecs)
        _c1 = get_net_io(CLIENT_CONTAINER); _s1 = get_net_io(GIT_SSH_CONTAINER)
        result.clone_net_bytes = _net_rxtx(_c0, _c1)
        result.clone_ssh_net_bytes = _net_rxtx(_s0, _s1)
        result.clone_time_s = clone_time + pull_time
        log_ok(f"Clone: {result.clone_time_s:.1f}s  wire: {fmt_bytes(result.clone_net_bytes[0])}")

        result.clone_lfs_count = count_lfs_objects(clone_dir)
        result.verified = verify_clone(checksums, clone_dir, source_lfs_count)
        log_ok(f"Verified: {result.verified}  LFS objects: {result.clone_lfs_count}/{source_lfs_count if source_lfs_count >= 0 else '?'}")

    except Exception as e:
        result.error = str(e)
        log_err(f"Scenario 1 error: {e}")
    finally:
        result.resources = monitor.stop()

    return result


def run_scenario_2(source_repo: Path, checksums: dict, raw_lfs_bytes: int,
                   tmpdir: Path, push_refspecs=None,
                   clone_extra_refspecs=None, source_lfs_count=-1,
                   container_cwd: str = None,
                   stop_after: int = None) -> ScenarioResult:
    """desync git-lfs-transfer, server local chunks (convention-based)."""
    result = ScenarioResult(
        id=2,
        name="desync on server → files",
        raw_lfs_bytes=raw_lfs_bytes,
        source_lfs_count=source_lfs_count,
    )
    repo_name = "repo2"
    clone_dir = tmpdir / "clone2"

    monitor = ResourceMonitor([CLIENT_CONTAINER, GIT_SSH_CONTAINER])
    try:
        create_bare_repo(repo_name)
        configure_desync_local_repo(repo_name)
        remote = client_remote_url(repo_name)

        monitor.start()

        log(f"  → S{result.id} Push")
        _c0 = get_net_io(CLIENT_CONTAINER); _s0 = get_net_io(GIT_SSH_CONTAINER)
        result.push_time_s = git_push_client(remote, source_repo, push_refspecs=push_refspecs,
                                              container_cwd=container_cwd,
                                              stop_after=stop_after)
        _c1 = get_net_io(CLIENT_CONTAINER); _s1 = get_net_io(GIT_SSH_CONTAINER)
        result.push_net_bytes = _net_rxtx(_c0, _c1)
        result.push_ssh_net_bytes = _net_rxtx(_s0, _s1)
        log_ok(f"Push: {result.push_time_s:.1f}s  wire: {fmt_bytes(result.push_net_bytes[1])}")

        # Storage: desync creates desync-lfs/{chunks,index} in repo dir
        result.storage_bytes = measure_local_storage_ssh(
            f"/home/git/{repo_name}.git/desync-lfs"
        )
        log_ok(f"Storage: {result.storage_bytes // (1024*1024)} MB")

        reclaim_all_page_cache()
        log(f"  → S{result.id} Clone")
        _c0 = get_net_io(CLIENT_CONTAINER); _s0 = get_net_io(GIT_SSH_CONTAINER)
        clone_time = git_clone_client(remote, clone_dir)
        pull_time = git_lfs_pull_client(clone_dir, extra_fetch_refspecs=clone_extra_refspecs)
        _c1 = get_net_io(CLIENT_CONTAINER); _s1 = get_net_io(GIT_SSH_CONTAINER)
        result.clone_net_bytes = _net_rxtx(_c0, _c1)
        result.clone_ssh_net_bytes = _net_rxtx(_s0, _s1)
        result.clone_time_s = clone_time + pull_time
        log_ok(f"Clone: {result.clone_time_s:.1f}s  wire: {fmt_bytes(result.clone_net_bytes[0])}")

        result.clone_lfs_count = count_lfs_objects(clone_dir)
        result.verified = verify_clone(checksums, clone_dir, source_lfs_count)
        log_ok(f"Verified: {result.verified}  LFS objects: {result.clone_lfs_count}/{source_lfs_count if source_lfs_count >= 0 else '?'}")

    except Exception as e:
        result.error = str(e)
        log_err(f"Scenario 2 error: {e}")
    finally:
        result.resources = monitor.stop()

    return result


def run_scenario_3(source_repo: Path, checksums: dict, raw_lfs_bytes: int,
                   tmpdir: Path, push_refspecs=None,
                   clone_extra_refspecs=None, source_lfs_count=-1,
                   container_cwd: str = None,
                   stop_after: int = None) -> ScenarioResult:
    """desync git-lfs-transfer, server stores chunks/index in MinIO."""
    result = ScenarioResult(
        id=3,
        name="desync on server → MinIO",
        raw_lfs_bytes=raw_lfs_bytes,
        source_lfs_count=source_lfs_count,
    )
    repo_name = "repo3"
    clone_dir = tmpdir / "clone3"
    bucket = "lfs"
    store_url = f"s3+{SHARED_MINIO_INTERNAL}/{bucket}/chunks/"
    index_url = f"s3+{SHARED_MINIO_INTERNAL}/{bucket}/index/"

    start_s3_service("shared-minio", setup_fn=setup_shared_minio)
    monitor = ResourceMonitor([CLIENT_CONTAINER, GIT_SSH_CONTAINER, SHARED_MINIO_CONTAINER])
    try:
        create_bare_repo(repo_name)
        server_cache = f"/home/git/{repo_name}.git/desync-lfs/cache"
        configure_desync_s3_repo(
            repo_name, store_url, index_url,
            SHARED_MINIO_INTERNAL, MINIO_USER, MINIO_PASS,
            cache_path=server_cache,
        )
        remote = client_remote_url(repo_name)

        result.s3_container = SHARED_MINIO_CONTAINER
        monitor.start()

        check_desync_cache(GIT_SSH_CONTAINER, server_cache, True, "S3 before push")
        log(f"  → S{result.id} Push")
        _c0 = get_net_io(CLIENT_CONTAINER); _s0 = get_net_io(GIT_SSH_CONTAINER); _m0 = get_net_io(SHARED_MINIO_CONTAINER)
        result.push_time_s = git_push_client(
            remote, source_repo,
            extra_env={"S3_ACCESS_KEY": MINIO_USER, "S3_SECRET_KEY": MINIO_PASS},
            push_refspecs=push_refspecs,
            container_cwd=container_cwd,
            stop_after=stop_after,
        )
        _c1 = get_net_io(CLIENT_CONTAINER); _s1 = get_net_io(GIT_SSH_CONTAINER); _m1 = get_net_io(SHARED_MINIO_CONTAINER)
        result.push_net_bytes = _net_rxtx(_c0, _c1)
        result.push_ssh_net_bytes = _net_rxtx(_s0, _s1)
        result.push_s3_net_bytes = _net_rxtx(_m0, _m1)
        log_ok(f"Push: {result.push_time_s:.1f}s  wire: {fmt_bytes(result.push_net_bytes[1])}")
        check_desync_cache(GIT_SSH_CONTAINER, server_cache, True, "S3 after push")

        result.storage_bytes = measure_volume_storage(f"{COMPOSE_PROJECT}_shared-minio-data")
        log_ok(f"Storage: {result.storage_bytes // (1024*1024)} MB")

        reclaim_all_page_cache(SHARED_MINIO_CONTAINER)
        run(["docker", "exec", GIT_SSH_CONTAINER,
             "sh", "-c", f"rm -rf /home/git/{repo_name}.git/desync-lfs/cache/*"],
            hide_output=True, check=False, timeout=30)
        log(f"  → S{result.id} Clone")
        _c0 = get_net_io(CLIENT_CONTAINER); _s0 = get_net_io(GIT_SSH_CONTAINER); _m0 = get_net_io(SHARED_MINIO_CONTAINER)
        clone_time = git_clone_client(remote, clone_dir)
        pull_time = git_lfs_pull_client(
            clone_dir,
            extra_fetch_refspecs=clone_extra_refspecs,
        )
        _c1 = get_net_io(CLIENT_CONTAINER); _s1 = get_net_io(GIT_SSH_CONTAINER); _m1 = get_net_io(SHARED_MINIO_CONTAINER)
        result.clone_net_bytes = _net_rxtx(_c0, _c1)
        result.clone_ssh_net_bytes = _net_rxtx(_s0, _s1)
        result.clone_s3_net_bytes = _net_rxtx(_m0, _m1)
        result.clone_time_s = clone_time + pull_time
        log_ok(f"Clone: {result.clone_time_s:.1f}s  wire: {fmt_bytes(result.clone_net_bytes[0])}")

        result.clone_lfs_count = count_lfs_objects(clone_dir)
        result.verified = verify_clone(checksums, clone_dir, source_lfs_count)
        log_ok(f"Verified: {result.verified}  LFS objects: {result.clone_lfs_count}/{source_lfs_count if source_lfs_count >= 0 else '?'}")

    except Exception as e:
        result.error = str(e)
        log_err(f"Scenario 3 error: {e}")
    finally:
        result.resources = monitor.stop()

    return result


def run_scenario_4(source_repo: Path, checksums: dict, raw_lfs_bytes: int,
                   tmpdir: Path, push_refspecs=None,
                   clone_extra_refspecs=None, source_lfs_count=-1,
                   container_cwd: str = None,
                   stop_after: int = None) -> ScenarioResult:
    """desync git-lfs-transfer, server stores chunks/index in Garage."""
    result = ScenarioResult(
        id=4,
        name="desync on server → Garage",
        raw_lfs_bytes=raw_lfs_bytes,
        source_lfs_count=source_lfs_count,
    )
    repo_name = "repo4"
    clone_dir = tmpdir / "clone4"
    bucket = "lfs"
    store_url = f"s3+{SHARED_GARAGE_INTERNAL}/{bucket}/chunks/"
    index_url = f"s3+{SHARED_GARAGE_INTERNAL}/{bucket}/index/"

    start_s3_service("shared-garage", setup_fn=setup_shared_garage)

    if not SHARED_GARAGE_KEY_ID or not SHARED_GARAGE_SECRET:
        result.error = "Garage credentials not available"
        return result

    monitor = ResourceMonitor([CLIENT_CONTAINER, GIT_SSH_CONTAINER, SHARED_GARAGE_CONTAINER])
    try:
        create_bare_repo(repo_name)
        server_cache = f"/home/git/{repo_name}.git/desync-lfs/cache"
        configure_desync_s3_repo(
            repo_name, store_url, index_url,
            SHARED_GARAGE_INTERNAL, SHARED_GARAGE_KEY_ID, SHARED_GARAGE_SECRET,
            s3_region="garage",
            cache_path=server_cache,
        )
        remote = client_remote_url(repo_name)

        result.s3_container = SHARED_GARAGE_CONTAINER
        monitor.start()

        check_desync_cache(GIT_SSH_CONTAINER, server_cache, True, "S4 before push")
        log(f"  → S{result.id} Push")
        _c0 = get_net_io(CLIENT_CONTAINER); _s0 = get_net_io(GIT_SSH_CONTAINER); _g0 = get_net_io(SHARED_GARAGE_CONTAINER)
        result.push_time_s = git_push_client(
            remote, source_repo,
            extra_env={"S3_ACCESS_KEY": SHARED_GARAGE_KEY_ID, "S3_SECRET_KEY": SHARED_GARAGE_SECRET},
            push_refspecs=push_refspecs,
            container_cwd=container_cwd,
            stop_after=stop_after,
        )
        _c1 = get_net_io(CLIENT_CONTAINER); _s1 = get_net_io(GIT_SSH_CONTAINER); _g1 = get_net_io(SHARED_GARAGE_CONTAINER)
        result.push_net_bytes = _net_rxtx(_c0, _c1)
        result.push_ssh_net_bytes = _net_rxtx(_s0, _s1)
        result.push_s3_net_bytes = _net_rxtx(_g0, _g1)
        log_ok(f"Push: {result.push_time_s:.1f}s  wire: {fmt_bytes(result.push_net_bytes[1])}")
        check_desync_cache(GIT_SSH_CONTAINER, server_cache, True, "S4 after push")

        _data = measure_volume_storage(f"{COMPOSE_PROJECT}_shared-garage-data")
        _meta = measure_volume_storage(f"{COMPOSE_PROJECT}_shared-garage-meta")
        result.storage_bytes = (_data if _data >= 0 else 0) + (_meta if _meta >= 0 else 0)
        if _data < 0 and _meta < 0:
            result.storage_bytes = -1
        log_ok(f"Storage: {result.storage_bytes // (1024*1024)} MB")

        reclaim_all_page_cache(SHARED_GARAGE_CONTAINER)
        run(["docker", "exec", GIT_SSH_CONTAINER,
             "sh", "-c", f"rm -rf /home/git/{repo_name}.git/desync-lfs/cache/*"],
            hide_output=True, check=False, timeout=30)
        log(f"  → S{result.id} Clone")
        _c0 = get_net_io(CLIENT_CONTAINER); _s0 = get_net_io(GIT_SSH_CONTAINER); _g0 = get_net_io(SHARED_GARAGE_CONTAINER)
        clone_time = git_clone_client(
            remote, clone_dir,
        )
        pull_time = git_lfs_pull_client(
            clone_dir,
            extra_fetch_refspecs=clone_extra_refspecs,
        )
        _c1 = get_net_io(CLIENT_CONTAINER); _s1 = get_net_io(GIT_SSH_CONTAINER); _g1 = get_net_io(SHARED_GARAGE_CONTAINER)
        result.clone_net_bytes = _net_rxtx(_c0, _c1)
        result.clone_ssh_net_bytes = _net_rxtx(_s0, _s1)
        result.clone_s3_net_bytes = _net_rxtx(_g0, _g1)
        result.clone_time_s = clone_time + pull_time
        log_ok(f"Clone: {result.clone_time_s:.1f}s  wire: {fmt_bytes(result.clone_net_bytes[0])}")
        check_desync_cache(GIT_SSH_CONTAINER, server_cache, False, "S4 after clone")

        result.clone_lfs_count = count_lfs_objects(clone_dir)
        result.verified = verify_clone(checksums, clone_dir, source_lfs_count)
        log_ok(f"Verified: {result.verified}  LFS objects: {result.clone_lfs_count}/{source_lfs_count if source_lfs_count >= 0 else '?'}")

    except Exception as e:
        result.error = str(e)
        log_err(f"Scenario 4 error: {e}")
    finally:
        result.resources = monitor.stop()

    return result


def run_scenario_5(source_repo: Path, checksums: dict, raw_lfs_bytes: int,
                   tmpdir: Path, push_refspecs=None,
                   clone_extra_refspecs=None, source_lfs_count=-1,
                   container_cwd: str = None,
                   stop_after: int = None) -> ScenarioResult:
    """git-lfs-desync client agent, non-pipelined (one process per concurrent transfer)."""
    result = ScenarioResult(
        id=5,
        name="desync on client (no pipeline)",
        raw_lfs_bytes=raw_lfs_bytes,
        source_lfs_count=source_lfs_count,
    )
    repo_name = "repo5"
    clone_dir = tmpdir / "clone5"
    bucket = "lfs"
    store_url = f"s3+{SHARED_GARAGE_INTERNAL}/{bucket}/chunks/"
    index_url = f"s3+{SHARED_GARAGE_INTERNAL}/{bucket}/index/"

    start_s3_service("shared-garage", setup_fn=setup_shared_garage)

    if not SHARED_GARAGE_KEY_ID or not SHARED_GARAGE_SECRET:
        result.error = "Garage credentials not available"
        return result

    monitor = ResourceMonitor([CLIENT_CONTAINER, GIT_SSH_CONTAINER, SHARED_GARAGE_CONTAINER])
    try:
        create_bare_repo(repo_name)
        server_cache = f"/home/git/{repo_name}.git/desync-lfs/cache"
        configure_desync_s3_repo(
            repo_name, store_url, index_url,
            SHARED_GARAGE_INTERNAL, SHARED_GARAGE_KEY_ID, SHARED_GARAGE_SECRET,
            s3_region="garage",
            cache_path=server_cache,
            advertise="with-credentials",
        )
        # Force non-pipelined mode via --no-pipelined on the agent.
        client_exec(["git", "config", "--global",
                     "lfs.customtransfer.desync.args", "--no-pipelined"])
        remote = client_remote_url(repo_name)

        # Clear chunk cache so each scenario starts fresh.
        clear_client_desync_cache()

        result.s3_container = SHARED_GARAGE_CONTAINER
        monitor.start()

        log(f"  → S{result.id} Push (client agent, non-pipelined)")
        _c0 = get_net_io(CLIENT_CONTAINER); _s0 = get_net_io(GIT_SSH_CONTAINER); _g0 = get_net_io(SHARED_GARAGE_CONTAINER)
        result.push_time_s = git_push_client(
            remote, source_repo,
            push_refspecs=push_refspecs,
            container_cwd=container_cwd,
            stop_after=stop_after,
        )
        _c1 = get_net_io(CLIENT_CONTAINER); _s1 = get_net_io(GIT_SSH_CONTAINER); _g1 = get_net_io(SHARED_GARAGE_CONTAINER)
        result.push_net_bytes = _net_rxtx(_c0, _c1)
        result.push_ssh_net_bytes = _net_rxtx(_s0, _s1)
        result.push_s3_net_bytes = _net_rxtx(_g0, _g1)
        log_ok(f"Push: {result.push_time_s:.1f}s  wire: {fmt_bytes(result.push_net_bytes[1])}")

        _data = measure_volume_storage(f"{COMPOSE_PROJECT}_shared-garage-data")
        _meta = measure_volume_storage(f"{COMPOSE_PROJECT}_shared-garage-meta")
        result.storage_bytes = (_data if _data >= 0 else 0) + (_meta if _meta >= 0 else 0)
        if _data < 0 and _meta < 0:
            result.storage_bytes = -1
        log_ok(f"Storage: {result.storage_bytes // (1024*1024)} MB")

        if not stop_after:
            # Clear cache so clone starts cold (dedup-within-fetch-all only).
            clear_client_desync_cache()
            reclaim_all_page_cache(SHARED_GARAGE_CONTAINER)
            log(f"  → S{result.id} Clone (client agent, non-pipelined)")
            _c0 = get_net_io(CLIENT_CONTAINER); _s0 = get_net_io(GIT_SSH_CONTAINER); _g0 = get_net_io(SHARED_GARAGE_CONTAINER)
            clone_time = git_clone_client(remote, clone_dir)
            pull_time = git_lfs_pull_client(clone_dir, extra_fetch_refspecs=clone_extra_refspecs)
            _c1 = get_net_io(CLIENT_CONTAINER); _s1 = get_net_io(GIT_SSH_CONTAINER); _g1 = get_net_io(SHARED_GARAGE_CONTAINER)
            result.clone_net_bytes = _net_rxtx(_c0, _c1)
            result.clone_ssh_net_bytes = _net_rxtx(_s0, _s1)
            result.clone_s3_net_bytes = _net_rxtx(_g0, _g1)
            result.clone_time_s = clone_time + pull_time
            log_ok(f"Clone: {result.clone_time_s:.1f}s  wire: {fmt_bytes(result.clone_net_bytes[0])}")

            result.clone_lfs_count = count_lfs_objects(clone_dir)
            result.verified = verify_clone(checksums, clone_dir, source_lfs_count)
            log_ok(f"Verified: {result.verified}  LFS objects: {result.clone_lfs_count}/{source_lfs_count if source_lfs_count >= 0 else '?'}")

    except Exception as e:
        result.error = str(e)
        log_err(f"Scenario 5 error: {e}")
    finally:
        result.resources = monitor.stop()
        # Reset the --no-pipelined override so other scenarios aren't affected.
        client_exec(["git", "config", "--global", "--unset",
                     "lfs.customtransfer.desync.args"], check=False)

    return result


def run_scenario_6(source_repo: Path, checksums: dict, raw_lfs_bytes: int,
                   tmpdir: Path, push_refspecs=None,
                   clone_extra_refspecs=None, source_lfs_count=-1,
                   container_cwd: str = None,
                   stop_after: int = None) -> ScenarioResult:
    """git-lfs-desync client agent, pipelined mode with conn-pool-size=100."""
    result = ScenarioResult(
        id=6,
        name="desync on client → Garage",
        raw_lfs_bytes=raw_lfs_bytes,
        source_lfs_count=source_lfs_count,
    )
    repo_name = "repo6"
    clone_dir = tmpdir / "clone6"
    bucket = "lfs"
    store_url = f"s3+{SHARED_GARAGE_INTERNAL}/{bucket}/chunks/"
    index_url = f"s3+{SHARED_GARAGE_INTERNAL}/{bucket}/index/"

    start_s3_service("shared-garage", setup_fn=setup_shared_garage)

    if not SHARED_GARAGE_KEY_ID or not SHARED_GARAGE_SECRET:
        result.error = "Garage2 credentials not available"
        return result

    monitor = ResourceMonitor([CLIENT_CONTAINER, GIT_SSH_CONTAINER, SHARED_GARAGE_CONTAINER])
    try:
        create_bare_repo(repo_name)
        # Configure server with desync S3 config + advertise credentials to client.
        server_cache = f"/home/git/{repo_name}.git/desync-lfs/cache"
        configure_desync_s3_repo(
            repo_name, store_url, index_url,
            SHARED_GARAGE_INTERNAL, SHARED_GARAGE_KEY_ID, SHARED_GARAGE_SECRET,
            s3_region="garage",
            cache_path=server_cache,
            advertise="with-credentials",
        )
        remote = client_remote_url(repo_name)

        # Clear chunk cache so each scenario starts fresh.
        clear_client_desync_cache()

        result.s3_container = SHARED_GARAGE_CONTAINER
        monitor.start()

        log(f"  → S{result.id} Push (client agent, pipelined)")
        _c0 = get_net_io(CLIENT_CONTAINER); _s0 = get_net_io(GIT_SSH_CONTAINER); _g0 = get_net_io(SHARED_GARAGE_CONTAINER)
        result.push_time_s = git_push_client(
            remote, source_repo,
            push_refspecs=push_refspecs,
            container_cwd=container_cwd,
            stop_after=stop_after,
        )
        _c1 = get_net_io(CLIENT_CONTAINER); _s1 = get_net_io(GIT_SSH_CONTAINER); _g1 = get_net_io(SHARED_GARAGE_CONTAINER)
        result.push_net_bytes = _net_rxtx(_c0, _c1)
        result.push_ssh_net_bytes = _net_rxtx(_s0, _s1)
        result.push_s3_net_bytes = _net_rxtx(_g0, _g1)
        log_ok(f"Push: {result.push_time_s:.1f}s  wire: {fmt_bytes(result.push_net_bytes[1])}")

        _data = measure_volume_storage(f"{COMPOSE_PROJECT}_shared-garage-data")
        _meta = measure_volume_storage(f"{COMPOSE_PROJECT}_shared-garage-meta")
        result.storage_bytes = (_data if _data >= 0 else 0) + (_meta if _meta >= 0 else 0)
        if _data < 0 and _meta < 0:
            result.storage_bytes = -1
        log_ok(f"Storage: {result.storage_bytes // (1024*1024)} MB")

        if not stop_after:
            # Clear cache so clone starts cold (dedup-within-fetch-all only).
            clear_client_desync_cache()
            reclaim_all_page_cache(SHARED_GARAGE_CONTAINER)
            log(f"  → S{result.id} Clone (client agent, pipelined)")
            _c0 = get_net_io(CLIENT_CONTAINER); _s0 = get_net_io(GIT_SSH_CONTAINER); _g0 = get_net_io(SHARED_GARAGE_CONTAINER)
            clone_time = git_clone_client(remote, clone_dir)
            pull_time = git_lfs_pull_client(clone_dir, extra_fetch_refspecs=clone_extra_refspecs)
            _c1 = get_net_io(CLIENT_CONTAINER); _s1 = get_net_io(GIT_SSH_CONTAINER); _g1 = get_net_io(SHARED_GARAGE_CONTAINER)
            result.clone_net_bytes = _net_rxtx(_c0, _c1)
            result.clone_ssh_net_bytes = _net_rxtx(_s0, _s1)
            result.clone_s3_net_bytes = _net_rxtx(_g0, _g1)
            result.clone_time_s = clone_time + pull_time
            log_ok(f"Clone: {result.clone_time_s:.1f}s  wire: {fmt_bytes(result.clone_net_bytes[0])}")

            result.clone_lfs_count = count_lfs_objects(clone_dir)
            result.verified = verify_clone(checksums, clone_dir, source_lfs_count)
            log_ok(f"Verified: {result.verified}  LFS objects: {result.clone_lfs_count}/{source_lfs_count if source_lfs_count >= 0 else '?'}")

    except Exception as e:
        result.error = str(e)
        log_err(f"Scenario 6 error: {e}")
    finally:
        result.resources = monitor.stop()

    return result


def run_scenario_8(source_repo: Path, checksums: dict, raw_lfs_bytes: int,
                   tmpdir: Path, push_refspecs=None,
                   clone_extra_refspecs=None, source_lfs_count=-1,
                   container_cwd: str = None,
                   stop_after: int = None) -> ScenarioResult:
    """Forgejo + standard S3 LFS (MinIO backend, HTTP batch API)."""
    result = ScenarioResult(
        id=8,
        name="Forgejo HTTP → MinIO",
        raw_lfs_bytes=raw_lfs_bytes,
        source_lfs_count=source_lfs_count,
    )
    repo_name = "repo8"
    clone_dir = tmpdir / "clone8"
    forgejo_host = "s8-forgejo"
    api_base = "http://localhost:3008"

    # Start S8 services (minio4 is in the forgejo profile).
    start_s3_service("shared-minio", setup_fn=setup_shared_minio)
    compose(["--profile", "forgejo", "up", "-d", "forgejo-db"], timeout=600)
    compose(["--profile", "forgejo", "up", "-d", "s8-forgejo"], timeout=600)
    _wait_forgejo_healthy(api_base, "Forgejo S3")
    setup_forgejo_instance(S8_FORGEJO_CONTAINER, api_base, "Forgejo S3")

    monitor = ResourceMonitor([CLIENT_CONTAINER, S8_FORGEJO_CONTAINER, SHARED_MINIO_CONTAINER])
    try:
        _forgejo_create_repo(api_base, repo_name)
        remote = _forgejo_remote_url(forgejo_host, repo_name)

        result.s3_container = SHARED_MINIO_CONTAINER
        monitor.start()

        log(f"  → S{result.id} Push")
        _c0 = get_net_io(CLIENT_CONTAINER); _s0 = get_net_io(S8_FORGEJO_CONTAINER); _m0 = get_net_io(SHARED_MINIO_CONTAINER)
        result.push_time_s = git_push_client(
            remote, source_repo,
            push_refspecs=push_refspecs,
            container_cwd=container_cwd,
            stop_after=stop_after,
        )
        _c1 = get_net_io(CLIENT_CONTAINER); _s1 = get_net_io(S8_FORGEJO_CONTAINER); _m1 = get_net_io(SHARED_MINIO_CONTAINER)
        result.push_net_bytes = _net_rxtx(_c0, _c1)
        result.push_ssh_net_bytes = _net_rxtx(_s0, _s1)
        result.push_s3_net_bytes = _net_rxtx(_m0, _m1)
        log_ok(f"Push: {result.push_time_s:.1f}s  wire: {fmt_bytes(result.push_net_bytes[1])}")

        result.storage_bytes = measure_volume_storage(f"{COMPOSE_PROJECT}_shared-minio-data")
        log_ok(f"Storage: {result.storage_bytes // (1024*1024)} MB")

        reclaim_all_page_cache(S8_FORGEJO_CONTAINER, SHARED_MINIO_CONTAINER)
        log(f"  → S{result.id} Clone")
        _c0 = get_net_io(CLIENT_CONTAINER); _s0 = get_net_io(S8_FORGEJO_CONTAINER); _m0 = get_net_io(SHARED_MINIO_CONTAINER)
        clone_time = git_clone_client(remote, clone_dir)
        pull_time = git_lfs_pull_client(clone_dir, extra_fetch_refspecs=clone_extra_refspecs)
        _c1 = get_net_io(CLIENT_CONTAINER); _s1 = get_net_io(S8_FORGEJO_CONTAINER); _m1 = get_net_io(SHARED_MINIO_CONTAINER)
        result.clone_net_bytes = _net_rxtx(_c0, _c1)
        result.clone_ssh_net_bytes = _net_rxtx(_s0, _s1)
        result.clone_s3_net_bytes = _net_rxtx(_m0, _m1)
        result.clone_time_s = clone_time + pull_time
        log_ok(f"Clone: {result.clone_time_s:.1f}s  wire: {fmt_bytes(result.clone_net_bytes[0])}")

        result.clone_lfs_count = count_lfs_objects(clone_dir)
        result.verified = verify_clone(checksums, clone_dir, source_lfs_count)
        log_ok(f"Verified: {result.verified}  LFS objects: {result.clone_lfs_count}/{source_lfs_count if source_lfs_count >= 0 else '?'}")

    except Exception as e:
        result.error = str(e)
        log_err(f"Scenario 8 error: {e}")
    finally:
        result.resources = monitor.stop()

    return result


def run_scenario_9(source_repo: Path, checksums: dict, raw_lfs_bytes: int,
                   tmpdir: Path, push_refspecs=None,
                   clone_extra_refspecs=None, source_lfs_count=-1,
                   container_cwd: str = None,
                   stop_after: int = None) -> ScenarioResult:
    """Forgejo + desync LFS (Garage S3 backend, HTTP batch API)."""
    result = ScenarioResult(
        id=9,
        name="Forgejo HTTP → desync Garage",
        raw_lfs_bytes=raw_lfs_bytes,
        source_lfs_count=source_lfs_count,
    )
    repo_name = "repo9"
    clone_dir = tmpdir / "clone9"
    forgejo_host = "s9-forgejo"
    api_base = "http://localhost:3009"

    # Start S9 services (garage3 is in the forgejo profile).
    start_s3_service("shared-garage", setup_fn=setup_s9_garage)
    compose(["--profile", "forgejo", "up", "-d", "forgejo-db"], timeout=600)
    compose(["--profile", "forgejo", "up", "-d", "s9-forgejo"], timeout=600)
    _wait_forgejo_healthy(api_base, "Forgejo desync")
    setup_forgejo_instance(S9_FORGEJO_CONTAINER, api_base, "Forgejo desync")

    monitor = ResourceMonitor([CLIENT_CONTAINER, S9_FORGEJO_CONTAINER, SHARED_GARAGE_CONTAINER])
    try:
        _forgejo_create_repo(api_base, repo_name)
        remote = _forgejo_remote_url(forgejo_host, repo_name)

        s9_cache = "/data/gitea/desync-cache"
        result.s3_container = SHARED_GARAGE_CONTAINER
        monitor.start()

        check_desync_cache(S9_FORGEJO_CONTAINER, s9_cache, True, "S9 before push")
        log(f"  → S{result.id} Push")
        _c0 = get_net_io(CLIENT_CONTAINER); _s0 = get_net_io(S9_FORGEJO_CONTAINER); _g0 = get_net_io(SHARED_GARAGE_CONTAINER)
        result.push_time_s = git_push_client(
            remote, source_repo,
            push_refspecs=push_refspecs,
            container_cwd=container_cwd,
            stop_after=stop_after,
        )
        _c1 = get_net_io(CLIENT_CONTAINER); _s1 = get_net_io(S9_FORGEJO_CONTAINER); _g1 = get_net_io(SHARED_GARAGE_CONTAINER)
        result.push_net_bytes = _net_rxtx(_c0, _c1)
        result.push_ssh_net_bytes = _net_rxtx(_s0, _s1)
        result.push_s3_net_bytes = _net_rxtx(_g0, _g1)
        log_ok(f"Push: {result.push_time_s:.1f}s  wire: {fmt_bytes(result.push_net_bytes[1])}")
        check_desync_cache(S9_FORGEJO_CONTAINER, s9_cache, True, "S9 after push")

        _data = measure_volume_storage(f"{COMPOSE_PROJECT}_shared-garage-data")
        _meta = measure_volume_storage(f"{COMPOSE_PROJECT}_shared-garage-meta")
        result.storage_bytes = (_data if _data > 0 else 0) + (_meta if _meta > 0 else 0)
        log_ok(f"Storage: {result.storage_bytes // (1024*1024)} MB")

        if not stop_after:
            reclaim_all_page_cache(S9_FORGEJO_CONTAINER, SHARED_GARAGE_CONTAINER)
            run(["docker", "exec", S9_FORGEJO_CONTAINER,
                 "sh", "-c", "rm -rf /data/gitea/desync-cache/*"],
                hide_output=True, check=False, timeout=30)
            log(f"  → S{result.id} Clone")
            _c0 = get_net_io(CLIENT_CONTAINER); _s0 = get_net_io(S9_FORGEJO_CONTAINER); _g0 = get_net_io(SHARED_GARAGE_CONTAINER)
            clone_time = git_clone_client(remote, clone_dir)
            pull_time = git_lfs_pull_client(clone_dir, extra_fetch_refspecs=clone_extra_refspecs)
            _c1 = get_net_io(CLIENT_CONTAINER); _s1 = get_net_io(S9_FORGEJO_CONTAINER); _g1 = get_net_io(SHARED_GARAGE_CONTAINER)
            result.clone_net_bytes = _net_rxtx(_c0, _c1)
            result.clone_ssh_net_bytes = _net_rxtx(_s0, _s1)
            result.clone_s3_net_bytes = _net_rxtx(_g0, _g1)
            result.clone_time_s = clone_time + pull_time
            log_ok(f"Clone: {result.clone_time_s:.1f}s  wire: {fmt_bytes(result.clone_net_bytes[0])}")
            check_desync_cache(S9_FORGEJO_CONTAINER, s9_cache, False, "S9 after clone")

            result.clone_lfs_count = count_lfs_objects(clone_dir)
            result.verified = verify_clone(checksums, clone_dir, source_lfs_count)
            log_ok(f"Verified: {result.verified}  LFS objects: {result.clone_lfs_count}/{source_lfs_count if source_lfs_count >= 0 else '?'}")

    except Exception as e:
        result.error = str(e)
        log_err(f"Scenario 9 error: {e}")
    finally:
        result.resources = monitor.stop()

    return result


def run_scenario_10(source_repo: Path, checksums: dict, raw_lfs_bytes: int,
                    tmpdir: Path, push_refspecs=None,
                    clone_extra_refspecs=None, source_lfs_count=-1,
                    container_cwd: str = None,
                    stop_after: int = None) -> ScenarioResult:
    """Forgejo + desync LFS via SSH pkt-line (Garage S3 backend)."""
    result = ScenarioResult(
        id=10,
        name="Forgejo SSH → desync Garage",
        raw_lfs_bytes=raw_lfs_bytes,
        source_lfs_count=source_lfs_count,
    )
    repo_name = "repo10"
    clone_dir = tmpdir / "clone10"
    forgejo_host = "s10-forgejo"
    api_base = "http://localhost:3010"

    start_s3_service("shared-garage", setup_fn=setup_s10_garage)
    compose(["--profile", "forgejo", "up", "-d", "forgejo-db", "s10-forgejo"], timeout=600)
    _wait_forgejo_healthy(api_base, "Forgejo SSH")
    setup_forgejo_instance(S10_FORGEJO_CONTAINER, api_base, "Forgejo SSH")

    monitor = ResourceMonitor([CLIENT_CONTAINER, S10_FORGEJO_CONTAINER, SHARED_GARAGE_CONTAINER])
    try:
        _forgejo_create_repo(api_base, repo_name)
        remote = _forgejo_remote_url(forgejo_host, repo_name)

        # No lfs.url — SSH pkt-line LFS transfer auto-negotiated.
        s10_cache = "/data/gitea/desync-cache"
        result.s3_container = SHARED_GARAGE_CONTAINER
        monitor.start()

        check_desync_cache(S10_FORGEJO_CONTAINER, s10_cache, True, "S10 before push")
        log(f"  → S{result.id} Push")
        _c0 = get_net_io(CLIENT_CONTAINER); _s0 = get_net_io(S10_FORGEJO_CONTAINER); _g0 = get_net_io(SHARED_GARAGE_CONTAINER)
        result.push_time_s = git_push_client(
            remote, source_repo,
            push_refspecs=push_refspecs,
            container_cwd=container_cwd,
            stop_after=stop_after,
        )
        _c1 = get_net_io(CLIENT_CONTAINER); _s1 = get_net_io(S10_FORGEJO_CONTAINER); _g1 = get_net_io(SHARED_GARAGE_CONTAINER)
        result.push_net_bytes = _net_rxtx(_c0, _c1)
        result.push_ssh_net_bytes = _net_rxtx(_s0, _s1)
        result.push_s3_net_bytes = _net_rxtx(_g0, _g1)
        log_ok(f"Push: {result.push_time_s:.1f}s  wire: {fmt_bytes(result.push_net_bytes[1])}")
        check_desync_cache(S10_FORGEJO_CONTAINER, s10_cache, True, "S10 after push")
        check_lfs_endpoint(source_repo)

        _data = measure_volume_storage(f"{COMPOSE_PROJECT}_shared-garage-data")
        _meta = measure_volume_storage(f"{COMPOSE_PROJECT}_shared-garage-meta")
        result.storage_bytes = (_data if _data > 0 else 0) + (_meta if _meta > 0 else 0)
        log_ok(f"Storage: {result.storage_bytes // (1024*1024)} MB")

        if not stop_after:
            reclaim_all_page_cache(S10_FORGEJO_CONTAINER, SHARED_GARAGE_CONTAINER)
            run(["docker", "exec", S10_FORGEJO_CONTAINER,
                 "sh", "-c", f"rm -rf {s10_cache}/*"],
                hide_output=True, check=False, timeout=30)
            log(f"  → S{result.id} Clone")
            _c0 = get_net_io(CLIENT_CONTAINER); _s0 = get_net_io(S10_FORGEJO_CONTAINER); _g0 = get_net_io(SHARED_GARAGE_CONTAINER)
            clone_time = git_clone_client(remote, clone_dir)
            pull_time = git_lfs_pull_client(clone_dir, extra_fetch_refspecs=clone_extra_refspecs)
            _c1 = get_net_io(CLIENT_CONTAINER); _s1 = get_net_io(S10_FORGEJO_CONTAINER); _g1 = get_net_io(SHARED_GARAGE_CONTAINER)
            result.clone_net_bytes = _net_rxtx(_c0, _c1)
            result.clone_ssh_net_bytes = _net_rxtx(_s0, _s1)
            result.clone_s3_net_bytes = _net_rxtx(_g0, _g1)
            result.clone_time_s = clone_time + pull_time
            log_ok(f"Clone: {result.clone_time_s:.1f}s  wire: {fmt_bytes(result.clone_net_bytes[0])}")
            check_desync_cache(S10_FORGEJO_CONTAINER, s10_cache, False, "S10 after clone")

            result.clone_lfs_count = count_lfs_objects(clone_dir)
            result.verified = verify_clone(checksums, clone_dir, source_lfs_count)
            log_ok(f"Verified: {result.verified}  LFS objects: {result.clone_lfs_count}/{source_lfs_count if source_lfs_count >= 0 else '?'}")

    except Exception as e:
        result.error = str(e)
        log_err(f"Scenario 10 error: {e}")
    finally:
        result.resources = monitor.stop()

    return result


def run_scenario_11(source_repo: Path, checksums: dict, raw_lfs_bytes: int,
                    tmpdir: Path, push_refspecs=None,
                    clone_extra_refspecs=None, source_lfs_count=-1,
                    container_cwd: str = None,
                    stop_after: int = None) -> ScenarioResult:
    """Forgejo + MinIO S3 LFS via SSH pkt-line (custom, non-desync storage)."""
    result = ScenarioResult(
        id=11,
        name="Forgejo SSH → MinIO",
        raw_lfs_bytes=raw_lfs_bytes,
        source_lfs_count=source_lfs_count,
    )
    repo_name = "repo11"
    clone_dir = tmpdir / "clone11"
    forgejo_host = "s11-forgejo"
    api_base = "http://localhost:3012"

    start_s3_service("shared-minio", setup_fn=setup_shared_minio)
    compose(["--profile", "forgejo", "up", "-d", "forgejo-db", "s11-forgejo"], timeout=600)
    _wait_forgejo_healthy(api_base, "Forgejo SSH+MinIO")
    setup_forgejo_instance(S11_FORGEJO_CONTAINER, api_base, "Forgejo SSH+MinIO")

    monitor = ResourceMonitor([CLIENT_CONTAINER, S11_FORGEJO_CONTAINER, SHARED_MINIO_CONTAINER])
    try:
        _forgejo_create_repo(api_base, repo_name)
        remote = _forgejo_remote_url(forgejo_host, repo_name)

        # No lfs.url — SSH pkt-line LFS transfer auto-negotiated.
        result.s3_container = SHARED_MINIO_CONTAINER
        monitor.start()

        log(f"  → S{result.id} Push")
        _c0 = get_net_io(CLIENT_CONTAINER); _s0 = get_net_io(S11_FORGEJO_CONTAINER); _m0 = get_net_io(SHARED_MINIO_CONTAINER)
        result.push_time_s = git_push_client(
            remote, source_repo,
            push_refspecs=push_refspecs,
            container_cwd=container_cwd,
            stop_after=stop_after,
        )
        _c1 = get_net_io(CLIENT_CONTAINER); _s1 = get_net_io(S11_FORGEJO_CONTAINER); _m1 = get_net_io(SHARED_MINIO_CONTAINER)
        result.push_net_bytes = _net_rxtx(_c0, _c1)
        result.push_ssh_net_bytes = _net_rxtx(_s0, _s1)
        result.push_s3_net_bytes = _net_rxtx(_m0, _m1)
        log_ok(f"Push: {result.push_time_s:.1f}s  wire: {fmt_bytes(result.push_net_bytes[1])}")

        result.storage_bytes = measure_volume_storage(f"{COMPOSE_PROJECT}_shared-minio-data")
        log_ok(f"Storage: {result.storage_bytes // (1024*1024)} MB")

        reclaim_all_page_cache(S11_FORGEJO_CONTAINER, SHARED_MINIO_CONTAINER)
        log(f"  → S{result.id} Clone")
        _c0 = get_net_io(CLIENT_CONTAINER); _s0 = get_net_io(S11_FORGEJO_CONTAINER); _m0 = get_net_io(SHARED_MINIO_CONTAINER)
        clone_time = git_clone_client(remote, clone_dir)
        pull_time = git_lfs_pull_client(clone_dir, extra_fetch_refspecs=clone_extra_refspecs)
        _c1 = get_net_io(CLIENT_CONTAINER); _s1 = get_net_io(S11_FORGEJO_CONTAINER); _m1 = get_net_io(SHARED_MINIO_CONTAINER)
        result.clone_net_bytes = _net_rxtx(_c0, _c1)
        result.clone_ssh_net_bytes = _net_rxtx(_s0, _s1)
        result.clone_s3_net_bytes = _net_rxtx(_m0, _m1)
        result.clone_time_s = clone_time + pull_time
        log_ok(f"Clone: {result.clone_time_s:.1f}s  wire: {fmt_bytes(result.clone_net_bytes[0])}")

        result.clone_lfs_count = count_lfs_objects(clone_dir)
        result.verified = verify_clone(checksums, clone_dir, source_lfs_count)
        log_ok(f"Verified: {result.verified}  LFS objects: {result.clone_lfs_count}/{source_lfs_count if source_lfs_count >= 0 else '?'}")

    except Exception as e:
        result.error = str(e)
        log_err(f"Scenario 11 error: {e}")
    finally:
        result.resources = monitor.stop()

    return result


def run_scenario_7(source_repo: Path, checksums: dict, raw_lfs_bytes: int,
                    tmpdir: Path, push_refspecs=None,
                    clone_extra_refspecs=None, source_lfs_count=-1,
                    container_cwd: str = None,
                    stop_after: int = None) -> ScenarioResult:
    """Stock Forgejo v14 + standard S3 LFS (MinIO backend, HTTP batch API)."""
    result = ScenarioResult(
        id=7,
        name="Stock Forgejo HTTP → MinIO",
        raw_lfs_bytes=raw_lfs_bytes,
        source_lfs_count=source_lfs_count,
    )
    repo_name = "repo7"
    clone_dir = tmpdir / "clone7"
    forgejo_host = "s7-forgejo"
    api_base = "http://localhost:3011"

    start_s3_service("shared-minio", setup_fn=setup_shared_minio)
    compose(["--profile", "forgejo", "up", "-d", "forgejo-db", "s7-forgejo"], timeout=600)
    _wait_forgejo_healthy(api_base, "Forgejo stock")
    setup_forgejo_instance(S7_FORGEJO_CONTAINER, api_base, "Forgejo stock")

    monitor = ResourceMonitor([CLIENT_CONTAINER, S7_FORGEJO_CONTAINER, SHARED_MINIO_CONTAINER])
    try:
        _forgejo_create_repo(api_base, repo_name)
        remote = _forgejo_remote_url(forgejo_host, repo_name)

        result.s3_container = SHARED_MINIO_CONTAINER
        monitor.start()

        log(f"  → S{result.id} Push")
        _c0 = get_net_io(CLIENT_CONTAINER); _s0 = get_net_io(S7_FORGEJO_CONTAINER); _m0 = get_net_io(SHARED_MINIO_CONTAINER)
        result.push_time_s = git_push_client(
            remote, source_repo,
            push_refspecs=push_refspecs,
            container_cwd=container_cwd,
            stop_after=stop_after,
        )
        _c1 = get_net_io(CLIENT_CONTAINER); _s1 = get_net_io(S7_FORGEJO_CONTAINER); _m1 = get_net_io(SHARED_MINIO_CONTAINER)
        result.push_net_bytes = _net_rxtx(_c0, _c1)
        result.push_ssh_net_bytes = _net_rxtx(_s0, _s1)
        result.push_s3_net_bytes = _net_rxtx(_m0, _m1)
        log_ok(f"Push: {result.push_time_s:.1f}s  wire: {fmt_bytes(result.push_net_bytes[1])}")

        result.storage_bytes = measure_volume_storage(f"{COMPOSE_PROJECT}_shared-minio-data")
        log_ok(f"Storage: {result.storage_bytes // (1024*1024)} MB")

        reclaim_all_page_cache(S7_FORGEJO_CONTAINER, SHARED_MINIO_CONTAINER)
        log(f"  → S{result.id} Clone")
        _c0 = get_net_io(CLIENT_CONTAINER); _s0 = get_net_io(S7_FORGEJO_CONTAINER); _m0 = get_net_io(SHARED_MINIO_CONTAINER)
        clone_time = git_clone_client(remote, clone_dir)
        pull_time = git_lfs_pull_client(clone_dir, extra_fetch_refspecs=clone_extra_refspecs)
        _c1 = get_net_io(CLIENT_CONTAINER); _s1 = get_net_io(S7_FORGEJO_CONTAINER); _m1 = get_net_io(SHARED_MINIO_CONTAINER)
        result.clone_net_bytes = _net_rxtx(_c0, _c1)
        result.clone_ssh_net_bytes = _net_rxtx(_s0, _s1)
        result.clone_s3_net_bytes = _net_rxtx(_m0, _m1)
        result.clone_time_s = clone_time + pull_time
        log_ok(f"Clone: {result.clone_time_s:.1f}s  wire: {fmt_bytes(result.clone_net_bytes[0])}")

        result.clone_lfs_count = count_lfs_objects(clone_dir)
        result.verified = verify_clone(checksums, clone_dir, source_lfs_count)
        log_ok(f"Verified: {result.verified}  LFS objects: {result.clone_lfs_count}/{source_lfs_count if source_lfs_count >= 0 else '?'}")

    except Exception as e:
        result.error = str(e)
        log_err(f"Scenario 7 error: {e}")
    finally:
        result.resources = monitor.stop()

    return result


# --- Scenario 12: Forgejo + Garage + client-side desync (auto-negotiated, pipelined) ---

def run_scenario_12(source_repo: Path, checksums: dict, raw_lfs_bytes: int,
                    tmpdir: Path, push_refspecs=None,
                    clone_extra_refspecs=None, source_lfs_count=-1,
                    container_cwd: str = None,
                    stop_after: int = None) -> ScenarioResult:
    """Forgejo + client-side desync via auto-negotiation (pipelined, Garage S3)."""
    result = ScenarioResult(
        id=12,
        name="Forgejo client desync → Garage",
        raw_lfs_bytes=raw_lfs_bytes,
        source_lfs_count=source_lfs_count,
    )
    repo_name = "repo12"
    clone_dir = tmpdir / "clone12"
    forgejo_host = "s12-forgejo"
    api_base = "http://localhost:3015"

    # S12 uses its own Forgejo instance with LFS_DESYNC_ADVERTISE=with-credentials.
    # This makes the server advertise the desync transfer agent to clients,
    # which then auto-negotiate and push/pull directly to Garage S3.
    start_s3_service("shared-garage", setup_fn=setup_s12_garage)
    compose(["--profile", "forgejo", "up", "-d", "forgejo-db", "s12-forgejo"], timeout=600)
    _wait_forgejo_healthy(api_base, "Forgejo desync client")
    setup_forgejo_instance(S12_FORGEJO_CONTAINER, api_base, "Forgejo desync client")

    monitor = ResourceMonitor([CLIENT_CONTAINER, S12_FORGEJO_CONTAINER, SHARED_GARAGE_CONTAINER])
    try:
        _forgejo_create_repo(api_base, repo_name)
        remote = _forgejo_remote_url(forgejo_host, repo_name)
        result.s3_container = SHARED_GARAGE_CONTAINER

        # Clear chunk cache so each scenario starts fresh.
        clear_client_desync_cache()

        monitor.start()

        log(f"  → S{result.id} Push (client agent via Forgejo auto-negotiation)")
        _c0 = get_net_io(CLIENT_CONTAINER); _s0 = get_net_io(S12_FORGEJO_CONTAINER); _g0 = get_net_io(SHARED_GARAGE_CONTAINER)
        result.push_time_s = git_push_client(
            remote, source_repo,
            push_refspecs=push_refspecs,
            container_cwd=container_cwd,
            stop_after=stop_after,
        )
        _c1 = get_net_io(CLIENT_CONTAINER); _s1 = get_net_io(S12_FORGEJO_CONTAINER); _g1 = get_net_io(SHARED_GARAGE_CONTAINER)
        result.push_net_bytes = _net_rxtx(_c0, _c1)
        result.push_ssh_net_bytes = _net_rxtx(_s0, _s1)
        result.push_s3_net_bytes = _net_rxtx(_g0, _g1)
        log_ok(f"Push: {result.push_time_s:.1f}s  wire: {fmt_bytes(result.push_net_bytes[1])}")

        _data = measure_volume_storage(f"{COMPOSE_PROJECT}_shared-garage-data")
        _meta = measure_volume_storage(f"{COMPOSE_PROJECT}_shared-garage-meta")
        result.storage_bytes = (_data if _data > 0 else 0) + (_meta if _meta > 0 else 0)
        log_ok(f"Storage: {result.storage_bytes // (1024*1024)} MB")

        if not stop_after:
            # Clear cache so clone starts cold (dedup-within-fetch-all only).
            clear_client_desync_cache()
            reclaim_all_page_cache(S12_FORGEJO_CONTAINER, SHARED_GARAGE_CONTAINER)
            log(f"  → S{result.id} Clone (client agent via Forgejo auto-negotiation)")
            _c0 = get_net_io(CLIENT_CONTAINER); _s0 = get_net_io(S12_FORGEJO_CONTAINER); _g0 = get_net_io(SHARED_GARAGE_CONTAINER)
            clone_time = git_clone_client(remote, clone_dir)
            pull_time = git_lfs_pull_client(clone_dir, extra_fetch_refspecs=clone_extra_refspecs)
            _c1 = get_net_io(CLIENT_CONTAINER); _s1 = get_net_io(S12_FORGEJO_CONTAINER); _g1 = get_net_io(SHARED_GARAGE_CONTAINER)
            result.clone_net_bytes = _net_rxtx(_c0, _c1)
            result.clone_ssh_net_bytes = _net_rxtx(_s0, _s1)
            result.clone_s3_net_bytes = _net_rxtx(_g0, _g1)
            result.clone_time_s = clone_time + pull_time
            log_ok(f"Clone: {result.clone_time_s:.1f}s  wire: {fmt_bytes(result.clone_net_bytes[0])}")

            result.clone_lfs_count = count_lfs_objects(clone_dir)
            result.verified = verify_clone(checksums, clone_dir, source_lfs_count)
            log_ok(f"Verified: {result.verified}  LFS objects: {result.clone_lfs_count}/{source_lfs_count if source_lfs_count >= 0 else '?'}")

    except Exception as e:
        result.error = str(e)
        log_err(f"Scenario 12 error: {e}")
    finally:
        result.resources = monitor.stop()

    return result


# --- Gitea scenarios (S13, S14) ---

def run_scenario_13(source_repo: Path, checksums: dict, raw_lfs_bytes: int,
                    tmpdir: Path, push_refspecs=None,
                    clone_extra_refspecs=None, source_lfs_count=-1,
                    container_cwd: str = None,
                    stop_after: int = None) -> ScenarioResult:
    """Stock Gitea v1.25.5 + standard S3 LFS (MinIO backend, HTTP batch API)."""
    result = ScenarioResult(
        id=13,
        name="Stock Gitea HTTP → MinIO",
        raw_lfs_bytes=raw_lfs_bytes,
        source_lfs_count=source_lfs_count,
    )
    repo_name = "repo13"
    clone_dir = tmpdir / "clone13"
    gitea_host = "s13-gitea"
    api_base = "http://localhost:3013"

    start_s3_service("shared-minio", setup_fn=setup_shared_minio)
    compose(["--profile", "forgejo", "up", "-d", "forgejo-db", "s13-gitea"], timeout=600)
    _wait_forgejo_healthy(api_base, "Gitea HTTP")
    setup_gitea_instance(S13_GITEA_CONTAINER, api_base, "Gitea HTTP")

    monitor = ResourceMonitor([CLIENT_CONTAINER, S13_GITEA_CONTAINER, SHARED_MINIO_CONTAINER])
    try:
        _forgejo_create_repo(api_base, repo_name)
        remote = _forgejo_remote_url(gitea_host, repo_name)

        result.s3_container = SHARED_MINIO_CONTAINER
        monitor.start()

        log(f"  → S{result.id} Push")
        _c0 = get_net_io(CLIENT_CONTAINER); _s0 = get_net_io(S13_GITEA_CONTAINER); _m0 = get_net_io(SHARED_MINIO_CONTAINER)
        result.push_time_s = git_push_client(
            remote, source_repo,
            push_refspecs=push_refspecs,
            container_cwd=container_cwd,
            stop_after=stop_after,
        )
        _c1 = get_net_io(CLIENT_CONTAINER); _s1 = get_net_io(S13_GITEA_CONTAINER); _m1 = get_net_io(SHARED_MINIO_CONTAINER)
        result.push_net_bytes = _net_rxtx(_c0, _c1)
        result.push_ssh_net_bytes = _net_rxtx(_s0, _s1)
        result.push_s3_net_bytes = _net_rxtx(_m0, _m1)
        log_ok(f"Push: {result.push_time_s:.1f}s  wire: {fmt_bytes(result.push_net_bytes[1])}")

        result.storage_bytes = measure_volume_storage(f"{COMPOSE_PROJECT}_shared-minio-data")
        log_ok(f"Storage: {result.storage_bytes // (1024*1024)} MB")

        reclaim_all_page_cache(S13_GITEA_CONTAINER, SHARED_MINIO_CONTAINER)
        log(f"  → S{result.id} Clone")
        _c0 = get_net_io(CLIENT_CONTAINER); _s0 = get_net_io(S13_GITEA_CONTAINER); _m0 = get_net_io(SHARED_MINIO_CONTAINER)
        clone_time = git_clone_client(remote, clone_dir)
        pull_time = git_lfs_pull_client(clone_dir, extra_fetch_refspecs=clone_extra_refspecs)
        _c1 = get_net_io(CLIENT_CONTAINER); _s1 = get_net_io(S13_GITEA_CONTAINER); _m1 = get_net_io(SHARED_MINIO_CONTAINER)
        result.clone_net_bytes = _net_rxtx(_c0, _c1)
        result.clone_ssh_net_bytes = _net_rxtx(_s0, _s1)
        result.clone_s3_net_bytes = _net_rxtx(_m0, _m1)
        result.clone_time_s = clone_time + pull_time
        log_ok(f"Clone: {result.clone_time_s:.1f}s  wire: {fmt_bytes(result.clone_net_bytes[0])}")

        result.clone_lfs_count = count_lfs_objects(clone_dir)
        result.verified = verify_clone(checksums, clone_dir, source_lfs_count)
        log_ok(f"Verified: {result.verified}  LFS objects: {result.clone_lfs_count}/{source_lfs_count if source_lfs_count >= 0 else '?'}")

    except Exception as e:
        result.error = str(e)
        log_err(f"Scenario 13 error: {e}")
    finally:
        result.resources = monitor.stop()

    return result


def run_scenario_14(source_repo: Path, checksums: dict, raw_lfs_bytes: int,
                    tmpdir: Path, push_refspecs=None,
                    clone_extra_refspecs=None, source_lfs_count=-1,
                    container_cwd: str = None,
                    stop_after: int = None) -> ScenarioResult:
    """Stock Gitea v1.25.5 + MinIO S3 LFS via SSH pkt-line transfer."""
    result = ScenarioResult(
        id=14,
        name="Stock Gitea SSH → MinIO",
        raw_lfs_bytes=raw_lfs_bytes,
        source_lfs_count=source_lfs_count,
    )
    repo_name = "repo14"
    clone_dir = tmpdir / "clone14"
    gitea_host = "s14-gitea"
    api_base = "http://localhost:3014"

    start_s3_service("shared-minio", setup_fn=setup_shared_minio)
    compose(["--profile", "forgejo", "up", "-d", "forgejo-db", "s14-gitea"], timeout=600)
    _wait_forgejo_healthy(api_base, "Gitea SSH")
    setup_gitea_instance(S14_GITEA_CONTAINER, api_base, "Gitea SSH")

    monitor = ResourceMonitor([CLIENT_CONTAINER, S14_GITEA_CONTAINER, SHARED_MINIO_CONTAINER])
    try:
        _forgejo_create_repo(api_base, repo_name)
        remote = _forgejo_remote_url(gitea_host, repo_name)

        # No lfs.url — SSH pkt-line LFS transfer auto-negotiated.
        result.s3_container = SHARED_MINIO_CONTAINER
        monitor.start()

        log(f"  → S{result.id} Push")
        _c0 = get_net_io(CLIENT_CONTAINER); _s0 = get_net_io(S14_GITEA_CONTAINER); _m0 = get_net_io(SHARED_MINIO_CONTAINER)
        result.push_time_s = git_push_client(
            remote, source_repo,
            push_refspecs=push_refspecs,
            container_cwd=container_cwd,
            stop_after=stop_after,
        )
        _c1 = get_net_io(CLIENT_CONTAINER); _s1 = get_net_io(S14_GITEA_CONTAINER); _m1 = get_net_io(SHARED_MINIO_CONTAINER)
        result.push_net_bytes = _net_rxtx(_c0, _c1)
        result.push_ssh_net_bytes = _net_rxtx(_s0, _s1)
        result.push_s3_net_bytes = _net_rxtx(_m0, _m1)
        log_ok(f"Push: {result.push_time_s:.1f}s  wire: {fmt_bytes(result.push_net_bytes[1])}")

        result.storage_bytes = measure_volume_storage(f"{COMPOSE_PROJECT}_shared-minio-data")
        log_ok(f"Storage: {result.storage_bytes // (1024*1024)} MB")

        reclaim_all_page_cache(S14_GITEA_CONTAINER, SHARED_MINIO_CONTAINER)
        log(f"  → S{result.id} Clone")
        _c0 = get_net_io(CLIENT_CONTAINER); _s0 = get_net_io(S14_GITEA_CONTAINER); _m0 = get_net_io(SHARED_MINIO_CONTAINER)
        clone_time = git_clone_client(remote, clone_dir)
        pull_time = git_lfs_pull_client(clone_dir, extra_fetch_refspecs=clone_extra_refspecs)
        _c1 = get_net_io(CLIENT_CONTAINER); _s1 = get_net_io(S14_GITEA_CONTAINER); _m1 = get_net_io(SHARED_MINIO_CONTAINER)
        result.clone_net_bytes = _net_rxtx(_c0, _c1)
        result.clone_ssh_net_bytes = _net_rxtx(_s0, _s1)
        result.clone_s3_net_bytes = _net_rxtx(_m0, _m1)
        result.clone_time_s = clone_time + pull_time
        log_ok(f"Clone: {result.clone_time_s:.1f}s  wire: {fmt_bytes(result.clone_net_bytes[0])}")

        result.clone_lfs_count = count_lfs_objects(clone_dir)
        result.verified = verify_clone(checksums, clone_dir, source_lfs_count)
        log_ok(f"Verified: {result.verified}  LFS objects: {result.clone_lfs_count}/{source_lfs_count if source_lfs_count >= 0 else '?'}")

    except Exception as e:
        result.error = str(e)
        log_err(f"Scenario 14 error: {e}")
    finally:
        result.resources = monitor.stop()

    return result


# ---------------------------------------------------------------------------
# Cross-compatibility tests
# ---------------------------------------------------------------------------
def run_cross_compat_3_client(checksums: dict, tmpdir: Path, source_lfs_count: int = -1) -> CrossCompatResult:
    """Clone scenario-3 data (uploaded by server to MinIO) using auto-negotiated client agent."""
    result = CrossCompatResult(name="3→client: server MinIO data via client agent")
    remote = client_remote_url("repo3")
    clone_dir = tmpdir / "xcompat3"

    try:
        client_exec(["rm", "-rf", str(clone_dir)], check=False)
        clone_time = git_clone_client(remote, clone_dir)
        pull_time = git_lfs_pull_client(clone_dir)
        result.clone_time_s = clone_time + pull_time
        result.verified = verify_clone(checksums, clone_dir, source_lfs_count)
    except Exception as e:
        result.error = str(e)
        log_err(f"Cross-compat 3→client error: {e}")

    return result


def run_cross_compat_4_client(checksums: dict, tmpdir: Path, source_lfs_count: int = -1) -> CrossCompatResult:
    """Clone scenario-4 data (uploaded by server to Garage) using auto-negotiated client agent."""
    result = CrossCompatResult(name="4→client: server Garage data via client agent")
    remote = client_remote_url("repo4")
    clone_dir = tmpdir / "xcompat4"

    try:
        client_exec(["rm", "-rf", str(clone_dir)], check=False)
        clone_time = git_clone_client(remote, clone_dir)
        pull_time = git_lfs_pull_client(clone_dir)
        result.clone_time_s = clone_time + pull_time
        result.verified = verify_clone(checksums, clone_dir, source_lfs_count)
    except Exception as e:
        result.error = str(e)
        log_err(f"Cross-compat 4→client error: {e}")

    return result


def run_cross_compat_5_server(checksums: dict, tmpdir: Path, source_lfs_count: int = -1) -> CrossCompatResult:
    """Clone scenario-5 data (uploaded by client to MinIO) using server SSH transfer."""
    result = CrossCompatResult(name="5→server: client MinIO data via server SSH transfer")
    # The server's repo5.git was configured with a desync-lfs.json (in run_scenario_5)
    # pointing to minio. So git clone via SSH + lfs pull uses the server-side transfer.
    remote = client_remote_url("repo5")
    clone_dir = tmpdir / "xcompat5"

    try:
        client_exec(["rm", "-rf", str(clone_dir)], check=False)
        clone_time = git_clone_client(remote, clone_dir)
        pull_time = git_lfs_pull_client(clone_dir)
        result.clone_time_s = clone_time + pull_time
        result.verified = verify_clone(checksums, clone_dir, source_lfs_count)
    except Exception as e:
        result.error = str(e)
        log_err(f"Cross-compat 5→server error: {e}")

    return result


def run_cross_compat_6_server(checksums: dict, tmpdir: Path, source_lfs_count: int = -1) -> CrossCompatResult:
    """Clone scenario-6 data (uploaded by client to Garage2) using server SSH transfer."""
    result = CrossCompatResult(name="6→server: client Garage data via server SSH transfer")
    if not SHARED_GARAGE_KEY_ID or not SHARED_GARAGE_SECRET:
        result.error = "Garage2 credentials not available"
        return result

    remote = client_remote_url("repo6")
    clone_dir = tmpdir / "xcompat6"

    try:
        client_exec(["rm", "-rf", str(clone_dir)], check=False)
        clone_time = git_clone_client(remote, clone_dir)
        pull_time = git_lfs_pull_client(clone_dir)
        result.clone_time_s = clone_time + pull_time
        result.verified = verify_clone(checksums, clone_dir, source_lfs_count)
    except Exception as e:
        result.error = str(e)
        log_err(f"Cross-compat 6→server error: {e}")

    return result


def run_cross_compat_9_client(checksums: dict, tmpdir: Path, source_lfs_count: int = -1) -> CrossCompatResult:
    """Clone scenario-9 data (uploaded by Forgejo desync to Garage3) using client agent."""
    result = CrossCompatResult(name="9→client: Forgejo desync data via client agent")
    remote = _forgejo_remote_url("s9-forgejo", "repo9")
    clone_dir = tmpdir / "xcompat9"

    try:
        client_exec(["rm", "-rf", str(clone_dir)], check=False)
        clone_time = git_clone_client(remote, clone_dir)
        pull_time = git_lfs_pull_client(clone_dir)
        result.clone_time_s = clone_time + pull_time
        result.verified = verify_clone(checksums, clone_dir, source_lfs_count)
    except Exception as e:
        result.error = str(e)
        log_err(f"Cross-compat 9→client error: {e}")

    return result


def run_cross_compat_10_http(checksums: dict, tmpdir: Path, source_lfs_count: int = -1) -> CrossCompatResult:
    """Clone scenario-10 data (uploaded via SSH pkt-line) using HTTP batch API."""
    result = CrossCompatResult(name="10→http: SSH desync data via HTTP batch API")

    forgejo_host = "s10-forgejo"
    repo_name = "repo10"
    remote = _forgejo_remote_url(forgejo_host, repo_name)
    clone_dir = tmpdir / "xcompat10"

    # Force HTTP data transfer by disabling the pure SSH protocol.
    # git-lfs falls back to hybrid mode: SSH git-lfs-authenticate for auth,
    # HTTP batch API for data transfer.
    lfs_config = ["lfs.sshtransfer=never"]

    try:
        client_exec(["rm", "-rf", str(clone_dir)], check=False)
        clone_time = git_clone_client(remote, clone_dir, extra_git_config=lfs_config)
        pull_time = git_lfs_pull_client(clone_dir, extra_git_config=lfs_config)
        result.clone_time_s = clone_time + pull_time
        result.verified = verify_clone(checksums, clone_dir, source_lfs_count)
    except Exception as e:
        result.error = str(e)
        log_err(f"Cross-compat 10→http error: {e}")

    return result


# ---------------------------------------------------------------------------
# Results reporting
# ---------------------------------------------------------------------------
def fmt_bytes(n: int) -> str:
    if n < 0:
        return "N/A"
    if n < 1024:
        return f"{n} B"
    if n < 1024 ** 2:
        return f"{n/1024:.1f} KB"
    if n < 1024 ** 3:
        return f"{n/1024/1024:.1f} MB"
    return f"{n/1024/1024/1024:.2f} GB"


def fmt_time(s: float) -> str:
    if s <= 0:
        return "N/A"
    if s < 60:
        return f"{s:.1f}s"
    total = int(s)
    if total >= 3600:
        h = total // 3600
        m = (total % 3600) // 60
        return f"{h}h{m:02d}m"
    m = total // 60
    sec = total % 60
    return f"{m}m{sec:02d}s"


def fmt_pct(v: float) -> str:
    if v <= 0:
        return "N/A"
    return f"{v*100:.0f}%"


def fmt_ratio(v: float) -> str:
    if v <= 0:
        return "N/A"
    return f"{v:.2f}×"


def _table_row(row, col_w, color):
    log("  ".join(v.ljust(w) for v, w in zip(row, col_w)), color)


def _table_header(headers, col_w):
    log("  ".join(h.ljust(w) for h, w in zip(headers, col_w)), BOLD)
    log("  ".join("-" * w for w in col_w))


def _fmt_cpu(s):
    if s.sample_count == 0:
        return "N/A"
    return f"{s.avg_cpu_percent:.0f}%/{s.max_cpu_percent:.0f}%"

def _fmt_mem(s):
    if s.sample_count == 0:
        return "N/A"
    total = fmt_bytes(s.max_mem_bytes)
    if s.max_anon_bytes > 0:
        return f"{total}({fmt_bytes(s.max_anon_bytes)})"
    return total


def _md_table(headers: list, rows: list) -> str:
    """Build a markdown table with | separators and right-aligned numeric columns."""
    all_rows = [headers] + rows
    col_w = [max(len(str(r[i])) for r in all_rows) for i in range(len(headers))]
    # Right-align all columns except # and Scenario (indices 0, 1)
    def _align(val, w, col_idx):
        return val.ljust(w) if col_idx <= 1 else val.rjust(w)
    def _fmt_row(row):
        cells = " | ".join(_align(str(v), col_w[i], i) for i, v in enumerate(row))
        return f"| {cells} |"
    sep_parts = []
    for i, w in enumerate(col_w):
        if i <= 1:
            sep_parts.append("-" * (w + 2))  # left-aligned
        else:
            sep_parts.append("-" * (w + 1) + ":")  # right-aligned
    sep = "|" + "|".join(sep_parts) + "|"
    lines = [_fmt_row(headers), sep] + [_fmt_row(r) for r in rows]
    return "\n".join(lines)


def _inject_ref_lines(svg_path: str, ref_lines: list, max_val: float):
    """Inject vertical reference lines into a pygal HorizontalBar SVG.

    ref_lines: list of (value, color, label) tuples.
    max_val: the caller's estimated x-axis maximum; used only as a fallback
        when the rendered axis scale can't be parsed from the SVG.  Pygal
        rounds axis ranges to nice numbers, so the rendered max may differ
        from the value passed to chart.range — we parse the actual axis
        label positions to pin reference lines to the correct x coordinates.
    """
    import re
    svg = Path(svg_path).read_text()
    # Find the plot area dimensions from the transform and background rect.
    plot_match = re.search(
        r'<g transform="translate\(([0-9.]+),\s*([0-9.]+)\)" class="plot">'
        r'\s*<rect x="0" y="0" width="([0-9.]+)" height="([0-9.]+)"',
        svg,
    )
    if not plot_match:
        return
    plot_x = float(plot_match.group(1))
    plot_y = float(plot_match.group(2))
    plot_w = float(plot_match.group(3))
    plot_h = float(plot_match.group(4))

    # Determine the actual plotted axis range by parsing the x-axis tick
    # labels.  Pygal places them inside the plot transform at x positions
    # that encode the linear mapping between data values and pixels.  The
    # relevant labels are numeric and sit just below the plot area
    # (y ≈ plot_h + a small offset).
    tick_y = plot_h + 14  # pygal's default tick label offset (y="404" for plot_h=389)
    ticks = []
    for m in re.finditer(r'<text x="([0-9.-]+)" y="([0-9.-]+)"[^>]*>([^<]+)</text>', svg):
        x_val, y_val, txt = float(m.group(1)), float(m.group(2)), m.group(3)
        if abs(y_val - tick_y) > 2:
            continue
        try:
            v = float(txt)
        except ValueError:
            continue
        ticks.append((v, x_val))
    ticks.sort()
    # Need at least two distinct values to interpolate.
    unique_vals = {v for v, _ in ticks}
    if len(unique_vals) >= 2:
        v_min, x_at_min = ticks[0]
        v_max, x_at_max = ticks[-1]
        scale = (x_at_max - x_at_min) / (v_max - v_min)
        def val_to_doc_x(v):
            return plot_x + x_at_min + (v - v_min) * scale
    else:
        # Fallback to the caller-provided max_val over the full plot width.
        def val_to_doc_x(v):
            return plot_x + (v / max_val) * plot_w

    lines_svg = ""
    text_y = plot_y - 4  # all labels at the same height, just above the plot
    for val, color, label in ref_lines:
        x = val_to_doc_x(val)
        lines_svg += (
            f'<line x1="{x:.1f}" y1="{plot_y}" x2="{x:.1f}" '
            f'y2="{plot_y + plot_h:.1f}" stroke="{color}" '
            f'stroke-width="2" stroke-dasharray="6,4" opacity="0.8"/>\n'
            f'<text x="{x - 3:.1f}" y="{text_y:.1f}" text-anchor="end" '
            f'font-size="10" fill="{color}" opacity="0.9">{label}</text>\n'
        )
    # Insert before the closing </svg> tag.
    svg = svg.replace("</svg>", lines_svg + "</svg>")
    Path(svg_path).write_text(svg)


def _generate_charts(results: list, raw_lfs_bytes: int, git_packed_bytes: int,
                     output_prefix: str):
    """Generate interactive SVG charts from results data using pygal."""
    try:
        import pygal
        import pygal.style
    except ImportError:
        log_warn("pygal not available, skipping chart generation")
        return []

    GB = 1024**3
    MB = 1024**2
    # Reverse so S0 appears at the top of horizontal bar charts.
    results_rev = list(reversed(results))
    labels = [f"{r.name} (S{r.id})" for r in results_rev]

    _font = '"Segoe UI", Roboto, "Noto Sans", Ubuntu, "Helvetica Neue", Arial, sans-serif'
    style = pygal.style.Style(
        background="#111",
        plot_background="#222",
        foreground="#999",
        foreground_strong="#eee",
        foreground_subtle="#555",
        colors=("#8cedff", "#ff5995", "#feed6c", "#b6e354", "#9e6ffe"),
        font_family=_font,
        title_font_family=_font,
        legend_font_family=_font,
        label_font_family=_font,
        major_label_font_family=_font,
        tooltip_font_family=_font,
        value_font_family=_font,
    )
    chart_files = []

    def _fmt_time(v, unit):
        if unit == "minutes":
            return f"{v:.1f} min"
        return f"{v:.1f} s"

    def _fmt_size(v, unit):
        return f"{v:.1f} {unit}"

    # ── Chart 1: Push + Clone times ──────────────────────────────────────
    push_vals = [r.push_time_s if r.push_time_s > 0 else 0 for r in results_rev]
    clone_vals = [r.clone_time_s if r.clone_time_s > 0 else 0 for r in results_rev]
    max_time = max(max(push_vals, default=0), max(clone_vals, default=0))
    if max_time > 120:
        push_data = [{"value": v / 60, "label": _fmt_time(v / 60, "minutes")} for v in push_vals]
        clone_data = [{"value": v / 60, "label": _fmt_time(v / 60, "minutes")} for v in clone_vals]
        time_unit = "minutes"
    else:
        push_data = [{"value": v, "label": _fmt_time(v, "seconds")} for v in push_vals]
        clone_data = [{"value": v, "label": _fmt_time(v, "seconds")} for v in clone_vals]
        time_unit = "seconds"

    chart = pygal.HorizontalBar(
        style=style, show_legend=True, human_readable=True,
        legend_at_bottom=True, legend_at_bottom_columns=2,
        title=f"Push and Clone Times ({time_unit})",
        x_title=time_unit,
        width=800, margin_left=-50,
        height=max(400, len(results_rev) * 40),
        print_values=False, truncate_label=-1,
    )
    chart.x_labels = labels
    chart.add("Push", push_data)
    chart.add("Clone", clone_data)
    path = f"{output_prefix}-times.svg"
    chart.render_to_file(path)
    chart_files.append(path)

    # ── Chart 2: Storage efficiency ──────────────────────────────────────
    storage_raw = [r.storage_bytes if r.storage_bytes > 0 else 0 for r in results_rev]
    max_storage = max(max(storage_raw, default=0), raw_lfs_bytes, git_packed_bytes)
    if max_storage > 2 * GB:
        divisor, size_unit = GB, "GB"
    else:
        divisor, size_unit = MB, "MB"

    desync_ids = {2, 3, 4, 5, 6, 9, 10, 12}
    git_only_ids = {0}
    # Build three separate series so the legend shows the color categories.
    git_data = []
    lfs_data = []
    desync_data = []
    for r, raw in zip(results_rev, storage_raw):
        v = raw / divisor
        entry = {"value": v, "label": _fmt_size(v, size_unit)}
        if r.id in git_only_ids:
            git_data.append(entry)
            lfs_data.append(None)
            desync_data.append(None)
        elif r.id in desync_ids:
            git_data.append(None)
            lfs_data.append(None)
            desync_data.append(entry)
        else:
            git_data.append(None)
            lfs_data.append(entry)
            desync_data.append(None)

    storage_style = pygal.style.Style(
        background="#111",
        plot_background="#222",
        foreground="#999",
        foreground_strong="#eee",
        foreground_subtle="#555",
        colors=("#899ca1", "#ff5995", "#b6e354"),
        font_family=_font,
        title_font_family=_font,
        legend_font_family=_font,
        label_font_family=_font,
        major_label_font_family=_font,
        tooltip_font_family=_font,
        value_font_family=_font,
    )
    chart = pygal.HorizontalStackedBar(
        style=storage_style, show_legend=True, human_readable=True,
        legend_at_bottom=True, legend_at_bottom_columns=3,
        title=f"Storage Size ({size_unit})",
        x_title=size_unit,
        width=800, margin_left=-50,
        height=max(400, len(results_rev) * 35),
        print_values=False, truncate_label=-1,
    )
    chart.x_labels = labels
    chart.add("Git objects", git_data)
    chart.add("LFS objects", lfs_data)
    chart.add("desync chunks", desync_data)
    # Collect reference lines to inject as vertical lines after rendering.
    ref_lines = []
    if raw_lfs_bytes > 0:
        ref_lines.append((raw_lfs_bytes / divisor, "#feed6c", f"Raw LFS ({raw_lfs_bytes / divisor:.1f} {size_unit})"))
    if git_packed_bytes > 0:
        ref_lines.append((git_packed_bytes / divisor, "#8cedff", f"Git repacked ({git_packed_bytes / divisor:.1f} {size_unit})"))
    all_vals = [v for v in (e["value"] for s in [git_data, lfs_data, desync_data] for e in s if e) if v > 0]
    max_val = max(max(all_vals, default=0), *(v for v, _, _ in ref_lines)) * 1.1
    chart.range = (0, max_val)
    path = f"{output_prefix}-storage.svg"
    chart.render_to_file(path)
    # Inject vertical reference lines into the SVG.
    if ref_lines:
        _inject_ref_lines(path, ref_lines, max_val)
    chart_files.append(path)

    # ── Chart 3: Network traffic (push: client, server, S3) ─────────────
    client_raw = [_net_sum(r.push_net_bytes) for r in results_rev]
    server_raw = [_net_sum(r.push_ssh_net_bytes) for r in results_rev]
    s3_raw = [_net_sum(r.push_s3_net_bytes) for r in results_rev]
    max_traffic = max(max(client_raw, default=0), max(server_raw, default=0), max(s3_raw, default=0))
    if max_traffic > 2 * GB:
        tdiv, tunit = GB, "GB"
    else:
        tdiv, tunit = MB, "MB"

    client_data = [{"value": v / tdiv, "label": _fmt_size(v / tdiv, tunit)} for v in client_raw]
    server_data = [{"value": v / tdiv, "label": _fmt_size(v / tdiv, tunit)} for v in server_raw]
    s3_data = [{"value": v / tdiv, "label": _fmt_size(v / tdiv, tunit)} for v in s3_raw]

    chart = pygal.HorizontalBar(
        style=style, show_legend=True, human_readable=True,
        legend_at_bottom=True, legend_at_bottom_columns=3,
        title=f"Push Network Traffic ({tunit})",
        x_title=tunit,
        width=800, margin_left=-50,
        height=max(400, len(results_rev) * 45),
        truncate_label=-1,
        print_values=False,
    )
    chart.x_labels = labels
    chart.add("Client", client_data)
    chart.add("Server", server_data)
    chart.add("S3", s3_data)
    path = f"{output_prefix}-traffic.svg"
    chart.render_to_file(path)
    chart_files.append(path)

    return chart_files


def _build_results_text(results: list, xcompat: list, raw_lfs_bytes: int,
                        source_lfs_count: int, git_packed_bytes: int = 0,
                        total_time_s: float = 0, chart_prefix: str = "") -> str:
    """Build markdown results tables (no ANSI codes). Returns the text."""
    lines = []

    lines.append(f"- Raw LFS data size: {fmt_bytes(raw_lfs_bytes)}")
    if git_packed_bytes > 0:
        ratio = raw_lfs_bytes / git_packed_bytes if git_packed_bytes > 0 else 0
        lines.append(f"- Git-only repacked size: {fmt_bytes(git_packed_bytes)} ({ratio:.2f}× dedup)")
    lines.append(f"- Chunk size: {CHUNK_SIZE} KB")
    lines.append(f"- LFS objects: {source_lfs_count}")
    if total_time_s > 0:
        lines.append(f"- Total experiment time: {fmt_time(total_time_s)}")
    lines.append("")

    # ── Charts (before tables) ───────────────────────────────────────────────
    if chart_prefix:
        prefix = Path(chart_prefix).name
        lines.append(f"![Push and Clone Times]({prefix}-times.svg)")
        lines.append("")
        lines.append(f"![Storage Size]({prefix}-storage.svg)")
        lines.append("")
        lines.append(f"![Push Network Traffic]({prefix}-traffic.svg)")
        lines.append("")

    # ── Table 1: Timing & Storage ─────────────────────────────────────────────
    lines.append("**Table 1 — Timing & Storage**")
    lines.append("")
    rows1 = []
    for r in results:
        if r.verified:
            status = "✓"
        elif r.error:
            status = "ERR"
        else:
            status = f"{r.clone_lfs_count}/{r.source_lfs_count}"
        rows1.append([
            str(r.id), r.name,
            fmt_time(r.push_time_s), fmt_time(r.clone_time_s),
            fmt_bytes(r.storage_bytes), fmt_ratio(r.storage_ratio), status,
        ])
    lines.append(_md_table(["#", "Scenario", "Push", "Clone", "Storage", "Ratio", "OK?"], rows1))
    for r in results:
        if r.error:
            lines.append(f"\n> **S{r.id} error:** {r.error}")
    lines.append("")

    # ── Table 2: Network Traffic ──────────────────────────────────────────────
    lines.append("**Table 2 — Network Traffic**")
    lines.append("")
    rows2 = []
    for r in results:
        rows2.append([
            str(r.id), r.name,
            fmt_bytes(_net_sum(r.push_net_bytes)),
            fmt_bytes(_net_sum(r.push_ssh_net_bytes)),
            fmt_bytes(_net_sum(r.push_s3_net_bytes)),
            fmt_bytes(_net_sum(r.clone_net_bytes)),
            fmt_bytes(_net_sum(r.clone_ssh_net_bytes)),
            fmt_bytes(_net_sum(r.clone_s3_net_bytes)),
        ])
    lines.append(_md_table(
        ["#", "Scenario", "push client", "push server", "push s3", "clone client", "clone server", "clone s3"], rows2))
    lines.append("")

    # ── Table 3: Resource Usage ───────────────────────────────────────────────
    has_resources = any(r.resources for r in results)
    if has_resources:
        lines.append("**Table 3 — CPU Usage (avg/peak)**")
        lines.append("")
        rows_cpu = []
        rows_mem = []
        for r in results:
            res = r.resources
            client_r = res.get(CLIENT_CONTAINER, ResourceSummary())
            # For Forgejo scenarios, the server is the Forgejo container, not git-ssh.
            ssh_r = res.get(GIT_SSH_CONTAINER, None)
            if ssh_r is None:
                for cname in (S8_FORGEJO_CONTAINER, S9_FORGEJO_CONTAINER, S10_FORGEJO_CONTAINER, S11_FORGEJO_CONTAINER, S12_FORGEJO_CONTAINER, S7_FORGEJO_CONTAINER, S13_GITEA_CONTAINER, S14_GITEA_CONTAINER):
                    ssh_r = res.get(cname, None)
                    if ssh_r is not None:
                        break
                if ssh_r is None:
                    ssh_r = ResourceSummary()
            s3_name = r.s3_container
            s3_r = res.get(s3_name, ResourceSummary()) if s3_name else ResourceSummary()
            rows_cpu.append([
                str(r.id), r.name,
                _fmt_cpu(client_r), _fmt_cpu(ssh_r),
                _fmt_cpu(s3_r) if s3_name else "N/A",
            ])
            rows_mem.append([
                str(r.id), r.name,
                _fmt_mem(client_r), _fmt_mem(ssh_r),
                _fmt_mem(s3_r) if s3_name else "N/A",
            ])
        lines.append(_md_table(
            ["#", "Scenario", "client CPU", "server CPU", "s3 CPU"], rows_cpu))
        lines.append("")
        lines.append("**Table 4 — Memory Usage (total, RSS in parentheses)**")
        lines.append("")
        lines.append(_md_table(
            ["#", "Scenario", "client mem", "server mem", "s3 mem"], rows_mem))
        lines.append("")

    # ── Cross-compatibility ───────────────────────────────────────────────────
    if xcompat:
        lines.append("**Cross-compatibility Tests**")
        lines.append("")
        xrows = []
        for xr in xcompat:
            status = "✓" if xr.verified else ("ERR" if xr.error else "✗")
            time_str = fmt_time(xr.clone_time_s) if xr.clone_time_s > 0 else "N/A"
            xrows.append([xr.name, time_str, status])
        lines.append(_md_table(["Test", "Clone", "OK?"], xrows))
        lines.append("")

    return "\n".join(lines)


def print_results(results: list, xcompat: list, raw_lfs_bytes: int,
                  source_lfs_count: int, git_packed_bytes: int = 0):
    W = 94
    log("\n" + "=" * W, BOLD)
    log("RESULTS", BOLD + BLUE)
    log("=" * W, BOLD)

    text = _build_results_text(results, xcompat, raw_lfs_bytes, source_lfs_count, git_packed_bytes)
    for line in text.splitlines():
        if line.startswith("Table ") or line.startswith("Cross-compat"):
            log(line, BOLD + CYAN)
        elif "ERR" in line or "Error:" in line:
            log(line, RED)
        elif "✗" in line:
            log(line, YELLOW)
        elif "✓" in line:
            log(line, GREEN)
        else:
            log(line)

    log("\n" + "=" * W, BOLD)


# ---------------------------------------------------------------------------
# Main
# ---------------------------------------------------------------------------
def parse_args():
    p = argparse.ArgumentParser(
        description="LFS transfer benchmark comparing git-lfs-transfer variants"
    )
    p.add_argument(
        "--dataset", choices=["small", "r0"], default="small",
        help="Dataset to use: 'small' (synthetic ~200MB, 50 commits) or 'r0' (R0.git, ~23GB). Default: small",
    )
    p.add_argument(
        "--data-dir", type=Path, default=EXPERIMENT_DIR / "data" / "R0.git",
        help="Path to R0.git bare repo (only with --dataset r0)",
    )
    p.add_argument(
        "--data-dir0", type=Path, default=EXPERIMENT_DIR / "data" / "R0-nolfs.git",
        help="Path to non-LFS R0 bare repo for scenario 0 (only with --dataset r0)",
    )
    p.add_argument(
        "--scenarios", default="0,1,2,3,4,5,6,7,8,9,10,11,12,13,14",
        help="Comma-separated list of scenario IDs to run. Default: 0,1,2,3,4,5,6,7,8,9,10,11",
    )
    p.add_argument(
        "--no-cross-compat", action="store_true",
        help="Skip cross-compatibility tests",
    )
    p.add_argument(
        "--stop-after", type=int, default=None,
        help="Stop after pushing N refs (skip clone/verify). For quick profiling runs.",
    )
    p.add_argument(
        "--markdown-only", action="store_true",
        help="Regenerate .md from existing .json results (no experiments run)",
    )
    p.add_argument(
        "--skip-build", action="store_true",
        help="Skip building binaries (use existing binaries in experiment/binaries/)",
    )
    p.add_argument(
        "--skip-setup", action="store_true",
        help="Skip docker-compose setup (assume services already running with correct mounts). Also skips WORK_DIR cleanup so the source repo is preserved.",
    )
    p.add_argument(
        "--keep-containers", action="store_true",
        help="Do not stop docker containers after the experiment",
    )
    p.add_argument(
        "--verbose", "-v", action="store_true",
        help="Show all subprocess output (builds, docker commands, etc.)",
    )
    p.add_argument(
        "--work-dir", type=Path, default=EXPERIMENT_DIR / "tmp" / "lfs-exp",
        help="Working directory for temporary repos. Default: <repo>/tmp/lfs-exp",
    )
    p.add_argument(
        "--output", type=str, default=None,
        help="Base name for output files (produces <name>.json and <name>.md). Default: results-<dataset>",
    )
    return p.parse_args()


def main():
    global WORK_DIR, VERBOSE
    args = parse_args()
    VERBOSE = args.verbose
    ALL_SCENARIOS = "0,1,2,3,4,5,6,7,8,9,10,11,12,13,14"
    if args.output is None:
        base = f"results-{args.dataset}"
        if args.scenarios != ALL_SCENARIOS:
            ids = ",".join(s.strip() for s in args.scenarios.split(","))
            base += f"-s{ids}"
        args.output = base

    log(f"\n{'LFS Transfer Benchmark':^60}", BOLD + BLUE)
    log(f"{'=' * 60}", BOLD)
    log(f"  GIT_LFS_CONCURRENCY = {GIT_LFS_CONCURRENCY}")
    log(f"  DESYNC_CONCURRENCY  = {DESYNC_CONCURRENCY}")
    log(f"  CHUNK_SIZE          = {CHUNK_SIZE}")
    log(f"  MAX_IN_FLIGHT       = {MAX_IN_FLIGHT // (1024**2)} MB" if MAX_IN_FLIGHT else "  MAX_IN_FLIGHT       = disabled")
    log(f"  MAX_STORAGE_OPS     = {MAX_STORAGE_OPS}" if MAX_STORAGE_OPS else "  MAX_STORAGE_OPS     = disabled")
    log(f"  S1_LFS_TRANSFER    = {S1_LFS_TRANSFER}")
    log(f"  Output              = {args.output}.json / {args.output}.md")

    results_file = EXPERIMENT_DIR / f"{args.output}.json"
    results_md = EXPERIMENT_DIR / f"{args.output}.md"

    # --markdown-only: reload JSON and regenerate markdown without running experiments.
    if args.markdown_only:
        if not results_file.exists():
            log(f"No results file found: {results_file}", RED)
            return 1
        data = json.loads(results_file.read_text())
        results = []
        for s in data["scenarios"]:
            r = ScenarioResult(id=s["id"], name=s["name"], raw_lfs_bytes=s.get("raw_lfs_bytes", 0),
                               source_lfs_count=s.get("source_lfs_count", -1))
            r.push_time_s = s.get("push_time_s", 0)
            r.clone_time_s = s.get("clone_time_s", 0)
            r.storage_bytes = s.get("storage_bytes", 0)
            r.push_net_bytes = tuple(s.get("push_net_bytes", (0, 0)))
            r.push_ssh_net_bytes = tuple(s.get("push_ssh_net_bytes", (0, 0)))
            r.push_s3_net_bytes = tuple(s.get("push_s3_net_bytes", (0, 0)))
            r.clone_net_bytes = tuple(s.get("clone_net_bytes", (0, 0)))
            r.clone_ssh_net_bytes = tuple(s.get("clone_ssh_net_bytes", (0, 0)))
            r.clone_s3_net_bytes = tuple(s.get("clone_s3_net_bytes", (0, 0)))
            r.s3_container = s.get("s3_container", "")
            r.clone_lfs_count = s.get("clone_lfs_count", 0)
            r.verified = s.get("verified", False)
            r.error = s.get("error", "")
            r.resources = {
                name: ResourceSummary(**vals)
                for name, vals in s.get("resources", {}).items()
            }
            results.append(r)
        xcompat = []
        for xc in data.get("cross_compat", []):
            xr = CrossCompatResult(name=xc["name"])
            xr.clone_time_s = xc.get("clone_time_s", 0)
            xr.verified = xc.get("verified", False)
            xr.error = xc.get("error", "")
            xcompat.append(xr)
        raw_lfs_bytes = data.get("raw_lfs_bytes", 0)
        git_packed_bytes = data.get("git_packed_bytes", 0)
        source_lfs_count = data.get("source_lfs_count",
                                    results[0].source_lfs_count if results else 0)
        total_time_s = data.get("total_time_s", 0)
        chart_prefix = str(EXPERIMENT_DIR / args.output)
        chart_files = _generate_charts(results, raw_lfs_bytes, git_packed_bytes, chart_prefix)
        text = _build_results_text(results, xcompat, raw_lfs_bytes,
                                   source_lfs_count, git_packed_bytes,
                                   total_time_s=total_time_s,
                                   chart_prefix=chart_prefix)
        results_md.write_text(text + "\n")
        log(f"Regenerated: {results_md}", CYAN)
        for cf in chart_files:
            log(f"  Chart: {cf}", CYAN)
        return 0

    scenarios_to_run = [int(s.strip()) for s in args.scenarios.split(",")]

    skip_setup = args.skip_setup

    # Working directory for repos (also mounted into the client container)
    WORK_DIR = args.work_dir
    if not skip_setup:
        if WORK_DIR.exists():
            # Clean contents but keep the directory itself so Docker bind mounts
            # (which track the inode) stay valid across runs.
            for item in WORK_DIR.iterdir():
                try:
                    shutil.rmtree(item) if item.is_dir() else item.unlink()
                except PermissionError:
                    # Docker may create dirs owned by root; remove via container.
                    run(["docker", "run", "--rm", "-v", f"{WORK_DIR}:/w",
                         "alpine", "rm", "-rf", f"/w/{item.name}"],
                        check=False, hide_output=True, timeout=30)
        else:
            WORK_DIR.mkdir(parents=True)
    else:
        WORK_DIR.mkdir(parents=True, exist_ok=True)

    # ---- Build binaries ----
    if not args.skip_build:
        build_binaries()
    else:
        log_step("Skipping binary build (--skip-build)")

    # ---- SSH keys ----
    setup_ssh_keys()

    # ---- Start services ----
    # Tear down all containers and volumes from any previous run to start clean.
    if not skip_setup:
        log_step("Cleaning up previous containers and volumes")
        compose(["--profile", "s3", "--profile", "forgejo", "down", "-v", "--remove-orphans"], check=False, timeout=300)

    # For R0 dataset, mount the bare repo's lfs/ directory into the client
    # container at /lfs-cache so setup_r0_direct can symlink .git/lfs to it.
    lfs_mount = args.data_dir / "lfs" if args.dataset == "r0" and (args.data_dir / "lfs").exists() else None
    if not skip_setup:
        start_services(WORK_DIR, lfs_dir=lfs_mount)
        setup_containers()
    else:
        log_step("Skipping service setup (--skip-setup)")

    # S3 services are started on demand per scenario (see start_s3_service).
    # When --skip-setup is used, we still need to ensure garage credentials
    # are loaded if the services happen to be running already.
    if skip_setup:
        # Try to load credentials from already-running garage container.
        try:
            setup_shared_garage()
        except Exception:
            pass

    # ---- Create dataset ----
    if args.dataset == "small":
        source_repo = WORK_DIR / "source-repo"
        raw_lfs_bytes, source_lfs_count = create_synthetic_repo(source_repo)
        checksums = load_checksums(source_repo)
    else:
        # R0: fast clone from bare repo with LFS symlink (no expensive lfs fetch)
        source_repo = WORK_DIR / "data"
        raw_lfs_bytes, source_lfs_count = setup_r0_direct(args.data_dir, source_repo)
        checksums = load_checksums(source_repo)

    # Ordered push refspecs covering all local branches and tags (works for both datasets).
    push_refspecs = get_ordered_push_refspecs(source_repo)
    clone_extra_refspecs = None

    # Prepare the non-LFS source repo for scenario 0 (plain git baseline).
    source_repo_nolfs = None
    push_refspecs_nolfs = None
    checksums_nolfs = None
    git_packed_bytes = 0
    if 0 in scenarios_to_run:
        if args.dataset == "small":
            source_repo_nolfs = WORK_DIR / "source-repo-nolfs"
            create_nolfs_repo(source_repo, source_repo_nolfs)
        else:
            source_repo_nolfs = WORK_DIR / "data-nolfs"
            setup_r0_nolfs(args.data_dir0, source_repo_nolfs)
        checksums_nolfs = load_checksums(source_repo_nolfs)
        push_refspecs_nolfs = get_ordered_push_refspecs(source_repo_nolfs)
    # Measure the git-repacked size (full gc'd .git directory) for the header.
    if source_repo_nolfs and source_repo_nolfs.exists():
        git_dir = source_repo_nolfs / ".git"
        if git_dir.exists():
            r = run(["du", "-sk", str(git_dir)], capture=True)
            try:
                git_packed_bytes = int(r.stdout.strip().split()[0]) * 1024
            except (ValueError, IndexError):
                pass

    # ---- Stop containers before scenario loop ----
    # Setup and dataset creation are done; stop all containers so each
    # scenario starts from a clean state (stopped containers preserve their
    # state — SSH keys, git config, volumes are all intact).
    if not skip_setup:
        stop_all_containers()

    # ---- Run scenarios ----
    scenario_funcs = {
        0: run_scenario_0,
        1: run_scenario_1,
        2: run_scenario_2,
        3: run_scenario_3,
        4: run_scenario_4,
        5: run_scenario_5,
        6: run_scenario_6,
        7: run_scenario_7,
        8: run_scenario_8,
        9: run_scenario_9,
        10: run_scenario_10,
        11: run_scenario_11,
        12: run_scenario_12,
        13: run_scenario_13,
        14: run_scenario_14,
    }

    results = []

    def _save_results(results, xcompat=None):
        elapsed = time.perf_counter() - experiment_start
        output = {
            "raw_lfs_bytes": raw_lfs_bytes,
            "git_packed_bytes": git_packed_bytes,
            "source_lfs_count": source_lfs_count,
            "total_time_s": round(elapsed, 1),
            "scenarios": [
                {
                    "id": r.id, "name": r.name,
                    "push_time_s": round(r.push_time_s, 2),
                    "clone_time_s": round(r.clone_time_s, 2),
                    "raw_lfs_bytes": r.raw_lfs_bytes,
                    "storage_bytes": r.storage_bytes,
                    "storage_ratio": round(r.storage_ratio, 4),
                    "push_net_bytes": r.push_net_bytes,
                    "push_ssh_net_bytes": r.push_ssh_net_bytes,
                    "push_s3_net_bytes": r.push_s3_net_bytes,
                    "clone_net_bytes": r.clone_net_bytes,
                    "clone_ssh_net_bytes": r.clone_ssh_net_bytes,
                    "clone_s3_net_bytes": r.clone_s3_net_bytes,
                    "s3_container": r.s3_container,
                    "source_lfs_count": r.source_lfs_count,
                    "clone_lfs_count": r.clone_lfs_count,
                    "verified": r.verified, "error": r.error,
                    "resources": {
                        name: {
                            "avg_cpu_percent": round(s.avg_cpu_percent, 1),
                            "max_cpu_percent": round(s.max_cpu_percent, 1),
                            "avg_mem_bytes": s.avg_mem_bytes,
                            "max_mem_bytes": s.max_mem_bytes,
                            "max_anon_bytes": s.max_anon_bytes,
                            "max_cache_bytes": s.max_cache_bytes,
                            "mem_limit_bytes": s.mem_limit_bytes,
                            "sample_count": s.sample_count,
                        }
                        for name, s in r.resources.items()
                    } if r.resources else {},
                }
                for r in sorted(results, key=lambda x: x.id)
            ],
            "cross_compat": [
                {
                    "name": xr.name,
                    "clone_time_s": round(xr.clone_time_s, 2),
                    "verified": xr.verified, "error": xr.error,
                }
                for xr in (xcompat or [])
            ],
        }
        results_file.write_text(json.dumps(output, indent=2))

    def _save_markdown(results, xcompat=None, with_charts=False):
        elapsed = time.perf_counter() - experiment_start
        chart_prefix = str(EXPERIMENT_DIR / args.output)
        if with_charts:
            _generate_charts(results, raw_lfs_bytes, git_packed_bytes, chart_prefix)
        text = _build_results_text(results, xcompat or [], raw_lfs_bytes,
                                   source_lfs_count, git_packed_bytes,
                                   total_time_s=elapsed, chart_prefix=chart_prefix)
        results_md.write_text(text + "\n")

    # Cross-compat tests run inline (right after their parent scenario).
    xcompat = []
    xcompat_map = {
        3: run_cross_compat_3_client,
        4: run_cross_compat_4_client,
        5: run_cross_compat_5_server,
        6: run_cross_compat_6_server,
        9: run_cross_compat_9_client,
        10: run_cross_compat_10_http,
    }

    experiment_start = time.perf_counter()

    for sid in scenarios_to_run:
        if sid not in scenario_funcs:
            log_warn(f"Unknown scenario ID: {sid}, skipping")
            continue
        start_core_containers()
        log_step(f"Running scenario {sid}")
        fn = scenario_funcs[sid]
        if sid == 0 and source_repo_nolfs is not None:
            res = fn(source_repo_nolfs, checksums_nolfs, raw_lfs_bytes, WORK_DIR,
                     push_refspecs=push_refspecs_nolfs,
                     clone_extra_refspecs=None,
                     source_lfs_count=-1,
                     stop_after=args.stop_after)
        else:
            res = fn(source_repo, checksums, raw_lfs_bytes, WORK_DIR,
                     push_refspecs=push_refspecs,
                     clone_extra_refspecs=clone_extra_refspecs,
                     source_lfs_count=source_lfs_count,
                     stop_after=args.stop_after)
        results.append(res)
        _save_results(results, xcompat)
        _save_markdown(results, xcompat)
        log_ok(f"Scenario {sid} result saved")

        # Run cross-compat test immediately (S3 services still running)
        if not args.no_cross_compat and not args.stop_after and sid in xcompat_map:
            log(f"  Running cross-compat for scenario {sid}")
            xc = xcompat_map[sid](checksums, WORK_DIR, source_lfs_count)
            xcompat.append(xc)
            _save_results(results, xcompat)
            _save_markdown(results, xcompat)
            log_ok(f"Cross-compat {sid} done: verified={xc.verified}")
            # Clean up xcompat clone and cache
            client_exec(["rm", "-rf", f"/work/xcompat{sid}"], check=False, timeout=120)
            client_exec(["rm", "-rf", f"/work/.cache/xcompat{sid}"], check=False, timeout=120)

        # Clean up clone directory and desync cache to free disk space
        client_exec(["rm", "-rf", f"/work/clone{sid}"], check=False, timeout=120)
        client_exec(["rm", "-rf", f"/work/.cache/desync{sid}"], check=False, timeout=120)

        is_last = (sid == scenarios_to_run[-1])
        if not (is_last and args.keep_containers):
            stop_all_containers()

            # Wipe shared S3 volumes to free disk space for next scenario.
            _MINIO_SCENARIOS = {3, 7, 8, 11, 13, 14}
            _GARAGE_SCENARIOS = {4, 5, 6, 9, 10, 12}
            if sid in _MINIO_SCENARIOS:
                wipe_shared_minio()
            if sid in _GARAGE_SCENARIOS:
                wipe_shared_garage()

    # Save final results (including cross-compat) and print report.
    _save_results(results, xcompat)
    _save_markdown(results, xcompat, with_charts=True)

    # ---- Report ----
    print_results(results, xcompat, raw_lfs_bytes, source_lfs_count, git_packed_bytes)

    # ---- Cleanup ----
    if not args.keep_containers and not skip_setup:
        stop_services()
    else:
        log_step("Containers left running (--keep-containers or --skip-setup)")

    log(f"\nResults: {results_file}", CYAN)
    log(f"         {results_md}", CYAN)

    # Return exit code based on whether any scenario failed
    # When --stop-after is set, unverified results are expected (clone was skipped).
    failed = [r for r in results if r.error or (not r.verified and not args.stop_after)]
    if failed:
        log(f"\n{len(failed)} scenario(s) failed or could not be verified.", RED)
        return 1
    return 0


if __name__ == "__main__":
    sys.exit(main())