"""Instrument tests for suites/scale (generator, sizer, tiers, op oracles, rows).

Hermetic: no Oak binary and no network. Git-backed tests use the resolved

real git and skip when it is absent. Oak-side oracles are exercised with

canned subject output (a stub timed runner), so every op's untimed oracle is

tested for both a passing and a failing observation.

"""

from __future__ import annotations

import hashlib

import io

import json

import re

import shutil

import subprocess

import sys

import tempfile

import unittest

from pathlib import Path

from types import SimpleNamespace

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

sys.path.insert(0, str(ROOT / "suites" / "scale"))

sys.path.insert(0, str(ROOT / "scripts"))

from oakbench.execution import TimedCapture, run_timed  # noqa: E402

from oakbench.rows import validate_rows  # noqa: E402

from scalesuite import generator as gen  # noqa: E402

from scalesuite import ops as ops_mod  # noqa: E402

from scalesuite import runner as runner_mod  # noqa: E402

from scalesuite import sizer as sz  # noqa: E402

from scalesuite import tiers as tiers_mod  # noqa: E402

from scalesuite import workspace as ws  # noqa: E402

try:

    from oakbench.subjects import which_measured_git

    GIT = which_measured_git()

except Exception:  # pragma: no cover

    GIT = shutil.which("git")

SMALL = gen.GenSpec(name="t", seed="unit-seed", files=60, file_bytes=96, depth=4, fanout=4, commits=30,

                    touches_per_commit=3, branches=4, branch_commit_files=2, tags=3, binaries=1,

                    binary_bytes=(2 << 20) + 17, binary_revisions=2, ignore_patterns=("build/",),

                    conflict_files=3)

def stream_sha(spec: gen.GenSpec) -> str:

    buffer = io.BytesIO()

    gen.write_stream(spec, buffer)

    return hashlib.sha256(buffer.getvalue()).hexdigest()

def git(*args: str, cwd: Path) -> str:

    return subprocess.run([GIT, *args], cwd=cwd, check=True, capture_output=True, text=True,

                          env=gen.git_env()).stdout

class GeneratorDeterminismTest(unittest.TestCase):

    def test_same_spec_same_stream(self) -> None:

        self.assertEqual(stream_sha(SMALL), stream_sha(SMALL))

    def test_seed_changes_stream_and_spec_hash(self) -> None:

        other = gen.GenSpec(**{**SMALL.to_json(), "seed": "other-seed", "ignore_patterns": ("build/",)})

        self.assertNotEqual(stream_sha(SMALL), stream_sha(other))

        self.assertNotEqual(SMALL.spec_sha256(), other.spec_sha256())

    def test_spec_hash_is_stable_text(self) -> None:

        self.assertEqual(SMALL.spec_sha256(), gen.GenSpec(**{**SMALL.to_json(), "ignore_patterns": ("build/",)}).spec_sha256())

        self.assertRegex(SMALL.spec_sha256(), r"^[0-9a-f]{64}$")

    def test_validation(self) -> None:

        with self.assertRaises(ValueError):

            gen.GenSpec(name="x", seed="s", files=0).validate()

        with self.assertRaises(ValueError):

            gen.GenSpec(name="x", seed="s", binary_revisions=2, commits=3).validate()

        with self.assertRaises(ValueError):

            gen.GenSpec(name="x", seed="s", file_bytes=8).validate()

    def test_every_path_has_exact_depth_and_is_unique(self) -> None:

        for spec in (SMALL, gen.GenSpec(name="d", seed="s", files=5000, depth=40, fanout=8),

                     gen.GenSpec(name="w", seed="s", files=5000, depth=2, fanout=10),

                     gen.GenSpec(name="r", seed="s", files=7, depth=1, fanout=4)):

            cache = gen.layout(spec)

            paths = [gen.file_path(spec, i, cache) for i in range(spec.files)]

            self.assertEqual(len(set(paths)), spec.files)

            self.assertTrue(all(p.count("/") + 1 == spec.depth for p in paths), spec)

    def test_content_unique_sized_and_deterministic(self) -> None:

        factory = gen.ContentFactory(SMALL)

        blobs = {factory.text(i, "r1") for i in range(SMALL.files)}

        self.assertEqual(len(blobs), SMALL.files)

        self.assertTrue(all(len(b) == SMALL.file_bytes for b in blobs))

        self.assertEqual(factory.text(3, "r9"), gen.ContentFactory(SMALL).text(3, "r9"))

        self.assertNotEqual(factory.text(3, "r9"), factory.text(3, "r8"))

    def test_binary_is_sparse_friendly_and_exact_size(self) -> None:

        factory = gen.ContentFactory(SMALL)

        data = b"".join(factory.binary_segments(0, 1))

        self.assertEqual(len(data), SMALL.binary_bytes)

        self.assertEqual(data[gen.SEGMENT_RANDOM:gen.SEGMENT], bytes(gen.SEGMENT - gen.SEGMENT_RANDOM))

        self.assertNotEqual(data[:gen.SEGMENT_RANDOM], bytes(gen.SEGMENT_RANDOM))

        sha1, _ = gen.binary_sha1_and_sha256(factory, 0, 1)

        self.assertEqual(sha1, gen.git_blob_id(data))

    def test_branch_changes_match_stream_facts(self) -> None:

        buffer = io.BytesIO()

        facts = gen.write_stream(SMALL, buffer)

        self.assertEqual(facts["branch_sample"]["changed_paths"],

                         sorted(path for _, path in gen.branch_changes(SMALL, 0)))

        self.assertEqual(facts["total_commits"], SMALL.commits + SMALL.branches + 2)

@unittest.skipUnless(GIT, "git not available")

class GeneratedFixtureTest(unittest.TestCase):

    @classmethod

    def setUpClass(cls) -> None:

        cls.tmp = Path(tempfile.mkdtemp(prefix="scale-gen-"))

        cls.manifest = gen.generate(SMALL, cls.tmp / "a", GIT)

        cls.repo = cls.tmp / "a" / "src.git"

    @classmethod

    def tearDownClass(cls) -> None:

        shutil.rmtree(cls.tmp, ignore_errors=True)

    def test_same_seed_same_fixture_hash(self) -> None:

        again = gen.generate(SMALL, self.tmp / "b", GIT)

        self.assertEqual(again["refs_sha256"], self.manifest["refs_sha256"])

        self.assertEqual(again["head_commit"], self.manifest["head_commit"])

    def test_reuse_is_detected(self) -> None:

        self.assertTrue(gen.generate(SMALL, self.tmp / "a", GIT)["reused"])

    def test_manifest_facts_agree_with_git(self) -> None:

        probe = self.manifest["probe_path"]

        log = git("log", "--oneline", "main", "--", probe, cwd=self.repo).splitlines()

        self.assertEqual(len(log), self.manifest["probe_path_commit_count"])

        self.assertEqual(int(git("rev-list", "--count", "main", cwd=self.repo)), SMALL.commits)

        changed = git("diff", "--name-only", "main...b00000", cwd=self.repo).split()

        self.assertEqual(sorted(changed), self.manifest["branch_sample"]["changed_paths"])

        self.assertEqual(len(git("tag", "--list", cwd=self.repo).split()), SMALL.tags)

        conflicts = subprocess.run([GIT, "merge-tree", "--write-tree", "--name-only", "bench-ours", "bench-theirs"],

                                   cwd=self.repo, capture_output=True, text=True)

        self.assertEqual(conflicts.returncode, 1)

        self.assertEqual(len(conflicts.stdout.split("\n\n")[0].splitlines()[1:]), SMALL.conflict_files)

    def test_binary_blob_id_matches_generator(self) -> None:

        blob = git("rev-parse", f"b00000:{gen.binary_path(0)}", cwd=self.repo).strip()

        # branch b00000 forks from the tip, where the binary is at revision binary_revisions

        sha1, _ = gen.binary_sha1_and_sha256(gen.ContentFactory(SMALL), 0, SMALL.binary_revisions)

        self.assertEqual(blob, sha1)

    def test_sizer_values(self) -> None:

        stats = sz.measure(self.repo, GIT)

        values = stats["values"]

        self.assertEqual(stats["checkout_scope"], "all_commits")

        self.assertEqual(values["uniqueCommitCount"], self.manifest["total_commits"])

        self.assertEqual(values["maxCheckoutPathDepth"], SMALL.depth)

        self.assertEqual(values["maxCheckoutBlobCount"], SMALL.files + SMALL.binaries + 1)  # + .gitignore

        self.assertEqual(values["referenceCount"], 1 + SMALL.branches + 2 + SMALL.tags)

        self.assertEqual(values["maxBlobSize"], SMALL.binary_bytes)

        self.assertEqual(values["maxHistoryDepth"], SMALL.commits + 1)

        self.assertEqual(values["maxCommitParentCount"], 1)

        self.assertEqual(values["uniqueTagCount"], 0)

        longest = max(len(p) for p in git("ls-tree", "-r", "--name-only", "main", cwd=self.repo).split())

        self.assertEqual(values["maxCheckoutPathLength"], longest)

        self.assertEqual(stats["level_of_concern"]["maxCheckoutPathDepth"],

                         round(SMALL.depth / 10, 4))

        compact = sz.compact(stats)

        self.assertEqual(compact["maxCheckoutPathDepth"], SMALL.depth)

    def test_sizer_ref_tips_scope_and_tag_depth(self) -> None:

        clone = self.tmp / "tags.git"

        subprocess.run([GIT, "clone", "-q", "--bare", str(self.repo), str(clone)], check=True, env=gen.git_env())

        env = {**gen.git_env(), "GIT_COMMITTER_NAME": "t", "GIT_COMMITTER_EMAIL": "t@t"}

        subprocess.run([GIT, "-C", str(clone), "tag", "-a", "-m", "one", "a1", "main"], check=True, env=env)

        subprocess.run([GIT, "-C", str(clone), "tag", "-a", "-m", "two", "a2", "a1"], check=True, env=env)

        stats = sz.measure(clone, GIT, max_checkout_roots=2)

        self.assertEqual(stats["checkout_scope"], "ref_tips")

        self.assertEqual(stats["values"]["maxTagDepth"], 2)

        self.assertEqual(stats["values"]["uniqueTagCount"], 2)

        self.assertIsNone(stats["values"]["uniqueTreeEntries"])

@unittest.skipUnless(GIT, "git not available")

class RegistryPinTest(unittest.TestCase):

    """Cross-machine determinism: regenerating a tier reproduces the pinned identity."""

    def test_pinned_tiers_regenerate_identically(self) -> None:

        registry = json.loads((ROOT / "config" / "fixtures.json").read_text())["fixtures"]

        tmp = Path(tempfile.mkdtemp(prefix="scale-pin-"))

        try:

            checked = 0

            for name in ("fresh", "default"):

                entry = registry[f"scale-{name}-v1"]

                tier = tiers_mod.TIERS[name]

                self.assertEqual(entry["spec_sha256"], tier.spec.spec_sha256())

                self.assertEqual(entry["seed"], tier.spec.seed)

                manifest = gen.generate(tier.spec, tmp / name, GIT)

                self.assertEqual(manifest["refs_sha256"], entry["refs_sha256"], name)

                self.assertEqual(manifest["head_commit"], entry["head_commit"], name)

                checked += 1

            self.assertEqual(checked, 2)

        finally:

            shutil.rmtree(tmp, ignore_errors=True)

    def test_every_tier_is_registered_with_its_spec_hash(self) -> None:

        registry = json.loads((ROOT / "config" / "fixtures.json").read_text())["fixtures"]

        for name, tier in tiers_mod.TIERS.items():

            self.assertEqual(registry[f"scale-{name}-v1"]["spec_sha256"], tier.spec.spec_sha256(), name)

class SizerPureTest(unittest.TestCase):

    def test_parse_tree(self) -> None:

        oid = bytes(range(20))

        raw = b"100644 a.txt\0" + oid + b"40000 dir\0" + oid[::-1]

        self.assertEqual(sz.parse_tree(raw), [(b"100644", b"a.txt", oid), (b"40000", b"dir", oid[::-1])])

    def test_level_of_concern_uses_git_sizer_reference_values(self) -> None:

        self.assertEqual(sz.level_of_concern("maxCheckoutBlobCount", 1_000_000), 20.0)

        self.assertEqual(sz.level_of_concern("referenceCount", 25_000), 1.0)

        self.assertIsNone(sz.level_of_concern("maxBlobSize", None))

class TierTest(unittest.TestCase):

    def test_all_14_core_shapes_are_runnable_specs(self) -> None:

        shapes = tiers_mod.core_shape_tiers(0.05)

        self.assertEqual(len(shapes), 14)

        for shape_id, tier in shapes.items():

            tier.spec.validate()

            self.assertGreaterEqual(tier.spec.branches, 1, shape_id)

            self.assertEqual(tier.kind, "shape")

        self.assertEqual(shapes["deep-tree"].spec.depth, 64)

        self.assertEqual(shapes["conflict-heavy"].spec.conflict_files, 250)

        self.assertEqual(shapes["binary-history"].spec.binary_revisions, shapes["binary-history"].spec.commits)

        self.assertIn("build/", shapes["ignored-heavy"].spec.ignore_patterns)

    def test_scale_is_part_of_identity(self) -> None:

        a = tiers_mod.core_shape_tiers(0.05)["normal"]

        b = tiers_mod.core_shape_tiers(0.1)["normal"]

        self.assertNotEqual(a.fixture_id, b.fixture_id)

    def test_tier_ids_unique_and_xl_opt_in(self) -> None:

        ids = [tier.fixture_id for tier in tiers_mod.TIERS.values()]

        self.assertEqual(len(ids), len(set(ids)))

        with self.assertRaises(SystemExit):

            tiers_mod.resolve("xl-tree")

        self.assertEqual(tiers_mod.resolve("xl-tree", allow_xl=True).spec.files, 1_000_000)

        with self.assertRaises(SystemExit):

            tiers_mod.resolve("xl-combined", allow_xl=True)

        combined = tiers_mod.resolve("xl-combined", allow_xl=True, allow_xl_combined=True).spec

        self.assertGreaterEqual(combined.commits, 100_000)

        self.assertGreaterEqual(combined.files, 1_000_000)

        self.assertGreaterEqual(combined.depth, 32)

        self.assertGreaterEqual(combined.branches, 10_000)

        self.assertGreaterEqual(combined.binary_bytes, 2 << 30)

        for tier in tiers_mod.TIERS.values():

            tier.spec.validate()

class CatalogTest(unittest.TestCase):

    def test_scale_yaml_matches_catalog(self) -> None:

        self.assertEqual((ROOT / "scenarios" / "scale.yaml").read_text(), ops_mod.catalog_yaml())

    def test_tperf_refs_are_file_names_only(self) -> None:

        pattern = re.compile(r"^p\d{4}-[a-z0-9-]+\.sh$")

        for info in ops_mod.OP_INFO.values():

            for ref in info.tperf:

                self.assertRegex(ref, pattern)

    def test_units_cover_ops(self) -> None:

        self.assertEqual(sorted(ops_mod.ALL_OPS), sorted(ops_mod.OPS))

    def test_porcelain_paths(self) -> None:

        text = ' M a/b.txt\n D c.txt\nR  old.txt -> new.txt\n?? added/x/\nM "sp ace.txt"\n'

        self.assertEqual(ops_mod.porcelain_paths(text),

                         ["a/b.txt", "c.txt", "old.txt", "new.txt", "added/x/", "sp ace.txt"])

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

# Oracle tests with canned subject output

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

def cap(stdout: str = "", returncode: int = 0, raw: bytes | None = None) -> TimedCapture:

    data = stdout.encode()

    return TimedCapture(elapsed_ms=1.0, returncode=returncode, stdout_text=stdout[:4000], stdout_bytes=len(data),

                        stdout_truncated=False, stderr_text="", stderr_bytes=0, stderr_truncated=False,

                        stdout_full_text=stdout, stdout_raw=raw if raw is not None else data)

class OracleTest(unittest.TestCase):

    def setUp(self) -> None:

        self.tmp = Path(tempfile.mkdtemp(prefix="scale-oracle-"))

        self.spec = gen.GenSpec(name="o", seed="oracle", files=40, file_bytes=64, depth=3, fanout=4, commits=12,

                                branches=2, branch_commit_files=2)

        facts = gen.write_stream(self.spec, io.BytesIO())

        factory = gen.ContentFactory(self.spec)

        for kind in ("git", "oak"):

            work = self.tmp / kind / "work"

            for i in range(self.spec.files):

                path = work / gen.file_path(self.spec, i)

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

                path.write_bytes(factory.text(i, "r1"))

        branch_paths = facts["branch_sample"]["changed_paths"]

        self.expected = ws.Expected(

            main_file_count=self.spec.files, sample={}, probe_path=facts["probe_path"], probe_bytes=b"probe\nx\n",

            probe_lines=2, probe_commit_count=facts["probe_path_commit_count"], main_commits=self.spec.commits,

            branch="b00000", branch_paths=branch_paths, branch_bytes={}, main_bytes_of_branch_paths={},

            conflict={}, protected=[facts["probe_path"], *branch_paths], all_paths_sample_size=0)

        tier = tiers_mod.Tier(tier_id="t", kind="tier", spec=self.spec, dirty=tiers_mod.DirtyShape(3, 2, 1, 1))

        root = self.tmp

        self.prepared = SimpleNamespace(tier=tier, expected=self.expected, state={}, env={}, git_bin=GIT or "git",

                                        oak_base_branch="base", oak_branches_created=2,

                                        path=lambda *parts: root.joinpath(*parts))

        self.untimed_calls: list = []

    def tearDown(self) -> None:

        shutil.rmtree(self.tmp, ignore_errors=True)

    def ctx(self, kind: str, *outputs, run: int = 1) -> ops_mod.Ctx:

        queue = list(outputs)

        subject = SimpleNamespace(name=kind, kind=kind, bin=kind)

        def fake_runner(command, cwd, env, admitted, **kwargs):

            item = queue.pop(0)

            return item() if callable(item) else item

        ctx = ops_mod.Ctx(self.prepared, subject, run, fake_runner, self.tmp / "scratch", 60.0)

        ctx.untimed = lambda args, cwd, check=True, binary=None: (  # type: ignore[assignment]

            self.untimed_calls.append(list(args)) or SimpleNamespace(returncode=0, stdout=b"", stderr=b""))

        ctx.git = lambda args, cwd, check=True: (  # type: ignore[assignment]

            self.untimed_calls.append(["git", *args]) or SimpleNamespace(returncode=0, stdout=b"", stderr=b""))

        return ctx

    def run_op(self, op: str, kind: str, *outputs: TimedCapture) -> ops_mod.Outcome:

        return ops_mod.execute(self.ctx(kind, *outputs), op)

    def test_status_clean(self) -> None:

        self.assertTrue(self.run_op("status.clean", "git", cap("")).oracle_ok)

        self.assertFalse(self.run_op("status.clean", "oak", cap("M a.txt\n")).oracle_ok)

    def test_log_counts(self) -> None:

        lines = "".join(f"h{i} c{i}\n" for i in range(self.spec.commits))

        self.assertTrue(self.run_op("log.full", "git", cap(lines)).oracle_ok)

        self.assertFalse(self.run_op("log.full", "oak", cap(lines + "extra\n")).oracle_ok)

        self.assertTrue(self.run_op("log.n", "oak", cap(lines)).oracle_ok)

        self.assertFalse(self.run_op("log.n", "oak", cap("h1 c1\n")).oracle_ok)

        probe = "".join(f"p{i}\n" for i in range(self.expected.probe_commit_count))

        self.assertTrue(self.run_op("log.path", "git", cap(probe)).oracle_ok)

        self.assertFalse(self.run_op("log.path", "oak", cap("No matching commits\n")).oracle_ok)

    def test_blame(self) -> None:

        self.assertTrue(self.run_op("blame", "git", cap("a 1) probe\nb 2) x\n")).oracle_ok)

        self.assertFalse(self.run_op("blame", "git", cap("a 1) probe\n")).oracle_ok)

        self.assertEqual(self.run_op("blame", "oak").skip_reason, "oak_cli_has_no_blame_or_annotate_command")

    def test_file_show(self) -> None:

        self.assertTrue(self.run_op("file.show", "git", cap(raw=b"probe\nx\n")).oracle_ok)

        self.assertFalse(self.run_op("file.show", "git", cap(raw=b"other\n")).oracle_ok)

        # oak writes --output; nothing written -> mismatch

        self.assertFalse(self.run_op("file.show", "oak", cap("{}")).oracle_ok)

    def test_branch_list(self) -> None:

        self.assertTrue(self.run_op("branch.list", "git", cap("* main\n  b00000\n  b00001\n")).oracle_ok)

        self.assertFalse(self.run_op("branch.list", "git", cap("* main\n")).oracle_ok)

        listing = json.dumps([{"name": "main"}, {"name": "b00000"}, {"name": "b00001"}])

        self.assertTrue(self.run_op("branch.list", "oak", cap(listing)).oracle_ok)

        self.assertFalse(self.run_op("branch.list", "oak", cap("not json")).oracle_ok)

    def test_review_branch_diff(self) -> None:

        paths = self.expected.branch_paths

        self.assertTrue(self.run_op("review.branch_diff", "git",

                                    cap("".join(f"M\t{p}\n" for p in paths))).oracle_ok)

        oak_ok = json.dumps({"changed_files": [{"path": p} for p in paths]})

        self.assertTrue(self.run_op("review.branch_diff", "oak", cap(oak_ok)).oracle_ok)

        self.assertFalse(self.run_op("review.branch_diff", "oak", cap(json.dumps({"changed_files": []}))).oracle_ok)

    def test_merge_preview_oak(self) -> None:

        good = json.dumps({"clean": True, "changed_files": [{"path": p} for p in self.expected.branch_paths],

                           "merge_safety": {"certified": False}})

        outcome = self.run_op("merge.preview", "oak", cap(good, returncode=7))

        self.assertTrue(outcome.oracle_ok)

        self.assertFalse(outcome.extra["merge_certified"])

        self.assertIn(["switch", "b00000"], self.untimed_calls)

        self.assertEqual(self.untimed_calls[-1], ["switch", "base"])  # always switched back

        bad = json.dumps({"clean": False, "changed_files": []})

        self.assertFalse(self.run_op("merge.preview", "oak", cap(bad)).oracle_ok)

        self.assertIsNotNone(self.run_op("merge.preview", "oak", cap(good, returncode=2)).failed)

    def test_local_merge_ops_skip_for_oak(self) -> None:

        self.assertIn("oak_serve_lacks_merge_to_main", self.run_op("merge.local", "oak").skip_reason)

        self.assertEqual(self.run_op("merge.conflict", "git").skip_reason, "fixture_has_no_conflict_shape")

    def _listing(self, kind: str = "git") -> str:

        state = self.prepared.state[kind]["dirty"]

        text = "".join(f" M {p}\n" for p in state["modified"]) + "".join(f" D {p}\n" for p in state["deleted"])

        return text + "".join(f" D {old}\n" for old, _ in state["renamed"]) + "?? added/\n?? renamed/\n"

    def test_dirty_cycle_status_and_diff_oracles(self) -> None:

        good = ops_mod.execute(self.ctx("git", lambda: cap(self._listing())), "status.dirty")

        self.assertTrue(good.oracle_ok, good.oracle_detail)

        dirty = self.prepared.state["git"]["dirty"]

        self.assertEqual(dirty["applied"], [3, 2, 1, 1])

        self.assertFalse(set(dirty["modified"]) & set(self.expected.protected))

        diff = "".join(f"diff --git a/{p} b/{p}\n" for p in dirty["modified"] + dirty["deleted"])

        self.assertTrue(ops_mod.execute(self.ctx("git", cap(diff)), "diff.dirty").oracle_ok)

        self.assertFalse(ops_mod.execute(self.ctx("git", cap("diff --git a/x b/x\n")), "diff.dirty").oracle_ok)

        partial = ops_mod.execute(self.ctx("oak", lambda: cap(self._listing().split("\n", 1)[1]), run=2),

                                  "status.dirty")

        self.assertFalse(partial.oracle_ok)

    def test_dirty_cycle_runs_on_throwaway_branch(self) -> None:

        ops_mod.execute(self.ctx("oak", lambda: cap(self._listing("oak"))), "status.dirty")

        self.assertIn(["switch", "-c", "dirty-oak-r1"], self.untimed_calls)

        self.untimed_calls.clear()

        out = ops_mod.execute(self.ctx("oak", cap("{}")), "commit.dirty")

        self.assertIsNone(out.failed, out.failed)

        self.assertEqual(self.untimed_calls[-1], ["switch", "base"])

        self.assertEqual(self.prepared.state["oak"]["harness_commits"], 0)

    def test_recover_returns_to_base(self) -> None:

        ctx = self.ctx("oak")

        self.assertIsNone(ops_mod.recover(ctx))

        self.assertEqual(self.untimed_calls[-1], ["switch", "base"])

    def test_oak_state_oracles_use_independent_reader(self) -> None:

        from unittest import mock

        from oakbench import oak_sqlite_reader as reader

        probe = self.expected.probe_path

        self.prepared.state["oak"] = {"harness_commits": 0, "deleted": set(), "server_extra": 0, "branch_seq": 0}

        with mock.patch.object(reader, "committed_payloads",

                               return_value=({probe: self.expected.probe_bytes}, {"head": "h", "reader_version": "v"})):

            outcome = ops_mod.execute(self.ctx("oak", cap("")), "gc")

        self.assertEqual(outcome.oracle_reader, "oak_sqlite_reader_independent")

        self.assertTrue(outcome.oracle_ok, outcome.oracle_detail)

        with mock.patch.object(reader, "committed_payloads", return_value=({probe: b"torn"}, {"head": "h"})):

            self.assertFalse(ops_mod.execute(self.ctx("oak", cap("")), "gc").oracle_ok)

        with mock.patch.object(reader, "committed_payloads", side_effect=reader.Unavailable("database too big")):

            unmeasured = ops_mod.execute(self.ctx("oak", cap("")), "gc")

        self.assertIsNone(unmeasured.oracle_ok)

        self.assertIsNone(unmeasured.failed)

        self.assertIn("reader_unavailable", unmeasured.oracle_detail)

    def test_commit_needs_dirty_state(self) -> None:

        self.assertIn("needs status.dirty", self.run_op("commit.dirty", "git").failed)

    def test_timed_failure_is_recorded_not_raised(self) -> None:

        outcome = self.run_op("status.clean", "oak", cap("", returncode=3))

        self.assertIn("exit 3", outcome.failed)

        self.assertIsNone(outcome.oracle_ok)

    def test_check_worktree(self) -> None:

        root = self.tmp / "git" / "work"

        path = gen.file_path(self.spec, 5)

        self.expected.sample[path] = gen.git_blob_id((root / path).read_bytes())

        ctx = self.ctx("git")

        ok, detail = ops_mod.check_worktree(ctx, root, self.spec.files)

        self.assertTrue(ok, detail)

        (root / path).write_bytes(b"changed")

        ok, detail = ops_mod.check_worktree(ctx, root, self.spec.files)

        self.assertFalse(ok)

        self.assertEqual(detail["sample_mismatch_count"], 1)

class RowTest(unittest.TestCase):

    def prepared(self) -> SimpleNamespace:

        tier = tiers_mod.TIERS["fresh"]

        fields = {"tier": "fresh", "tier_kind": "tier", "fixture_id": tier.fixture_id,

                  "fixture_manifest_sha256": "a" * 64, "fixture_spec_sha256": "b" * 64}

        return SimpleNamespace(tier=tier, fixture_fields=lambda: fields, phases={"ingest.oak_import": 1.5},

                               notes=[], manifest={"refs_sha256": "a" * 64, "spec_sha256": "b" * 64})

    def row(self, subject: str, kind: str, op: str, run: int, outcome: ops_mod.Outcome) -> dict:

        return runner_mod.build_row(bench_id="b", prepared=self.prepared(),

                                    subject=SimpleNamespace(name=subject, kind=kind), op=op, run=run,

                                    outcome=outcome, details={}, env_sample={}, lock_wait_ms=0.0,

                                    order_position=0, runner_fields={})

    def test_rows_validate_for_all_outcomes(self) -> None:

        ok = ops_mod.Outcome(timed=[ops_mod.Timed("main", ["git", "status"], cap(""))], oracle_ok=True)

        skip = ops_mod.Outcome(skip_reason="no_equivalent")

        fail = ops_mod.Outcome(timed=[ops_mod.Timed("main", ["oak", "status"], cap("", 2))], failed="exit 2")

        bad = ops_mod.Outcome(timed=[ops_mod.Timed("main", ["oak", "log"], cap(""))], oracle_ok=False)

        rows = [self.row("git", "git", "status.clean", 1, ok), self.row("oak_installed", "oak", "blame", 1, skip),

                self.row("oak_installed", "oak", "status.clean", 1, fail),

                self.row("oak_installed", "oak", "log.path", 1, bad)]

        rows.append(runner_mod.abort_row(rows[0], "diff.dirty", "status.dirty"))

        prepared = self.prepared()

        prepared.oak_identity = {"binary_sha256": "c" * 64}

        rows += runner_mod.ingest_rows(prepared, "b", {})

        self.assertEqual(validate_rows(rows, "scale"), [])

        self.assertEqual([r["returncode"] for r in rows[:5]], [0, 77, 1, 1, 77])

        self.assertEqual(rows[3]["failure"], "oracle_failed")

        self.assertFalse(any(r["claim_eligible"] for r in rows))

        self.assertTrue(all(r["oracle_timed"] is False for r in rows[:4]))

        import publish_gate

        for row in rows:

            if row["subject_kind"] == "oak" and row["operation"].startswith("ingest."):

                self.assertEqual(publish_gate.oak_binary_identity(row), "c" * 64)

    def test_cell_report_gives_floors_and_mde(self) -> None:

        rows = []

        for run in range(1, 7):

            for subject, kind, base in (("git", "git", 10.0), ("git_aa", "git", 10.0),

                                        ("oak_installed", "oak", 20.0), ("oak_installed_aa", "oak", 20.0)):

                outcome = ops_mod.Outcome(timed=[ops_mod.Timed("main", [kind], cap(""))], oracle_ok=True)

                row = self.row(subject, kind, "status.clean", run, outcome)

                row["elapsed_ms"] = base + (run % 3) * 0.2

                rows.append(row)

        report = runner_mod.cell_report(rows, ["git", "git_aa", "oak_installed", "oak_installed_aa"])

        item = report[0]

        self.assertIsNotNone(item["git_aa_floor_pct"])

        self.assertIsNotNone(item["oak_vs_git_mde_pct_80"])

        self.assertIsNotNone(item["oak_vs_oak_mde_pct_80"])

        self.assertAlmostEqual(item["oak_vs_git_delta_pct_dev_only"], 100.0, delta=5)

        self.assertFalse(item["claim_eligible"])

        text = runner_mod.summary_markdown(self.prepared(), report,

                                           {"bench_id": "b", "runs": 6, "subjects": ["git"]})

        self.assertIn("DEVELOPMENT ONLY", text)

        self.assertRegex(text, r"\| 1\.9\dx \|")

@unittest.skipUnless(GIT, "git not available")

class GitOnlyScheduleTest(unittest.TestCase):

    """End to end on git only: prepare, interleave git + git_aa, every oracle passes."""

    def test_schedule(self) -> None:

        from oakbench.environment import base_env

        tmp = Path(tempfile.mkdtemp(prefix="scale-e2e-"))

        try:

            spec = gen.GenSpec(name="e2e", seed="e2e", files=24, file_bytes=64, depth=3, fanout=4, commits=9,

                               branches=2, conflict_files=2)

            tier = tiers_mod.Tier(tier_id="e2e", kind="tier", spec=spec, dirty=tiers_mod.DirtyShape(2, 1, 1, 1))

            env = base_env(home=tmp / "home")

            # default logger (stderr) on purpose: it must be importable and callable

            prepared = ws.prepare(tier, tmp / "work", git_bin=GIT, oak_bin="oak", env=env, kinds=("git",))

            subjects = [SimpleNamespace(name=n, kind="git", bin=Path(GIT)) for n in ("git", "git_aa")]

            try:

                rows = runner_mod.run_schedule(prepared, subjects, runs=1, ops=ops_mod.ALL_OPS, bench_id="e2e",

                                               seed=1, scratch=tmp / "scratch", details={}, runner_fields={},

                                               timeout_s=120, lock_wait_ms=None, log=lambda *_: None)

            finally:

                prepared.close()

            self.assertEqual(validate_rows(rows, "scale"), [])

            failed = [(r["operation"], r.get("failure"), r.get("oracle_detail")) for r in rows if r["returncode"] == 1]

            self.assertEqual(failed, [])

            measured = {r["operation"] for r in rows if r["returncode"] == 0}

            self.assertEqual(measured, set(ops_mod.ALL_OPS))

            self.assertTrue(all(r["oracle_ok"] for r in rows if r["returncode"] == 0))

        finally:

            shutil.rmtree(tmp, ignore_errors=True)

if __name__ == "__main__":

    unittest.main()