#!/usr/bin/env python3

"""Mount-vs-clone: time-to-work and bytes-to-work on a large repo.

The acquisition-model comparison marketing needs: `oak mount` (lazy

hydration) against credible git acquisition strategies on byte-identical

mirrors of the same large fixture (scripts/make_large_fixture.py):

- oak_mount:           oak mount ORG/REPO DEST

- git_clone_full:      git clone (full history, full tree)

- git_clone_shallow:   git clone --depth 1 (no history, full tree)

- git_clone_blobless:  git clone --filter=blob:none (lazy history blobs;

                       checkout still hydrates the full working tree)

- git_sparse_task:     blobless + --no-checkout + cone sparse-checkout of the

                       task path — git's strongest lazy mode, paid in extra

                       tool calls (recorded; that cost is part of the story)

Both mirrors are network remotes (oak.space vs github.com), so unlike the

core lane's local-file remote rows this is a cross-network comparison: reps

are interleaved per rep round, every row records the server, and published

medians must carry the different-network caveat. The claim this lane backs is

about the acquisition model (lazy vs eager hydration), which at this fixture

size dominates network variance — verify with the recorded spread before

publishing.

Setup:

  OAK_BENCH_MIRROR_REPO=oak/bench-large-mirror \\

  GIT_BENCH_MIRROR_URL=https://github.com/oakdotspace/bench-large-mirror.git \\

  python3 scripts/mount_vs_clone.py --reps 5

Unconfigured subjects emit skip rows (returncode 77), never silent absence.

"""

from __future__ import annotations

import argparse

import hashlib

import platform

import random

import shlex

import shutil

import statistics

import tempfile

from datetime import datetime, timezone

from pathlib import Path

from typing import Any

import sys

sys.path.insert(0, str(Path(__file__).resolve().parent))

from oakbench import cachectl as oakbench_cachectl

from oakbench import remotes as oakbench_remotes

from oakbench.diskprobe import bounded_tree_usage

from oakbench.environment import ENV_ISOLATION_VERSION, base_env

from oakbench.execution import run_timed

from oakbench.integrity import parse_manifest

from oakbench.reporting import fmt_num

from oakbench.results import ResultsStore

from oakbench.rows import row_returncode

from oakbench.runlock import measurement_lock

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

DEFAULT_WORKDIR = Path(tempfile.gettempdir()) / "oak-mount-vs-clone"

ADMITTED_OUTPUT_CHARS = 20_000

SKIP_RETURNCODE = 77

OAK_CACHE_STATE_UNCHECKED = "global_cache_not_purged"

UNKNOWN_CACHE_STATE = "unknown"

VARIANTS = ("oak_mount", "git_clone_full", "git_clone_shallow", "git_clone_blobless", "git_sparse_task")

FIRST_READ_TARGET = "README.md"

MANIFEST_TARGET = "MANIFEST.sha256"

def parse_args() -> argparse.Namespace:

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

    parser.add_argument("--variants", default=",".join(VARIANTS), help="Comma-separated acquisition variants.")

    parser.add_argument("--reps", type=int, default=5)

    parser.add_argument("--task-prefix", default="docs", help="Repo subtree a scoped task reads from.")

    parser.add_argument("--task-files", type=int, default=50, help="Files read by the scoped task.")

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

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

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

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

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

    return parser.parse_args()

def acquire_commands(

    variant: str, oak_bin: str, git_bin: str, oak_repo: str, git_url: str, dest: Path, task_prefix: str

) -> list[list[str]]:

    if variant == "oak_mount":

        return [[oak_bin, "mount", oak_repo, str(dest)]]

    if variant == "git_clone_full":

        return [[git_bin, "clone", "-q", git_url, str(dest)]]

    if variant == "git_clone_shallow":

        return [[git_bin, "clone", "-q", "--depth", "1", git_url, str(dest)]]

    if variant == "git_clone_blobless":

        return [[git_bin, "clone", "-q", "--filter=blob:none", git_url, str(dest)]]

    if variant == "git_sparse_task":

        return [

            [git_bin, "clone", "-q", "--filter=blob:none", "--no-checkout", git_url, str(dest)],

            [git_bin, "-C", str(dest), "sparse-checkout", "set", "--cone", task_prefix],

            [git_bin, "-C", str(dest), "checkout", "-q"],

        ]

    raise SystemExit(f"unknown variant {variant!r}")

def scoped_task_files(dest: Path, prefix: str, limit: int) -> list[str]:

    """Deterministic task file list: sorted walk of the prefix, names only.

    Directory listing does not hydrate file content on a lazy mount, so

    building this list does not pre-pay the bytes the task step measures.

    """

    base = dest / prefix

    if not base.is_dir():

        return []

    found = sorted(

        str(p.relative_to(dest)) for p in base.rglob("*") if p.is_file()

    )

    return found[:limit]

def read_manifest(dest: Path) -> tuple[dict[str, str], str | None]:

    manifest = dest / MANIFEST_TARGET

    try:

        return parse_manifest(manifest.read_text()), None

    except OSError as exc:

        return {}, str(exc)

def _safe_relative_path(rel: str) -> Path | None:

    candidate = Path(rel)

    if candidate.is_absolute() or ".." in candidate.parts:

        return None

    return candidate

def content_verification_fields(dest: Path, paths: list[str], manifest: dict[str, str]) -> dict[str, Any]:

    failures: list[dict[str, str]] = []

    verified = 0

    verified_bytes = 0

    actual_hashes: dict[str, str] = {}

    expected_hashes: dict[str, str] = {}

    missing_expected: list[str] = []

    for rel in paths:

        expected = manifest.get(rel)

        if not expected:

            missing_expected.append(rel)

            continue

        safe_rel = _safe_relative_path(rel)

        if safe_rel is None:

            failures.append({"path": rel, "reason": "unsafe_relative_path"})

            continue

        path = dest / safe_rel

        try:

            data = path.read_bytes()

        except OSError as exc:

            failures.append({"path": rel, "expected_sha256": expected, "reason": str(exc)})

            continue

        actual = hashlib.sha256(data).hexdigest()

        actual_hashes[rel] = actual

        expected_hashes[rel] = expected

        verified_bytes += len(data)

        if actual == expected:

            verified += 1

        else:

            failures.append({"path": rel, "expected_sha256": expected, "actual_sha256": actual})

    ok = bool(paths) and verified == len(paths) and not failures and not missing_expected

    status = "passed" if ok else "failed"

    if missing_expected:

        status = "missing_manifest_entries"

    return {

        "content_check_method": "manifest_sha256",

        "content_check_status": status,

        "content_verified": ok,

        "content_expected_files": len(paths),

        "content_verified_files": verified,

        "content_verified_bytes": verified_bytes,

        "content_expected_sha256": expected_hashes,

        "content_actual_sha256": actual_hashes,

        "content_missing_manifest_entries": missing_expected,

        "content_mismatches": failures,

        "checks": {"ok": ok, "content_verified": ok},

    }

def skipped_content_verification_fields(paths: list[str], reason: str) -> dict[str, Any]:

    return {

        "content_check_method": "manifest_sha256",

        "content_check_status": "skipped",

        "content_verified": False,

        "content_expected_files": len(paths),

        "content_verified_files": 0,

        "content_verified_bytes": 0,

        "content_skip_reason": reason,

        "checks": {"ok": False, "content_verified": False, "skip_reason": reason},

    }

def apply_content_verification(row: dict[str, Any], fields: dict[str, Any]) -> None:

    row.update(fields)

    if row.get("returncode") == 0 and not fields.get("content_verified"):

        row["returncode"] = 1

def disk_row_fields(dest: Path, variant: str) -> dict[str, Any]:

    total = bounded_tree_usage(dest)

    fields: dict[str, Any] = {

        "visible_tree_bytes": total["visible_tree_bytes"],

        "allocated_tree_bytes": total["allocated_tree_bytes"],

        "disk_entry_count": total["disk_entry_count"],

        "disk_usage_source": "stat_st_blocks_bounded_walk",

    }

    git_dir = dest / ".git"

    if git_dir.is_dir():

        store = bounded_tree_usage(git_dir)

        fields["vcs_store_allocated_bytes"] = store["allocated_tree_bytes"]

        fields["worktree_allocated_bytes"] = max(

            0, total["allocated_tree_bytes"] - store["allocated_tree_bytes"]

        )

    else:

        # Oak mount: per-dest store and any global chunk cache are not under

        # dest/.git; unmeasured here means null, never zero (ADR-0002).

        fields["vcs_store_allocated_bytes"] = None

        fields["worktree_allocated_bytes"] = total["allocated_tree_bytes"]

        fields["oak_global_cache_bytes"] = None

        fields["oak_global_cache_state"] = OAK_CACHE_STATE_UNCHECKED

    return fields

def cache_state_fields(variant: str, purge_result: dict[str, Any] | None = None) -> dict[str, Any]:

    fields = {

        "requested_cache_state": "cold",

        **oakbench_cachectl.cache_fields("cold", purge_result),

    }

    if variant == "oak_mount":

        # OS page cache purging is not enough to prove a cold Oak mount:

        # Oak's global chunk cache is outside this harness' control.

        if fields["cache_state"] == "cold":

            fields["cache_state"] = UNKNOWN_CACHE_STATE

            fields["cache_state_reason"] = (

                f"{fields['cache_state_reason']}; {OAK_CACHE_STATE_UNCHECKED}"

            )

        fields["oak_global_cache_state"] = OAK_CACHE_STATE_UNCHECKED

    return fields

def base_row(metadata: dict[str, Any], variant: str, kind: str, scenario: str, op: str, rep: int) -> dict[str, Any]:

    return {

        **metadata,

        "subject": variant,

        "subject_kind": kind,

        "scenario": scenario,

        "operation": op,

        "run": rep,

    }

def command_row(

    metadata: dict[str, Any],

    variant: str,

    kind: str,

    scenario: str,

    op: str,

    rep: int,

    command: list[str],

    cwd: Path,

    since_start_ms: float,

    expected_returncodes: tuple[int, ...] = (0,),

) -> dict[str, Any]:

    capture = run_timed(command, cwd, base_env(), ADMITTED_OUTPUT_CHARS)

    row = {

        **base_row(metadata, variant, kind, scenario, op, rep),

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

        "since_scenario_start_ms": round(since_start_ms + capture.elapsed_ms, 3),

        "returncode": capture.returncode,

        "command": command,

        "stdout_bytes": capture.stdout_bytes,

        "stderr_bytes": capture.stderr_bytes,

        "peak_rss_bytes": capture.peak_rss_bytes,

        "tool_call_count": 1,

    }

    if capture.returncode not in expected_returncodes:

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

    return row

def skip_row(metadata: dict[str, Any], variant: str, kind: str, scenario: str, rep: int, reason: str) -> dict[str, Any]:

    return {

        **base_row(metadata, variant, kind, scenario, "acquire.cmd", rep),

        "elapsed_ms": 0.0,

        "returncode": SKIP_RETURNCODE,

        "command": [],

        "skipped": True,

        "skip_reason": reason,

        "tool_call_count": 0,

    }

def teardown(variant: str, oak_bin: str, dest: Path) -> None:

    if variant == "oak_mount":

        run_timed([oak_bin, "mount", "end", str(dest), "-f"], dest.parent, base_env(), 2_000)

    shutil.rmtree(dest, ignore_errors=True)

def run_variant(

    args: argparse.Namespace,

    metadata: dict[str, Any],

    variant: str,

    rep: int,

    oak_remote: Any,

    git_remote: Any,

    run_root: Path,

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

    kind = "oak" if variant == "oak_mount" else "git"

    scenario = "acquire_cold"

    remote = oak_remote if kind == "oak" else git_remote

    if not remote.resolved:

        return [skip_row(metadata, variant, kind, scenario, rep, remote.skip_reason or "remote unresolved")]

    row_remote_fields = remote.row_fields()

    purge_result = oakbench_cachectl.purge_fs_caches()

    row_cache_fields = cache_state_fields(variant, purge_result)

    dest = run_root / f"{variant}-rep{rep}"

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

    shutil.rmtree(dest, ignore_errors=True)

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

    since = 0.0

    commands = acquire_commands(

        variant, args.oak_bin, args.git_bin, oak_remote.repo or "", git_remote.repo or "", dest, args.task_prefix

    )

    acquisition_failed = False

    for index, command in enumerate(commands):

        op = "acquire.cmd" if len(commands) == 1 else f"acquire.cmd_{index}"

        row = command_row(metadata, variant, kind, scenario, op, rep, command, dest.parent, since)

        row.update(row_remote_fields)

        row.update(row_cache_fields)

        rows.append(row)

        since = row["since_scenario_start_ms"]

        if row["returncode"] != 0:

            acquisition_failed = True

            break

    if not acquisition_failed:

        ready_row = command_row(

            metadata, variant, kind, scenario, "acquire.list_root", rep, ["/bin/ls", str(dest)], dest.parent, since

        )

        rows.append(ready_row)

        since = ready_row["since_scenario_start_ms"]

        time_to_task_ready_ms = since

        first_read = command_row(

            metadata, variant, kind, scenario, "acquire.first_read", rep,

            ["/bin/cat", FIRST_READ_TARGET], dest, since,

        )

        rows.append(first_read)

        since = first_read["since_scenario_start_ms"]

        time_to_first_read_ms = since

        manifest_read = command_row(

            metadata, variant, kind, scenario, "acquire.manifest_read", rep,

            ["/bin/cat", MANIFEST_TARGET], dest, since,

        )

        rows.append(manifest_read)

        since = manifest_read["since_scenario_start_ms"]

        manifest, manifest_error = read_manifest(dest)

        if manifest_error:

            manifest_read["content_check_method"] = "manifest_sha256"

            manifest_read["content_check_status"] = "manifest_unreadable"

            manifest_read["content_verified"] = False

            manifest_read["checks"] = {"ok": False, "error": manifest_error}

            if manifest_read["returncode"] == 0:

                manifest_read["returncode"] = 1

        else:

            manifest_read["content_check_method"] = "manifest_sha256"

            manifest_read["content_check_status"] = "manifest_loaded"

            manifest_read["content_manifest_entries"] = len(manifest)

            manifest_read["content_verified"] = True

            manifest_read["checks"] = {"ok": True, "manifest_entries": len(manifest)}

        if manifest:

            apply_content_verification(

                first_read,

                content_verification_fields(dest, [FIRST_READ_TARGET], manifest),

            )

        else:

            apply_content_verification(

                first_read,

                skipped_content_verification_fields(

                    [FIRST_READ_TARGET], manifest_error or "manifest_empty_or_unparseable"

                ),

            )

        post_acquire = {

            **base_row(metadata, variant, kind, scenario, "disk.usage.post_acquire", rep),

            "elapsed_ms": 0.0,

            "returncode": 0,

            "command": [],

            "tool_call_count": 0,

            **disk_row_fields(dest, variant),

        }

        rows.append(post_acquire)

        task_files = scoped_task_files(dest, args.task_prefix, args.task_files)

        if task_files:

            quoted = " ".join(shlex.quote(f) for f in task_files)

            task_row = command_row(

                metadata, variant, kind, scenario, "task.scoped_reads", rep,

                ["/bin/zsh", "-c", f"cat {quoted} > /dev/null"], dest, since,

            )

            task_row["task_prefix"] = args.task_prefix

            task_row["task_files_read"] = len(task_files)

            if manifest:

                apply_content_verification(

                    task_row,

                    content_verification_fields(dest, task_files, manifest),

                )

            else:

                apply_content_verification(

                    task_row,

                    skipped_content_verification_fields(

                        task_files, manifest_error or "manifest_empty_or_unparseable"

                    ),

                )

            rows.append(task_row)

            since = task_row["since_scenario_start_ms"]

        else:

            rows.append({

                **base_row(metadata, variant, kind, scenario, "task.scoped_reads", rep),

                "elapsed_ms": 0.0,

                "returncode": 1,

                "command": [],

                "tool_call_count": 0,

                "task_prefix": args.task_prefix,

                "task_files_read": 0,

                "notes": "task prefix missing from acquired tree",

                **skipped_content_verification_fields([], "task_prefix_missing"),

            })

        post_task = {

            **base_row(metadata, variant, kind, scenario, "disk.usage.post_task", rep),

            "elapsed_ms": 0.0,

            "returncode": 0,

            "command": [],

            "tool_call_count": 0,

            **disk_row_fields(dest, variant),

        }

        rows.append(post_task)

        rows.append({

            **base_row(metadata, variant, kind, scenario, "acquire.total", rep),

            **row_remote_fields,

            "elapsed_ms": round(since, 3),

            "returncode": 1 if any(r.get("returncode") not in (0,) for r in rows) else 0,

            "command": [r["command"] for r in rows if r.get("command")],

            "tool_call_count": sum(int(r.get("tool_call_count", 0)) for r in rows),

            "acquire_tool_calls": len(commands),

            "time_to_task_ready_ms": round(time_to_task_ready_ms, 3),

            "time_to_first_read_ms": round(time_to_first_read_ms, 3),

            "post_acquire_allocated_bytes": post_acquire["allocated_tree_bytes"],

            "post_task_allocated_bytes": post_task["allocated_tree_bytes"],

            "task_hydrated_delta_bytes": post_task["allocated_tree_bytes"] - post_acquire["allocated_tree_bytes"],

        })

    teardown(variant, args.oak_bin, dest)

    return rows

def summary_text(rows: list[dict[str, Any]], variants: list[str], task_prefix: str) -> str:

    def med(variant: str, field: str) -> float | None:

        values = [

            float(r[field]) for r in rows

            if r.get("subject") == variant and r.get("operation") == "acquire.total"

            and r.get(field) is not None and row_returncode(r) == 0

        ]

        return statistics.median(values) if values else None

    lines = [

        "# Mount vs Clone (acquire_cold)",

        "",

        "Different networks by construction (oak.space vs github.com): medians "

        "support acquisition-model claims, not transport benchmarking. Reps are "

        "interleaved; verify spread in the raw rows before publishing.",

        f"Scoped task: reads under `{task_prefix}/` (task_files_limit in row metadata).",

        "",

        "| Variant | n | Time to first read (ms) | Time task-ready (ms) | Acquire calls | Allocated after acquire (MB) | Task hydration delta (MB) |",

        "| --- | ---: | ---: | ---: | ---: | ---: | ---: |",

    ]

    for variant in variants:

        n = sum(

            1 for r in rows

            if r.get("subject") == variant and r.get("operation") == "acquire.total" and row_returncode(r) == 0

        )

        ttfr = med(variant, "time_to_first_read_ms")

        ready = med(variant, "time_to_task_ready_ms")

        calls = med(variant, "acquire_tool_calls")

        alloc = med(variant, "post_acquire_allocated_bytes")

        delta = med(variant, "task_hydrated_delta_bytes")

        mb = 1024 * 1024

        lines.append(

            f"| `{variant}` | {n} | {fmt_num(ttfr, 0)} | {fmt_num(ready, 0)} | {fmt_num(calls, 0)} | "

            f"{fmt_num(alloc / mb if alloc is not None else None, 1)} | "

            f"{fmt_num(delta / mb if delta is not None else None, 1)} |"

        )

    skips = [r for r in rows if r.get("returncode") == SKIP_RETURNCODE]

    if skips:

        lines += ["", f"Skipped rows: {len(skips)} (see skip_reason; skips are work items)."]

    return "\n".join(lines) + "\n"

def rows_contain_failures(rows: list[dict[str, Any]]) -> bool:

    return any(

        row.get("returncode") not in {0, SKIP_RETURNCODE}

        and not row.get("skipped")

        for row in rows

    )

def main() -> int:

    args = parse_args()

    variants = [v.strip() for v in args.variants.split(",") if v.strip()]

    unknown = [v for v in variants if v not in VARIANTS]

    if unknown:

        raise SystemExit(f"unknown variants: {', '.join(unknown)}")

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

    run_root = args.workdir / "runs" / timestamp

    oak_remote = oakbench_remotes.resolve_oak_remote("mirror")

    git_remote = oakbench_remotes.resolve_git_github_remote(oakbench_remotes.GIT_MIRROR_ENV)

    metadata = {

        "bench_id": timestamp,

        "profile": "mount-vs-clone",

        "timestamp_utc": timestamp,

        "host": platform.node(),

        "platform": platform.platform(),

        "machine": platform.machine(),

        "python": platform.python_version(),

        "env_isolation_version": ENV_ISOLATION_VERSION,

        "task_prefix": args.task_prefix,

        "task_files_limit": args.task_files,

    }

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

    with measurement_lock("mount_vs_clone") as lock_info:

        metadata["measurement_lock_wait_ms"] = lock_info.wait_ms

        metadata["measurement_lock"] = "held" if lock_info.enabled else "disabled"

        for rep in range(args.reps):

            order = list(variants)

            random.Random(f"{timestamp}:{rep}").shuffle(order)

            for variant in order:

                print(f"[run] rep={rep} variant={variant}", flush=True)

                rows.extend(run_variant(args, metadata, variant, rep, oak_remote, git_remote, run_root))

    store = ResultsStore(args.results, lane="mount", filename_suffix="acquire")

    raw_path, summary_path = store.write(timestamp, rows, summary_text(rows, variants, args.task_prefix))

    if not args.keep_workdirs:

        shutil.rmtree(run_root, ignore_errors=True)

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

    print(f"[summary] {summary_path}")

    return 1 if rows_contain_failures(rows) else 0

if __name__ == "__main__":

    raise SystemExit(main())