lazaret/tools/ignored_workspace_benchmark.py

import argparse
import base64
import hashlib
import json
import os
import platform
import resource
import stat
import struct
import subprocess
import sys
import time
import zipfile
from pathlib import Path, PurePosixPath


ROOT = Path(__file__).resolve().parents[1]
CONTRACT_PATH = ROOT / "tests/fixtures/ignored-workspace-v1/benchmark.json"
FIXTURE_CONTEXT = b"lazaret-ignored-workspace-fixture-v1\0"
RESULT_CONTEXT = b"lazaret-ignored-workspace-result-v1\0"


def require(condition, message):
    if not condition:
        raise AssertionError(message)


def load_contract(path):
    contract = json.loads(Path(path).read_text())
    require(contract["schema"] == "lazaret-ignored-workspace-benchmark-v1", "unsupported benchmark contract")
    return contract


def workload(contract, identifier):
    matches = [candidate for candidate in contract["workloads"] if candidate["id"] == identifier]
    require(len(matches) == 1, f"unknown workload: {identifier}")
    return matches[0]


def derived_bytes(seed, path, length=64):
    digest = hashlib.sha256(seed.encode() + b"/" + os.fsencode(path)).digest()
    return (digest * ((length + len(digest) - 1) // len(digest)))[:length]


def write(path, data, mode=0o644):
    path.parent.mkdir(parents=True, exist_ok=True)
    path.write_bytes(data)
    path.chmod(mode)


def backing_paths(arguments):
    visible = Path(arguments.visible_root)
    if arguments.host_root is None:
        require(arguments.isolated_root is None, "isolated root requires a host root")
        visible.mkdir(parents=True, exist_ok=False)
        return visible, visible, visible
    host = Path(arguments.host_root)
    isolated = Path(arguments.isolated_root)
    require(host != isolated, "host and isolated roots must differ")
    host.mkdir(parents=True, exist_ok=False)
    isolated.mkdir(parents=True, exist_ok=False)
    return visible, host, isolated


def initialize_routed_workspace(arguments, host):
    host.mkdir(parents=True, exist_ok=False)
    git(arguments, host, "init", "--quiet", "--initial-branch=benchmark")
    patterns = [f"/bench/metadata/d{index:03d}/" for index in range(1, 100, 2)]
    patterns.extend(["/bench/uv/.venv/", "/bench/cargo/target/"])
    write(host / ".gitignore", ("\n".join(patterns) + "\n").encode())


def materialize_metadata(specification, host, isolated, direct):
    relative = Path("bench/metadata")
    seed = specification["fixture_seed"]
    for directory_index in range(100):
        directory = Path(f"d{directory_index:03d}")
        backing = host if direct or directory_index % 2 == 0 else isolated
        for file_index in range(100):
            file = directory / f"f{file_index:03d}"
            write(backing / relative / file, derived_bytes(seed, file))
    return relative


def git(arguments, root, *values, check=True):
    environment = dict(os.environ)
    environment.update({
        "GIT_CONFIG_NOSYSTEM": "1",
        "GIT_AUTHOR_NAME": "Lazaret Benchmark",
        "GIT_AUTHOR_EMAIL": "[email protected]",
        "GIT_COMMITTER_NAME": "Lazaret Benchmark",
        "GIT_COMMITTER_EMAIL": "[email protected]",
        "GIT_AUTHOR_DATE": "1700000000 +0000",
        "GIT_COMMITTER_DATE": "1700000000 +0000",
    })
    return subprocess.run(
        [arguments.git, *values], cwd=root, env=environment, check=check,
        stdout=subprocess.PIPE, stderr=subprocess.PIPE,
    )


def materialize_git(arguments, specification, host, isolated, direct):
    root = host / "bench/git"
    root.mkdir(parents=True)
    git(arguments, root, "init", "--quiet", "--initial-branch=benchmark")
    git(arguments, root, "config", "user.name", "Lazaret Benchmark")
    git(arguments, root, "config", "user.email", "[email protected]")
    write(root / ".gitignore", b"ignored/\n")
    seed = specification["fixture_seed"]
    for index in range(1000):
        path = Path("tracked") / f"t{index:04d}"
        write(root / path, derived_bytes(seed, path))
    git(arguments, root, "add", "--", ".gitignore", "tracked")
    git(arguments, root, "commit", "--quiet", "-m", "fixture")
    for index in range(500):
        ordinary = Path("ordinary") / f"u{index:04d}"
        ignored = Path("ignored") / f"i{index:04d}"
        write(root / ordinary, derived_bytes(seed, ordinary))
        if direct:
            write(root / ignored, b"isolated:" + derived_bytes(seed, ignored, 32))
        else:
            write(root / ignored, b"host-sentinel:" + derived_bytes(seed, ignored, 32))
            write(isolated / "bench/git" / ignored, b"isolated:" + derived_bytes(seed, ignored, 32))
    return Path("bench/git")


def wheel_record_line(path, payload):
    digest = base64.urlsafe_b64encode(hashlib.sha256(payload).digest()).rstrip(b"=").decode()
    return f"{path},sha256={digest},{len(payload)}"


def build_wheel(wheelhouse, package_index, seed):
    distribution = f"lazaret_bench_pkg_{package_index:02d}"
    module = distribution
    version = "1.0.0"
    dist_info = f"{distribution}-{version}.dist-info"
    module_path = f"{module}/__init__.py"
    value = hashlib.sha256(seed.encode() + distribution.replace("_", "-").encode()).hexdigest()
    module_bytes = f'VALUE = "{value}"\n'.encode()
    requirements = "" if package_index else "".join(
        f"Requires-Dist: lazaret-bench-pkg-{dependency:02d} == 1.0.0\n" for dependency in range(1, 32)
    )
    metadata = (
        "Metadata-Version: 2.1\n"
        f"Name: lazaret-bench-pkg-{package_index:02d}\n"
        f"Version: {version}\n"
        f"{requirements}\n"
    ).encode()
    wheel = b"Wheel-Version: 1.0\nGenerator: lazaret\nRoot-Is-Purelib: true\nTag: py3-none-any\n"
    entries = {
        module_path: module_bytes,
        f"{dist_info}/METADATA": metadata,
        f"{dist_info}/WHEEL": wheel,
    }
    record_path = f"{dist_info}/RECORD"
    records = [wheel_record_line(path, payload) for path, payload in sorted(entries.items())]
    records.append(f"{record_path},,")
    entries[record_path] = ("\n".join(records) + "\n").encode()
    destination = wheelhouse / f"{distribution}-{version}-py3-none-any.whl"
    with zipfile.ZipFile(destination, "w", compression=zipfile.ZIP_DEFLATED, compresslevel=9) as archive:
        for path, payload in sorted(entries.items()):
            info = zipfile.ZipInfo(path, (2020, 1, 1, 0, 0, 0))
            info.compress_type = zipfile.ZIP_DEFLATED
            info.external_attr = 0o100644 << 16
            archive.writestr(info, payload)


def materialize_uv(arguments, specification, host, isolated, direct):
    root = host / "bench/uv"
    wheelhouse = root / "wheelhouse"
    wheelhouse.mkdir(parents=True)
    for index in range(32):
        build_wheel(wheelhouse, index, specification["fixture_seed"])
    venv = root / ".venv" if direct else isolated / "bench/uv/.venv"
    venv.parent.mkdir(parents=True, exist_ok=True)
    subprocess.run(
        [arguments.uv, "venv", "--offline", "--no-project", "--python", sys.executable, venv],
        check=True, stdout=subprocess.PIPE, stderr=subprocess.PIPE,
    )
    return Path("bench/uv")


def cargo_package_name(index):
    return f"lazaret-bench-crate-{index:02d}"


def materialize_cargo(arguments, specification, host, isolated, direct):
    root = host / "bench/cargo"
    root.mkdir(parents=True)
    members = [cargo_package_name(index) for index in range(32)]
    write(root / "Cargo.toml", ("[workspace]\nresolver = \"2\"\nmembers = [\n" + "".join(f'  "{member}",\n' for member in members) + "]\n").encode())
    seed = specification["fixture_seed"].encode()
    for index, name in enumerate(members):
        package = root / name
        dependencies = ""
        if index == 0:
            dependencies = "\n[dependencies]\n" + "".join(
                f'{cargo_package_name(value).replace("-", "_")} = {{ package = "{cargo_package_name(value)}", path = "../{cargo_package_name(value)}", version = "=1.0.0" }}\n'
                for value in range(1, 32)
            )
        write(package / "Cargo.toml", (f'[package]\nname = "{name}"\nversion = "1.0.0"\nedition = "2024"\n{dependencies}').encode())
        if index == 0:
            expression = " ^ ".join(f"{cargo_package_name(value).replace('-', '_')}::VALUE" for value in range(1, 32))
            write(package / "src/main.rs", f"fn main() {{ println!(\"{{}}\", {expression}); }}\n".encode())
        else:
            value = int.from_bytes(hashlib.sha256(seed + name.encode()).digest()[:8], "big")
            write(package / "src/lib.rs", f"pub const VALUE: u64 = {value};\n".encode())
    subprocess.run(
        [arguments.cargo, "generate-lockfile", "--offline"], cwd=root, check=True,
        stdout=subprocess.PIPE, stderr=subprocess.PIPE,
    )
    target = root / "target" if direct else isolated / "bench/cargo/target"
    target.mkdir(parents=True)
    return Path("bench/cargo")


def materialize(arguments):
    contract = load_contract(arguments.contract)
    specification = workload(contract, arguments.workload)
    visible = Path(arguments.visible_root)
    if arguments.host_root is None:
        visible, host, isolated = backing_paths(arguments)
    else:
        require(arguments.isolated_root is not None, "routed materialization requires an isolated root")
        host = Path(arguments.host_root)
        isolated = Path(arguments.isolated_root)
        require(host != isolated, "host and isolated roots must differ")
        initialize_routed_workspace(arguments, host)
        isolated.mkdir(parents=True, exist_ok=False)
    direct = arguments.host_root is None
    if specification["category"] == "metadata":
        relative = materialize_metadata(specification, host, isolated, direct)
    elif specification["category"] == "git":
        relative = materialize_git(arguments, specification, host, isolated, direct)
    elif specification["category"] == "package-manager":
        relative = materialize_uv(arguments, specification, host, isolated, direct)
    elif specification["category"] == "build":
        relative = materialize_cargo(arguments, specification, host, isolated, direct)
    else:
        raise AssertionError("unsupported workload category")
    manifest = {
        "schema": "lazaret-ignored-workspace-materialization-v1",
        "workload": arguments.workload,
        "placement": "direct" if direct else "routed",
        "visible_workload_root": os.fspath(visible / relative),
    }
    Path(arguments.output).parent.mkdir(parents=True, exist_ok=True)
    Path(arguments.output).write_text(json.dumps(manifest, indent=2, sort_keys=True) + "\n")


def fixture_identity(root, excluded_roots=()):
    records = []
    excluded = {os.fsencode(value) for value in excluded_roots}
    for directory, directories, files in os.walk(root, topdown=True, followlinks=False):
        directories.sort(key=os.fsencode)
        files.sort(key=os.fsencode)
        relative_directory = Path(directory).relative_to(root)
        if relative_directory == Path("."):
            directories[:] = [name for name in directories if os.fsencode(name) not in excluded]
            files[:] = [name for name in files if os.fsencode(name) not in excluded]
        for name in [*directories, *files]:
            path = Path(directory) / name
            relative = relative_directory / name
            metadata = os.lstat(path)
            if stat.S_ISDIR(metadata.st_mode):
                kind, payload = 1, b""
            elif stat.S_ISREG(metadata.st_mode):
                kind, payload = 2, path.read_bytes()
            elif stat.S_ISLNK(metadata.st_mode):
                kind, payload = 3, os.fsencode(os.readlink(path))
            else:
                raise AssertionError(f"unsupported fixture entry: {relative}")
            records.append((os.fsencode(relative), kind, metadata.st_mode & 0o7777, payload))
    records.sort(key=lambda record: record[0])
    stream = bytearray(FIXTURE_CONTEXT)
    stream.extend(struct.pack(">I", len(records)))
    for path, kind, mode, payload in records:
        stream.extend(struct.pack(">I", len(path)))
        stream.extend(path)
        stream.extend(struct.pack(">BIQ", kind, mode, len(payload)))
        stream.extend(hashlib.sha256(payload).digest())
    return hashlib.sha256(stream).digest()


def workload_commands(arguments, identifier, root):
    if identifier == "metadata-10000":
        return [[arguments.metadata_benchmark, "metadata", "--root", root, "--bytewise-order", "--statx-mask", "basic-stats"]]
    if identifier == "git-2000-status-add-reset":
        return [
            [arguments.git, "status", "--porcelain=v2", "--untracked-files=all"],
            [arguments.git, "add", "--", "ordinary"],
            [arguments.git, "reset", "--quiet", "--mixed"],
            [arguments.git, "status", "--porcelain=v2", "--untracked-files=all"],
        ]
    if identifier == "uv-offline-32-wheels":
        return [[arguments.uv, "pip", "install", "--offline", "--no-index", "--find-links", root / "wheelhouse", "--python", root / ".venv/bin/python", "lazaret-bench-pkg-00==1.0.0"]]
    if identifier == "cargo-offline-32-crates":
        return [[arguments.cargo, "check", "--offline", "--locked", "--workspace", "--target-dir", root / "target"]]
    raise AssertionError(f"unsupported workload: {identifier}")


def workload_state(arguments, specification, root):
    if specification["category"] == "git":
        index = root / ".git/index"
        state = {"index_sha256": hashlib.sha256(index.read_bytes()).hexdigest()}
        if arguments.host_root is not None:
            sentinels = Path(arguments.host_root) / "bench/git/ignored"
            state["ignored_host_sentinels_identity"] = fixture_identity(sentinels).hex()
        return state
    if specification["category"] == "build":
        return {
            "source_identity": fixture_identity(root, ("target",)).hex(),
            "lock_sha256": hashlib.sha256((root / "Cargo.lock").read_bytes()).hexdigest(),
        }
    if specification["category"] == "package-manager":
        candidates = list((root / ".venv").glob("lib/python*/site-packages"))
        if not candidates:
            return {"installed_distribution_count": 0, "installed_values_identity": None}
        site_packages = candidates[0]
        distributions = sorted(site_packages.glob("lazaret_bench_pkg_*-1.0.0.dist-info"))
        values = []
        for index in range(32):
            module = site_packages / f"lazaret_bench_pkg_{index:02d}/__init__.py"
            if module.exists():
                values.append(module.read_bytes())
        return {
            "installed_distribution_count": len(distributions),
            "installed_values_identity": hashlib.sha256(b"".join(values)).hexdigest() if len(values) == 32 else None,
        }
    return {}


def verify_uv_installation(root, seed):
    site_packages = next((root / ".venv").glob("lib/python*/site-packages"), None)
    require(site_packages is not None, "installed virtual environment has no site-packages")
    distributions = list(site_packages.glob("lazaret_bench_pkg_*-1.0.0.dist-info"))
    require(len(distributions) == 32, "offline installation did not install all 32 distributions")
    for index in range(32):
        normalized = f"lazaret-bench-pkg-{index:02d}"
        expected = hashlib.sha256(seed.encode() + normalized.encode()).hexdigest()
        module = site_packages / normalized.replace("-", "_") / "__init__.py"
        require(module.read_text() == f'VALUE = "{expected}"\n', f"installed module {normalized} differs")


def invoke(arguments, specification, capture):
    root = Path(arguments.root)
    initial_identity = fixture_identity(root)
    initial_state = workload_state(arguments, specification, root)
    records = []
    started = time.clock_gettime_ns(time.CLOCK_MONOTONIC_RAW)
    timeout_seconds = 4 * specification["absolute_median_ms"] / 1000
    for argv in workload_commands(arguments, arguments.workload, root):
        elapsed_seconds = (time.clock_gettime_ns(time.CLOCK_MONOTONIC_RAW) - started) / 1_000_000_000
        remaining = timeout_seconds - elapsed_seconds
        require(remaining > 0, "workload exceeded its sample timeout")
        result = subprocess.run(
            [os.fspath(value) for value in argv], cwd=root, stdout=subprocess.PIPE,
            stderr=subprocess.PIPE, timeout=remaining,
        )
        require(result.returncode == 0, f"workload command failed: {result.stderr.decode(errors='replace')}")
        if capture:
            records.append({"status": result.returncode, "stdout": result.stdout, "stderr": result.stderr})
    elapsed = time.clock_gettime_ns(time.CLOCK_MONOTONIC_RAW) - started
    post_identity = fixture_identity(root)
    post_state = workload_state(arguments, specification, root)
    if specification["category"] == "package-manager":
        verify_uv_installation(root, specification["fixture_seed"])
    if not capture:
        return None
    stream = bytearray(RESULT_CONTEXT)
    workload_bytes = arguments.workload.encode()
    stream.extend(struct.pack(">I", len(workload_bytes)))
    stream.extend(workload_bytes)
    stream.extend(initial_identity)
    stream.extend(struct.pack(">I", len(records)))
    for record in records:
        stream.extend(struct.pack(">iQ", record["status"], len(record["stdout"])))
        stream.extend(record["stdout"])
        stream.extend(struct.pack(">Q", len(record["stderr"])))
        stream.extend(record["stderr"])
    stream.extend(post_identity)
    return {
        "elapsed_ns": elapsed,
        "initial_fixture_identity": initial_identity.hex(),
        "post_fixture_identity": post_identity.hex(),
        "initial_workload_state": initial_state,
        "post_workload_state": post_state,
        "result_identity": hashlib.sha256(stream).hexdigest(),
        "commands": [
            {
                "status": record["status"],
                "stdout_base64": base64.b64encode(record["stdout"]).decode(),
                "stderr_base64": base64.b64encode(record["stderr"]).decode(),
            }
            for record in records
        ],
    }


def cpu_snapshot():
    fields = Path("/proc/stat").read_text().splitlines()[0].split()[1:]
    usage = resource.getrusage(resource.RUSAGE_SELF)
    children = resource.getrusage(resource.RUSAGE_CHILDREN)
    return {
        "system_ticks": sum(int(value) for value in fields),
        "idle_ticks": int(fields[3]) + int(fields[4]),
        "owned_seconds": usage.ru_utime + usage.ru_stime + children.ru_utime + children.ru_stime,
    }


def memory_pressure_snapshot():
    pressure = {}
    path = Path("/sys/fs/cgroup/memory.pressure")
    if path.exists():
        for line in path.read_text().splitlines():
            category, *fields = line.split()
            pressure[category] = {key: int(value) for key, value in (field.split("=") for field in fields) if key == "total"}["total"]
    return pressure


def run_sample(arguments):
    contract = load_contract(arguments.contract)
    specification = workload(contract, arguments.workload)
    require(arguments.preparation_state in {"unprimed", "primed"}, "invalid preparation state")
    sequence_started_ns = time.clock_gettime_ns(time.CLOCK_MONOTONIC_RAW)
    if arguments.preparation_state == "primed":
        invoke(arguments, specification, False)
    cpu_before = cpu_snapshot()
    pressure_before = memory_pressure_snapshot()
    sample = invoke(arguments, specification, True)
    pressure_after = memory_pressure_snapshot()
    cpu_after = cpu_snapshot()
    elapsed_seconds = sample["elapsed_ns"] / 1_000_000_000
    ticks_per_second = os.sysconf("SC_CLK_TCK")
    total_busy_seconds = (
        (cpu_after["system_ticks"] - cpu_before["system_ticks"])
        - (cpu_after["idle_ticks"] - cpu_before["idle_ticks"])
    ) / ticks_per_second
    owned_seconds = max(0.0, cpu_after["owned_seconds"] - cpu_before["owned_seconds"])
    external_seconds = max(0.0, total_busy_seconds - owned_seconds)
    pressure_delta = {key: pressure_after.get(key, 0) - pressure_before.get(key, 0) for key in pressure_after.keys() | pressure_before.keys()}
    sample.update({
        "schema": "lazaret-ignored-workspace-benchmark-sample-v1",
        "workload": arguments.workload,
        "placement": arguments.placement,
        "preparation_state": arguments.preparation_state,
        "independent_run": arguments.independent_run,
        "sample_kind": arguments.sample_kind,
        "sample_index": arguments.sample_index,
        "pair_index": arguments.pair_index,
        "fixture_id": arguments.fixture_id,
        "sequence_started_ns": sequence_started_ns,
        "background_load": {
            "external_cpu_seconds": external_seconds,
            "reference_elapsed_seconds": elapsed_seconds,
            "memory_pressure_total_delta": pressure_delta,
        },
    })
    Path(arguments.output).parent.mkdir(parents=True, exist_ok=True)
    Path(arguments.output).write_text(json.dumps(sample, indent=2, sort_keys=True) + "\n")


def plan(arguments):
    contract = load_contract(arguments.contract)
    sampling = contract["sampling"]
    entries = []
    for independent_run in range(sampling["independent_runs"]):
        for specification in contract["workloads"]:
            for preparation_state in specification["preparation_states"]:
                for sample_kind, count in (
                    ("warmup", sampling["warmup_samples_per_cell"]),
                    ("measured", sampling["measured_samples_per_cell"]),
                ):
                    for sample_index in range(count):
                        pair_index = len(entries) // 2
                        placements = ["direct", "routed"] if pair_index % 2 == 0 else ["routed", "direct"]
                        for placement in placements:
                            entries.append({
                                "workload": specification["id"],
                                "preparation_state": preparation_state,
                                "independent_run": independent_run,
                                "sample_kind": sample_kind,
                                "sample_index": sample_index,
                                "pair_index": pair_index,
                                "placement": placement,
                                "fixture_id": f"run-{independent_run}-{specification['id']}-{preparation_state}-{sample_kind}-{sample_index}-{placement}",
                            })
    output = {"schema": "lazaret-ignored-workspace-benchmark-plan-v1", "samples": entries}
    Path(arguments.output).parent.mkdir(parents=True, exist_ok=True)
    Path(arguments.output).write_text(json.dumps(output, indent=2, sort_keys=True) + "\n")


def version_output(argv):
    result = subprocess.run(argv, check=True, stdout=subprocess.PIPE, stderr=subprocess.STDOUT, text=True)
    return result.stdout.strip()


def cpu_information():
    values = {}
    processors = []
    current = {}
    for line in Path("/proc/cpuinfo").read_text().splitlines() + [""]:
        if not line:
            if current:
                processors.append(current)
                current = {}
            continue
        key, separator, value = line.partition(":")
        if separator:
            current[key.strip()] = value.strip()
    allowed = sorted(os.sched_getaffinity(0))
    topology = []
    for cpu in allowed:
        base = Path(f"/sys/devices/system/cpu/cpu{cpu}")
        topology.append({
            "logical_cpu": cpu,
            "package": int((base / "topology/physical_package_id").read_text()),
            "core": int((base / "topology/core_id").read_text()),
            "governor": (base / "cpufreq/scaling_governor").read_text().strip(),
        })
    first = processors[0]
    values["cpu_model"] = first.get("model name", "")
    values["cpu_microcode"] = first.get("microcode", "")
    values["allocated_cpu_topology"] = topology
    return values


def decode_mount_path(value):
    return value.replace("\\040", " ").replace("\\011", "\t").replace("\\012", "\n").replace("\\134", "\\")


def mount_information(root):
    root = Path(root).resolve()
    candidates = []
    for line in Path("/proc/self/mountinfo").read_text().splitlines():
        left, right = line.split(" - ", 1)
        fields = left.split()
        mountpoint = Path(decode_mount_path(fields[4]))
        try:
            root.relative_to(mountpoint)
        except ValueError:
            continue
        right_fields = right.split()
        candidates.append((len(mountpoint.parts), {
            "mountpoint": os.fspath(mountpoint),
            "source": decode_mount_path(right_fields[1]),
            "filesystem_type": right_fields[0],
            "filesystem_mount_options": sorted(set(fields[5].split(",")) | set(right_fields[2].split(","))),
        }))
    require(candidates, "benchmark root has no mountinfo entry")
    return max(candidates, key=lambda candidate: candidate[0])[1]


def storage_information(source):
    require(source.startswith("/dev/"), "qualification storage is not a local block device")
    tree = json.loads(version_output(["lsblk", "--json", "--output", "PATH,TYPE,MODEL,TRAN"]))["blockdevices"]
    flattened = []

    def visit(device, parents):
        flattened.append((device, parents))
        for child in device.get("children", []):
            visit(child, [*parents, device])

    for device in tree:
        visit(device, [])
    source_real = os.path.realpath(source)
    matches = [(device, parents) for device, parents in flattened if os.path.realpath(device.get("path", "")) == source_real]
    require(matches, f"cannot identify storage device for {source}")
    device, parents = matches[0]
    stack = [*parents, device]
    models = [model.strip() for entry in stack if (model := entry.get("model")) and model.strip()]
    transports = [entry.get("tran", "") for entry in stack if entry.get("tran")]
    return {
        "storage_model": models[-1] if models else "unreported",
        "storage_transport": transports[-1] if transports else "unreported",
        "storage_stack": [entry.get("type", "unknown") for entry in stack],
    }


def current_cgroup_swap_max(
    membership_path=Path("/proc/self/cgroup"),
    cgroup_root=Path("/sys/fs/cgroup"),
):
    memberships = Path(membership_path).read_text().splitlines()
    require(len(memberships) == 1, "current cgroup data is not unified cgroup v2")
    fields = memberships[0].split(":", 2)
    require(
        len(fields) == 3 and fields[0] == "0" and fields[1] == "" and fields[2].startswith("/"),
        "current cgroup data is not unified cgroup v2",
    )
    membership = PurePosixPath(fields[2])
    require(
        all(component not in {"", ".", ".."} for component in membership.parts[1:]),
        "current cgroup path is malformed",
    )
    root = Path(cgroup_root).resolve(strict=True)
    directory = root.joinpath(*membership.parts[1:]).resolve(strict=True)
    try:
        directory.relative_to(root)
    except ValueError as error:
        raise AssertionError("current cgroup path escapes cgroup root") from error
    value = (directory / "memory.swap.max").read_text().strip()
    require(value == "max" or value.isascii() and value.isdigit(), "invalid memory.swap.max value")
    return value


def qualify(arguments):
    mount = mount_information(arguments.root)
    filesystem = os.statvfs(arguments.root)
    cpu = cpu_information()
    memory_kib = int(next(line.split()[1] for line in Path("/proc/meminfo").read_text().splitlines() if line.startswith("MemTotal:")))
    memory_max_path = Path("/sys/fs/cgroup/memory.max")
    memory_max = memory_max_path.read_text().strip() if memory_max_path.exists() else "max"
    memory_bytes = memory_kib * 1024
    if memory_max != "max":
        memory_bytes = min(memory_bytes, int(memory_max))
    swap_max = current_cgroup_swap_max()
    xfs = version_output([arguments.xfs_info, arguments.root]) if mount["filesystem_type"].lower() == "xfs" else ""
    result = {
        "schema": "lazaret-ignored-workspace-qualification-v1",
        "architecture": platform.machine(),
        "kernel_release": platform.release(),
        **cpu,
        "memory_bytes": memory_bytes,
        "benchmark_cgroup_swap_max": swap_max,
        **storage_information(mount["source"]),
        "storage_kind": "local-block-device",
        "storage_network_backed": False,
        "storage_copy_on_write": "reflink=0" not in xfs,
        "filesystem_type": mount["filesystem_type"].upper(),
        "filesystem_block_bytes": filesystem.f_frsize,
        "filesystem_mount_options": mount["filesystem_mount_options"],
        "fuse_userspace_version": version_output([arguments.fusermount, "--version"]),
        "runsc_version": version_output([arguments.runsc, "--version"]),
        "git_version": version_output([arguments.git, "--version"]),
        "uv_version": version_output([arguments.uv, "--version"]),
        "rust_version": version_output([arguments.rustc, "--version"]),
    }
    Path(arguments.output).parent.mkdir(parents=True, exist_ok=True)
    Path(arguments.output).write_text(json.dumps(result, indent=2, sort_keys=True) + "\n")


def kernel_at_least(release, minimum):
    numbers = []
    for component in release.split("."):
        digits = "".join(character for character in component if character.isdigit())
        if not digits:
            break
        numbers.append(int(digits))
        if len(numbers) == 2:
            break
    return tuple(numbers) >= minimum


def validate_qualification(contract, qualification, failures):
    expected_fields = contract["qualification_environment"]["qualification_result_fields"]
    missing = [field for field in expected_fields if field not in qualification]
    if missing:
        failures.append(f"qualification result omits fields: {missing}")
        return
    if qualification.get("architecture") != "x86_64":
        failures.append("qualification architecture is not x86_64")
    if not kernel_at_least(qualification["kernel_release"], (6, 9)):
        failures.append("qualification kernel is older than Linux 6.9")
    topology = qualification["allocated_cpu_topology"]
    physical = {(cpu["package"], cpu["core"]) for cpu in topology}
    if len(physical) < 8 or len(physical) != len(topology):
        failures.append("CPU allocation is not one hardware thread from at least eight physical cores")
    if any(cpu["governor"] != "performance" for cpu in topology):
        failures.append("an allocated CPU is not using the performance governor")
    if qualification["memory_bytes"] < 17_179_869_184:
        failures.append("qualification memory is below 16 GiB")
    if qualification.get("benchmark_cgroup_swap_max") != "0":
        failures.append("swap is not disabled for the benchmark cgroup")
    if qualification.get("storage_kind") != "local-block-device" or qualification.get("storage_network_backed") is not False:
        failures.append("qualification storage is not a local non-network block device")
    if qualification.get("storage_copy_on_write") is not False:
        failures.append("qualification storage has copy-on-write enabled")
    if qualification["filesystem_type"] != "XFS" or qualification["filesystem_block_bytes"] != 4096:
        failures.append("qualification filesystem is not 4096-byte XFS")
    mount_options = set(qualification["filesystem_mount_options"])
    if not {"relatime", "inode64"}.issubset(mount_options):
        failures.append("qualification filesystem omits relatime or inode64")
    if mount_options & {"quota", "usrquota", "grpquota", "prjquota", "uquota", "gquota", "pquota"}:
        failures.append("qualification filesystem enables quota")


def lower_median(values):
    ordered = sorted(values)
    require(ordered, "cannot calculate an empty median")
    return ordered[len(ordered) // 2]


def relative_mad(values):
    median = lower_median(values)
    require(median > 0, "non-positive median")
    return lower_median([abs(value - median) for value in values]) / median


def decode_command_records(sample):
    return [
        {
            "status": record["status"],
            "stdout": base64.b64decode(record["stdout_base64"], validate=True),
            "stderr": base64.b64decode(record["stderr_base64"], validate=True),
        }
        for record in sample["commands"]
    ]


def validate_result_identity(sample, failures):
    try:
        records = decode_command_records(sample)
        stream = bytearray(RESULT_CONTEXT)
        identifier = sample["workload"].encode()
        stream.extend(struct.pack(">I", len(identifier)))
        stream.extend(identifier)
        stream.extend(bytes.fromhex(sample["initial_fixture_identity"]))
        stream.extend(struct.pack(">I", len(records)))
        for record in records:
            stream.extend(struct.pack(">iQ", record["status"], len(record["stdout"])))
            stream.extend(record["stdout"])
            stream.extend(struct.pack(">Q", len(record["stderr"])))
            stream.extend(record["stderr"])
        stream.extend(bytes.fromhex(sample["post_fixture_identity"]))
        require(hashlib.sha256(stream).hexdigest() == sample["result_identity"], "result identity differs")
        require(all(record["status"] == 0 for record in records), "a command status is nonzero")
        return records
    except (AssertionError, KeyError, ValueError, TypeError) as error:
        failures.append(f"invalid sample {sample.get('fixture_id', '<unknown>')}: {error}")
        return []


def validate_workload_oracle(specification, sample, records, failures):
    prefix = sample.get("fixture_id", "sample")
    if not records:
        return
    if specification["category"] == "metadata":
        try:
            output = json.loads(records[0]["stdout"])
            require(output["regular_files"] == 10_000, "metadata file count differs")
            require(output["content_bytes"] == 640_000, "metadata byte count differs")
            require(output["fixture_identity"] == sample["initial_fixture_identity"], "metadata fixture identity differs")
        except (AssertionError, KeyError, ValueError, TypeError) as error:
            failures.append(f"{prefix}: metadata oracle failed: {error}")
    elif specification["category"] == "git":
        if len(records) != 4 or records[0]["stdout"] != records[3]["stdout"]:
            failures.append(f"{prefix}: Git status streams differ")
        initial = sample.get("initial_workload_state", {})
        post = sample.get("post_workload_state", {})
        if initial.get("index_sha256") != post.get("index_sha256"):
            failures.append(f"{prefix}: Git index did not return to its initial identity")
        if sample["placement"] == "routed" and initial.get("ignored_host_sentinels_identity") != post.get("ignored_host_sentinels_identity"):
            failures.append(f"{prefix}: ignored host sentinels changed")
    elif specification["category"] == "build":
        if sample.get("initial_workload_state") != sample.get("post_workload_state"):
            failures.append(f"{prefix}: Cargo changed the lockfile or source tree")
    elif specification["category"] == "package-manager":
        post = sample.get("post_workload_state", {})
        if post.get("installed_distribution_count") != 32 or post.get("installed_values_identity") is None:
            failures.append(f"{prefix}: offline installation oracle differs")


def verify_results(arguments):
    contract = load_contract(arguments.contract)
    samples = [json.loads(path.read_text()) for path in sorted(Path(arguments.samples).glob("*.json"))]
    sampling = contract["sampling"]
    failures = []
    qualification = json.loads(Path(arguments.qualification).read_text())
    validate_qualification(contract, qualification, failures)
    seen = set()
    fixture_ids = set()
    for sample in samples:
        if sample.get("schema") != "lazaret-ignored-workspace-benchmark-sample-v1":
            failures.append(f"unsupported sample schema: {sample.get('schema')}")
        key = tuple(sample.get(field) for field in (
            "workload", "preparation_state", "independent_run", "sample_kind",
            "sample_index", "pair_index", "placement",
        ))
        if key in seen:
            failures.append(f"duplicate sample coordinate: {key}")
        seen.add(key)
        fixture_id = sample.get("fixture_id")
        if fixture_id in fixture_ids:
            failures.append(f"reused fixture identity: {fixture_id}")
        fixture_ids.add(fixture_id)
        records = validate_result_identity(sample, failures)
        try:
            validate_workload_oracle(workload(contract, sample["workload"]), sample, records, failures)
        except (AssertionError, KeyError) as error:
            failures.append(f"invalid sample workload: {error}")
    cells = []
    for specification in contract["workloads"]:
        for state in specification["preparation_states"]:
            run_medians = {"direct": [], "routed": [], "ratio": []}
            for run in range(sampling["independent_runs"]):
                matching = [sample for sample in samples if sample.get("workload") == specification["id"] and sample.get("preparation_state") == state and sample.get("independent_run") == run]
                for kind, expected_count in (("warmup", sampling["warmup_samples_per_cell"]), ("measured", sampling["measured_samples_per_cell"])):
                    for placement in ("direct", "routed"):
                        observed_count = sum(sample.get("sample_kind") == kind and sample.get("placement") == placement for sample in matching)
                        if observed_count != expected_count:
                            failures.append(f"{specification['id']}/{state}/run-{run}: expected {expected_count} {kind} {placement} samples")
                for pair_index in {sample.get("pair_index") for sample in matching}:
                    pair = [sample for sample in matching if sample.get("pair_index") == pair_index]
                    if len(pair) == 2 and {sample.get("placement") for sample in pair} == {"direct", "routed"}:
                        ordered = sorted(pair, key=lambda sample: sample.get("sequence_started_ns", 0))
                        expected_first = "direct" if pair_index % 2 == 0 else "routed"
                        if ordered[0].get("placement") != expected_first:
                            failures.append(f"{specification['id']}/{state}/run-{run}/pair-{pair_index}: placement order differs")
                        if pair[0].get("initial_fixture_identity") != pair[1].get("initial_fixture_identity"):
                            failures.append(f"{specification['id']}/{state}/run-{run}/pair-{pair_index}: initial fixtures differ")
                        direct_sample = next(sample for sample in pair if sample["placement"] == "direct")
                        routed_sample = next(sample for sample in pair if sample["placement"] == "routed")
                        if specification["category"] in {"metadata", "git"}:
                            direct_output = [record["stdout"] for record in decode_command_records(direct_sample)]
                            routed_output = [record["stdout"] for record in decode_command_records(routed_sample)]
                            if direct_output != routed_output:
                                failures.append(f"{specification['id']}/{state}/run-{run}/pair-{pair_index}: routed output differs from direct oracle")
                for sample in matching:
                    if sample.get("elapsed_ns", 0) > 4 * specification["absolute_median_ms"] * 1_000_000:
                        failures.append(f"{specification['id']}/{state}/run-{run}: sample exceeded timeout")
                    background = sample.get("background_load", {})
                    if background.get("external_cpu_seconds", float("inf")) > 0.05 * background.get("reference_elapsed_seconds", 0):
                        failures.append(f"{specification['id']}/{state}/run-{run}: background CPU threshold exceeded")
                    if any(background.get("memory_pressure_total_delta", {"missing": 1}).values()):
                        failures.append(f"{specification['id']}/{state}/run-{run}: memory pressure caused reclaim")
                measured = [sample for sample in matching if sample.get("sample_kind") == "measured"]
                direct = sorted((sample for sample in measured if sample["placement"] == "direct"), key=lambda sample: sample["pair_index"])
                routed = sorted((sample for sample in measured if sample["placement"] == "routed"), key=lambda sample: sample["pair_index"])
                expected = sampling["measured_samples_per_cell"]
                if len(direct) != expected or len(routed) != expected:
                    failures.append(f"{specification['id']}/{state}/run-{run}: expected {expected} measured samples per placement")
                    continue
                if [sample["pair_index"] for sample in direct] != [sample["pair_index"] for sample in routed]:
                    failures.append(f"{specification['id']}/{state}/run-{run}: paired sample identities differ")
                    continue
                direct_times = [sample["elapsed_ns"] / 1_000_000 for sample in direct]
                routed_times = [sample["elapsed_ns"] / 1_000_000 for sample in routed]
                ratios = [right / left for left, right in zip(direct_times, routed_times, strict=True)]
                medians = {"direct": lower_median(direct_times), "routed": lower_median(routed_times), "ratio": lower_median(ratios)}
                for key in run_medians:
                    run_medians[key].append(medians[key])
                calibration = specification["absolute_median_ms"] / specification["relative_median_ratio"]
                if medians["direct"] > calibration:
                    failures.append(f"{specification['id']}/{state}/run-{run}: direct calibration {medians['direct']:.3f} ms > {calibration:.3f} ms")
                if medians["routed"] > specification["absolute_median_ms"]:
                    failures.append(f"{specification['id']}/{state}/run-{run}: routed median exceeds absolute budget")
                if medians["ratio"] > specification["relative_median_ratio"]:
                    failures.append(f"{specification['id']}/{state}/run-{run}: median ratio exceeds relative budget")
                for label, values in (("direct", direct_times), ("routed", routed_times), ("ratio", ratios)):
                    if relative_mad(values) > sampling["within_run_relative_mad_limit"]:
                        failures.append(f"{specification['id']}/{state}/run-{run}: {label} relative MAD exceeds limit")
                cells.append({"workload": specification["id"], "preparation_state": state, "independent_run": run, "medians": medians})
            if all(len(values) == sampling["independent_runs"] for values in run_medians.values()):
                for label, values in run_medians.items():
                    spread = (max(values) - min(values)) / lower_median(values)
                    if spread > sampling["between_run_median_spread_limit"]:
                        failures.append(f"{specification['id']}/{state}: {label} between-run spread exceeds limit")
    expected_samples = sum(
        len(specification["preparation_states"])
        * sampling["independent_runs"]
        * (sampling["warmup_samples_per_cell"] + sampling["measured_samples_per_cell"])
        * 2
        for specification in contract["workloads"]
    )
    if len(samples) != expected_samples:
        failures.append(f"expected {expected_samples} retained samples, observed {len(samples)}")
    result = {
        "schema": "lazaret-ignored-workspace-benchmark-result-v1",
        "disposition": "PASS" if not failures else "FAIL",
        "qualification": qualification,
        "cells": cells,
        "failures": failures,
    }
    Path(arguments.output).parent.mkdir(parents=True, exist_ok=True)
    Path(arguments.output).write_text(json.dumps(result, indent=2, sort_keys=True) + "\n")
    require(not failures, "benchmark qualification failed")


def add_tool_arguments(parser):
    parser.add_argument("--git", default="git")
    parser.add_argument("--uv", default="uv")
    parser.add_argument("--cargo", default="cargo")


def main():
    parser = argparse.ArgumentParser()
    subparsers = parser.add_subparsers(dest="command", required=True)
    prepare = subparsers.add_parser("prepare")
    prepare.add_argument("--contract", type=Path, default=CONTRACT_PATH)
    prepare.add_argument("--workload", required=True)
    prepare.add_argument("--visible-root", type=Path, required=True)
    prepare.add_argument("--host-root", type=Path)
    prepare.add_argument("--isolated-root", type=Path)
    prepare.add_argument("--output", type=Path, required=True)
    add_tool_arguments(prepare)
    schedule = subparsers.add_parser("plan")
    schedule.add_argument("--contract", type=Path, default=CONTRACT_PATH)
    schedule.add_argument("--output", type=Path, required=True)
    sample = subparsers.add_parser("sample")
    sample.add_argument("--contract", type=Path, default=CONTRACT_PATH)
    sample.add_argument("--workload", required=True)
    sample.add_argument("--root", type=Path, required=True)
    sample.add_argument("--placement", choices=("direct", "routed"), required=True)
    sample.add_argument("--preparation-state", choices=("unprimed", "primed"), required=True)
    sample.add_argument("--independent-run", type=int, required=True)
    sample.add_argument("--sample-kind", choices=("warmup", "measured"), required=True)
    sample.add_argument("--sample-index", type=int, required=True)
    sample.add_argument("--pair-index", type=int, required=True)
    sample.add_argument("--fixture-id", required=True)
    sample.add_argument("--host-root", type=Path)
    sample.add_argument("--metadata-benchmark", default="lazaret-workspace-benchmark")
    sample.add_argument("--output", type=Path, required=True)
    add_tool_arguments(sample)
    verify = subparsers.add_parser("verify")
    verify.add_argument("--contract", type=Path, default=CONTRACT_PATH)
    verify.add_argument("--samples", type=Path, required=True)
    verify.add_argument("--qualification", type=Path, required=True)
    verify.add_argument("--output", type=Path, required=True)
    qualification = subparsers.add_parser("qualify")
    qualification.add_argument("--root", type=Path, required=True)
    qualification.add_argument("--output", type=Path, required=True)
    qualification.add_argument("--fusermount", default="fusermount3")
    qualification.add_argument("--runsc", default="runsc")
    qualification.add_argument("--git", default="git")
    qualification.add_argument("--uv", default="uv")
    qualification.add_argument("--rustc", default="rustc")
    qualification.add_argument("--xfs-info", default="xfs_info")
    arguments = parser.parse_args()
    if arguments.command == "prepare":
        materialize(arguments)
    elif arguments.command == "plan":
        plan(arguments)
    elif arguments.command == "sample":
        run_sample(arguments)
    elif arguments.command == "qualify":
        qualify(arguments)
    else:
        verify_results(arguments)


if __name__ == "__main__":
    main()