#!/usr/bin/env python3
"""Strict, offline provenance primitives for installed Golden Path runtimes."""
from __future__ import annotations

import base64
import configparser
import csv
import hashlib
import io
import json
import os
import stat
import sys
import zipfile
from email.parser import BytesParser
from email.policy import compat32
from pathlib import Path
from typing import Any, Iterable, Mapping, Sequence, cast
from urllib.parse import unquote_to_bytes, urlsplit


def canonical(value: object) -> bytes:
    return json.dumps(
        value,
        ensure_ascii=False,
        sort_keys=True,
        separators=(",", ":"),
        allow_nan=False,
    ).encode()


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


def _normal(name: str) -> str:
    return "-".join(name.lower().replace("_", "-").split("-"))


def _within(path: Path, roots: Iterable[Path]) -> bool:
    resolved = path.resolve(strict=True)
    return any(resolved == root.resolve(strict=True) or resolved.is_relative_to(root.resolve(strict=True)) for root in roots)


def _regular(path: Path, label: str, *, private: bool = False) -> None:
    info = path.lstat()
    if stat.S_ISLNK(info.st_mode):
        raise ValueError(f"{label} is a symlink")
    if not stat.S_ISREG(info.st_mode) or info.st_nlink != 1 or info.st_uid != os.geteuid():
        raise ValueError(f"{label} is not an owned regular file")
    if private and stat.S_IMODE(info.st_mode) not in {0o600, 0o400}:
        raise ValueError(f"{label} permissions are not private")


def _private_directory(path: Path, label: str) -> None:
    info = path.lstat()
    if stat.S_ISLNK(info.st_mode):
        raise ValueError(f"{label} is a symlink")
    if not stat.S_ISDIR(info.st_mode) or info.st_uid != os.geteuid():
        raise ValueError(f"{label} is not an owned directory")
    if stat.S_IMODE(info.st_mode) not in {0o700, 0o500}:
        raise ValueError(f"{label} permissions are not private")


def read_private_json(path: Path) -> dict[str, Any]:
    _regular(path, "provenance", private=True)
    value = json.loads(path.read_text(encoding="utf-8"))
    if not isinstance(value, dict) or any(not isinstance(key, str) for key in value):
        raise ValueError("provenance is not an object")
    return cast(dict[str, Any], value)


def _decode_record_hash(value: str, label: str) -> str:
    algorithm, separator, encoded = value.partition("=")
    if algorithm != "sha256" or not separator or not encoded:
        raise ValueError(f"{label} RECORD hash is invalid")
    try:
        raw = base64.urlsafe_b64decode(encoded + "=" * (-len(encoded) % 4))
    except Exception as exc:
        raise ValueError(f"{label} RECORD hash is invalid") from exc
    if len(raw) != hashlib.sha256().digest_size:
        raise ValueError(f"{label} RECORD hash is invalid")
    return raw.hex()


def _record_rows(data: bytes, record_relative: str, label: str) -> list[tuple[str, str, int | None]]:
    try:
        decoded = data.decode("utf-8")
        parsed = list(csv.reader(io.StringIO(decoded, newline="")))
    except (UnicodeDecodeError, csv.Error) as exc:
        raise ValueError(f"{label} RECORD is invalid") from exc
    rows: list[tuple[str, str, int | None]] = []
    seen: set[str] = set()
    for row in parsed:
        if len(row) != 3:
            raise ValueError(f"{label} RECORD row is invalid")
        relative, encoded_hash, raw_size = row
        candidate = Path(relative)
        if not relative or candidate.is_absolute() or relative in seen:
            raise ValueError(f"{label} RECORD path is invalid")
        seen.add(relative)
        if relative == record_relative:
            if encoded_hash or raw_size:
                raise ValueError(f"{label} RECORD self-row is invalid")
            rows.append((relative, "", None))
            continue
        if not encoded_hash or not raw_size.isdecimal():
            raise ValueError(f"{label} RECORD hash or size is missing")
        rows.append((relative, _decode_record_hash(encoded_hash, label), int(raw_size)))
    if record_relative not in seen:
        raise ValueError(f"{label} RECORD self-row is missing")
    return rows


def _wheel_inventory(
    wheel: Path,
    dist_info_name: str,
    installed_rows: Sequence[tuple[str, str, int | None]],
    label: str,
    *,
    site_packages: Path,
    venv: Path,
) -> str:
    _regular(wheel, f"{label} wheel", private=True)
    try:
        with zipfile.ZipFile(wheel) as archive:
            names = archive.namelist()
            if len(names) != len(set(names)) or any(name.startswith("/") or ".." in Path(name).parts for name in names):
                raise ValueError(f"{label} wheel members are invalid")
            record_relative = f"{dist_info_name}/RECORD"
            wheel_record = archive.read(record_relative)
            rows = _record_rows(wheel_record, record_relative, label)
            installed_by_path = {row[0]: row for row in installed_rows}
            if set(names) != {row[0] for row in rows}:
                raise ValueError(f"{label} wheel inventory differs from wheel RECORD")
            data_prefix = f"{dist_info_name.removesuffix('.dist-info')}.data/"
            for relative, expected_hash, expected_size in rows:
                if relative == record_relative:
                    continue
                try:
                    payload = archive.read(relative)
                except KeyError as exc:
                    raise ValueError(f"{label} wheel RECORD member is missing: {relative}") from exc
                if len(payload) != expected_size:
                    raise ValueError(f"{label} wheel RECORD size mismatch: {relative}")
                if hashlib.sha256(payload).hexdigest() != expected_hash:
                    raise ValueError(f"{label} wheel RECORD hash mismatch: {relative}")
                installed_relative = relative
                if relative.startswith(data_prefix):
                    scheme, separator, suffix = relative[len(data_prefix):].partition("/")
                    if not separator or not suffix:
                        raise ValueError(f"{label} wheel data member is invalid: {relative}")
                    if scheme in {"purelib", "platlib"}:
                        installed_relative = suffix
                    elif scheme == "data":
                        installed_relative = Path(os.path.relpath(venv / suffix, site_packages)).as_posix()
                    else:
                        raise ValueError(f"{label} unsupported wheel data scheme: {scheme}")
                installed = installed_by_path.get(installed_relative)
                if installed != (installed_relative, expected_hash, expected_size):
                    raise ValueError(f"{label} installed RECORD differs for wheel member: {relative}")
            return hashlib.sha256(wheel_record).hexdigest()
    except zipfile.BadZipFile as exc:
        raise ValueError(f"{label} wheel is invalid") from exc


def _strict_json_object(data: bytes, label: str) -> dict[str, object]:
    def object_pairs(
        pairs: list[tuple[str, object]],
    ) -> dict[str, object]:
        result: dict[str, object] = {}
        for key, value in pairs:
            if key in result:
                raise ValueError(f"{label} contains a duplicate key")
            result[key] = value
        return result

    try:
        value = json.loads(
            data.decode("utf-8", errors="strict"), object_pairs_hook=object_pairs
        )
    except (UnicodeDecodeError, json.JSONDecodeError) as exc:
        raise ValueError(f"{label} is invalid") from exc
    if not isinstance(value, dict):
        raise ValueError(f"{label} is invalid")
    return cast(dict[str, object], value)


def _portable_direct_url_provenance(
    *,
    payload: bytes,
    record_hash: str,
    record_size: str,
    record_path: str,
    wheel: Path,
    distribution_name: str,
) -> tuple[dict[str, object], dict[str, object]]:
    value = _strict_json_object(payload, "generated direct URL metadata")
    archive_value = value.get("archive_info")
    if set(value) != {"archive_info", "url"} or not isinstance(
        archive_value, dict
    ):
        raise ValueError("generated direct URL metadata is not a wheel archive")
    archive = cast(dict[str, object], archive_value)
    hashes_value = archive.get("hashes")
    if set(archive) != {"hash", "hashes"} or not isinstance(hashes_value, dict):
        raise ValueError("generated direct URL archive hash is ambiguous")
    hashes = cast(dict[str, object], hashes_value)
    wheel_sha256 = sha256_file(wheel)
    if (
        set(hashes) != {"sha256"}
        or archive.get("hash") != f"sha256={wheel_sha256}"
        or hashes.get("sha256") != wheel_sha256
    ):
        raise ValueError("generated direct URL archive hash differs from sealed wheel")
    raw_url = value.get("url")
    if not isinstance(raw_url, str):
        raise ValueError("generated direct URL wheel provenance is invalid")
    parsed = urlsplit(raw_url)
    if (
        parsed.scheme != "file"
        or parsed.netloc
        or parsed.query
        or parsed.fragment
        or not parsed.path
    ):
        raise ValueError("generated direct URL is not an offline local wheel")
    try:
        decoded_path = unquote_to_bytes(parsed.path).decode("utf-8", errors="strict")
    except UnicodeDecodeError as exc:
        raise ValueError("generated direct URL wheel path is invalid") from exc
    archive_path = Path(decoded_path)
    if (
        not archive_path.is_absolute()
        or archive_path.suffix != ".whl"
        or archive_path.name != wheel.name
    ):
        raise ValueError("generated direct URL wheel provenance is invalid")
    try:
        same_archive = os.path.samefile(archive_path, wheel)
    except OSError:
        same_archive = False
    if not same_archive:
        raise ValueError("generated direct URL differs from sealed wheel provenance")
    payload_sha256 = hashlib.sha256(payload).hexdigest()
    if (
        not record_size.isdecimal()
        or int(record_size) != len(payload)
        or _decode_record_hash(record_hash, "direct URL") != payload_sha256
    ):
        raise ValueError("generated direct URL RECORD row differs")
    portable: dict[str, object] = {
        "schema": "task26-portable-direct-url-provenance-v1",
        "distribution": distribution_name,
        "dist_info_member_path": record_path,
        "sealed_wheel_basename": wheel.name,
        "sealed_wheel_sha256": wheel_sha256,
        "archive_sha256": wheel_sha256,
    }
    raw: dict[str, object] = {
        "dist_info_member_path": record_path,
        "byte_sha256": payload_sha256,
        "record_row_sha256": _decode_record_hash(record_hash, "direct URL"),
        "size": len(payload),
        "url_path_classification": "absolute_local_file_wheel_url",
        "raw_equality_required_or_claimed": False,
    }
    return portable, raw


def portable_record_projection(
    *,
    wheel: Path,
    site_packages: Path,
    dist_info: Path,
    venv: Path,
    distribution_name: str,
    require_direct_url: bool = False,
) -> dict[str, object]:
    """Project path-independent installed RECORD semantics without hiding raw drift."""
    record = dist_info / "RECORD"
    raw_record = record.read_bytes()
    try:
        installed_csv = list(
            csv.reader(io.StringIO(raw_record.decode("utf-8"), newline=""))
        )
        with zipfile.ZipFile(wheel) as archive:
            wheel_record_name = f"{dist_info.name}/RECORD"
            wheel_rows = _record_rows(
                archive.read(wheel_record_name),
                wheel_record_name,
                distribution_name,
            )
            entry_points_data = (
                archive.read(f"{dist_info.name}/entry_points.txt")
                if f"{dist_info.name}/entry_points.txt" in archive.namelist()
                else b""
            )
    except (UnicodeDecodeError, csv.Error, zipfile.BadZipFile) as exc:
        raise ValueError("portable RECORD input is invalid") from exc
    installed_rows: dict[str, tuple[str, str]] = {}
    for row in installed_csv:
        if len(row) != 3 or not row[0] or row[0] in installed_rows:
            raise ValueError("installed RECORD row is invalid")
        installed_rows[row[0]] = (row[1], row[2])
    wheel_owned: list[dict[str, object]] = []
    generated: list[dict[str, object]] = []
    consumed: set[str] = {record.relative_to(site_packages).as_posix()}
    data_prefix = f"{dist_info.name.removesuffix('.dist-info')}.data/"
    for relative, expected_hash, expected_size in wheel_rows:
        if relative == wheel_record_name:
            continue
        installed_relative = relative
        generated_launcher = False
        if relative.startswith(data_prefix):
            scheme, separator, suffix = relative[len(data_prefix):].partition("/")
            if not separator:
                raise ValueError("unsupported path-dependent wheel entry")
            if scheme == "data":
                suffix_path = Path(suffix)
                if (
                    not suffix
                    or suffix_path.is_absolute()
                    or ".." in suffix_path.parts
                ):
                    raise ValueError("unsupported path-dependent wheel entry")
                target = venv / suffix_path
                installed_relative = Path(
                    os.path.relpath(target, site_packages)
                ).as_posix()
                row = installed_rows.get(installed_relative)
                if (
                    row is None
                    or not row[0]
                    or not row[1].isdecimal()
                    or _decode_record_hash(
                        row[0], "relocated static data"
                    )
                    != expected_hash
                    or int(row[1]) != expected_size
                ):
                    raise ValueError(
                        "wheel-owned installed RECORD row differs for relocated static data"
                    )
                consumed.add(installed_relative)
                _regular(target, "relocated static data")
                payload = target.read_bytes()
                with zipfile.ZipFile(wheel) as archive:
                    wheel_payload = archive.read(relative)
                if (
                    len(wheel_payload) != expected_size
                    or hashlib.sha256(wheel_payload).hexdigest()
                    != expected_hash
                    or payload != wheel_payload
                ):
                    raise ValueError(
                        "wheel-owned installed row differs from relocated static data"
                    )
                wheel_owned.append(
                    {
                        "kind": "wheel_owned_relocated_static_data",
                        "installed_path": installed_relative,
                        "wheel_path": relative,
                        "canonical_destination": (
                            f"{{data}}/{suffix_path.as_posix()}"
                        ),
                        "sha256": expected_hash,
                        "size": expected_size,
                    }
                )
                continue
            if scheme != "scripts":
                raise ValueError("unsupported path-dependent wheel entry")
            installed_relative = Path(
                os.path.relpath(venv / "bin" / suffix, site_packages)
            ).as_posix()
            generated_launcher = True
        row = installed_rows.get(installed_relative)
        if row is None:
            raise ValueError("wheel-owned installed RECORD row is missing")
        consumed.add(installed_relative)
        target = site_packages / installed_relative
        _regular(target, "portable RECORD installed entry")
        payload = target.read_bytes()
        if generated_launcher:
            wheel_payload = zipfile.ZipFile(wheel).read(relative)
            first, separator, remainder = payload.partition(b"\n")
            _wheel_first, wheel_separator, wheel_remainder = wheel_payload.partition(b"\n")
            expected_shebang = f"#!{venv / 'bin/python'}".encode()
            if (
                not separator
                or not wheel_separator
                or first != expected_shebang
                or wheel_remainder != remainder
            ):
                raise ValueError("generated launcher semantics are invalid")
            template = b"#!<VENV>/bin/python\n" + remainder
            generated.append(
                {
                    "kind": "path_dependent_launcher",
                    "record_path": installed_relative,
                    "wheel_path": relative,
                    "owner": distribution_name,
                    "template_sha256": hashlib.sha256(template).hexdigest(),
                    "size_without_shebang_path": len(template),
                }
            )
        else:
            if len(payload) != expected_size or hashlib.sha256(payload).hexdigest() != expected_hash:
                raise ValueError("wheel-owned installed row differs from wheel RECORD")
            wheel_owned.append(
                {
                    "installed_path": installed_relative,
                    "wheel_path": relative,
                    "sha256": expected_hash,
                    "size": expected_size,
                }
            )
    if entry_points_data:
        parser = configparser.ConfigParser(interpolation=None)
        try:
            parser.read_string(entry_points_data.decode("utf-8"))
        except (UnicodeDecodeError, configparser.Error) as exc:
            raise ValueError("wheel console entry points are invalid") from exc
        for script_name, target_spec in (
            parser.items("console_scripts")
            if parser.has_section("console_scripts")
            else ()
        ):
            target = target_spec.split("[", 1)[0].strip()
            module, separator, function = target.partition(":")
            if (
                not separator
                or not script_name
                or any(character not in "abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ0123456789._-" for character in script_name)
                or not module.replace(".", "_").isidentifier()
                or not function.isidentifier()
            ):
                raise ValueError("wheel console entry point is invalid")
            relative = Path(
                os.path.relpath(venv / "bin" / script_name, site_packages)
            ).as_posix()
            if relative not in installed_rows:
                raise ValueError("generated console launcher RECORD row is missing")
            launcher = site_packages / relative
            _regular(launcher, "generated console launcher")
            payload = launcher.read_bytes()
            first, separator_bytes, remainder = payload.partition(b"\n")
            expected_remainder = (
                "# -*- coding: utf-8 -*-\n"
                "import re\n"
                "import sys\n"
                f"from {module} import {function}\n"
                "if __name__ == '__main__':\n"
                "    sys.argv[0] = re.sub(r'(-script\\.pyw|\\.exe)?$', '', sys.argv[0])\n"
                f"    sys.exit({function}())\n"
            ).encode()
            if (
                not separator_bytes
                or first != f"#!{venv / 'bin/python'}".encode()
                or remainder != expected_remainder
            ):
                raise ValueError("generated console launcher template is invalid")
            consumed.add(relative)
            template = b"#!<VENV>/bin/python\n" + remainder
            generated.append(
                {
                    "kind": "console_entry_point_launcher",
                    "record_path": relative,
                    "owner": distribution_name,
                    "entry_point": f"{script_name}={target_spec}",
                    "template_sha256": hashlib.sha256(template).hexdigest(),
                    "size_without_shebang_path": len(template),
                }
            )
    direct_url_provenance: dict[str, object] | None = None
    raw_direct_url: dict[str, object] | None = None
    allowed_generated = {
        f"{dist_info.name}/INSTALLER": "installer_metadata",
        f"{dist_info.name}/REQUESTED": "requested_marker",
        f"{dist_info.name}/direct_url.json": "direct_url_metadata",
    }
    for relative in sorted(set(installed_rows) - consumed):
        kind = allowed_generated.get(relative)
        if kind is None:
            raise ValueError("unexpected generated installed RECORD entry")
        target = site_packages / relative
        _regular(target, "generated RECORD entry")
        payload = target.read_bytes()
        if kind == "installer_metadata" and payload.strip() != b"pip":
            raise ValueError("generated installer metadata is invalid")
        if kind == "requested_marker" and payload not in {b"", b"\n"}:
            raise ValueError("generated requested marker is invalid")
        if kind == "direct_url_metadata":
            row_hash, row_size = installed_rows[relative]
            direct_url_provenance, raw_direct_url = (
                _portable_direct_url_provenance(
                    payload=payload,
                    record_hash=row_hash,
                    record_size=row_size,
                    record_path=relative,
                    wheel=wheel,
                    distribution_name=distribution_name,
                )
            )
            generated.append(
                {
                    "kind": "canonical_direct_url_archive_provenance",
                    **direct_url_provenance,
                }
            )
            continue
        generated.append(
            {
                "kind": kind,
                "record_path": relative,
                "owner": distribution_name,
                "content_sha256": hashlib.sha256(payload).hexdigest(),
                "size": len(payload),
            }
        )
    if require_direct_url and (
        direct_url_provenance is None or raw_direct_url is None
    ):
        raise ValueError("generated direct URL wheel provenance is missing")
    core = {
        "schema": "task26-portable-installed-record-projection-v2",
        "distribution_name": distribution_name,
        "wheel_record_sha256": hashlib.sha256(
            zipfile.ZipFile(wheel).read(f"{dist_info.name}/RECORD")
        ).hexdigest(),
        "wheel_owned_rows": sorted(wheel_owned, key=lambda row: str(row["wheel_path"])),
        "generated_entries": sorted(
            generated,
            key=lambda row: str(
                row.get("record_path", row.get("dist_info_member_path", ""))
            ),
        ),
        "direct_url_provenance": direct_url_provenance,
    }
    return {
        **core,
        "projection_sha256": hashlib.sha256(canonical(core)).hexdigest(),
        "raw_installed_record_sha256": hashlib.sha256(raw_record).hexdigest(),
        "raw_installed_record_portable": False,
        "raw_direct_url": raw_direct_url,
    }


def verify_packaged_module_parity(
    *,
    source: Path,
    wheel: Path,
    wheel_member: str,
    installed: Path,
) -> dict[str, object]:
    """Prove one packaged helper is byte-identical in source, wheel, and install."""
    _regular(source, "module source")
    _regular(wheel, "module wheel", private=True)
    _regular(installed, "installed module")
    member = Path(wheel_member)
    if not wheel_member or member.is_absolute() or ".." in member.parts:
        raise ValueError("wheel module path is invalid")
    try:
        with zipfile.ZipFile(wheel) as archive:
            names = archive.namelist()
            if wheel_member not in names:
                raise ValueError("packaged module is absent from wheel")
            payload = archive.read(wheel_member)
            record_names = [name for name in names if name.endswith(".dist-info/RECORD")]
            if len(record_names) != 1:
                raise ValueError("wheel RECORD cardinality is invalid")
            rows = _record_rows(archive.read(record_names[0]), record_names[0], "module wheel")
            row = next((item for item in rows if item[0] == wheel_member), None)
            if row is None or row[1] != hashlib.sha256(payload).hexdigest() or row[2] != len(payload):
                raise ValueError("packaged module wheel RECORD binding is invalid")
    except zipfile.BadZipFile as exc:
        raise ValueError("module wheel is invalid") from exc
    source_bytes = source.read_bytes()
    installed_bytes = installed.read_bytes()
    if source_bytes != payload or installed_bytes != payload:
        raise ValueError("source, wheel, and installed module bytes differ")
    digest = hashlib.sha256(payload).hexdigest()
    return {
        "module": wheel_member.removesuffix(".py").replace("/", "."),
        "sha256": digest,
        "size": len(payload),
        "source_path": str(source.resolve(strict=True)),
        "wheel_path": str(wheel.resolve(strict=True)),
        "installed_path": str(installed.resolve(strict=True)),
    }


def inspect_installed_distribution(
    *,
    venv: Path,
    site_packages: Path,
    wheel: Path,
    distribution_name: str,
    import_packages: Sequence[str],
) -> dict[str, object]:
    """Recompute one wheel's installed inventory from METADATA and RECORD."""
    venv = venv.absolute()
    site_packages = site_packages.absolute()
    wheel = wheel.absolute()
    _private_directory(venv, "installed venv")
    if not site_packages.is_dir() or not _within(site_packages, (venv,)):
        raise ValueError("site-packages is outside the installed venv")
    matches: list[tuple[Path, bytes, str, str]] = []
    for dist_info in sorted(site_packages.glob("*.dist-info")):
        if dist_info.is_symlink() or not dist_info.is_dir():
            continue
        metadata_path = dist_info / "METADATA"
        if not metadata_path.is_file() or metadata_path.is_symlink():
            continue
        metadata = metadata_path.read_bytes()
        message = BytesParser(policy=compat32).parsebytes(metadata)
        name = str(message.get("Name", ""))
        version = str(message.get("Version", ""))
        if _normal(name) == _normal(distribution_name):
            matches.append((dist_info, metadata, name, version))
    if len(matches) != 1:
        raise ValueError(f"installed distribution cardinality is invalid: {distribution_name}")
    dist_info, metadata, declared_name, version = matches[0]
    if not version:
        raise ValueError(f"installed distribution version is unavailable: {distribution_name}")
    record = dist_info / "RECORD"
    _regular(record, f"{distribution_name} RECORD")
    record_data = record.read_bytes()
    record_relative = record.relative_to(site_packages).as_posix()
    rows = _record_rows(record_data, record_relative, distribution_name)
    inventory: list[dict[str, object]] = []
    for relative, expected_hash, expected_size in rows:
        target = site_packages / relative
        try:
            target.resolve(strict=True)
        except OSError as exc:
            raise ValueError(f"{distribution_name} RECORD file is unavailable: {relative}") from exc
        if not _within(target, (venv,)):
            raise ValueError(f"{distribution_name} RECORD path escapes the installed venv")
        _regular(target, f"{distribution_name} RECORD file")
        if relative == record_relative:
            actual_hash = hashlib.sha256(record_data).hexdigest()
            actual_size = len(record_data)
        else:
            actual_size = target.stat().st_size
            if actual_size != expected_size:
                raise ValueError(f"{distribution_name} RECORD size mismatch: {relative}")
            actual_hash = sha256_file(target)
            if actual_hash != expected_hash:
                raise ValueError(f"{distribution_name} RECORD hash mismatch: {relative}")
        inventory.append({"path": relative, "sha256": actual_hash, "size": actual_size})
    wheel_record_sha256 = _wheel_inventory(
        wheel,
        dist_info.name,
        rows,
        distribution_name,
        site_packages=site_packages,
        venv=venv,
    )
    record_projection = portable_record_projection(
        wheel=wheel,
        site_packages=site_packages,
        dist_info=dist_info,
        venv=venv,
        distribution_name=declared_name,
        require_direct_url=True,
    )
    direct_url_value = record_projection.get("direct_url_provenance")
    if not isinstance(direct_url_value, dict):
        raise ValueError("generated direct URL wheel provenance is missing")
    direct_url_provenance = cast(dict[str, object], direct_url_value)
    direct_url_path = str(direct_url_provenance["dist_info_member_path"])
    inventory = [
        (
            {
                "path": direct_url_path,
                "kind": "canonical_direct_url_archive_provenance",
                "provenance": direct_url_provenance,
            }
            if row.get("path") == direct_url_path
            else row
        )
        for row in inventory
    ]
    package_roots: list[str] = []
    for package in import_packages:
        if not package or "." in package or "/" in package:
            raise ValueError("import package name is invalid")
        root = site_packages / package
        if root.is_symlink() or not root.is_dir():
            raise ValueError(f"installed package root is unavailable: {package}")
        package_roots.append(str(root.resolve(strict=True)))
    return {
        "distribution_name": declared_name,
        "distribution_version": version,
        "dist_info": str(dist_info.resolve(strict=True)),
        "metadata_sha256": hashlib.sha256(metadata).hexdigest(),
        "package_roots": package_roots,
        "record_path": str(record.resolve(strict=True)),
        "record_sha256": hashlib.sha256(record_data).hexdigest(),
        "wheel_filename": wheel.name,
        "wheel_path": str(wheel),
        "wheel_sha256": sha256_file(wheel),
        "wheel_record_sha256": wheel_record_sha256,
        "installed_inventory": inventory,
        "installed_inventory_sha256": hashlib.sha256(canonical(inventory)).hexdigest(),
        "portable_record_projection": record_projection,
    }


def validate_import_environment(
    *,
    venv: Path,
    site_packages: Path,
    import_packages: Sequence[str],
    search_path: Sequence[str] | None = None,
    executable: Path | None = None,
) -> dict[str, object]:
    """Reject another import authority ahead of the declared site-packages."""
    venv = venv.absolute()
    site_packages = site_packages.resolve(strict=True)
    declared_python = venv / ("Scripts/python.exe" if os.name == "nt" else "bin/python")
    if not declared_python.exists():
        raise ValueError("installed interpreter is unavailable")
    actual = (executable or Path(sys.executable)).absolute()
    try:
        same = os.path.samefile(actual, declared_python)
    except OSError:
        same = False
    if not same:
        raise ValueError("running interpreter differs from declared installed interpreter")
    entries = tuple(sys.path if search_path is None else search_path)
    resolved_entries: list[Path] = []
    for entry in entries:
        candidate = Path(entry or os.getcwd()).absolute()
        if candidate.exists():
            resolved_entries.append(candidate.resolve(strict=True))
    try:
        site_index = resolved_entries.index(site_packages)
    except ValueError as exc:
        raise ValueError("declared site-packages is absent from interpreter search path") from exc
    for entry in resolved_entries[:site_index]:
        if entry == site_packages:
            continue
        if any((entry / package).exists() for package in import_packages):
            raise ValueError(f"import authority precedes declared site-packages: {entry}")
    interpreter = declared_python.resolve(strict=True)
    interpreter_info = interpreter.lstat()
    if (
        not stat.S_ISREG(interpreter_info.st_mode)
        or stat.S_IMODE(interpreter_info.st_mode) & 0o022
    ):
        raise ValueError("selected toolchain interpreter is unsafe")
    return {
        "declared_path": str(declared_python.absolute()),
        "resolved_path": str(interpreter),
        "sha256": sha256_file(interpreter),
        "version": list(sys.version_info[:3]),
        "site_packages": str(site_packages),
    }


def loaded_module_receipts(
    modules: Mapping[str, Path], *, allowed_roots: Sequence[Path]
) -> list[dict[str, object]]:
    roots = tuple(root.resolve(strict=True) for root in allowed_roots)
    receipts: list[dict[str, object]] = []
    for name, raw_path in sorted(modules.items()):
        path = raw_path.absolute()
        if not _within(path, roots):
            raise ValueError(f"loaded module is outside declared installed roots: {name}")
        _regular(path, f"loaded module {name}")
        receipts.append(
            {
                "module": name,
                "path": str(path.resolve(strict=True)),
                "sha256": sha256_file(path),
                "size": path.stat().st_size,
            }
        )
    return receipts


def portable_installed_runtime(runtime: Mapping[str, object]) -> dict[str, object]:
    """Retain only path-independent installed facts and projection identities."""
    distributions = cast(Mapping[str, object], runtime.get("distributions"))
    portable_distributions: dict[str, object] = {}
    for role in ("profile", "hermes"):
        distribution = cast(Mapping[str, object], distributions.get(role))
        inventory = [
            row
            for row in cast(Sequence[Mapping[str, object]], distribution.get("installed_inventory"))
            if ".." not in Path(str(row.get("path", ""))).parts
            and not str(row.get("path", "")).endswith(".dist-info/RECORD")
        ]
        portable_distribution: dict[str, object] = {
            key: distribution.get(key)
            for key in (
                "distribution_name", "distribution_version", "metadata_sha256",
                "wheel_filename", "wheel_sha256", "wheel_record_sha256",
            )
        }
        projection = cast(
            Mapping[str, object], distribution.get("portable_record_projection")
        )
        portable_distribution["portable_record_projection_sha256"] = projection.get(
            "projection_sha256"
        )
        portable_distribution["installed_inventory"] = inventory
        portable_distribution["installed_inventory_sha256"] = hashlib.sha256(
            canonical(inventory)
        ).hexdigest()
        portable_distributions[role] = portable_distribution
    interpreter = cast(Mapping[str, object], runtime.get("interpreter"))
    return {
        "schema": "installed-golden-runtime-portable-v2",
        "interpreter": {
            "sha256": interpreter.get("sha256"),
            "version": interpreter.get("version"),
        },
        "distributions": portable_distributions,
    }


def _hex_digest(value: object) -> bool:
    return (
        isinstance(value, str)
        and len(value) == 64
        and all(character in "0123456789abcdef" for character in value)
    )


def nonportable_installed_record_audit(
    runtime: Mapping[str, object],
    *,
    install_role: str,
    candidate_digest: str,
) -> dict[str, object]:
    """Bind raw RECORD hashes without making them portable equality inputs."""
    if install_role not in {"original", "rehydrated"} or not _hex_digest(
        candidate_digest
    ):
        raise ValueError("nonportable RECORD audit identity is invalid")
    distributions_value = runtime.get("distributions")
    if not isinstance(distributions_value, Mapping):
        raise ValueError("nonportable RECORD audit distributions are invalid")
    distributions = cast(Mapping[str, object], distributions_value)
    audit_distributions: dict[str, object] = {}
    for role in ("profile", "hermes"):
        distribution_value = distributions.get(role)
        if not isinstance(distribution_value, Mapping):
            raise ValueError("nonportable RECORD audit distribution is invalid")
        distribution = cast(Mapping[str, object], distribution_value)
        projection_value = distribution.get("portable_record_projection")
        if not isinstance(projection_value, Mapping):
            raise ValueError("nonportable RECORD projection is invalid")
        projection = cast(Mapping[str, object], projection_value)
        generated_value = projection.get("generated_entries")
        if not isinstance(generated_value, Sequence) or isinstance(
            generated_value, (str, bytes)
        ):
            raise ValueError("nonportable RECORD classifications are invalid")
        classification_values: set[str] = set()
        for generated_row in generated_value:
            if not isinstance(generated_row, Mapping):
                continue
            generated = cast(Mapping[str, object], generated_row)
            kind = generated.get("kind")
            if kind in {
                "path_dependent_launcher",
                "console_entry_point_launcher",
            }:
                classification_values.add(str(kind))
        raw_direct_url_value = projection.get("raw_direct_url")
        if not isinstance(raw_direct_url_value, Mapping):
            raise ValueError("nonportable direct URL audit is invalid")
        raw_direct_url = cast(Mapping[str, object], raw_direct_url_value)
        classification_values.update(
            {"direct_url_absolute_file_url", "direct_url_raw_record_hash"}
        )
        classifications = sorted(classification_values)
        audit_distributions[role] = {
            "distribution": distribution.get("distribution_name"),
            "wheel_sha256": distribution.get("wheel_sha256"),
            "portable_projection_sha256": projection.get("projection_sha256"),
            "raw_installed_record_sha256": projection.get(
                "raw_installed_record_sha256"
            ),
            "raw_installed_record_portable": False,
            "raw_direct_url_sha256": raw_direct_url.get("byte_sha256"),
            "raw_direct_url_record_row_sha256": raw_direct_url.get(
                "record_row_sha256"
            ),
            "raw_direct_url_equality_required_or_claimed": False,
            "reason": "validated_path_dependent_installer_material_may_change_raw_record",
            "path_dependent_classifications": classifications,
        }
    audit: dict[str, object] = {
        "schema": "task26-nonportable-installed-record-audit-v1",
        "install_role": install_role,
        "candidate_digest": candidate_digest,
        "distributions": audit_distributions,
        "raw_direct_url_hash_equality_required_or_claimed": False,
    }
    audit["audit_sha256"] = hashlib.sha256(canonical(audit)).hexdigest()
    return audit


def validate_nonportable_installed_record_audit(
    value: object,
    *,
    install_role: str,
    candidate_digest: str,
    runtime_portable: Mapping[str, object],
) -> dict[str, object]:
    if not isinstance(value, dict) or set(value) != {
        "schema",
        "install_role",
        "candidate_digest",
        "distributions",
        "raw_direct_url_hash_equality_required_or_claimed",
        "audit_sha256",
    }:
        raise ValueError("nonportable installed RECORD audit schema is invalid")
    audit = cast(dict[str, object], value)
    unsigned = {key: item for key, item in audit.items() if key != "audit_sha256"}
    portable_distributions_value = runtime_portable.get("distributions")
    audit_distributions_value = audit.get("distributions")
    if not isinstance(portable_distributions_value, Mapping) or not isinstance(
        audit_distributions_value, Mapping
    ):
        raise ValueError("nonportable installed RECORD audit distributions are invalid")
    portable_distributions = cast(Mapping[str, object], portable_distributions_value)
    audit_distributions = cast(Mapping[str, object], audit_distributions_value)
    if (
        audit.get("schema") != "task26-nonportable-installed-record-audit-v1"
        or audit.get("install_role") != install_role
        or audit.get("candidate_digest") != candidate_digest
        or audit.get("raw_direct_url_hash_equality_required_or_claimed") is not False
        or audit.get("audit_sha256")
        != hashlib.sha256(canonical(unsigned)).hexdigest()
        or set(audit_distributions) != {"profile", "hermes"}
    ):
        raise ValueError("nonportable installed RECORD audit binding is invalid")
    for role in ("profile", "hermes"):
        row_value = audit_distributions.get(role)
        portable_value = portable_distributions.get(role)
        if not isinstance(row_value, Mapping) or not isinstance(
            portable_value, Mapping
        ):
            raise ValueError("nonportable installed RECORD audit row is invalid")
        row = cast(Mapping[str, object], row_value)
        portable = cast(Mapping[str, object], portable_value)
        classifications = row.get("path_dependent_classifications")
        if (
            set(row)
            != {
                "distribution",
                "wheel_sha256",
                "portable_projection_sha256",
                "raw_installed_record_sha256",
                "raw_installed_record_portable",
                "raw_direct_url_sha256",
                "raw_direct_url_record_row_sha256",
                "raw_direct_url_equality_required_or_claimed",
                "reason",
                "path_dependent_classifications",
            }
            or row.get("distribution") != portable.get("distribution_name")
            or row.get("wheel_sha256") != portable.get("wheel_sha256")
            or row.get("portable_projection_sha256")
            != portable.get("portable_record_projection_sha256")
            or not _hex_digest(row.get("raw_installed_record_sha256"))
            or row.get("raw_installed_record_portable") is not False
            or not _hex_digest(row.get("raw_direct_url_sha256"))
            or row.get("raw_direct_url_record_row_sha256")
            != row.get("raw_direct_url_sha256")
            or row.get("raw_direct_url_equality_required_or_claimed") is not False
            or row.get("reason")
            != "validated_path_dependent_installer_material_may_change_raw_record"
            or not isinstance(classifications, list)
            or classifications != sorted(set(classifications))
            or any(
                item
                not in {
                    "path_dependent_launcher",
                    "console_entry_point_launcher",
                    "direct_url_absolute_file_url",
                    "direct_url_raw_record_hash",
                }
                for item in classifications
            )
        ):
            raise ValueError("nonportable installed RECORD audit row is invalid")
    return audit


def collect_installed_runtime(
    *,
    venv: Path,
    site_packages: Path,
    profile_wheel: Path,
    hermes_wheel: Path,
) -> dict[str, object]:
    interpreter = validate_import_environment(
        venv=venv,
        site_packages=site_packages,
        import_packages=("checkin_cli", "gateway"),
    )
    profile = inspect_installed_distribution(
        venv=venv,
        site_packages=site_packages,
        wheel=profile_wheel,
        distribution_name="physique-checkin-cli",
        import_packages=("checkin_cli",),
    )
    hermes = inspect_installed_distribution(
        venv=venv,
        site_packages=site_packages,
        wheel=hermes_wheel,
        distribution_name="hermes-agent",
        import_packages=("gateway",),
    )
    return {
        "schema": "installed-golden-runtime-v1",
        "venv": str(venv.absolute()),
        "site_packages": str(site_packages.resolve(strict=True)),
        "interpreter": interpreter,
        "distributions": {"profile": profile, "hermes": hermes},
    }