"""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[-2:], [["switch", "base"], ["close", "dirty-oak-r1"]])

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

    def test_oak_harness_branches_are_closed_on_teardown(self) -> None:

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

        ops_mod.execute(self.ctx("oak", cap("")), "branch.create")

        self.assertIn(["close", "new-oak-r1-1"], self.untimed_calls)

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

        self.untimed_calls.clear()

        seen = []

        ctx = self.ctx("oak", cap(listing))

        original = ctx._run_timed

        ctx._run_timed = lambda command, *a, **k: (seen.append(command), original(command, *a, **k))[1]

        ops_mod.execute(ctx, "branch.list")

        self.assertIn("--status", seen[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()

class WatchdogTimingTest(unittest.TestCase):

    """QA P5-1: a bounded command must not pay Popen.wait(timeout) poll quantization."""

    def test_watchdog_bound_does_not_change_timing(self) -> None:

        import statistics

        env = {"PATH": "/usr/bin:/bin"}

        plain, bounded = [], []

        for _ in range(25):  # interleaved pairs share host drift

            plain.append(run_timed(["/bin/sleep", "0.013"], Path("/"), env, 100).elapsed_ms)

            bounded.append(run_timed(["/bin/sleep", "0.013"], Path("/"), env, 100, watchdog_seconds=60.0).elapsed_ms)

        delta = statistics.median(bounded) - statistics.median(plain)

        self.assertLess(abs(delta), 2.0, (statistics.median(plain), statistics.median(bounded)))

    def test_watchdog_kills_and_reports_timeout(self) -> None:

        capture = run_timed(["/bin/sleep", "5"], Path("/"), {"PATH": "/usr/bin:/bin"}, 100, watchdog_seconds=0.3)

        self.assertTrue(capture.timed_out)

        self.assertEqual(capture.returncode, 124)

        self.assertLess(capture.elapsed_ms, 3000)

    def test_both_bounds_rejected(self) -> None:

        with self.assertRaises(ValueError):

            run_timed(["/usr/bin/true"], Path("/"), {}, 10, timeout_seconds=1, watchdog_seconds=1)

    def test_scale_ops_use_the_watchdog(self) -> None:

        seen = {}

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

            seen.update(kwargs)

            return cap("")

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

        prepared = SimpleNamespace(env={}, state={}, path=lambda *p: Path("/"))

        ops_mod.Ctx(prepared, subject, 1, fake, Path("/"), 30.0).timed(["status"], Path("/"))

        self.assertEqual(seen.get("watchdog_seconds"), 30.0)

        self.assertNotIn("timeout_seconds", seen)

def _cell_rows(tier: str, op: str, seed, base_git: float = 100.0, base_oak: float = 700.0) -> list:

    rows = []

    for run in range(1, 7):

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

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

            row = {"scenario": tier, "operation": op, "run": run, "subject": subject, "subject_kind": kind,

                   "returncode": 0, "elapsed_ms": base + (run % 3) * 0.5}

            if kind == "oak" and seed is not None:

                row["oak_server_seed"] = seed

            rows.append(row)

    return rows

class ComparabilityTest(unittest.TestCase):

    """QA P5-2: a snapshot-seeded Oak server is not the repository Git serves."""

    SUBJECTS = ["git", "git_aa", "oak_installed", "oak_installed_aa"]

    def test_snapshot_seeded_sync_cells_suppress_oak_vs_git(self) -> None:

        rows = []

        tiers = ("default", "xl-deep", "xl-history", "xl-refs")

        ops = ("push.branch", "pull.ff", "pull.diverged", "clone.full")

        for tier in tiers:

            for op in ops:

                rows += _cell_rows(tier, op, "snapshot_1_commit")

            rows += _cell_rows(tier, "status.clean", None)

        report = {(c["scenario"], c["operation"]): c for c in runner_mod.cell_report(rows, self.SUBJECTS)}

        for tier in tiers:

            for op in ops:

                cell = report[(tier, op)]

                self.assertEqual(cell["oak_vs_git_comparability"], "not_comparable:oak_server_seed_snapshot_1_commit")

                self.assertIsNone(cell["oak_vs_git_delta_pct_dev_only"])

                self.assertIsNone(cell["oak_vs_git_mde_pct_80"])

                self.assertIsNotNone(cell["oak_vs_oak_mde_pct_80"])

            comparable = report[(tier, "status.clean")]

            self.assertEqual(comparable["oak_vs_git_comparability"], "comparable")

            self.assertIsNotNone(comparable["oak_vs_git_delta_pct_dev_only"])

        prepared = SimpleNamespace(tier=tiers_mod.TIERS["default"], phases={}, notes=[],

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

        text = runner_mod.summary_markdown(prepared, [report[("default", "pull.ff")], report[("default", "status.clean")]],

                                           {"bench_id": "b", "runs": 6, "subjects": self.SUBJECTS})

        pull_line = next(line for line in text.splitlines() if line.startswith("| pull.ff "))

        self.assertIn("| n/c |", pull_line)

        self.assertIn("oak-vs-oak only", pull_line)

        self.assertNotIn("7.00x", pull_line)

    def test_full_history_seed_stays_comparable(self) -> None:

        rows = _cell_rows("fresh", "clone.full", "full_history")

        cell = runner_mod.cell_report(rows, self.SUBJECTS)[0]

        self.assertEqual(cell["oak_vs_git_comparability"], "comparable")

class CalibrationGateTest(unittest.TestCase):

    """QA P5-3 (rounds 1-2): claimability is verified by RECOMPUTING the calibration

    campaign from its own JSONL; nothing a record states is trusted."""

    PIN = "scale-fresh-v1"

    def setUp(self) -> None:

        from oakbench import scale_calibration as cal

        self.cal = cal

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

        self.addCleanup(shutil.rmtree, self.tmp, True)

        entry = json.loads((ROOT / "config" / "fixtures.json").read_text())["fixtures"][self.PIN]

        self.fixture = {"tier": "fresh", "fixture_id": "scale-fresh-x", "fixture_spec_sha256": entry["spec_sha256"],

                        "fixture_manifest_sha256": entry["refs_sha256"], "fixture_pin_status": "verified"}

        self.volume = {"mount_point": "/System/Volumes/Data", "device": "/dev/disk3s5", "fstype": "apfs",

                       "volume_uuid": "VOL-1", "bus_protocol": "Apple Fabric", "physical_store": "disk0s2",

                       "medium": "physical"}

        self.identity = {"runner_id": "runner-1", "host_cpu_count": 14, "host_mem_bytes": 1 << 35,

                         "work_volume": self.volume, "runner_profile": {"system": "Darwin"}}

    def rows(self, bench_id: str, *, load: float = 0.5, jitter: float = 0.05, fsync_ms: float = 1.0,

             identity=None, calibrated=None, flush_kind: str = "F_FULLFSYNC", slow_aa=None,

             late_aa_shift: float = 0.0) -> list:

        ident = identity or self.identity

        rows = []

        for run in range(1, 11):

            for op_index in range(12):

                for subject, kind in (("git", "git"), ("git_aa", "git"), ("oak_installed", "oak"),

                                      ("oak_installed_aa", "oak")):

                    rows.append({"bench_id": bench_id, "profile": "scale", "scenario": "fresh",

                                 "operation": f"op{op_index}", "run": run, "subject": subject, "subject_kind": kind,

                                 "elapsed_ms": 10.0 + op_index + ((run + subject.endswith("_aa")) % 3) * jitter,

                                 "returncode": 0, "command": [],

                                 "binary_sha256": ("a" if kind == "git" else "b") * 64,

                                 "environment_sample": {"loadavg": [load, load, load]},

                                 **self.fixture, **ident})

                    if subject == "oak_installed_aa" and slow_aa and op_index in slow_aa:

                        rows[-1]["elapsed_ms"] *= 1.5

                    if subject == "oak_installed_aa" and late_aa_shift and run > 5:

                        rows[-1]["elapsed_ms"] *= 1.0 + late_aa_shift

        base = {"bench_id": bench_id, "profile": "scale", "scenario": "fresh", **self.fixture}

        probe = {"fullfsync_ms": [fsync_ms] * 20, "fullfsync_kind": flush_kind, "create_files": 500,

                 "create_elapsed_s": 0.05, "work_volume": ident["work_volume"]}

        probes = self.cal.probe_rows(probe, base, "start")

        for row in probes:

            row.update(ident)

        rows = probes + rows

        if calibrated is not None:

            record, facts, reasons = calibrated

            for row in rows:

                if not row.get("diagnostic"):

                    self.cal.stamp(row, record, facts, reasons)

        return rows

    def write(self, name: str, rows: list) -> Path:

        path = self.tmp / name

        path.write_text("".join(json.dumps(row, sort_keys=True) + "\n" for row in rows))

        return path

    def campaign(self, name: str = "calibration.jsonl", **kwargs):

        path = self.write(name, self.rows("cal", **kwargs))

        rows = self.cal.load_rows(path)

        facts, problems = self.cal.recompute(rows, self.registry())

        record = self.cal.seal({**{k: facts.get(k) for k in ("runner_id", "host_cpu_count", "host_mem_bytes",

                                                              "oak_binary_sha256", "git_binary_sha256",

                                                              "work_volume_key", "aa_cells",

                                                              "aa_floor_median_pct", "aa_false_claims")},

                                "calibration_rows_path": str(path),

                                "calibration_rows_sha256": self.cal.file_sha256(path)})

        return path, record, facts, problems

    def registry(self):

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

    def gate_text(self, *paths: Path) -> str:

        import contextlib

        import publish_gate

        out = io.StringIO()

        with contextlib.redirect_stdout(out), contextlib.redirect_stderr(io.StringIO()):

            publish_gate.main([str(p) for p in paths])

        return out.getvalue()

    def published(self, record, facts, problems, **kwargs) -> Path:

        return self.write("published.jsonl", self.rows("pub", calibrated=(record, facts, problems), **kwargs))

    def test_quiet_campaign_recomputes_and_publishes(self) -> None:

        cal_path, record, facts, problems = self.campaign()

        self.assertEqual(problems, [])

        self.assertEqual(facts["aa_cells"], 24)

        self.assertEqual(facts["aa_per_arm"]["git"]["held_out_claims"], 0)

        self.assertEqual(facts["aa_false_claims"], 0)

        pub = self.published(record, facts, problems)

        rows = self.cal.load_rows(pub)

        self.assertTrue(all(r["claim_eligible"] for r in rows if not r.get("diagnostic")))

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

        text = self.gate_text(pub, cal_path)

        self.assertIn("PASS   scale-calibration", text)

        self.assertNotIn("FAIL   scale-calibration", text)

    def test_campaign_jsonl_missing_from_inputs_fails(self) -> None:

        cal_path, record, facts, problems = self.campaign()

        self.assertIn("FAIL   scale-calibration", self.gate_text(self.published(record, facts, problems)))

    def test_calibration_jsonl_not_matching_record_sha_fails(self) -> None:

        cal_path, record, facts, problems = self.campaign()

        pub = self.published(record, facts, problems)

        cal_path.write_text(cal_path.read_text().replace('"elapsed_ms": 10.0', '"elapsed_ms": 10.5', 1))

        text = self.gate_text(pub, cal_path)

        self.assertIn("FAIL   scale-calibration", text)

        self.assertIn("calibration campaign rows not provided", text)  # no input matches the record's sha

    def test_resealed_edited_record_fails(self) -> None:

        cal_path, record, facts, problems = self.campaign(jitter=4.0)  # noisy campaign: floors too wide

        self.assertTrue(problems)

        forged = self.cal.seal({**{k: v for k, v in record.items() if k not in ("record_sha256",)},

                                "aa_floor_median_pct": 0.0})

        pub = self.published(forged, facts, [])  # stamped as if valid

        text = self.gate_text(pub, cal_path)

        self.assertIn("FAIL   scale-calibration", text)

    def test_resealed_record_numbers_are_never_trusted(self) -> None:

        cal_path, record, facts, problems = self.campaign()

        forged = self.cal.seal({**{k: v for k, v in record.items() if k != "record_sha256"}, "aa_cells": 999})

        text = self.gate_text(self.published(forged, facts, []), cal_path)

        self.assertIn("FAIL   scale-calibration", text)

        self.assertIn("recomputed", text)

    def test_copied_runner_record_fails(self) -> None:

        cal_path, record, facts, problems = self.campaign()

        other = dict(self.identity, runner_id="runner-2")

        text = self.gate_text(self.published(record, facts, [], identity=other), cal_path)

        self.assertIn("FAIL   scale-calibration", text)

        resealed = self.cal.seal({**{k: v for k, v in record.items() if k != "record_sha256"},

                                  "runner_id": "runner-2"})

        text = self.gate_text(self.published(resealed, facts, [], identity=other), cal_path)

        self.assertIn("FAIL   scale-calibration", text)

    def test_host_mismatch_and_loaded_host_fail(self) -> None:

        cal_path, record, facts, problems = self.campaign()

        bigger = dict(self.identity, host_cpu_count=100000)

        self.assertIn("FAIL   scale-calibration", self.gate_text(self.published(record, facts, [], identity=bigger),

                                                                   cal_path))

        loaded = self.campaign("loaded.jsonl", load=2.5)

        self.assertTrue(any("not quiet" in r for r in loaded[3]))

        disk = self.campaign("disk.jsonl", fsync_ms=4.0)

        self.assertTrue(any("disk contended" in r for r in disk[3]))

        text = self.gate_text(self.published(record, facts, [], fsync_ms=4.0), cal_path)

        self.assertIn("published run: disk contended", text)

    def test_broken_aa_arm_fails(self) -> None:

        # QA r3 probe: the identical oak binary 50% slower in 11 of 12 oak_aa cells.

        _, _, facts, problems = self.campaign("slow-aa.jsonl", slow_aa=set(range(11)))

        self.assertTrue(any("oak_installed" in p and ("floor" in p or "cells have" in p) for p in problems),

                        problems)

        self.assertEqual(facts["aa_per_arm"]["git"]["cells_over_threshold"], 0)

    def test_held_out_aa_claims_are_counted(self) -> None:

        # A shift that appears only in the held-out half cannot hide inside its own floor.

        _, _, facts, problems = self.campaign("late-shift.jsonl", late_aa_shift=0.5)

        self.assertGreater(facts["aa_false_claims"], 0)

        self.assertTrue(any("held-out claim" in p for p in problems), problems)

    def test_quiet_campaign_has_no_held_out_claims(self) -> None:

        _, _, facts, problems = self.campaign("quiet.jsonl")

        self.assertEqual(problems, [])

        self.assertEqual(facts["aa_per_arm"]["oak_installed"]["held_out_claims"], 0)

    def test_too_few_runs_fails(self) -> None:

        path = self.write("short.jsonl", [r for r in self.rows("cal") if r.get("diagnostic") or r["run"] <= 6])

        self.assertTrue(any("fewer than 10 runs" in p for p in self.cal.recompute(self.cal.load_rows(path),

                                                                                   self.registry())[1]))

    def test_ram_or_image_volume_and_volume_mismatch_fail(self) -> None:

        ram = dict(self.volume, medium="ram_or_disk_image", bus_protocol="Disk Image", volume_uuid="RAM")

        _, _, _, problems = self.campaign("ram.jsonl", identity=dict(self.identity, work_volume=ram))

        self.assertTrue(any("cannot certify" in p for p in problems), problems)

        cal_path, record, facts, problems = self.campaign()

        self.assertEqual(problems, [])

        other = dict(self.volume, volume_uuid="VOL-2", device="/dev/disk9s1")

        rows = self.rows("pub", calibrated=(record, facts, []))

        self.assertTrue(rows[-1]["claim_eligible"])

        for row in rows:  # hand-edit: claim the published run used another (physical) volume

            row["work_volume"] = other

        text = self.gate_text(self.write("published.jsonl", rows), cal_path)

        self.assertIn("FAIL   scale-calibration", text)

        self.assertIn("different work volume", text)

    def test_flush_probe_must_use_platform_full_flush(self) -> None:

        _, _, _, problems = self.campaign("plainfsync.jsonl", flush_kind="fsync")

        self.assertTrue(any("full-flush primitive" in p for p in problems), problems)

        linux = dict(self.identity, runner_profile={"system": "Linux"})

        _, _, _, problems = self.campaign("linux.jsonl", identity=linux, flush_kind="fsync")

        self.assertFalse(any("full-flush" in p for p in problems), problems)

    def test_real_volume_identity_is_physical_here(self) -> None:

        volume = self.cal.volume_identity(ROOT)

        self.assertIn(volume["medium"], ("physical", "unknown"))

        self.assertTrue(volume["mount_point"])

    def test_classify_medium(self) -> None:

        self.assertEqual(self.cal.classify_medium({"fstype": "tmpfs"}, {}), "tmpfs")

        self.assertEqual(self.cal.classify_medium({"fstype": "apfs"}, {"BusProtocol": "Disk Image"}),

                         "ram_or_disk_image")

        self.assertEqual(self.cal.classify_medium({"fstype": "apfs"}, {"BusProtocol": "Apple Fabric"}), "physical")

        self.assertEqual(self.cal.classify_medium({"fstype": "apfs"}, {}), "unknown")

    def test_qa_forged_record_fixture_fails(self) -> None:

        import gzip

        forged = self.tmp / "forgedC.fresh.jsonl"

        forged.write_bytes(gzip.decompress((ROOT / "tests" / "fixtures" / "scale" /

                                            "qa-r2-forgedC.fresh.jsonl.gz").read_bytes()))

        rows = self.cal.load_rows(forged)

        self.assertTrue(any(r.get("claim_eligible") is True for r in rows))  # the forgery claims eligibility

        self.assertIn("FAIL   scale-calibration", self.gate_text(forged))

    def test_without_record_rows_are_dev_only(self) -> None:

        row = self.cal.stamp({"returncode": 0}, None, None, [])

        self.assertFalse(row["claim_eligible"])

        self.assertEqual(row["evidence_class"], "development_only_loaded_host")

class PinAndTeardownTest(unittest.TestCase):

    def test_check_pin(self) -> None:

        registry = json.loads((ROOT / "config" / "fixtures.json").read_text())["fixtures"]["scale-default-v1"]

        manifest = {"spec_sha256": registry["spec_sha256"], "refs_sha256": registry["refs_sha256"]}

        self.assertEqual(gen.check_pin(manifest), "verified")

        with self.assertRaises(gen.FixturePinMismatch):

            gen.check_pin(dict(manifest, refs_sha256="0" * 64))

        self.assertEqual(gen.check_pin({"spec_sha256": "f" * 64, "refs_sha256": "0" * 64}), "unpinned")

    def test_results_file_naming(self) -> None:

        from oakbench.results import ResultsStore

        with tempfile.TemporaryDirectory() as tmp:

            store = ResultsStore(Path(tmp), "scale", filename_suffix="fresh")

            raw, _ = store.write("20260929T000000Z", [], "x")

            self.assertEqual(raw.name, "20260929T000000Z.fresh.jsonl")