#!/usr/bin/env python3

"""Probe Oak mount startup and lazy-hydration behavior.

The default run is intentionally safe: if `oak mount` is unavailable or no

mount repo is configured, the script writes capability and skip rows instead of

failing. Real mount work starts only when the spec or environment provides a

throwaway OWNER/REPO.

"""

from __future__ import annotations

import argparse

import concurrent.futures

import hashlib

import json

import os

import platform

import re

import shutil

import subprocess

import sys

import tempfile

import time

from datetime import datetime, timezone

from pathlib import Path

from typing import Any, Callable, Optional, Tuple

from oakbench import cachectl as oakbench_cachectl

from oakbench import diskprobe as oakbench_diskprobe

from oakbench.environment import ENV_ISOLATION_VERSION

from oakbench.integrity import parse_manifest

from oakbench.runlock import measurement_lock

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

DEFAULT_SPEC = ROOT / "scenarios" / "mount.yaml"

DEFAULT_DEST = Path(tempfile.gettempdir()) / "oak-mount-probe"

SKIP_RETURNCODE = 77

MANIFEST_TARGET = "MANIFEST.sha256"

OAK_CACHE_STATE_NOT_PURGED = "oak_global_cache_not_purged"

UNKNOWN_CACHE_STATE = "unknown"

UNSET_CACHE_STATE = "unset"

CACHE_STATE_UNSET = "cache_state_unset"

EXPLICIT_CACHE_STATE = "cache_state_declared"

class SpecError(ValueError):

    pass

def parse_args() -> argparse.Namespace:

    parser = argparse.ArgumentParser(description=__doc__)

    parser.add_argument("--spec", type=Path, default=DEFAULT_SPEC)

    parser.add_argument("--dest", type=Path, default=DEFAULT_DEST)

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

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

    parser.add_argument("--keep", action="store_true", help="Keep probe workdirs and mounts for inspection")

    return parser.parse_args()

def strip_comment(line: str) -> str:

    quote: Optional[str] = None

    escaped = False

    for index, char in enumerate(line):

        if escaped:

            escaped = False

            continue

        if char == "\\":

            escaped = True

            continue

        if quote:

            if char == quote:

                quote = None

            continue

        if char in {"'", '"'}:

            quote = char

            continue

        if char == "#" and (index == 0 or line[index - 1].isspace()):

            return line[:index]

    return line

def parse_scalar(value: str) -> Any:

    raw = value.strip()

    if raw in {"null", "Null", "NULL", "~"}:

        return None

    if raw in {"true", "True", "TRUE"}:

        return True

    if raw in {"false", "False", "FALSE"}:

        return False

    if raw.startswith('"') and raw.endswith('"'):

        try:

            return json.loads(raw)

        except json.JSONDecodeError:

            return raw[1:-1]

    if raw.startswith("'") and raw.endswith("'"):

        return raw[1:-1]

    if re.fullmatch(r"-?\d+", raw):

        return int(raw)

    if re.fullmatch(r"-?\d+\.\d+", raw):

        return float(raw)

    return raw

def split_key_value(text: str) -> Tuple[str, str]:

    if ":" not in text:

        raise SpecError(f"expected key: value line, got {text!r}")

    key, value = text.split(":", 1)

    key = key.strip()

    if not key:

        raise SpecError(f"empty key in line {text!r}")

    return key, value.strip()

def looks_like_inline_mapping(text: str) -> bool:

    if ":" not in text:

        return False

    key = text.split(":", 1)[0].strip()

    return bool(re.fullmatch(r"[A-Za-z0-9_.-]+", key))

BLOCK_SCALAR_MARKERS = {">", ">-", ">+", "|", "|-", "|+"}

def _block_scalar_line(text: str) -> tuple[str, str] | None:

    prefix = ""

    body = text

    if body.startswith("- "):

        prefix = "- "

        body = body[2:].strip()

    if ":" not in body:

        return None

    key, value = body.split(":", 1)

    marker = value.strip()

    if marker not in BLOCK_SCALAR_MARKERS:

        return None

    return prefix + key.strip(), marker

def yaml_lines(text: str) -> list[Tuple[int, str]]:

    rows: list[Tuple[int, str]] = []

    raw_lines = text.splitlines()

    index = 0

    while index < len(raw_lines):

        line_no = index + 1

        raw = raw_lines[index]

        without_comment = strip_comment(raw).rstrip()

        if not without_comment.strip():

            index += 1

            continue

        indent_text = without_comment[: len(without_comment) - len(without_comment.lstrip(" "))]

        if "\t" in indent_text:

            raise SpecError(f"tabs are not supported for indentation on line {line_no}")

        indent = len(indent_text)

        stripped = without_comment.strip()

        block = _block_scalar_line(stripped)

        if block is not None:

            key_text, marker = block

            index += 1

            block_lines: list[tuple[int, str]] = []

            while index < len(raw_lines):

                child_raw = raw_lines[index]

                child_without_comment = child_raw.rstrip()

                if not child_without_comment.strip():

                    block_lines.append((indent + 1, ""))

                    index += 1

                    continue

                child_indent_text = child_without_comment[

                    : len(child_without_comment) - len(child_without_comment.lstrip(" "))

                ]

                if "\t" in child_indent_text:

                    raise SpecError(f"tabs are not supported for indentation on line {index + 1}")

                child_indent = len(child_indent_text)

                if child_indent <= indent:

                    break

                block_lines.append((child_indent, child_without_comment))

                index += 1

            content_indent = min((child_indent for child_indent, line in block_lines if line.strip()), default=indent + 1)

            normalized = [

                line[content_indent:] if len(line) >= content_indent else ""

                for _, line in block_lines

            ]

            if marker.startswith("|"):

                block_value = "\n".join(normalized)

            else:

                block_value = " ".join(line.strip() for line in normalized if line.strip())

            if marker.endswith("-"):

                block_value = block_value.rstrip("\n")

            elif marker.endswith("+"):

                pass

            elif marker.startswith("|") and block_lines:

                block_value += "\n"

            rows.append((indent, f"{key_text}: {json.dumps(block_value)}"))

            continue

        rows.append((indent, stripped))

        index += 1

    return rows

def parse_block(rows: list[Tuple[int, str]], index: int, indent: int) -> Tuple[Any, int]:

    if index >= len(rows):

        return {}, index

    current_indent, current_text = rows[index]

    if current_indent < indent:

        return {}, index

    if current_text.startswith("- "):

        return parse_list(rows, index, current_indent)

    return parse_map(rows, index, current_indent)

def parse_map(rows: list[Tuple[int, str]], index: int, indent: int) -> Tuple[dict[str, Any], int]:

    result: dict[str, Any] = {}

    while index < len(rows):

        current_indent, text = rows[index]

        if current_indent < indent:

            break

        if current_indent > indent:

            raise SpecError(f"unexpected indentation before {text!r}")

        if text.startswith("- "):

            break

        key, value = split_key_value(text)

        index += 1

        if value:

            result[key] = parse_scalar(value)

            continue

        if index < len(rows) and rows[index][0] > current_indent:

            child, index = parse_block(rows, index, rows[index][0])

            result[key] = child

        else:

            result[key] = {}

    return result, index

def parse_list(rows: list[Tuple[int, str]], index: int, indent: int) -> Tuple[list[Any], int]:

    result: list[Any] = []

    while index < len(rows):

        current_indent, text = rows[index]

        if current_indent < indent:

            break

        if current_indent > indent:

            raise SpecError(f"unexpected indentation before {text!r}")

        if not text.startswith("- "):

            break

        item_text = text[2:].strip()

        index += 1

        if not item_text:

            if index < len(rows) and rows[index][0] > current_indent:

                item, index = parse_block(rows, index, rows[index][0])

            else:

                item = None

            result.append(item)

            continue

        if looks_like_inline_mapping(item_text):

            key, value = split_key_value(item_text)

            item_dict: dict[str, Any] = {}

            if value:

                item_dict[key] = parse_scalar(value)

            elif index < len(rows) and rows[index][0] > current_indent:

                child, index = parse_block(rows, index, rows[index][0])

                item_dict[key] = child

            else:

                item_dict[key] = {}

            if index < len(rows) and rows[index][0] > current_indent:

                extra, index = parse_block(rows, index, rows[index][0])

                if isinstance(extra, dict):

                    item_dict.update(extra)

                else:

                    item_dict["_items"] = extra

            result.append(item_dict)

        else:

            result.append(parse_scalar(item_text))

    return result, index

def parse_yaml_subset(text: str) -> dict[str, Any]:

    stripped = text.lstrip()

    if stripped.startswith("{"):

        data = json.loads(text)

        if not isinstance(data, dict):

            raise SpecError("JSON spec root must be an object")

        return data

    rows = yaml_lines(text)

    if not rows:

        return {}

    parsed, index = parse_block(rows, 0, rows[0][0])

    if index != len(rows):

        raise SpecError(f"unparsed content starting at {rows[index][1]!r}")

    if not isinstance(parsed, dict):

        raise SpecError("spec root must be a mapping")

    return parsed

def default_spec(reason: str) -> dict[str, Any]:

    return {

        "version": 1,

        "suite": "oak_mount_lazy_hydration",

        "spec_missing": reason,

        "remote": {"owner_repo": "", "owner_repo_env": "OAK_BENCH_MOUNT_REPO", "safe_push": False},

        "defaults": {

            "runs": 1,

            "timeout_seconds": 120,

            "retries": 0,

            "target_read": "README.md",

            "target_read_bytes": 4096,

            "target_read_contains": "# oak-benchmarks-tmp",

            "content_manifest": MANIFEST_TARGET,

            "require_content_manifest": False,

            "edit_path": ".oak-bench/mount-edit.txt",

            "huge_file_path": "",

            "huge_file_read_bytes": 4096,

            "parallelism": 4,

            "max_disk_entries": 100000,

            "sparse_paths": ["README.md"],

        },

        "scenarios": [

            {"name": "capability_probe", "requires_remote": False, "operations": ["oak.version", "oak.mount.help"]},

            {"name": "cold_mount_startup", "requires_remote": True, "operations": ["mount.start"]},

        ],

    }

def load_spec(path: Path) -> dict[str, Any]:

    if not path.exists():

        return default_spec(f"missing spec: {path}")

    try:

        return parse_yaml_subset(path.read_text())

    except Exception as exc:  # noqa: BLE001 - preserve graceful benchmark behavior

        return default_spec(f"could not parse spec {path}: {exc}")

def resolve_oak_bin(value: str) -> Optional[str]:

    expanded = os.path.expanduser(value)

    if os.sep in expanded or expanded.startswith("."):

        path = Path(expanded)

        return str(path) if path.exists() else None

    found = shutil.which(expanded)

    return found

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

    env = os.environ.copy()

    env.update(

        {

            "OAK_AUTHOR": "oakbench",

            "NO_COLOR": "1",

            "CLICOLOR": "0",

        }

    )

    return env

def run_process(

    command: list[str],

    cwd: Optional[Path],

    timeout: int,

    retries: int,

    capture_stdout: bool = True,

) -> dict[str, Any]:

    attempts = 0

    total_elapsed_ms = 0.0

    last: dict[str, Any] = {}

    while True:

        attempts += 1

        start = time.perf_counter()

        try:

            proc = subprocess.run(

                command,

                cwd=str(cwd) if cwd else None,

                env=base_env(),

                stdout=subprocess.PIPE if capture_stdout else subprocess.DEVNULL,

                stderr=subprocess.PIPE,

                text=True,

                timeout=timeout,

                check=False,

            )

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

            stdout = proc.stdout or ""

            stderr = proc.stderr or ""

            last = {

                "elapsed_ms": elapsed_ms,

                "returncode": proc.returncode,

                "stdout": stdout,

                "stderr": stderr,

            }

        except FileNotFoundError as exc:

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

            last = {"elapsed_ms": elapsed_ms, "returncode": 127, "stdout": "", "stderr": str(exc)}

        except subprocess.TimeoutExpired as exc:

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

            stdout = exc.stdout if isinstance(exc.stdout, str) else ""

            stderr = exc.stderr if isinstance(exc.stderr, str) else ""

            last = {

                "elapsed_ms": elapsed_ms,

                "returncode": 124,

                "stdout": stdout,

                "stderr": stderr or f"timed out after {timeout}s",

            }

        total_elapsed_ms += float(last["elapsed_ms"])

        if last["returncode"] == 0 or attempts >= retries + 1:

            break

    last["elapsed_ms"] = total_elapsed_ms

    last["attempts"] = attempts

    last["retry_count"] = attempts - 1

    return last

def parse_observed_bytes(text: str) -> dict[str, int]:

    found: dict[str, int] = {}

    for label in ("hydrated", "downloaded"):

        pattern = rf"{label}[^\d]([0-9][0-9_,]*)\s*(bytes|b|kb|kib|mb|mib|gb|gib)?"

        match = re.search(pattern, text, flags=re.IGNORECASE)

        if not match:

            continue

        value = int(match.group(1).replace("_", "").replace(",", ""))

        unit = (match.group(2) or "bytes").lower()

        scale = {

            "bytes": 1,

            "b": 1,

            "kb": 1000,

            "mb": 1000 * 1000,

            "gb": 1000 * 1000 * 1000,

            "kib": 1024,

            "mib": 1024 * 1024,

            "gib": 1024 * 1024 * 1024,

        }[unit]

        found[f"bytes_{label}"] = value * scale

    return found

def tail(text: str, limit: int = 4000) -> str:

    return text[-limit:] if len(text) > limit else text

def metadata(timestamp: str, spec_path: Path, spec: dict[str, Any], oak_bin: Optional[str]) -> dict[str, Any]:

    return {

        "bench_id": timestamp,

        "profile": "mount",

        "timestamp_utc": timestamp,

        "host": platform.node(),

        "platform": platform.platform(),

        "machine": platform.machine(),

        "python": platform.python_version(),

        "env_isolation_version": ENV_ISOLATION_VERSION,

        "subject": "oak_mount",

        "subject_kind": "oak",

        "subject_label": "Oak mount",

        "spec": {

            "path": str(spec_path),

            "suite": spec.get("suite"),

            "version": spec.get("version"),

            "spec_missing": spec.get("spec_missing"),

        },

        "oak_bin": oak_bin,

    }

def command_row(

    meta: dict[str, Any],

    scenario: str,

    operation: str,

    run_index: int,

    command: list[str],

    result: dict[str, Any],

    scenario_start: Optional[float] = None,

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

) -> dict[str, Any]:

    stdout = result.get("stdout", "")

    stderr = result.get("stderr", "")

    row = {

        **meta,

        "scenario": scenario,

        "run": run_index,

        "operation": operation,

        "elapsed_ms": round(float(result.get("elapsed_ms", 0.0)), 3),

        "returncode": int(result.get("returncode", 0)),

        "command": command,

        "attempts": int(result.get("attempts", 1)),

        "retry_count": int(result.get("retry_count", 0)),

        "stdout_bytes": len(stdout.encode("utf-8", "replace")),

        "stderr_bytes": len(stderr.encode("utf-8", "replace")),

    }

    if scenario_start is not None:

        row["since_scenario_start_ms"] = round((time.perf_counter() - scenario_start) * 1000, 3)

    if stderr and row["returncode"] != 0:

        row["stderr"] = tail(stderr)

    if stdout and operation in {"oak.version", "oak.mount.help"}:

        row["stdout"] = tail(stdout, 1200)

    row.update(parse_observed_bytes(stdout + "\n" + stderr))

    if extra:

        row.update(extra)

    return row

def skip_row(

    meta: dict[str, Any],

    scenario: str,

    operation: str,

    run_index: int,

    reason: str,

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

) -> dict[str, Any]:

    row = {

        **meta,

        "scenario": scenario,

        "run": run_index,

        "operation": operation,

        "elapsed_ms": 0.0,

        "returncode": SKIP_RETURNCODE,

        "command": [],

        "skipped": True,

        "skip_reason": reason,

        "attempts": 0,

        "retry_count": 0,

    }

    if extra:

        row.update(extra)

    return row

def fs_row(

    meta: dict[str, Any],

    scenario: str,

    operation: str,

    run_index: int,

    command: list[str],

    scenario_start: float,

    func: Callable[[], dict[str, Any]],

    first_useful: bool,

) -> dict[str, Any]:

    start = time.perf_counter()

    try:

        extra = func()

        returncode = 0 if extra.get("checks", {}).get("ok", True) else 1

        stderr = ""

    except Exception as exc:  # noqa: BLE001 - turn filesystem failures into rows

        extra = {"checks": {"ok": False, "error": str(exc)}}

        returncode = 1

        stderr = str(exc)

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

    result = {

        "elapsed_ms": elapsed_ms,

        "returncode": returncode,

        "stdout": "",

        "stderr": stderr,

        "attempts": 1,

        "retry_count": 0,

    }

    row = command_row(meta, scenario, operation, run_index, command, result, scenario_start, extra)

    if first_useful:

        row["time_to_first_useful_work_ms"] = row.get("since_scenario_start_ms", row["elapsed_ms"])

    return row

def bounded_tree_usage(root: Path, max_entries: int) -> dict[str, Any]:

    # Shared measurement policy (oakbench.diskprobe); re-exported here so

    # existing callers and tests keep their entry point.

    return oakbench_diskprobe.bounded_tree_usage(root, max_entries)

def spec_defaults(spec: dict[str, Any]) -> dict[str, Any]:

    defaults = spec.get("defaults") or {}

    if not isinstance(defaults, dict):

        return {}

    return defaults

def int_default(defaults: dict[str, Any], key: str, fallback: int) -> int:

    value = defaults.get(key, fallback)

    try:

        return int(value)

    except (TypeError, ValueError):

        return fallback

def str_default(defaults: dict[str, Any], key: str, fallback: str = "") -> str:

    value = defaults.get(key, fallback)

    return fallback if value is None else str(value)

def list_default(defaults: dict[str, Any], key: str) -> list[str]:

    value = defaults.get(key, [])

    if isinstance(value, list):

        return [str(item) for item in value]

    if isinstance(value, str) and value:

        return [value]

    return []

def bool_default(defaults: dict[str, Any], key: str, fallback: bool = False) -> bool:

    value = defaults.get(key, fallback)

    if isinstance(value, str):

        return value.strip().lower() in {"1", "true", "yes", "on"}

    return bool(value)

def manifest_expected_hash(root: Path, manifest_rel: str, target_rel: str) -> tuple[str | None, str | None]:

    manifest_path = root / manifest_rel

    try:

        manifest = parse_manifest(manifest_path.read_text())

    except OSError as exc:

        return None, str(exc)

    expected = manifest.get(target_rel)

    if not expected:

        return None, f"{target_rel} missing from {manifest_rel}"

    return expected, None

def content_check_fields(

    root: Path,

    target_rel: str,

    data: bytes,

    defaults: dict[str, Any],

) -> dict[str, Any]:

    expected_contains = defaults.get("target_read_contains")

    expected_sha = str_default(defaults, "target_read_sha256", "")

    manifest_rel = str_default(defaults, "content_manifest", MANIFEST_TARGET)

    require_manifest = bool_default(defaults, "require_content_manifest", False)

    if expected_contains is not None:

        needle = str(expected_contains).encode("utf-8")

        ok = needle in data

        return {

            "content_check_method": "contains",

            "content_check_status": "passed" if ok else "failed",

            "content_verified": ok,

            "content_expected_contains": str(expected_contains),

            "content_actual_sha256": hashlib.sha256(data).hexdigest(),

        }

    if expected_sha:

        actual = hashlib.sha256((root / target_rel).read_bytes()).hexdigest()

        ok = actual == expected_sha.lower()

        return {

            "content_check_method": "sha256",

            "content_check_status": "passed" if ok else "failed",

            "content_verified": ok,

            "content_expected_sha256": expected_sha.lower(),

            "content_actual_sha256": actual,

        }

    expected_from_manifest, manifest_error = manifest_expected_hash(root, manifest_rel, target_rel)

    if expected_from_manifest:

        actual = hashlib.sha256((root / target_rel).read_bytes()).hexdigest()

        ok = actual == expected_from_manifest

        return {

            "content_check_method": "manifest_sha256",

            "content_check_status": "passed" if ok else "failed",

            "content_verified": ok,

            "content_manifest": manifest_rel,

            "content_expected_sha256": expected_from_manifest,

            "content_actual_sha256": actual,

        }

    if require_manifest:

        return {

            "content_check_method": "manifest_sha256",

            "content_check_status": "failed",

            "content_verified": False,

            "content_manifest": manifest_rel,

            "content_error": manifest_error or "manifest_missing_or_unparseable",

        }

    return {

        "content_check_method": "none",

        "content_check_status": "skipped_unconfigured",

        "content_verified": False,

    }

def mount_readiness_probe(path: Path, defaults: dict[str, Any]) -> dict[str, Any]:

    start = time.perf_counter()

    target = str_default(defaults, "target_read", "README.md")

    read_bytes = int_default(defaults, "target_read_bytes", 4096)

    fields: dict[str, Any] = {

        "mount_ready_path": str(path),

        "mount_ready_target": target,

    }

    try:

        entries = sorted(os.listdir(path))

        fields["mount_ready_list_ok"] = True

        fields["mount_ready_entry_count"] = len(entries)

    except OSError as exc:

        fields["mount_ready_list_ok"] = False

        fields["mount_ready_error"] = str(exc)

        fields["mount_ready_probe_elapsed_ms"] = round((time.perf_counter() - start) * 1000, 3)

        fields["checks"] = {"ok": False, "ready_to_read": False, "error": str(exc)}

        return fields

    try:

        with (path / target).open("rb") as fh:

            data = fh.read(read_bytes)

        fields["mount_ready_read_ok"] = True

        fields["mount_ready_read_bytes"] = len(data)

    except OSError as exc:

        fields["mount_ready_read_ok"] = False

        fields["mount_ready_error"] = str(exc)

        data = b""

    ready = bool(fields.get("mount_ready_list_ok") and fields.get("mount_ready_read_ok"))

    fields["mount_ready_probe_elapsed_ms"] = round((time.perf_counter() - start) * 1000, 3)

    fields["checks"] = {"ok": ready, "ready_to_read": ready}

    return fields

def scenario_cache_fields(

    scenario: dict[str, Any],

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

) -> dict[str, Any]:

    requested = scenario.get("cache_state")

    if requested == "cold":

        return {

            "requested_cache_state": "cold",

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

        }

    if not requested:

        return {

            "requested_cache_state": UNSET_CACHE_STATE,

            "cache_state": UNKNOWN_CACHE_STATE,

            "cache_state_reason": scenario.get("cache_state_reason") or CACHE_STATE_UNSET,

        }

    return {

        "requested_cache_state": requested,

        "cache_state": requested,

        "cache_state_reason": scenario.get("cache_state_reason") or EXPLICIT_CACHE_STATE,

    }

def apply_scenario_cache_state(

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

    scenario: dict[str, Any],

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

) -> None:

    fields = scenario_cache_fields(scenario, purge_result)

    for row in rows:

        row.update(fields)

_SUPPORT_CACHE: dict[Tuple[str, str], bool] = {}

def oak_supports(oak_bin: str, subcommand: str) -> bool:

    """Untimed capability probe: does this binary know the subcommand?

    Keeps planned lifecycle operations (e.g. `oak finish`) as explicit skip

    rows on older binaries instead of guaranteed-failure rows, and makes them

    light up automatically once the binary ships the command.

    """

    key = (oak_bin, subcommand)

    if key not in _SUPPORT_CACHE:

        try:

            probe = subprocess.run(

                [oak_bin, subcommand, "--help"],

                capture_output=True,

                text=True,

                timeout=15,

                check=False,

            )

            _SUPPORT_CACHE[key] = probe.returncode == 0

        except (OSError, subprocess.TimeoutExpired):

            _SUPPORT_CACHE[key] = False

    return _SUPPORT_CACHE[key]

def mount_repo(spec: dict[str, Any]) -> str:

    remote = spec.get("remote") or {}

    if not isinstance(remote, dict):

        remote = {}

    configured = str(remote.get("owner_repo") or "").strip()

    env_name = str(remote.get("owner_repo_env") or "OAK_BENCH_MOUNT_REPO")

    return configured or os.environ.get(env_name, "").strip()

def safe_push_enabled(spec: dict[str, Any]) -> bool:

    remote = spec.get("remote") or {}

    return bool(isinstance(remote, dict) and remote.get("safe_push"))

def scenario_runs(defaults: dict[str, Any]) -> range:

    return range(max(1, int_default(defaults, "runs", 1)))

def mount_path(run_root: Path, scenario_name: str, run_index: int) -> Path:

    safe = re.sub(r"[^A-Za-z0-9_.-]+", "-", scenario_name).strip("-") or "mount"

    return run_root / f"{safe}-run-{run_index}"

def finish_desc_file_path(run_root: Path, scenario_name: str, run_index: int) -> Path:

    safe = re.sub(r"[^A-Za-z0-9_.-]+", "-", scenario_name).strip("-") or "mount"

    return run_root / f"{safe}-run-{run_index}.finish-desc.md"

def run_timed_command(

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

    meta: dict[str, Any],

    scenario_name: str,

    operation: str,

    run_index: int,

    command: list[str],

    cwd: Optional[Path],

    defaults: dict[str, Any],

    scenario_start: float,

    capture_stdout: bool = True,

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

) -> dict[str, Any]:

    result = run_process(

        command,

        cwd=cwd,

        timeout=int_default(defaults, "timeout_seconds", 120),

        retries=int_default(defaults, "retries", 0),

        capture_stdout=capture_stdout,

    )

    row = command_row(meta, scenario_name, operation, run_index, command, result, scenario_start, extra)

    rows.append(row)

    return row

def run_operation_sequence(

    meta: dict[str, Any],

    oak_bin: str,

    owner_repo: str,

    allow_push: bool,

    scenario: dict[str, Any],

    defaults: dict[str, Any],

    run_root: Path,

    run_index: int,

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

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

    cleanup_safe = True

    scenario_name = str(scenario.get("name", "mount"))

    path = mount_path(run_root, scenario_name, run_index)

    operations = scenario.get("operations") or []

    mounted = False

    first_useful_recorded = False

    scenario_start = time.perf_counter()

    purge_result = oakbench_cachectl.purge_fs_caches() if scenario.get("cache_state") == "cold" else None

    for operation in operations:

        operation = str(operation)

        if operation == "mount.start":

            command = [oak_bin, "mount", owner_repo, str(path)]

            row = run_timed_command(rows, meta, scenario_name, operation, run_index, command, None, defaults, scenario_start)

            command_return_since_start = row.get("since_scenario_start_ms")

            row["mount_command_returncode"] = row["returncode"]

            row["mount_command_elapsed_ms"] = row["elapsed_ms"]

            row["time_to_mount_command_return_ms"] = command_return_since_start

            if row["returncode"] == 0:

                readiness = mount_readiness_probe(path, defaults)

                row.update(readiness)

                row["since_scenario_start_ms"] = round((time.perf_counter() - scenario_start) * 1000, 3)

                row["time_to_mount_ready_to_read_ms"] = row["since_scenario_start_ms"]

                if not readiness.get("checks", {}).get("ok"):

                    row["returncode"] = 1

            else:

                row["mount_ready_probe_elapsed_ms"] = 0.0

                row["mount_ready_read_ok"] = False

                row["mount_ready_list_ok"] = False

                row["time_to_mount_ready_to_read_ms"] = None

                row["checks"] = {"ok": False, "ready_to_read": False, "command_returncode": row["returncode"]}

            mounted = row["returncode"] == 0

            cleanup_safe = cleanup_safe and mounted

            continue

        if operation == "mount.teardown":

            if not mounted:

                rows.append(skip_row(meta, scenario_name, operation, run_index, "mount_not_started"))

                cleanup_safe = False

                continue

            command = [oak_bin, "mount", "end", str(path)]

            row = run_timed_command(rows, meta, scenario_name, operation, run_index, command, None, defaults, scenario_start)

            released = not path.exists()

            row["checks"] = {"ok": row["returncode"] == 0 and released, "path_released": released}

            mounted = False

            cleanup_safe = cleanup_safe and row["returncode"] == 0

            continue

        if operation == "oak.status":

            if not mounted:

                rows.append(skip_row(meta, scenario_name, operation, run_index, "mount_not_started"))

                continue

            run_timed_command(rows, meta, scenario_name, operation, run_index, [oak_bin, "status"], path, defaults, scenario_start)

            continue

        if operation == "oak.diff":

            if not mounted:

                rows.append(skip_row(meta, scenario_name, operation, run_index, "mount_not_started"))

                continue

            run_timed_command(rows, meta, scenario_name, operation, run_index, [oak_bin, "diff"], path, defaults, scenario_start)

            continue

        if operation == "oak.commit":

            if not mounted:

                rows.append(skip_row(meta, scenario_name, operation, run_index, "mount_not_started"))

                continue

            run_timed_command(

                rows,

                meta,

                scenario_name,

                operation,

                run_index,

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

                path,

                defaults,

                scenario_start,

            )

            continue

        if operation == "oak.push":

            if not mounted:

                rows.append(skip_row(meta, scenario_name, operation, run_index, "mount_not_started"))

                continue

            if not allow_push:

                rows.append(skip_row(meta, scenario_name, operation, run_index, "safe_push_disabled"))

                continue

            run_timed_command(rows, meta, scenario_name, operation, run_index, [oak_bin, "push"], path, defaults, scenario_start)

            continue

        if operation == "fs.first_ls":

            if not mounted:

                rows.append(skip_row(meta, scenario_name, operation, run_index, "mount_not_started"))

                continue

            def first_ls() -> dict[str, Any]:

                entries = sorted(os.listdir(path))

                return {"entry_count": len(entries), "sample_entries": entries[:10], "checks": {"ok": True, "exists": path.exists()}}

            rows.append(

                fs_row(

                    meta,

                    scenario_name,

                    operation,

                    run_index,

                    ["python:os.listdir", "."],

                    scenario_start,

                    first_ls,

                    not first_useful_recorded,

                )

            )

            first_useful_recorded = True

            continue

        if operation == "fs.targeted_read":

            if not mounted:

                rows.append(skip_row(meta, scenario_name, operation, run_index, "mount_not_started"))

                continue

            target = str_default(defaults, "target_read", "README.md")

            read_bytes = int_default(defaults, "target_read_bytes", 4096)

            expected = defaults.get("target_read_contains")

            def targeted_read() -> dict[str, Any]:

                target_path = path / target

                with target_path.open("rb") as fh:

                    data = fh.read(read_bytes)

                checks = {"ok": target_path.exists(), "exists": target_path.exists(), "read_bytes": len(data)}

                content_fields = content_check_fields(path, target, data, defaults)

                checks["content_verified"] = bool(content_fields.get("content_verified"))

                checks["ok"] = checks["ok"] and checks["content_verified"]

                if expected is not None:

                    checks["contains_expected"] = bool(content_fields.get("content_verified"))

                return {"target_path": target, "read_bytes": len(data), "checks": checks, **content_fields}

            rows.append(

                fs_row(

                    meta,

                    scenario_name,

                    operation,

                    run_index,

                    ["python:read_bytes", target, str(read_bytes)],

                    scenario_start,

                    targeted_read,

                    not first_useful_recorded,

                )

            )

            first_useful_recorded = True

            continue

        if operation == "fs.first_edit":

            if not mounted:

                rows.append(skip_row(meta, scenario_name, operation, run_index, "mount_not_started"))

                continue

            edit_path = str_default(defaults, "edit_path", ".oak-bench/mount-edit.txt")

            def first_edit() -> dict[str, Any]:

                target = path / edit_path

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

                text = f"oak mount benchmark {scenario_name} run {run_index}\n"

                target.write_text(text)

                round_trip = target.read_text()

                return {

                    "edit_path": edit_path,

                    "write_bytes": len(text.encode("utf-8")),

                    "checks": {"ok": round_trip == text, "round_trip": round_trip == text},

                }

            rows.append(

                fs_row(

                    meta,

                    scenario_name,

                    operation,

                    run_index,

                    ["python:write_text", edit_path],

                    scenario_start,

                    first_edit,

                    not first_useful_recorded,

                )

            )

            first_useful_recorded = True

            continue

        if operation == "fs.sparse_paths":

            if not mounted:

                rows.append(skip_row(meta, scenario_name, operation, run_index, "mount_not_started"))

                continue

            sparse_paths = list_default(defaults, "sparse_paths")

            def sparse_probe() -> dict[str, Any]:

                checked = []

                for rel in sparse_paths:

                    item = path / rel

                    exists = item.exists()

                    is_dir = item.is_dir() if exists else False

                    checked.append({"path": rel, "exists": exists, "is_dir": is_dir})

                    if exists and not is_dir:

                        with item.open("rb") as fh:

                            fh.read(128)

                return {"sparse_paths": checked, "checks": {"ok": all(item["exists"] for item in checked)}}

            rows.append(

                fs_row(

                    meta,

                    scenario_name,

                    operation,

                    run_index,

                    ["python:sparse_paths"] + sparse_paths,

                    scenario_start,

                    sparse_probe,

                    not first_useful_recorded,

                )

            )

            first_useful_recorded = True

            continue

        if operation == "fs.huge_partial_read":

            if not mounted:

                rows.append(skip_row(meta, scenario_name, operation, run_index, "mount_not_started"))

                continue

            huge_path = str_default(defaults, "huge_file_path", "")

            if not huge_path:

                rows.append(skip_row(meta, scenario_name, operation, run_index, "huge_file_path_not_configured"))

                continue

            read_bytes = int_default(defaults, "huge_file_read_bytes", 4096)

            def huge_partial_read() -> dict[str, Any]:

                target = path / huge_path

                with target.open("rb") as fh:

                    data = fh.read(read_bytes)

                return {

                    "huge_file_path": huge_path,

                    "requested_bytes": read_bytes,

                    "read_bytes": len(data),

                    "checks": {"ok": target.exists() and len(data) > 0, "exists": target.exists()},

                }

            rows.append(

                fs_row(

                    meta,

                    scenario_name,

                    operation,

                    run_index,

                    ["python:read_prefix", huge_path, str(read_bytes)],

                    scenario_start,

                    huge_partial_read,

                    not first_useful_recorded,

                )

            )

            first_useful_recorded = True

            continue

        if operation == "disk.usage":

            if not mounted:

                rows.append(skip_row(meta, scenario_name, operation, run_index, "mount_not_started"))

                continue

            max_entries = int_default(defaults, "max_disk_entries", 100000)

            rows.append(

                fs_row(

                    meta,

                    scenario_name,

                    operation,

                    run_index,

                    ["python:bounded_tree_usage", str(path), str(max_entries)],

                    scenario_start,

                    lambda: bounded_tree_usage(path, max_entries),

                    False,

                )

            )

            continue

        if operation == "oak.desc":

            if not mounted:

                rows.append(skip_row(meta, scenario_name, operation, run_index, "mount_not_started"))

                continue

            message = str_default(defaults, "desc_message", f"bench {scenario_name} run {run_index}")

            run_timed_command(

                rows, meta, scenario_name, operation, run_index, [oak_bin, "desc", message], path, defaults, scenario_start

            )

            continue

        if operation == "oak.finish":

            if not mounted:

                rows.append(skip_row(meta, scenario_name, operation, run_index, "mount_not_started"))

                continue

            if not oak_supports(oak_bin, "finish"):

                rows.append(

                    skip_row(meta, scenario_name, operation, run_index, "finish_not_supported_by_binary")

                )

                continue

            message = str_default(defaults, "desc_message", f"bench {scenario_name} run {run_index}")

            desc_file = finish_desc_file_path(run_root, scenario_name, run_index).resolve()

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

            desc_file.write_text(message + "\n")

            row = run_timed_command(

                rows,

                meta,

                scenario_name,

                operation,

                run_index,

                [oak_bin, "finish", "--desc-file", str(desc_file)],

                path,

                defaults,

                scenario_start,

                extra={"desc_file": str(desc_file)},

            )

            released = not path.exists()

            row["checks"] = {"ok": row["returncode"] == 0, "path_released": released}

            mounted = mounted and not released

            continue

        if operation == "space.clean":

            # space clean operates on the space (parent of the mounts); it

            # should tear down this mount only if its tree is clean

            # (committed + pushed), which is the lifecycle property under test.

            row = run_timed_command(

                rows,

                meta,

                scenario_name,

                operation,

                run_index,

                [oak_bin, "space", "clean"],

                path.parent,

                defaults,

                scenario_start,

            )

            released = not path.exists()

            row["checks"] = {"ok": row["returncode"] == 0, "mount_released_by_clean": released}

            mounted = mounted and not released

            continue

        if operation == "loop.push_desc":

            # The steady-state agent loop on a mount: edit, commit, desc,

            # push, repeat. Lifecycle cost lives in the repetition, not the

            # first call — per-iteration rows keep the trend, the .total row

            # keeps the loop-level price.

            if not mounted:

                rows.append(skip_row(meta, scenario_name, operation, run_index, "mount_not_started"))

                continue

            iterations = int_default(defaults, "loop_iterations", 3)

            per_iteration_ms: list[float] = []

            loop_failures = 0

            loop_start = time.perf_counter()

            for iteration in range(iterations):

                iteration_start = time.perf_counter()

                def loop_edit(iteration: int = iteration) -> dict[str, Any]:

                    target = path / ".oak-bench" / "loop-edit.txt"

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

                    text = f"loop iteration {iteration} of {scenario_name} run {run_index}\n"

                    target.write_text(text)

                    return {"iteration": iteration, "checks": {"ok": target.read_text() == text}}

                rows.append(

                    fs_row(

                        meta,

                        scenario_name,

                        "loop.edit",

                        run_index,

                        ["python:write_text", ".oak-bench/loop-edit.txt"],

                        scenario_start,

                        loop_edit,

                        False,

                    )

                )

                commit_row = run_timed_command(

                    rows,

                    meta,

                    scenario_name,

                    "loop.commit",

                    run_index,

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

                    path,

                    defaults,

                    scenario_start,

                    extra={"iteration": iteration},

                )

                desc_row = run_timed_command(

                    rows,

                    meta,

                    scenario_name,

                    "loop.desc",

                    run_index,

                    [oak_bin, "desc", f"loop iteration {iteration}"],

                    path,

                    defaults,

                    scenario_start,

                    extra={"iteration": iteration},

                )

                loop_failures += int(commit_row["returncode"] != 0) + int(desc_row["returncode"] != 0)

                if allow_push:

                    push_row = run_timed_command(

                        rows,

                        meta,

                        scenario_name,

                        "loop.push",

                        run_index,

                        [oak_bin, "push"],

                        path,

                        defaults,

                        scenario_start,

                        extra={"iteration": iteration},

                    )

                    loop_failures += int(push_row["returncode"] != 0)

                else:

                    rows.append(

                        skip_row(

                            meta,

                            scenario_name,

                            "loop.push",

                            run_index,

                            "safe_push_disabled",

                            extra={"iteration": iteration},

                        )

                    )

                per_iteration_ms.append(round((time.perf_counter() - iteration_start) * 1000, 3))

            rows.append(

                {

                    **meta,

                    "scenario": scenario_name,

                    "run": run_index,

                    "operation": "loop.push_desc.total",

                    "elapsed_ms": round((time.perf_counter() - loop_start) * 1000, 3),

                    "returncode": 0 if loop_failures == 0 else 1,

                    "command": ["loop:edit+commit+desc+push", str(iterations)],

                    "iterations": iterations,

                    "iteration_failures": loop_failures,

                    "push_enabled": allow_push,

                    "per_iteration_ms": per_iteration_ms,

                    "iteration_ms_max": max(per_iteration_ms) if per_iteration_ms else None,

                    "since_scenario_start_ms": round((time.perf_counter() - scenario_start) * 1000, 3),

                    "attempts": 1,

                    "retry_count": 0,

                }

            )

            continue

        rows.append(skip_row(meta, scenario_name, operation, run_index, "operation_not_implemented"))

    if mounted:

        command = [oak_bin, "mount", "end", str(path)]

        row = run_timed_command(rows, meta, scenario_name, "mount.teardown.auto", run_index, command, None, defaults, scenario_start)

        cleanup_safe = cleanup_safe and row["returncode"] == 0

    apply_scenario_cache_state(rows, scenario, purge_result)

    return rows, cleanup_safe

def run_recovery_probe(

    meta: dict[str, Any],

    oak_bin: str,

    owner_repo: str,

    scenario: dict[str, Any],

    defaults: dict[str, Any],

    run_root: Path,

    run_index: int,

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

    scenario_name = str(scenario.get("name", "interrupted_recovery"))

    path = mount_path(run_root, scenario_name, run_index)

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

    (path / ".oak-bench-partial").write_text("simulated interrupted mount destination\n")

    scenario_start = time.perf_counter()

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

    command = [oak_bin, "mount", owner_repo, str(path)]

    row = run_timed_command(rows, meta, scenario_name, "recovery.existing_partial_dir", run_index, command, None, defaults, scenario_start)

    if row["returncode"] != 0 and "not empty" in str(row.get("stderr", "")):

        rows[-1] = skip_row(

            meta,

            scenario_name,

            "recovery.existing_partial_dir",

            run_index,

            "mount_partial_destination_recovery_not_supported",

            {

                "attempted_command": command,

                "process_returncode": row["returncode"],

                "process_stderr": row.get("stderr"),

                "recovery_supported": False,

            },

        )

        shutil.rmtree(path, ignore_errors=True)

        apply_scenario_cache_state(rows, scenario)

        return rows, True

    mounted = row["returncode"] == 0

    if mounted:

        end_row = run_timed_command(

            rows,

            meta,

            scenario_name,

            "mount.teardown",

            run_index,

            [oak_bin, "mount", "end", str(path)],

            None,

            defaults,

            scenario_start,

        )

        apply_scenario_cache_state(rows, scenario)

        return rows, end_row["returncode"] == 0

    apply_scenario_cache_state(rows, scenario)

    return rows, False

def run_parallel_probe(

    meta: dict[str, Any],

    oak_bin: str,

    owner_repo: str,

    scenario: dict[str, Any],

    defaults: dict[str, Any],

    run_root: Path,

    run_index: int,

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

    scenario_name = str(scenario.get("name", "parallel_mounts"))

    parallelism = max(1, int_default(defaults, "parallelism", 4))

    timeout = int_default(defaults, "timeout_seconds", 120)

    scenario_start = time.perf_counter()

    def worker(worker_index: int) -> dict[str, Any]:

        path = run_root / f"{scenario_name}-run-{run_index}-worker-{worker_index}"

        command = [oak_bin, "mount", owner_repo, str(path)]

        mount_result = run_process(command, None, timeout, int_default(defaults, "retries", 0))

        first_ls_ok = False

        teardown_returncode: Optional[int] = None

        if mount_result["returncode"] == 0:

            try:

                os.listdir(path)

                first_ls_ok = True

            except OSError:

                first_ls_ok = False

            teardown = run_process([oak_bin, "mount", "end", str(path)], None, timeout, 0)

            teardown_returncode = int(teardown["returncode"])

        return {

            "worker": worker_index,

            "returncode": mount_result["returncode"],

            "elapsed_ms": mount_result["elapsed_ms"],

            "first_ls_ok": first_ls_ok,

            "teardown_returncode": teardown_returncode,

            "stderr": mount_result.get("stderr", ""),

        }

    start = time.perf_counter()

    with concurrent.futures.ThreadPoolExecutor(max_workers=parallelism) as executor:

        worker_rows = list(executor.map(worker, range(parallelism)))

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

    failures = [

        item

        for item in worker_rows

        if item["returncode"] != 0 or not item["first_ls_ok"] or item["teardown_returncode"] not in {0}

    ]

    result = {

        "elapsed_ms": elapsed_ms,

        "returncode": 0 if not failures else 1,

        "stdout": "",

        "stderr": "\n".join(tail(str(item.get("stderr", "")), 500) for item in failures),

        "attempts": parallelism,

        "retry_count": 0,

    }

    row = command_row(

        meta,

        scenario_name,

        "parallel.mounts",

        run_index,

        [oak_bin, "mount", owner_repo, "<parallel-dests>"],

        result,

        scenario_start,

        {

            "parallelism": parallelism,

            "workers": worker_rows,

            **scenario_cache_fields(scenario),

            "checks": {"ok": not failures, "successful_workers": parallelism - len(failures)},

        },

    )

    return [row], not failures

def scenario_skip_reason(

    scenario: dict[str, Any],

    mount_available: bool,

    owner_repo: str,

    spec: dict[str, Any],

) -> Optional[str]:

    if not mount_available:

        return "oak_mount_unavailable"

    if scenario.get("requires_remote") and not owner_repo:

        return "mount_repo_not_configured"

    if scenario.get("requires_safe_push") and not safe_push_enabled(spec):

        return "safe_push_disabled"

    if scenario.get("requires_huge_file_path"):

        defaults = spec_defaults(spec)

        if not str_default(defaults, "huge_file_path", ""):

            return "huge_file_path_not_configured"

    if scenario.get("requires_external_offline_harness"):

        return "requires_external_offline_harness"

    return None

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

    """Persist mount rows through the shared lane-validated write path.

    Mount rows go through the same row-contract validation as every other

    lane (fail loudly after persisting); the explicit-.jsonl-file form keeps

    that guarantee with a direct write plus the same validation.

    """

    from oakbench.results import ResultsStore

    from oakbench.rows import RowContractViolation, validate_rows

    if results_arg.suffix == ".jsonl":

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

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

            for row in rows:

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

        errors = validate_rows(rows, "mount")

        if errors:

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

            raise RowContractViolation(

                f"Row contract violation (mount lane). Rows were written to {results_arg} "

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

            )

        return results_arg

    store = ResultsStore(results_arg, lane="mount", filename_suffix="mount")

    raw_path, _ = store.write(timestamp, rows)

    return raw_path

def main() -> int:

    args = parse_args()

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

    spec = load_spec(args.spec)

    defaults = spec_defaults(spec)

    oak_bin = resolve_oak_bin(args.oak_bin)

    meta = metadata(timestamp, args.spec, spec, oak_bin)

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

    owner_repo = mount_repo(spec)

    meta["mount_repo_configured"] = bool(owner_repo)

    meta["mount_repo"] = owner_repo if owner_repo else None

    run_root = args.dest / timestamp

    cleanup_safe = True

    scenarios = spec.get("scenarios") or []

    if not isinstance(scenarios, list):

        scenarios = []

    if oak_bin is not None:

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

    with measurement_lock("mount_probe") as lock_info:

        meta["measurement_lock_wait_ms"] = lock_info.wait_ms

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

        if oak_bin is None:

            rows.append(skip_row(meta, "capability_probe", "oak.version", 0, f"oak binary not found: {args.oak_bin}"))

            rows.append(

                skip_row(meta, "capability_probe", "oak.mount.help", 0, f"oak binary not found: {args.oak_bin}")

            )

        else:

            version_result = run_process([oak_bin, "--version"], None, int_default(defaults, "timeout_seconds", 120), 0)

            version_text = (version_result.get("stdout") or version_result.get("stderr") or "").strip()

            meta["subject_versions"] = {"oak_mount": version_text or "unknown"}

            rows.append(command_row(meta, "capability_probe", "oak.version", 0, [oak_bin, "--version"], version_result))

            mount_help_result = run_process(

                [oak_bin, "mount", "--help"], None, int_default(defaults, "timeout_seconds", 120), 0

            )

            mount_available = mount_help_result["returncode"] == 0

            rows.append(

                command_row(

                    meta,

                    "capability_probe",

                    "oak.mount.help",

                    0,

                    [oak_bin, "mount", "--help"],

                    mount_help_result,

                    extra={"mount_available": mount_available},

                )

            )

            for scenario in scenarios:

                if not isinstance(scenario, dict):

                    continue

                scenario_name = str(scenario.get("name", "unnamed"))

                if scenario_name == "capability_probe":

                    continue

                reason = scenario_skip_reason(scenario, mount_available, owner_repo, spec)

                if reason:

                    for run_index in scenario_runs(defaults):

                        rows.append(

                            skip_row(

                                meta,

                                scenario_name,

                                "scenario.skip",

                                run_index,

                                reason,

                                scenario_cache_fields(scenario),

                            )

                        )

                    continue

                for run_index in scenario_runs(defaults):

                    operations = [str(item) for item in (scenario.get("operations") or [])]

                    if operations == ["recovery.existing_partial_dir"]:

                        scenario_rows, scenario_cleanup_safe = run_recovery_probe(

                            meta, oak_bin, owner_repo, scenario, defaults, run_root, run_index

                        )

                    elif operations == ["parallel.mounts"]:

                        scenario_rows, scenario_cleanup_safe = run_parallel_probe(

                            meta, oak_bin, owner_repo, scenario, defaults, run_root, run_index

                        )

                    else:

                        scenario_rows, scenario_cleanup_safe = run_operation_sequence(

                            meta, oak_bin, owner_repo, safe_push_enabled(spec), scenario, defaults, run_root, run_index

                        )

                    rows.extend(scenario_rows)

                    cleanup_safe = cleanup_safe and scenario_cleanup_safe

    raw_path = write_results(args.results, timestamp, rows)

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

    if oak_bin is None:

        return 0

    if not args.keep and cleanup_safe:

        shutil.rmtree(run_root, ignore_errors=True)

    elif not args.keep:

        print(f"[keep] cleanup skipped because a mount command failed: {run_root}", file=sys.stderr)

    else:

        print(f"[keep] {run_root}")

    real_failures = [

        row

        for row in rows

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

        and row.get("scenario") != "capability_probe"

        and not row.get("skipped")

    ]

    return 1 if real_failures else 0

if __name__ == "__main__":

    raise SystemExit(main())