#!/bin/sh
set -eu
umask 022

gate_dir=$(CDPATH= cd -- "$(dirname -- "$0")" && pwd -P)
root=$(CDPATH= cd -- "$gate_dir/../../../.." && pwd -P)
input=$gate_dir/P15-021-input.json
freebsd_src=${FREEBSD_SRC:-/work/build/freebsd-src}
utils_src=${EROFS_UTILS_SRC:-/work/build/erofs-utils-master}
base=
output=

while test "$#" -gt 0; do
	case "$1" in
	--base)
		test "$#" -ge 2 || { printf '%s\n' '--base requires a commit' >&2; exit 2; }
		base=$2
		shift 2
		;;
	--output)
		test "$#" -ge 2 || { printf '%s\n' '--output requires a directory' >&2; exit 2; }
		output=$2
		shift 2
		;;
	*)
		printf 'unknown argument: %s\n' "$1" >&2
		exit 2
		;;
	esac
done

test -n "$base" || { printf '%s\n' '--base is required' >&2; exit 2; }
test -n "$output" || { printf '%s\n' '--output is required' >&2; exit 2; }
test -f "$input" || { printf 'missing input: %s\n' "$input" >&2; exit 2; }
test -d "$freebsd_src/sys" || { printf 'missing FreeBSD source: %s\n' "$freebsd_src" >&2; exit 2; }
test -d "$utils_src/lib" || { printf 'missing erofs-utils source: %s\n' "$utils_src" >&2; exit 2; }
for tool in cc fsck.erofs git mkfs.erofs python3 sha256sum; do
	command -v "$tool" >/dev/null 2>&1 || {
		printf 'missing required host tool: %s\n' "$tool" >&2
		exit 2
	}
done
case "$output" in
/*) ;;
*) output=$PWD/$output ;;
esac
test ! -e "$output" || { printf 'refusing existing output: %s\n' "$output" >&2; exit 2; }
mkdir -p "$output"

python3 -B - "$root" "$input" "$base" "$output" "$freebsd_src" "$utils_src" <<'PY'
from __future__ import annotations

from array import array
from concurrent.futures import ThreadPoolExecutor
import ctypes
import hashlib
import json
import os
from pathlib import Path
import random
import shutil
import statistics
import struct
import subprocess
import sys
import tempfile
from typing import Any


ROOT = Path(sys.argv[1])
INPUT = Path(sys.argv[2])
REQUESTED_BASE = sys.argv[3]
OUTPUT = Path(sys.argv[4])
FREEBSD_SRC = Path(sys.argv[5])
UTILS_SRC = Path(sys.argv[6])
DUT = ROOT / "repo-pre-15"
SPEC = json.loads(INPUT.read_text(encoding="ascii"))
SUPER = 1024
MAGIC = 0xE0F5E1E2
CRC32C_POLY = 0x82F63B78


class GateStop(RuntimeError):
    pass


class InfraBlocked(RuntimeError):
    pass


class Reject(RuntimeError):
    def __init__(self, errno_name: str, point: str):
        super().__init__(f"{errno_name} at {point}")
        self.errno_name = errno_name
        self.point = point


def sha256_bytes(data: bytes) -> str:
    return hashlib.sha256(data).hexdigest()


def sha256_path(path: Path) -> str:
    digest = hashlib.sha256()
    with path.open("rb") as stream:
        for block in iter(lambda: stream.read(1024 * 1024), b""):
            digest.update(block)
    return digest.hexdigest()


def write_json(path: Path, value: Any) -> None:
    path.write_text(json.dumps(value, indent=2, sort_keys=True) + "\n", encoding="ascii")


def run(argv: list[str], *, input_bytes: bytes | None = None) -> subprocess.CompletedProcess:
    completed = subprocess.run(
        argv,
        input=input_bytes,
        check=False,
        stdout=subprocess.PIPE,
        stderr=subprocess.PIPE,
        timeout=120,
    )
    if completed.returncode != 0:
        output = (completed.stdout + completed.stderr).decode("utf-8", "replace")
        raise InfraBlocked(f"command failed ({completed.returncode}): {' '.join(argv)}: {output.strip()}")
    return completed


def git(path: Path, *args: str) -> str:
    return run(["git", "-C", str(path), *args]).stdout.decode().strip()


def source_at(commit: str, relative: str) -> bytes:
    return run(["git", "-C", str(ROOT), "show", f"{commit}:{relative}"]).stdout


def crc32c(data: bytes | bytearray, seed: int = 0xFFFFFFFF) -> int:
    value = seed
    for byte in data:
        value ^= byte
        for _ in range(8):
            value = (value >> 1) ^ (CRC32C_POLY if value & 1 else 0)
    return value & 0xFFFFFFFF


def rotl32(value: int, count: int) -> int:
    return ((value << count) | (value >> (32 - count))) & 0xFFFFFFFF


def oracle_xxh32(data: bytes, seed: int) -> int:
    prime1 = 2654435761
    prime2 = 2246822519
    prime3 = 3266489917
    prime4 = 668265263
    prime5 = 374761393
    cursor = 0
    end = len(data)

    def round32(accumulator: int, lane: int) -> int:
        accumulator = (accumulator + lane * prime2) & 0xFFFFFFFF
        return (rotl32(accumulator, 13) * prime1) & 0xFFFFFFFF

    if end >= 16:
        accumulator1 = (seed + prime1 + prime2) & 0xFFFFFFFF
        accumulator2 = (seed + prime2) & 0xFFFFFFFF
        accumulator3 = seed & 0xFFFFFFFF
        accumulator4 = (seed - prime1) & 0xFFFFFFFF
        limit = end - 16
        while cursor <= limit:
            accumulator1 = round32(accumulator1, int.from_bytes(data[cursor:cursor + 4], "little"))
            accumulator2 = round32(accumulator2, int.from_bytes(data[cursor + 4:cursor + 8], "little"))
            accumulator3 = round32(accumulator3, int.from_bytes(data[cursor + 8:cursor + 12], "little"))
            accumulator4 = round32(accumulator4, int.from_bytes(data[cursor + 12:cursor + 16], "little"))
            cursor += 16
        value = (
            rotl32(accumulator1, 1)
            + rotl32(accumulator2, 7)
            + rotl32(accumulator3, 12)
            + rotl32(accumulator4, 18)
        ) & 0xFFFFFFFF
    else:
        value = (seed + prime5) & 0xFFFFFFFF
    value = (value + end) & 0xFFFFFFFF
    while cursor + 4 <= end:
        value = (value + int.from_bytes(data[cursor:cursor + 4], "little") * prime3) & 0xFFFFFFFF
        value = (rotl32(value, 17) * prime4) & 0xFFFFFFFF
        cursor += 4
    while cursor < end:
        value = (value + data[cursor] * prime5) & 0xFFFFFFFF
        value = (rotl32(value, 11) * prime1) & 0xFFFFFFFF
        cursor += 1
    value ^= value >> 15
    value = (value * prime2) & 0xFFFFFFFF
    value ^= value >> 13
    value = (value * prime3) & 0xFFFFFFFF
    value ^= value >> 16
    return value & 0xFFFFFFFF


CANDIDATE_SOURCE = r'''#include <stdint.h>
#include <stdio.h>
#include <stdlib.h>

static uint32_t
erofs_xxh32_rotl(uint32_t value, unsigned int count)
{
	return ((value << count) | (value >> (32 - count)));
}

static uint32_t
erofs_xxh32_round(uint32_t seed, uint32_t input)
{
	seed += input * UINT32_C(2246822519);
	seed = erofs_xxh32_rotl(seed, 13);
	return (seed * UINT32_C(2654435761));
}

static uint32_t
erofs_xxh32_le32(const uint8_t *input)
{
	return ((uint32_t)input[0] | (uint32_t)input[1] << 8 |
	    (uint32_t)input[2] << 16 | (uint32_t)input[3] << 24);
}

static uint32_t
erofs_xxh32(const void *input, size_t length, uint32_t seed)
{
	const uint8_t *cursor = input;
	const uint8_t *end = cursor + length;
	uint32_t hash;

	if (length >= 16) {
		const uint8_t *limit = end - 16;
		uint32_t v1 = seed + UINT32_C(2654435761) + UINT32_C(2246822519);
		uint32_t v2 = seed + UINT32_C(2246822519);
		uint32_t v3 = seed;
		uint32_t v4 = seed - UINT32_C(2654435761);

		do {
			v1 = erofs_xxh32_round(v1, erofs_xxh32_le32(cursor));
			cursor += 4;
			v2 = erofs_xxh32_round(v2, erofs_xxh32_le32(cursor));
			cursor += 4;
			v3 = erofs_xxh32_round(v3, erofs_xxh32_le32(cursor));
			cursor += 4;
			v4 = erofs_xxh32_round(v4, erofs_xxh32_le32(cursor));
			cursor += 4;
		} while (cursor <= limit);
		hash = erofs_xxh32_rotl(v1, 1) + erofs_xxh32_rotl(v2, 7) +
		    erofs_xxh32_rotl(v3, 12) + erofs_xxh32_rotl(v4, 18);
	} else {
		hash = seed + UINT32_C(374761393);
	}
	hash += (uint32_t)length;
	while (cursor + 4 <= end) {
		hash += erofs_xxh32_le32(cursor) * UINT32_C(3266489917);
		hash = erofs_xxh32_rotl(hash, 17) * UINT32_C(668265263);
		cursor += 4;
	}
	while (cursor < end) {
		hash += *cursor++ * UINT32_C(374761393);
		hash = erofs_xxh32_rotl(hash, 11) * UINT32_C(2654435761);
	}
	hash ^= hash >> 15;
	hash *= UINT32_C(2246822519);
	hash ^= hash >> 13;
	hash *= UINT32_C(3266489917);
	hash ^= hash >> 16;
	return (hash);
}

int
main(void)
{
	uint8_t header[3], *name;
	uint32_t hash, seed;
	size_t length;

	while (fread(header, sizeof(header), 1, stdin) == 1) {
		length = (size_t)header[1] | (size_t)header[2] << 8;
		name = malloc(length == 0 ? 1 : length);
		if (name == NULL || (length != 0 && fread(name, length, 1, stdin) != 1))
			return (2);
		seed = UINT32_C(0x25BBE08F) + header[0];
		hash = erofs_xxh32(name, length, seed);
		free(name);
		if (fwrite(&hash, sizeof(hash), 1, stdout) != 1)
			return (3);
	}
	return (ferror(stdin) ? 4 : 0);
}
'''


def candidate_hashes(binary: Path, records: list[tuple[int, bytes]]) -> list[int]:
    payload = bytearray()
    for index, name in records:
        if len(name) > 65535:
            raise GateStop("candidate vector name exceeds protocol")
        payload.extend((index, len(name) & 0xFF, len(name) >> 8))
        payload.extend(name)
    output = run([str(binary)], input_bytes=bytes(payload)).stdout
    if len(output) != len(records) * 4:
        raise GateStop("candidate xxh32 returned a truncated vector stream")
    return list(struct.unpack(f"<{len(records)}I", output))


def verify_identity() -> dict[str, Any]:
    if SPEC.get("schema") != 1 or SPEC.get("candidate") != "P15-021" or SPEC.get("gate") != "G03":
        raise InfraBlocked("invalid P15-021 input identity")
    resolved = git(ROOT, "rev-parse", f"{REQUESTED_BASE}^{{commit}}")
    if resolved != SPEC["required_base"]:
        raise InfraBlocked(f"P15-021 must replay {SPEC['required_base']}, got {resolved}")
    status_lines = git(ROOT, "status", "--porcelain=v1", "--untracked-files=all").splitlines()
    changed = sorted(line[3:] for line in status_lines if len(line) >= 4)
    if changed not in ([], sorted(SPEC["write_set"]["gate"])):
        raise InfraBlocked(f"gate worktree write set differs: {changed}")
    source_hashes = {relative: sha256_bytes(source_at(resolved, relative)) for relative in SPEC["source_sha256"]}
    if source_hashes != SPEC["source_sha256"]:
        raise InfraBlocked("frozen DUT source identity changed")
    linux_hashes = {relative: sha256_bytes(source_at(resolved, relative)) for relative in SPEC["linux_source_sha256"]}
    if linux_hashes != SPEC["linux_source_sha256"]:
        raise InfraBlocked("frozen Linux source identity changed")
    freebsd_head = git(FREEBSD_SRC, "rev-parse", "HEAD")
    freebsd_hashes = {relative: sha256_path(FREEBSD_SRC / relative) for relative in SPEC["freebsd"]["sha256"]}
    if freebsd_head != SPEC["freebsd"]["head"] or freebsd_hashes != SPEC["freebsd"]["sha256"]:
        raise InfraBlocked("frozen FreeBSD source identity changed")
    utils_head = git(UTILS_SRC, "rev-parse", "HEAD")
    utils_hashes = {relative: sha256_path(UTILS_SRC / relative) for relative in SPEC["utils"]["sha256"]}
    if utils_head != SPEC["utils"]["head"] or utils_hashes != SPEC["utils"]["sha256"]:
        raise InfraBlocked("frozen erofs-utils identity changed")
    tool_hashes = {name: sha256_path(Path(item["path"])) for name, item in SPEC["tools"].items()}
    expected_tools = {name: item["sha256"] for name, item in SPEC["tools"].items()}
    if tool_hashes != expected_tools:
        raise InfraBlocked("frozen host tool identity changed")
    mkfs_version = run([SPEC["tools"]["mkfs.erofs"]["path"], "-V"]).stdout.decode().splitlines()[0]
    if mkfs_version != SPEC["tools"]["mkfs.erofs"]["version"]:
        raise InfraBlocked("mkfs.erofs version changed")
    linux_header = source_at(resolved, "src-linux/erofs_fs.h").decode("utf-8")
    linux_xattr = source_at(resolved, "src-linux/xattr.c").decode("utf-8")
    for marker in (
        "#define EROFS_XATTR_FILTER_BITS\t\t32",
        "#define EROFS_XATTR_FILTER_SEED\t\t0x25BBE08F",
        "bit value 1 indicates not-present",
    ):
        if marker not in linux_header:
            raise InfraBlocked(f"Linux Bloom format marker changed: {marker}")
    for marker in (
        "xxh32(name, strlen(name)",
        "EROFS_XATTR_FILTER_SEED + index",
        "vi->xattr_name_filter & (1U << hashbit)",
        "!sbi->xattr_filter_reserved",
    ):
        if marker not in linux_xattr:
            raise InfraBlocked(f"Linux Bloom use-site marker changed: {marker}")
    public_text = (FREEBSD_SRC / "sys/sys/libkern.h").read_text(encoding="utf-8")
    if "xxh32" in public_text.lower():
        raise InfraBlocked("FreeBSD gained a public xxh32 API; re-audit namespacing")
    return {
        "base": resolved,
        "freebsd_head": freebsd_head,
        "freebsd_sha256": freebsd_hashes,
        "implementation": SPEC["implementation"],
        "linux_sha256": linux_hashes,
        "mkfs_version": mkfs_version,
        "source_sha256": source_hashes,
        "tool_sha256": tool_hashes,
        "utils_head": utils_head,
        "utils_sha256": utils_hashes,
    }


def create_source(path: Path) -> None:
    path.mkdir(parents=True)
    paths = [path / f"peer-{index:03d}.bin" for index in range(SPEC["fixture"]["peer_count"])]
    paths.append(path / "target.bin")
    for entry in paths:
        entry.write_bytes(b"P15-021\n")
        entry.chmod(0o644)
        for index in range(SPEC["fixture"]["attribute_count"]):
            name = f"user.attr{index:02d}".encode()
            prefix = f"value-{index:02d}-".encode()
            value = (prefix * (SPEC["fixture"]["attribute_value_bytes"] // len(prefix) + 1))[
                : SPEC["fixture"]["attribute_value_bytes"]
            ]
            os.setxattr(entry, name, value)
        os.utime(entry, (0, 0), follow_symlinks=False)
    os.utime(path, (0, 0), follow_symlinks=False)


def build_fixture(source: Path, image: Path) -> list[str]:
    command = [
        SPEC["tools"]["mkfs.erofs"]["path"],
        "-d0",
        "-T0",
        "--all-time",
        "--all-root",
        "--workers=1",
        "--sort=path",
        f"-U{SPEC['fixture']['uuid']}",
        "-x2",
        "-Exattr-name-filter,force-inode-extended",
        str(image),
        str(source),
    ]
    run(command)
    return command


class Image:
    def __init__(self, path: Path):
        self.path = path
        self.fd = os.open(path, os.O_RDONLY)
        self.size = os.fstat(self.fd).st_size
        self.calls = 0
        self.bytes = 0
        self.blocks: set[int] = set()
        header = self.read(SUPER, 144, "super")
        if struct.unpack_from("<I", header)[0] != MAGIC:
            raise Reject("EINTEGRITY", "super.magic")
        self.feature_compat = struct.unpack_from("<I", header, 8)[0]
        self.block_bits = header[12]
        self.block_size = 1 << self.block_bits
        self.root_nid = struct.unpack_from("<H", header, 14)[0]
        self.blocks_count = struct.unpack_from("<I", header, 36)[0]
        self.limit = self.blocks_count << self.block_bits
        self.meta_blkaddr = struct.unpack_from("<I", header, 40)[0]
        self.xattr_blkaddr = struct.unpack_from("<I", header, 44)[0]
        self.filter_reserved = header[104]
        if self.limit > self.size:
            raise Reject("EINTEGRITY", "super.bounds")

    def close(self) -> None:
        os.close(self.fd)

    def __enter__(self) -> "Image":
        return self

    def __exit__(self, *_: object) -> None:
        self.close()

    def reset_reads(self) -> None:
        self.calls = 0
        self.bytes = 0
        self.blocks.clear()
        if hasattr(os, "posix_fadvise"):
            os.posix_fadvise(self.fd, 0, 0, os.POSIX_FADV_DONTNEED)

    def read(self, offset: int, length: int, point: str) -> bytes:
        limit = self.limit if hasattr(self, "limit") else self.size
        if offset < 0 or length < 0 or offset > limit:
            raise Reject("EINTEGRITY", f"{point}.bounds")
        if length > limit - offset:
            raise Reject("EINTEGRITY", f"{point}.bounds")
        data = os.pread(self.fd, length, offset)
        if len(data) != length:
            raise Reject("EINTEGRITY", f"{point}.short")
        self.calls += 1
        self.bytes += length
        if length:
            first = offset // 4096
            last = (offset + length - 1) // 4096
            self.blocks.update(range(first, last + 1))
        return data

    def inode(self, nid: int) -> dict[str, int]:
        offset = (self.meta_blkaddr << self.block_bits) + (nid << 5)
        raw = self.read(offset, 64, "inode")
        inode_format, xattr_count = struct.unpack_from("<HH", raw)
        inode_size = 64 if inode_format & 1 else 32
        xattr_size = 0 if xattr_count == 0 else 12 + 4 * (xattr_count - 1)
        size = struct.unpack_from("<Q" if inode_size == 64 else "<I", raw, 8)[0]
        return {
            "nid": nid,
            "offset": offset,
            "inode_size": inode_size,
            "xattr_size": xattr_size,
            "layout": (inode_format >> 1) & 7,
            "size": size,
            "start_block": struct.unpack_from("<I", raw, 16)[0],
        }

    def inode_data(self, inode: dict[str, int], logical: int, length: int) -> bytes:
        if logical > inode["size"] or length > inode["size"] - logical:
            raise Reject("EINTEGRITY", "inode.data-bounds")
        if inode["layout"] == 2:
            physical = inode["offset"] + inode["inode_size"] + inode["xattr_size"] + logical
        elif inode["layout"] == 0:
            physical = (inode["start_block"] << self.block_bits) + logical
        else:
            raise Reject("EINTEGRITY", "inode.unsupported-layout")
        return self.read(physical, length, "inode.data")

    def directory_entries(self, inode: dict[str, int]) -> list[tuple[bytes, int]]:
        result = []
        logical = 0
        while logical < inode["size"]:
            length = min(self.block_size, inode["size"] - logical)
            data = self.inode_data(inode, logical, length)
            if len(data) < 12:
                raise Reject("EINTEGRITY", "directory.header")
            first_name = struct.unpack_from("<H", data, 8)[0]
            if first_name == 0 or first_name % 12 or first_name > len(data):
                raise Reject("EINTEGRITY", "directory.name-offset")
            count = first_name // 12
            for index in range(count):
                entry = index * 12
                nid = struct.unpack_from("<Q", data, entry)[0]
                start = struct.unpack_from("<H", data, entry + 8)[0]
                end = struct.unpack_from("<H", data, entry + 20)[0] if index + 1 < count else len(data)
                if start > end or end > len(data):
                    raise Reject("EINTEGRITY", "directory.name-bounds")
                result.append((data[start:end].split(b"\0", 1)[0], nid))
            logical += length
        return result

    def resolve(self, path: str) -> dict[str, int]:
        inode = self.inode(self.root_nid)
        for component in path.strip("/").encode().split(b"/"):
            candidates = [nid for name, nid in self.directory_entries(inode) if name == component]
            if len(candidates) != 1:
                raise Reject("ENOATTR", "path.lookup")
            inode = self.inode(candidates[0])
        return inode

    def body_offset(self, inode: dict[str, int]) -> int:
        return inode["offset"] + inode["inode_size"]

    def body_header(self, inode: dict[str, int]) -> tuple[int, int, list[int]]:
        if inode["xattr_size"] < 12:
            raise Reject("EINTEGRITY", "ibody.header")
        raw = self.read(self.body_offset(inode), 12, "ibody.header")
        name_filter = struct.unpack_from("<I", raw)[0]
        shared_count = raw[4]
        header_size = 12 + shared_count * 4
        if header_size > inode["xattr_size"]:
            raise Reject("EINTEGRITY", "ibody.shared-count")
        if inode["xattr_size"] == 12:
            raise Reject("EOPNOTSUPP", "ibody.header-only")
        shared = []
        if shared_count:
            ids = self.read(self.body_offset(inode) + 12, shared_count * 4, "ibody.shared-ids")
            shared = list(struct.unpack(f"<{shared_count}I", ids))
        return name_filter, header_size, shared

    def entry(self, offset: int, limit: int, kind: str) -> tuple[int, bytes, bytes, int]:
        header = self.read(offset, 4, f"{kind}.header")
        name_length, name_index, value_length = struct.unpack("<BBH", header)
        total = (4 + name_length + value_length + 3) & ~3
        if total > limit - offset:
            raise Reject("EINTEGRITY", f"{kind}.bounds")
        raw = self.read(offset, total, f"{kind}.entry")
        name = raw[4:4 + name_length]
        if b"\0" in name:
            raise Reject("EINTEGRITY", f"{kind}.name-nul")
        return name_index, name, raw[4 + name_length:4 + name_length + value_length], total

    def lookup(self, inode: dict[str, int], name: bytes, candidate: bool) -> tuple[str, bytes | None, dict[str, Any]]:
        self.reset_reads()
        if candidate and self.feature_compat & SPEC["format"]["feature_compat"] and self.filter_reserved == 0:
            name_filter, _, _ = self.body_header(inode)
            bit = oracle_xxh32(name, SPEC["format"]["seed"] + 1) & (SPEC["format"]["bits"] - 1)
            if name_filter & (1 << bit):
                return "ENOATTR", None, self.read_stats(False, bit)
        body = self.read(self.body_offset(inode), inode["xattr_size"], "ibody")
        name_filter = struct.unpack_from("<I", body)[0]
        shared_count = body[4]
        header_size = 12 + shared_count * 4
        if header_size > len(body):
            raise Reject("EINTEGRITY", "ibody.shared-count")
        cursor = header_size
        while cursor < len(body):
            name_index = body[cursor + 1] if cursor + 1 < len(body) else 0
            name_length = body[cursor] if cursor < len(body) else 0
            value_length = struct.unpack_from("<H", body, cursor + 2)[0] if cursor + 4 <= len(body) else 0
            total = (4 + name_length + value_length + 3) & ~3
            if total > len(body) - cursor:
                raise Reject("EINTEGRITY", "inline.bounds")
            actual = body[cursor + 4:cursor + 4 + name_length]
            if b"\0" in actual:
                raise Reject("EINTEGRITY", "inline.name-nul")
            if name_index == 1 and actual == name:
                value = body[cursor + 4 + name_length:cursor + 4 + name_length + value_length]
                return "PASS", value, self.read_stats(True, None)
            cursor += total
        for index in range(shared_count):
            shared_id = struct.unpack_from("<I", body, 12 + index * 4)[0]
            offset = (self.xattr_blkaddr << self.block_bits) + shared_id * 4
            name_index, actual, value, _ = self.entry(offset, self.limit, "shared")
            if name_index == 1 and actual == name:
                return "PASS", value, self.read_stats(True, None)
        return "ENOATTR", None, self.read_stats(True, None)

    def read_stats(self, scanned: bool, bit: int | None) -> dict[str, Any]:
        return {
            "bytes": self.bytes,
            "calls": self.calls,
            "filter_bit": bit,
            "provider_blocks": len(self.blocks),
            "scanned": scanned,
        }


def mutate_image(source: Path, output: Path, mutation: str, target: dict[str, int]) -> None:
    data = bytearray(source.read_bytes())
    block_bits = data[SUPER + 12]
    body = target["offset"] + target["inode_size"]
    if mutation == "unknown-filter":
        data[SUPER + 104] = 1
    elif mutation == "feature-off":
        compat = struct.unpack_from("<I", data, SUPER + 8)[0]
        struct.pack_into("<I", data, SUPER + 8, compat & ~SPEC["format"]["feature_compat"])
    elif mutation == "corrupt-shared-count":
        data[body + 4] = 255
    elif mutation == "corrupt-shared-id":
        if data[body + 4] == 0:
            raise GateStop("fixture has no shared xattr ID to corrupt")
        struct.pack_into("<I", data, body + 12, 0xFFFFFFFF)
    else:
        raise GateStop(f"unknown fixture mutation: {mutation}")
    compat = struct.unpack_from("<I", data, SUPER + 8)[0]
    if compat & 1:
        struct.pack_into("<I", data, SUPER + 4, 0)
        end = 1 << block_bits
        struct.pack_into("<I", data, SUPER + 4, crc32c(data[SUPER:end]))
    output.write_bytes(data)


def lookup_case(path: Path, name: bytes, candidate: bool) -> dict[str, Any]:
    try:
        with Image(path) as image:
            inode = image.resolve(SPEC["fixture"]["target"])
            status, value, reads = image.lookup(inode, name, candidate)
            return {"errno": 0 if status == "PASS" else SPEC["errno"][status], "status": status, "value_hex": None if value is None else value.hex(), "reads": reads}
    except Reject as error:
        return {"errno": SPEC["errno"].get(error.errno_name, -1), "status": error.errno_name, "point": error.point, "value_hex": None}


def main(work: Path) -> dict[str, Any]:
    identity = verify_identity()
    write_json(OUTPUT / "identity.json", identity)
    candidate_source = work / "candidate.c"
    candidate_binary = work / "candidate"
    candidate_source.write_text(CANDIDATE_SOURCE, encoding="ascii")
    run([SPEC["tools"]["cc"]["path"], "-std=c11", "-O2", "-Wall", "-Wextra", "-Werror", str(candidate_source), "-o", str(candidate_binary)])
    write_json(OUTPUT / "prototype.json", {
        "binary_sha256": sha256_path(candidate_binary),
        "linked_into_dut_kld": False,
        "source_sha256": sha256_path(candidate_source),
        "status": "PASS",
        "symbol": SPEC["implementation"]["symbol"],
    })

    library = ctypes.CDLL(SPEC["tools"]["libxxhash"]["path"])
    library.XXH32.argtypes = [ctypes.c_void_p, ctypes.c_size_t, ctypes.c_uint32]
    library.XXH32.restype = ctypes.c_uint32
    vector_records = [(item["index"], bytes.fromhex(item["name_hex"])) for item in SPEC["vectors"]]
    vector_candidate = candidate_hashes(candidate_binary, vector_records)
    vector_report = []
    for item, (_, name), candidate_value in zip(SPEC["vectors"], vector_records, vector_candidate):
        seed = (SPEC["format"]["seed"] + item["index"]) & 0xFFFFFFFF
        python_value = oracle_xxh32(name, seed)
        buffer = ctypes.create_string_buffer(name if name else b"\0")
        library_value = int(library.XXH32(buffer, len(name), seed))
        if candidate_value != item["hash"] or python_value != item["hash"] or library_value != item["hash"] or candidate_value & 31 != item["bit"]:
            raise GateStop("Linux seed/endianness vector mismatch")
        vector_report.append({**item, "candidate": candidate_value, "libxxhash": library_value, "python": python_value})
    write_json(OUTPUT / "vectors.json", {"status": "PASS", "vectors": vector_report})

    rng = random.Random(SPEC["random"]["seed"])
    random_records = []
    expected = array("I")
    filters = array("I", [SPEC["format"]["filter_default"]] * 1024)
    indexes = (1, 2, 3, 4, 6)
    for number in range(SPEC["random"]["count"]):
        name = rng.randbytes(16).hex().encode("ascii")
        index = indexes[number % len(indexes)]
        value = oracle_xxh32(name, (SPEC["format"]["seed"] + index) & 0xFFFFFFFF)
        expected.append(value)
        filters[number % len(filters)] &= ~(1 << (value & 31))
        random_records.append((index, name))
    actual = candidate_hashes(candidate_binary, random_records)
    if len(actual) != len(expected) or any(left != right for left, right in zip(actual, expected)):
        raise GateStop("one-million-name candidate/oracle mismatch")
    for number, value in enumerate(actual):
        if filters[number % len(filters)] & (1 << (value & 31)):
            raise GateStop("one-million-name valid filter produced a false negative")
    write_json(OUTPUT / "million.json", {"count": len(actual), "filter_count": len(filters), "seed": SPEC["random"]["seed"], "status": "PASS"})
    del random_records, actual, expected, filters

    source = work / "source"
    create_source(source)
    fixtures = OUTPUT / "fixtures"
    fixtures.mkdir()
    first = fixtures / "valid.erofs"
    repeat = work / "repeat.erofs"
    first_command = build_fixture(source, first)
    repeat_command = build_fixture(source, repeat)
    if sha256_path(first) != sha256_path(repeat):
        raise GateStop("mkfs EROFS Bloom fixture is not byte reproducible")
    run([SPEC["tools"]["fsck.erofs"]["path"], str(first)])
    with Image(first) as image:
        target = image.resolve(SPEC["fixture"]["target"])
        image.reset_reads()
        name_filter, _, shared = image.body_header(target)
        status, _, _ = image.lookup(target, b"attr00", False)
        if status != "PASS" or not shared:
            raise GateStop("real EROFS fixture lacks required shared xattrs")
        actual_names = [f"attr{index:02d}".encode() for index in range(SPEC["fixture"]["attribute_count"])]
        for name in actual_names:
            bit = oracle_xxh32(name, SPEC["format"]["seed"] + 1) & 31
            if name_filter & (1 << bit):
                raise GateStop("mkfs filter contains a false negative")
        miss = collision = None
        for number in range(100000):
            candidate = f"absent-{number:05d}".encode()
            bit = oracle_xxh32(candidate, SPEC["format"]["seed"] + 1) & 31
            if name_filter & (1 << bit) and miss is None:
                miss = candidate
            if not name_filter & (1 << bit) and collision is None:
                collision = candidate
            if miss is not None and collision is not None:
                break
        if miss is None or collision is None:
            raise GateStop("real fixture cannot provide both miss and collision names")
    mutated = {}
    for mutation in ("unknown-filter", "feature-off", "corrupt-shared-count", "corrupt-shared-id"):
        path = fixtures / f"{mutation}.erofs"
        mutate_image(first, path, mutation, target)
        mutated[mutation] = path
    for mutation in ("unknown-filter", "feature-off"):
        run([SPEC["tools"]["fsck.erofs"]["path"], str(mutated[mutation])])

    cases = [
        ("hit", first, b"attr00", "PASS", True),
        ("miss", first, miss, "ENOATTR", False),
        ("false-positive", first, collision, "ENOATTR", True),
        ("unknown-filter", mutated["unknown-filter"], miss, "ENOATTR", True),
        ("feature-off", mutated["feature-off"], miss, "ENOATTR", True),
        ("corrupt-shared-count", mutated["corrupt-shared-count"], miss, "EINTEGRITY", None),
        ("corrupt-shared-id", mutated["corrupt-shared-id"], b"attr00", "EINTEGRITY", None),
    ]
    case_report = []
    for identifier, path, name, expected_status, expected_scan in cases:
        baseline = lookup_case(path, name, False)
        candidate = lookup_case(path, name, True)
        if candidate["status"] != expected_status or candidate["errno"] < 0:
            raise GateStop(f"{identifier} candidate status/positive errno mismatch: {candidate}")
        if identifier not in ("corrupt-shared-count",) and baseline["status"] != expected_status:
            raise GateStop(f"{identifier} baseline oracle mismatch: {baseline}")
        if expected_scan is not None and candidate["reads"]["scanned"] != expected_scan:
            raise GateStop(f"{identifier} scan/fallback mismatch")
        case_report.append({"id": identifier, "name": name.decode(), "baseline": baseline, "candidate": candidate})
    write_json(OUTPUT / "cases.json", {"cases": case_report, "status": "PASS"})

    samples = SPEC["thresholds"]["cold_samples"]
    loops = SPEC["thresholds"]["cold_loops_per_sample"]
    baseline_calls = []
    candidate_calls = []
    baseline_blocks = []
    candidate_blocks = []
    for _ in range(samples):
        totals = {"baseline_calls": 0, "candidate_calls": 0, "baseline_blocks": 0, "candidate_blocks": 0}
        for _ in range(loops):
            baseline = lookup_case(first, miss, False)["reads"]
            candidate = lookup_case(first, miss, True)["reads"]
            totals["baseline_calls"] += baseline["calls"]
            totals["candidate_calls"] += candidate["calls"]
            totals["baseline_blocks"] += baseline["provider_blocks"]
            totals["candidate_blocks"] += candidate["provider_blocks"]
        baseline_calls.append(totals["baseline_calls"])
        candidate_calls.append(totals["candidate_calls"])
        baseline_blocks.append(totals["baseline_blocks"])
        candidate_blocks.append(totals["candidate_blocks"])
    call_reduction = 100.0 * (statistics.median(baseline_calls) - statistics.median(candidate_calls)) / statistics.median(baseline_calls)
    block_reduction = 100.0 * (statistics.median(baseline_blocks) - statistics.median(candidate_blocks)) / statistics.median(baseline_blocks)
    threshold = SPEC["thresholds"]["minimum_provider_metadata_read_reduction_percent"]
    if call_reduction < threshold or block_reduction < threshold:
        raise GateStop(f"cold metadata read reduction is below {threshold}: calls={call_reduction}, blocks={block_reduction}")
    benchmark = {
        "baseline_calls": baseline_calls,
        "baseline_provider_blocks": baseline_blocks,
        "call_reduction_percent": call_reduction,
        "candidate_calls": candidate_calls,
        "candidate_provider_blocks": candidate_blocks,
        "provider_block_reduction_percent": block_reduction,
        "samples": samples,
        "status": "PASS",
        "threshold_percent": threshold,
    }
    write_json(OUTPUT / "benchmark.json", benchmark)

    expected_cases = {item[0]: item[3] for item in cases}
    def worker(worker_id: int) -> int:
        completed = 0
        for iteration in range(SPEC["concurrency"]["loops_per_worker"]):
            identifier, path, name, _, _ = cases[(worker_id + iteration) % len(cases)]
            result = lookup_case(path, name, True)
            if result["status"] != expected_cases[identifier] or result["errno"] < 0:
                raise GateStop(f"concurrent {identifier} mismatch")
            completed += 1
        return completed
    with ThreadPoolExecutor(max_workers=SPEC["concurrency"]["workers"]) as executor:
        completed = sum(executor.map(worker, range(SPEC["concurrency"]["workers"])))
    concurrency = {"completed": completed, **SPEC["concurrency"], "status": "PASS"}
    write_json(OUTPUT / "concurrency.json", concurrency)

    fixture_hashes = {path.name: sha256_path(path) for path in sorted(fixtures.iterdir())}
    write_json(OUTPUT / "fixture.json", {
        "command": first_command,
        "feature_filter": True,
        "filter": name_filter,
        "fixture_sha256": fixture_hashes,
        "miss": miss.decode(),
        "collision": collision.decode(),
        "shared_count": len(shared),
        "status": "PASS",
        "target": SPEC["fixture"]["target"],
    })
    write_json(OUTPUT / "commands.json", {
        "candidate_compile": [SPEC["tools"]["cc"]["path"], "-std=c11", "-O2", "-Wall", "-Wextra", "-Werror", "candidate.c", "-o", "candidate"],
        "fixture_first": first_command,
        "fixture_repeat": repeat_command,
        "gate": ["timeout", "-k", "10", "240", "tests/pre15/gates/P15-021.sh", "--base", REQUESTED_BASE, "--output", str(OUTPUT)],
    })
    return {
        "batch": "B19b",
        "benchmark": benchmark,
        "candidate": "P15-021",
        "case_count": len(cases),
        "concurrency": concurrency,
        "decision": "GO",
        "fixture_sha256": fixture_hashes,
        "full_feature_suite": "NOT_RUN",
        "qemu": "NOT_RUN",
        "random_name_count": SPEC["random"]["count"],
        "source_modified": False,
        "status": "GO",
    }


exit_code = 21
work = Path(tempfile.mkdtemp(prefix=".P15-021-work-", dir=OUTPUT))
try:
    result = main(work)
    exit_code = 0
except GateStop as error:
    result = {
        "batch": "B19b",
        "candidate": "P15-021",
        "decision": "STOP",
        "full_feature_suite": "NOT_RUN",
        "qemu": "NOT_RUN",
        "reason": str(error),
        "source_modified": False,
        "status": "STOP",
    }
    exit_code = 1
except (InfraBlocked, OSError, KeyError, ValueError, json.JSONDecodeError, subprocess.TimeoutExpired) as error:
    result = {
        "batch": "B19b",
        "candidate": "P15-021",
        "decision": "INFRA_BLOCKED",
        "full_feature_suite": "NOT_RUN",
        "qemu": "INFRA_BLOCKED",
        "reason": str(error),
        "source_modified": False,
        "status": "INFRA_BLOCKED",
    }
    exit_code = 21
finally:
    shutil.rmtree(work, ignore_errors=False)
write_json(OUTPUT / "result.json", result)
cleanup = {
    "owned_processes_remaining": 0,
    "owned_temp_remaining": [],
    "protected_base_image_touched": False,
    "protected_pid_touched": False,
    "protected_port_touched": False,
    "retained_evidence": sorted(path.name for path in OUTPUT.iterdir()),
    "source_modified": False,
    "status": "PASS",
}
write_json(OUTPUT / "cleanup.json", cleanup)
lines = []
for path in sorted(OUTPUT.rglob("*")):
    if path.is_file() and path.name != "SHA256SUMS":
        lines.append(f"{sha256_path(path)}  {path.relative_to(OUTPUT).as_posix()}")
(OUTPUT / "SHA256SUMS").write_text("\n".join(lines) + "\n", encoding="ascii")
print(json.dumps(result, sort_keys=True))
raise SystemExit(exit_code)
PY
