distinct_agent/runtime.py

"""Getting llama.cpp onto a volunteer's machine, with the same rules as weights.

WHY THIS EXISTS.

The setup used to end with a wall. A volunteer cloned the repository, ran the
install, ran the worker, and was told to go to another project's releases page,
pick the right build for their machine out of a dozen, unzip it somewhere, and
either edit their PATH or pass a path back on the command line. Their reaction
was the correct one: *why do I have to provide this, it should be baked into
the repo*.

It cannot literally be in the repository. A llama.cpp release is around a
hundred megabytes per platform, there is one per operating system and per
accelerator, and it is a separate project with its own release cadence. Putting
those in a Space repository would make every clone enormous and every llama.cpp
release a commit here.

``llama-cpp-python`` was the other obvious answer and is worse: with no
matching wheel it compiles from source, so on a Windows machine without a
build toolchain the failure moves from "run the worker" to ``pip install``
itself, which is a wall one step earlier and harder to read.

So this does what the project already does for model weights: fetches a
**pinned** release over TLS and refuses to install it unless the bytes match a
**recorded SHA-256**. One command, nothing to go and get, and the thing that
runs the models is verified rather than trusted.

WHAT IS DELIBERATELY NOT DONE.

There is no trust-on-first-use path. Reading a digest out of the same response
that carried the download proves only that the bytes arrived intact, which TLS
already said. This project withholds a *model* for exactly that reason --
``llama-2-7b-chat`` is refused with "no upstream repository, pinned revision or
SHA-256 digest recorded, so this worker cannot obtain the exact published file
or confirm what it ran" -- and the program that executes those models cannot be
held to a lower standard than the models.

A platform with no recorded digest therefore gets no automatic download. It is
told so plainly, and keeps both manual routes: ``--llama-server`` for a build
the operator already has, and ``--demo-runner`` to run the library's tools and
skills with no model at all.
"""

from __future__ import annotations

import hashlib
import json
import platform
import re
import shutil
import stat
import subprocess
import sys
import tempfile
import urllib.error
import urllib.parse
import urllib.request
import zipfile
from dataclasses import dataclass
from pathlib import Path
from typing import Any, Callable, Mapping, Optional

#: Where the pins live. A data file rather than a literal so that recording a
#: new platform, or moving to a newer llama.cpp, is a data change somebody can
#: review as a diff of digests.
PIN_FILE = Path(__file__).resolve().parent.parent / "llama_runtime.json"

#: Releases are served from github.com and redirected to their object store.
#: Anywhere else and the download is refused, for the same reason the weights
#: fetcher refuses a redirect off Hugging Face: a download that has been bounced
#: somewhere unexpected is no longer the download that was asked for.
RUNTIME_REDIRECT_HOSTS = ("github.com", "githubusercontent.com", "github-releases.githubusercontent.com")

_CHUNK = 1024 * 1024

#: A CPU llama.cpp Windows archive is about 20 MB. A CUDA one is an order of
#: magnitude larger, and the CUDA runtime beside it larger still. These are
#: bomb guards, not size expectations: an archive claiming to be larger than
#: this is refused before a byte of it is written. Set with room above the
#: largest real asset rather than snugly, because a guard that trips on a
#: legitimate release is a guard that gets removed.
MAX_ARCHIVE_BYTES = 900 * 1024 * 1024

#: The unpacked tree is checked too, so an archive that is small on the wire
#: and enormous on disk is refused rather than unpacked.
MAX_UNPACKED_BYTES = 2 * 1024 * 1024 * 1024


class RuntimeUnavailable(RuntimeError):
    """The runtime could not be obtained, with a sentence saying why."""


class RuntimeVerificationError(RuntimeUnavailable):
    """The bytes arrived but did not match the recorded digest."""


def permitted_source(url: str) -> bool:
    """Whether this worker is willing to fetch a runtime from ``url`` at all.

    THE FIRST URL NEEDS THE SAME CHECK AS EVERY REDIRECT.

    The redirect handler below refuses to follow a hop off GitHub, which is the
    obvious half of the rule; the half that a penetration test found missing was
    the URL the download *starts* at. That one comes out of a JSON file, so
    anybody who could alter ``llama_runtime.json`` -- a bad commit, a merged
    pull request nobody read closely, a tampered checkout -- could point every
    volunteer's worker at a host of their choosing and have it make the request.

    The recorded digest still means no substituted *file* is ever installed, so
    this was not a path to running someone else's code. It was still an
    arbitrary outbound request made by every worker that updated, which is a
    thing worth having and not a thing worth leaving lying around. Now the
    allowlist is applied where the URL is read, so a pin pointing anywhere else
    is not a download that fails: it is a pin that does not count as recorded.
    """

    parsed = urllib.parse.urlsplit(url)
    if parsed.scheme != "https":
        return False
    host = (parsed.hostname or "").casefold().rstrip(".")
    if not host:
        return False
    return any(
        host == allowed or host.endswith("." + allowed)
        for allowed in RUNTIME_REDIRECT_HOSTS
    )


class _RuntimeRedirects(urllib.request.HTTPRedirectHandler):
    max_redirections = 10

    _permitted = staticmethod(permitted_source)

    def redirect_request(self, req, fp, code, msg, headers, newurl):  # noqa: D102
        if not self._permitted(newurl):
            raise urllib.error.HTTPError(
                newurl,
                code,
                "the llama.cpp download was redirected off GitHub and was refused",
                headers,
                fp,
            )
        return super().redirect_request(req, fp, code, msg, headers, newurl)


def _opener() -> Callable[..., Any]:
    return urllib.request.build_opener(_RuntimeRedirects).open


@dataclass(frozen=True)
class RuntimeArchive:
    """One file that has to arrive, intact, for a build to run."""

    asset: str
    url: str
    sha256: str
    bytes: int = 0

    @property
    def recorded(self) -> bool:
        return bool(self.url and len(self.sha256) == 64 and permitted_source(self.url))


@dataclass(frozen=True)
class RuntimePin:
    """One build recorded for one kind of machine.

    ``archives`` is a list rather than a single file because an accelerated
    Windows build is not self-contained: llama.cpp ships the CUDA binaries in
    one archive and the CUDA runtime libraries they link against in another,
    and a worker that fetched only the first gets ``cudart64_12.dll was not
    found`` at start-up instead of a model. Both are pinned, both are
    verified, and both unpack into the same directory.
    """

    key: str
    tag: str
    archives: tuple[RuntimeArchive, ...] = ()
    #: What the machine must have for this build to run at all. ``cpu`` runs
    #: anywhere and is the floor every platform falls back to.
    accelerator: str = "cpu"
    #: The lowest CUDA version the driver must advertise, in NVML's integer
    #: form: 12040 is CUDA 12.4. Zero when there is no such requirement.
    requires_cuda: int = 0

    @property
    def recorded(self) -> bool:
        return bool(self.archives) and all(item.recorded for item in self.archives)

    @property
    def asset(self) -> str:
        return self.archives[0].asset if self.archives else ""

    @property
    def url(self) -> str:
        return self.archives[0].url if self.archives else ""

    @property
    def sha256(self) -> str:
        return self.archives[0].sha256 if self.archives else ""

    @property
    def bytes(self) -> int:
        return sum(item.bytes for item in self.archives)


def platform_key(system: str = "", machine: str = "") -> str:
    """The base key this machine looks itself up under: OS and architecture.

    Accelerated builds are recorded under this key with a suffix --
    ``windows-x64-cuda`` beside ``windows-x64`` -- and chosen by
    :func:`preferred_keys`, which checks the machine can run them. This
    function answers only "what kind of computer is this", and every platform
    falls back to the plain key, which is the build that runs anywhere.
    """

    system = (system or platform.system()).casefold()
    machine = (machine or platform.machine()).casefold()
    if machine in {"amd64", "x86_64", "x64"}:
        arch = "x64"
    elif machine in {"arm64", "aarch64"}:
        arch = "arm64"
    else:
        arch = machine or "unknown"
    if system.startswith("win"):
        return f"windows-{arch}"
    if system == "darwin":
        return f"macos-{arch}"
    if system == "linux":
        return f"linux-{arch}"
    return f"{system or 'unknown'}-{arch}"


def cuda_capability() -> int:
    """The highest CUDA version this machine's driver can run, or zero.

    In NVML's integer form -- 12040 for CUDA 12.4 -- because that is the shape
    a pin's ``requires_cuda`` is compared against. Zero means *there is no
    usable NVIDIA GPU here*, and covers no card, no driver, and a driver too
    broken to answer, all of which have the same consequence: do not install a
    CUDA build.

    THIS IS THE DRIVER'S CEILING, NOT THE CARD'S SPEED. A CUDA 12.4 build on a
    driver that tops out at 12.0 does not run slowly, it fails to start, and
    the volunteer sees a missing-symbol error rather than a worker. That is
    why this is a gate rather than a preference.
    """

    return _cuda_from_nvml() or _cuda_from_smi()


def accelerator_memory() -> tuple[int, int]:
    """``(free, total)`` bytes on the first NVIDIA board, or ``(0, 0)``.

    Free rather than total, and it matters: on a laptop the desktop compositor
    is already holding a third of a 4 GB card, and an offload sized on the
    total is an offload that does not fit. What happens then is worse than
    failing -- see :func:`distinct_agent.server_runner.plan_offload`.
    """

    try:
        import pynvml  # noqa: PLC0415
    except Exception:  # noqa: BLE001
        return (0, 0)
    try:
        pynvml.nvmlInit()
    except Exception:  # noqa: BLE001
        return (0, 0)
    try:
        if int(pynvml.nvmlDeviceGetCount()) < 1:
            return (0, 0)
        handle = pynvml.nvmlDeviceGetHandleByIndex(0)
        info = pynvml.nvmlDeviceGetMemoryInfo(handle)
        return (int(info.free), int(info.total))
    except Exception:  # noqa: BLE001
        return (0, 0)
    finally:
        try:
            pynvml.nvmlShutdown()
        except Exception:  # noqa: BLE001
            pass


def accelerator_name() -> str:
    """The first NVIDIA board's product name, or an empty string."""

    try:
        import pynvml  # noqa: PLC0415
    except Exception:  # noqa: BLE001
        return ""
    try:
        pynvml.nvmlInit()
    except Exception:  # noqa: BLE001
        return ""
    try:
        if int(pynvml.nvmlDeviceGetCount()) < 1:
            return ""
        name = pynvml.nvmlDeviceGetName(pynvml.nvmlDeviceGetHandleByIndex(0))
        return name.decode("utf-8", "replace") if isinstance(name, bytes) else str(name)
    except Exception:  # noqa: BLE001
        return ""
    finally:
        try:
            pynvml.nvmlShutdown()
        except Exception:  # noqa: BLE001
            pass


def _cuda_from_nvml() -> int:
    try:
        import pynvml  # noqa: PLC0415 - optional, and probing is the point
    except Exception:  # noqa: BLE001
        return 0
    try:
        pynvml.nvmlInit()
    except Exception:  # noqa: BLE001
        return 0
    try:
        if int(pynvml.nvmlDeviceGetCount()) < 1:
            return 0
        for name in ("nvmlSystemGetCudaDriverVersion_v2", "nvmlSystemGetCudaDriverVersion"):
            probe = getattr(pynvml, name, None)
            if probe is None:
                continue
            try:
                return int(probe())
            except Exception:  # noqa: BLE001
                continue
        return 0
    except Exception:  # noqa: BLE001
        return 0
    finally:
        try:
            pynvml.nvmlShutdown()
        except Exception:  # noqa: BLE001
            pass


#: ``nvidia-smi`` prints "CUDA Version: 12.4" in its header. Parsed only as a
#: fallback, for a machine with the driver but without the Python bindings.
_SMI_CUDA = re.compile(r"CUDA Version:\s*(\d+)\.(\d+)")


def _cuda_from_smi() -> int:
    executable = shutil.which("nvidia-smi")
    if not executable:
        return 0
    try:
        finished = subprocess.run(  # noqa: S603 - resolved from PATH, fixed argv
            [executable],
            capture_output=True,
            text=True,
            timeout=20,
            check=False,
            creationflags=getattr(subprocess, "CREATE_NO_WINDOW", 0),
        )
    except (OSError, subprocess.SubprocessError):
        return 0
    found = _SMI_CUDA.search(f"{finished.stdout}\n{finished.stderr}")
    if not found:
        return 0
    return int(found.group(1)) * 1000 + int(found.group(2)) * 10


def preferred_keys(key: str = "", path: Optional[Path] = None) -> list[str]:
    """Which recorded builds suit this machine, best first, CPU last.

    The list always ends with the plain platform key, so there is always
    something to fall back to and the fallback is always the build that asks
    least of the machine.
    """

    base = key or platform_key()
    pins = load_pins(path)
    capability: Optional[int] = None
    accelerated: list[RuntimePin] = []
    for candidate, pin in pins.items():
        if candidate == base or not candidate.startswith(f"{base}-"):
            continue
        if not pin.recorded:
            continue
        if pin.accelerator == "cuda":
            if capability is None:
                capability = cuda_capability()
            if capability < max(pin.requires_cuda, 1):
                continue
        else:
            # An accelerator this build of the selector does not know how to
            # check for is not offered. Silently installing a HIP build on an
            # unexamined machine is the failure this whole function exists to
            # prevent.
            continue
        accelerated.append(pin)
    accelerated.sort(key=lambda pin: pin.requires_cuda, reverse=True)
    return [pin.key for pin in accelerated] + [base]


def starts(executable: str, timeout: float = 60.0) -> bool:
    """Whether this llama-server can load its libraries and run at all.

    A CUDA build on a machine missing the CUDA runtime does not fail at
    inference, it fails at process start: the loader cannot resolve
    ``cudart64_12.dll`` and the image never runs, with no output and an exit
    code most people have never seen. ``--version`` loads exactly the same
    libraries as a real start and exits immediately, which separates "this
    build cannot run here" from "this build is slow here" in about a second,
    before a volunteer has waited out a model load to find out.

    Lenient about the exit code on purpose: a build that answered at all
    resolved its libraries, and that is the question being asked.
    """

    try:
        finished = subprocess.run(  # noqa: S603 - our own installed binary
            [executable, "--version"],
            capture_output=True,
            timeout=timeout,
            check=False,
            creationflags=getattr(subprocess, "CREATE_NO_WINDOW", 0),
        )
    except (OSError, subprocess.SubprocessError):
        return False
    return finished.returncode == 0 or bool(finished.stdout or finished.stderr)


def load_pins(path: Optional[Path] = None) -> dict[str, RuntimePin]:
    """Read the recorded builds, or an empty mapping if none are recorded."""

    source = path or PIN_FILE
    try:
        raw = json.loads(source.read_text(encoding="utf-8"))
    except (OSError, ValueError):
        return {}
    builds = raw.get("builds") if isinstance(raw, Mapping) else None
    if not isinstance(builds, Mapping):
        return {}
    pins: dict[str, RuntimePin] = {}
    for key, entry in builds.items():
        if not isinstance(entry, Mapping):
            continue
        pins[str(key)] = RuntimePin(
            key=str(key),
            tag=str(entry.get("tag", "")),
            archives=_archives(entry),
            accelerator=str(entry.get("accelerator", "cpu")).casefold() or "cpu",
            requires_cuda=int(entry.get("requires_cuda", 0) or 0),
        )
    return pins


def _archives(entry: Mapping[str, Any]) -> tuple[RuntimeArchive, ...]:
    """The files a pin needs, accepting the one-file shape it started as."""

    listed = entry.get("archives")
    if not isinstance(listed, (list, tuple)):
        listed = [entry] if entry.get("url") else []
    found: list[RuntimeArchive] = []
    for item in listed:
        if not isinstance(item, Mapping):
            continue
        found.append(
            RuntimeArchive(
                asset=str(item.get("asset", "")),
                url=str(item.get("url", "")),
                sha256=str(item.get("sha256", "")).casefold(),
                bytes=int(item.get("bytes", 0) or 0),
            )
        )
    return tuple(found)


def pin_for(key: str = "", path: Optional[Path] = None) -> Optional[RuntimePin]:
    pin = load_pins(path).get(key or platform_key())
    return pin if pin is not None and pin.recorded else None


def runtime_directory(root: Optional[Path] = None, key: str = "") -> Path:
    """Where an installed llama.cpp lives: ``runtime/<key>/``.

    One directory per build, so a machine that has a CUDA build and a CPU
    build to fall back to has both, rather than whichever was fetched last.
    """

    base = (root or PIN_FILE.parent) / "runtime"
    return base / key if key else base


def installed_server(root: Optional[Path] = None, key: str = "") -> str:
    """The path to an already-installed llama-server for ``key``, or empty."""

    directory = runtime_directory(root, key)
    if not directory.is_dir():
        return ""
    for name in ("llama-server.exe", "llama-server"):
        candidate = directory / name
        if candidate.is_file():
            return str(candidate)
    # Some archives nest everything one level down under build/bin.
    for candidate in sorted(directory.glob("**/llama-server*")):
        if candidate.is_file() and candidate.suffix in {"", ".exe"}:
            return str(candidate)
    return ""


def best_installed(
    root: Optional[Path] = None, pins_path: Optional[Path] = None
) -> str:
    """The best already-installed build for this machine, or empty.

    Same order as :func:`preferred_keys`, so a worker that has both a CUDA
    build and a CPU build picks the one it would have installed today rather
    than the one it happened to install first.
    """

    for candidate in preferred_keys(path=pins_path):
        found = installed_server(root, candidate)
        if found:
            return found
    return ""


def _migrate_flat_install(root: Optional[Path] = None) -> None:
    """Move a pre-keyed install into the directory it would have today.

    Builds used to unpack straight into ``runtime/``, because there was only
    ever one. Left there it would be invisible to the keyed lookup and
    re-downloaded, and worse, it would sit in the parent of every keyed
    directory where a glob could still find it and hand back a CPU build to a
    machine that had just installed a CUDA one. Moving it once is cheaper and
    less surprising than either.
    """

    base = runtime_directory(root)
    stray = [
        name
        for name in ("llama-server.exe", "llama-server")
        if (base / name).is_file()
    ]
    if not stray:
        return
    destination = base / platform_key()
    destination.mkdir(parents=True, exist_ok=True)
    for item in list(base.iterdir()):
        if item.is_dir():
            continue
        try:
            shutil.move(str(item), str(destination / item.name))
        except OSError:
            # A file we cannot move is left where it is; the keyed install
            # will simply be fetched. Losing a download is not worth an
            # exception on somebody's start-up.
            continue


def _safe_members(archive: zipfile.ZipFile) -> list[zipfile.ZipInfo]:
    """Every member that may be written, refusing traversal and bombs.

    ``ZipFile.extractall`` sanitises paths in modern Python, but it has no
    opinion about total size, and this archive comes from outside the project.
    Both are checked here so the refusal names the member rather than filling a
    volunteer's disk and failing later.
    """

    total = 0
    members: list[zipfile.ZipInfo] = []
    for info in archive.infolist():
        name = info.filename
        if name.endswith("/"):
            continue
        pure = Path(name)
        if pure.is_absolute() or ".." in pure.parts or (len(name) > 1 and name[1] == ":"):
            raise RuntimeUnavailable(
                f"the llama.cpp archive contains an unsafe path ({name!r}) and was not unpacked"
            )
        total += info.file_size
        if info.file_size > MAX_UNPACKED_BYTES or total > MAX_UNPACKED_BYTES:
            raise RuntimeUnavailable(
                "the llama.cpp archive unpacks to more than "
                f"{MAX_UNPACKED_BYTES // (1024 * 1024)} MB and was refused"
            )
        members.append(info)
    return members


def ensure(
    *,
    root: Optional[Path] = None,
    key: str = "",
    progress: Optional[Callable[[float, str], None]] = None,
    opener: Optional[Callable[..., Any]] = None,
    pins_path: Optional[Path] = None,
) -> str:
    """Install the pinned llama.cpp for ``key`` and return its path.

    Returns the path to ``llama-server``. Raises :class:`RuntimeUnavailable`
    with a sentence a volunteer can act on when there is nothing recorded for
    this platform, and :class:`RuntimeVerificationError` when the bytes do not
    match -- in which case nothing is installed.
    """

    resolved = key or platform_key()
    _migrate_flat_install(root)
    existing = installed_server(root, resolved)
    if existing:
        return existing

    raw = load_pins(pins_path).get(resolved)
    for archive in raw.archives if raw is not None else ():
        if archive.url and not permitted_source(archive.url):
            raise RuntimeUnavailable(
                f"the recorded llama.cpp source for {resolved} is {archive.url!r}, which "
                "is not GitHub. Nothing was fetched. A pin that points anywhere else has "
                "been tampered with; report it rather than working around it."
            )
    pin = pin_for(resolved, pins_path)
    if pin is None:
        raise RuntimeUnavailable(
            f"no llama.cpp build is recorded for {resolved}, so this worker will not "
            "download one: an unverified build of the program that runs the models is "
            "exactly what this project refuses to install. Point at a build you already "
            "have with --llama-server, or start without a model using --demo-runner."
        )

    target = runtime_directory(root, resolved)
    fetch = opener or _opener()
    total = len(pin.archives)
    moved = 0

    with tempfile.TemporaryDirectory(prefix="distinct-runtime-") as scratch:
        # EVERY ARCHIVE VERIFIES BEFORE ANY OF THEM IS INSTALLED.
        #
        # A CUDA pin is two downloads, and the failure that matters is the
        # second one: binaries unpacked, CUDA runtime missing, a directory
        # that looks installed and a worker that will not start. Staging the
        # lot and moving once means a pin either arrives whole or leaves no
        # trace, which is also what makes a failed verification testable.
        staged = Path(scratch) / "staged"
        staged.mkdir()
        for index, archive in enumerate(pin.archives):
            # Each archive gets its own slice of the progress bar, so a CUDA
            # build fetching binaries and then the CUDA runtime beside them
            # reads as one download rather than as one that restarts.
            _fetch_archive(
                archive,
                resolved,
                Path(scratch) / f"archive-{index}",
                staged,
                fetch,
                progress,
                index / total,
                (index + 1) / total,
            )
        target.mkdir(parents=True, exist_ok=True)
        for source in sorted(staged.glob("**/*")):
            if not source.is_file():
                continue
            destination = target / source.name
            shutil.copy2(source, destination)
            if not sys.platform.startswith("win"):
                destination.chmod(destination.stat().st_mode | stat.S_IXUSR)
            moved += 1

    found = installed_server(root, resolved)
    if not found:
        raise RuntimeUnavailable(
            f"the llama.cpp archive for {resolved} unpacked {moved} file(s) but none of "
            "them was llama-server. The recorded asset is probably the wrong one; "
            "report it rather than working around it."
        )
    if progress is not None:
        progress(1.0, f"llama.cpp {pin.tag} installed in {target}")
    return found


def _fetch_archive(
    archive: RuntimeArchive,
    resolved: str,
    scratch: Path,
    staged: Path,
    fetch: Callable[..., Any],
    progress: Optional[Callable[[float, str], None]],
    floor: float,
    ceiling: float,
) -> None:
    """Fetch one archive, verify its digest, and unpack it into ``staged``."""

    scratch.mkdir(parents=True, exist_ok=True)
    archive_path = scratch / "llama.zip"
    digest = hashlib.sha256()
    written = 0
    label = archive.asset or resolved

    if progress is not None:
        progress(floor, f"Fetching {label} for {resolved}")

    try:
        request = urllib.request.Request(
            archive.url,
            headers={"User-Agent": "distinct-agent", "Accept": "application/octet-stream"},
        )
        with fetch(request, timeout=60) as response:
            try:
                declared = int(response.headers.get("Content-Length") or 0)
            except (TypeError, ValueError):
                declared = 0
            if declared and declared > MAX_ARCHIVE_BYTES:
                raise RuntimeUnavailable(
                    "the llama.cpp download declares more than "
                    f"{MAX_ARCHIVE_BYTES // (1024 * 1024)} MB and was refused"
                )
            with archive_path.open("wb") as handle:
                while True:
                    chunk = response.read(_CHUNK)
                    if not chunk:
                        break
                    handle.write(chunk)
                    digest.update(chunk)
                    written += len(chunk)
                    if written > MAX_ARCHIVE_BYTES:
                        raise RuntimeUnavailable(
                            "the llama.cpp download exceeded "
                            f"{MAX_ARCHIVE_BYTES // (1024 * 1024)} MB and was stopped"
                        )
                    if progress is not None and declared:
                        progress(
                            min(ceiling - 0.001, floor + (ceiling - floor) * written / declared),
                            f"Fetching {label}",
                        )
    except RuntimeUnavailable:
        raise
    except (OSError, urllib.error.URLError, ValueError) as exc:
        raise RuntimeUnavailable(
            f"could not fetch {label} for {resolved}: "
            f"{type(exc).__name__}: {exc}. Point at a build you already have with "
            "--llama-server, or start without a model using --demo-runner."
        ) from exc

    actual = digest.hexdigest()
    if actual != archive.sha256:
        raise RuntimeVerificationError(
            f"{label} for {resolved} does not match the recorded SHA-256: expected "
            f"{archive.sha256}, computed {actual}. Nothing was installed. This is what "
            "a substituted or corrupted build looks like; report it rather than "
            "working around it."
        )

    if progress is not None:
        progress(ceiling - 0.001, f"{label} verified, unpacking")

    # The layout differs between releases, so the binaries are found rather
    # than assumed, and land flat in the staging directory. Flat is what puts
    # the CUDA runtime libraries beside the executable that needs them, which
    # on Windows is the only place the loader will look without an
    # environment variable.
    unpacked = scratch / "unpacked"
    unpacked.mkdir()
    try:
        with zipfile.ZipFile(archive_path) as opened:
            members = _safe_members(opened)
            opened.extractall(unpacked, members=members)
    except zipfile.BadZipFile as exc:
        raise RuntimeUnavailable(
            f"{label} for {resolved} is not a readable archive: {exc}"
        ) from exc
    for source in sorted(unpacked.glob("**/*")):
        if source.is_file():
            shutil.copy2(source, staged / source.name)
    shutil.rmtree(unpacked, ignore_errors=True)
    archive_path.unlink(missing_ok=True)


def ensure_best(
    *,
    root: Optional[Path] = None,
    key: str = "",
    progress: Optional[Callable[[float, str], None]] = None,
    opener: Optional[Callable[..., Any]] = None,
    pins_path: Optional[Path] = None,
    verify: Optional[Callable[[str], bool]] = None,
) -> str:
    """Install the fastest recorded build this machine will actually run.

    THE DIFFERENCE THIS MAKES IS NOT SMALL. A twenty-workload benchmark on a
    CUDA laptop ran at 3.1 tokens per second with the GPU at its idle draw,
    because the only recorded build was the CPU one. Every joule that
    benchmark reported was real; the machine was simply doing the work the
    slowest way it could.

    Stepping down rather than choosing once, because "can this machine run
    this build" has two answers and only one of them can be read off a driver
    version. The other is found by starting it: a CUDA build whose runtime
    libraries did not unpack, or whose driver is newer than the ceiling
    reported, fails at process start, and this notices and moves to the next
    rung instead of leaving a volunteer with a worker that will not boot.
    """

    check = starts if verify is None else verify
    problems: list[str] = []
    for candidate in preferred_keys(key, pins_path):
        try:
            path = ensure(
                root=root,
                key=candidate,
                progress=progress,
                opener=opener,
                pins_path=pins_path,
            )
        except (RuntimeUnavailable, RuntimeVerificationError) as exc:
            problems.append(f"{candidate}: {exc}")
            continue
        if check(path):
            return path
        problems.append(
            f"{candidate}: installed, but would not start on this machine"
        )
    raise RuntimeUnavailable(
        "no recorded llama.cpp build could be installed and started here. "
        + " ".join(problems)
    )


def describe_availability(key: str = "", pins_path: Optional[Path] = None) -> str:
    """One line about what this machine can install, for the start-up report."""

    order = [key] if key else preferred_keys(path=pins_path)
    for candidate in order:
        pin = pin_for(candidate, pins_path)
        if pin is None:
            continue
        how = "CPU" if pin.accelerator == "cpu" else pin.accelerator.upper()
        return f"llama.cpp: {pin.tag} recorded for {candidate} ({how}), verified on install"
    where = key or platform_key()
    return f"llama.cpp: no build recorded for {where}; supply --llama-server"