#!/bin/sh
set -eu

gate_dir=$(CDPATH= cd -- "$(dirname -- "$0")" && pwd -P)
root=$(CDPATH= cd -- "$gate_dir/../../../.." && pwd -P)
input=$gate_dir/P15-087-input.json
freebsd_src=${FREEBSD_SRC:-/work/dev-freebsd-releng}
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 tree: %s\n' "$freebsd_src" >&2
	exit 2
}
for tool in cc fsck.erofs git mkfs.erofs python3 qemu-img qemu-system-x86_64 \
    scp sha256sum ssh tar; 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 - "$root" "$input" "$base" "$output" "$freebsd_src" <<'PY'
from __future__ import annotations

import difflib
import hashlib
import json
import os
from pathlib import Path
import re
import shlex
import shutil
import socket
import statistics
import struct
import subprocess
import sys
import tarfile
import tempfile
import time
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])
SPEC = json.loads(INPUT.read_text(encoding="ascii"))
SUPER = 1024
MAGIC = 0xE0F5E1E2
DT_BY_EROFS = {1: 8, 2: 4, 3: 2, 4: 6, 5: 1, 6: 12, 7: 10}


class GateFailure(RuntimeError):
    def __init__(self, status: str, reason: str):
        super().__init__(reason)
        self.status = status
        self.reason = reason


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


def sha256_path(path: Path) -> str:
    return sha256_bytes(path.read_bytes())


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


def run(argv: list[str], *, cwd: Path | None = None, env: dict[str, str] | None = None,
        timeout: int | None = None, check: bool = True) -> subprocess.CompletedProcess[str]:
    try:
        completed = subprocess.run(
            argv,
            cwd=cwd,
            env=env,
            text=True,
            stdin=subprocess.DEVNULL,
            stdout=subprocess.PIPE,
            stderr=subprocess.STDOUT,
            timeout=timeout,
            check=False,
        )
    except subprocess.TimeoutExpired as error:
        raise GateFailure(
            "INFRA_BLOCKED", f"command timed out: {' '.join(argv)}"
        ) from error
    if check and completed.returncode != 0:
        raise GateFailure(
            "INFRA_BLOCKED",
            f"command failed ({completed.returncode}): {' '.join(argv)}\n{completed.stdout}",
        )
    return completed


def git(*args: str) -> str:
    return subprocess.check_output(["git", "-C", str(ROOT), *args], text=True).strip()


def source_at(commit: str, path: str) -> str:
    completed = run(["git", "-C", str(ROOT), "show", f"{commit}:{path}"], check=False)
    if completed.returncode != 0:
        raise GateFailure("INFRA_BLOCKED", f"cannot read frozen source {path}: {completed.stdout}")
    return completed.stdout


def replace_once(source: str, before: str, after: str, label: str) -> str:
    if source.count(before) != 1:
        raise GateFailure("INFRA_BLOCKED", f"candidate anchor changed for {label}")
    return source.replace(before, after, 1)


def fnv_update(value: int, data: bytes) -> int:
    for byte in data:
        value ^= byte
        value = (value * 1099511628211) & ((1 << 64) - 1)
    return value


def record_hash(value: int, nid: int, dtype: int, name: bytes, cookie: int) -> int:
    value = fnv_update(value, struct.pack("<Q", nid))
    value = fnv_update(value, bytes((dtype,)))
    value = fnv_update(value, struct.pack("<H", len(name)))
    value = fnv_update(value, struct.pack("<Q", cookie))
    return fnv_update(value, name)


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


def promote_root_tail_to_plain(path: Path) -> dict[str, Any]:
    data = bytearray(path.read_bytes())
    image = Image(bytes(data))
    inode = image.root_inode()
    if inode["layout"] != 2:
        raise GateFailure(
            "INFRA_BLOCKED", f"expected mkfs flat-inline directory, got {inode['layout']}"
        )
    block_size = image.block_size
    size = int(inode["size"])
    tail_size = size % block_size
    if tail_size == 0:
        raise GateFailure("INFRA_BLOCKED", "mkfs flat-inline directory has no tail")
    full_blocks = size // block_size
    startblk = int(inode["startblk"])
    if (startblk + full_blocks) * block_size != len(data):
        raise GateFailure(
            "INFRA_BLOCKED", "root full blocks are not the contiguous image suffix"
        )
    inode_offset = int(inode["offset"])
    inline_offset = inode_offset + int(inode["inode_size"]) + int(inode["xattr_size"])
    if inline_offset + tail_size > len(data):
        raise GateFailure("INFRA_BLOCKED", "root inline tail exceeds the source image")
    tail = bytes(data[inline_offset:inline_offset + tail_size])
    data.extend(b"\0" * block_size)
    data[-block_size:-block_size + tail_size] = tail
    ifmt = struct.unpack_from("<H", data, inode_offset)[0]
    ifmt &= ~(0x7 << 1)
    struct.pack_into("<H", data, inode_offset, ifmt)
    blocks = struct.unpack_from("<I", data, SUPER + 36)[0]
    if blocks != len(data) // block_size - 1:
        raise GateFailure("INFRA_BLOCKED", "superblock count does not match mkfs image")
    struct.pack_into("<I", data, SUPER + 36, blocks + 1)
    struct.pack_into("<I", data, SUPER + 4, 0)
    checksum_span = block_size - SUPER if block_size > SUPER else block_size
    checksum_end = SUPER + checksum_span
    struct.pack_into("<I", data, SUPER + 4, crc32c(data[SUPER:checksum_end]))
    path.write_bytes(data)
    promoted = Image(path.read_bytes()).root_inode()
    if promoted["layout"] != 0:
        raise GateFailure("INFRA_BLOCKED", "derived directory did not become flat plain")
    return {
        "appended_blocks": 1,
        "full_blocks_before_tail": full_blocks,
        "inline_tail_bytes": tail_size,
        "source_layout": "flat-inline",
        "target_layout": "flat-plain",
    }


class Image:
    def __init__(self, data: bytes):
        self.data = data
        if len(data) < SUPER + 144 or self.u32(SUPER) != MAGIC:
            raise GateFailure("INFRA_BLOCKED", "generated fixture is not an EROFS image")
        self.block_bits = data[SUPER + 12]
        self.block_size = 1 << self.block_bits
        self.meta_blkaddr = self.u32(SUPER + 40)
        self.root_nid = self.u16(SUPER + 14) or self.u64(SUPER + 112)

    def u16(self, offset: int) -> int:
        return struct.unpack_from("<H", self.data, offset)[0]

    def u32(self, offset: int) -> int:
        return struct.unpack_from("<I", self.data, offset)[0]

    def u64(self, offset: int) -> int:
        return struct.unpack_from("<Q", self.data, offset)[0]

    def root_inode(self) -> dict[str, int | bool]:
        offset = (self.meta_blkaddr << self.block_bits) + (self.root_nid << 5)
        ifmt = self.u16(offset)
        version = ifmt & 1
        inode_size = 32 if version == 0 else 64
        xattr_count = self.u16(offset + 2)
        xattr_size = 0 if xattr_count == 0 else 12 + (xattr_count - 1) * 4
        size = self.u32(offset + 8) if version == 0 else self.u64(offset + 8)
        return {
            "nid": self.root_nid,
            "offset": offset,
            "layout": (ifmt >> 1) & 7,
            "dot_omitted": bool((ifmt >> 4) & 1),
            "inode_size": inode_size,
            "xattr_size": xattr_size,
            "mode": self.u16(offset + 4),
            "size": size,
            "startblk": self.u32(offset + 16),
        }

    def directory_oracle(self) -> dict[str, Any]:
        inode = self.root_inode()
        if inode["layout"] != 0:
            raise GateFailure("STOP", f"root directory is not flat plain: layout={inode['layout']}")
        if int(inode["mode"]) & 0o170000 != 0o040000:
            raise GateFailure("INFRA_BLOCKED", "root inode is not a directory")
        data_offset = int(inode["startblk"]) << self.block_bits
        size = int(inode["size"])
        if data_offset + size > len(self.data):
            raise GateFailure("INFRA_BLOCKED", "root directory bytes exceed the fixture")
        records: list[dict[str, Any]] = []
        previous = b""
        value = 1469598103934665603
        for block_off in range(0, size, self.block_size):
            maxsize = min(self.block_size, size - block_off)
            base = data_offset + block_off
            if maxsize < 12:
                raise GateFailure("INFRA_BLOCKED", "short directory block")
            first_nameoff = self.u16(base + 8)
            if first_nameoff < 12 or first_nameoff % 12 != 0 or first_nameoff >= maxsize:
                raise GateFailure("INFRA_BLOCKED", f"bad name offset at block {block_off}")
            count = first_nameoff // 12
            for index in range(count):
                entry = base + index * 12
                nameoff = self.u16(entry + 8)
                endoff = self.u16(entry + 20) if index + 1 < count else maxsize
                if not (first_nameoff <= nameoff < endoff <= maxsize):
                    raise GateFailure("INFRA_BLOCKED", "invalid directory name span")
                raw = self.data[base + nameoff:base + endoff]
                name = raw.split(b"\0", 1)[0]
                if not name or any(raw[len(name):]):
                    raise GateFailure("INFRA_BLOCKED", "invalid directory name padding")
                if previous and previous >= name:
                    raise GateFailure("INFRA_BLOCKED", "fixture directory is not strictly ordered")
                previous = name
                nid = self.u64(entry)
                erofs_type = self.data[entry + 10]
                if erofs_type not in DT_BY_EROFS:
                    raise GateFailure("INFRA_BLOCKED", "fixture uses an unknown file type")
                cookie = block_off + (index + 1) * 12 if index + 1 < count else block_off + maxsize
                value = record_hash(value, nid, DT_BY_EROFS[erofs_type], name, cookie)
                records.append({
                    "cookie": cookie,
                    "file_type": erofs_type,
                    "name": name.decode("ascii"),
                    "nid": nid,
                })
        if inode["dot_omitted"]:
            cookie = size + 1
            value = record_hash(value, int(inode["nid"]), 4, b".", cookie)
            records.append({"cookie": cookie, "file_type": 2, "name": ".", "nid": inode["nid"]})
        random_offset = (size // (2 * self.block_size)) * self.block_size
        if random_offset == 0:
            raise GateFailure("INFRA_BLOCKED", "fixture is too small for a random seek")
        return {
            "block_count": (size + self.block_size - 1) // self.block_size,
            "block_size": self.block_size,
            "data_offset": data_offset,
            "directory_bytes": size,
            "dot_omitted": inode["dot_omitted"],
            "entry_count": len(records),
            "final_cookie": records[-1]["cookie"],
            "fnv64": f"{value:016x}",
            "layout": "flat-plain",
            "random_offset": random_offset,
            "root_nid": inode["nid"],
            "startblk": inode["startblk"],
        }


def make_candidate(data_source: str, dir_source: str, internal_source: str) -> dict[str, str]:
    data_candidate = replace_once(
        data_source,
        '#include "internal.h"\n',
        '#include "internal.h"\n\n#define EROFS_DIR_READAHEAD_BYTES\t(1024 * 1024)\n#define EROFS_DIR_READAHEAD_SLOTS\t256\n',
        "data constants",
    )
    data_candidate = replace_once(
        data_candidate,
        "static int\nerofs_bread_device(struct erofs_sb_info *sbi, struct erofs_device_info *dif,\n    erofs_blk_t blocks, erofs_off_t off, size_t len, void **bufp)\n",
        "static int\nerofs_bread_device(struct erofs_sb_info *sbi, struct erofs_device_info *dif,\n    erofs_blk_t blocks, erofs_off_t off, size_t len, daddr_t *rablkno,\n    int *rabsize, int racnt, void **bufp)\n",
        "bread device signature",
    )
    data_candidate = replace_once(
        data_candidate,
        "\t\terror = bread(dif->devvp, btodb(blkoff), iosize, NOCRED, &bp);\n",
        "\t\tif (done == 0 && racnt != 0)\n\t\t\terror = breadn(dif->devvp, btodb(blkoff), iosize,\n\t\t\t    rablkno, rabsize, racnt, NOCRED, &bp);\n\t\telse\n\t\t\terror = bread(dif->devvp, btodb(blkoff), iosize, NOCRED, &bp);\n",
        "backing vnode breadn",
    )
    data_candidate = replace_once(
        data_candidate,
        "\treturn (erofs_bread_device(sbi, &sbi->dif0, sbi->dif0.blocks, off, len,\n\t    bufp));\n",
        "\treturn (erofs_bread_device(sbi, &sbi->dif0, sbi->dif0.blocks, off, len,\n\t    NULL, NULL, 0, bufp));\n",
        "primary bread caller",
    )
    data_candidate = replace_once(
        data_candidate,
        "\treturn (erofs_bread_device(sbi, map.m_dif, blocks, map.m_pa, len, bufp));\n}\n\n/* Release a contiguous buffer returned by erofs_bread(). */\n",
        "\treturn (erofs_bread_device(sbi, map.m_dif, blocks, map.m_pa, len,\n\t    NULL, NULL, 0, bufp));\n}\n\nstatic int\nerofs_read_physical_readahead(struct erofs_sb_info *sbi,\n    unsigned int device_id, erofs_off_t off, size_t len,\n    unsigned int rablocks, void **bufp)\n{\n\tstruct erofs_map_dev current, future;\n\tdaddr_t rablkno[EROFS_DIR_READAHEAD_SLOTS];\n\tint rabsize[EROFS_DIR_READAHEAD_SLOTS];\n\terofs_off_t step;\n\terofs_blk_t blocks;\n\tunsigned int count;\n\tint error;\n\n\tcurrent = (struct erofs_map_dev) {\n\t\t.m_pa = off,\n\t\t.m_deviceid = device_id,\n\t\t.m_plen = len,\n\t};\n\terror = erofs_map_dev(sbi, &current);\n\tif (error != 0)\n\t\treturn (error);\n\tblocks = current.m_dif->blocks;\n\tif (current.m_dif == &sbi->dif0 && sbi->flatdev)\n\t\tblocks = sbi->flatdev_blocks;\n\trablocks = MIN(rablocks, (unsigned int)nitems(rablkno));\n\tfor (count = 0; count < rablocks; count++) {\n\t\tstep = (erofs_off_t)(count + 1) * sbi->block_size;\n\t\tif (off > UINT64_MAX - step || current.m_pa > UINT64_MAX - step)\n\t\t\tbreak;\n\t\tfuture = (struct erofs_map_dev) {\n\t\t\t.m_pa = off + step,\n\t\t\t.m_deviceid = device_id,\n\t\t\t.m_plen = sbi->block_size,\n\t\t};\n\t\tif (erofs_map_dev(sbi, &future) != 0 ||\n\t\t    future.m_dif != current.m_dif ||\n\t\t    future.m_pa != current.m_pa + step)\n\t\t\tbreak;\n\t\trablkno[count] = btodb(future.m_pa);\n\t\trabsize[count] = sbi->block_size;\n\t}\n\treturn (erofs_bread_device(sbi, current.m_dif, blocks, current.m_pa, len,\n\t    rablkno, rabsize, count, bufp));\n}\n\n/* Release a contiguous buffer returned by erofs_bread(). */\n",
        "physical readahead helper",
    )
    data_candidate = replace_once(
        data_candidate,
        "int\nerofs_read_data(struct erofs_sb_info *sbi, struct erofs_inode *vi, erofs_off_t loff,\n    size_t len, void **bufp)\n",
        "static int\nerofs_read_data_impl(struct erofs_sb_info *sbi, struct erofs_inode *vi,\n    erofs_off_t loff, size_t len, unsigned int rablocks, void **bufp)\n",
        "read data implementation",
    )
    data_candidate = replace_once(
        data_candidate,
        "\t\t\t} else {\n\t\t\t\terror = erofs_read_physical(sbi, map.m_deviceid, map.m_pa,\n\t\t\t\t    want, &blk);\n\t\t\t}\n",
        "\t\t\t} else if (done == 0 && rablocks != 0 &&\n\t\t\t    map.m_flags == EROFS_MAP_MAPPED) {\n\t\t\t\terror = erofs_read_physical_readahead(sbi,\n\t\t\t\t    map.m_deviceid, map.m_pa, want, rablocks, &blk);\n\t\t\t} else {\n\t\t\t\terror = erofs_read_physical(sbi, map.m_deviceid, map.m_pa,\n\t\t\t\t    want, &blk);\n\t\t\t}\n",
        "read data readahead dispatch",
    )
    data_candidate = replace_once(
        data_candidate,
        "\t*bufp = out;\n\treturn (0);\n}\n\n/*\n * Transfer the logical content of an inode directly into a uio.\n",
        "\t*bufp = out;\n\treturn (0);\n}\n\nint\nerofs_read_data(struct erofs_sb_info *sbi, struct erofs_inode *vi,\n    erofs_off_t loff, size_t len, void **bufp)\n{\n\treturn (erofs_read_data_impl(sbi, vi, loff, len, 0, bufp));\n}\n\nint\nerofs_read_data_readahead(struct erofs_sb_info *sbi, struct erofs_inode *vi,\n    erofs_off_t loff, size_t len, bool sequential, void **bufp)\n{\n\tuint64_t remaining;\n\tunsigned int rablocks;\n\n\trablocks = 0;\n\tif (sequential && vi->datalayout == EROFS_INODE_FLAT_PLAIN &&\n\t    sbi->block_size != 0 && (loff & (sbi->block_size - 1)) == 0 &&\n\t    len <= sbi->block_size && loff <= vi->size && len <= vi->size - loff) {\n\t\tremaining = vi->size - loff - len;\n\t\trablocks = MIN(howmany(remaining, sbi->block_size),\n\t\t    (uint64_t)EROFS_DIR_READAHEAD_SLOTS);\n\t}\n\treturn (erofs_read_data_impl(sbi, vi, loff, len, rablocks, bufp));\n}\n\n/*\n * Transfer the logical content of an inode directly into a uio.\n",
        "read data wrappers",
    )

    internal_candidate = replace_once(
        internal_source,
        "int erofs_read_data(struct erofs_sb_info *sbi, struct erofs_inode *vi,\n    erofs_off_t loff, size_t len, void **bufp);\n",
        "int erofs_read_data(struct erofs_sb_info *sbi, struct erofs_inode *vi,\n    erofs_off_t loff, size_t len, void **bufp);\nint erofs_read_data_readahead(struct erofs_sb_info *sbi,\n    struct erofs_inode *vi, erofs_off_t loff, size_t len, bool sequential,\n    void **bufp);\n",
        "internal readahead prototype",
    )

    dir_candidate = replace_once(
        dir_source,
        "\tbool have_previous;\n\tint error;\n",
        "\tbool have_previous, sequential;\n\tint error;\n",
        "directory sequential state",
    )
    dir_candidate = replace_once(
        dir_candidate,
        "\tlogical_off = uio->uio_offset;\n\tuiodir.last_cookie = logical_off;\n",
        "\tlogical_off = uio->uio_offset;\n\tsequential = logical_off == 0;\n\tuiodir.last_cookie = logical_off;\n",
        "directory sequential initialization",
    )
    dir_candidate = replace_once(
        dir_candidate,
        "\t\terror = erofs_read_data(sbi, dir, block_off, maxsize,\n\t\t    (void **)&blk);\n",
        "\t\terror = erofs_read_data_readahead(sbi, dir, block_off,\n\t\t    maxsize, sequential, (void **)&blk);\n",
        "directory readahead hint",
    )
    return {
        "src/data.c": data_candidate,
        "src/dir.c": dir_candidate,
        "src/internal.h": internal_candidate,
    }


BENCHMARK_C = r'''#include <sys/types.h>
#include <sys/dirent.h>
#include <sys/endian.h>
#include <sys/stat.h>

#include <errno.h>
#include <fcntl.h>
#include <inttypes.h>
#include <stdint.h>
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <time.h>
#include <unistd.h>
#include <dirent.h>

static uint64_t
fnv_update(uint64_t value, const void *buffer, size_t length)
{
	const unsigned char *bytes = buffer;
	size_t index;

	for (index = 0; index < length; index++) {
		value ^= bytes[index];
		value *= UINT64_C(1099511628211);
	}
	return (value);
}

static uint64_t
record_hash(uint64_t value, const struct dirent *entry)
{
	uint64_t inode, cookie;
	uint16_t namelen;
	uint8_t type;

	inode = htole64(entry->d_fileno);
	type = entry->d_type;
	namelen = htole16(entry->d_namlen);
	cookie = htole64(entry->d_off);
	value = fnv_update(value, &inode, sizeof(inode));
	value = fnv_update(value, &type, sizeof(type));
	value = fnv_update(value, &namelen, sizeof(namelen));
	value = fnv_update(value, &cookie, sizeof(cookie));
	return (fnv_update(value, entry->d_name, entry->d_namlen));
}

static uint64_t
elapsed_ns(const struct timespec *start, const struct timespec *end)
{
	return ((uint64_t)(end->tv_sec - start->tv_sec) * UINT64_C(1000000000) +
	    (uint64_t)(end->tv_nsec - start->tv_nsec));
}

int
main(int argc, char **argv)
{
	struct timespec start, end;
	struct dirent *entry;
	off_t base, last_cookie, seek_offset;
	uint64_t count, hash;
	char *buffer, *cursor;
	long buffer_size;
	ssize_t amount;
	int fd, random_mode;

	if (argc != 5) {
		fprintf(stderr, "usage: benchmark sequential|random DIR BUFFER SEEK\n");
		return (2);
	}
	random_mode = strcmp(argv[1], "random") == 0;
	if (!random_mode && strcmp(argv[1], "sequential") != 0)
		return (2);
	buffer_size = strtol(argv[3], NULL, 10);
	seek_offset = strtoll(argv[4], NULL, 10);
	if (buffer_size <= 0 || buffer_size > INT32_MAX || seek_offset < 0)
		return (2);
	buffer = malloc((size_t)buffer_size);
	if (buffer == NULL)
		return (3);
	fd = open(argv[2], O_RDONLY | O_DIRECTORY);
	if (fd < 0)
		return (4);
	if (random_mode && lseek(fd, seek_offset, SEEK_SET) != seek_offset)
		return (5);
	hash = UINT64_C(1469598103934665603);
	count = 0;
	last_cookie = random_mode ? seek_offset : 0;
	if (clock_gettime(CLOCK_MONOTONIC, &start) != 0)
		return (6);
	for (;;) {
		base = 0;
		amount = getdirentries(fd, buffer, (size_t)buffer_size, &base);
		if (amount < 0)
			return (7);
		if (amount == 0)
			break;
		cursor = buffer;
		while (cursor < buffer + amount) {
			entry = (struct dirent *)cursor;
			if (entry->d_reclen < _GENERIC_DIRSIZ(entry) ||
			    cursor + entry->d_reclen > buffer + amount ||
			    entry->d_off <= last_cookie)
				return (8);
			hash = record_hash(hash, entry);
			last_cookie = entry->d_off;
			count++;
			cursor += entry->d_reclen;
		}
		if (random_mode)
			break;
	}
	if (clock_gettime(CLOCK_MONOTONIC, &end) != 0)
		return (9);
	printf("%" PRIu64 "\t%" PRIu64 "\t%016" PRIx64 "\t%jd\n",
	    elapsed_ns(&start, &end), count, hash, (intmax_t)last_cookie);
	close(fd);
	free(buffer);
	return (0);
}
'''


def guest_script(oracle: dict[str, Any]) -> str:
    return f'''#!/bin/sh
set -eu

work=$1
phase=$2
result=$work/result
mkdir -p "$result/baseline" "$result/candidate" "$work/freebsd-src" "$work/package"
cleanup()
{{
	for mountpoint in "$work"/mnt-*; do
		test -d "$mountpoint" || continue
		mount | grep -F " on $mountpoint " >/dev/null 2>&1 && umount "$mountpoint" || true
	done
	kldstat -n erofs.ko >/dev/null 2>&1 && kldunload erofs.ko || true
}}
trap cleanup EXIT HUP INT TERM

if test ! -f "$result/prepared"; then
	tar -xzf "$work/freebsd-sys.tar.gz" -C "$work/freebsd-src"
	tar -xzf "$work/gate-package.tar.gz" -C "$work/package"
	cc -O2 -Wall -Wextra -Werror "$work/package/benchmark.c" -o "$work/benchmark"
	uname -a > "$result/uname.txt"
	sha256 -q "$work/freebsd-src/sys/sys/buf.h" > "$result/guest-buf-h.sha256"
	for variant in baseline candidate; do
		(
			cd "$work/package/$variant/repo-pre-15"
			env FREEBSD_SRC="$work/freebsd-src" WITH_ZSTDIO=0 sh ./build.sh
		) > "$result/build-$variant.log" 2>&1
		cp "$work/package/$variant/repo-pre-15/build/erofs.ko" "$result/$variant/erofs.ko"
		nm -g "$result/$variant/erofs.ko" | awk '{{print $NF}}' | sort -u > "$result/globals-$variant.txt"
		nm -u "$result/$variant/erofs.ko" | awk '{{print $NF}}' | sort -u > "$result/undefined-$variant.txt"
	done
	comm -13 "$result/globals-baseline.txt" "$result/globals-candidate.txt" > "$result/globals-added.txt"
	comm -23 "$result/globals-baseline.txt" "$result/globals-candidate.txt" > "$result/globals-removed.txt"
	diff -u "$result/undefined-baseline.txt" "$result/undefined-candidate.txt" > "$result/undefined.diff" || true
	dmesg > "$result/dmesg-before.txt"
	: > "$result/runs.tsv"
	printf '%s\n' reached > "$result/target.marker"
	printf '%s\n' prepared > "$result/prepared"
fi

iostat_read()
{{
	iostat -Ix -d "$1" | awk -v device="$1" '$1 == device {{printf "%.0f %.0f\\n", $2, $4}}'
}}

run_one()
{{
	run_id=$1
	device=$2
	mode=$3
	mountpoint="$work/mnt-$run_id"
	mkdir "$mountpoint"
	set -- $(iostat_read "$device")
	reads_before=$1
	kb_before=$2
	mount -t erofs -o ro "/dev/$device" "$mountpoint"
	if test "$mode" = sequential; then
		bench=$($work/benchmark sequential "$mountpoint" {SPEC['benchmark']['buffer_bytes']} 0)
	else
		bench=$($work/benchmark random "$mountpoint" {SPEC['benchmark']['random_buffer_bytes']} {oracle['random_offset']})
	fi
	sleep 1
	umount "$mountpoint"
	set -- $(iostat_read "$device")
	reads_after=$1
	kb_after=$2
	rmdir "$mountpoint"
	set -- $bench
	printf '%s\t%s\t%s\t%s\t%s\t%s\t%s\t%s\t%s\t%s\n' \
	    "$variant" "$run_id" "$device" "$mode" "$1" \
	    "$((reads_after - reads_before))" "$((kb_after - kb_before))" \
	    "$2" "$3" "$4" >> "$result/runs.tsv"
}}

case "$phase" in
baseline-a)
	variant=baseline
	runs='seq-1:da0:sequential seq-4:da3:sequential'
	;;
candidate-a)
	variant=candidate
	runs='seq-2:da1:sequential seq-3:da2:sequential seq-6:da5:sequential'
	;;
baseline-b)
	variant=baseline
	runs='seq-5:da4:sequential seq-8:da7:sequential seq-9:da8:sequential random-baseline:da10:random'
	;;
candidate-b)
	variant=candidate
	runs='seq-7:da6:sequential seq-10:da9:sequential random-candidate:da11:random'
	;;
*)
	exit 2
	;;
esac

kldstat -v > "$result/kldstat-$phase-before.txt"
sysctl -n vfs.conflist > "$result/vfs-conflist-$phase-before.txt"
if ! kldload "$result/$variant/erofs.ko" > "$result/kldload-$phase.log" 2>&1; then
	dmesg > "$result/dmesg-$phase-load-failure.txt"
	exit 70
fi
for spec in $runs; do
	oldifs=$IFS
	IFS=:
	set -- $spec
	IFS=$oldifs
	run_one "$1" "$2" "$3"
done
cleanup
trap - EXIT HUP INT TERM
if test "$phase" = candidate-b; then
	! kldstat -n erofs.ko >/dev/null 2>&1
	dmesg > "$result/dmesg-after.txt"
	kldstat > "$result/kldstat-after.txt"
	mount > "$result/mount-after.txt"
	sha256 -q "$result/baseline/erofs.ko" > "$result/erofs-baseline.sha256"
	sha256 -q "$result/candidate/erofs.ko" > "$result/erofs-candidate.sha256"
fi
'''


def prepare_fixture(temp: Path) -> tuple[Path, dict[str, Any], list[str]]:
    source = temp / "fixture-source"
    source.mkdir()
    count = int(SPEC["benchmark"]["entry_count"])
    name_length = int(SPEC["benchmark"]["name_length"])
    for index in range(count):
        prefix = f"f{index:05d}-"
        name = prefix + chr(ord("a") + index % 26) * (name_length - len(prefix))
        descriptor = os.open(source / name, os.O_WRONLY | os.O_CREAT | os.O_EXCL, 0o444)
        os.close(descriptor)
    fixture_a = temp / "directory-a.erofs"
    fixture_b = temp / "directory-b.erofs"
    command = [
        "mkfs.erofs", "-d0", "-T0", "--all-time", "--all-root",
        "--workers=1", "-x-1", "-E", "noinline_data", "-U",
        SPEC["fixture"]["uuid"], str(fixture_a), str(source),
    ]
    env = dict(os.environ)
    env["SOURCE_DATE_EPOCH"] = "0"
    first = run(command, env=env)
    command[-2] = str(fixture_b)
    second = run(command, env=env)
    if fixture_a.read_bytes() != fixture_b.read_bytes() or first.stdout != second.stdout:
        raise GateFailure("INFRA_BLOCKED", "large-directory fixture is not byte reproducible")
    transform_a = promote_root_tail_to_plain(fixture_a)
    transform_b = promote_root_tail_to_plain(fixture_b)
    if transform_a != transform_b or fixture_a.read_bytes() != fixture_b.read_bytes():
        raise GateFailure("INFRA_BLOCKED", "plain-directory derivation is not byte reproducible")
    fsck = run(["fsck.erofs", "-d0", str(fixture_a)], check=False)
    if fsck.returncode != 0:
        raise GateFailure("INFRA_BLOCKED", f"derived plain fixture fails fsck: {fsck.stdout}")
    fixture_out = fixture_a
    image = Image(fixture_out.read_bytes())
    oracle = image.directory_oracle()
    if oracle["directory_bytes"] < SPEC["fixture"]["minimum_directory_bytes"]:
        raise GateFailure("STOP", "generated flat-plain directory is below the frozen workload size")
    if oracle["entry_count"] < count:
        raise GateFailure("INFRA_BLOCKED", "independent oracle lost generated entries")
    oracle["generator_transform"] = transform_a
    oracle["fsck_exit"] = fsck.returncode
    canonical = command[:-2] + ["DIRECTORY.erofs", "SOURCE"]
    return fixture_out, oracle, canonical


def create_package(temp: Path, resolved: str, sources: dict[str, str],
                   oracle: dict[str, Any]) -> tuple[Path, str, dict[str, Any]]:
    archive = temp / "baseline.tar"
    with archive.open("wb") as output_file:
        completed = subprocess.run(
            ["git", "-C", str(ROOT), "archive", "--format=tar", resolved, "repo-pre-15"],
            stdout=output_file,
            stderr=subprocess.PIPE,
            check=False,
        )
    if completed.returncode != 0:
        raise GateFailure("INFRA_BLOCKED", completed.stderr.decode("utf-8", "replace"))
    package_root = temp / "package-root"
    baseline = package_root / "baseline"
    candidate = package_root / "candidate"
    baseline.mkdir(parents=True)
    with tarfile.open(archive) as tar:
        tar.extractall(baseline, filter="data")
    shutil.copytree(baseline, candidate)
    candidate_sources = make_candidate(
        sources["repo-pre-15/src/data.c"],
        sources["repo-pre-15/src/dir.c"],
        sources["repo-pre-15/src/internal.h"],
    )
    patch_lines: list[str] = []
    for relative, after in candidate_sources.items():
        before = sources[f"repo-pre-15/{relative}"]
        target = candidate / "repo-pre-15" / relative
        target.write_text(after, encoding="utf-8")
        patch_lines.extend(difflib.unified_diff(
            before.splitlines(keepends=True), after.splitlines(keepends=True),
            fromfile=f"a/{relative}", tofile=f"b/{relative}",
        ))
    patch = "".join(patch_lines)
    (OUTPUT / "candidate.patch").write_text(patch, encoding="utf-8")
    semantic = {
        "backing_vnode_only": "breadn(dif->devvp" in candidate_sources["src/data.c"],
        "cluster_read_absent": "cluster_read" not in patch,
        "directory_vnode_breadn_absent": "breadn" not in candidate_sources["src/dir.c"],
        "errno_token_multiset_unchanged": sorted(re.findall(r"return \(([A-Z][A-Z0-9_]*)\);", sources["repo-pre-15/src/data.c"])) == sorted(re.findall(r"return \(([A-Z][A-Z0-9_]*)\);", candidate_sources["src/data.c"])),
        "geom_ownership_calls_not_added": not re.search(r"\b(?:g_attach|g_detach|g_access|g_destroy_consumer)\s*\(", patch),
        "locking_calls_not_added": not re.search(r"\b(?:VOP_LOCK|vn_lock|lockmgr|mtx_lock|sx_xlock)\s*\(", patch),
        "max_readahead_bytes": SPEC["prototype"]["max_readahead_bytes"],
        "random_seek_gate": "sequential = logical_off == 0;" in candidate_sources["src/dir.c"],
        "read_hint_is_directory_only": candidate_sources["src/dir.c"].count("erofs_read_data_readahead(") == 1,
        "vm_entrypoints_not_added": not re.search(r"\b(?:vnode_create_vobject|vm_object|VOP_BMAP)\b", patch),
        "write_set": sorted(candidate_sources),
    }
    semantic["pass"] = all(
        value for key, value in semantic.items()
        if key not in {"max_readahead_bytes", "write_set"}
    ) and semantic["max_readahead_bytes"] <= 1024 * 1024
    write_json(OUTPUT / "semantic-ledger.json", semantic)
    (package_root / "benchmark.c").write_text(BENCHMARK_C, encoding="ascii")
    write_json(package_root / "oracle.json", oracle)
    (OUTPUT / "benchmark.c").write_text(BENCHMARK_C, encoding="ascii")
    return package_root, patch, semantic


def build_modules(temp: Path, package_root: Path) -> dict[str, Path]:
    helper = ROOT / "repo-pre-15/tests/pre15/fixtures/B12-build-kld.sh"
    if not helper.is_file():
        raise GateFailure("INFRA_BLOCKED", f"missing B12 KLD helper: {helper}")
    modules: dict[str, Path] = {}
    symbols: dict[str, list[str]] = {}
    undefined: dict[str, list[str]] = {}
    module_root = temp / "modules"
    for variant in ("baseline", "candidate"):
        output_dir = module_root / variant
        output_dir.mkdir(parents=True)
        output = output_dir / "erofs.ko"
        work = temp / f"kld-work-{variant}"
        completed = run([
            "/bin/sh", str(helper),
            str(package_root / variant / "repo-pre-15"),
            str(FREEBSD_SRC), str(output), str(work),
        ], timeout=600)
        (OUTPUT / f"build-{variant}.log").write_text(
            completed.stdout, encoding="utf-8"
        )
        modules[variant] = output
        globals_output = run(["nm", "-g", str(output)]).stdout.splitlines()
        undefined_output = run(["nm", "-u", str(output)]).stdout.splitlines()
        symbols[variant] = sorted({line.split()[-1] for line in globals_output if line.split()})
        undefined[variant] = sorted({line.split()[-1] for line in undefined_output if line.split()})
        (OUTPUT / f"module-{variant}.sha256").write_text(
            f"{sha256_path(output)}  erofs.ko\n", encoding="ascii"
        )
    added = sorted(set(symbols["candidate"]) - set(symbols["baseline"]))
    removed = sorted(set(symbols["baseline"]) - set(symbols["candidate"]))
    (OUTPUT / "globals-added.txt").write_text("\n".join(added) + ("\n" if added else ""), encoding="ascii")
    (OUTPUT / "globals-removed.txt").write_text("\n".join(removed) + ("\n" if removed else ""), encoding="ascii")
    undefined_added = sorted(set(undefined["candidate"]) - set(undefined["baseline"]))
    undefined_removed = sorted(set(undefined["baseline"]) - set(undefined["candidate"]))
    (OUTPUT / "undefined-added.txt").write_text(
        "\n".join(undefined_added) + ("\n" if undefined_added else ""),
        encoding="ascii",
    )
    (OUTPUT / "undefined-removed.txt").write_text(
        "\n".join(undefined_removed) + ("\n" if undefined_removed else ""),
        encoding="ascii",
    )
    undefined_diff = list(difflib.unified_diff(
        [line + "\n" for line in undefined["baseline"]],
        [line + "\n" for line in undefined["candidate"]],
        fromfile="baseline", tofile="candidate",
    ))
    (OUTPUT / "undefined.diff").write_text("".join(undefined_diff), encoding="ascii")
    return modules


def ssh_options(port: int) -> list[str]:
    return [
        "-q",
        "-o", "BatchMode=no",
        "-o", "PubkeyAuthentication=no",
        "-o", "PreferredAuthentications=keyboard-interactive,password",
        "-o", "NumberOfPasswordPrompts=1",
        "-o", "StrictHostKeyChecking=no",
        "-o", "UserKnownHostsFile=/dev/null",
        "-o", "ConnectTimeout=5",
        "-p", str(port),
    ]


def scp_options(port: int) -> list[str]:
    return [
        "-q",
        "-o", "BatchMode=no",
        "-o", "PubkeyAuthentication=no",
        "-o", "PreferredAuthentications=keyboard-interactive,password",
        "-o", "NumberOfPasswordPrompts=1",
        "-o", "StrictHostKeyChecking=no",
        "-o", "UserKnownHostsFile=/dev/null",
        "-o", "ConnectTimeout=5",
        "-P", str(port),
    ]


def qemu_replay(temp: Path, fixture: Path, modules: dict[str, Path],
                oracle: dict[str, Any]) -> dict[str, Any]:
    inputs = SPEC["host_inputs"]
    base_image = Path(inputs["base_image"])
    askpass = Path(inputs["askpass_path"])
    ssh_key = Path(inputs["ssh_key_path"])
    if base_image.stat().st_size != inputs["base_image_size"]:
        raise GateFailure("INFRA_BLOCKED", "base .bp size changed")
    if sha256_path(askpass) != inputs["askpass_sha256"] or sha256_path(ssh_key) != inputs["ssh_key_sha256"]:
        raise GateFailure("INFRA_BLOCKED", "QEMU authentication input changed")
    protected_pid = int(SPEC["protected"]["pid"])
    protected_port = int(SPEC["protected"]["port"])
    protected_proc = Path(f"/proc/{protected_pid}")
    if not protected_proc.exists():
        raise GateFailure("INFRA_BLOCKED", "protected QEMU PID is absent")
    protected_start = (protected_proc / "stat").read_text().split()[21]
    protected_cmd = sha256_path(protected_proc / "cmdline")
    base_before = base_image.stat()
    auth_env = dict(os.environ)
    auth_env.update({"DISPLAY": ":0", "SSH_ASKPASS": str(askpass), "SSH_ASKPASS_REQUIRE": "force"})
    rows: list[str] = []
    ownership_rounds: list[dict[str, Any]] = []
    sequential_counts = {"baseline": 0, "candidate": 0}
    prepared_overlay = temp / "prepared-overlay.qcow2"

    def protected_ok() -> bool:
        return (
            protected_proc.exists()
            and (protected_proc / "stat").read_text().split()[21] == protected_start
            and sha256_path(protected_proc / "cmdline") == protected_cmd
        )

    def pick_port() -> int:
        while True:
            with socket.socket() as listener:
                listener.bind(("127.0.0.1", 0))
                selected = listener.getsockname()[1]
            if selected != protected_port:
                return selected

    def prepare_overlay() -> None:
        prepare_dir = OUTPUT / "prepare"
        prepare_dir.mkdir()
        run([
            "qemu-img", "create", "-q", "-f", "qcow2", "-F",
            inputs["base_image_format"], "-b", str(base_image),
            str(prepared_overlay),
        ])
        port = pick_port()
        serial = prepare_dir / "serial.log"
        argv = [
            "qemu-system-x86_64", "-accel", "tcg,thread=multi", "-cpu", "qemu64",
            "-m", str(SPEC["qemu"]["memory_mb"]), "-smp", str(SPEC["qemu"]["cpus"]),
            "-drive", f"file={prepared_overlay},if=virtio,format=qcow2,cache=none",
            "-netdev", f"user,id=net0,hostfwd=tcp:127.0.0.1:{port}-:22",
            "-device", "virtio-net-pci,netdev=net0", "-display", "none",
            "-serial", f"file:{serial}", "-monitor", "none",
        ]
        write_json(prepare_dir / "qemu-argv.json", argv)
        qemu_stdout = (prepare_dir / "qemu-process.log").open("w", encoding="utf-8")
        process = subprocess.Popen(
            argv, stdin=subprocess.DEVNULL, stdout=qemu_stdout,
            stderr=subprocess.STDOUT, text=True,
        )
        if process.pid == protected_pid:
            process.terminate()
            qemu_stdout.close()
            raise GateFailure("INFRA_BLOCKED", "prepare QEMU reused protected PID")
        (prepare_dir / "owned-pid.txt").write_text(f"{process.pid}\n", encoding="ascii")
        (prepare_dir / "owned-port.txt").write_text(f"{port}\n", encoding="ascii")
        options = ssh_options(port)
        copy_options = scp_options(port)
        guest_root = "/root/pre15-p15087"
        pending_error: Exception | None = None

        def ssh(command: str, timeout: int = 60, check: bool = True) -> subprocess.CompletedProcess[str]:
            return run(
                ["ssh", *options, "root@127.0.0.1", command],
                env=auth_env, timeout=timeout, check=check,
            )

        def scp_to(local: Path, remote: str) -> None:
            run(
                ["scp", *copy_options, str(local), f"root@127.0.0.1:{remote}"],
                env=auth_env, timeout=120,
            )

        try:
            deadline = time.monotonic() + int(SPEC["qemu"]["boot_timeout_seconds"])
            while time.monotonic() < deadline:
                if process.poll() is not None:
                    raise GateFailure("INFRA_BLOCKED", "prepare QEMU exited before SSH")
                if ssh("true", timeout=8, check=False).returncode == 0:
                    break
                time.sleep(2)
            else:
                raise GateFailure("INFRA_BLOCKED", "prepare QEMU SSH boot deadline expired")
            ssh(f"mkdir -p {guest_root}/baseline {guest_root}/candidate {guest_root}/evidence")
            scp_to(modules["baseline"], f"{guest_root}/baseline/erofs.ko")
            scp_to(modules["candidate"], f"{guest_root}/candidate/erofs.ko")
            scp_to(OUTPUT / "benchmark.c", f"{guest_root}/benchmark.c")
            scp_to(fixture, f"{guest_root}/fixture-1.erofs")
            scp_to(fixture, f"{guest_root}/fixture-2.erofs")
            setup = ssh(
                f"cc -O2 -Wall -Wextra -Werror {guest_root}/benchmark.c -o {guest_root}/benchmark && "
                f"uname -a > {guest_root}/evidence/uname.txt && "
                f"sysctl -n kern.osreldate > {guest_root}/evidence/osreldate.txt && "
                f"sha256 -q {guest_root}/fixture-1.erofs > {guest_root}/evidence/fixture-1.sha256 && "
                f"sha256 -q {guest_root}/fixture-2.erofs > {guest_root}/evidence/fixture-2.sha256 && "
                f"sha256 -q {guest_root}/baseline/erofs.ko > {guest_root}/evidence/baseline.sha256 && "
                f"sha256 -q {guest_root}/candidate/erofs.ko > {guest_root}/evidence/candidate.sha256 && sync",
                timeout=180, check=False,
            )
            if setup.returncode != 0:
                raise GateFailure("INFRA_BLOCKED", f"prepare guest setup failed: {setup.stdout}")
            prepare_guest = prepare_dir / "guest"
            prepare_guest.mkdir()
            run(
                ["scp", *copy_options, "-r", f"root@127.0.0.1:{guest_root}/evidence/.", str(prepare_guest)],
                env=auth_env, timeout=120,
            )
            ssh("shutdown -p now", timeout=10, check=False)
            try:
                process.wait(timeout=90)
            except subprocess.TimeoutExpired:
                process.terminate()
                process.wait(timeout=15)
        except Exception as error:
            pending_error = error
        finally:
            if process.poll() is None:
                process.terminate()
                try:
                    process.wait(timeout=15)
                except subprocess.TimeoutExpired:
                    process.kill()
                    process.wait(timeout=10)
            qemu_stdout.close()
        port_released = False
        for _ in range(50):
            with socket.socket() as check_socket:
                port_released = check_socket.connect_ex(("127.0.0.1", port)) != 0
            if port_released:
                break
            time.sleep(0.1)
        ownership = {
            "owned_pid": process.pid,
            "owned_pid_stopped": process.poll() is not None,
            "owned_port": port,
            "owned_port_free_after": port_released,
            "prepared_overlay_present": prepared_overlay.exists(),
        }
        write_json(prepare_dir / "ownership.json", ownership)
        if not all((ownership["owned_pid_stopped"], port_released,
                    ownership["prepared_overlay_present"])):
            raise GateFailure("INFRA_BLOCKED", "prepare ownership audit failed")
        if pending_error is not None:
            raise pending_error

    def run_round(round_name: str, samples: list[tuple[str, str]]) -> None:
        round_dir = OUTPUT / "rounds" / round_name
        round_dir.mkdir(parents=True)
        overlay = temp / f"{round_name}-overlay.qcow2"
        run([
            "qemu-img", "create", "-q", "-f", "qcow2", "-F", "qcow2",
            "-b", str(prepared_overlay), str(overlay),
        ])
        port = pick_port()
        serial = round_dir / "serial.log"
        qemu_log = round_dir / "qemu.log"
        argv = [
            "qemu-system-x86_64", "-accel", "tcg,thread=multi", "-cpu", "qemu64",
            "-m", str(SPEC["qemu"]["memory_mb"]), "-smp", str(SPEC["qemu"]["cpus"]),
            "-drive", f"file={overlay},if=virtio,format=qcow2,cache=none",
            "-netdev", f"user,id=net0,hostfwd=tcp:127.0.0.1:{port}-:22",
            "-device", "virtio-net-pci,netdev=net0",
            "-display", "none", "-serial", f"file:{serial}", "-monitor", "none",
        ]
        write_json(round_dir / "qemu-argv.json", argv)
        qemu_stdout = (round_dir / "qemu-process.log").open("w", encoding="utf-8")
        process = subprocess.Popen(
            argv, stdin=subprocess.DEVNULL, stdout=qemu_stdout,
            stderr=subprocess.STDOUT, text=True,
        )
        if process.pid == protected_pid:
            process.terminate()
            qemu_stdout.close()
            raise GateFailure("INFRA_BLOCKED", "owned QEMU reused protected PID")
        (round_dir / "owned-pid.txt").write_text(f"{process.pid}\n", encoding="ascii")
        (round_dir / "owned-port.txt").write_text(f"{port}\n", encoding="ascii")
        options = ssh_options(port)
        copy_options = scp_options(port)
        guest_root = "/root/pre15-p15087"
        guest_reached = False
        guest_md_removed = False
        pending_error: Exception | None = None

        def ssh(command: str, timeout: int = 60, check: bool = True) -> subprocess.CompletedProcess[str]:
            return run(
                ["ssh", *options, "root@127.0.0.1", command],
                env=auth_env, timeout=timeout, check=check,
            )

        def scp_to(local: Path, remote: str) -> None:
            run(
                ["scp", *copy_options, str(local), f"root@127.0.0.1:{remote}"],
                env=auth_env, timeout=120,
            )

        try:
            deadline = time.monotonic() + int(SPEC["qemu"]["boot_timeout_seconds"])
            while time.monotonic() < deadline:
                if process.poll() is not None:
                    raise GateFailure("INFRA_BLOCKED", f"{round_name}: QEMU exited before SSH")
                if ssh("true", timeout=8, check=False).returncode == 0:
                    break
                time.sleep(2)
            else:
                raise GateFailure("INFRA_BLOCKED", f"{round_name}: SSH boot deadline expired")
            ssh(f"mkdir -p {guest_root}/evidence")
            setup = ssh(
                f"test -x {guest_root}/benchmark && "
                f"test -f {guest_root}/fixture-1.erofs && "
                f"test -f {guest_root}/fixture-2.erofs && "
                f"uname -a > {guest_root}/evidence/uname-round.txt && "
                f"sysctl -n kern.osreldate > {guest_root}/evidence/osreldate.txt && "
                f"kldstat > {guest_root}/evidence/kldstat-before.txt && "
                f"mount > {guest_root}/evidence/mount-before.txt && "
                f"mdconfig -l > {guest_root}/evidence/md-before.txt && "
                f"dmesg > {guest_root}/evidence/dmesg-before.txt && "
                f"sha256 -q {guest_root}/baseline/erofs.ko > {guest_root}/evidence/baseline.sha256 && "
                f"sha256 -q {guest_root}/candidate/erofs.ko > {guest_root}/evidence/candidate.sha256 && "
                f"sha256 -q {guest_root}/fixture-1.erofs > {guest_root}/evidence/fixture-1.sha256 && "
                f"sha256 -q {guest_root}/fixture-2.erofs > {guest_root}/evidence/fixture-2.sha256",
                timeout=120, check=False,
            )
            if setup.returncode != 0:
                raise GateFailure("INFRA_BLOCKED", f"{round_name}: guest setup failed: {setup.stdout}")
            guest_reached = True
            for device_index, (variant, mode) in enumerate(samples):
                fixture_index = device_index + 1
                if mode == "sequential":
                    sequential_counts[variant] += 1
                    run_id = f"seq-{sequential_counts[variant]}"
                    buffer_bytes = int(SPEC["benchmark"]["buffer_bytes"])
                    seek_offset = 0
                else:
                    run_id = f"random-{variant}"
                    buffer_bytes = int(SPEC["benchmark"]["random_buffer_bytes"])
                    seek_offset = int(oracle["random_offset"])
                evidence_id = run_id if mode == "random" else f"{run_id}-{variant}"
                command = f'''set -eu
work={shlex.quote(guest_root)}
variant={shlex.quote(variant)}
fixture="$work/fixture-{fixture_index}.erofs"
mode={shlex.quote(mode)}
run_id={shlex.quote(run_id)}
evidence_id={shlex.quote(evidence_id)}
mountpoint="$work/mnt-$evidence_id"
device=
cleanup()
{{
	mount | grep -F " on $mountpoint " >/dev/null 2>&1 && umount "$mountpoint" || true
	kldstat -n erofs.ko >/dev/null 2>&1 && kldunload erofs.ko || true
	test -z "$device" || mdconfig -d -u "${{device#md}}" >/dev/null 2>&1 || true
	rmdir "$mountpoint" >/dev/null 2>&1 || true
}}
trap cleanup EXIT HUP INT TERM
mkdir "$mountpoint"
device=$(mdconfig -a -t vnode -f "$fixture")
test -c "/dev/$device"
set -- $(iostat -Ix -d "$device" | awk -v device="$device" '$1 == device {{printf "%.0f %.0f\\n", $2, $4; found=1}} END {{exit !found}}')
reads_before=$1
kb_before=$2
kldload "$work/$variant/erofs.ko"
kldstat -v > "$work/evidence/kldstat-$evidence_id.txt"
mdconfig -lv > "$work/evidence/md-$evidence_id-before.txt"
mount -t erofs -o ro "/dev/$device" "$mountpoint"
bench=$("$work/benchmark" "$mode" "$mountpoint" {buffer_bytes} {seek_offset})
printf '%s\n' "$bench" > "$work/evidence/benchmark-$evidence_id.txt"
sleep 1
umount "$mountpoint"
set -- $(iostat -Ix -d "$device" | awk -v device="$device" '$1 == device {{printf "%.0f %.0f\\n", $2, $4; found=1}} END {{exit !found}}')
reads_after=$1
kb_after=$2
kldunload erofs.ko
record_device=$device
mdconfig -d -u "${{device#md}}"
device=
rmdir "$mountpoint"
trap - EXIT HUP INT TERM
set -- $bench
	printf '%s\t%s\t%s\t%s\t%s\t%s\t%s\t%s\t%s\t%s\n' \
	    "$variant" "$run_id" "$record_device" "$mode" "$1" \
    "$((reads_after - reads_before))" "$((kb_after - kb_before))" \
    "$2" "$3" "$4"
'''
                sample = ssh(command, timeout=int(SPEC["qemu"]["guest_timeout_seconds"]), check=False)
                (round_dir / f"sample-{evidence_id}.log").write_text(sample.stdout, encoding="utf-8")
                if sample.returncode != 0:
                    raise GateFailure("INFRA_BLOCKED", f"{round_name}: {variant} {mode} sample failed: {sample.stdout}")
                sample_rows = [line for line in sample.stdout.splitlines() if line.count("\t") == 9]
                if len(sample_rows) != 1:
                    raise GateFailure("INFRA_BLOCKED", f"{round_name}: malformed sample output: {sample.stdout}")
                rows.append(sample_rows[0])
            ssh(
                f"dmesg > {guest_root}/evidence/dmesg-after.txt && "
                f"kldstat > {guest_root}/evidence/kldstat-after.txt && "
                f"mount > {guest_root}/evidence/mount-after.txt && "
                f"mdconfig -l > {guest_root}/evidence/md-after.txt",
                timeout=60,
            )
            guest_evidence = round_dir / "guest"
            guest_evidence.mkdir()
            run(
                ["scp", *copy_options, "-r", f"root@127.0.0.1:{guest_root}/evidence/.", str(guest_evidence)],
                env=auth_env, timeout=120,
            )
            cleanup = ssh(
                f"test -z \"$(mount -t erofs)\" && "
                f"! kldstat -n erofs.ko >/dev/null 2>&1 && "
                f"test -z \"$(mdconfig -l)\"",
                timeout=60, check=False,
            )
            if cleanup.returncode != 0:
                raise GateFailure("INFRA_BLOCKED", f"{round_name}: guest cleanup failed: {cleanup.stdout}")
            guest_md_removed = True
            ssh("shutdown -p now", timeout=10, check=False)
            try:
                process.wait(timeout=90)
            except subprocess.TimeoutExpired:
                process.terminate()
                process.wait(timeout=15)
        except Exception as error:
            pending_error = error
            if guest_reached:
                ssh(f"dmesg > {guest_root}/evidence/dmesg-failure.txt", timeout=30, check=False)
                partial = round_dir / "guest-partial"
                partial.mkdir(exist_ok=True)
                run(
                    ["scp", *copy_options, "-r", f"root@127.0.0.1:{guest_root}/evidence/.", str(partial)],
                    env=auth_env, timeout=60, check=False,
                )
        finally:
            if process.poll() is None:
                process.terminate()
                try:
                    process.wait(timeout=15)
                except subprocess.TimeoutExpired:
                    process.kill()
                    process.wait(timeout=10)
            qemu_stdout.close()
            overlay.unlink(missing_ok=True)
        port_free = False
        for _ in range(50):
            with socket.socket() as check_socket:
                port_free = check_socket.connect_ex(("127.0.0.1", port)) != 0
            if port_free:
                break
            time.sleep(0.1)
        round_ownership = {
            "name": round_name,
            "owned_pid": process.pid,
            "owned_pid_stopped": process.poll() is not None,
            "owned_port": port,
            "owned_port_free_after": port_free,
            "overlay_removed": not overlay.exists(),
            "guest_md_removed": guest_md_removed,
        }
        ownership_rounds.append(round_ownership)
        write_json(round_dir / "ownership.json", round_ownership)
        if not all((round_ownership["owned_pid_stopped"], port_free,
                    round_ownership["overlay_removed"], round_ownership["guest_md_removed"])):
            raise GateFailure("INFRA_BLOCKED", f"{round_name}: ownership cleanup failed")
        if pending_error is not None:
            raise pending_error

    prepare_overlay()
    if not prepared_overlay.is_file():
        raise GateFailure("INFRA_BLOCKED", "prepared overlay disappeared before sampling")
    for round_index in range(1, 6):
        order = [("baseline", "sequential"), ("candidate", "sequential")]
        if round_index % 2 == 0:
            order.reverse()
        run_round(f"sequential-{round_index}", order)
    run_round("random", [("baseline", "random"), ("candidate", "random")])
    (OUTPUT / "runs.tsv").write_text("\n".join(rows) + "\n", encoding="ascii")
    (OUTPUT / "target.marker").write_text("reached\n", encoding="ascii")
    base_after = base_image.stat()
    base_ok = (
        base_before.st_ino == base_after.st_ino
        and base_before.st_size == base_after.st_size
        and base_before.st_mtime_ns == base_after.st_mtime_ns
        and base_before.st_ctime_ns == base_after.st_ctime_ns
    )
    ownership = {
        "base_bp_metadata_unchanged": base_ok,
        "base_bp_path": str(base_image),
        "protected_pid": protected_pid,
        "protected_pid_identity_unchanged": protected_ok(),
        "protected_port": protected_port,
        "protected_port_used": False,
        "rounds": ownership_rounds,
    }
    write_json(OUTPUT / "ownership.json", ownership)
    if not base_ok or not ownership["protected_pid_identity_unchanged"]:
        raise GateFailure("INFRA_BLOCKED", f"QEMU ownership audit failed: {ownership}")
    return ownership


def evaluate(oracle: dict[str, Any], semantic: dict[str, Any]) -> dict[str, Any]:
    run_path = OUTPUT / "runs.tsv"
    rows = []
    for line in run_path.read_text(encoding="ascii").splitlines():
        fields = line.split("\t")
        if len(fields) != 10:
            raise GateFailure("INFRA_BLOCKED", f"bad guest run row: {line}")
        rows.append({
            "variant": fields[0], "run": fields[1], "device": fields[2],
            "mode": fields[3], "elapsed_ns": int(fields[4]),
            "provider_reads": int(fields[5]), "provider_kb": int(fields[6]),
            "entry_count": int(fields[7]), "hash": fields[8],
            "final_cookie": int(fields[9]),
        })
    sequential = [row for row in rows if row["mode"] == "sequential"]
    baseline = [row for row in sequential if row["variant"] == "baseline"]
    candidate = [row for row in sequential if row["variant"] == "candidate"]
    random_rows = [row for row in rows if row["mode"] == "random"]
    if len(baseline) != 5 or len(candidate) != 5 or len(random_rows) != 2:
        raise GateFailure("INFRA_BLOCKED", "guest did not produce the frozen 5+5+2 run matrix")
    correctness = all(
        row["entry_count"] == oracle["entry_count"]
        and row["hash"] == oracle["fnv64"]
        and row["final_cookie"] == oracle["final_cookie"]
        for row in sequential
    )
    baseline_ns = statistics.median(row["elapsed_ns"] for row in baseline)
    candidate_ns = statistics.median(row["elapsed_ns"] for row in candidate)
    improvement = (baseline_ns - candidate_ns) * 100.0 / baseline_ns
    baseline_reads = statistics.median(row["provider_reads"] for row in baseline)
    candidate_reads = statistics.median(row["provider_reads"] for row in candidate)
    if baseline_reads <= 0:
        raise GateFailure("INFRA_BLOCKED", "provider read counter did not observe baseline I/O")
    extra_reads = max(0.0, (candidate_reads - baseline_reads) * 100.0 / baseline_reads)
    random_baseline = next(row for row in random_rows if row["variant"] == "baseline")
    random_candidate = next(row for row in random_rows if row["variant"] == "candidate")
    random_noop = (
        random_baseline["entry_count"] == random_candidate["entry_count"]
        and random_baseline["hash"] == random_candidate["hash"]
        and random_baseline["final_cookie"] == random_candidate["final_cookie"]
        and random_baseline["provider_reads"] == random_candidate["provider_reads"]
    )
    added = (OUTPUT / "globals-added.txt").read_text().split()
    removed = (OUTPUT / "globals-removed.txt").read_text().split()
    undefined_diff = (OUTPUT / "undefined.diff").read_text().strip()
    symbol_contract = added == ["erofs_read_data_readahead"] and not removed and not undefined_diff
    dmesg = "\n".join(
        path.read_text(encoding="utf-8", errors="replace")
        for path in sorted((OUTPUT / "rounds").glob("*/guest/dmesg-after.txt"))
    )
    kernel_clean = not re.search(
        r"panic:|lock order reversal|KASAN:|Witness.*warning|link_elf_obj:.*(?:error|undefined)|linker_load_file:.*error",
        dmesg, re.IGNORECASE,
    )
    thresholds = SPEC["thresholds"]
    checks = {
        "cold_median_improvement": improvement >= thresholds["minimum_cold_median_improvement_percent"],
        "correctness_hash_cookie": correctness,
        "extra_provider_reads": extra_reads <= thresholds["maximum_extra_provider_reads_percent"],
        "kernel_log_clean": kernel_clean,
        "random_seek_noop": random_noop,
        "semantic_contract": semantic["pass"],
        "symbol_contract": symbol_contract,
        "window_bound": semantic["max_readahead_bytes"] <= 1024 * 1024,
    }
    status = "GO" if all(checks.values()) else "STOP"
    result = {
        "b12": "AUTHORIZED" if status == "GO" else "STOP-NO-SOURCE",
        "baseline_cold_elapsed_ns": [row["elapsed_ns"] for row in baseline],
        "baseline_median_elapsed_ns": baseline_ns,
        "baseline_median_provider_reads": baseline_reads,
        "candidate": "P15-087",
        "candidate_cold_elapsed_ns": [row["elapsed_ns"] for row in candidate],
        "candidate_median_elapsed_ns": candidate_ns,
        "candidate_median_provider_reads": candidate_reads,
        "checks": checks,
        "cold_median_improvement_percent": improvement,
        "extra_provider_reads_percent": extra_reads,
        "full_feature_suite": "NOT_RUN",
        "gate": "G11",
        "oracle": oracle,
        "qemu": "PASS",
        "random_rows": random_rows,
        "requested_base": REQUESTED_BASE,
        "schema": 1,
        "sequential_rows": sequential,
        "status": status,
        "thresholds": thresholds,
    }
    write_json(OUTPUT / "result.json", result)
    return result


def finalize() -> None:
    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)}")
    (OUTPUT / "SHA256SUMS").write_text("\n".join(lines) + "\n", encoding="ascii")


result: dict[str, Any] | None = None
exit_code = 0
try:
    if SPEC.get("schema") != 1 or SPEC.get("candidate") != "P15-087" or SPEC.get("gate") != "G11":
        raise GateFailure("INFRA_BLOCKED", "invalid P15-087 input schema")
    resolved = git("rev-parse", f"{REQUESTED_BASE}^{{commit}}")
    if resolved != SPEC["required_base"]:
        raise GateFailure("INFRA_BLOCKED", f"P15-087 must replay {SPEC['required_base']}, got {resolved}")
    sources = {path: source_at(resolved, path) for path in SPEC["source_sha256"]}
    source_hashes = {path: sha256_bytes(text.encode("utf-8")) for path, text in sources.items()}
    if source_hashes != SPEC["source_sha256"]:
        raise GateFailure("INFRA_BLOCKED", "frozen DUT/Linux source identity changed")
    freebsd_head = subprocess.check_output(["git", "-C", str(FREEBSD_SRC), "rev-parse", "HEAD"], text=True).strip()
    freebsd_hashes = {path: sha256_path(FREEBSD_SRC / path) for path in SPEC["freebsd_sha256"]}
    if freebsd_head != SPEC["freebsd_head"] or freebsd_hashes != SPEC["freebsd_sha256"]:
        raise GateFailure("INFRA_BLOCKED", "FreeBSD source identity changed")
    version = run(["mkfs.erofs", "-V"]).stdout.splitlines()[0]
    if version != SPEC["mkfs_version"]:
        raise GateFailure("INFRA_BLOCKED", f"mkfs version changed: {version}")
    write_json(OUTPUT / "source-sha256.json", source_hashes)
    write_json(OUTPUT / "freebsd-source.json", {"head": freebsd_head, "sha256": freebsd_hashes})
    with tempfile.TemporaryDirectory(prefix="p15-087-gate-") as temporary:
        temp = Path(temporary)
        fixture, oracle, command = prepare_fixture(temp)
        oracle["fixture_sha256"] = sha256_path(fixture)
        oracle["mkfs_command"] = command
        write_json(OUTPUT / "oracle.json", oracle)
        package_root, patch, semantic = create_package(temp, resolved, sources, oracle)
        modules = build_modules(temp, package_root)
        qemu_replay(temp, fixture, modules, oracle)
        result = evaluate(oracle, semantic)
        exit_code = 0 if result["status"] == "GO" else 1
except GateFailure as failure:
    status = failure.status
    result = {
        "b12": "STOP-NO-SOURCE" if status == "STOP" else "NOT_RUN",
        "candidate": "P15-087",
        "full_feature_suite": "NOT_RUN",
        "gate": "G11",
        "qemu": "INFRA_BLOCKED" if status == "INFRA_BLOCKED" else "NOT_RUN",
        "reason": failure.reason,
        "requested_base": REQUESTED_BASE,
        "schema": 1,
        "status": status,
    }
    write_json(OUTPUT / "result.json", result)
    exit_code = 21 if status == "INFRA_BLOCKED" else 1
except Exception as failure:
    result = {
        "b12": "NOT_RUN",
        "candidate": "P15-087",
        "full_feature_suite": "NOT_RUN",
        "gate": "G11",
        "qemu": "INFRA_BLOCKED",
        "reason": f"unexpected gate runner failure: {failure}",
        "requested_base": REQUESTED_BASE,
        "schema": 1,
        "status": "INFRA_BLOCKED",
    }
    write_json(OUTPUT / "result.json", result)
    exit_code = 21
finally:
    finalize()

print(json.dumps(result, sort_keys=True))
raise SystemExit(exit_code)
PY
