"""The scale-lane operation catalog: every user-facing op, git and oak.

Each op is one function per subject kind. Inside it, ONLY ``ctx.timed(...)``

calls are measured (their elapsed times are summed into the row); setup,

oracle and teardown use ``ctx.untimed(...)`` and run outside the timed path

(ADR-0011: oracles untimed). An op whose timed command fails emits no oracle

verdict (the row carries the failure); an op with no equivalent for a kind

returns a skip (returncode 77 with a reason) so coverage gaps are rows, never

silent omissions.

Oracle evidence classes (``oracle_reader``):

- ``git_fixture_instrument``: expected values read from the generated

  fixture with git plumbing, compared against the subject's worktree bytes

  hashed by the harness (independent of the subject under test);

- ``subject_output``: the subject's own output parsed by the harness

  (counts, names, JSON fields);

- ``subject_file_inspect``: Oak's ``oak file inspect`` (subject-produced)

  or git's ``git show``/``cat-file`` of the subject repository.

t/perf references cite upstream git test FILE NAMES only (GPL-2.0); the ops

are re-implemented here, nothing is copied (docs/prior-art.md).

"""

from __future__ import annotations

import hashlib

import json

import os

import random

import shutil

from dataclasses import dataclass, field

from pathlib import Path

from typing import Any, Callable, Dict, List, Optional, Sequence, Tuple

from . import generator as gen

UNIT_ORDER: Tuple[Tuple[str, ...], ...] = (

    ("status.clean",),

    ("log.full",),

    ("log.n",),

    ("log.path",),

    ("blame",),

    ("file.show",),

    ("branch.list",),

    ("review.branch_diff",),

    ("merge.preview",),

    ("merge.preview.conflict",),

    ("switch.existing", "switch.back"),

    ("branch.create",),

    ("status.dirty", "diff.dirty", "commit.dirty"),

    ("merge.local",),

    ("merge.conflict",),

    ("push.branch", "pull.ff"),

    ("pull.diverged",),

    ("clone.full",),

    ("gc",),

)

@dataclass(frozen=True)

class OpInfo:

    name: str

    description: str

    tperf: Tuple[str, ...] = ()

    core_yaml: Tuple[str, ...] = ()

    mutates: bool = False

OP_INFO: Dict[str, OpInfo] = {info.name: info for info in (

    OpInfo("status.clean", "Status of a clean working tree (porcelain).",

           ("p0005-status.sh", "p7519-fsmonitor.sh"), ("status.clean",)),

    OpInfo("status.dirty", "Status after the tier's dirty shape (modify/add/delete/rename).",

           ("p0005-status.sh", "p7519-fsmonitor.sh"), ("status.dirty",)),

    OpInfo("diff.dirty", "Full worktree diff of the dirty shape.",

           ("p4000-diff-algorithms.sh", "p0009-diff-pathspec.sh"), ("diff.dirty",)),

    OpInfo("commit.dirty", "Commit the dirty shape (git: add -A + commit, summed).",

           ("p0007-write-cache.sh", "p0090-cache-tree.sh"), ("snapshot.dirty",), mutates=True),

    OpInfo("log.full", "Entire history reachable from the current branch, one line per commit.",

           ("p0001-rev-list.sh", "p4205-log-pretty-formats.sh"), ("history.log.range",)),

    OpInfo("log.n", "The newest 20 commits.", ("p4205-log-pretty-formats.sh",), ("history.log.tip",)),

    OpInfo("log.path", "History of one path (the fixture's probe path).",

           ("p4211-line-log.sh", "p0001-rev-list.sh"), ("history.log.range",)),

    OpInfo("blame", "Line authorship of the probe path.", ("p8020-last-modified.sh",), ()),

    OpInfo("file.show", "Committed bytes of the probe path at HEAD.", ("p1006-cat-file.sh",), ()),

    OpInfo("branch.list", "Enumerate branches (oak: open branches; harness branches are closed, git deletes them).", ("p6300-for-each-ref.sh",), ("branch.list",)),

    OpInfo("branch.create", "Create and switch to a new branch.", ("p1400-update-ref.sh",),

           ("branch.create",), mutates=True),

    OpInfo("switch.existing", "Switch to the fixture's side branch b00000.",

           ("p0006-read-tree-checkout.sh", "p2000-sparse-operations.sh"), ("branch.switch",), mutates=True),

    OpInfo("switch.back", "Switch back to the base branch.", ("p0006-read-tree-checkout.sh",),

           ("history.restore_head",), mutates=True),

    OpInfo("review.branch_diff", "Checkout-free diff of b00000 against main.",

           ("p4000-diff-algorithms.sh", "p6010-merge-base.sh"), ()),

    OpInfo("merge.preview", "Dry-run merge prediction of b00000 into main.",

           ("p6010-merge-base.sh",), ()),

    OpInfo("merge.preview.conflict", "Dry-run merge prediction of the conflict-shape branches.",

           ("p6010-merge-base.sh",), ("merge.conflict",)),

    OpInfo("merge.local", "Local (non-fast-forward) merge of b00000 into a scratch branch.",

           ("p6010-merge-base.sh", "p3400-rebase.sh"), (), mutates=True),

    OpInfo("merge.conflict", "Local merge of the conflict-shape branches (expected to conflict), then abort.",

           ("p6010-merge-base.sh",), ("merge.conflict", "merge.abort"), mutates=True),

    OpInfo("push.branch", "Push one new commit over loopback HTTP.",

           ("p5313-pack-objects.sh", "p5310-pack-bitmaps.sh"), (), mutates=True),

    OpInfo("pull.ff", "Pull one upstream commit (fast-forward).",

           ("p5311-pack-bitmaps-fetch.sh", "p5551-fetch-rescan.sh"), (), mutates=True),

    OpInfo("pull.diverged", "Pull when local and upstream both have one new commit.",

           ("p5311-pack-bitmaps-fetch.sh", "p3400-rebase.sh"), (), mutates=True),

    OpInfo("clone.full", "Full clone over loopback HTTP into a new directory.",

           ("p5600-partial-clone.sh", "p5601-clone-reference.sh"), ()),

    OpInfo("gc", "Repository maintenance (git gc / oak maintenance compact).",

           ("p5304-prune.sh", "p5313-pack-objects.sh", "p5303-many-packs.sh"), (), mutates=True),

)}

ALL_OPS: Tuple[str, ...] = tuple(op for unit in UNIT_ORDER for op in unit)

assert set(ALL_OPS) == set(OP_INFO), "UNIT_ORDER and OP_INFO must list the same operations"

@dataclass

class Timed:

    label: str

    command: List[str]

    capture: Any

@dataclass

class Outcome:

    timed: List[Timed] = field(default_factory=list)

    failed: Optional[str] = None

    skip_reason: Optional[str] = None

    oracle_ok: Optional[bool] = None

    oracle_method: Optional[str] = None

    oracle_reader: Optional[str] = None

    oracle_detail: Dict[str, Any] = field(default_factory=dict)

    semantic_class: Optional[str] = None

    extra: Dict[str, Any] = field(default_factory=dict)

class OpFailed(Exception):

    """A timed command failed; the row records it and the unit aborts."""

class OpSkipped(Exception):

    pass

class Ctx:

    """Per (subject, run, op) execution context."""

    def __init__(self, prepared: Any, subject: Any, run: int, runner: Callable[..., Any],

                 scratch: Path, timeout_s: float) -> None:

        self.p = prepared

        self.subject = subject

        self.kind = subject.kind

        self.bin = str(subject.bin)

        self.run = run

        self._run_timed = runner

        self.scratch = scratch

        self.timeout_s = timeout_s

        self.outcome = Outcome()

        self.tag = f"{subject.name}-r{run}"

    # -- command helpers ----------------------------------------------------

    def timed(self, args: Sequence[str], cwd: Path, *, label: str = "main", accept: Sequence[int] = (0,),

              raw: bool = False) -> Any:

        command = [self.bin, *args]

        capture = self._run_timed(command, cwd, self.p.env, 4000, full_output_bytes=64 << 20,

                                  watchdog_seconds=self.timeout_s, raw_output_bytes=(64 << 20) if raw else None)

        self.outcome.timed.append(Timed(label, command, capture))

        if capture.returncode not in accept:

            tail = (capture.stderr_text or capture.stdout_text or "")[-400:]

            raise OpFailed(f"{label}: exit {capture.returncode}: {tail}")

        return capture

    def untimed(self, args: Sequence[str], cwd: Path, *, check: bool = True, binary: Optional[str] = None

                ) -> Any:

        from .workspace import run

        return run([binary or self.bin, *args], cwd=cwd, env=self.p.env, check=check)

    def git(self, args: Sequence[str], cwd: Path, check: bool = True) -> Any:

        """The oracle's git instrument (the resolved real git), never the subject binary."""

        return self.untimed(args, cwd, check=check, binary=self.p.git_bin)

    def oracle(self, ok: bool, method: str, reader: str, **detail: Any) -> None:

        self.outcome.oracle_ok = None if "reader_unavailable" in detail else bool(ok)

        self.outcome.oracle_method = method

        self.outcome.oracle_reader = reader

        self.outcome.oracle_detail.update(detail)

    # -- workspace paths ----------------------------------------------------

    @property

    def work(self) -> Path:

        return self.p.path(self.kind, "work")

    def client(self, name: str) -> Path:

        return self.p.path(self.kind, name)

    @property

    def state(self) -> Dict[str, Any]:

        return self.p.state.setdefault(self.kind, {"harness_commits": 0, "deleted": set(), "server_extra": 0,

                                                     "branch_seq": 0})

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

# Shared oracle helpers (harness-side, independent of the subject)

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

def worktree_files(root: Path) -> List[str]:

    out = []

    for base, dirs, files in os.walk(root):

        dirs[:] = [d for d in dirs if d not in (".git", ".oak")]

        rel = os.path.relpath(base, root)

        for name in files:

            out.append(name if rel == "." else f"{rel}/{name}".replace(os.sep, "/"))

    return out

def blob_id_of(path: Path) -> str:

    digest = hashlib.sha1()

    size = path.stat().st_size

    digest.update(b"blob %d\0" % size)

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

        for chunk in iter(lambda: fh.read(1 << 20), b""):

            digest.update(chunk)

    return digest.hexdigest()

def check_worktree(ctx: Ctx, root: Path, expected_count: Optional[int], ignore_prefixes: Sequence[str] = ()

                   ) -> Tuple[bool, Dict[str, Any]]:

    """Sampled content check against the fixture plus an exact file count."""

    exp = ctx.p.expected

    files = [f for f in worktree_files(root) if not any(f.startswith(prefix) for prefix in ignore_prefixes)]

    mismatched = []

    for rel, blob in exp.sample.items():

        if rel in ctx.state["deleted"]:

            continue

        target = root / rel

        if not target.is_file() or blob_id_of(target) != blob:

            mismatched.append(rel)

    detail = {"files_seen": len(files), "files_expected": expected_count, "sample_checked": len(exp.sample),

              "sample_mismatched": mismatched[:10], "sample_mismatch_count": len(mismatched)}

    ok = not mismatched and (expected_count is None or len(files) == expected_count)

    return ok, detail

def porcelain_paths(text: str) -> List[str]:

    paths = []

    for line in text.splitlines():

        line = line.rstrip()

        if len(line) < 3:

            continue

        rest = line[2:].strip() if line[2] == " " or line[1] == " " else line.split(None, 1)[-1]

        if " -> " in rest:

            old, new = rest.split(" -> ", 1)

            paths.extend([old.strip('"'), new.strip('"')])

        else:

            paths.append(rest.strip('"'))

    return paths

def committed_bytes(ctx: Ctx, repo: Path, paths: Sequence[str]) -> Tuple[Dict[str, bytes], str, Dict[str, Any]]:

    """Committed HEAD bytes of ``paths`` read WITHOUT the subject binary.

    git: ``git cat-file`` with the oracle's own git. oak: the independent

    stdlib reader of Oak's SQLite store (oakbench.oak_sqlite_reader). Returns

    (payloads, reader label, meta). A reader bound (Unavailable) puts

    ``reader_unavailable`` in meta; Ctx.oracle then records oracle_ok None

    (unmeasured, ADR-0002) instead of a guess.

    """

    if _is_git(ctx):

        out: Dict[str, bytes] = {}

        for rel in paths:

            proc = ctx.git(["cat-file", "blob", f"HEAD:{rel}"], repo, check=False)

            if proc.returncode == 0:

                out[rel] = proc.stdout

        return out, "git_cat_file_oracle_instrument", {}

    from oakbench import oak_sqlite_reader as reader

    try:

        payloads, meta = reader.committed_payloads(Path(repo), list(paths))

    except reader.Unavailable as exc:

        # A harness/reader bound, not a property of Oak's data: the oracle

        # becomes unmeasured (Ctx.oracle turns this into oracle_ok None).

        return {}, "oak_sqlite_reader_independent", {"reader_unavailable": str(exc)[:300]}

    except reader.Invalid as exc:

        return {}, "oak_sqlite_reader_independent", {"invalid": str(exc)[:300]}

    return payloads, "oak_sqlite_reader_independent", {"head": meta.get("head"),

                                                       "reader_version": meta.get("reader_version")}

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

# Dirty shape (untimed setup)

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

def apply_dirty(ctx: Ctx) -> Dict[str, Any]:

    spec = ctx.p.tier.spec

    shape = ctx.p.tier.dirty

    cache = gen.layout(spec)

    factory = gen.ContentFactory(spec)

    protected = set(ctx.p.expected.protected)

    rng = random.Random(f"{spec.seed}:dirty:{ctx.tag}")

    deleted: set = ctx.state["deleted"]

    wanted = shape.modified + shape.deleted + shape.renamed

    pool: List[int] = []

    attempts = 0

    seen = set()

    while len(pool) < wanted and attempts < wanted * 20 + 100:

        attempts += 1

        index = rng.randrange(spec.files)

        path = gen.file_path(spec, index, cache)

        if index in seen or path in protected or path in deleted:

            continue

        seen.add(index)

        pool.append(index)

    modified = pool[:shape.modified]

    removed = pool[shape.modified:shape.modified + shape.deleted]

    renamed = pool[shape.modified + shape.deleted:]

    root = ctx.work

    mod_paths, del_paths, ren_pairs, add_paths = [], [], [], []

    for index in modified:

        path = gen.file_path(spec, index, cache)

        (root / path).write_bytes(factory.text(index, f"dirty-{ctx.tag}"))

        mod_paths.append(path)

    for index in removed:

        path = gen.file_path(spec, index, cache)

        (root / path).unlink()

        del_paths.append(path)

    for index in renamed:

        path = gen.file_path(spec, index, cache)

        target = f"renamed/{ctx.tag}/f{index}.txt"

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

        os.replace(root / path, root / target)

        ren_pairs.append((path, target))

    for number in range(shape.added):

        rel = f"added/{ctx.tag}/n{number}.txt"

        (root / rel).parent.mkdir(parents=True, exist_ok=True)

        (root / rel).write_bytes(factory.text(10_000_000 + number, f"add-{ctx.tag}"))

        add_paths.append(rel)

    dirty = {"modified": mod_paths, "deleted": del_paths, "renamed": ren_pairs, "added": add_paths,

             "requested": [shape.modified, shape.added, shape.deleted, shape.renamed],

             "applied": [len(mod_paths), len(add_paths), len(del_paths), len(ren_pairs)]}

    ctx.state["dirty"] = dirty

    return dirty

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

# Operations. Each returns None; results land in ctx.outcome.

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

def _is_git(ctx: Ctx) -> bool:

    return ctx.kind == "git"

def op_status_clean(ctx: Ctx) -> None:

    args = ["status", "--porcelain=v1"] if _is_git(ctx) else ["status", "--porcelain"]

    cap = ctx.timed(args, ctx.work)

    paths = porcelain_paths(cap.stdout_full_text or "")

    ctx.oracle(not paths, "clean_status_reports_no_paths", "subject_output", reported_paths=paths[:10])

def _dirty_branch_enter(ctx: Ctx) -> None:

    """The dirty cycle runs on a throwaway branch (untimed switch), so the base

    branch never accumulates harness commits: every run's switch/log/merge ops

    see the same repository state (no run-over-run drift)."""

    name = f"dirty-{ctx.tag}"

    if _is_git(ctx):

        ctx.git(["switch", "-q", "-c", name, "main"], ctx.work)

    else:

        ctx.untimed(["switch", "-c", name], ctx.work)

    ctx.state["dirty_branch"] = name

def _dirty_branch_leave(ctx: Ctx) -> None:

    name = ctx.state.pop("dirty_branch", None)

    if name is None:

        return

    _drop_branch(ctx, name)

def _drop_branch(ctx: Ctx, name: str) -> None:

    """Untimed teardown: back to base, then remove the harness branch so branch

    counts (branch.list) stay fixed across runs. git deletes it; Oak has no

    local delete, so it is closed (and oak branch.list lists open branches)."""

    if _is_git(ctx):

        ctx.git(["switch", "-q", "-f", "main"], ctx.work)

        ctx.git(["branch", "-q", "-D", name], ctx.work, check=False)

    else:

        ctx.untimed(["switch", _base_branch(ctx)], ctx.work)

        ctx.untimed(["close", name], ctx.work, check=False)

def recover(ctx: Ctx) -> Optional[str]:

    """Untimed repair after a failed unit: discard worktree changes, return to base."""

    try:

        ctx.state.pop("dirty", None)

        stale = ctx.state.pop("dirty_branch", None)

        if _is_git(ctx):

            ctx.git(["merge", "--abort"], ctx.work, check=False)

            ctx.git(["reset", "-q", "--hard"], ctx.work, check=False)

            ctx.git(["clean", "-fdq", "-e", "build/"], ctx.work, check=False)

            ctx.git(["switch", "-q", "-f", "main"], ctx.work)

        else:

            ctx.untimed(["reset", "-f"], ctx.work, check=False)

            for extra in ("added", "renamed"):

                shutil.rmtree(ctx.work / extra, ignore_errors=True)

            ctx.untimed(["switch", _base_branch(ctx)], ctx.work)

        if stale:

            _drop_branch(ctx, stale)

        return None

    except Exception as exc:  # noqa: BLE001 - reported on the row, never hidden

        return f"{type(exc).__name__}: {str(exc)[-300:]}"

def op_status_dirty(ctx: Ctx) -> None:

    _dirty_branch_enter(ctx)

    dirty = apply_dirty(ctx)

    ctx.outcome.extra["dirty_applied"] = dirty["applied"]

    args = ["status", "--porcelain=v1"] if _is_git(ctx) else ["status", "--porcelain"]

    cap = ctx.timed(args, ctx.work)

    text = cap.stdout_full_text or ""

    reported = set(porcelain_paths(text))

    must = set(dirty["modified"]) | set(dirty["deleted"]) | {old for old, _ in dirty["renamed"]}

    missing = sorted(must - reported)

    # Git collapses untracked directories ("?? added/<tag>/"); accept a

    # reported ancestor directory for added/renamed targets.

    new_paths = dirty["added"] + [new for _, new in dirty["renamed"]]

    uncovered = [p for p in new_paths if p not in reported and not any(

        p.startswith(r.rstrip("/") + "/") for r in reported if r.endswith("/") or r in ("added", "renamed"))]

    ctx.oracle(not missing and not uncovered, "dirty_paths_reported", "subject_output",

               missing=missing[:10], missing_count=len(missing), uncovered_new=uncovered[:10],

               uncovered_new_count=len(uncovered), stdout_bytes=cap.stdout_bytes)

def op_diff_dirty(ctx: Ctx) -> None:

    dirty = ctx.state.get("dirty")

    if not dirty:

        raise OpFailed("diff.dirty needs status.dirty's dirty shape")

    args = ["diff", "--no-ext-diff"] if _is_git(ctx) else ["diff", "--print"]

    cap = ctx.timed(args, ctx.work)

    text = cap.stdout_full_text

    if text is None:

        ctx.oracle(cap.stdout_bytes > 0, "diff_output_nonempty_over_8MiB", "subject_output",

                   stdout_bytes=cap.stdout_bytes)

        return

    must = dirty["modified"] + dirty["deleted"]

    missing = [p for p in must if p not in text]

    ctx.oracle(not missing, "diff_mentions_modified_and_deleted_paths", "subject_output",

               missing=missing[:10], missing_count=len(missing), stdout_bytes=cap.stdout_bytes)

def op_commit_dirty(ctx: Ctx) -> None:

    dirty = ctx.state.pop("dirty", None)

    if not dirty:

        raise OpFailed("commit.dirty needs status.dirty's dirty shape")

    if _is_git(ctx):

        ctx.timed(["add", "-A"], ctx.work, label="add")

        ctx.timed(["commit", "-q", "--no-verify", "-m", f"dirty {ctx.tag}"], ctx.work, label="commit")

    else:

        ctx.timed(["commit", "--json"], ctx.work)

    probe = dirty["modified"][0] if dirty["modified"] else None

    gone = dirty["deleted"][0] if dirty["deleted"] else None

    status_args = ["status", "--porcelain=v1"] if _is_git(ctx) else ["status", "--porcelain"]

    left = porcelain_paths(ctx.untimed(status_args, ctx.work).stdout.decode(errors="replace"))

    wanted = [p for p in (probe, gone) if p]

    payloads, reader, meta = committed_bytes(ctx, ctx.work, wanted)

    committed_ok = probe is None or payloads.get(probe) == (ctx.work / probe).read_bytes()

    deleted_ok = gone is None or gone not in payloads

    ctx.oracle(not left and committed_ok and deleted_ok and "invalid" not in meta and "reader_unavailable" not in meta,

               "tree_clean_modified_bytes_committed_deleted_path_absent", reader,

               left_after_commit=left[:10], committed_bytes_match=committed_ok, deleted_absent=deleted_ok, **meta)

    _dirty_branch_leave(ctx)

def _log_count(text: str) -> int:

    return sum(1 for line in text.splitlines() if line.strip() and not line.startswith("No matching commits"))

def op_log_full(ctx: Ctx) -> None:

    args = ["log", "--oneline"] if _is_git(ctx) else ["log", "--oneline", "-n", "100000000"]

    cap = ctx.timed(args, ctx.work)

    count = _log_count(cap.stdout_full_text) if cap.stdout_full_text is not None else None

    expected = ctx.p.expected.main_commits + ctx.state["harness_commits"]

    ctx.oracle(count == expected, "log_line_count_equals_history_length", "subject_output",

               lines=count, expected=expected, stdout_bytes=cap.stdout_bytes,

               output_over_capture_cap=count is None)

def op_log_n(ctx: Ctx) -> None:

    cap = ctx.timed(["log", "--oneline", "-n", "20"], ctx.work)

    count = _log_count(cap.stdout_full_text or "")

    expected = min(20, ctx.p.expected.main_commits + ctx.state["harness_commits"])

    ctx.oracle(count == expected, "log_n_line_count", "subject_output", lines=count, expected=expected)

def op_log_path(ctx: Ctx) -> None:

    probe = ctx.p.expected.probe_path

    args = ["log", "--oneline", "--", probe] if _is_git(ctx) else ["log", "--oneline", "-n", "100000000", probe]

    cap = ctx.timed(args, ctx.work)

    count = _log_count(cap.stdout_full_text or "")

    expected = ctx.p.expected.probe_commit_count

    ctx.oracle(count == expected, "path_log_count_equals_generator_touch_count", "subject_output",

               lines=count, expected=expected, probe_path=probe)

def op_blame(ctx: Ctx) -> None:

    if not _is_git(ctx):

        raise OpSkipped("oak_cli_has_no_blame_or_annotate_command")

    probe = ctx.p.expected.probe_path

    cap = ctx.timed(["blame", "-s", "--", probe], ctx.work)

    lines = _log_count(cap.stdout_full_text or "")

    ctx.oracle(lines == ctx.p.expected.probe_lines, "blame_line_count_equals_file_lines", "subject_output",

               lines=lines, expected=ctx.p.expected.probe_lines)

def op_file_show(ctx: Ctx) -> None:

    probe = ctx.p.expected.probe_path

    if _is_git(ctx):

        cap = ctx.timed(["show", f"HEAD:{probe}"], ctx.work, raw=True)

        got = cap.stdout_raw or b""

        reader = "subject_output"

    else:

        out = ctx.scratch / f"show-{ctx.tag}.bin"

        ctx.timed(["file", "inspect", "--at", "HEAD", "--json", probe, "--output", str(out)], ctx.work)

        got = out.read_bytes() if out.exists() else b""

        if out.exists():

            out.unlink()

        reader = "subject_file_inspect"

    ctx.oracle(got == ctx.p.expected.probe_bytes, "shown_bytes_equal_fixture_blob", reader,

               bytes=len(got), expected_bytes=len(ctx.p.expected.probe_bytes))

def op_branch_list(ctx: Ctx) -> None:

    if _is_git(ctx):

        cap = ctx.timed(["branch", "--list"], ctx.work)

        names = [line[2:].strip() for line in (cap.stdout_full_text or "").splitlines() if line.strip()]

        reader = "subject_output"

    else:

        # Open branches: harness branches are closed on teardown (Oak has no local delete).

        cap = ctx.timed(["branch", "list", "--status", "open", "--json"], ctx.work)

        try:

            names = [item.get("name") for item in json.loads(cap.stdout_full_text or "[]")]

        except ValueError:

            names = []

        reader = "subject_output_json"

    spec = ctx.p.tier.spec

    fixture_branches = spec.branches if _is_git(ctx) else ctx.p.oak_branches_created

    need = [gen.branch_name(0)] if fixture_branches else []

    ok = len(names) >= fixture_branches and all(n in names for n in need)

    ctx.oracle(ok, "branch_count_at_least_fixture_branches", reader, listed=len(names),

               fixture_branches=fixture_branches)

def _branch_paths_match(ctx: Ctx, root: Path, want: Dict[str, bytes]) -> Tuple[bool, List[str]]:

    bad = [rel for rel, data in want.items() if not (root / rel).is_file() or (root / rel).read_bytes() != data]

    return not bad, bad

def op_switch_existing(ctx: Ctx) -> None:

    exp = ctx.p.expected

    args = ["switch", "-q", exp.branch] if _is_git(ctx) else ["switch", exp.branch]

    ctx.timed(args, ctx.work)

    ok, bad = _branch_paths_match(ctx, ctx.work, exp.branch_bytes)

    ctx.oracle(ok, "branch_paths_bytes_equal_fixture", "git_fixture_instrument", mismatched=bad[:10])

def _base_branch(ctx: Ctx) -> str:

    return "main" if _is_git(ctx) else str(ctx.p.oak_base_branch)

def op_switch_back(ctx: Ctx) -> None:

    exp = ctx.p.expected

    args = ["switch", "-q", _base_branch(ctx)] if _is_git(ctx) else ["switch", _base_branch(ctx)]

    ctx.timed(args, ctx.work)

    ok, bad = _branch_paths_match(ctx, ctx.work, exp.main_bytes_of_branch_paths)

    ctx.oracle(ok, "base_paths_bytes_equal_fixture_main", "git_fixture_instrument", mismatched=bad[:10])

def current_branch(ctx: Ctx, root: Path) -> str:

    if _is_git(ctx):

        return ctx.git(["symbolic-ref", "--short", "HEAD"], root, check=False).stdout.decode().strip()

    out = ctx.untimed(["status", "--json"], root, check=False).stdout

    try:

        return str(json.loads(out).get("branch"))

    except ValueError:

        return ""

def op_branch_create(ctx: Ctx) -> None:

    ctx.state["branch_seq"] += 1

    name = f"new-{ctx.tag}-{ctx.state['branch_seq']}"

    args = ["switch", "-q", "-c", name] if _is_git(ctx) else ["switch", "-c", name]

    ctx.timed(args, ctx.work)

    now = current_branch(ctx, ctx.work)

    ctx.outcome.semantic_class = "git_branch_from_HEAD" if _is_git(ctx) else "oak_branch_from_main"

    _drop_branch(ctx, name)

    ctx.oracle(now == name, "current_branch_is_new_branch", "subject_output", current=now, wanted=name)

def op_review_branch_diff(ctx: Ctx) -> None:

    exp = ctx.p.expected

    if _is_git(ctx):

        cap = ctx.timed(["diff", "--name-status", f"main...{exp.branch}"], ctx.work)

        paths = [line.split("\t", 1)[-1] for line in (cap.stdout_full_text or "").splitlines() if "\t" in line]

        reader = "subject_output"

    else:

        cap = ctx.timed(["branch", "diff", exp.branch, "--json"], ctx.work)

        try:

            paths = [item["path"] for item in json.loads(cap.stdout_full_text or "{}").get("changed_files", [])]

        except (ValueError, KeyError, AttributeError):

            paths = []

        reader = "subject_output_json"

    ctx.oracle(sorted(paths) == sorted(exp.branch_paths), "changed_paths_equal_generator_branch_paths", reader,

               reported=sorted(paths)[:10], expected=sorted(exp.branch_paths)[:10])

def op_merge_preview(ctx: Ctx) -> None:

    exp = ctx.p.expected

    if _is_git(ctx):

        cap = ctx.timed(["merge-tree", "--write-tree", "main", exp.branch], ctx.work)

        tree = (cap.stdout_full_text or "").split("\n", 1)[0].strip()

        bad = []

        for rel, data in exp.branch_bytes.items():

            got = ctx.git(["cat-file", "blob", f"{tree}:{rel}"], ctx.work, check=False).stdout

            if got != data:

                bad.append(rel)

        ctx.oracle(bool(tree) and not bad, "clean_prediction_tree_has_branch_bytes", "git_fixture_instrument",

                   mismatched=bad[:10])

        ctx.outcome.semantic_class = "git_merge_tree_two_refs"

        return

    ctx.untimed(["switch", exp.branch], ctx.work)

    try:

        # Exit 7 = prediction computed but not certified against the remote

        # target head (documented exit code); the prediction itself is timed.

        cap = ctx.timed(["merge", "--dry-run", "--json"], ctx.work, accept=(0, 7))

    finally:

        ctx.untimed(["switch", _base_branch(ctx)], ctx.work)

    try:

        data = json.loads((cap.stdout_full_text or "").splitlines()[0])

    except (ValueError, IndexError):

        data = {}

    changed = sorted(item.get("path") if isinstance(item, dict) else item for item in data.get("changed_files", []))

    ctx.outcome.semantic_class = "oak_merge_dry_run_current_into_main"

    ctx.outcome.extra["merge_certified"] = (data.get("merge_safety") or {}).get("certified")

    ctx.oracle(data.get("clean") is True and changed == sorted(exp.branch_paths),

               "clean_prediction_lists_branch_paths", "subject_output_json", clean=data.get("clean"),

               changed=changed[:10], expected=sorted(exp.branch_paths)[:10])

def op_merge_preview_conflict(ctx: Ctx) -> None:

    conflict = ctx.p.expected.conflict

    if not conflict:

        raise OpSkipped("fixture_has_no_conflict_shape")

    if not _is_git(ctx):

        raise OpSkipped("oak_merge_dry_run_predicts_current_branch_into_main_only")

    cap = ctx.timed(["merge-tree", "--write-tree", "--name-only", conflict["ours"], conflict["theirs"]],

                    ctx.work, accept=(1,))

    lines = (cap.stdout_full_text or "").split("\n\n", 1)[0].splitlines()[1:]

    ctx.oracle(len(lines) == conflict["paths"], "conflicted_path_count_equals_fixture", "subject_output",

               conflicted=len(lines), expected=conflict["paths"])

def op_merge_local(ctx: Ctx) -> None:

    if not _is_git(ctx):

        raise OpSkipped("oak_local_merge_needs_non_main_parent_unreachable_from_cli; "

                        "oak_serve_lacks_merge_to_main")

    exp = ctx.p.expected

    scratch = f"merge-{ctx.tag}"

    ctx.git(["switch", "-q", "-c", scratch, "main"], ctx.work)

    try:

        ctx.timed(["merge", "-q", "--no-ff", "--no-edit", exp.branch], ctx.work)

        parents = ctx.git(["rev-list", "--parents", "-n", "1", "HEAD"], ctx.work).stdout.split()

        ok, bad = _branch_paths_match(ctx, ctx.work, exp.branch_bytes)

        ctx.oracle(ok and len(parents) == 3, "merge_commit_two_parents_and_branch_bytes", "git_fixture_instrument",

                   parents=len(parents) - 1, mismatched=bad[:10])

    finally:

        ctx.git(["merge", "--abort"], ctx.work, check=False)

        ctx.git(["switch", "-q", "-f", "main"], ctx.work)

        ctx.git(["branch", "-q", "-D", scratch], ctx.work, check=False)

def op_merge_conflict(ctx: Ctx) -> None:

    conflict = ctx.p.expected.conflict

    if not conflict:

        raise OpSkipped("fixture_has_no_conflict_shape")

    if not _is_git(ctx):

        raise OpSkipped("oak_local_merge_needs_non_main_parent_unreachable_from_cli")

    scratch = f"conflict-{ctx.tag}"

    ctx.git(["switch", "-q", "-c", scratch, conflict["ours"]], ctx.work)

    try:

        ctx.timed(["merge", "-q", "--no-edit", conflict["theirs"]], ctx.work, accept=(1,))

        unmerged = ctx.git(["diff", "--name-only", "--diff-filter=U"], ctx.work).stdout.decode().split()

        ctx.oracle(len(unmerged) == conflict["paths"], "unmerged_path_count_equals_fixture", "subject_output",

                   unmerged=len(unmerged), expected=conflict["paths"])

    finally:

        ctx.git(["merge", "--abort"], ctx.work, check=False)

        ctx.git(["switch", "-q", "-f", "main"], ctx.work)

        ctx.git(["branch", "-q", "-D", scratch], ctx.work, check=False)

def _client_commit(ctx: Ctx, client: Path, rel: str, content: bytes) -> str:

    target = client / rel

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

    target.write_bytes(content)

    if _is_git(ctx):

        ctx.git(["add", "-A"], client)

        ctx.git(["commit", "-q", "--no-verify", "-m", f"sync {rel}"], client)

        return ctx.git(["rev-parse", "HEAD"], client).stdout.decode().strip()

    ctx.untimed(["commit", "--json"], client)

    return json.loads(ctx.untimed(["status", "--json"], client).stdout)["head"]

def _head(ctx: Ctx, client: Path) -> str:

    if _is_git(ctx):

        return ctx.git(["rev-parse", "HEAD"], client).stdout.decode().strip()

    return json.loads(ctx.untimed(["status", "--json"], client).stdout)["head"]

def _oak_remote(ctx: Ctx) -> None:

    """Sync ops need seeded loopback servers; label an Oak snapshot seed."""

    if not getattr(ctx.p.tier, "sync", True):

        raise OpSkipped("tier_excludes_sync_clients_disk_budget")

    if _is_git(ctx):

        return

    if getattr(ctx.p, "oak_remote_unavailable", None):

        raise OpSkipped(str(ctx.p.oak_remote_unavailable))

    seed = getattr(ctx.p, "oak_seed", None)

    ctx.outcome.extra["oak_server_seed"] = seed

    if seed and seed != "full_history":

        ctx.outcome.semantic_class = f"oak_server_seed_{seed}"

def op_push_branch(ctx: Ctx) -> None:

    _oak_remote(ctx)

    client_b = ctx.client("clientB")

    rel = f"sync/push-{ctx.tag}.txt"

    head = _client_commit(ctx, client_b, rel, f"push {ctx.tag}\n".encode())

    ctx.state["pushed"] = (rel, head)

    if _is_git(ctx):

        ctx.timed(["push", "-q", "origin", "HEAD:main"], client_b)

        remote = ctx.git(["ls-remote", "origin", "refs/heads/main"], client_b).stdout.decode().split()

        ctx.state["server_extra"] += 1

        ctx.oracle(bool(remote) and remote[0] == head, "server_ref_equals_pushed_head", "git_ls_remote",

                   remote=remote[:1], pushed=head)

    else:

        cap = ctx.timed(["push", "--json"], client_b)

        try:

            receipt = json.loads((cap.stdout_full_text or "").splitlines()[-1])

        except (ValueError, IndexError):

            receipt = {}

        # Server-side confirmation comes from clientA's pull.ff oracle in the

        # same unit (independent checkout sees the pushed commit and bytes).

        ctx.outcome.extra["push_receipt_pushed"] = receipt.get("pushed")

        ctx.oracle(receipt.get("remote_contacted") is True, "push_receipt_remote_contacted_confirmed_by_pull_ff",

                   "subject_output_json", pushed=receipt.get("pushed"))

def op_pull_ff(ctx: Ctx) -> None:

    _oak_remote(ctx)

    pushed = ctx.state.pop("pushed", None)

    if not pushed:

        raise OpFailed("pull.ff needs push.branch in the same unit")

    rel, head = pushed

    client_a = ctx.client("clientA")

    if _is_git(ctx):

        ctx.timed(["pull", "-q", "--ff-only"], client_a)

    else:

        ctx.timed(["pull", "--json"], client_a)

    now = _head(ctx, client_a)

    want = f"push {ctx.tag}\n".encode()

    data_ok = (client_a / rel).is_file() and (client_a / rel).read_bytes() == want

    payloads, reader, meta = committed_bytes(ctx, client_a, [rel])

    committed_ok = payloads.get(rel) == want

    ctx.oracle(now == head and data_ok and committed_ok, "puller_head_worktree_and_committed_bytes_equal_pusher",

               reader, puller_head=now, pusher_head=head, bytes_ok=data_ok, committed_ok=committed_ok, **meta)

def op_pull_diverged(ctx: Ctx) -> None:

    _oak_remote(ctx)

    client_a, client_b = ctx.client("clientA"), ctx.client("clientB")

    local_rel, up_rel = f"sync/local-{ctx.tag}.txt", f"sync/up-{ctx.tag}.txt"

    _client_commit(ctx, client_a, local_rel, b"local\n")

    _client_commit(ctx, client_b, up_rel, b"upstream\n")

    if _is_git(ctx):

        ctx.git(["push", "-q", "origin", "HEAD:main"], client_b)

        ctx.state["server_extra"] += 1

        ctx.timed(["pull", "-q", "--no-rebase", "--no-edit"], client_a)

        ctx.outcome.semantic_class = "git_merge_commit"

    else:

        ctx.untimed(["push", "--json"], client_b)

        cap = ctx.timed(["pull", "--json"], client_a)

        try:

            receipt = json.loads((cap.stdout_full_text or "").splitlines()[-1])

        except (ValueError, IndexError):

            receipt = {}

        ctx.outcome.semantic_class = (ctx.outcome.semantic_class or "") + f"|oak_pull_{receipt.get('convergence', 'unknown')}"

    both = (client_a / local_rel).is_file() and (client_a / up_rel).is_file()

    status_args = ["status", "--porcelain=v1"] if _is_git(ctx) else ["status", "--porcelain"]

    left = porcelain_paths(ctx.untimed(status_args, client_a).stdout.decode(errors="replace"))

    payloads, reader, meta = committed_bytes(ctx, client_a, [local_rel, up_rel])

    committed_ok = payloads.get(local_rel) == b"local\n" and payloads.get(up_rel) == b"upstream\n"

    ctx.oracle(both and not left and committed_ok, "both_sides_committed_present_and_tree_clean", reader,

               local_present=(client_a / local_rel).is_file(), upstream_present=(client_a / up_rel).is_file(),

               committed_ok=committed_ok, left=left[:10], **meta)

    # Re-converge (untimed): A publishes its merge, B catches up. A failure

    # here is an observed subject defect in an UNTIMED step: it is recorded on

    # the row (teardown_error) without discarding the timed sample.

    if _is_git(ctx):

        ctx.git(["push", "-q", "origin", "HEAD:main"], client_a)

        ctx.state["server_extra"] += 1

        ctx.git(["pull", "-q", "--ff-only"], client_b)

        return

    ctx.untimed(["push", "--json"], client_a)

    done = ctx.untimed(["pull", "--json"], client_b, check=False)

    if done.returncode != 0:

        try:

            receipt = json.loads(done.stdout.decode(errors="replace").splitlines()[-1])

        except (ValueError, IndexError):

            receipt = {}

        ctx.outcome.extra["teardown_error"] = (

            f"peer pull after reparent exited {done.returncode}: branch_update={receipt.get('branch_update')} "

            f"parent_sync_error_code={receipt.get('parent_sync_error_code')}")

        if _head(ctx, client_b) != _head(ctx, client_a):

            raise OpFailed("peer checkout did not converge after reparent: " + ctx.outcome.extra["teardown_error"])

def op_clone_full(ctx: Ctx) -> None:

    _oak_remote(ctx)

    dest = ctx.scratch / f"clone-{ctx.tag}"

    if dest.exists():

        shutil.rmtree(dest)

    try:

        if _is_git(ctx):

            ctx.timed(["clone", "-q", str(ctx.p.git_url), str(dest)], ctx.scratch)

            # server main = fixture + files pushed to main by the sync ops

            expected = ctx.p.expected.main_file_count + ctx.state["server_extra"]

            extra_prefix: Tuple[str, ...] = ()

        else:

            ctx.timed(["clone", "-r", str(ctx.p.oak_url), "bench/fx", str(dest)], ctx.scratch)

            expected = ctx.p.expected.main_file_count  # oak sync ops publish to a branch, not main

            # (a snapshot seed has the same main tree, one commit of history)

            extra_prefix = ()

        ok, detail = check_worktree(ctx, dest, expected, extra_prefix)

        ctx.oracle(ok, "clone_file_count_and_sampled_blob_ids_equal_fixture", "git_fixture_instrument", **detail)

    finally:

        shutil.rmtree(dest, ignore_errors=True)

def op_gc(ctx: Ctx) -> None:

    status_args = ["status", "--porcelain=v1"] if _is_git(ctx) else ["status", "--porcelain"]

    if _is_git(ctx):

        before = ctx.git(["rev-list", "--count", "HEAD"], ctx.work).stdout.strip()

        ctx.timed(["gc", "--quiet"], ctx.work)

        after = ctx.git(["rev-list", "--count", "HEAD"], ctx.work).stdout.strip()

        ctx.outcome.semantic_class = "git_gc"

    else:

        before = ctx.untimed(["hash"], ctx.work).stdout.strip()

        ctx.timed(["maintenance", "compact"], ctx.work)

        after = ctx.untimed(["hash"], ctx.work).stdout.strip()

        ctx.outcome.semantic_class = "oak_maintenance_compact"

    left = porcelain_paths(ctx.untimed(status_args, ctx.work).stdout.decode(errors="replace"))

    probe = ctx.p.expected.probe_path

    payloads, reader, meta = committed_bytes(ctx, ctx.work, [probe])

    probe_ok = payloads.get(probe) == ctx.p.expected.probe_bytes

    ctx.oracle(before == after and not left and probe_ok, "head_status_and_committed_probe_unchanged",

               reader, before=before.decode() if isinstance(before, bytes) else before,

               after=after.decode() if isinstance(after, bytes) else after, left=left[:10],

               committed_probe_ok=probe_ok, **meta)

OPS: Dict[str, Callable[[Ctx], None]] = {

    "status.clean": op_status_clean,

    "status.dirty": op_status_dirty,

    "diff.dirty": op_diff_dirty,

    "commit.dirty": op_commit_dirty,

    "log.full": op_log_full,

    "log.n": op_log_n,

    "log.path": op_log_path,

    "blame": op_blame,

    "file.show": op_file_show,

    "branch.list": op_branch_list,

    "branch.create": op_branch_create,

    "switch.existing": op_switch_existing,

    "switch.back": op_switch_back,

    "review.branch_diff": op_review_branch_diff,

    "merge.preview": op_merge_preview,

    "merge.preview.conflict": op_merge_preview_conflict,

    "merge.local": op_merge_local,

    "merge.conflict": op_merge_conflict,

    "push.branch": op_push_branch,

    "pull.ff": op_pull_ff,

    "pull.diverged": op_pull_diverged,

    "clone.full": op_clone_full,

    "gc": op_gc,

}

assert set(OPS) == set(OP_INFO)

def catalog_yaml() -> str:

    """scenarios/scale.yaml, generated from this catalog (a test keeps them equal)."""

    lines = [

        "schema_version: 1",

        "scenario_set: scale",

        "description: Scale and coverage lane (suites/scale). Every user-facing VCS operation timed for git and oak"

        " across repository tiers. Generated from suites/scale/scalesuite/ops.py (catalog_yaml); "

        "tests/test_scale_suite.py keeps the two equal.",

        "",

        "operation_catalog:",

    ]

    for unit in UNIT_ORDER:

        for op in unit:

            info = OP_INFO[op]

            lines += [f"  {op}:", f"    description: {info.description}", "    timed: true",

                      f"    mutates: {str(info.mutates).lower()}"]

            for key, values in (("tperf_refs", info.tperf), ("core_yaml_ops", info.core_yaml)):

                if values:

                    lines.append(f"    {key}:")

                    lines.extend(f"      - {value}" for value in values)

                else:

                    lines.append(f"    {key}: []")

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

def execute(ctx: Ctx, op: str) -> Outcome:

    try:

        OPS[op](ctx)

    except OpSkipped as exc:

        ctx.outcome.skip_reason = str(exc)

    except OpFailed as exc:

        ctx.outcome.failed = str(exc)

    except Exception as exc:  # noqa: BLE001 - harness errors are recorded, never hidden

        ctx.outcome.failed = f"harness_error: {type(exc).__name__}: {str(exc)[-800:]}"

        ctx.outcome.extra["harness_error"] = True

    return ctx.outcome