"""Roundtrip conformance matrix: author -> commit -> push -> fresh clone (+ pull) -> compare.

Every behaviour has an immutable ID (ADR-0005). The expected state is what the

harness itself wrote to the filesystem (never a VCS answer). Each subject run

compares that expectation against:

- ``commit_read`` - the committed tree of the authoring repository, read by an

  independent instrument: Oak through ``oak_sqlite_reader`` (pure-Python

  BLAKE3/zstd, no Oak code), Git through its plumbing (``ls-tree``/``cat-file``);

  checked after every step, so intermediate commits (deletes, transitions) count;

- ``source_preserved`` - the authoring worktree is not rewritten by commit/push

  (checked after every step);

- ``clone_fs`` / ``clone_read`` / ``clone_status_clean`` - a fresh clone of the

  pushed branch: filesystem bytes, independently read committed state, and the

  subject's own claim that the fresh clone is clean;

- ``pull_fs`` / ``pull_read`` / ``pull_status_clean`` - multi-step behaviours

  only: a clone taken after step 1 and updated with ``pull``.

A Git differential is recorded on Oak rows: Oak's independently read committed

tree against Git's committed tree for the same authored fixture (paths, bytes,

executable bit, symlink targets). Records are diagnostic conformance evidence

(``evidence_class: diagnostic_conformance``), never timing rows.

"""

from __future__ import annotations

from dataclasses import dataclass

import hashlib

import json

import os

from pathlib import Path

import shutil

import signal

import socket

import stat

import subprocess

import tempfile

import time

from oakbench.environment import base_env

MATRIX_VERSION = "2026-09-28.2"

ORACLE = "roundtrip_independent_v1"

SCHEMA_VERSION = 3

HUGE_ENV = "OAK_BENCH_CONFORMANCE_HUGE"

COMMAND_TIMEOUT_S = 180

HUGE_TIMEOUT_S = 3600

LEGS = ("commit_read", "source_preserved", "clone_fs", "clone_read", "clone_status_clean",

        "pull_fs", "pull_read", "pull_status_clean")

class CapabilityUnavailable(Exception):

    pass

class LegFailure(AssertionError):

    pass

# --- fixture DSL ------------------------------------------------------------

# Each step is a tuple of operations applied in order to the authoring tree:

#   ("write", path, bytes[, executable]) ("symlink", path, target)

#   ("mkdir", path) ("remove", path) ("rename", src, dst) ("chmod", path, mode)

#   ("sparse", path, size, tail) ("inplace", path, bytes)  # same size, mtime kept

def apply_op(root: Path, op: tuple) -> None:

    kind = op[0]

    path = root / op[1]

    if kind == "write":

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

        path.write_bytes(op[2])

        path.chmod(0o755 if len(op) > 3 and op[3] else 0o644)

    elif kind == "symlink":

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

        os.symlink(op[2], path)

    elif kind == "mkdir":

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

    elif kind == "remove":

        if path.is_symlink() or path.is_file():

            path.unlink()

        else:

            shutil.rmtree(path)

    elif kind == "rename":

        target = root / op[2]

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

        os.rename(path, target)

    elif kind == "chmod":

        path.chmod(op[2])

    elif kind == "sparse":

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

        with path.open("wb") as stream:

            stream.truncate(op[2] - len(op[3]))

            stream.seek(op[2] - len(op[3]))

            stream.write(op[3])

    elif kind == "inplace":

        # Same size, same mtime: a stat-cache that trusts (size, mtime) misses it.

        before = path.stat()

        if len(op[2]) != before.st_size:

            raise ValueError("inplace edit must keep the size")

        with path.open("r+b") as stream:

            stream.write(op[2])

        os.utime(path, ns=(before.st_atime_ns, before.st_mtime_ns))

    else:

        raise ValueError(f"unknown fixture op {kind!r}")

@dataclass(frozen=True)

class Behaviour:

    id: str

    group: str

    summary: str

    steps: tuple

    requires: frozenset = frozenset()

    gated: bool = False

    prior_art: tuple = ()

    observe_dirs: bool = False

    def describe(self) -> dict:

        return {"behaviour": self.id, "group": self.group, "summary": self.summary,

                "steps": len(self.steps), "requires": sorted(self.requires), "gated": self.gated,

                "prior_art": list(self.prior_art)}

def _w(path, data, executable=False):

    return ("write", path, data if isinstance(data, bytes) else data.encode(), executable)

def _files(*names, payload=b"conformance payload\n"):

    return tuple(_w(name, str(index).encode() + b":" + payload) for index, name in enumerate(names))

def _stream(seed: bytes, size: int) -> bytes:

    """Deterministic incompressible bytes (SHA-256 counter stream)."""

    out = bytearray()

    counter = 0

    while len(out) < size:

        out += hashlib.sha256(seed + counter.to_bytes(8, "little")).digest()

        counter += 1

    return bytes(out[:size])

def _many_small():

    ops = []

    for index in range(1200):

        size = (index * 7919) % 65

        ops.append(_w(f"d{index % 30:02d}/f{index:05d}.txt", _stream(b"small%d" % index, size)))

    return tuple(ops)

T3902 = "git t3902-quoted (unambiguous odd path names)"

T0050 = "git t0050-filesystem (case and Unicode-normalization filesystems)"

T3910 = "git t3910-mac-os-precompose (NFC/NFD names)"

T3700 = "git t3700-add (executable bit, core.filemode)"

T4129 = "git t4129-apply-samemode (mode-only changes)"

T4011 = "git t4011-diff-symlink (symlink targets as content)"

T2007 = "git t2007-checkout-symlink (directory replaced by symlink)"

T4114 = "git t4114-apply-typechange (file/symlink/directory type changes)"

T2021 = "git t2021-checkout-overwrite (file/directory replacement on checkout)"

T2003 = "git t2003-checkout-cache-mkdir (leading-path creation, symlinked parents)"

T0020 = "git t0020-crlf (byte-exact line endings when conversion is off)"

T0021 = "git t0021-conversion (content conversion must not alter stored bytes)"

T0027 = "git t0027-auto-crlf (LF/CRLF/mixed/no-final-newline matrix)"

T4012 = "git t4012-diff-binary (binary detection)"

T0010 = "git t0010-racy-git (same-size edits within timestamp granularity)"

T7001 = "git t7001-mv (renames, case-only renames)"

T4102 = "git t4102-apply-rename (moves across directories)"

DULWICH = "dulwich tests/compat (differential against real git; pattern only)"

LONG_COMPONENT = "l" * 251 + ".txt"

LONG_UNICODE_COMPONENT = "é" * 127  # 254 UTF-8 bytes

LONG_PATH = "/".join(f"segment-{index:02d}-" + "p" * 48 for index in range(12)) + "/leaf.txt"

DEEP_PATH = "/".join("d" for _ in range(40)) + "/bottom.txt"

BEHAVIOURS = (

    # --- paths ---------------------------------------------------------------

    Behaviour("path_space_v1", "paths", "Leading, trailing and inner spaces in files and directories.",

              (_files("space name", " leading space", "trailing space ", "dir with space/inner file"),),

              prior_art=(T3902,)),

    Behaviour("path_tab_v1", "paths", "Tab inside a file name.", (_files("tab\tname"),), prior_art=(T3902,)),

    Behaviour("path_newline_v1", "paths", "Newline and carriage return inside a file name.",

              (_files("line\nbreak", "carriage\rreturn"),), prior_art=(T3902,)),

    Behaviour("path_control_chars_v1", "paths", "Other C0/DEL control characters inside file names.",

              (_files("bell\x07name", "escape\x1b[0m", "delete\x7fname", "unit\x1fsep"),), prior_art=(T3902,)),

    Behaviour("path_leading_dash_v1", "paths", "Names that look like command-line options.",

              (_files("-leading", "--double-dash", "-n", "--", "dir/-inner"),), prior_art=(T3902,)),

    Behaviour("path_shell_metachars_v1", "paths", "Quotes, backslash, glob, shell and URL metacharacters.",

              (_files("quote\"name", "single'quote", "glob*star?[x]", "hash#bang!",

                      "$HOME", "semi;amp&pipe|", "percent%20", "colon:name", "brace{a,b}", "tilde~", "a=b"),),

              prior_art=(T3902,)),

    Behaviour("path_backslash_v1", "paths", "Backslash inside a POSIX file name is a literal byte, not a separator.",

              (_files("back\\slash", "trailing\\", "dir/in\\ner"),), requires=frozenset({"posix_backslash_names"}),

              prior_art=(T3902,)),

    Behaviour("path_unicode_nfc_v1", "paths", "Precomposed and non-Latin Unicode names, astral and format characters.",

              (_files("café", "日本語/文件", "λ-lambda", "emoji-\U0001F600",

                      "zwj-‍-joiner", "rtl-‮-override"),), prior_art=(T3910, T0050)),

    Behaviour("path_unicode_nfd_v1", "paths", "Decomposed (NFD) names must keep their exact code points.",

              (_files("café", "Å-ring"),), prior_art=(T3910, T0050)),

    Behaviour("path_unicode_nfc_nfd_pair_v1", "paths", "NFC and NFD spellings of one name as two distinct paths.",

              (_files("café", "café"),), requires=frozenset({"normalization_sensitive_fs"}),

              prior_art=(T3910, T0050)),

    Behaviour("path_non_utf8_v1", "paths", "A file name that is not valid UTF-8.",

              ((("write", os.fsdecode(b"latin1-\xe9.txt"), b"bytes name\n"),),),

              requires=frozenset({"non_utf8_names"}), prior_art=(T3902,)),

    Behaviour("path_long_component_v1", "paths", "255-byte path components (ASCII and multi-byte).",

              (_files(LONG_COMPONENT, LONG_UNICODE_COMPONENT),), requires=frozenset({"long_component_255"})),

    Behaviour("path_long_path_v1", "paths", "A ~700-byte relative path.", (_files(LONG_PATH),)),

    Behaviour("path_deep_tree_v1", "paths", "A file 40 directories deep.", (_files(DEEP_PATH),)),

    Behaviour("path_reserved_names_v1", "paths", "Windows-reserved and dot-heavy names that are legal on POSIX.",

              (_files("CON", "prn.txt", "aux.c", "NUL", "com1", "LPT9.log", "trailing-dot.", "..double-leading",

                      "...", ".hidden", "-", "_", "@", "clock$"),), prior_art=(T0050,)),

    Behaviour("path_case_only_rename_v1", "paths", "Case-only rename on a case-insensitive filesystem.",

              ((_w("File.txt", "case rename payload\n"),), (("rename", "File.txt", "file.txt"),)),

              requires=frozenset({"case_insensitive_fs"}), prior_art=(T7001, T0050)),

    Behaviour("path_case_distinct_v1", "paths", "Names differing only by case on a case-sensitive filesystem.",

              (_files("README", "readme", "ReadMe"),), requires=frozenset({"case_sensitive_fs"}),

              prior_art=(T0050,)),

    # --- modes ---------------------------------------------------------------

    Behaviour("mode_exec_new_v1", "modes", "New executable file next to a regular file with identical bytes.",

              ((_w("tool.sh", "#!/bin/sh\necho conformance\n", True), _w("copy.txt", "#!/bin/sh\necho conformance\n")),),

              requires=frozenset({"exec_bit"}), prior_art=(T3700,)),

    Behaviour("mode_exec_add_v1", "modes", "Executable bit added without a content change.",

              ((_w("run.sh", "#!/bin/sh\ntrue\n"),), (("chmod", "run.sh", 0o755),)),

              requires=frozenset({"exec_bit"}), prior_art=(T3700, T4129)),

    Behaviour("mode_exec_remove_v1", "modes", "Executable bit removed without a content change.",

              ((_w("run.sh", "#!/bin/sh\ntrue\n", True),), (("chmod", "run.sh", 0o644),)),

              requires=frozenset({"exec_bit"}), prior_art=(T3700, T4129)),

    Behaviour("mode_exec_and_content_v1", "modes", "Executable bit and content change in one commit.",

              ((_w("run.sh", "#!/bin/sh\ntrue\n"),), (_w("run.sh", "#!/bin/sh\nfalse\n", True),)),

              requires=frozenset({"exec_bit"}), prior_art=(T3700,)),

    # --- symlinks ------------------------------------------------------------

    Behaviour("symlink_relative_v1", "symlinks", "Relative links at several depths.",

              ((_w("target.txt", "link target\n"), ("symlink", "link", "target.txt"),

                ("symlink", "sub/uplink", "../target.txt"), ("symlink", "sub/deep/ln", "../../target.txt")),),

              requires=frozenset({"symlink"}), prior_art=(T4011,)),

    Behaviour("symlink_absolute_v1", "symlinks", "Absolute link targets are stored verbatim, never followed.",

              ((_w("keep.txt", "keep\n"), ("symlink", "abs", "/nonexistent-oak-conformance/absolute/target"),

                ("symlink", "abs-root", "/")),), requires=frozenset({"symlink"}), prior_art=(T4011,)),

    Behaviour("symlink_dangling_v1", "symlinks", "Dangling, self-referential and cyclic links.",

              ((_w("keep.txt", "keep\n"), ("symlink", "dangling", "absent-target"), ("symlink", "self", "self"),

                ("symlink", "loop-a", "loop-b"), ("symlink", "loop-b", "loop-a")),),

              requires=frozenset({"symlink"}), prior_art=(T4011,)),

    Behaviour("symlink_to_dir_v1", "symlinks", "Links to directories (including '.') are stored as links.",

              ((_w("real/file.txt", "inside real dir\n"), ("symlink", "alias", "real"), ("symlink", "here", ".")),),

              requires=frozenset({"symlink"}), prior_art=(T2003, T4011)),

    Behaviour("symlink_target_edge_v1", "symlinks", "Targets with spaces, Unicode, a newline and 1000 bytes.",

              ((_w("keep.txt", "keep\n"), ("symlink", "spaced", "target with space"),

                ("symlink", "unicode", "tärget-λ"), ("symlink", "newline", "nl\ntarget"),

                ("symlink", "long", "x" * 1000)),), requires=frozenset({"symlink"}), prior_art=(T4011,)),

    Behaviour("symlink_retarget_v1", "symlinks", "A link's target changes.",

              ((_w("a", "a\n"), _w("b", "b\n"), ("symlink", "l", "a")), (("remove", "l"), ("symlink", "l", "b"))),

              requires=frozenset({"symlink"}), prior_art=(T4011,)),

    Behaviour("symlink_to_file_v1", "symlinks", "A symlink replaced by a regular file.",

              ((_w("t", "t\n"), ("symlink", "x", "t")), (("remove", "x"), _w("x", "now a file\n"))),

              requires=frozenset({"symlink"}), prior_art=(T4114,)),

    Behaviour("file_to_symlink_v1", "symlinks", "A regular file replaced by a symlink.",

              ((_w("t", "t\n"), _w("x", "a file\n")), (("remove", "x"), ("symlink", "x", "t"))),

              requires=frozenset({"symlink"}), prior_art=(T4114,)),

    Behaviour("dir_to_symlink_v1", "symlinks", "A directory replaced by a symlink.",

              ((_w("p/inner.txt", "inner\n"), _w("elsewhere/e.txt", "e\n")),

               (("remove", "p"), ("symlink", "p", "elsewhere"))),

              requires=frozenset({"symlink"}), prior_art=(T2007, T4114)),

    Behaviour("symlink_to_dir_real_v1", "symlinks", "A symlink replaced by a real directory.",

              ((_w("elsewhere/e.txt", "e\n"), ("symlink", "p", "elsewhere")),

               (("remove", "p"), _w("p/inner.txt", "inner\n"))),

              requires=frozenset({"symlink"}), prior_art=(T2007, T4114)),

    # --- file <-> directory --------------------------------------------------

    Behaviour("file_to_dir_v1", "file_dir", "A file replaced by a directory of the same name.",

              ((_w("node", "file\n"), _w("keep", "k\n")), (("remove", "node"), _w("node/child.txt", "child\n"))),

              prior_art=(T2021, T4114)),

    Behaviour("dir_to_file_v1", "file_dir", "A directory tree replaced by a file of the same name.",

              ((_w("node/child.txt", "child\n"), _w("node/sub/g.txt", "g\n"), _w("keep", "k\n")),

               (("remove", "node"), _w("node", "file now\n"))), prior_art=(T2021, T4114)),

    Behaviour("file_to_deep_dir_v1", "file_dir", "A nested file replaced by a deeper directory chain.",

              ((_w("a/b", "b file\n"),), (("remove", "a/b"), _w("a/b/c/d.txt", "deep\n"))), prior_art=(T2021,)),

    # --- empty ---------------------------------------------------------------

    Behaviour("empty_file_v1", "empty", "Empty regular and executable files at several depths.",

              ((_w("empty", b""), _w("dir/empty", b""), _w("empty-exec", b"", True)),),

              requires=frozenset({"exec_bit"})),

    Behaviour("empty_dir_semantics_v1", "empty", "Empty directories are not content (recorded, never failed).",

              ((_w("keep.txt", "keep\n"), ("mkdir", "void"), ("mkdir", "nested/void/deeper")),), observe_dirs=True),

    Behaviour("dir_emptied_v1", "empty", "Deleting a directory's last file leaves no stale file.",

              ((_w("d/f.txt", "f\n"), _w("keep", "k\n")), (("remove", "d/f.txt"),)), observe_dirs=True),

    Behaviour("delete_all_v1", "empty", "Deleting every tracked file commits an empty tree.",

              ((_w("a", "a\n"), _w("b/c", "c\n")), (("remove", "a"), ("remove", "b")))),

    # --- content -------------------------------------------------------------

    Behaviour("bytes_all_values_v1", "content", "Every byte value, repeated.",

              ((_w("binary", bytes(range(256)) * 17),),), prior_art=(T4012,)),

    Behaviour("eol_lf_v1", "content", "LF line endings.", ((_w("lf.txt", b"one\ntwo\nthree\n"),),),

              prior_art=(T0020, T0027)),

    Behaviour("eol_crlf_v1", "content", "CRLF line endings, byte-exact.", ((_w("crlf.txt", b"one\r\ntwo\r\nthree\r\n"),),),

              prior_art=(T0020, T0027)),

    Behaviour("eol_cr_only_v1", "content", "Classic-Mac CR-only line endings.", ((_w("cr.txt", b"one\rtwo\rthree\r"),),),

              prior_art=(T0020,)),

    Behaviour("eol_mixed_v1", "content", "LF, CRLF and CR mixed in one file.",

              ((_w("mixed.txt", b"lf\ncrlf\r\ncr\rlf-again\n\r\n\n"),),), prior_art=(T0027,)),

    Behaviour("eol_no_final_newline_v1", "content", "Text without a final newline (LF and CRLF bodies).",

              ((_w("tail-lf.txt", b"a\nb"), _w("tail-crlf.txt", b"a\r\nb")),), prior_art=(T0027,)),

    Behaviour("eol_lf_to_crlf_v1", "content", "Changing only LF to CRLF is a committed change.",

              ((_w("eol.txt", b"x\ny\n"),), (_w("eol.txt", b"x\r\ny\r\n"),)), prior_art=(T0020, T0021)),

    Behaviour("whitespace_only_change_v1", "content", "Trailing-space and tab/space-only edits are committed.",

              ((_w("ws.txt", b"a\n\tb\n"),), (_w("ws.txt", b"a \n    b\n"),)), prior_art=(T0021,)),

    Behaviour("utf8_bom_v1", "content", "UTF-8 and UTF-16 byte-order marks are preserved.",

              ((_w("bom8.txt", b"\xef\xbb\xbfbom text\n"), _w("bom16.txt", "utf16 text\n".encode("utf-16-le"))),),

              prior_art=(T0021,)),

    Behaviour("binary_detection_v1", "content", "Binary-looking and text-looking payloads roundtrip byte-exact.",

              ((_w("nul-in-text.txt", b"text\x00more\n"), _w("latin1.txt", b"caf\xe9\n"),

                _w("late-nul.txt", b"a" * 9000 + b"\x00tail\n"), _w("image.png", b"\x89PNG\r\n\x1a\n" + _stream(b"png", 2048)),

                _w("high-bytes", bytes(range(128, 256)) * 8), _w("random.bin", _stream(b"random", 65536))),),

              prior_art=(T4012,)),

    Behaviour("binary_text_transition_v1", "content", "The same path changes from text to binary.",

              ((_w("blob", b"plain text\n"),), (_w("blob", _stream(b"bin", 4096)),)), prior_art=(T4012,)),

    Behaviour("modify_same_size_same_mtime_v1", "content",

              "Same-size in-place edit with the mtime restored must still be committed.",

              ((_w("racy.txt", b"aaaa-original\n"),), (("inplace", "racy.txt", b"bbbb-modified\n"),)),

              prior_art=(T0010,)),

    Behaviour("large_file_chunked_v1", "content", "A 10 MiB + 64 KiB incompressible file (chunked storage path).",

              ((_w("large.bin", _stream(b"large", 10 * 1024 * 1024 + 64 * 1024)),),)),

    Behaviour("huge_file_sparse_v1", "content", "A 1 GiB + 5 byte sparse file (gated).",

              ((("sparse", "huge.bin", 1024 * 1024 * 1024 + 5, b"tail\n"),),),

              requires=frozenset({"sparse_files"}), gated=True),

    # --- history -------------------------------------------------------------

    Behaviour("delete_readd_v1", "history", "Delete then re-add identical bytes at the same path.",

              ((_w("f.txt", "same\n"), _w("keep", "k\n")), (("remove", "f.txt"),), (_w("f.txt", "same\n"),))),

    Behaviour("rename_move_v1", "history", "Move a file to a new directory with a new name.",

              ((_w("old/name.txt", "moving payload\n"),), (("rename", "old/name.txt", "new/dir/name2.txt"),)),

              prior_art=(T4102, T7001)),

    Behaviour("shared_blob_across_modes_v1", "history",

              "One byte string as regular file, executable, duplicate and symlink target.",

              ((_w("a.txt", "abc"), _w("b.sh", "abc", True), _w("c.txt", "abc"), ("symlink", "link", "abc")),),

              requires=frozenset({"symlink", "exec_bit"})),

    Behaviour("many_small_files_v1", "history", "1,200 small files across 30 directories.", (_many_small(),)),

)

BY_ID = {behaviour.id: behaviour for behaviour in BEHAVIOURS}

assert len(BY_ID) == len(BEHAVIOURS), "behaviour IDs must be unique"

# --- capability probes ------------------------------------------------------

def probe_capabilities(workdir: Path) -> dict:

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

    caps = {}

    with tempfile.TemporaryDirectory(prefix="caps-", dir=workdir) as temporary:

        root = Path(temporary)

        try:

            (root / "t").write_text("t")

            (root / "l").symlink_to("t")

            caps["symlink"] = (root / "l").is_symlink()

        except (OSError, NotImplementedError):

            caps["symlink"] = False

        probe = root / "x"

        probe.write_text("x")

        probe.chmod(0o755)

        caps["exec_bit"] = bool(probe.stat().st_mode & 0o100) and os.name == "posix"

        (root / "CaseProbe").write_text("c")

        insensitive = (root / "caseprobe").exists()

        caps["case_insensitive_fs"] = insensitive

        caps["case_sensitive_fs"] = not insensitive

        (root / "café").write_text("n")

        caps["normalization_sensitive_fs"] = not (root / "café").exists()

        try:

            (root / os.fsdecode(b"\xe9-probe")).write_text("b")

            caps["non_utf8_names"] = True

        except (OSError, UnicodeError):

            caps["non_utf8_names"] = False

        try:

            (root / LONG_COMPONENT).write_text("l")

            (root / LONG_UNICODE_COMPONENT).write_text("l")

            caps["long_component_255"] = True

        except OSError:

            caps["long_component_255"] = False

        try:

            sparse = root / "sparse"

            with sparse.open("wb") as stream:

                stream.truncate(64 * 1024 * 1024)

            caps["sparse_files"] = sparse.stat().st_blocks * 512 < 1024 * 1024

        except OSError:

            caps["sparse_files"] = False

        caps["posix_backslash_names"] = os.name == "posix"

        free = shutil.disk_usage(root).free

        caps["free_bytes"] = free

    return caps

def fs_profile(caps: dict) -> str:

    """Host filesystem class keying expected failures and coverage (e.g. darwin-ci-ni)."""

    import sys

    return "-".join((sys.platform, "ci" if caps.get("case_insensitive_fs") else "cs",

                     "ns" if caps.get("normalization_sensitive_fs") else "ni"))

# --- manifests --------------------------------------------------------------

def _sha256_file(path: Path) -> tuple[str, int]:

    digest = hashlib.sha256()

    size = 0

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

        for block in iter(lambda: stream.read(1024 * 1024), b""):

            digest.update(block)

            size += len(block)

    return digest.hexdigest(), size

def fs_manifest(root: Path, dirs: list | None = None) -> dict:

    """Exact path -> entry projection of a worktree, never following links.

    Empty directories are not content; pass ``dirs`` to collect every

    directory path as an observation. Root ``.git``/``.oak`` are excluded.

    """

    result = {}

    def visit(directory: Path, prefix: str) -> None:

        with os.scandir(directory) as iterator:

            children = sorted(iterator, key=lambda entry: entry.name)

        for entry in children:

            if not prefix and entry.name in (".git", ".oak"):

                continue

            relative = prefix + entry.name

            mode = entry.stat(follow_symlinks=False).st_mode

            if stat.S_ISLNK(mode):

                result[relative] = {"kind": "symlink", "target": os.readlink(entry.path)}

            elif stat.S_ISDIR(mode):

                if dirs is not None:

                    dirs.append(relative)

                visit(Path(entry.path), relative + "/")

            elif stat.S_ISREG(mode):

                digest, size = _sha256_file(Path(entry.path))

                result[relative] = {"kind": "file", "size": size, "sha256": digest,

                                    "executable": bool(mode & 0o111)}

            else:

                result[relative] = {"kind": "special"}

    visit(root, "")

    return result

def reader_projection(entries: list) -> dict:

    result = {}

    for entry in entries:

        if entry["kind"] == "file":

            result[entry["path"]] = {"kind": "file", "size": entry["size"], "sha256": entry["sha256"],

                                     "executable": entry["executable"]}

        elif entry["kind"] == "symlink":

            result[entry["path"]] = {"kind": "symlink",

                                     "target": os.fsdecode(bytes.fromhex(entry["target_hex"]))}

    return result

def diff_manifests(expected: dict, actual: dict, limit: int = 8) -> str | None:

    if expected == actual:

        return None

    missing = sorted(expected.keys() - actual.keys())

    extra = sorted(actual.keys() - expected.keys())

    changed = sorted(key for key in expected.keys() & actual.keys() if expected[key] != actual[key])

    parts = []

    for label, items in (("missing", missing), ("extra", extra), ("changed", changed)):

        if items:

            parts.append(f"{label}={items[:limit]!r}" + (f"(+{len(items) - limit})" if len(items) > limit else ""))

    return "; ".join(parts)

# --- subjects ---------------------------------------------------------------

class Runner:

    def __init__(self, binary: str, env: dict, timeout: float):

        self.binary, self.env, self.timeout = binary, env, timeout

        self.commands = []

    def run(self, cwd: Path, *args, check=True, timeout=None) -> subprocess.CompletedProcess:

        started = time.monotonic()

        try:

            process = subprocess.run([self.binary, *args], cwd=cwd, env=self.env, stdin=subprocess.DEVNULL,

                                     capture_output=True, timeout=timeout or self.timeout)

        except subprocess.TimeoutExpired as exc:

            self.commands.append({"argv": [str(a) for a in args], "returncode": None, "timeout": True})

            raise LegFailure(f"command timed out: {args[:3]!r}") from exc

        self.commands.append({

            "argv": [str(a) for a in args], "returncode": process.returncode,

            "seconds": round(time.monotonic() - started, 3),

            "stderr": process.stderr.decode("utf-8", "backslashreplace")[-600:]})

        if check and process.returncode != 0:

            raise LegFailure(f"{Path(self.binary).name} {' '.join(map(str, args[:2]))} exited "

                             f"{process.returncode}: {process.stderr.decode('utf-8', 'backslashreplace').strip()[-300:]}")

        return process

class GitSubject:

    kind = "git"

    reader_label = "git_plumbing_committed_tree"

    def __init__(self, runner: Runner, workdir: Path):

        self.git, self.workdir = runner, workdir

    CONFIG = ("core.autocrlf=false", "core.precomposeunicode=false", "core.filemode=true",

              "core.symlinks=true", "core.quotepath=false")

    def init(self, repo: Path):

        self.git.run(repo, "init", "-q", "-b", "main")

        for item in self.CONFIG:

            key, value = item.split("=")

            self.git.run(repo, "config", key, value)

        self.remote = repo.parent / "remote.git"

        self.git.run(repo.parent, "init", "-q", "--bare", str(self.remote))

        self.git.run(repo, "remote", "add", "origin", str(self.remote))

    def commit(self, repo: Path, message: str):

        self.git.run(repo, "add", "--all")

        self.git.run(repo, "commit", "-q", "--no-verify", "--allow-empty-message", "-m", message)

    def push(self, repo: Path):

        self.git.run(repo, "push", "-q", "origin", "HEAD:refs/heads/main")

    def clone(self, repo: Path, destination: Path):

        config = [argument for item in self.CONFIG for argument in ("--config", item)]

        self.git.run(destination.parent, "clone", "-q", *config, "--branch", "main", str(self.remote),

                     str(destination))

    def pull(self, repo: Path):

        self.git.run(repo, "pull", "-q", "--ff-only", "origin", "main")

    def status_clean(self, repo: Path):

        output = self.git.run(repo, "status", "--porcelain=v1", "-z", "--untracked-files=all").stdout

        if output:

            raise LegFailure("git status reports changes: " + output.decode("utf-8", "backslashreplace")[:300])

    def committed(self, repo: Path) -> tuple[dict, dict]:

        listing = self.git.run(repo, "ls-tree", "-r", "-z", "--full-tree", "HEAD", check=False)

        if listing.returncode != 0:

            head = self.git.run(repo, "rev-parse", "--verify", "-q", "HEAD", check=False)

            if head.returncode != 0:

                raise LegFailure("no committed HEAD")

            raise LegFailure("git ls-tree failed")

        entries = []

        for record in listing.stdout.split(b"\0"):

            if not record:

                continue

            meta, path = record.split(b"\t", 1)

            mode, kind, object_id = meta.split(b" ")

            entries.append((mode.decode(), kind.decode(), object_id.decode(), os.fsdecode(path)))

        manifest = {}

        if entries:

            process = subprocess.run([self.git.binary, "cat-file", "--batch"], cwd=repo, env=self.git.env,

                                     input=b"".join(e[2].encode() + b"\n" for e in entries),

                                     capture_output=True, timeout=self.git.timeout)

            if process.returncode != 0:

                raise LegFailure("git cat-file --batch failed")

            data, offset = process.stdout, 0

            for mode, kind, object_id, path in entries:

                end = data.index(b"\n", offset)

                header = data[offset:end].split(b" ")

                size = int(header[2])

                payload = data[end + 1:end + 1 + size]

                offset = end + 2 + size

                if mode == "120000":

                    manifest[path] = {"kind": "symlink", "target": os.fsdecode(payload)}

                elif mode in ("100644", "100755"):

                    manifest[path] = {"kind": "file", "size": size, "sha256": hashlib.sha256(payload).hexdigest(),

                                      "executable": mode == "100755"}

                else:

                    manifest[path] = {"kind": "special:" + mode}

        head = self.git.run(repo, "rev-parse", "HEAD").stdout.decode().strip()

        return manifest, {"head": head, "reader": self.reader_label}

class OakSubject:

    kind = "oak"

    reader_label = "independent_oak_sqlite_reader"

    def __init__(self, runner: Runner, workdir: Path, server):

        self.oak, self.workdir, self.server = runner, workdir, server

        self.branch = None

    def init(self, repo: Path):

        self.oak.run(repo, "init", ".")

        self.spec = self.server.new_repo_spec()

    def commit(self, repo: Path, message: str):

        self.oak.run(repo, "commit", "--no-verify")

        if self.branch is None:

            self.branch = json.loads(self.oak.run(repo, "status", "--json").stdout)["branch"]

    def push(self, repo: Path):

        self.oak.run(repo, "push", "--repo", self.spec, "--remote", self.server.url)

    def clone(self, repo: Path, destination: Path):

        self.oak.run(destination.parent, "clone", self.spec, str(destination), "--remote", self.server.url,

                     "--branch", self.branch)

    def pull(self, repo: Path):

        self.oak.run(repo, "pull")

    def status_clean(self, repo: Path):

        status = json.loads(self.oak.run(repo, "status", "--json").stdout)

        dirty = list(status.get("changes") or []) + list((status.get("working_changes") or {}).get("changes") or [])

        if dirty:

            raise LegFailure("oak status reports changes: " + json.dumps(dirty)[:300])

    def committed(self, repo: Path) -> tuple[dict, dict]:

        from oakbench import oak_sqlite_reader as reader

        try:

            result = reader.read_committed_tree(repo, None, manifest=True, deep=True, complete=True)

        except reader.Invalid as exc:

            raise LegFailure("independent reader: stored state invalid: " + str(exc)) from exc

        except reader.Unavailable as exc:

            raise CapabilityUnavailable("independent reader unavailable: " + str(exc)) from exc

        info = {"head": result["head"], "reader": self.reader_label, "reader_version": result["reader_version"],

                "reader_backend": result["reader_backend"], "history": result.get("history"),

                "chunked_blobs_reassembled": result["chunked_blobs_reassembled"],

                "chunk_mappings_crosschecked": result["chunk_mappings_crosschecked"]}

        return reader_projection(result["manifest"]), info

class LoopbackOakServer:

    """One ``oak serve`` on 127.0.0.1 per matrix run; only this PID is ever stopped."""

    def __init__(self, binary: str, root: Path, env: dict):

        self.binary, self.root, self.env = binary, root, env

        self.process = None

        self.counter = 0

    def __enter__(self):

        self.root.mkdir(parents=True, exist_ok=True)

        with socket.socket() as listener:

            listener.bind(("127.0.0.1", 0))

            self.port = listener.getsockname()[1]

        self.url = f"http://127.0.0.1:{self.port}"

        self.process = subprocess.Popen([self.binary, "serve", "--dir", str(self.root), "--host", "127.0.0.1",

                                         "--port", str(self.port)], cwd=self.root, env=self.env,

                                        stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL,

                                        start_new_session=os.name == "posix")

        deadline = time.monotonic() + 20

        while time.monotonic() < deadline:

            if self.process.poll() is not None:

                raise CapabilityUnavailable("oak serve exited before accepting connections")

            try:

                with socket.create_connection(("127.0.0.1", self.port), timeout=0.2):

                    return self

            except OSError:

                time.sleep(0.05)

        self.__exit__(None, None, None)

        raise CapabilityUnavailable("oak serve did not accept connections")

    def new_repo_spec(self) -> str:

        self.counter += 1

        return f"conformance/r{self.counter}-{os.getpid()}"

    def __exit__(self, *exc):

        """Stop the server's whole process group (a wrapper script's child too).

        The group was created by this run (start_new_session), so only

        processes this run started are signalled.

        """

        if self.process is None:

            return

        group = self.process.pid if os.name == "posix" else None

        for sig, wait in ((signal.SIGTERM, 10), (signal.SIGKILL, 10)):

            try:

                if group is not None:

                    os.killpg(group, sig)

                elif self.process.poll() is None:

                    self.process.send_signal(sig)

            except ProcessLookupError:

                break

            try:

                self.process.wait(timeout=wait)

            except subprocess.TimeoutExpired:

                continue

            if group is None:

                break

            try:

                os.killpg(group, 0)  # members other than the direct child left?

            except ProcessLookupError:

                break

# --- behaviour execution ----------------------------------------------------

def _leg(legs: dict, name: str, function) -> None:

    try:

        detail = function()

        legs[name] = {"status": "fail", "detail": detail} if detail else {"status": "pass"}

    except CapabilityUnavailable as exc:

        legs[name] = {"status": "unmeasured", "detail": str(exc)}

    except LegFailure as exc:

        legs[name] = {"status": "fail", "detail": str(exc)}

def run_behaviour(subject, behaviour: Behaviour, root: Path, observations: dict) -> tuple[dict, dict | None]:

    """Execute one behaviour; returns (legs, committed-manifest-of-final-state)."""

    legs = {}

    source = root / "src"

    source.mkdir()

    subject.init(source)

    step_manifests = []

    early_clone = root / "early"

    final_committed = None

    for index, step in enumerate(behaviour.steps):

        for op in step:

            apply_op(source, op)

        expected = fs_manifest(source)

        step_manifests.append(expected)

        label = f"commit_read@{index + 1}" if len(behaviour.steps) > 1 else "commit_read"

        try:

            subject.commit(source, f"conformance {behaviour.id} step {index + 1}")

            subject.push(source)

        except LegFailure as exc:

            legs["commit_push" + (f"@{index + 1}" if len(behaviour.steps) > 1 else "")] = {

                "status": "fail", "detail": str(exc)}

            return legs, None

        committed_holder = {}

        def check_commit(expected=expected):

            manifest, info = subject.committed(source)

            committed_holder["manifest"], committed_holder["info"] = manifest, info

            return diff_manifests(expected, manifest)

        _leg(legs, label, check_commit)

        suffix = f"@{index + 1}" if len(behaviour.steps) > 1 else ""

        _leg(legs, "source_preserved" + suffix, lambda expected=expected: diff_manifests(expected, fs_manifest(source)))

        final_committed = committed_holder.get("manifest")

        if committed_holder.get("info"):

            observations.setdefault("committed_reader", committed_holder["info"])

        if index == 0 and len(behaviour.steps) > 1:

            try:

                subject.clone(source, early_clone)

            except LegFailure as exc:

                legs["pull_fs"] = {"status": "fail", "detail": "clone for pull leg failed: " + str(exc)}

    expected = step_manifests[-1]

    clone = root / "clone"

    try:

        subject.clone(source, clone)

    except LegFailure as exc:

        legs["clone_fs"] = {"status": "fail", "detail": str(exc)}

    else:

        dirs = [] if behaviour.observe_dirs else None

        _leg(legs, "clone_fs", lambda: diff_manifests(expected, fs_manifest(clone, dirs)))

        if dirs is not None:

            observations["clone_dirs"] = dirs

        def check_clone_read():

            manifest, info = subject.committed(clone)

            observations["clone_reader"] = info

            return diff_manifests(expected, manifest)

        _leg(legs, "clone_read", check_clone_read)

        _leg(legs, "clone_status_clean", lambda: subject.status_clean(clone))

    if len(behaviour.steps) > 1 and "pull_fs" not in legs:

        try:

            subject.pull(early_clone)

        except LegFailure as exc:

            legs["pull_fs"] = {"status": "fail", "detail": str(exc)}

        else:

            dirs = [] if behaviour.observe_dirs else None

            _leg(legs, "pull_fs", lambda: diff_manifests(expected, fs_manifest(early_clone, dirs)))

            if dirs is not None:

                observations["pull_dirs"] = dirs

            _leg(legs, "pull_read", lambda: diff_manifests(expected, subject.committed(early_clone)[0]))

            _leg(legs, "pull_status_clean", lambda: subject.status_clean(early_clone))

    return legs, final_committed

def _binary_identity(binary: str, kind: str = "oak") -> dict:

    resolved = shutil.which(binary)

    if resolved is None:

        return {}

    extra = {}

    if kind == "git":

        # Same rule as the timed lanes (ADR-0011): run the real Git, never the xcrun shim.

        from oakbench.subjects import is_xcrun_git_shim, resolve_real_git

        if is_xcrun_git_shim(resolved):

            real = resolve_real_git()

            extra = {"git_requested_bin": resolved, "git_shim_resolved": real is not None}

            resolved = str(real) if real else resolved

    resolved = str(Path(resolved).resolve())

    return {"binary": resolved, "binary_sha256": hashlib.sha256(Path(resolved).read_bytes()).hexdigest(), **extra}

def base_row(kind: str, behaviour: Behaviour) -> dict:

    return {"conformance_schema_version": SCHEMA_VERSION, "evidence_class": "diagnostic_conformance",

            "matrix_version": MATRIX_VERSION, "oracle": ORACLE, "subject": kind, "scenario": behaviour.id,

            "behaviour_group": behaviour.group, "summary": behaviour.summary, "fixture_version": 1,

            "prior_art": list(behaviour.prior_art), "wall_ms": None, "tokens": None, "integrity_ok": None}

def run_matrix(workdir: Path, subjects: dict, behaviours=None, include_gated: bool | None = None,

               progress=None) -> list[dict]:

    """Run behaviours for {kind: binary}; git first so Oak rows carry the differential."""

    workdir = Path(workdir).resolve()

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

    if include_gated is None:

        include_gated = os.environ.get(HUGE_ENV) == "1"

    selected = [BY_ID[name] for name in behaviours] if behaviours else list(BEHAVIOURS)

    caps = probe_capabilities(workdir)

    profile = fs_profile(caps)

    rows = []

    git_committed = {}

    order = [kind for kind in ("git", "oak") if kind in subjects]

    for kind in order:

        identity = _binary_identity(subjects[kind], kind)

        home = workdir / f"home-{kind}"

        home.mkdir(exist_ok=True)

        env = base_env(home=home)

        env["OAK_NO_UPDATE_CHECK"] = "1"

        if not identity:

            rows += [{**base_row(kind, b), "fs_profile": profile, "binary_sha256": None, "returncode": 77,

                      "skip_reason": f"binary unavailable: {subjects[kind]}"} for b in selected]

            continue

        version = subprocess.run([identity["binary"], "--version"], capture_output=True, env=env,

                                 timeout=30).stdout.decode("utf-8", "replace").strip()

        server = LoopbackOakServer(identity["binary"], workdir / "oak-serve-data", env) if kind == "oak" else None

        try:

            if server is not None:

                server.__enter__()

            for behaviour in selected:

                row = {**base_row(kind, behaviour), **identity, "version": version, "fs_profile": profile,

                       "capabilities": {k: v for k, v in caps.items() if k in behaviour.requires}}

                missing = sorted(cap for cap in behaviour.requires if not caps.get(cap))

                if behaviour.gated and not include_gated:

                    row.update(returncode=77, skip_reason=f"gated: set {HUGE_ENV}=1 to run")

                elif missing:

                    row.update(returncode=77, skip_reason="filesystem capability unavailable: " + ",".join(missing))

                elif behaviour.gated and caps.get("free_bytes", 0) < 8 * 1024 ** 3:

                    row.update(returncode=77, skip_reason="gated behaviour needs >= 8 GiB free disk")

                else:

                    runner = Runner(identity["binary"], env, HUGE_TIMEOUT_S if behaviour.gated else COMMAND_TIMEOUT_S)

                    subject = GitSubject(runner, workdir) if kind == "git" else OakSubject(runner, workdir, server)

                    observations = {}

                    started = time.monotonic()

                    with tempfile.TemporaryDirectory(prefix=f"{kind}-{behaviour.id}-", dir=workdir) as temporary:

                        try:

                            legs, committed = run_behaviour(subject, behaviour, Path(temporary), observations)

                        except (OSError, ValueError, KeyError, json.JSONDecodeError) as exc:

                            legs, committed = {"harness": {"status": "error", "detail": repr(exc)}}, None

                    row["legs"] = legs

                    row["observations"] = observations

                    row["commands"] = runner.commands

                    row["harness_seconds"] = round(time.monotonic() - started, 2)

                    statuses = {leg["status"] for leg in legs.values()}

                    failed = sorted(name for name, leg in legs.items() if leg["status"] in ("fail", "error"))

                    row["failed_legs"] = failed

                    if "error" in statuses:

                        row.update(returncode=2, failure="harness error: " + legs["harness"]["detail"])

                    elif failed:

                        first = legs[failed[0]]

                        row.update(returncode=1, integrity_ok=False, failure=f"{failed[0]}: {first.get('detail')}")

                    elif behaviour.gated and "unmeasured" in statuses and all(

                            leg["status"] == "pass" or "reader bound" in str(leg.get("detail"))

                            for leg in legs.values()):

                        # Gated huge behaviour: filesystem legs are measured; the

                        # reader legs are over the reader's bounds by design.

                        row.update(returncode=77, skip_reason="gated: filesystem legs passed; reader legs over "

                                   "reader bound (content not independently re-hashed)")

                    elif "unmeasured" in statuses:

                        row.update(returncode=77, skip_reason="leg unmeasured: " + "; ".join(

                            f"{name}: {leg['detail']}" for name, leg in legs.items() if leg["status"] == "unmeasured"))

                    else:

                        row.update(returncode=0, integrity_ok=True)

                    if kind == "git" and committed is not None:

                        git_committed[behaviour.id] = committed

                    if kind == "oak":

                        reference = git_committed.get(behaviour.id)

                        if reference is None or committed is None:

                            row["git_differential"] = "unavailable"

                        else:

                            difference = diff_manifests(reference, committed)

                            row["git_differential"] = "identical" if difference is None else "differs: " + difference

                rows.append(row)

                if progress:

                    progress(row)

        finally:

            if server is not None:

                server.__exit__(None, None, None)

    return rows