run

#!/usr/bin/env python3
# SPDX-License-Identifier: Apache-2.0
"""
virtio-villain test runner.

Spawns one VM per test to isolate VMM crashes. Collects serial output
and exit codes, prints a summary.
"""

import argparse
from collections import Counter
from contextlib import ExitStack
import fnmatch
import json
import hashlib
import mmap
import os
import random
import re
import shlex
import signal
import stat
import subprocess
import sys
import fcntl
import shutil
import tempfile
import threading
import platform
import time
from concurrent.futures import ThreadPoolExecutor

_SCRIPT_DIR = os.path.dirname(os.path.abspath(__file__))
sys.path.insert(0, os.path.join(_SCRIPT_DIR, "lib", "python"))
from villain import execution, images, process, reporting  # noqa: E402
from villain.cli import (  # noqa: E402
    ShortFirstHelpFormatter, print_banner, normalize_memory as _normalize_mem,
    canonicalize_tests as _canonicalize_tests,
)
from villain.console import (  # noqa: E402
    merge_streams as _merge_streams,
    restore_terminal as _restore_term, save_terminal as _save_term,
    drain_stream,
)
from villain.diagnostics import security_triage as _security_triage  # noqa: E402
from villain.host import enable_sysconfigs  # noqa: E402
from villain.process import (  # noqa: E402
    install_termination_as_interrupt as _install_termination_as_interrupt,
    kill_all_processes as _kill_all_live_procs, kill_process as _kill_proc,
)
from villain.vmm import (  # noqa: E402
    DEFAULT_VMM_TRACE_FILE, CloudHypervisor, Qemu, QemuMicrovm,
    detect_vmm, fetch_kernel,
)

sys.stdout.reconfigure(line_buffering=True)

_print_lock = threading.Lock()


INIT_BINARY = os.path.join(_SCRIPT_DIR, "target", "init")
INITRAMFS = os.path.join(_SCRIPT_DIR, "target", "initramfs.cpio.gz")
DISK_SIZE = 16 * 1024 * 1024  # 16 MiB
PMEM_SIZE = 128 * 1024 * 1024
DISK_FORMATS = images.FORMATS


def get_test_list(init_binary=INIT_BINARY):
    """Get available tests by running target/init --list-tsv on the host.

    Returns (list of test IDs, dict mapping ID to (desc, ver, section),
    dict mapping ID to {device_id, flags}).
    The --list-tsv output is TAB separated:
        name<TAB>desc<TAB>version<TAB>section[<TAB>device_id<TAB>flags]
    """
    result = subprocess.run(
        [init_binary, "--list-tsv"],
        capture_output=True, text=True
    )
    if result.returncode != 0:
        sys.exit(f"Failed to list tests: {result.stderr}")
    tests = []
    info = {}
    meta = {}
    for line in result.stdout.splitlines():
        if not line:
            continue
        fields = line.split("\t")
        if len(fields) < 4:
            continue
        tid, desc, ver, sec = fields[0], fields[1], fields[2], fields[3]
        tests.append(tid)
        info[tid] = (desc, ver, sec)
        if len(fields) >= 6:
            meta[tid] = {
                "device_id": int(fields[4], 16),
                "flags": int(fields[5]),
                "required_features": int(fields[6], 16) if len(fields) >= 7 else 0,
                "min_queues": int(fields[7]) if len(fields) >= 8 else 0,
            }
    return tests, info, meta


def _parse_result(output, returncode):
    """Parse VM output into a result status string.

    The verdict line is `[TAG] <name>` printed by the guest harness,
    but the VMM may interleave its own log lines into the same stdout
    stream at any character boundary (CH writes its own warnings
    inline). Strip those log lines first, then look for the tag.
    """
    cleaned = _clean_result_output(output)
    m = re.search(r"\[(XPASS|XFAIL|FAIL|REJECT|WEDGED|SKIP|PASS)\]", cleaned)
    if m:
        return m.group(1)
    if returncode < 0:
        return "FAIL"
    if "[vv]" not in cleaned:
        return "FAIL"
    return "WEDGED"


def _parse_batch_results(output, test_names):
    """Parse VM output for multiple verdict lines.

    Returns a dict mapping test name to (status, output). Tests not
    found in the output are marked WEDGED (the VM died before
    reaching them).
    """
    cleaned = _clean_result_output(output)
    results = {}
    for name in test_names:
        marker = re.search(
            rf"\[(XPASS|XFAIL|FAIL|REJECT|WEDGED|SKIP|PASS)\]\s+"
            rf"{re.escape(name)}(?![A-Za-z0-9_])",
            cleaned)
        results[name] = marker.group(1) if marker else "WEDGED"
    return results


_VERDICT_MARKER_RE = re.compile(
    r"\[(?:XPASS|XFAIL|FAIL|REJECT|WEDGED|SKIP|PASS)\]")


def _remove_spliced_log_lines(output, pattern):
    """Remove matching logs while retaining any verdict text inside them."""
    def replace(match):
        line = match.group(0)
        marker = _VERDICT_MARKER_RE.search(line)
        if not marker:
            return ""
        return line[marker.start():].rstrip("\n")

    return re.sub(pattern, replace, output)


def _clean_result_output(output):
    """Remove host and guest log lines that can splice into verdict markers."""
    output = _remove_spliced_log_lines(
        output, r"cloud-hypervisor:[^\n]*(?:\n|$)")
    return _remove_spliced_log_lines(
        output, r"\[\s*\d+\.\d+\][^\n]*(?:\n|$)")


def _find_virtiofsd():
    """Locate a virtiofsd binary on the host. Returns path or None.

    Checked locations in order. The first match wins.
    """
    for cand in ("virtiofsd",
                 "/usr/libexec/virtiofsd",
                 "/usr/lib/virtiofsd",
                 "/usr/lib/qemu/virtiofsd"):
        p = shutil.which(cand) if "/" not in cand else cand
        if p and os.path.exists(p) and os.access(p, os.X_OK):
            return p
    return None


# Default timeout for VMM API control calls. Scaled at startup by the
# host oversubscription factor so sidecars do not flake with a hard
# 5s ceiling when ch-remote and the CH API thread are CPU starved.
_VM_API_TIMEOUT = 5.0
_SIDECAR_JOIN_TIMEOUT = 5.0
_PROBE_CACHE_DIR = f"/dev/shm/virtio-villain-probes-{os.getuid()}"
_PROBE_CACHE_SIZE = 65536


def _probe_cache_key(backend, kernel, initramfs, opts):
    identities = []
    for path in (backend.binary, kernel, initramfs, __file__):
        path = os.path.realpath(path)
        stat = os.stat(path)
        identities.append((path, stat.st_dev, stat.st_ino, stat.st_size,
                           stat.st_mtime_ns, stat.st_ctime_ns))
    with open("/proc/sys/kernel/random/boot_id") as stream:
        boot_id = stream.read().strip()
    data = (boot_id, platform.machine(), backend.name, identities, opts)
    return hashlib.sha256(json.dumps(data, sort_keys=True).encode()).hexdigest()


def _probe_records(output):
    records = []
    cleaned = _clean_result_output(output.decode(errors="replace"))
    for line in cleaned.splitlines():
        if not line.startswith("PROBE\t"):
            continue
        parts = line.split("\t")
        if len(parts) < 4:
            raise ValueError(f"incomplete probe record: {line!r}")
        records.append((
            int(parts[1], 16), int(parts[2], 16), int(parts[3]),
            int(parts[4]) if len(parts) >= 5 else 1,
            int(parts[5]) if len(parts) >= 6 else 1))
    if not records:
        raise ValueError("probe returned no device records")
    return records


def _probe_payload(records):
    return "".join(
        f"PROBE\t{dev:x}\t{features:x}\t{queues}\t{isr}\t{cfg}\n"
        for dev, features, queues, isr, cfg in records).encode()


def _read_probe_cache(fd, decode=_probe_records):
    if os.fstat(fd).st_size != _PROBE_CACHE_SIZE:
        return None
    with mmap.mmap(fd, _PROBE_CACHE_SIZE,
                   access=mmap.ACCESS_READ) as memory:
        length = int.from_bytes(memory[:8], "little")
        if length == 0:
            return None
        return decode(memory[8:8 + length])


def _cached_probe(key, probe, decode=_probe_records, encode=_probe_payload):
    """Reuse shared probe results independently of the launching process."""
    os.makedirs(_PROBE_CACHE_DIR, mode=0o700, exist_ok=True)
    directory = os.stat(_PROBE_CACHE_DIR, follow_symlinks=False)
    if (not stat.S_ISDIR(directory.st_mode) or directory.st_uid != os.getuid()
            or stat.S_IMODE(directory.st_mode) != 0o700):
        raise RuntimeError(f"unsafe probe cache directory: {_PROBE_CACHE_DIR}")
    cache_key = key() if callable(key) else key
    path = os.path.join(_PROBE_CACHE_DIR, cache_key)
    fd = os.open(path, os.O_RDWR | os.O_CREAT | os.O_NOFOLLOW, 0o600)
    try:
        try:
            fcntl.flock(fd, fcntl.LOCK_SH | fcntl.LOCK_NB)
        except BlockingIOError:
            pass
        else:
            try:
                cached = _read_probe_cache(fd, decode)
                if cached is not None:
                    return cached
            finally:
                fcntl.flock(fd, fcntl.LOCK_UN)

        records = decode(probe())
        if records is None:
            return None
        payload = encode(records)
        if len(payload) > _PROBE_CACHE_SIZE - 8:
            raise ValueError("capability probe exceeds shared memory size")
        try:
            fcntl.flock(fd, fcntl.LOCK_EX | fcntl.LOCK_NB)
        except BlockingIOError:
            return records
        if _read_probe_cache(fd, decode) is not None:
            return records
        os.ftruncate(fd, _PROBE_CACHE_SIZE)
        with mmap.mmap(fd, _PROBE_CACHE_SIZE) as memory:
            memory[:8] = bytes(8)
            memory[8:8 + len(payload)] = payload
            memory[:8] = len(payload).to_bytes(8, "little")
        return records
    finally:
        os.close(fd)


def _mmio_boots(backend, kernel, initramfs):
    def probe():
        with tempfile.TemporaryDirectory(prefix="vv-mmio-boot-") as directory:
            disk = os.path.join(directory, "disk.raw")
            images.create_raw(disk, DISK_SIZE)
            command = backend.build_cmd(
                kernel, initramfs, disk,
                f"console={backend.console_device} vv.test=list")
            try:
                result = process.capture(command, timeout=2)
                return result.stdout
            except subprocess.TimeoutExpired as error:
                return error.stdout or b""
            except OSError as error:
                print(f"Note: MMIO boot probe failed: {error}",
                      file=sys.stderr)
                return b""

    def ready(output):
        return True if b"[vv]" in output else None

    try:
        return bool(_cached_probe(
            lambda: _probe_cache_key(
                backend, kernel, initramfs, {"probe": "mmio-boot"}),
            probe, decode=ready, encode=lambda result: b"[vv]"))
    except (OSError, ValueError, RuntimeError) as error:
        print(f"Warning: MMIO boot probe cache unavailable: {error}. "
              "Checking boot without cache.", file=sys.stderr)
        return bool(ready(probe()))


def vm_api(api_socket, backend, command, args=None, timeout=None):
    """Send a control request to a running VM through its API socket.

    For CH, backend must be a CloudHypervisor instance and the call
    shells out to ch-remote. command is the subcommand name, for
    example 'ping', 'info', 'snapshot', 'pause', 'add-disk'. args is
    a list of extra positional or option arguments passed verbatim.
    Returns CompletedProcess with captured stdout and stderr.

    For QEMU, backend must be a Qemu instance and the call speaks
    raw QMP JSON over the unix socket. command is the QMP execute
    name, for example 'query-status' or 'stop'. args is a dict of
    QMP arguments. Returns parsed JSON response or raw bytes on
    parse failure. Raises OSError on connect failure.
    """
    if timeout is None:
        timeout = _VM_API_TIMEOUT
    deadline = time.time() + timeout
    while not os.path.exists(api_socket):
        if time.time() > deadline:
            raise OSError(f"api socket not ready: {api_socket}")
        time.sleep(0.02)

    if isinstance(backend, CloudHypervisor):
        ch_remote = _find_ch_remote(backend.binary)
        if not ch_remote:
            raise FileNotFoundError(
                "ch-remote not found next to the cloud-hypervisor binary "
                "or in PATH")
        cmd = [ch_remote, "--api-socket", api_socket, command]
        if args:
            cmd += list(args)
        return subprocess.run(cmd, capture_output=True, text=True,
                              timeout=timeout)

    if isinstance(backend, (Qemu, QemuMicrovm)):
        import json
        import socket as _socket
        s = _socket.socket(_socket.AF_UNIX, _socket.SOCK_STREAM)
        s.settimeout(timeout)
        s.connect(api_socket)
        try:
            with s.makefile("rb") as stream:
                def read_reply():
                    while True:
                        remaining = deadline - time.time()
                        if remaining <= 0:
                            raise TimeoutError(f"QMP {command} timed out")
                        s.settimeout(remaining)
                        line = stream.readline().rstrip(b"\r\n")
                        try:
                            reply = json.loads(line.decode())
                        except (ValueError, UnicodeDecodeError):
                            return line
                        if not isinstance(reply, dict) or "event" not in reply:
                            return reply

                greeting = read_reply()
                if not isinstance(greeting, dict) or "QMP" not in greeting:
                    raise RuntimeError(f"invalid QMP greeting: {greeting!r}")
                s.sendall(b'{"execute":"qmp_capabilities"}\n')
                capabilities = read_reply()
                if (not isinstance(capabilities, dict)
                        or "return" not in capabilities
                        or "error" in capabilities):
                    raise RuntimeError(
                        f"QMP capabilities failed: {capabilities!r}")
                req = {"execute": command}
                if args is not None:
                    req["arguments"] = args
                s.sendall((json.dumps(req) + "\n").encode())
                return read_reply()
        finally:
            s.close()

    raise ValueError(f"vm_api unsupported backend: {backend}")


def _find_ch_remote(ch_binary):
    """Locate the ch-remote client. Looks next to the cloud-hypervisor
    binary first, then PATH.
    """
    sibling = os.path.join(os.path.dirname(os.path.abspath(ch_binary)),
                           "ch-remote")
    if os.path.isfile(sibling) and os.access(sibling, os.X_OK):
        return sibling
    return shutil.which("ch-remote")


# ---------------------------------------------------------------------------
# Sidecar mechanism. A sidecar is an optional Python file next to a C
# test source. The runner imports it and calls run(ctx) on a daemon
# thread alongside the guest test. The sidecar can drive anything on
# the host side, for example flap a tap link, kill a vhost backend,
# truncate the disk file, or call the VMM control socket. It is not
# tied to the API socket. Tests that do not need a host actor simply
# do not ship a sidecar.

class _SidecarSkip(Exception):
    pass


class _Ctx:
    """Context object passed to a sidecar's run() function."""

    def __init__(self, *, test_id, backend, vmm_pid, tmpdir, api_sock,
                 disk, log, output_buf, output_lock, stop_event, control=None):
        self.test_id = test_id
        self.backend = backend
        self.vmm_pid = vmm_pid
        self.tmpdir = tmpdir
        self.api_sock = api_sock
        self.disk = disk
        self.log = log
        self.stop = stop_event
        self._buf = output_buf
        self._lock = output_lock
        self._control = control

    def skip(self, reason):
        """Stop the host actor because a required prerequisite is absent."""
        raise _SidecarSkip(reason)

    def vm_api(self, command, args=None, timeout=None):
        """Convenience wrapper. Calls vm_api with this context's
        api_sock and backend. Raises RuntimeError if the api socket
        is disabled or the VMM has already exited. API responses and
        connection errors are passed through without changing them.
        """
        if self.api_sock is None:
            raise RuntimeError("api socket disabled for this test")
        if not self.vm_alive():
            raise RuntimeError(f"VMM exited before API command {command}")
        if self._control is not None:
            return self._control(command, args, timeout)
        return vm_api(self.api_sock, self.backend, command, args,
                      timeout=timeout)

    def vm_alive(self):
        """Return True if the VMM process is still around. The guest
        test may finish and the VMM may exit before the sidecar
        reaches a later step. Sidecars should use this to skip
        unnecessary follow up calls instead of erroring out.
        """
        if self.vmm_pid is None:
            return False
        try:
            os.kill(self.vmm_pid, 0)
        except OSError:
            return False
        return True

    def wait_text(self, pattern, timeout=10.0):
        """Block until the VMM stdout buffer contains pattern. Returns
        True if seen, False on timeout or stop signal.
        """
        needle = pattern.encode() if isinstance(pattern, str) else pattern
        deadline = time.time() + timeout
        while time.time() < deadline:
            with self._lock:
                joined = b"".join(self._buf)
            if needle in joined:
                return True
            if self.stop.is_set():
                return False
            time.sleep(0.05)
        return False


def _find_sidecar(test_name):
    """Return absolute path to a sidecar .py for test_name, or None.

    Sidecars live next to the C source under tests/<dir>/. Match is
    case insensitive and accepts the bare ID stem or a stem followed
    by an underscore and free form text.
    """
    td = _test_dir(test_name)
    if not td:
        return None
    base = os.path.join(_SCRIPT_DIR, "tests", td)
    if not os.path.isdir(base):
        return None
    low = test_name.lower()
    for entry in os.listdir(base):
        if not entry.endswith(".py"):
            continue
        stem = entry[:-3].lower()
        if stem == low or stem.startswith(low + "_"):
            return os.path.join(base, entry)
    return None


def _load_sidecar(path):
    """Import a sidecar file as an anonymous module."""
    import importlib.util
    spec = importlib.util.spec_from_file_location(
        "vv_sidecar_" + os.path.splitext(os.path.basename(path))[0], path)
    if spec is None or spec.loader is None:
        raise ImportError(f"cannot load sidecar: {path}")
    mod = importlib.util.module_from_spec(spec)
    spec.loader.exec_module(mod)
    return mod


def run_test(backend, kernel, test_name, timeout, verbose=0, console=False,
             disk_format="raw", opts=None, stop=None):
    """Run one or more tests in a fresh VM.

    test_name is a single test ID string or a list of IDs. When a
    list is given, the tests are submitted as a comma separated
    batch and each gets its own verdict line in the output. Sidecar
    scripts are only supported for single test runs.

    opts["initramfs"] selects a guest archive without changing source paths.

    Returns a list of (name, status, output) tuples.
    """
    if stop is not None and stop.is_set():
        raise KeyboardInterrupt
    if isinstance(test_name, str):
        test_names = [test_name]
    else:
        test_names = list(test_name)
    batch = len(test_names) > 1
    sidecar_path = _find_sidecar(test_names[0]) if not batch else None
    sidecar_mod = _load_sidecar(sidecar_path) if sidecar_path else None
    supported = getattr(sidecar_mod, "SUPPORTED_BACKENDS", None)
    if supported is not None and backend.name not in supported:
        output = (f"[vv-sidecar] SKIP: backend {backend.name} is not "
                  f"supported by {os.path.basename(sidecar_path)}\n")
        return [(test_names[0], "SKIP", output)]
    required_tools = getattr(sidecar_mod, "REQUIRED_TOOLS", {})
    for tool in required_tools.get(backend.name, ()):
        executable = (_find_ch_remote(backend.binary)
                      if backend.name == "ch" and tool == "ch-remote"
                      else shutil.which(tool))
        if not executable:
            output = f"[vv-sidecar] SKIP: required host tool {tool} not found\n"
            return [(test_names[0], "SKIP", output)]
    # A batch runs every test in one VM, so the timeout budget scales
    # with the number of tests. A single per test timeout routinely
    # trips under heavy parallelism and kills the VM mid batch, wedging
    # the tests that had not run yet. A single test keeps its own
    # budget since the factor is one.
    effective_timeout = timeout * len(test_names)
    suffix = {"raw": ".raw", "qcow2": ".qcow2", "vhd": ".vhd",
              "vhdx": ".vhdx"}[disk_format]
    disk_path = tempfile.mktemp(suffix=suffix)
    images.create(disk_path, DISK_SIZE, image_format=disk_format)

    # Optional per test sidecar (single test only). It may declare launch
    # requirements via vmm_config() and/or a run(ctx) host actor.
    sidecar_cfg = {}
    if sidecar_mod is not None and hasattr(sidecar_mod, "vmm_config"):
        sidecar_cfg = sidecar_mod.vmm_config() or {}

    # Guest kernel initcalls to skip. The block device is always present,
    # so keep its guest driver from claiming it. A test whose sidecar
    # drives a device the kernel would otherwise own adds more here (for
    # example the net repro that owns the net and iommu devices itself).
    skip_initcalls = [
        "virtio_blk_init",
        "virtio_scsi_init",
        "virtio_net_driver_init",
        "virtio_console_init",
        "virtio_balloon_driver_init",
        "virtio_rng_driver_init",
        "virtio_mem_driver_init",
        "virtio_pmem_driver_init",
        "virtio_iommu_drv_init",
        "virtio_fs_init",
        "p9_virtio_init",
        "virtio_vsock_init",
        "virtio_rtc_drv_init",
        "virtio_watchdog_driver_init",
        "virtio_input_driver_init",
        "virtio_crypto_driver_init",
        # Subsystems unused by the harness
        "ext4_init_fs",
        "fuse_init",
        "audit_init",
        "cpufreq_core_init",
    ]
    skip_initcalls += list(sidecar_cfg.get("skip_initcalls", []))
    wait_scale = (opts or {}).get("wait_scale", 1)
    transport = "mmio" if backend.name == "qemu-mmio" else "pci"
    cmdline = (f"console={backend.console_device} "
               f"initcall_blacklist={','.join(skip_initcalls)} "
               f"vv.wait_scale={wait_scale} "
               f"vv.transport={transport} "
               f"vv.test={','.join(test_names)}")
    # Create pmem backing file
    pmem_path = disk_path.rsplit(".", 1)[0] + ".pmem"
    images.create_raw(pmem_path, PMEM_SIZE)
    # Copy so per test mutations do not leak into the shared opts
    # dict that the caller reuses across every test.
    opts = dict(opts) if opts else {}
    opts["pmem_path"] = pmem_path

    # Create the virtio-scsi backing file so a SCSI READ has real data
    # to return from the block backend.
    scsi_path = disk_path.rsplit(".", 1)[0] + ".scsi"
    images.create_raw(scsi_path, DISK_SIZE)
    opts["scsi_path"] = scsi_path
    # Second backing file for the 4K block LUN at LUN 1.
    scsi_path2 = disk_path.rsplit(".", 1)[0] + ".scsi2"
    images.create_raw(scsi_path2, DISK_SIZE)
    opts["scsi_path2"] = scsi_path2

    api_socket = None
    openvmm_control = bool(opts.get("api_enabled") and backend.name == "openvmm")
    if opts.get("api_enabled") and backend.name in ("ch", "qemu", "qemu-mmio"):
        api_socket = disk_path.rsplit(".", 1)[0] + ".api"
        opts["api_socket"] = api_socket
    else:
        opts.pop("api_socket", None)
    opts.pop("api_enabled", None)

    # Spawn virtiofsd if this test targets virtio-fs and a daemon is
    # available. The shared directory is a fresh empty tmpdir per test.
    fs_proc = None
    fs_socket = None
    fs_sharedir = None
    if test_names[0].upper().startswith("F"):
        fs_daemon = _find_virtiofsd()
        if fs_daemon:
            fs_socket = disk_path.rsplit(".", 1)[0] + ".fsd"
            fs_sharedir = tempfile.mkdtemp(prefix="vvfs.")
            fs_proc = subprocess.Popen(
                [fs_daemon, "--socket-path", fs_socket,
                 "--shared-dir", fs_sharedir,
                 "--cache", "never"],
                stdin=subprocess.DEVNULL,
                stdout=subprocess.DEVNULL,
                stderr=subprocess.DEVNULL,
                start_new_session=True,
            )
            # Wait briefly for the socket to appear so CH/QEMU can connect.
            for _ in range(50):
                if os.path.exists(fs_socket):
                    break
                time.sleep(0.02)
            opts["fs_socket"] = fs_socket
    p9_sharedir = None
    if test_names[0].upper().startswith("PLAN9"):
        p9_sharedir = tempfile.mkdtemp(prefix="vvp9.")
        opts["p9_sharedir"] = p9_sharedir
    # Apply any VMM opts the sidecar declared (for example placing the
    # net device behind the vIOMMU). These are generic opts the backend
    # build_cmd already understands, so the runner stays test agnostic.
    for key, value in sidecar_cfg.get("opts", {}).items():
        opts[key] = value
    cmd = backend.build_cmd(
        kernel, opts.get("initramfs", INITRAMFS), disk_path, cmdline, opts)

    if verbose >= 1:
        with _print_lock:
            print(f"  $ {shlex.join(cmd)}")

    # Result markers printed by the in-VM harness. As soon as we see one
    # we kill the VM, since not every VMM cleanly exits on guest reboot
    # (notably CH on triple fault, and any teardown wedged by a test).
    # The VMM writes its own log lines into the same stdout stream and
    # can split a verdict marker at any character boundary; strip those
    # lines first, then match the tag.
    vmm_noise_re = re.compile(rb"cloud-hypervisor:[^\n]*(?:\n|$)")
    kernel_noise_re = re.compile(rb"\[\s*\d+\.\d+\][^\n]*(?:\n|$)")
    verdict_marker_re = re.compile(
        rb"\[(?:XPASS|XFAIL|FAIL|REJECT|WEDGED|SKIP|PASS)\]")

    def _remove_stream_log(match):
        line = match.group(0)
        marker = verdict_marker_re.search(line)
        if not marker:
            return b""
        return line[marker.start():].rstrip(b"\n")

    wanted = {t.encode() for t in test_names}
    escaped_names = sorted((re.escape(t) for t in wanted), key=len,
                           reverse=True)
    # Wait for the full line so killing the VM cannot truncate skip reasons.
    marker_name_re = re.compile(
        rb"\[(?:PASS|FAIL|REJECT|WEDGED|SKIP|XFAIL|XPASS)\]\s+("
        + rb"|".join(escaped_names)
        + rb")(?![A-Za-z0-9_])[^\r\n]*[\r\n]")

    saved_term = _save_term() if console else None
    proc = None
    openvmm_master = None
    processes = ExitStack()
    timer = None
    sidecar_thread = None
    sidecar_state = {"err": None, "skip": None}
    sidecar_stop = threading.Event()
    sidecar_tmpdir = None
    sidecar_log_lines = []
    output_lock = threading.Lock()
    output = ""
    parse_text = ""
    status = "FAIL"
    # The VMM writes the guest serial, which carries the verdict
    # markers, to its stdout, and its own log lines to stderr. Merging
    # the two into one pipe lets a log line splice into a marker at a
    # byte boundary and corrupt it, which surfaces as a spurious WEDGED
    # for whichever test provoked a VMM log at verdict time. Keep stderr
    # on its own pipe so the marker stream stays clean, and drain it in
    # a thread. In console mode the drain thread also streams stderr live.
    err_chunks = []
    err_thread = None
    try:
        old_sigint = None
        if console:
            old_sigint = signal.signal(signal.SIGINT, signal.SIG_IGN)
        proc, openvmm_master = processes.enter_context(process.spawn(
            cmd, terminal=backend.name == "openvmm", start_new_session=not console,
            stop=stop, stdout=subprocess.PIPE, stderr=subprocess.PIPE))
        if proc.stderr is not None:
            err_thread = threading.Thread(
                target=drain_stream,
                args=(proc.stderr, err_chunks, console),
                name=f"stderr-{test_names[0]}", daemon=True)
            err_thread.start()
        chunks = []
        if sidecar_mod is not None and hasattr(sidecar_mod, "run"):
            sidecar_tmpdir = tempfile.mkdtemp(prefix=f"vv-sc-{test_names[0]}.")
            sidecar_log_lines = []

            def _sc_log(msg):
                line = f"[vv-sidecar] {msg}"
                sidecar_log_lines.append(line)
                if console:
                    print(line)

            ctx = _Ctx(
                test_id=test_names[0],
                backend=backend,
                vmm_pid=proc.pid,
                tmpdir=sidecar_tmpdir,
                api_sock="repl" if openvmm_control else opts.get("api_socket"),
                disk=disk_path,
                log=_sc_log,
                output_buf=chunks,
                output_lock=output_lock,
                stop_event=sidecar_stop,
            )
            if openvmm_control:
                command_mode = False

                def _openvmm_command(command, arguments, timeout):
                    nonlocal command_mode
                    acknowledgements = {
                        "pause": b"pause complete",
                        "resume": b"resumed complete",
                        "pulse-save-restore": b"pulse save/restore complete",
                    }
                    if command not in acknowledgements or arguments:
                        raise ValueError(f"unsupported OpenVMM command: {command}")
                    deadline = time.monotonic() + (
                        _VM_API_TIMEOUT if timeout is None else timeout)
                    if not command_mode:
                        os.write(openvmm_master, b"\x11")
                        while b"openvmm> " not in (
                                b"".join(chunks) + b"".join(err_chunks)):
                            if proc.poll() is not None:
                                raise RuntimeError(
                                    "OpenVMM exited before command prompt")
                            if time.monotonic() >= deadline:
                                raise TimeoutError("OpenVMM command prompt timed out")
                            time.sleep(0.01)
                        command_mode = True
                    start = len(err_chunks)
                    os.write(openvmm_master, command.encode() + b"\n")
                    while True:
                        stderr_closed = not err_thread.is_alive()
                        reply = b"".join(err_chunks[start:])
                        if acknowledgements[command] in reply:
                            return subprocess.CompletedProcess(
                                [command], 0, "", reply.decode(errors="replace"))
                        complete_reply = (
                            reply if stderr_closed
                            else reply[:reply.rfind(b"\n") + 1])
                        for line in complete_reply.splitlines():
                            if b"ERROR" not in line and b"error:" not in line:
                                continue
                            if command == "pulse-save-restore" and re.search(
                                    rb"\bERROR\b[^\r\n]*pulse save/restore failed "
                                    rb'error=(?:"reset not supported"|reset not supported)'
                                    rb"[ \t\r]*$", line):
                                ctx.skip(
                                    "OpenVMM pulse-save-restore unavailable: "
                                    "reset not supported")
                            raise RuntimeError(
                                f"OpenVMM {command} failed: "
                                f"{reply.decode(errors='replace')}")
                        if proc.poll() is not None and stderr_closed:
                            raise RuntimeError(f"OpenVMM exited during {command}")
                        if time.monotonic() >= deadline:
                            raise TimeoutError(f"OpenVMM {command} timed out")
                        time.sleep(0.01)

                ctx._control = _openvmm_command

            def _sc_run():
                try:
                    sidecar_mod.run(ctx)
                except _SidecarSkip as e:
                    sidecar_state["skip"] = str(e)
                except BaseException as e:  # capture for the runner
                    import traceback as _tb
                    sidecar_state["err"] = (e, _tb.format_exc())

            sidecar_thread = threading.Thread(
                target=_sc_run, name=f"sidecar-{test_names[0]}", daemon=True)
            sidecar_thread.start()
        timer_fired = [False]

        def _on_timeout():
            timer_fired[0] = True
            if console:
                proc.kill()
            else:
                _kill_proc(proc)
        timer = threading.Timer(effective_timeout, _on_timeout)
        timer.start()
        reported = set()
        scan_from = 0
        marker_seen = False
        for chunk in iter(lambda: proc.stdout.read1(4096), b""):
            if console:
                sys.stdout.buffer.write(chunk)
                sys.stdout.buffer.flush()
            with output_lock:
                chunks.append(chunk)
            if not marker_seen:
                with output_lock:
                    joined = b"".join(chunks)
                # Scan only the new tail, with an overlap so a verdict
                # line split across a chunk boundary or a CH log line is
                # still matched. Rescanning the whole growing buffer
                # every chunk is O(n^2) and dominates a long batch run.
                # Accumulate the reported test names and kill once every
                # expected test has printed a full "[TAG] NAME" line,
                # matching how the results are parsed. A bare tag count
                # could reach the target before a verdict line was
                # complete and kill with a name not yet captured, which
                # then parses as a spurious WEDGED.
                tail = joined[max(scan_from - 8192, 0):]
                cleaned_tail = vmm_noise_re.sub(_remove_stream_log, tail)
                cleaned_tail = kernel_noise_re.sub(
                    _remove_stream_log, cleaned_tail)
                for m in marker_name_re.finditer(cleaned_tail):
                    reported.add(m.group(1))
                scan_from = len(joined)
                if wanted <= reported:
                    marker_seen = True
                    if console:
                        proc.kill()
                    else:
                        _kill_proc(proc)
        proc.wait()
        timer.cancel()
        timed_out = not marker_seen and timer_fired[0]
        if old_sigint is not None:
            signal.signal(signal.SIGINT, old_sigint)
        if timed_out:
            raise subprocess.TimeoutExpired(cmd, effective_timeout)
        if err_thread is not None:
            err_thread.join(timeout=5)
        parse_text = b"".join(chunks).decode(errors="replace")
        status = _parse_result(parse_text, proc.returncode)
        output = _merge_streams(parse_text, err_chunks)
    except subprocess.TimeoutExpired:
        if proc:
            proc.kill()
            proc.wait()
        if err_thread is not None:
            err_thread.join(timeout=5)
        # Keep everything the VM already printed. A batch runs many
        # tests in one VM, so discarding the captured output here would
        # erase the verdicts of tests that already reported and mark the
        # whole batch WEDGED. Preserve the real output so only the tests
        # that never printed a marker stay WEDGED; append a timeout note
        # for the single test case where the run itself is the failure.
        parse_text = b"".join(chunks).decode(errors="replace") + \
            f"\n[vv] TIMEOUT after {effective_timeout}s\n"
        output = _merge_streams(parse_text, err_chunks)
        status = "FAIL"
    except KeyboardInterrupt:
        if proc:
            if console:
                proc.kill()
            else:
                _kill_proc(proc)
            proc.wait()
        raise
    except Exception as e:
        output = str(e)
        status = "FAIL"
    finally:
        if timer is not None:
            timer.cancel()
            if timer.ident is not None:
                timer.join()
        if proc and proc.poll() is None:
            _kill_proc(proc, process_group=not console or backend.name == "openvmm")
        if err_thread is not None and err_thread.is_alive():
            err_thread.join(timeout=5)
        sidecar_stop.set()
        if sidecar_thread is not None:
            sidecar_thread.join(timeout=_SIDECAR_JOIN_TIMEOUT)
            err = sidecar_state.get("err")
            if sidecar_thread.is_alive():
                err = (RuntimeError(
                    "host action did not exit within "
                    f"{_SIDECAR_JOIN_TIMEOUT:g}s"), "")
            extra = ""
            if sidecar_log_lines:
                extra += "\n" + "\n".join(sidecar_log_lines) + "\n"
            skip_reason = sidecar_state["skip"]
            if skip_reason is not None:
                extra += f"\n[vv-sidecar] SKIP: {skip_reason}\n"
                if status not in ("FAIL", "WEDGED", "XPASS"):
                    status = "SKIP"
            if err is not None:
                missing_device = status == "SKIP" and re.search(
                    r"\[SKIP\]\s+" + re.escape(test_names[0])
                    + r"\s+\(no device 0x[0-9a-fA-F]+\)", parse_text)
                if missing_device:
                    extra += ("\n[vv-sidecar] SKIP: guest device unavailable"
                              f"; host action stopped: {err[0]}\n")
                else:
                    extra += f"\n[vv-sidecar] FAIL: {err[0]}\n{err[1]}\n"
                    if status != "WEDGED":
                        status = "FAIL"
            output += extra
        if sidecar_tmpdir:
            shutil.rmtree(sidecar_tmpdir, ignore_errors=True)
        processes.close()
        _restore_term(saved_term)
        if console:
            signal.signal(signal.SIGINT, signal.default_int_handler)
        if fs_proc:
            try:
                fs_proc.terminate()
                fs_proc.wait(timeout=2)
            except (subprocess.TimeoutExpired, OSError):
                try:
                    fs_proc.kill()
                except OSError:
                    pass
        if fs_socket:
            try:
                os.unlink(fs_socket)
            except FileNotFoundError:
                pass
            try:
                os.unlink(fs_socket + ".pid")
            except FileNotFoundError:
                pass
        if fs_sharedir:
            shutil.rmtree(fs_sharedir, ignore_errors=True)
        if p9_sharedir:
            shutil.rmtree(p9_sharedir, ignore_errors=True)
        os.unlink(disk_path)
        try:
            os.unlink(pmem_path)
        except FileNotFoundError:
            pass
        try:
            os.unlink(scsi_path)
        except FileNotFoundError:
            pass
        try:
            os.unlink(scsi_path2)
        except FileNotFoundError:
            pass
        if api_socket:
            try:
                os.unlink(api_socket)
            except FileNotFoundError:
                pass
        vsock_sock = disk_path.replace(suffix, ".vsock")
        try:
            os.unlink(vsock_sock)
        except FileNotFoundError:
            pass

    if batch:
        verdicts = _parse_batch_results(parse_text, test_names)
        return [(n, verdicts[n], output) for n in test_names]
    return [(test_names[0], status, output)]


def _retry_wedged(batch_results, backend, kernel, timeout, verbose,
                  disk_format, opts):
    """Re-run WEDGED tests from a batch individually.

    When a test crashes or wedges inside a batch, subsequent tests in
    the same VM never run and appear as WEDGED. A genuine crash has a
    culprit: the first WEDGED test is the one that was running when the
    VM died, so it keeps its verdict and the tests after it are re-run
    individually.

    A batch that was killed by the timeout has no culprit. The first
    WEDGED test is just where the clock ran out, not a test that
    misbehaved, so it must be re-run individually too. Otherwise a
    benign test that happened to be live at the deadline reports a
    spurious WEDGED on every loaded run.
    """
    wedged = [name for name, st, _ in batch_results if st == "WEDGED"]
    if not wedged:
        return batch_results
    # The whole batch shares one captured output. run_test appends a
    # timeout note when the VM was killed by the deadline rather than by
    # a test crashing it.
    batch_output = batch_results[0][2] if batch_results else ""
    timed_out = "[vv] TIMEOUT after" in batch_output
    # Find the first WEDGED test. Everything before it has a real
    # verdict. On a crash the first one stays WEDGED as the culprit; on
    # a timeout it is retried like the rest.
    first_wedged_idx = None
    for i, (name, st, _) in enumerate(batch_results):
        if st == "WEDGED":
            first_wedged_idx = i
            break
    keep_upto = first_wedged_idx if timed_out else first_wedged_idx + 1
    final = list(batch_results[:keep_upto])
    for name, st, out in batch_results[keep_upto:]:
        if st == "WEDGED":
            solo = run_test(backend, kernel, name, timeout, verbose,
                            False, disk_format, opts)
            final.extend(solo)
        else:
            final.append((name, st, out))
    return final


# Maps test name prefix (leading uppercase letters) to the source
# directory under tests/. Keep in sync with the layout in README.md.
PREFIX_DIR = {
    "T": "vring",
    "P": "packed",
    "PCI": "pci",
    "M": "mmio",
    "S": "state",
    "A": "admin",
    "B": "blk",
    "Z": "blk",
    "SCSI": "scsi",
    "N": "net",
    "C": "console",
    "RNG": "rng",
    "V": "vsock",
    "L": "balloon",
    "E": "pmem",
    "R": "mem",
    "D": "watchdog",
    "I": "iommu",
    "RTC": "rtc",
    "F": "fs",
    "PLAN9": "plan9",
    "CR": "crypto",
    "SND": "sound",
}

# Reverse: directory name to set of accepted prefixes.
DIR_PREFIXES = {}
for _p, _d in PREFIX_DIR.items():
    DIR_PREFIXES.setdefault(_d, set()).add(_p)

# Maps test name prefix to the PCI device ID that must be present.
# Transport tests (T, PCI, S) use BLK as the default probe device.
PREFIX_DEVICE_ID = {
    "B": 0x1042, "Z": 0x1042,   # blk
    "SCSI": 0x1048,             # scsi
    "N": 0x1041,                 # net
    "C": 0x1043,                 # console
    "RNG": 0x1044,               # rng
    "L": 0x1045,                 # balloon
    "V": 0x1053,                 # vsock
    "I": 0x1057,                 # iommu
    "R": 0x1058,                 # mem
    "E": 0x105b,                 # pmem
    "D": 0x1063,                 # watchdog
    "RTC": 0x1051,               # rtc
    "F": 0x105a,                 # fs
    "PLAN9": 0x1049,             # 9p
    "CR": 0x1054,                # crypto
    "SND": 0x1059,               # sound
}


def _test_prefix(name):
    """Return the recognized leading test prefix, or ''."""
    m = re.match(r"^(?:PLAN9|[A-Z]+)", name)
    return m.group(0) if m else ""


def _test_dir(name):
    """Return source directory for a test name, or None if unknown.

    Matches the longest known prefix so that 'RTC0016' maps to 'rtc'
    rather than 'R' or 'RT'.
    """
    pref = _test_prefix(name)
    while pref:
        if pref in PREFIX_DIR:
            return PREFIX_DIR[pref]
        pref = pref[:-1]
    return None


def _expand_dir_globs(patterns, flag_name):
    """Expand each pattern against known directory names.

    Plain names (no glob meta) must match exactly an entry in
    DIR_PREFIXES, otherwise sys.exit. Glob patterns may match zero
    or more entries and an empty match is allowed silently because
    the user may legitimately want to test that a category is empty.
    Matching is case insensitive.
    """
    expanded = set()
    for pat in patterns or []:
        low = pat.lower()
        if any(c in low for c in "*?["):
            for name in DIR_PREFIXES:
                if fnmatch.fnmatchcase(name, low):
                    expanded.add(name)
        else:
            if low not in DIR_PREFIXES:
                sys.exit(f"Unknown test directory for {flag_name}: {pat}. "
                         f"Known: {', '.join(sorted(DIR_PREFIXES))}")
            expanded.add(low)
    return expanded


def filter_tests(tests, includes, excludes, dirs, exclude_dirs):
    """Apply include/exclude pattern and directory filters.

    Includes and dirs are unioned to form the whitelist. If both are
    empty, every test is initially in the whitelist. Excludes and
    exclude_dirs then remove entries. Pattern arguments accept shell
    globs through fnmatch.
    """
    wanted_dirs = _expand_dir_globs(dirs, "--dir")
    excl_dirs = _expand_dir_globs(exclude_dirs, "--exclude-dir")

    if includes or dirs:
        result = []
        for t in tests:
            hit = False
            for pat in includes or []:
                if fnmatch.fnmatchcase(t.upper(), pat.upper()):
                    hit = True
                    break
            if not hit and wanted_dirs:
                td = _test_dir(t)
                if td and td in wanted_dirs:
                    hit = True
            if hit:
                result.append(t)
    else:
        result = list(tests)

    if excludes:
        result = [t for t in result
                  if not any(fnmatch.fnmatchcase(t.upper(), p.upper())
                             for p in excludes)]
    if excl_dirs:
        result = [t for t in result
                  if (_test_dir(t) or "") not in excl_dirs]
    return result


_JUNIT_FAILURE = {"FAIL", "WEDGED", "XPASS"}
_JUNIT_SKIPPED = {"SKIP", "XFAIL"}

# Verdict severity, most severe first. Used to pick a representative
# verdict when every retry of a test failed but with a different kind.
_STATUS_RANK = {
    "FAIL": 0, "WEDGED": 1, "XPASS": 2,
    "REJECT": 4, "PASS": 5, "SKIP": 6, "XFAIL": 7,
}


def _status_rank(status):
    return _STATUS_RANK.get(status, 99)


# Prefix marking a result whose first attempt failed and was retried.
# The rest of the line lists the verdict of every attempt in order. A
# retried test that recovered to an acceptable verdict is transient,
# not a stable property of the test; one that failed every attempt keeps
# its failure verdict and still gates.
_ATTEMPTS_PREFIX = "[vv] attempts: "


def _fold_retries(name, statuses, details):
    """Combine a failing test with its solo reruns into one result.

    statuses lists the verdict of every attempt in order, starting with
    the first attempt. The result keeps the real resolved verdict, not a
    synthetic one. When a retry reached an acceptable verdict that
    verdict is reported and the attempts note marks the result as
    retried so the caller can list it separately. When every attempt
    failed the most severe failure is reported and the run gates.
    details holds the (status, output) of the attempts whose output is
    specific to this test, which excludes a batched first attempt
    because its log is shared with the whole batch. The attached output
    is a solo attempt's own log, preferring the most severe solo failure
    and otherwise the last solo rerun, so there is always test-specific
    context and never the shared batch log.
    """
    note = _ATTEMPTS_PREFIX + ", ".join(statuses) + "\n"
    if statuses[-1] not in _JUNIT_FAILURE:
        verdict = statuses[-1]
    else:
        verdict = min(statuses, key=_status_rank)
    fails = [(st, o) for st, o in details if st in _JUNIT_FAILURE and o]
    if fails:
        detail = min(fails, key=lambda p: _status_rank(p[0]))[1]
    elif details:
        detail = details[-1][1] or ""
    else:
        detail = ""
    return (name, verdict, note + detail)


def _recovered_on_retry(status, output):
    """True if a result failed first and recovered on a rerun.

    Such a result carries the attempts note and a non failing verdict.
    It is reported as its real verdict and listed separately as a
    retried result, not claimed to be an inherent property of the test.
    A retried test that kept failing has a failing verdict and gates.
    """
    return (status not in _JUNIT_FAILURE
            and bool(output) and output.startswith(_ATTEMPTS_PREFIX))


def _apply_retries(batch_results, retries, backend, kernel, timeout,
                   verbose, disk_format, opts, batched=False):
    """Retry only the failing tests to absorb transient failures.

    Acceptable verdicts (PASS, REJECT, SKIP, XFAIL) are left untouched,
    so a green run and batching pay no extra cost. A failing verdict
    (FAIL, WEDGED, XPASS) is retried solo up to `retries` more times,
    stopping at the first acceptable verdict. A test that fails then
    recovers keeps that real verdict and is marked retried; a test that
    fails every attempt keeps its most severe failure and still gates.
    When the first attempt came from a batch its output is shared with
    the batch, so it is not attached to the single test; only solo
    reruns provide detail.
    """
    if retries <= 0:
        return batch_results
    out = []
    for name, status, output in batch_results:
        if status not in _JUNIT_FAILURE:
            out.append((name, status, output))
            continue
        statuses = [status]
        # Attempts whose output belongs to this test alone. A batched
        # first attempt is excluded because its log covers every test in
        # the batch.
        details = [] if batched else [(status, output)]
        for _ in range(retries):
            r = run_test(backend, kernel, name, timeout, verbose,
                         False, disk_format, opts)
            if len(r) != 1:
                break
            _, st, o = r[0]
            statuses.append(st)
            details.append((st, o))
            if st not in _JUNIT_FAILURE:
                break
        out.append(_fold_retries(name, statuses, details))
    return out


def _xml_safe(text):
    """Drop code points that XML 1.0 forbids so the report stays valid."""
    return "".join(
        ch for ch in text
        if ch in ("\t", "\n", "\r")
        or 0x20 <= ord(ch) <= 0xD7FF
        or 0xE000 <= ord(ch) <= 0xFFFD
        or 0x10000 <= ord(ch) <= 0x10FFFF
    )


def build_results_doc(results, test_info, *, backend_name, vmm):
    """Normalize run results into a JSON serializable document.

    results is a list of (name, status, output) tuples. test_info maps
    a test name to (description, spec_version, spec_section). Output is
    kept only for non PASS verdicts to keep the report lean. Each failing
    verdict (FAIL, WEDGED, XPASS) also carries a security tier from the
    triage in _security_triage; other verdicts carry security None. A
    result that failed first and recovered on a rerun keeps that real
    verdict and is flagged retried.
    """
    tests = []
    for name, status, output in results:
        desc, ver, sec = test_info.get(name, ("", "", ""))
        retried = _recovered_on_retry(status, output or "")
        out = (output or "") if status != "PASS" else ""
        security = None
        if status in _JUNIT_FAILURE:
            tier, why = _security_triage(out)
            security = {"tier": tier, "rationale": why}
        tests.append({
            "name": name,
            "status": status,
            "retried": retried,
            "description": desc,
            "spec_version": ver,
            "spec_section": sec,
            "output": out,
            "security": security,
        })
    counts = {
        key: sum(1 for t in tests if t["status"] == st)
        for key, st in (
            ("pass", "PASS"), ("fail", "FAIL"), ("reject", "REJECT"),
            ("wedged", "WEDGED"), ("skip", "SKIP"), ("xfail", "XFAIL"),
            ("xpass", "XPASS"),
        )
    }
    counts["retried"] = sum(1 for t in tests if t["retried"])
    return {
        "tool": "virtio-villain",
        "backend": backend_name,
        "vmm": vmm,
        "total": len(tests),
        "counts": counts,
        "tests": tests,
    }


def format_results_json(doc):
    """Render the results document as indented JSON."""
    return json.dumps(doc, indent=2)


def format_results_junit(doc):
    """Render the results document as a JUnit XML report.

    FAIL, WEDGED and XPASS map to <failure> so a CI gate fails on them,
    matching the runner exit code. SKIP and XFAIL map to <skipped>.
    PASS and REJECT are reported as plain passing testcases.
    """
    import xml.etree.ElementTree as ET

    counts = doc["counts"]
    failures = counts["fail"] + counts["wedged"] + counts["xpass"]
    skipped = counts["skip"] + counts["xfail"]
    attrs = {
        "name": "virtio-villain",
        "tests": str(doc["total"]),
        "failures": str(failures),
        "errors": "0",
        "skipped": str(skipped),
    }
    root = ET.Element("testsuites", attrs)
    suite = ET.SubElement(root, "testsuite", attrs)
    classname = f"virtio-villain.{doc['backend']}"
    for t in doc["tests"]:
        case = ET.SubElement(suite, "testcase", {
            "classname": classname,
            "name": t["name"],
            "time": "0",
        })
        status = t["status"]
        if status in _JUNIT_FAILURE:
            sec = t.get("security")
            attrs = {
                "type": status,
                "message": t["description"] or status,
            }
            prefix = ""
            if sec:
                attrs["security"] = sec["tier"]
                prefix = f"security: {sec['tier']} -- {sec['rationale']}\n\n"
            node = ET.SubElement(case, "failure", attrs)
            node.text = _xml_safe(prefix + t["output"])
        elif status in _JUNIT_SKIPPED:
            node = ET.SubElement(case, "skipped", {"message": status})
            if t["output"]:
                node.text = _xml_safe(t["output"])
    body = ET.tostring(root, encoding="unicode")
    return '<?xml version="1.0" encoding="UTF-8"?>\n' + body


def _desc_column_width(cols, longest_desc, fixed):
    """Width of the description column in the result table.

    Capped at the longest description in the run so a wide terminal does
    not pad each line with a long run of spaces before the version and
    section columns, and floored at 10 so a narrow terminal still shows
    something. fixed is the width taken by the other columns.
    """
    return reporting.desc_column_width(cols, longest_desc, fixed)


def _qemu_aio_forces_direct(backend_name, io_engine, direct):
    """Whether the runner must enable direct IO implicitly and warn.

    QEMU aio=native needs O_DIRECT, so selecting the aio engine turns
    direct IO on, but only for the QEMU backends. CH has no such
    requirement, so a CH run must not print the note.
    """
    return (io_engine == "aio" and not direct
            and backend_name.startswith("qemu"))


def main():
    if sys.argv[1:2] == ["--suite"]:
        from villain.appliance import main as run_appliance
        sys.exit(run_appliance(sys.argv[1:]))
    parser = argparse.ArgumentParser(
        description="virtio-villain test runner",
        formatter_class=ShortFirstHelpFormatter,
        add_help=False,
        epilog=(
            "examples\n"
            "  run --suite /path/to/suite --list\n"
            "        load an external appliance suite without virtio setup\n"
            "  run -m ./cloud-hypervisor\n"
            "        run all tests on Cloud Hypervisor\n"
            "  run -m ./qemu-system-x86_64 --order fast -j 4\n"
            "        QEMU, pool friendly order, four in parallel\n"
            "  run -m ./cloud-hypervisor -i 'B*' -x 'B02*'\n"
            "        block tests, excluding the B02 hotplug family\n"
            "  run -m ./qemu-system-x86_64 --io-engine io_uring B0001\n"
            "        one test with the io_uring block engine\n"
        ),
    )

    g_vmm = parser.add_argument_group("VMM and guest")
    g_vmm.add_argument("-m", "--vmm", required=True, metavar="PATH",
                       help="Path to the VMM binary (required)")
    g_vmm.add_argument("-k", "--kernel", default=None, metavar="PATH",
                       help="Guest kernel, fetched automatically if omitted")
    g_vmm.add_argument("--transport", choices=("pci", "mmio"), default=None,
                       help="Select the QEMU virtio transport")
    g_vmm.add_argument("--accel", choices=("kvm", "tcg"), default=None,
                       help="QEMU accelerator (default: kvm)")
    g_vmm.add_argument("--init", default=None, metavar="PATH",
                       help="Host executable for test discovery "
                            "(default: target/init)")
    g_vmm.add_argument("--initramfs", default=INITRAMFS, metavar="PATH",
                       help="Guest archive (default: target/initramfs.cpio.gz)")
    g_vmm.add_argument("-s", "--cpus", type=int, default=None, metavar="N",
                       help="vCPUs per VM (default: 4 CH, 1 QEMU)")
    g_vmm.add_argument("-M", "--memory", default="512M", type=_normalize_mem,
                       metavar="SIZE",
                       help="Guest memory size (default: 512M)")
    g_vmm.add_argument("-d", "--disk-type", choices=DISK_FORMATS,
                       default="raw",
                       help="Disk image format (default: raw)")
    g_vmm.add_argument("--blk-queues", type=int, default=4, metavar="N",
                       help="Block device queues (default: 4)")
    g_vmm.add_argument("--net-queues", type=int, default=4, metavar="N",
                       help="Net device queue pairs (default: 4)")
    g_vmm.add_argument("--packed", action="store_true", default=True,
                       help="Use packed virtqueues (default)")
    g_vmm.add_argument("--no-packed", action="store_true",
                       help="Use split virtqueues")
    g_vmm.add_argument("--io-engine",
                       choices=["default", "io_uring", "aio", "sync"],
                       default="default",
                       help="Block IO engine, one of default, io_uring, aio "
                            "(Linux AIO) or sync (default: default)")
    g_vmm.add_argument("--direct", action="store_true",
                       help="Open block backing files with O_DIRECT")

    g_sel = parser.add_argument_group("test selection")
    g_sel.add_argument("tests", nargs="*", metavar="TEST",
                       help="Test IDs to run (default: all)")
    g_sel.add_argument("-i", "--include", action="append", default=[],
                       metavar="GLOB",
                       help="Run tests matching shell GLOB, repeatable")
    g_sel.add_argument("-x", "--exclude", action="append", default=[],
                       metavar="GLOB",
                       help="Skip tests matching shell GLOB, repeatable")
    g_sel.add_argument("--dir", action="append", default=[], dest="dirs",
                       metavar="NAME",
                       help="Run tests from directory NAME under tests/, "
                            "repeatable")
    g_sel.add_argument("--exclude-dir", action="append", default=[],
                       metavar="NAME",
                       help="Skip tests from directory NAME, repeatable")
    g_sel.add_argument("--skip-unsupported", action="store_true",
                       help="Probe the VMM and drop tests it cannot run")
    g_sel.add_argument("--order",
                       choices=["default", "fast", "alpha", "random",
                                "reverse"],
                       default="default",
                       help="Execution order, one of default, fast, alpha, "
                            "random or reverse (default: default)")
    g_sel.add_argument("--batch", type=int, default=0, metavar="N",
                       help="Run N tests per VM boot, 0 disables (default: 0)")

    g_exec = parser.add_argument_group("execution")
    g_exec.add_argument("-j", "--jobs", type=int, default=1, metavar="N",
                        help="Run N tests in parallel (default: 1)")
    g_exec.add_argument("-t", "--timeout", type=int, default=30,
                        metavar="SEC",
                        help="Per test timeout in seconds (default: 30)")
    g_exec.add_argument("--retries", type=int, default=0, metavar="N",
                        help="Retry a failing test up to N times to spot "
                             "flakiness (default: 0)")

    g_out = parser.add_argument_group("output and reporting")
    g_out.add_argument("-c", "--console", action="store_true",
                       help="Stream VM output directly to stdout")
    g_out.add_argument("-l", "--log-dir", default=None, metavar="DIR",
                       help="Save per test output into this directory")
    g_out.add_argument("-v", "--verbose", action="count", default=0,
                       help="Increase verbosity (-v commands, -vv full "
                            "output)")
    g_out.add_argument("--format", choices=["text", "json", "junit"],
                       default="text",
                       help="Report format, one of text, json or junit "
                            "(default: text)")
    g_out.add_argument("--output", default=None, metavar="FILE",
                       help="Write the report to FILE instead of stdout")

    g_diag = parser.add_argument_group("diagnostics and host")
    g_diag.add_argument("--vmm-trace", action="append", default=[],
                        metavar="EVENTS",
                        help="Enable VMM tracepoints, comma separated globs, "
                             "repeatable")
    g_diag.add_argument("--vmm-trace-file", default=None, metavar="PATH",
                        help=f"Trace log path for --vmm-trace "
                             f"(default: {DEFAULT_VMM_TRACE_FILE})")
    g_diag.add_argument("--apply-sysconfig", action="store_true",
                        help="Apply sanitizer host settings via sudo, "
                             "reverted on exit")
    g_diag.add_argument("--no-api-socket", action="store_true",
                        help="Disable the per VM control socket")

    g_other = parser.add_argument_group("other")
    g_other.add_argument("-h", "--help", action="help",
                         default=argparse.SUPPRESS,
                         help="Show this help message and exit")
    args = parser.parse_args()
    if args.init is not None:
        if not os.path.isfile(args.init) or not os.access(args.init, os.X_OK):
            sys.exit(f"{args.init} is not an executable init binary")
    print_banner(sys.stderr, init_binary=args.init or INIT_BINARY)

    _install_termination_as_interrupt()

    if args.retries < 0:
        sys.exit("--retries must not be negative")

    if args.apply_sysconfig:
        enable_sysconfigs(args.vmm)

    machine_report = args.format != "text"
    human = sys.stderr if (machine_report and not args.output) else sys.stdout
    profraw_dir = os.path.join(_SCRIPT_DIR, "target", "profraw")
    os.makedirs(profraw_dir, exist_ok=True)
    os.environ["LLVM_PROFILE_FILE"] = os.path.join(
        profraw_dir, "%m_%p.profraw")

    if args.no_packed:
        args.packed = False
        if not args.exclude_dir:
            args.exclude_dir = []
        args.exclude_dir.append("packed")

    backend = detect_vmm(args.vmm)
    if args.transport is not None:
        if not isinstance(backend, Qemu):
            sys.exit("--transport is only supported for QEMU")
        if args.transport == "mmio":
            backend = QemuMicrovm(args.vmm)
    if args.accel is not None and not isinstance(backend, (Qemu, QemuMicrovm)):
        sys.exit("--accel is only supported for QEMU")
    kernel = args.kernel

    vmm_trace = []
    for spec in args.vmm_trace:
        vmm_trace += [e.strip() for e in spec.split(",") if e.strip()]
    if vmm_trace and not backend.supports_trace:
        sys.exit(f"--vmm-trace is not supported by the '{backend.name}' "
                 f"backend")

    vm_opts = {
        "initramfs": args.initramfs,
        "cpus": args.cpus if args.cpus else (
            4 if backend.name in ("ch", "openvmm") else 1),
        "memory": args.memory,
        "blk_queues": args.blk_queues,
        "net_queues": args.net_queues,
        "packed": args.packed,
        "api_enabled": not args.no_api_socket,
        "vmm_trace": vmm_trace,
        "vmm_trace_file": args.vmm_trace_file,
        "io_engine": args.io_engine,
        "direct": args.direct,
    }
    if args.accel is not None:
        vm_opts["accel"] = args.accel

    # Scale the per VM wall clock deadline by the host oversubscription
    # factor. When the user requests more total guest vCPUs than the
    # host has cores, each VM thread gets only a fraction of real CPU
    # time, so a test that normally completes in a few seconds of wall
    # clock can exceed the fixed --timeout and be killed before it can
    # emit a verdict marker. That manifests as random WEDGED results
    # across unrelated tests. Honor an explicit user override by only
    # extending, never shrinking, the requested timeout.
    host_cpus = os.cpu_count() or 1
    per_vm_cpus = max(1, int(vm_opts["cpus"]))
    oversub = max(1.0, (args.jobs * per_vm_cpus) / host_cpus)
    scaled_timeout = int(args.timeout * oversub)
    if scaled_timeout > args.timeout:
        print(f"Note: scaling per test timeout from {args.timeout}s to "
              f"{scaled_timeout}s for oversubscription factor "
              f"{oversub:.1f}x ({args.jobs} jobs x {per_vm_cpus} vCPUs "
              f"on {host_cpus} host CPUs)",
              file=sys.stderr)
        args.timeout = scaled_timeout

    # Propagate the oversubscription factor to the guest so the positive
    # wait helpers extend their window under the same CPU starvation. A
    # device response that arrives late because the guest vCPUs are
    # starved is then still caught instead of being misread as silence
    # and reported as a spurious REJECT. The expect reject path in the
    # guest ignores this and keeps its short silence window.
    vm_opts["wait_scale"] = max(1, int(oversub))

    # Scale the sidecar VMM API call timeout by the same factor so
    # ch-remote and the CH API thread have headroom when CPU starved.
    global _VM_API_TIMEOUT
    _VM_API_TIMEOUT = 5.0 * oversub

    # QEMU refuses aio=native without O_DIRECT, so the aio engine turns
    # direct IO on. Warn when the QEMU backend does so implicitly.
    if _qemu_aio_forces_direct(backend.name, args.io_engine, args.direct):
        print("Note: --io-engine aio requires direct IO on QEMU, "
              "enabling cache.direct=on", file=sys.stderr)

    if args.log_dir:
        os.makedirs(args.log_dir, exist_ok=True)

    if not os.path.isfile(args.initramfs):
        sys.exit(f"{args.initramfs} not found. Run 'make initramfs' first "
                 "or select an archive with --initramfs.")

    if args.tests:
        tests = args.tests
        if args.init is not None:
            known, test_info, _test_meta = get_test_list(args.init)
        else:
            try:
                known, test_info, _test_meta = get_test_list()
            except (FileNotFoundError, OSError, SystemExit):
                known, test_info, _test_meta = [], {}, {}
        if known and not os.environ.get("VV_ALLOW_UNKNOWN_TESTS"):
            tests, unknown = _canonicalize_tests(tests, known)
            if unknown:
                sys.exit(f"Unknown test name(s): {', '.join(unknown)}")
    else:
        tests, test_info, _test_meta = get_test_list(args.init or INIT_BINARY)

    tests = filter_tests(tests, args.include, args.exclude,
                         args.dirs, args.exclude_dir)

    if not tests:
        sys.exit("No tests found.")

    if args.skip_unsupported:
        supported_tests = []
        for name in tests:
            sidecar_path = _find_sidecar(name)
            sidecar = _load_sidecar(sidecar_path) if sidecar_path else None
            supported = getattr(sidecar, "SUPPORTED_BACKENDS", None)
            if supported is None or backend.name in supported:
                supported_tests.append(name)
        if len(supported_tests) != len(tests):
            print("Note: skipping tests whose sidecars do not support "
                  f"{backend.name}.", file=human)
        tests = supported_tests

    mmio_tests = [t for t in tests if t.startswith("M")]
    skipped_mmio = []
    if mmio_tests and not backend.mmio_transport:
        reason = (
            f"MMIO transport is unavailable on the selected VMM {args.vmm}.")
        print(f"Note: skipping MMIO tests. {reason}", file=human)
        if not args.skip_unsupported:
            skipped_mmio = [(name, "SKIP", reason) for name in mmio_tests]
        tests = [t for t in tests if not t.startswith("M")]

    if tests and kernel is None:
        try:
            kernel = fetch_kernel(mmio=backend.name == "qemu-mmio")
        except RuntimeError as error:
            sys.exit(str(error))

    if backend.name == "qemu-mmio" and mmio_tests:
        probe_ok = _mmio_boots(backend, kernel, args.initramfs)
        if not probe_ok:
            reason = f"MMIO guest boot probe failed for {args.vmm}."
            print(f"Note: skipping MMIO tests. {reason}", file=human)
            skipped_mmio = [(name, "SKIP", reason) for name in mmio_tests]
            tests = [t for t in tests if not t.startswith("M")]

    # Probe VMM for available devices when --skip-unsupported is set.
    if args.skip_unsupported:
        def probe(probe_backend, probe_kernel, options):
            with tempfile.TemporaryDirectory(prefix="vv-probe-") as directory:
                probe_opts = dict(options)
                probe_disk = os.path.join(directory, "disk.raw")
                for path, size in (
                        (probe_disk, DISK_SIZE),
                        (os.path.join(directory, "pmem"), PMEM_SIZE),
                        (os.path.join(directory, "scsi"), DISK_SIZE),
                        (os.path.join(directory, "scsi2"), DISK_SIZE)):
                    images.create_raw(path, size)
                probe_opts["pmem_path"] = os.path.join(directory, "pmem")
                probe_opts["scsi_path"] = os.path.join(directory, "scsi")
                probe_opts["scsi_path2"] = os.path.join(directory, "scsi2")
                probe_opts["exit_on_guest_end"] = True
                probe_opts.pop("api_enabled", None)
                probe_opts.pop("api_socket", None)
                skip_ic = [
                    "virtio_blk_init", "virtio_scsi_init",
                    "virtio_net_driver_init",
                    "virtio_console_init", "virtio_balloon_driver_init",
                    "virtio_rng_driver_init", "virtio_mem_driver_init",
                    "virtio_pmem_driver_init", "virtio_iommu_drv_init",
                    "virtio_fs_init", "virtio_vsock_init",
                    "virtio_rtc_drv_init", "virtio_watchdog_driver_init",
                    "virtio_input_driver_init",
                ]
                transport = ("vv.transport=mmio "
                             if probe_backend.name == "qemu-mmio" else "")
                probe_cmdline = (f"console={probe_backend.console_device} "
                                 f"initcall_blacklist={','.join(skip_ic)} "
                                 f"{transport}vv.test=probe")
                probe_cmd = probe_backend.build_cmd(
                    probe_kernel, args.initramfs, probe_disk,
                    probe_cmdline, probe_opts)
                result = process.capture(
                    probe_cmd, timeout=args.timeout, terminal=probe_backend.name == "openvmm")
                cleaned = _clean_result_output(
                    result.stdout.decode(errors="replace"))
                if "PROBE\t" not in cleaned:
                    detail = (result.stdout + result.stderr).decode(
                        errors="replace").strip()
                    raise RuntimeError(
                        f"probe exited with status {result.returncode} "
                        f"without device records\n{detail[-2000:]}")
                return cleaned.encode()

        probes = {}
        filtered = []
        for t in tests:
            meta = _test_meta.get(t)
            if not meta:
                filtered.append(t)
                continue
            probe_backend, probe_kernel = backend, kernel
            options = dict(vm_opts)
            sidecar_path = _find_sidecar(t)
            if sidecar_path:
                sidecar = _load_sidecar(sidecar_path)
                if hasattr(sidecar, "vmm_config"):
                    config = sidecar.vmm_config() or {}
                    options.update(config.get("opts") or {})
            identity = (probe_backend.name, probe_kernel,
                        json.dumps(options, sort_keys=True))
            if identity not in probes:
                try:
                    records = _cached_probe(
                        lambda: _probe_cache_key(
                            probe_backend, probe_kernel,
                            args.initramfs, options),
                        lambda: probe(probe_backend, probe_kernel, options))
                    probes[identity] = {
                        dev_id: (features, nq, has_isr, has_cfg)
                        for dev_id, features, nq, has_isr, has_cfg in records}
                except (subprocess.TimeoutExpired, OSError,
                        ValueError, RuntimeError) as error:
                    print(f"Warning: capability probe unavailable for "
                          f"{probe_backend.name}: {error}. "
                          "Running without capability filtering for this "
                          "configuration.", file=human)
                    probes[identity] = None
            devices = probes[identity]
            if devices is None:
                filtered.append(t)
                continue
            dev_id = meta["device_id"]
            if dev_id and dev_id not in devices:
                continue
            features, nq, has_isr, has_cfg = devices.get(dev_id, (0, 0, 1, 1))
            required = meta["required_features"]
            if (required & features) != required:
                continue
            if nq < meta["min_queues"]:
                continue
            if (meta["flags"] & 8) and not has_isr:  # NEEDS_ISR
                continue
            if (meta["flags"] & 16) and not has_cfg:  # NEEDS_CFG
                continue
            filtered.append(t)
        skipped = len(tests) - len(filtered)
        tests = filtered
        if skipped:
            print(f"Note: {skipped} test(s) skipped "
                  f"(device or feature not available)", file=human)

    total_tests = len(tests) + len(skipped_mmio)
    print(f"Running {total_tests} test(s) against {args.vmm} ({backend.name})\n",
          file=human)

    saved_term = _save_term()
    results = []
    # Names of tests that failed then recovered on a retry. They are
    # reported under their real verdict and listed here as retried, a
    # transient failure rather than a claimed property of the test.
    retried = []
    retried_attempts = {}
    is_tty = sys.stdout.isatty()

    def _progress(done, total, current=""):
        """Draw a progress line at the bottom (overwritten each time)."""
        if not is_tty or args.console:
            return
        cols = reporting.terminal_width(sys.stdout)
        counts = Counter(status for _, status, _ in results)
        counts["RETRIED"] = len(retried)
        sys.stdout.write(reporting.format_progress(counts, done, total, current, cols))
        sys.stdout.flush()

    def _clear_progress():
        if is_tty and not args.console:
            sys.stdout.write("\r\033[K")
            sys.stdout.flush()

    _RESET = reporting.RESET
    _DIM = reporting.DIM

    # Longest description among the tests in this run. The result table
    # pads the description column to this width so it stays aligned
    # without stretching to the full terminal width.
    _desc_col = max(
        (len(test_info.get(t, ("", "", ""))[0])
         for t in tests + [name for name, _, _ in skipped_mmio]),
        default=10)

    def _format_test_line(name, status):
        """Format a full test result line fitting in terminal width."""
        cols = reporting.terminal_width(sys.stdout)
        return reporting.format_test_line(
            name, status, test_info.get(name, ("", "", "")), _desc_col, cols, is_tty)

    def _report(name, status, output):
        """Print one test result immediately."""
        results.append((name, status, output))
        recovered = _recovered_on_retry(status, output or "")
        lines = output.splitlines() if output else []
        attempts = (lines[0] if lines and lines[0].startswith(_ATTEMPTS_PREFIX)
                    else "")
        if recovered:
            retried.append(name)
            retried_attempts[name] = attempts[len(_ATTEMPTS_PREFIX):]
        if not args.console:
            _clear_progress()
            print(_format_test_line(name, status), file=human)

        def _emit(line):
            print(reporting.format_detail(line, is_tty), file=human)

        if args.log_dir and output:
            log_path = os.path.join(args.log_dir, f"{name}.log")
            with open(log_path, "w") as f:
                f.write(output)
            if status == "FAIL":
                print(f"       log: {log_path}", file=human)
        # A retried result is shown under its real verdict and listed at
        # the end, so it renders like any normal result with no detail.
        if recovered or args.console:
            return
        if attempts:
            _emit(attempts)
        if status in _JUNIT_FAILURE and output:
            tier, why = _security_triage(output)
            sev = f"       security: {tier} -- {why}"
            print(f"{_DIM}{sev}{_RESET}" if is_tty else sev, file=human)
        if output and (status == "FAIL" or attempts):
            for line in lines[-10:]:
                if line and line != attempts:
                    _emit(line)

    for name, status, output in skipped_mmio:
        _report(name, status, output)

    pool = None
    stopped = threading.Event()
    # Reorder tests according to --test-order.
    if args.order == "fast":
        _sidecar_re = re.compile(r'[A-Z]+0[23]\d\d', re.IGNORECASE)

        def _is_slow(t):
            if _sidecar_re.match(t):
                return True
            if t.upper().startswith("F0"):
                return True
            if t.upper().startswith("P0"):
                return True
            if t.upper().startswith(("N0", "I0")):
                return True
            meta = _test_meta.get(t)
            if meta and meta["flags"] & 4:  # TEST_FLAG_XFAIL
                return True
            return False

        slow = [t for t in tests if _is_slow(t)]
        fast = [t for t in tests if not _is_slow(t)]
        random.shuffle(slow)
        random.shuffle(fast)
        if slow:
            step = max(1, len(fast) // (len(slow) + 1))
            merged = []
            si = 0
            for i, t in enumerate(fast):
                if si < len(slow) and i > 0 and i % step == 0:
                    merged.append(slow[si])
                    si += 1
                merged.append(t)
            merged.extend(slow[si:])
            tests = merged
        else:
            tests = fast
    elif args.order == "alpha":
        tests.sort()
    elif args.order == "random":
        random.shuffle(tests)
    elif args.order == "reverse":
        tests.sort(reverse=True)

    # Group tests into batches. Sidecar tests are never
    # batched because they need per test host actors and VM config.
    batch_size = args.batch
    if batch_size > 1:
        _sc_re = re.compile(r'[A-Z]+0[23]\d\d', re.IGNORECASE)

        def _must_solo(t):
            """Tests that must not be batched."""
            if _sc_re.match(t) or _find_sidecar(t):
                return True
            meta = _test_meta.get(t)
            if meta and meta["flags"] & 4:  # TEST_FLAG_XFAIL
                return True
            return False

        solo = [t for t in tests if _must_solo(t)]
        batchable = [t for t in tests if not _must_solo(t)]
        work_items = []
        for i in range(0, len(batchable), batch_size):
            chunk = batchable[i:i + batch_size]
            work_items.append(chunk)
        for t in solo:
            work_items.append([t])
    else:
        work_items = [[t] for t in tests]

    def run_item(item):
        return run_test(
            backend, kernel, item if len(item) > 1 else item[0],
            args.timeout, args.verbose, args.console, args.disk_type, vm_opts, stopped)

    def finish_results(batch_results):
        if batch_size > 1:
            batch_results = _retry_wedged(
                batch_results, backend, kernel, args.timeout, args.verbose,
                args.disk_type, vm_opts)
        if args.retries > 0:
            batch_results = _apply_retries(
                batch_results, args.retries, backend, kernel, args.timeout,
                args.verbose, args.disk_type, vm_opts, batch_size > 1)
        return batch_results

    try:
        if args.jobs > 1 and not args.console:
            pool = ThreadPoolExecutor(max_workers=args.jobs)
            work = execution.CompletionQueue(
                pool, run_item, work_items, limit=max(args.jobs, len(work_items)), stop=stopped)
            done = len(skipped_mmio)
            _progress(done, total_tests, "starting...")
            try:
                for _, future in work:
                    batch_results = finish_results(future.result())
                    for name, status, output in batch_results:
                        done += 1
                        with _print_lock:
                            _report(name, status, output)
                            _progress(done, total_tests)
            except KeyboardInterrupt:
                stopped.set()
                _clear_progress()
                print("\n\nInterrupted. Killing live VMMs...")
                # Drop unstarted work and SIGKILL everything still
                # running so worker threads can wake up and finish
                # their finally blocks promptly.
                try:
                    pool.shutdown(wait=False, cancel_futures=True)
                except TypeError:
                    # Python < 3.9 fallback (no cancel_futures kw)
                    pool.shutdown(wait=False)
                _kill_all_live_procs()
                # Now drain whatever finishes naturally. Don't propagate
                # exceptions from cancelled futures.
                for fut in work.pending:
                    if fut.cancelled():
                        continue
                    try:
                        fut.result(timeout=10)
                    except (Exception, KeyboardInterrupt):
                        pass
                raise
            finally:
                if pool is not None:
                    try:
                        pool.shutdown(wait=True)
                    except Exception:
                        pass
        else:
            try:
                done = len(skipped_mmio)
                for item in work_items:
                    _progress(done, total_tests, item[0])
                    batch_results = finish_results(run_item(item))
                    for name, status, output in batch_results:
                        done += 1
                        _report(name, status, output)
                        _progress(done, total_tests)
            except KeyboardInterrupt:
                stopped.set()
                _clear_progress()
                print("\n\nInterrupted. Killing live VMMs...")
                _kill_all_live_procs()
                raise
    except KeyboardInterrupt:
        _clear_progress()
        print("\nInterrupted.")
    finally:
        _clear_progress()
        _restore_term(saved_term)
        # Final safety net: anything still tracked gets SIGKILL'd.
        _kill_all_live_procs()

    counts = Counter(status for _, status, _ in results)
    counts["RETRIED"] = len(retried)
    total = len(results)

    print(reporting.format_summary(counts, total, is_tty), file=human)

    failed = [name for name, s, _ in results if s == "FAIL"]
    if failed:
        print(reporting.format_failure_list(
            [(name, test_info.get(name, ("", "", ""))[0]) for name in failed],
            color=is_tty), file=human)

    wedged = [name for name, s, _ in results if s == "WEDGED"]
    if wedged:
        hdr = "\033[33mWEDGED (device needs reset):\033[0m" if is_tty \
              else "WEDGED (device needs reset):"
        print(f"\n{hdr}", file=human)
        for name in wedged:
            desc, ver, sec = test_info.get(name, ("", "", ""))
            print(f"  {name:<7} {desc}", file=human)

    xfail = [name for name, s, _ in results if s == "XFAIL"]
    if xfail:
        hdr = "\033[34mXFAIL (expected failure, known bug or limitation):\033[0m" \
              if is_tty else "XFAIL (expected failure, known bug or limitation):"
        print(f"\n{hdr}", file=human)
        for name in xfail:
            desc, ver, sec = test_info.get(name, ("", "", ""))
            print(f"  {name:<7} {desc}", file=human)

    xpass = [name for name, s, _ in results if s == "XPASS"]
    if xpass:
        hdr = "\033[31mXPASS (xfail marker is stale, remove it):\033[0m" \
              if is_tty else "XPASS (xfail marker is stale, remove it):"
        print(f"\n{hdr}", file=human)
        for name in xpass:
            desc, ver, sec = test_info.get(name, ("", "", ""))
            print(f"  {name:<7} {desc}", file=human)

    if retried:
        hdr = ("\033[93mRETRIED (failed first, recovered on rerun):\033[0m"
               if is_tty else "RETRIED (failed first, recovered on rerun):")
        print(f"\n{hdr}", file=human)
        for name in retried:
            desc, ver, sec = test_info.get(name, ("", "", ""))
            seq = retried_attempts.get(name, "")
            tail = f"   [{seq}]" if seq else ""
            print(f"  {name:<7} {desc}{tail}", file=human)

    if machine_report:
        doc = build_results_doc(results, test_info,
                                backend_name=backend.name, vmm=args.vmm)
        report = (format_results_json(doc) if args.format == "json"
                  else format_results_junit(doc))
        if args.output:
            with open(args.output, "w") as f:
                f.write(report + "\n")
            print(f"Wrote {args.format} report to {args.output}", file=human)
        else:
            sys.stdout.write(report + "\n")

    sys.exit(0 if not failed and not wedged and not xpass else 1)


if __name__ == "__main__":
    # Handle lightweight utility subcommands before full arg parsing.
    if len(sys.argv) >= 2 and sys.argv[1] == "fetch-kernel":
        try:
            print(fetch_kernel())
        except RuntimeError as error:
            sys.exit(str(error))
        sys.exit(0)
    main()