oak/benchmarks

"""Independent behavioral conformance probes, never performance evidence.

See docs/vcs-conformance.md for the contract and upstream behavioral indexes.

"""

from __future__ import annotations

import hashlib

import os

from pathlib import Path

import shutil

import stat

import subprocess

import tempfile

from oakbench.environment import base_env

SCENARIOS = (

    "path_roundtrip_v1", "byte_roundtrip_v1", "mode_symlink_roundtrip_v1",

    "scoped_restore_preservation_v1", "failed_restore_preservation_v1",

)

class CapabilityUnavailable(Exception):

    pass

def tree_manifest(root: Path) -> dict:

    """Exact path/type/content projection; never follows directory symlinks.

    VCS directories at the repository root are excluded. Empty directories and

    non-executable permission bits are outside the portable VCS contract.

    """

    result = {}

    def visit(directory: Path) -> None:

        for path in sorted(directory.iterdir()):

            if directory == root and path.name in {".git", ".oak"}:

                continue

            mode = path.lstat().st_mode

            name = path.relative_to(root).as_posix()

            if stat.S_ISLNK(mode):

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

            elif stat.S_ISDIR(mode):

                visit(path)

            elif stat.S_ISREG(mode):

                data = path.read_bytes()

                result[name] = {"kind": "file", "size": len(data),

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

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

            else:

                raise AssertionError(f"unexpected file type: {name!r}")

    visit(root)

    return result

def assert_state(expected: dict, actual: dict) -> None:

    if expected != actual:

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

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

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

        raise AssertionError(f"state mismatch: missing={missing!r}, extra={extra!r}, changed={changed!r}")

class Repository:

    def __init__(self, root: Path, kind: str, binary: str, home: Path):

        self.root, self.kind, self.binary = root, kind, binary

        self.env = base_env(home=home)

        self.commands = []

    def run(self, *args: str, success: bool = True) -> subprocess.CompletedProcess:

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

                                 stdin=subprocess.DEVNULL, capture_output=True, timeout=30)

        self.commands.append({"argv": list(args), "returncode": process.returncode,

                              "stdout": process.stdout.decode("utf-8", "backslashreplace"),

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

        if success and process.returncode:

            raise AssertionError(f"command {args!r} failed ({process.returncode}): {process.stderr!r}")

        return process

    def initialize(self):

        self.run("init")

        if self.kind == "git":

            self.run("config", "core.autocrlf", "false")

            self.run("config", "core.filemode", "true")

    def commit(self):

        if self.kind == "git":

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

            self.run("commit", "--no-verify", "-m", "independent conformance fixture v1")

        else:

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

    def identity(self):

        if self.kind == "git":

            return {"head": self.run("rev-parse", "HEAD").stdout.hex(),

                    "branch": self.run("symbolic-ref", "HEAD").stdout.hex(),

                    "index": self.run("ls-files", "--stage", "-z").stdout.hex()}

        # Read-only logical identity; storage bookkeeping is not an atomicity oracle.

        # The subject's own claim (status) must agree with the committed state

        # read independently of the Oak binary (HEAD, branch, root tree).

        import json

        from oakbench import oak_sqlite_reader as reader

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

        identity = {"head": status["head"], "branch": status["branch"]}

        try:

            committed = reader.read_committed_tree(self.root, None)

        except reader.Invalid as exc:

            raise AssertionError(f"independent reader: committed state invalid: {exc}") from exc

        except reader.Unavailable as exc:

            identity["independent_reader"] = f"unavailable: {exc}"

            return identity

        if (committed["head"], committed["branch"]) != (status["head"], status["branch"]):

            raise AssertionError("oak status disagrees with independently read HEAD/branch")

        identity["root_tree"] = committed["root_tree"]

        return identity

    def restore(self, paths: list[str], invalid_source: bool = False):

        source = "conformance-definitely-missing-source-v1" if invalid_source else "HEAD"

        if self.kind == "git":

            return self.run("restore", "--source", source, "--", *paths,

                            success=not invalid_source)

        source_args = ["--source", source] if invalid_source else []

        return self.run("restore", "--force", *source_args, "--", *paths,

                        success=not invalid_source)

def fixture(root: Path, scenario: str) -> None:

    files = {"target.txt": b"baseline payload\n", "other.txt": b"separate payload\n"}

    if scenario == "path_roundtrip_v1":

        files = {name: (str(i) + "\nindependent path sentinel\n").encode()

                 for i, name in enumerate(("space name", "tab\tname", "line\nbreak",

                                           "quote\"name", "-leading", "unicode-λ", "nested/leaf"))}

    elif scenario == "byte_roundtrip_v1":

        files = {"empty": b"", "binary": bytes(range(256)) * 17,

                 "mixed-newlines": b"first\r\nsecond\nthird\rfinal", "no-final-lf": b"tail"}

    for name, payload in files.items():

        path = root / name

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

        path.write_bytes(payload)

    if scenario == "mode_symlink_roundtrip_v1":

        if os.name != "posix":

            raise CapabilityUnavailable("POSIX executable-bit contract unavailable")

        (root / "target.txt").chmod(0o755)

        if not ((root / "target.txt").stat().st_mode & 0o111):

            raise CapabilityUnavailable("filesystem does not retain executable bit")

        try:

            (root / "link").symlink_to("target.txt")

            (root / "dangling").symlink_to("absent-target")

        except (NotImplementedError, OSError) as exc:

            raise CapabilityUnavailable(f"filesystem symlink creation unavailable: {exc}") from exc

def exercise(repo: Repository, scenario: str) -> None:

    fixture(repo.root, scenario)

    expected = tree_manifest(repo.root)

    repo.commit()

    identity = repo.identity()

    if scenario.endswith("roundtrip_v1"):

        # Remove every fixture entry: a no-op restore cannot pass this oracle.

        for name in expected:

            (repo.root / name).unlink()

        repo.restore(list(expected))

        assert_state(expected, tree_manifest(repo.root))

        assert_state(identity, repo.identity())

        return

    (repo.root / "target.txt").write_bytes(b"discard only this change\n")

    (repo.root / "other.txt").write_bytes(b"preserve my unrelated work\x00\n")

    (repo.root / "untracked.txt").write_bytes(b"not yet checkpointed\n")

    before = tree_manifest(repo.root)

    identity = repo.identity()

    if scenario == "scoped_restore_preservation_v1":

        repo.restore(["target.txt"])

        before["target.txt"] = expected["target.txt"]

    else:

        failed = repo.restore(["target.txt"], invalid_source=True)

        if failed.returncode <= 0:

            raise AssertionError("invalid source must fail normally, not succeed or terminate by signal")

    assert_state(before, tree_manifest(repo.root))

    assert_state(identity, repo.identity())

def run_scenario(workdir: Path, kind: str, binary: str, scenario: str) -> dict:

    if kind not in {"git", "oak"} or scenario not in SCENARIOS:

        raise ValueError("unsupported subject or scenario")

    row = {"conformance_schema_version": 1, "evidence_class": "diagnostic_conformance",

           "subject": kind, "scenario": scenario, "fixture_version": 1,

           "wall_ms": None, "tokens": None, "integrity_ok": None,

           "oracle": "exact_tree_and_logical_identity_v2", "commands": []}

    resolved = shutil.which(binary)

    if resolved is None:

        return {**row, "returncode": 77, "skip_reason": f"binary unavailable: {binary}"}

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

    row["binary"] = resolved

    row["binary_sha256"] = hashlib.sha256(Path(resolved).read_bytes()).hexdigest()

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

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

        root = Path(temporary) / "repo"

        root.mkdir()

        repo = Repository(root, kind, resolved, Path(temporary) / "home")

        row["commands"] = repo.commands

        try:

            row["version"] = repo.run("--version").stdout.decode("utf-8", "replace").strip()

            repo.initialize()

            exercise(repo, scenario)

        except CapabilityUnavailable as exc:

            row.update(returncode=77, skip_reason=str(exc))

        except (AssertionError, OSError, ValueError, KeyError, subprocess.TimeoutExpired) as exc:

            row.update(returncode=1, integrity_ok=False, failure=str(exc))

        else:

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

    return row