ai-contract-platform/yuqei-infra/deploy/release-package.py

#!/usr/bin/env python3
"""Verify, plan, and explicitly apply one immutable Yuqei backend package.

The default operation is read-only verification.  Install, rollback, and
forward operations are plans unless ``--apply`` and the operation's exact
confirmation string are supplied; those controls are necessary but not
sufficient.  Production apply currently fails closed while the declared
release blockers remain unresolved.  The executor is deliberately offline: it
never builds or pulls images and it never manages traffic, DNS, volumes, or
database downgrades.  Contract Web identity is package-bound but is deployed by
its separate web orchestrator, so no result from this tool claims completion of
the full cross-host release.
"""

from __future__ import annotations

import argparse
import hashlib
import importlib.util
import json
import os
import re
import stat
import subprocess
import sys
from contextlib import contextmanager
from dataclasses import dataclass
from datetime import UTC, datetime
from pathlib import Path, PurePosixPath
from typing import Any, Callable, Iterator, Mapping


REPO_ROOT = Path(__file__).resolve().parents[2]
PACKAGE_FILE = "release-package.json"
BASE_COMPOSE_RELATIVE = "yuqei-infra/compose/docker-compose.yml"
PRODUCTION_COMPOSE_RELATIVE = "yuqei-infra/compose/docker-compose.production.yml"
OFFLINE_COMPOSE_RELATIVE = "yuqei-infra/compose/docker-compose.production.offline.yml"

PACKAGE_SCHEMA = "yuqei.release-package.v1"
CHECKSUM_SCHEMA = "yuqei.release-package-checksums.v1"
SOURCE_MANIFEST_SCHEMA = "yuqei.release-source-manifest.v1"
IMAGE_IDENTITY_SCHEMA = "yuqei.release-image-identity.v1"
BACKUP_PROOF_SCHEMA = "yuqei.release-backup-proof.v1"
COMPATIBILITY_PROOF_SCHEMA = "yuqei.release-rollback-compatibility.v1"
INSTALLABLE_MANIFEST_SCHEMA = "yuqei.installable-release-manifest.v1"
DOCKER_IMPORT_PLAN_SCHEMA = "yuqei.docker-image-import-plan.v1"
PLAN_SCHEMA = "yuqei.release-package-plan.v2"
RECEIPT_SCHEMA = "yuqei.release-package-receipt.v2"
PENDING_SCHEMA = "yuqei.release-package-pending.v2"
POST_APPLY_SCHEMA = "yuqei.release-post-apply-check.v1"
PREFLIGHT_BINDING_SCHEMA = "yuqei.production-preflight-binding.v2"

DOCKER_CLI = "/usr/bin/docker"
LOCAL_DOCKER_HOST = "unix:///var/run/docker.sock"
FIXED_DOCKER_HOME = "/var/empty"
REAL_APPLY_BLOCKERS = (
    "infrastructure_docker_import_plan_not_available",
    "backup_and_compatibility_external_signatures_not_verified",
)

CONFIRMATIONS = {
    "install": "I_AUTHORIZE_THIS_EXACT_RELEASE_INSTALL",
    "rollback": "I_AUTHORIZE_THIS_EXACT_RELEASE_ROLLBACK",
    "forward": "I_AUTHORIZE_THIS_EXACT_RELEASE_FORWARD",
}

APPLICATION_IMAGE_KEYS = (
    "CONTRACT_API_IMAGE",
    "CONTRACT_WEB_IMAGE",
    "AI_PLATFORM_API_IMAGE",
    "AI_WORKER_IMAGE",
)
BACKEND_APPLICATION_IMAGE_KEYS = (
    "CONTRACT_API_IMAGE",
    "AI_PLATFORM_API_IMAGE",
    "AI_WORKER_IMAGE",
)
INFRASTRUCTURE_IMAGE_KEYS = (
    "POSTGRES_IMAGE",
    "POSTGRES_CLIENT_IMAGE",
    "REDIS_IMAGE",
    "MINIO_IMAGE",
    "MINIO_MC_IMAGE",
    "NGINX_IMAGE",
    "BACKUP_UTILITY_IMAGE",
)
BACKEND_INSTALL_INFRASTRUCTURE_IMAGE_KEYS = (
    "POSTGRES_IMAGE",
    "POSTGRES_CLIENT_IMAGE",
    "REDIS_IMAGE",
    "MINIO_IMAGE",
)
DEPLOYMENT_SCOPE = "backend_only"
SEPARATELY_DEPLOYED_IMAGE_KEYS = ("CONTRACT_WEB_IMAGE",)
RUNTIME_SERVICES = (
    "contract-api",
    "contract-review-worker",
    "contract-playbook-governance-worker",
    "ai-platform-api",
    "worker",
)
INSTALL_SERVICES = (
    "postgres",
    "postgres-init-databases",
    "redis",
    "ai-redis",
    "minio",
    "contract-api-migrate",
    "ai-platform-api-migrate",
    *RUNTIME_SERVICES,
)
REQUIRED_ARTIFACTS = (
    "release_manifest",
    "checksum_manifest",
    "image_identity",
    "installable_manifest",
)
OPTIONAL_EVIDENCE_ARTIFACTS = ("backup_proof", "compatibility_proof")

IMPORT_ROLE_ENV_KEYS = (
    ("contract-web", "CONTRACT_WEB_IMAGE"),
    ("contract-api", "CONTRACT_API_IMAGE"),
    ("ai-platform-api", "AI_PLATFORM_API_IMAGE"),
    ("ai-worker", "AI_WORKER_IMAGE"),
)

SHA256_RE = re.compile(r"^[0-9a-f]{64}$")
COMMIT_RE = re.compile(r"^(?:[0-9a-f]{40}|[0-9a-f]{64})$")
RELEASE_ID_RE = re.compile(r"^[A-Za-z0-9][A-Za-z0-9._-]{0,127}$")
PROJECT_RE = re.compile(r"^[a-z0-9][a-z0-9_-]{0,62}$")
IDEMPOTENCY_RE = re.compile(r"^[A-Za-z0-9][A-Za-z0-9._:-]{0,127}$")
APPROVAL_RE = re.compile(r"^[A-Za-z0-9][A-Za-z0-9._:/-]{0,255}$")
ENV_KEY_RE = re.compile(r"^[A-Za-z_][A-Za-z0-9_]*$")
IMAGE_REFERENCE_RE = re.compile(
    r"^[a-z0-9](?:[a-z0-9._:/-]{0,510}[a-z0-9])?@sha256:[0-9a-f]{64}$"
)
IMAGE_DIGEST_RE = re.compile(r"^sha256:[0-9a-f]{64}$")
MAX_JSON_BYTES = 4 * 1024 * 1024
MAX_ENV_BYTES = 1024 * 1024

FORBIDDEN_COMMAND_TOKENS = frozenset(
    {
        "down",
        "rm",
        "prune",
        "build",
        "pull",
        "push",
        "volume",
        "volumes",
        "downgrade",
        "edge",
        "dns",
        "--volumes",
        "-v",
    }
)


class ReleasePackageError(RuntimeError):
    """Raised when a package or requested transition fails closed."""


@dataclass(frozen=True)
class ArtifactPin:
    path: str
    sha256: str


@dataclass(frozen=True)
class ValidatedPackage:
    root: Path
    package: dict[str, Any]
    package_sha256: str
    checksum_manifest: dict[str, Any]
    checksum_manifest_sha256: str
    release_manifest: dict[str, Any]
    release_manifest_sha256: str
    image_identity: dict[str, Any]
    image_identity_sha256: str
    installable_manifest: dict[str, Any]
    installable_manifest_sha256: str
    installable_root: Path
    trusted_build_receipt_sha256: str
    docker_import_plan: tuple[dict[str, Any], ...]
    backup_proof: dict[str, Any] | None
    backup_proof_sha256: str | None
    compatibility_proof: dict[str, Any] | None
    compatibility_proof_sha256: str | None
    application_images: dict[str, dict[str, str]]
    infrastructure_images: dict[str, dict[str, str]]
    rollback_images: dict[str, dict[str, str]]
    rollback_release_id: str


@dataclass(frozen=True)
class ExecutionOutcome:
    payload: dict[str, Any]
    exit_code: int = 0
    replayed: bool = False


@dataclass(frozen=True)
class RuntimeEvidence:
    env_file: Path
    env_values: dict[str, str]
    env_sha256: str
    preflight_summary: dict[str, Any]
    compose_gate_summary: dict[str, Any]
    daemon_identity: dict[str, str]


@dataclass(frozen=True)
class SnapshotInputs:
    root: Path
    source_env_file: Path
    env_file: Path
    compose_files: tuple[Path, Path, Path]
    docker_archives: dict[str, Path]
    tls_files: dict[str, Path]
    manifest: dict[str, Any]


_PREFLIGHT_MODULE: Any | None = None
_COMPOSE_GATE_MODULE: Any | None = None
_INSTALLABLE_VALIDATOR_MODULE: Any | None = None


def _exec_module_without_bytecode(spec: Any, module: Any) -> None:
    """Execute a local validator without mutating its source tree."""
    previous = sys.dont_write_bytecode
    sys.dont_write_bytecode = True
    try:
        spec.loader.exec_module(module)
    finally:
        sys.dont_write_bytecode = previous


def _load_local_module(filename: str, module_name: str) -> Any:
    path = Path(__file__).resolve().with_name(filename)
    spec = importlib.util.spec_from_file_location(module_name, path)
    if spec is None or spec.loader is None:
        raise ReleasePackageError(f"Cannot load required deployment gate: {filename}")
    module = importlib.util.module_from_spec(spec)
    sys.modules[module_name] = module
    try:
        _exec_module_without_bytecode(spec, module)
    except Exception as exc:
        sys.modules.pop(module_name, None)
        raise ReleasePackageError(f"Cannot load required deployment gate: {filename}") from exc
    return module


def _load_preflight_module() -> Any:
    global _PREFLIGHT_MODULE
    if _PREFLIGHT_MODULE is None:
        _PREFLIGHT_MODULE = _load_local_module(
            "preflight-production.py",
            "yuqei_release_package_preflight",
        )
    return _PREFLIGHT_MODULE


def _load_compose_gate_module() -> Any:
    global _COMPOSE_GATE_MODULE
    if _COMPOSE_GATE_MODULE is None:
        _COMPOSE_GATE_MODULE = _load_local_module(
            "verify-production-compose.py",
            "yuqei_release_package_compose_gate",
        )
    return _COMPOSE_GATE_MODULE


def _load_installable_validator_module() -> Any:
    global _INSTALLABLE_VALIDATOR_MODULE
    if _INSTALLABLE_VALIDATOR_MODULE is None:
        path = Path(__file__).resolve().parents[1] / "scripts" / "build-installable-release.py"
        spec = importlib.util.spec_from_file_location(
            "yuqei_release_package_installable_validator",
            path,
        )
        if spec is None or spec.loader is None:
            raise ReleasePackageError("Cannot load the strict installable manifest validator.")
        module = importlib.util.module_from_spec(spec)
        sys.modules[spec.name] = module
        try:
            _exec_module_without_bytecode(spec, module)
        except Exception as exc:
            sys.modules.pop(spec.name, None)
            raise ReleasePackageError(
                "Cannot load the strict installable manifest validator."
            ) from exc
        _INSTALLABLE_VALIDATOR_MODULE = module
    return _INSTALLABLE_VALIDATOR_MODULE


def canonical_json_bytes(value: object) -> bytes:
    try:
        return json.dumps(
            value,
            ensure_ascii=False,
            sort_keys=True,
            separators=(",", ":"),
            allow_nan=False,
        ).encode("utf-8")
    except (TypeError, ValueError) as exc:
        raise ReleasePackageError("Value is not canonical JSON.") from exc


def canonical_sha256(value: object) -> str:
    return hashlib.sha256(canonical_json_bytes(value)).hexdigest()


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


def confirmation_for(operation: str) -> str:
    try:
        return CONFIRMATIONS[operation]
    except KeyError as exc:
        raise ReleasePackageError(f"Unsupported write operation: {operation}") from exc


def _real_apply_blockers(operation: str) -> list[str]:
    blockers = list(REAL_APPLY_BLOCKERS)
    if operation == "rollback":
        blockers.append("rollback_docker_import_plan_not_available")
    return blockers


def _reject_json_constant(value: str) -> None:
    raise ReleasePackageError(f"Non-finite JSON value is forbidden: {value}")


def _reject_duplicate_pairs(pairs: list[tuple[str, Any]]) -> dict[str, Any]:
    result: dict[str, Any] = {}
    for key, value in pairs:
        if key in result:
            raise ReleasePackageError(f"Duplicate JSON key is forbidden: {key}")
        result[key] = value
    return result


def _read_json(path: Path, *, label: str) -> dict[str, Any]:
    try:
        size = path.stat().st_size
        if size <= 0 or size > MAX_JSON_BYTES:
            raise ReleasePackageError(f"{label} must be non-empty and no larger than 4 MiB.")
        raw = path.read_bytes()
        payload = json.loads(
            raw.decode("utf-8"),
            object_pairs_hook=_reject_duplicate_pairs,
            parse_constant=_reject_json_constant,
        )
    except ReleasePackageError:
        raise
    except (OSError, UnicodeError, json.JSONDecodeError) as exc:
        raise ReleasePackageError(f"{label} is not readable strict UTF-8 JSON.") from exc
    if not isinstance(payload, dict):
        raise ReleasePackageError(f"{label} must be a JSON object.")
    return payload


def _verify_embedded_hash(
    payload: Mapping[str, Any],
    field: str,
    *,
    label: str,
) -> None:
    expected = payload.get(field)
    if not isinstance(expected, str) or not SHA256_RE.fullmatch(expected):
        raise ReleasePackageError(f"{label} has no valid {field}.")
    unsigned = {key: value for key, value in payload.items() if key != field}
    if canonical_sha256(unsigned) != expected:
        raise ReleasePackageError(f"{label} canonical payload hash mismatch.")


def _absolute_lexical(path: Path) -> Path:
    return Path(os.path.abspath(os.fspath(path)))


def _reject_symlink_components(path: Path, *, label: str) -> None:
    absolute = _absolute_lexical(path)
    current = Path(absolute.anchor)
    for part in absolute.parts[1:]:
        current = current / part
        try:
            if current.exists() and current.is_symlink():
                raise ReleasePackageError(f"{label} must not contain symbolic links: {current}")
        except OSError as exc:
            raise ReleasePackageError(f"Unable to inspect {label} path components.") from exc


def _validated_directory(path: Path, *, label: str) -> Path:
    lexical = _absolute_lexical(path)
    _reject_symlink_components(lexical, label=label)
    try:
        resolved = lexical.resolve(strict=True)
    except OSError as exc:
        raise ReleasePackageError(f"{label} does not exist.") from exc
    if lexical.is_symlink() or not resolved.is_dir():
        raise ReleasePackageError(f"{label} must be a real directory, not a symbolic link.")
    return resolved


def _validated_file(path: Path, *, label: str, max_bytes: int | None = None) -> Path:
    lexical = _absolute_lexical(path)
    if not lexical.is_absolute():  # pragma: no cover - os.path.abspath guarantees this.
        raise ReleasePackageError(f"{label} must be absolute.")
    _reject_symlink_components(lexical, label=label)
    try:
        resolved = lexical.resolve(strict=True)
        size = resolved.stat().st_size
    except OSError as exc:
        raise ReleasePackageError(f"{label} is unavailable.") from exc
    if lexical.is_symlink() or not resolved.is_file():
        raise ReleasePackageError(f"{label} must be a real regular file, not a symbolic link.")
    if max_bytes is not None and (size <= 0 or size > max_bytes):
        raise ReleasePackageError(f"{label} has an invalid size.")
    return resolved


def _safe_relative_path(value: object, *, label: str) -> str:
    if not isinstance(value, str) or not value or "\\" in value or "\x00" in value:
        raise ReleasePackageError(f"{label} must use a non-empty canonical POSIX relative path.")
    pure = PurePosixPath(value)
    if (
        pure.is_absolute()
        or pure.as_posix() != value
        or any(part in {"", ".", ".."} for part in pure.parts)
    ):
        raise ReleasePackageError(f"{label} contains path traversal or is not canonical: {value!r}")
    return value


def _artifact_file(root: Path, relative_path: str, *, label: str) -> Path:
    relative_path = _safe_relative_path(relative_path, label=label)
    lexical = root.joinpath(*PurePosixPath(relative_path).parts)
    _reject_symlink_components(lexical, label=label)
    try:
        resolved = lexical.resolve(strict=True)
        resolved.relative_to(root)
    except (OSError, ValueError) as exc:
        raise ReleasePackageError(f"{label} escapes the release package or is missing.") from exc
    if lexical.is_symlink() or not resolved.is_file():
        raise ReleasePackageError(f"{label} must be a real regular file, not a symbolic link.")
    return resolved


def _artifact_pin(package: Mapping[str, Any], name: str) -> ArtifactPin:
    artifacts = package.get("artifacts")
    if not isinstance(artifacts, dict):
        raise ReleasePackageError("release-package.json has no artifacts object.")
    raw = artifacts.get(name)
    if not isinstance(raw, dict):
        raise ReleasePackageError(f"Required release package artifact is missing: {name}")
    if set(raw) != {"path", "sha256"}:
        raise ReleasePackageError(f"Artifact {name} fields are missing or unknown.")
    path = _safe_relative_path(raw.get("path"), label=f"artifact {name} path")
    digest = raw.get("sha256")
    if not isinstance(digest, str) or not SHA256_RE.fullmatch(digest):
        raise ReleasePackageError(f"Artifact {name} has no valid lowercase SHA-256.")
    return ArtifactPin(path=path, sha256=digest)


def _validate_identifier(value: object, pattern: re.Pattern[str], *, label: str) -> str:
    if not isinstance(value, str) or not pattern.fullmatch(value):
        raise ReleasePackageError(f"{label} is missing or unsafe.")
    return value


def _validate_image_map(
    value: object,
    expected_keys: tuple[str, ...],
    *,
    label: str,
) -> dict[str, dict[str, str]]:
    if not isinstance(value, dict) or set(value) != set(expected_keys):
        raise ReleasePackageError(f"{label} must contain exactly: {', '.join(expected_keys)}")
    result: dict[str, dict[str, str]] = {}
    for key in expected_keys:
        raw = value[key]
        if not isinstance(raw, dict):
            raise ReleasePackageError(f"{label}.{key} must be an object.")
        reference = raw.get("reference")
        digest = raw.get("digest")
        if not isinstance(reference, str) or not IMAGE_REFERENCE_RE.fullmatch(reference):
            raise ReleasePackageError(f"{label}.{key} is not an immutable safe image reference.")
        if not isinstance(digest, str) or not IMAGE_DIGEST_RE.fullmatch(digest):
            raise ReleasePackageError(f"{label}.{key} has no valid image digest.")
        if reference.rsplit("@", 1)[1] != digest:
            raise ReleasePackageError(f"{label}.{key} reference and digest do not match.")
        result[key] = {"reference": reference, "digest": digest}
    return result


def _validate_installable_manifest(
    root: Path,
    payload: Mapping[str, Any],
    *,
    source_commit: str,
    application_images: Mapping[str, Mapping[str, str]],
) -> tuple[dict[str, Any], ...]:
    if payload.get("schema_version") != INSTALLABLE_MANIFEST_SCHEMA:
        raise ReleasePackageError("Unsupported installable release manifest schema.")
    _verify_embedded_hash(
        payload,
        "manifest_payload_sha256",
        label="installable release manifest",
    )
    candidate = payload.get("candidate")
    if not isinstance(candidate, dict) or candidate.get("commit") != source_commit:
        raise ReleasePackageError("Installable manifest is not bound to the release source commit.")
    images = payload.get("images")
    expected_roles = [role for role, _key in IMPORT_ROLE_ENV_KEYS]
    if (
        not isinstance(images, list)
        or any(not isinstance(item, dict) for item in images)
        or [item.get("role") for item in images] != expected_roles
    ):
        raise ReleasePackageError("Installable manifest must contain the four application roles in fixed order.")
    images_by_role = {item["role"]: item for item in images}

    raw_plan = payload.get("docker_import_plan")
    if not isinstance(raw_plan, dict) or raw_plan.get("schema_version") != DOCKER_IMPORT_PLAN_SCHEMA:
        raise ReleasePackageError("Installable manifest has no supported docker_import_plan.")
    if raw_plan.get("implicit_preloaded_images_allowed") is not False:
        raise ReleasePackageError("Docker import plan must forbid implicit preloaded images.")
    raw_imports = raw_plan.get("images")
    if not isinstance(raw_imports, list) or len(raw_imports) != len(IMPORT_ROLE_ENV_KEYS):
        raise ReleasePackageError("Docker import plan must contain exactly four application archives.")

    normalized: list[dict[str, Any]] = []
    seen_paths: set[str] = set()
    seen_ids: set[str] = set()
    for sequence, ((expected_role, expected_key), raw) in enumerate(
        zip(IMPORT_ROLE_ENV_KEYS, raw_imports, strict=True),
        start=1,
    ):
        if not isinstance(raw, dict):
            raise ReleasePackageError("Docker import plan entries must be objects.")
        if (
            raw.get("sequence") != sequence
            or raw.get("role") != expected_role
            or raw.get("environment_key") != expected_key
        ):
            raise ReleasePackageError("Docker import plan role, sequence, or environment key drifted.")
        image = images_by_role[expected_role]
        if image.get("environment_key") != expected_key:
            raise ReleasePackageError("Installable image environment key drifted.")

        archive = raw.get("archive")
        docker_archive = image.get("docker_archive")
        if not isinstance(archive, dict) or archive != docker_archive:
            raise ReleasePackageError("Docker import plan archive is not bound to its image record.")
        if set(archive) != {"path", "sha256", "size_bytes"}:
            raise ReleasePackageError("Docker archive record has unexpected or missing fields.")
        relative = _safe_relative_path(archive.get("path"), label="Docker archive path")
        digest = archive.get("sha256")
        size = archive.get("size_bytes")
        if not isinstance(digest, str) or not SHA256_RE.fullmatch(digest):
            raise ReleasePackageError("Docker archive record has an invalid SHA-256.")
        if not isinstance(size, int) or isinstance(size, bool) or size <= 0:
            raise ReleasePackageError("Docker archive record has an invalid size.")
        if relative in seen_paths:
            raise ReleasePackageError("Docker import plan reuses an archive path.")
        seen_paths.add(relative)
        archive_path = _artifact_file(root, relative, label=f"{expected_role} Docker archive")
        archive_stat = archive_path.stat()
        if archive_stat.st_nlink != 1:
            raise ReleasePackageError(f"{expected_role} Docker archive must not be hard linked.")
        if archive_stat.st_size != size or file_sha256(archive_path) != digest:
            raise ReleasePackageError(f"{expected_role} Docker archive size or SHA-256 mismatch.")

        config_digest = image.get("config_digest")
        manifest_digest = image.get("manifest_digest")
        registry_reference = image.get("registry_reference")
        expected_tag = raw.get("expected_repo_tag")
        if (
            not isinstance(expected_tag, str)
            or not expected_tag
            or len(expected_tag) > 512
            or any(character.isspace() or ord(character) < 32 for character in expected_tag)
        ):
            raise ReleasePackageError("Docker import plan has an unsafe expected repository tag.")
        if not isinstance(config_digest, str) or not IMAGE_DIGEST_RE.fullmatch(config_digest):
            raise ReleasePackageError("Installable image has an invalid config digest.")
        if config_digest in seen_ids:
            raise ReleasePackageError("Docker import plan reuses a config-based image ID.")
        seen_ids.add(config_digest)
        if not isinstance(manifest_digest, str) or not IMAGE_DIGEST_RE.fullmatch(manifest_digest):
            raise ReleasePackageError("Installable image has an invalid OCI manifest digest.")
        if not isinstance(registry_reference, str) or not IMAGE_REFERENCE_RE.fullmatch(
            registry_reference
        ):
            raise ReleasePackageError("Installable image has an invalid online registry reference.")
        if (
            raw.get("command_argv")
            != [
                str(DOCKER_CLI),
                "--host",
                LOCAL_DOCKER_HOST,
                "image",
                "load",
                "--input",
                relative,
            ]
            or raw.get("expected_loaded_image_id") != config_digest
            or raw.get("expected_oci_manifest_digest") != manifest_digest
            or raw.get("offline_compose_override") != config_digest
            or raw.get("registry_reference_for_online_release") != registry_reference
            or raw.get("repo_digests_expected_after_load") != []
            or image.get("offline_image_id") != config_digest
            or image.get("image_ref") != expected_tag
        ):
            raise ReleasePackageError("Docker import plan does not match the installable image identity.")
        identity = application_images[expected_key]
        if identity["reference"] != registry_reference or identity["digest"] != manifest_digest:
            raise ReleasePackageError(
                f"Docker import plan online identity does not match application_images.{expected_key}."
            )
        normalized.append(
            {
                "sequence": sequence,
                "role": expected_role,
                "environment_key": expected_key,
                "archive_path": relative,
                "archive_sha256": digest,
                "archive_size_bytes": size,
                "expected_repo_tag": expected_tag,
                "expected_loaded_image_id": config_digest,
                "expected_oci_manifest_digest": manifest_digest,
                "offline_compose_override": config_digest,
                "registry_reference_for_online_release": registry_reference,
            }
        )
    return tuple(normalized)


def _validate_checksum_manifest(
    root: Path,
    payload: Mapping[str, Any],
    *,
    package_file_path: str,
    checksum_file_path: str,
) -> dict[str, dict[str, Any]]:
    if payload.get("schema_version") != CHECKSUM_SCHEMA:
        raise ReleasePackageError("Unsupported checksum manifest schema.")
    _verify_embedded_hash(payload, "manifest_payload_sha256", label="checksum manifest")
    raw_items = payload.get("files")
    if not isinstance(raw_items, list) or not raw_items:
        raise ReleasePackageError("Checksum manifest must list package artifacts.")
    records: dict[str, dict[str, Any]] = {}
    casefolded: set[str] = set()
    observed_order: list[str] = []
    for raw in raw_items:
        if not isinstance(raw, dict):
            raise ReleasePackageError("Checksum manifest file entries must be objects.")
        relative = _safe_relative_path(raw.get("path"), label="checksum entry path")
        digest = raw.get("sha256")
        size = raw.get("size_bytes")
        if not isinstance(digest, str) or not SHA256_RE.fullmatch(digest):
            raise ReleasePackageError(f"Checksum entry has invalid SHA-256: {relative}")
        if not isinstance(size, int) or isinstance(size, bool) or size < 0:
            raise ReleasePackageError(f"Checksum entry has invalid size: {relative}")
        if relative in records or relative.casefold() in casefolded:
            raise ReleasePackageError(f"Checksum manifest has a duplicate path: {relative}")
        if relative in {package_file_path, checksum_file_path}:
            raise ReleasePackageError("The package and checksum manifests must not checksum themselves.")
        records[relative] = {"sha256": digest, "size_bytes": size}
        casefolded.add(relative.casefold())
        observed_order.append(relative)
    if observed_order != sorted(observed_order, key=lambda item: item.encode("utf-8")):
        raise ReleasePackageError("Checksum manifest file entries must be bytewise path sorted.")

    for relative, record in records.items():
        target = _artifact_file(root, relative, label=f"checksummed artifact {relative}")
        if target.stat().st_size != record["size_bytes"]:
            raise ReleasePackageError(f"Artifact size mismatch: {relative}")
        if file_sha256(target) != record["sha256"]:
            raise ReleasePackageError(f"Artifact checksum mismatch: {relative}")

    actual_files: set[str] = set()
    for path in root.rglob("*"):
        if path.is_symlink():
            raise ReleasePackageError(f"Release package contains a symbolic link: {path}")
        if path.is_file():
            actual_files.add(path.relative_to(root).as_posix())
        elif not path.is_dir():
            raise ReleasePackageError(f"Release package contains a non-regular entry: {path}")
    expected_files = {package_file_path, checksum_file_path, *records}
    if actual_files != expected_files:
        missing = sorted(expected_files - actual_files)
        unexpected = sorted(actual_files - expected_files)
        raise ReleasePackageError(
            "Release package file set differs from its checksum manifest: "
            f"missing={missing}, unexpected={unexpected}"
        )
    return records


def _validate_source_manifest(payload: Mapping[str, Any]) -> str:
    if payload.get("schema_version") != SOURCE_MANIFEST_SCHEMA:
        raise ReleasePackageError("Unsupported release source manifest schema.")
    _verify_embedded_hash(payload, "manifest_payload_sha256", label="release source manifest")
    candidate = payload.get("candidate")
    if not isinstance(candidate, dict):
        raise ReleasePackageError("Release source manifest has no candidate identity.")
    commit = _validate_identifier(candidate.get("commit"), COMMIT_RE, label="source commit")
    if candidate.get("worktree_was_clean") is not True:
        raise ReleasePackageError("Release source manifest is not from a clean candidate.")
    files = payload.get("files")
    if not isinstance(files, dict) or not isinstance(files.get("items"), list):
        raise ReleasePackageError("Release source manifest has no file inventory.")
    if files.get("count") != len(files["items"]):
        raise ReleasePackageError("Release source manifest file count is inconsistent.")
    return commit


def _validate_compose_inputs(source_manifest: Mapping[str, Any], repo_root: Path) -> None:
    root = _validated_directory(repo_root, label="repository root")
    raw_items = source_manifest["files"]["items"]
    by_path: dict[str, Mapping[str, Any]] = {}
    for item in raw_items:
        if not isinstance(item, dict) or not isinstance(item.get("path"), str):
            raise ReleasePackageError("Release source manifest contains an invalid file record.")
        path = _safe_relative_path(item["path"], label="release source file path")
        if path in by_path:
            raise ReleasePackageError(f"Duplicate release source file path: {path}")
        by_path[path] = item
    for relative in (
        BASE_COMPOSE_RELATIVE,
        PRODUCTION_COMPOSE_RELATIVE,
        OFFLINE_COMPOSE_RELATIVE,
    ):
        item = by_path.get(relative)
        if not isinstance(item, Mapping):
            raise ReleasePackageError(f"Release source manifest does not bind Compose input: {relative}")
        expected = item.get("sha256")
        if not isinstance(expected, str) or not SHA256_RE.fullmatch(expected):
            raise ReleasePackageError(f"Release source manifest has an invalid Compose hash: {relative}")
        target = _artifact_file(root, relative, label=f"repository Compose input {relative}")
        if file_sha256(target) != expected:
            raise ReleasePackageError(f"Repository Compose input differs from release manifest: {relative}")


def _validate_backup_proof(
    payload: Mapping[str, Any],
    *,
    release_id: str,
    rollback_release_id: str,
) -> None:
    if payload.get("schema_version") != BACKUP_PROOF_SCHEMA:
        raise ReleasePackageError("Unsupported backup proof schema.")
    _verify_embedded_hash(payload, "proof_payload_sha256", label="backup proof")
    if payload.get("release_id") != release_id or payload.get("rollback_target_release_id") != rollback_release_id:
        raise ReleasePackageError("Backup proof is not bound to this release transition.")
    if payload.get("status") != "unverified" or payload.get("restore_tested") is not False:
        raise ReleasePackageError(
            "Self-hashed backup material must remain explicitly unverified."
        )
    backup_id = payload.get("backup_id")
    if backup_id is not None:
        _validate_identifier(backup_id, APPROVAL_RE, label="backup proof backup_id")
    backup_manifest_sha = payload.get("backup_manifest_sha256")
    if backup_manifest_sha is not None and (
        not isinstance(backup_manifest_sha, str)
        or not SHA256_RE.fullmatch(backup_manifest_sha)
    ):
        raise ReleasePackageError("Backup proof has an invalid backup manifest SHA-256.")


def _validate_compatibility_proof(
    payload: Mapping[str, Any],
    *,
    release_id: str,
    rollback_release_id: str,
    rollback_images: Mapping[str, Mapping[str, str]],
) -> None:
    if payload.get("schema_version") != COMPATIBILITY_PROOF_SCHEMA:
        raise ReleasePackageError("Unsupported rollback compatibility proof schema.")
    _verify_embedded_hash(payload, "proof_payload_sha256", label="rollback compatibility proof")
    if payload.get("from_release_id") != release_id or payload.get("to_release_id") != rollback_release_id:
        raise ReleasePackageError("Compatibility proof is not bound to this release transition.")
    if payload.get("status") != "unverified":
        raise ReleasePackageError(
            "Self-hashed compatibility material must remain explicitly unverified."
        )
    if payload.get("approval_id") is not None:
        raise ReleasePackageError(
            "Untrusted compatibility material must not claim an approval identifier."
        )
    if payload.get("database_downgrade_required") is not False:
        raise ReleasePackageError("Database downgrade is forbidden.")
    if payload.get("current_schema_is_backward_compatible") not in {None, False}:
        raise ReleasePackageError(
            "Untrusted compatibility material must not claim schema compatibility."
        )
    approved = payload.get("approved_application_digests")
    if approved is not None and approved != {} and approved != []:
        raise ReleasePackageError(
            "Untrusted compatibility material must not claim approved image digests."
        )


def validate_release_package(
    package_dir: Path,
    *,
    expected_package_sha256: str,
    repo_root: Path = REPO_ROOT,
) -> ValidatedPackage:
    if not SHA256_RE.fullmatch(expected_package_sha256):
        raise ReleasePackageError("--expected-package-sha256 must be a lowercase SHA-256.")
    root = _validated_directory(package_dir, label="release package directory")
    package_path = _artifact_file(root, PACKAGE_FILE, label="release-package.json")
    package_sha = file_sha256(package_path)
    if package_sha != expected_package_sha256:
        raise ReleasePackageError("release-package.json does not match the externally approved SHA-256.")
    package = _read_json(package_path, label="release-package.json")
    if package.get("schema_version") != PACKAGE_SCHEMA:
        raise ReleasePackageError("Unsupported release package schema.")
    _verify_embedded_hash(package, "package_payload_sha256", label="release-package.json")
    expected_package_fields = {
        "schema_version",
        "release_id",
        "compose_project_name",
        "deployment_scope",
        "deployed_service_set",
        "separately_deployed_artifacts",
        "trusted_build_receipt_sha256",
        "artifacts",
        "package_payload_sha256",
    }
    if set(package) != expected_package_fields:
        raise ReleasePackageError("release-package.json fields are missing or unknown.")
    trusted_build_receipt_sha256 = package.get("trusted_build_receipt_sha256")
    if not isinstance(trusted_build_receipt_sha256, str) or not SHA256_RE.fullmatch(
        trusted_build_receipt_sha256
    ):
        raise ReleasePackageError(
            "release-package.json has no trusted build receipt raw SHA-256."
        )
    release_id = _validate_identifier(package.get("release_id"), RELEASE_ID_RE, label="release_id")
    _validate_identifier(package.get("compose_project_name"), PROJECT_RE, label="compose_project_name")
    if package.get("deployment_scope") != DEPLOYMENT_SCOPE:
        raise ReleasePackageError("release-package.json must declare deployment_scope=backend_only.")
    if package.get("deployed_service_set") != list(INSTALL_SERVICES):
        raise ReleasePackageError(
            "release-package.json deployed_service_set must equal the fixed backend install allowlist."
        )
    if package.get("separately_deployed_artifacts") != list(SEPARATELY_DEPLOYED_IMAGE_KEYS):
        raise ReleasePackageError(
            "release-package.json must identify Contract Web as a separately deployed artifact."
        )
    artifacts = package.get("artifacts")
    artifact_names = set(artifacts) if isinstance(artifacts, dict) else set()
    if (
        not isinstance(artifacts, dict)
        or not set(REQUIRED_ARTIFACTS).issubset(artifact_names)
        or not artifact_names.issubset(
            set(REQUIRED_ARTIFACTS + OPTIONAL_EVIDENCE_ARTIFACTS)
        )
    ):
        raise ReleasePackageError(
            "release-package.json must contain exactly the required release manifest, "
            "checksum manifest, image identity, and installable manifest artifacts, "
            "plus only optional unverified backup/compatibility material."
        )
    pins = {name: _artifact_pin(package, name) for name in artifacts}
    if len({pin.path.casefold() for pin in pins.values()}) != len(pins):
        raise ReleasePackageError("Release package artifact paths must be unique.")

    checksum_path = _artifact_file(root, pins["checksum_manifest"].path, label="checksum manifest")
    checksum_sha = file_sha256(checksum_path)
    if checksum_sha != pins["checksum_manifest"].sha256:
        raise ReleasePackageError("Checksum manifest SHA-256 does not match release-package.json.")
    checksum_payload = _read_json(checksum_path, label="checksum manifest")
    if checksum_payload.get("release_id") != release_id:
        raise ReleasePackageError("Checksum manifest release_id mismatch.")
    checksum_records = _validate_checksum_manifest(
        root,
        checksum_payload,
        package_file_path=PACKAGE_FILE,
        checksum_file_path=pins["checksum_manifest"].path,
    )
    for name, pin in pins.items():
        if name == "checksum_manifest":
            continue
        record = checksum_records.get(pin.path)
        if record is None or record["sha256"] != pin.sha256:
            raise ReleasePackageError(f"Checksum manifest does not bind artifact {name}.")

    release_path = _artifact_file(root, pins["release_manifest"].path, label="release source manifest")
    release_sha = file_sha256(release_path)
    if release_sha != pins["release_manifest"].sha256:
        raise ReleasePackageError("Release source manifest SHA-256 mismatch.")
    release_manifest = _read_json(release_path, label="release source manifest")
    source_commit = _validate_source_manifest(release_manifest)
    _validate_compose_inputs(release_manifest, repo_root)

    identity_path = _artifact_file(root, pins["image_identity"].path, label="image identity")
    identity_sha = file_sha256(identity_path)
    if identity_sha != pins["image_identity"].sha256:
        raise ReleasePackageError("Image identity SHA-256 mismatch.")
    image_identity = _read_json(identity_path, label="image identity")
    if image_identity.get("schema_version") != IMAGE_IDENTITY_SCHEMA:
        raise ReleasePackageError("Unsupported image identity schema.")
    _verify_embedded_hash(image_identity, "identity_payload_sha256", label="image identity")
    if (
        image_identity.get("release_id") != release_id
        or image_identity.get("source_commit") != source_commit
        or image_identity.get("source_manifest_sha256") != release_sha
    ):
        raise ReleasePackageError("Image identity is not bound to the release manifest and source commit.")
    application_images = _validate_image_map(
        image_identity.get("application_images"),
        APPLICATION_IMAGE_KEYS,
        label="application_images",
    )
    infrastructure_images = _validate_image_map(
        image_identity.get("infrastructure_images"),
        INFRASTRUCTURE_IMAGE_KEYS,
        label="infrastructure_images",
    )
    rollback = image_identity.get("rollback")
    if not isinstance(rollback, dict) or set(rollback) != {
        "target_release_id",
        "evidence_status",
        "application_images",
    }:
        raise ReleasePackageError("Image identity has no complete rollback candidate.")
    if rollback.get("evidence_status") != "external_approval_missing":
        raise ReleasePackageError(
            "Rollback image identity must remain unapproved until external evidence exists."
        )
    rollback_release_id = _validate_identifier(
        rollback.get("target_release_id"), RELEASE_ID_RE, label="rollback target release_id"
    )
    rollback_images = _validate_image_map(
        rollback.get("application_images"),
        APPLICATION_IMAGE_KEYS,
        label="rollback.application_images",
    )
    if rollback_images == application_images:
        raise ReleasePackageError("Rollback image identity must differ from the install image identity.")

    installable_path = _artifact_file(
        root,
        pins["installable_manifest"].path,
        label="installable release manifest",
    )
    installable_sha = file_sha256(installable_path)
    if installable_sha != pins["installable_manifest"].sha256:
        raise ReleasePackageError("Installable release manifest SHA-256 mismatch.")
    installable_validator = _load_installable_validator_module()
    try:
        installable_manifest = installable_validator.validate_installable_release_manifest(
            artifact_root=installable_path.parent,
            manifest_path=installable_path,
            expected_receipt_sha256=trusted_build_receipt_sha256,
            repo=_validated_directory(repo_root, label="repository root"),
        )
    except installable_validator.ReleaseValidationError as exc:
        raise ReleasePackageError(f"Strict installable manifest validation failed: {exc}") from exc
    bundle_prefix = installable_path.parent.relative_to(root)
    for record in installable_manifest.get("files", []):
        if not isinstance(record, dict):
            raise ReleasePackageError("Installable manifest file inventory is invalid.")
        relative = _safe_relative_path(
            record.get("path"),
            label="installable manifest file path",
        )
        package_relative = (bundle_prefix / PurePosixPath(relative)).as_posix()
        outer = checksum_records.get(package_relative)
        if outer != {
            "sha256": record.get("sha256"),
            "size_bytes": record.get("size_bytes"),
        }:
            raise ReleasePackageError(
                f"Outer checksum manifest does not bind installable file: {package_relative}"
            )
    receipt_record = installable_manifest.get("build_receipt")
    if (
        not isinstance(receipt_record, dict)
        or receipt_record.get("sha256") != trusted_build_receipt_sha256
    ):
        raise ReleasePackageError(
            "Installable manifest is not bound to the trusted build receipt raw SHA-256."
        )
    build_receipt_path = _artifact_file(
        installable_path.parent,
        str(receipt_record["path"]),
        label="trusted build receipt",
    )
    build_receipt = _read_json(build_receipt_path, label="trusted build receipt")
    receipt_candidate = build_receipt.get("candidate")
    source_record = (
        receipt_candidate.get("source_manifest")
        if isinstance(receipt_candidate, dict)
        else None
    )
    if not isinstance(source_record, dict):
        raise ReleasePackageError("Trusted build receipt has no source manifest record.")
    receipt_source_path = _artifact_file(
        installable_path.parent,
        str(source_record.get("path")),
        label="trusted build receipt source manifest",
    )
    if receipt_source_path != release_path or source_record.get("sha256") != release_sha:
        raise ReleasePackageError(
            "Release package source manifest is not the build receipt source manifest."
        )
    docker_import_plan = _validate_installable_manifest(
        installable_path.parent,
        installable_manifest,
        source_commit=source_commit,
        application_images=application_images,
    )

    backup_proof: dict[str, Any] | None = None
    backup_sha: str | None = None
    if "backup_proof" in pins:
        backup_path = _artifact_file(root, pins["backup_proof"].path, label="backup proof")
        backup_sha = file_sha256(backup_path)
        if backup_sha != pins["backup_proof"].sha256:
            raise ReleasePackageError("Backup proof SHA-256 mismatch.")
        backup_proof = _read_json(backup_path, label="backup proof")
        _validate_backup_proof(
            backup_proof,
            release_id=release_id,
            rollback_release_id=rollback_release_id,
        )

    compatibility_proof: dict[str, Any] | None = None
    compatibility_sha: str | None = None
    if "compatibility_proof" in pins:
        compatibility_path = _artifact_file(
            root, pins["compatibility_proof"].path, label="rollback compatibility proof"
        )
        compatibility_sha = file_sha256(compatibility_path)
        if compatibility_sha != pins["compatibility_proof"].sha256:
            raise ReleasePackageError("Rollback compatibility proof SHA-256 mismatch.")
        compatibility_proof = _read_json(
            compatibility_path,
            label="rollback compatibility proof",
        )
        _validate_compatibility_proof(
            compatibility_proof,
            release_id=release_id,
            rollback_release_id=rollback_release_id,
            rollback_images=rollback_images,
        )

    return ValidatedPackage(
        root=root,
        package=package,
        package_sha256=package_sha,
        checksum_manifest=checksum_payload,
        checksum_manifest_sha256=checksum_sha,
        release_manifest=release_manifest,
        release_manifest_sha256=release_sha,
        image_identity=image_identity,
        image_identity_sha256=identity_sha,
        installable_manifest=installable_manifest,
        installable_manifest_sha256=installable_sha,
        installable_root=installable_path.parent,
        trusted_build_receipt_sha256=trusted_build_receipt_sha256,
        docker_import_plan=docker_import_plan,
        backup_proof=backup_proof,
        backup_proof_sha256=backup_sha,
        compatibility_proof=compatibility_proof,
        compatibility_proof_sha256=compatibility_sha,
        application_images=application_images,
        infrastructure_images=infrastructure_images,
        rollback_images=rollback_images,
        rollback_release_id=rollback_release_id,
    )


def parse_env_file(path: Path) -> dict[str, str]:
    compose_gate = _load_compose_gate_module()
    try:
        _target, values = compose_gate.parse_strict_env(path)
    except compose_gate.ComposeGateError as exc:
        raise ReleasePackageError(str(exc)) from exc
    return values


def _fixed_runtime_environment(overrides: Mapping[str, str] | None = None) -> dict[str, str]:
    environment = {
        "HOME": FIXED_DOCKER_HOME,
        "LANG": "C.UTF-8",
        "LC_ALL": "C.UTF-8",
    }
    if overrides:
        reserved = {
            "PATH",
            "DOCKER_HOST",
            "DOCKER_CONTEXT",
            "DOCKER_CONFIG",
            "COMPOSE_FILE",
            "COMPOSE_PATH_SEPARATOR",
            "COMPOSE_PROFILES",
            "COMPOSE_PROJECT_NAME",
        }
        conflict = sorted(reserved.intersection(overrides))
        if conflict:
            raise ReleasePackageError(
                f"Runtime overrides contain forbidden ambient-control keys: {', '.join(conflict)}"
            )
        environment.update(overrides)
    return environment


def _run_production_preflight(env_values: dict[str, str]) -> dict[str, Any]:
    preflight = _load_preflight_module()
    try:
        environment_checks = preflight.validate_environment(dict(env_values))
        file_checks = preflight.validate_files(dict(env_values))
    except Exception as exc:
        raise ReleasePackageError("Production preflight raised while validating protected inputs.") from exc

    def bound_record(check: Any) -> dict[str, Any]:
        record: dict[str, Any] = {"key": check.key, "status": check.status}
        details = getattr(check, "details", None)
        if details:
            record["details"] = details
        return record

    environment_records = [bound_record(check) for check in environment_checks]
    file_records = [bound_record(check) for check in file_checks]
    failed = [
        f"environment:{record['key']}"
        for record in environment_records
        if record["status"] == "fail"
    ] + [
        f"files:{record['key']}"
        for record in file_records
        if record["status"] == "fail"
    ]
    if failed:
        raise ReleasePackageError(
            "Production preflight failed before any state write: " + ", ".join(failed)
        )
    summary: dict[str, Any] = {
        "schema_version": PREFLIGHT_BINDING_SCHEMA,
        "status": "passed",
        "environment_checks": environment_records,
        "file_checks": file_records,
    }
    summary["summary_sha256"] = canonical_sha256(summary)
    return summary


def _tls_material_binding(preflight_summary: Mapping[str, Any]) -> dict[str, str]:
    records = preflight_summary.get("file_checks")
    if not isinstance(records, list):
        raise ReleasePackageError("Production preflight has no TLS file-check inventory.")
    matches = [
        record
        for record in records
        if isinstance(record, dict) and record.get("key") == "tls_material_binding"
    ]
    if len(matches) != 1 or matches[0].get("status") != "pass":
        raise ReleasePackageError("Production preflight has no unique passing TLS material binding.")
    details = matches[0].get("details")
    expected_fields = {
        "certificate_sha256",
        "private_key_sha256",
        "trusted_ca_sha256",
        "server_name",
    }
    if not isinstance(details, dict) or set(details) != expected_fields:
        raise ReleasePackageError("Production preflight TLS material binding is incomplete.")
    for field in (
        "certificate_sha256",
        "private_key_sha256",
        "trusted_ca_sha256",
    ):
        value = details.get(field)
        if not isinstance(value, str) or not SHA256_RE.fullmatch(value):
            raise ReleasePackageError(f"Production preflight has an invalid TLS {field}.")
    server_name = details.get("server_name")
    if (
        not isinstance(server_name, str)
        or not server_name
        or server_name != server_name.strip()
        or any(ord(character) < 32 or ord(character) == 127 for character in server_name)
    ):
        raise ReleasePackageError("Production preflight has an invalid TLS server name.")
    return {field: str(details[field]) for field in sorted(expected_fields)}


def _parse_json_text(value: object, *, label: str, max_bytes: int = MAX_JSON_BYTES) -> dict[str, Any]:
    raw = _output_bytes(value)
    if len(raw) > max_bytes:
        raise ReleasePackageError(f"{label} exceeds the maximum allowed size.")
    try:
        payload = json.loads(
            raw.decode("utf-8"),
            object_pairs_hook=_reject_duplicate_pairs,
            parse_constant=_reject_json_constant,
        )
    except ReleasePackageError:
        raise
    except (UnicodeError, ValueError, TypeError) as exc:
        raise ReleasePackageError(f"{label} is not strict UTF-8 JSON.") from exc
    if not isinstance(payload, dict):
        raise ReleasePackageError(f"{label} must be a JSON object.")
    return payload


def _docker_daemon_identity(
    *,
    runner: Callable[..., Any],
    repo_root: Path,
) -> dict[str, str]:
    command = [
        str(DOCKER_CLI),
        "--host",
        LOCAL_DOCKER_HOST,
        "info",
        "--format",
        "{{json .}}",
    ]
    try:
        completed = runner(
            command,
            check=False,
            capture_output=True,
            text=True,
            shell=False,
            cwd=str(repo_root),
            env=_fixed_runtime_environment(),
            timeout=30,
        )
    except (OSError, subprocess.TimeoutExpired) as exc:
        raise ReleasePackageError("Cannot query the fixed local Docker daemon identity.") from exc
    if int(completed.returncode) != 0:
        raise ReleasePackageError("Cannot query the fixed local Docker daemon identity.")
    payload = _parse_json_text(completed.stdout, label="Docker daemon identity")
    fields = {
        "daemon_id": payload.get("ID"),
        "daemon_name": payload.get("Name"),
        "docker_root_dir": payload.get("DockerRootDir"),
        "server_version": payload.get("ServerVersion"),
    }
    if any(not isinstance(value, str) or not value.strip() for value in fields.values()):
        raise ReleasePackageError("Docker daemon identity is incomplete.")
    identity: dict[str, str] = {
        "docker_cli": str(DOCKER_CLI),
        "docker_host": LOCAL_DOCKER_HOST,
        **{key: str(value) for key, value in fields.items()},
    }
    identity["identity_sha256"] = canonical_sha256(identity)
    return identity


def _collect_runtime_evidence(
    *,
    env_file: Path,
    repo_root: Path,
    inspection_runner: Callable[..., Any],
) -> RuntimeEvidence:
    env_target = _validated_file(env_file, label="production env file", max_bytes=MAX_ENV_BYTES)
    env_values = parse_env_file(env_target)
    preflight_summary = _run_production_preflight(env_values)

    compose_gate = _load_compose_gate_module()
    try:
        compose_report = compose_gate.verify_production_compose(
            env_file=env_target,
            repo_root=repo_root,
            runner=inspection_runner,
        )
    except compose_gate.ComposeGateError as exc:
        raise ReleasePackageError(f"Production Compose merge gate failed: {exc}") from exc
    compose_summary = dict(compose_report)
    compose_summary["summary_sha256"] = canonical_sha256(compose_summary)
    daemon_identity = _docker_daemon_identity(runner=inspection_runner, repo_root=repo_root)
    return RuntimeEvidence(
        env_file=env_target,
        env_values=env_values,
        env_sha256=file_sha256(env_target),
        preflight_summary=preflight_summary,
        compose_gate_summary=compose_summary,
        daemon_identity=daemon_identity,
    )


def _run_offline_execution_gate(
    *,
    env_file: Path,
    repo_root: Path,
    compose_files: tuple[Path, Path, Path],
    project_name: str,
    offline_application_overrides: Mapping[str, str],
    inspection_runner: Callable[..., Any],
) -> dict[str, Any]:
    compose_gate = _load_compose_gate_module()
    env_values = parse_env_file(env_file)
    env_values.update(offline_application_overrides)
    child_env = _fixed_runtime_environment(offline_application_overrides)
    cases: list[dict[str, Any]] = []
    try:
        for profile_case in compose_gate.PROFILE_CASES:
            command = compose_gate._command(
                project_name=project_name,
                env_file=env_file,
                compose_files=compose_files,
                profile_case=profile_case,
            )
            payload = compose_gate._run_config(
                command,
                repo_root=repo_root,
                runner=inspection_runner,
                child_env=child_env,
                label=f"offline-execution/{profile_case.name}",
            )
            cases.append(
                compose_gate._validate_model(
                    payload,
                    env_values=env_values,
                    mode="offline",
                    profile_case=profile_case,
                )
            )
    except compose_gate.ComposeGateError as exc:
        raise ReleasePackageError(f"Offline execution Compose gate failed: {exc}") from exc
    summary: dict[str, Any] = {
        "schema_version": "yuqei.offline-compose-execution-gate.v1",
        "status": "passed",
        "docker_host": LOCAL_DOCKER_HOST,
        "project_name": project_name,
        "offline_application_overrides": dict(offline_application_overrides),
        "cases": cases,
    }
    summary["summary_sha256"] = canonical_sha256(summary)
    return summary


def _assert_safe_compose_command(command: list[str], *, operation: str) -> None:
    if command[:4] != [str(DOCKER_CLI), "--host", LOCAL_DOCKER_HOST, "compose"]:
        raise ReleasePackageError("Only the fixed local Docker Compose target is allowed.")
    if any("\x00" in item or "\r" in item or "\n" in item for item in command):
        raise ReleasePackageError("Deployment argv contains a forbidden control character.")
    if command.count("up") != 1 or command.count("--pull") != 1 or "--no-build" not in command:
        raise ReleasePackageError("Offline execution requires compose up --pull never --no-build.")
    pull_index = command.index("--pull")
    if pull_index + 1 >= len(command) or command[pull_index + 1] != "never":
        raise ReleasePackageError("Offline execution requires the exact --pull never policy.")
    lowered = [item.casefold() for item in command]
    for index, token in enumerate(lowered):
        if token in FORBIDDEN_COMMAND_TOKENS:
            pull_pair = token == "pull" and index > 0 and lowered[index - 1] == "--pull"
            if not pull_pair:
                raise ReleasePackageError(f"Forbidden deployment command token: {command[index]}")
    if "edge" in lowered or any("dns" in item for item in lowered):
        raise ReleasePackageError("Automatic edge or DNS operations are forbidden.")
    if operation in {"rollback", "forward"} and "--no-deps" not in command:
        raise ReleasePackageError("Rollback and forward must switch application services with --no-deps.")
    expected_services = RUNTIME_SERVICES if operation in {"rollback", "forward"} else INSTALL_SERVICES
    if command[-len(expected_services) :] != list(expected_services):
        raise ReleasePackageError("Deployment command service set is not the fixed operation allowlist.")


def _compose_command(
    *,
    operation: str,
    project_name: str,
    env_file: Path,
    repo_root: Path,
    compose_files: tuple[Path, Path, Path] | None = None,
) -> list[str]:
    if compose_files is None:
        root = _validated_directory(repo_root, label="repository root")
        base = _artifact_file(root, BASE_COMPOSE_RELATIVE, label="base Compose file")
        production = _artifact_file(
            root,
            PRODUCTION_COMPOSE_RELATIVE,
            label="production Compose file",
        )
        offline = _artifact_file(
            root,
            OFFLINE_COMPOSE_RELATIVE,
            label="offline Compose override",
        )
    else:
        base, production, offline = compose_files
        for label, path in (
            ("base Compose snapshot", base),
            ("production Compose snapshot", production),
            ("offline Compose snapshot", offline),
        ):
            _validated_file(path, label=label, max_bytes=MAX_JSON_BYTES)
    command = [
        str(DOCKER_CLI),
        "--host",
        LOCAL_DOCKER_HOST,
        "compose",
        "--project-name",
        project_name,
        "--env-file",
        str(env_file),
        "-f",
        str(base),
        "-f",
        str(production),
        "-f",
        str(offline),
        "up",
        "-d",
        "--pull",
        "never",
        "--no-build",
    ]
    if operation in {"rollback", "forward"}:
        command.append("--no-deps")
        command.extend(RUNTIME_SERVICES)
    else:
        command.extend(INSTALL_SERVICES)
    _assert_safe_compose_command(command, operation=operation)
    return command


def build_plan(
    package: ValidatedPackage,
    *,
    operation: str,
    env_file: Path,
    repo_root: Path = REPO_ROOT,
    inspection_runner: Callable[..., Any] = subprocess.run,
) -> dict[str, Any]:
    if operation not in CONFIRMATIONS:
        raise ReleasePackageError("Operation must be install, rollback, or forward.")
    if not env_file.is_absolute():
        raise ReleasePackageError("--env-file must be an absolute path.")
    evidence = _collect_runtime_evidence(
        env_file=env_file,
        repo_root=repo_root,
        inspection_runner=inspection_runner,
    )
    tls_binding = _tls_material_binding(evidence.preflight_summary)
    env_target = evidence.env_file
    env_values = evidence.env_values
    project_name = package.package["compose_project_name"]
    if env_values.get("COMPOSE_PROJECT_NAME") != project_name:
        raise ReleasePackageError("Production env COMPOSE_PROJECT_NAME does not match the release package.")

    if operation == "rollback":
        target_release_id = package.rollback_release_id
        application_images = package.rollback_images
    else:
        target_release_id = package.package["release_id"]
        application_images = package.application_images
    environment_overrides = {
        key: application_images[key]["reference"] for key in BACKEND_APPLICATION_IMAGE_KEYS
    }
    environment_overrides["AI_PLATFORM_RELEASE_IMAGE_DIGEST"] = application_images[
        "AI_PLATFORM_API_IMAGE"
    ]["digest"]
    if operation == "install":
        environment_overrides.update(
            {
                key: package.infrastructure_images[key]["reference"]
                for key in BACKEND_INSTALL_INFRASTRUCTURE_IMAGE_KEYS
            }
        )
    import_plan = [dict(item) for item in package.docker_import_plan]
    offline_application_overrides = (
        {
            item["environment_key"]: item["offline_compose_override"]
            for item in import_plan
        }
        if operation in {"install", "forward"}
        else {}
    )
    apply_blockers = _real_apply_blockers(operation)
    source_root = _validated_directory(repo_root, label="repository root")
    source_compose_files = (
        _artifact_file(source_root, BASE_COMPOSE_RELATIVE, label="base Compose file"),
        _artifact_file(
            source_root,
            PRODUCTION_COMPOSE_RELATIVE,
            label="production Compose file",
        ),
        _artifact_file(
            source_root,
            OFFLINE_COMPOSE_RELATIVE,
            label="offline Compose override",
        ),
    )
    offline_execution_gate = _run_offline_execution_gate(
        env_file=env_target,
        repo_root=source_root,
        compose_files=source_compose_files,
        project_name=project_name,
        offline_application_overrides=offline_application_overrides,
        inspection_runner=inspection_runner,
    )
    command = _compose_command(
        operation=operation,
        project_name=project_name,
        env_file=env_target,
        repo_root=repo_root,
    )
    unsigned: dict[str, Any] = {
        "schema_version": PLAN_SCHEMA,
        "mode": "plan",
        "operation": operation,
        "deployment_scope": DEPLOYMENT_SCOPE,
        "deployed_service_set": list(
            RUNTIME_SERVICES if operation in {"rollback", "forward"} else INSTALL_SERVICES
        ),
        "separately_deployed_artifacts": [
            {
                "image_key": key,
                "reference": application_images[key]["reference"],
                "digest": application_images[key]["digest"],
                "deployment_scope": "separate_web_orchestrator",
                "applied_by_this_operation": False,
            }
            for key in SEPARATELY_DEPLOYED_IMAGE_KEYS
        ],
        "identity_bound_image_keys": list(APPLICATION_IMAGE_KEYS + INFRASTRUCTURE_IMAGE_KEYS),
        "release_completion_claimed": False,
        "release_id": package.package["release_id"],
        "target_release_id": target_release_id,
        "package_manifest_sha256": package.package_sha256,
        "release_manifest_sha256": package.release_manifest_sha256,
        "checksum_manifest_sha256": package.checksum_manifest_sha256,
        "image_identity_sha256": package.image_identity_sha256,
        "installable_manifest_sha256": package.installable_manifest_sha256,
        "trusted_build_receipt_sha256": package.trusted_build_receipt_sha256,
        "backup_proof_sha256": package.backup_proof_sha256,
        "compatibility_proof_sha256": package.compatibility_proof_sha256,
        "env_file": str(env_target),
        "env_file_sha256": evidence.env_sha256,
        "compose_project_name": project_name,
        "docker_daemon_identity": evidence.daemon_identity,
        "production_preflight": evidence.preflight_summary,
        "tls_material_binding": tls_binding,
        "tls_material_binding_sha256": canonical_sha256(tls_binding),
        "compose_merge_gate": evidence.compose_gate_summary,
        "offline_execution_gate": offline_execution_gate,
        "docker_import_plan": import_plan,
        "offline_application_overrides": offline_application_overrides,
        "apply_eligible": False,
        "backup_compatibility_evidence_status": (
            "optional_self_hashed_material_untrusted_external_signatures_missing_blocked"
        ),
        "backup_evidence_status": (
            "missing" if package.backup_proof is None else "unverified_self_hashed"
        ),
        "compatibility_evidence_status": (
            "missing"
            if package.compatibility_proof is None
            else "unverified_self_hashed"
        ),
        "real_apply_feature_gate": {
            "status": "blocked",
            "blockers": apply_blockers,
        },
        "offline": True,
        "pull_policy": "never",
        "build_allowed": False,
        "database_downgrade_allowed": False,
        "edge_or_dns_change_allowed": False,
        "environment_overrides": environment_overrides,
        "command_argv": command,
    }
    unsigned["plan_sha256"] = canonical_sha256(unsigned)
    return unsigned


def verification_payload(package: ValidatedPackage) -> dict[str, Any]:
    payload: dict[str, Any] = {
        "schema_version": PLAN_SCHEMA,
        "mode": "verify",
        "status": "structurally_verified_not_apply_eligible",
        "package_structure_valid": True,
        "apply_eligible": False,
        "real_apply_feature_gate": {
            "status": "blocked",
            "blockers": list(REAL_APPLY_BLOCKERS),
        },
        "deployment_scope": DEPLOYMENT_SCOPE,
        "deployed_service_set": [],
        "managed_backend_service_set": list(INSTALL_SERVICES),
        "separately_deployed_artifacts": [
            {
                "image_key": key,
                "reference": package.application_images[key]["reference"],
                "digest": package.application_images[key]["digest"],
                "deployment_scope": "separate_web_orchestrator",
                "applied_by_this_operation": False,
            }
            for key in SEPARATELY_DEPLOYED_IMAGE_KEYS
        ],
        "identity_bound_image_keys": list(APPLICATION_IMAGE_KEYS + INFRASTRUCTURE_IMAGE_KEYS),
        "release_completion_claimed": False,
        "release_id": package.package["release_id"],
        "rollback_target_release_id": package.rollback_release_id,
        "package_manifest_sha256": package.package_sha256,
        "release_manifest_sha256": package.release_manifest_sha256,
        "checksum_manifest_sha256": package.checksum_manifest_sha256,
        "image_identity_sha256": package.image_identity_sha256,
        "installable_manifest_sha256": package.installable_manifest_sha256,
        "trusted_build_receipt_sha256": package.trusted_build_receipt_sha256,
        "docker_import_plan_status": "validated_for_four_application_images",
        "infrastructure_import_plan_status": "missing_blocked",
        "backup_compatibility_evidence_status": (
            "optional_self_hashed_material_untrusted_external_signatures_missing_blocked"
        ),
        "backup_evidence_status": (
            "missing" if package.backup_proof is None else "unverified_self_hashed"
        ),
        "compatibility_evidence_status": (
            "missing"
            if package.compatibility_proof is None
            else "unverified_self_hashed"
        ),
        "backup_proof_sha256": package.backup_proof_sha256,
        "compatibility_proof_sha256": package.compatibility_proof_sha256,
        "offline": True,
        "writes_performed": False,
    }
    payload["plan_sha256"] = canonical_sha256(payload)
    return payload


def _validate_state_target(state_dir: Path, *, package_root: Path) -> Path:
    lexical = _absolute_lexical(state_dir)
    if not state_dir.is_absolute():
        raise ReleasePackageError("--state-dir must be an absolute path.")
    if lexical == Path(lexical.anchor):
        raise ReleasePackageError("--state-dir must not be a filesystem root.")
    _reject_symlink_components(lexical, label="state directory")
    existing = lexical
    while not existing.exists() and existing != existing.parent:
        existing = existing.parent
    try:
        existing_resolved = existing.resolve(strict=True)
    except OSError as exc:
        raise ReleasePackageError("State directory parent is unavailable.") from exc
    if not existing_resolved.is_dir():
        raise ReleasePackageError("State directory parent is not a directory.")
    try:
        lexical.resolve(strict=False).relative_to(package_root)
    except ValueError:
        pass
    else:
        raise ReleasePackageError("State directory must be outside the immutable release package.")
    if lexical.exists() and not lexical.is_dir():
        raise ReleasePackageError("State directory path is not a directory.")
    return lexical


def _fsync_directory(path: Path) -> None:
    if os.name == "nt":
        return
    flags = os.O_RDONLY | getattr(os, "O_DIRECTORY", 0)
    descriptor = os.open(path, flags)
    try:
        os.fsync(descriptor)
    finally:
        os.close(descriptor)


def _prepare_secure_state_root(state_dir: Path, *, require_root_owner: bool) -> None:
    existed = state_dir.exists()
    state_dir.mkdir(parents=True, exist_ok=True, mode=0o700)
    _reject_symlink_components(state_dir, label="state directory")
    if not state_dir.is_dir() or state_dir.is_symlink():
        raise ReleasePackageError("State root must be a real directory.")
    if os.name != "nt":
        if not existed:
            os.chmod(state_dir, 0o700)
        state_stat = state_dir.stat()
        expected_uid = 0 if require_root_owner else os.geteuid()
        if state_stat.st_uid != expected_uid:
            raise ReleasePackageError("State root has the wrong owner.")
        if stat.S_IMODE(state_stat.st_mode) != 0o700:
            raise ReleasePackageError("State root must have mode 0700.")
        _fsync_directory(state_dir.parent)


def _secure_snapshot_copy(
    source: Path,
    destination: Path,
    *,
    expected_sha256: str,
    expected_size: int | None = None,
    label: str,
) -> dict[str, Any]:
    source = _validated_file(source, label=label)
    flags = os.O_RDONLY | getattr(os, "O_BINARY", 0) | getattr(os, "O_NOFOLLOW", 0)
    try:
        source_descriptor = os.open(source, flags)
    except OSError as exc:
        raise ReleasePackageError(f"Cannot open {label} for an immutable snapshot.") from exc
    destination_descriptor = -1
    try:
        before = os.fstat(source_descriptor)
        if not stat.S_ISREG(before.st_mode) or before.st_nlink != 1:
            raise ReleasePackageError(f"{label} must be a single-link regular file.")
        if expected_size is not None and before.st_size != expected_size:
            raise ReleasePackageError(f"{label} size changed before snapshot.")
        destination.parent.mkdir(parents=True, exist_ok=True, mode=0o700)
        _reject_symlink_components(destination.parent, label="snapshot directory")
        destination_descriptor = os.open(
            destination,
            os.O_CREAT | os.O_EXCL | os.O_WRONLY | getattr(os, "O_BINARY", 0),
            0o600,
        )
        digest = hashlib.sha256()
        copied = 0
        while True:
            chunk = os.read(source_descriptor, 1024 * 1024)
            if not chunk:
                break
            digest.update(chunk)
            copied += len(chunk)
            view = memoryview(chunk)
            while view:
                written = os.write(destination_descriptor, view)
                view = view[written:]
        after = os.fstat(source_descriptor)
        if (
            before.st_dev,
            before.st_ino,
            before.st_size,
            getattr(before, "st_mtime_ns", None),
        ) != (
            after.st_dev,
            after.st_ino,
            after.st_size,
            getattr(after, "st_mtime_ns", None),
        ):
            raise ReleasePackageError(f"{label} changed while being snapshotted.")
        observed_sha256 = digest.hexdigest()
        if observed_sha256 != expected_sha256:
            raise ReleasePackageError(f"{label} SHA-256 changed before snapshot.")
        if expected_size is not None and copied != expected_size:
            raise ReleasePackageError(f"{label} size changed while being snapshotted.")
        os.fsync(destination_descriptor)
    except OSError as exc:
        raise ReleasePackageError(f"Cannot create immutable snapshot for {label}.") from exc
    finally:
        os.close(source_descriptor)
        if destination_descriptor >= 0:
            os.close(destination_descriptor)
    if os.name != "nt":
        os.chmod(destination, 0o600)
    destination_stat = destination.stat()
    if destination_stat.st_nlink != 1:
        raise ReleasePackageError(f"Snapshot for {label} is unexpectedly hard linked.")
    return {
        "path": destination.as_posix(),
        "sha256": expected_sha256,
        "size_bytes": destination_stat.st_size,
    }


def _write_exclusive_bytes(path: Path, payload: bytes, *, label: str) -> dict[str, Any]:
    path.parent.mkdir(parents=True, exist_ok=True, mode=0o700)
    _reject_symlink_components(path.parent, label=f"{label} directory")
    descriptor = -1
    try:
        descriptor = os.open(
            path,
            os.O_CREAT | os.O_EXCL | os.O_WRONLY | getattr(os, "O_BINARY", 0),
            0o600,
        )
        view = memoryview(payload)
        while view:
            written = os.write(descriptor, view)
            view = view[written:]
        os.fsync(descriptor)
    except OSError as exc:
        raise ReleasePackageError(f"Cannot create immutable {label}.") from exc
    finally:
        if descriptor >= 0:
            os.close(descriptor)
    metadata = os.lstat(path)
    if not stat.S_ISREG(metadata.st_mode) or metadata.st_nlink != 1:
        raise ReleasePackageError(f"Immutable {label} is not a single-link regular file.")
    return {
        "path": path.as_posix(),
        "sha256": hashlib.sha256(payload).hexdigest(),
        "size_bytes": len(payload),
    }


def _rewrite_execution_env(
    source: Path,
    destination: Path,
    *,
    replacements: Mapping[str, str],
) -> dict[str, Any]:
    try:
        text = source.read_text(encoding="utf-8-sig")
    except (OSError, UnicodeError) as exc:
        raise ReleasePackageError("Snapshotted production env is not readable UTF-8 text.") from exc
    remaining = set(replacements)
    rewritten: list[str] = []
    for raw_line in text.splitlines(keepends=True):
        ending = ""
        body = raw_line
        for candidate in ("\r\n", "\n", "\r"):
            if body.endswith(candidate):
                body = body[: -len(candidate)]
                ending = candidate
                break
        normalized = body.strip()
        if normalized and not normalized.startswith("#") and "=" in normalized:
            key = normalized.split("=", 1)[0]
            if key in replacements:
                value = replacements[key]
                if not value or any(character in value for character in "\r\n\x00"):
                    raise ReleasePackageError(f"Execution env replacement for {key} is unsafe.")
                body = f"{key}={value}"
                remaining.discard(key)
        rewritten.append(body + ending)
    if remaining:
        raise ReleasePackageError(
            "Production env is missing TLS keys required for execution: "
            + ", ".join(sorted(remaining))
        )
    payload = "".join(rewritten).encode("utf-8")
    record = _write_exclusive_bytes(destination, payload, label="execution env snapshot")
    parsed = parse_env_file(destination)
    if any(parsed.get(key) != value for key, value in replacements.items()):
        raise ReleasePackageError("Execution env TLS paths do not match the immutable snapshot.")
    return record


def _verify_immutable_snapshot_file(
    path: Path,
    *,
    expected_sha256: str,
    expected_size: int,
    label: str,
) -> None:
    target = _validated_file(path, label=label)
    flags = os.O_RDONLY | getattr(os, "O_BINARY", 0) | getattr(os, "O_NOFOLLOW", 0)
    descriptor = -1
    try:
        descriptor = os.open(target, flags)
        before = os.fstat(descriptor)
        if (
            not stat.S_ISREG(before.st_mode)
            or before.st_nlink != 1
            or before.st_size != expected_size
        ):
            raise ReleasePackageError(f"{label} metadata changed before execution.")
        digest = hashlib.sha256()
        while True:
            chunk = os.read(descriptor, 1024 * 1024)
            if not chunk:
                break
            digest.update(chunk)
        after = os.fstat(descriptor)
        if (
            before.st_dev,
            before.st_ino,
            before.st_size,
            getattr(before, "st_mtime_ns", None),
        ) != (
            after.st_dev,
            after.st_ino,
            after.st_size,
            getattr(after, "st_mtime_ns", None),
        ):
            raise ReleasePackageError(f"{label} changed during execution verification.")
        if digest.hexdigest() != expected_sha256:
            raise ReleasePackageError(f"{label} SHA-256 changed before execution.")
    except OSError as exc:
        raise ReleasePackageError(f"Cannot verify immutable {label} before execution.") from exc
    finally:
        if descriptor >= 0:
            os.close(descriptor)


def _verify_execution_snapshot(snapshot: SnapshotInputs, plan: Mapping[str, Any]) -> None:
    files = snapshot.manifest.get("files")
    if not isinstance(files, list):
        raise ReleasePackageError("Execution snapshot has no file inventory.")
    by_path = {
        str(record.get("path")): record
        for record in files
        if isinstance(record, dict) and isinstance(record.get("path"), str)
    }
    required = [snapshot.env_file, *snapshot.tls_files.values()]
    for path in required:
        record = by_path.get(path.as_posix())
        if (
            not isinstance(record, dict)
            or not isinstance(record.get("sha256"), str)
            or not SHA256_RE.fullmatch(str(record.get("sha256")))
            or not isinstance(record.get("size_bytes"), int)
        ):
            raise ReleasePackageError("Execution snapshot TLS or env inventory is incomplete.")
        _verify_immutable_snapshot_file(
            path,
            expected_sha256=str(record["sha256"]),
            expected_size=int(record["size_bytes"]),
            label=f"execution snapshot {path.name}",
        )
    if snapshot.manifest.get("tls_material_binding_sha256") != plan.get(
        "tls_material_binding_sha256"
    ):
        raise ReleasePackageError("Execution snapshot TLS binding does not match the release plan.")


def _freeze_snapshot_tree(root: Path) -> None:
    if os.name == "nt":
        return
    files = [path for path in root.rglob("*") if path.is_file()]
    directories = [path for path in root.rglob("*") if path.is_dir()]
    for path in files:
        os.chmod(path, 0o400)
    for path in sorted(directories, key=lambda item: len(item.parts), reverse=True):
        os.chmod(path, 0o500)
    os.chmod(root, 0o500)


def _snapshot_inputs(
    *,
    state_dir: Path,
    idempotency_key: str,
    plan: Mapping[str, Any],
    package: ValidatedPackage,
    repo_root: Path,
) -> SnapshotInputs:
    key_digest = hashlib.sha256(idempotency_key.encode("utf-8")).hexdigest()
    snapshots_root = state_dir / "snapshots"
    snapshots_root.mkdir(mode=0o700, exist_ok=True)
    _reject_symlink_components(snapshots_root, label="snapshot root")
    snapshot_root = snapshots_root / key_digest
    try:
        snapshot_root.mkdir(mode=0o700)
    except FileExistsError as exc:
        raise ReleasePackageError(
            "A prior snapshot exists for this idempotency key; reconciliation is required."
        ) from exc

    records: list[dict[str, Any]] = []
    source_env_destination = snapshot_root / "production.source.env"
    records.append(
        _secure_snapshot_copy(
            Path(str(plan["env_file"])),
            source_env_destination,
            expected_sha256=str(plan["env_file_sha256"]),
            label="production env file",
        )
    )

    binding = _tls_material_binding(plan["production_preflight"])
    binding_sha256 = canonical_sha256(binding)
    if (
        binding != plan.get("tls_material_binding")
        or binding_sha256 != plan.get("tls_material_binding_sha256")
    ):
        raise ReleasePackageError("Release plan TLS material binding is inconsistent.")
    source_values = parse_env_file(source_env_destination)
    source_tls_dir = Path(source_values.get("BACKEND_TLS_CERT_DIR", ""))
    source_ca_path = Path(source_values.get("BACKEND_TLS_TRUSTED_CA_FILE", ""))
    tls_directory = snapshot_root / "tls"
    tls_files = {
        "certificate": tls_directory / "tls.crt",
        "private_key": tls_directory / "tls.key",
        "trusted_ca": tls_directory / "trusted-ca.pem",
    }
    records.append(
        _secure_snapshot_copy(
            source_tls_dir / "tls.crt",
            tls_files["certificate"],
            expected_sha256=binding["certificate_sha256"],
            label="TLS certificate",
        )
    )
    records.append(
        _secure_snapshot_copy(
            source_tls_dir / "tls.key",
            tls_files["private_key"],
            expected_sha256=binding["private_key_sha256"],
            label="TLS private key",
        )
    )
    records.append(
        _secure_snapshot_copy(
            source_ca_path,
            tls_files["trusted_ca"],
            expected_sha256=binding["trusted_ca_sha256"],
            label="TLS trusted CA",
        )
    )
    env_destination = snapshot_root / "production.env"
    execution_env_record = _rewrite_execution_env(
        source_env_destination,
        env_destination,
        replacements={
            "BACKEND_TLS_CERT_DIR": tls_directory.as_posix(),
            "BACKEND_TLS_TRUSTED_CA_FILE": tls_files["trusted_ca"].as_posix(),
        },
    )
    records.append(execution_env_record)

    compose_hashes = {
        item["path"]: item["sha256"] for item in plan["compose_merge_gate"]["compose_inputs"]
    }
    compose_destinations: list[Path] = []
    for relative in (
        BASE_COMPOSE_RELATIVE,
        PRODUCTION_COMPOSE_RELATIVE,
        OFFLINE_COMPOSE_RELATIVE,
    ):
        destination = snapshot_root.joinpath(*PurePosixPath(relative).parts)
        records.append(
            _secure_snapshot_copy(
                repo_root.joinpath(*PurePosixPath(relative).parts),
                destination,
                expected_sha256=compose_hashes[relative],
                label=f"Compose input {relative}",
            )
        )
        compose_destinations.append(destination)

    archive_destinations: dict[str, Path] = {}
    for item in plan["docker_import_plan"]:
        role = item["role"]
        destination = snapshot_root / "docker-archives" / f"{role}.docker.tar"
        records.append(
            _secure_snapshot_copy(
                package.installable_root.joinpath(
                    *PurePosixPath(item["archive_path"]).parts
                ),
                destination,
                expected_sha256=item["archive_sha256"],
                expected_size=item["archive_size_bytes"],
                label=f"{role} Docker archive",
            )
        )
        archive_destinations[role] = destination

    manifest: dict[str, Any] = {
        "schema_version": "yuqei.release-input-snapshot.v2",
        "plan_sha256": plan["plan_sha256"],
        "tls_material_binding_sha256": binding_sha256,
        "execution_env_sha256": execution_env_record["sha256"],
        "files": sorted(records, key=lambda item: item["path"].encode("utf-8")),
    }
    manifest["snapshot_payload_sha256"] = canonical_sha256(manifest)
    _write_exclusive_canonical_json(snapshot_root / "snapshot-manifest.json", manifest)
    for directory in {
        snapshot_root,
        source_env_destination.parent,
        env_destination.parent,
        *(path.parent for path in tls_files.values()),
        *(path.parent for path in compose_destinations),
        *(path.parent for path in archive_destinations.values()),
    }:
        _fsync_directory(directory)
    snapshot = SnapshotInputs(
        root=snapshot_root,
        source_env_file=source_env_destination,
        env_file=env_destination,
        compose_files=(
            compose_destinations[0],
            compose_destinations[1],
            compose_destinations[2],
        ),
        docker_archives=archive_destinations,
        tls_files=tls_files,
        manifest=manifest,
    )
    _verify_execution_snapshot(snapshot, plan)
    _freeze_snapshot_tree(snapshot_root)
    return snapshot


@contextmanager
def _exclusive_lock(state_dir: Path) -> Iterator[Path]:
    _reject_symlink_components(state_dir, label="state directory")
    lock_path = state_dir / ".release-package.lock"
    try:
        descriptor = os.open(lock_path, os.O_CREAT | os.O_EXCL | os.O_WRONLY, 0o600)
    except FileExistsError as exc:
        raise ReleasePackageError("Another release package operation holds the exclusive lock.") from exc
    except OSError as exc:
        raise ReleasePackageError("Unable to acquire the release package lock.") from exc
    try:
        os.write(descriptor, f"pid={os.getpid()}\n".encode("ascii"))
        os.fsync(descriptor)
        _fsync_directory(state_dir)
        os.close(descriptor)
        descriptor = -1
        yield lock_path
    finally:
        if descriptor >= 0:
            os.close(descriptor)
        try:
            lock_path.unlink(missing_ok=True)
        except OSError as exc:  # pragma: no cover - an unhealthy filesystem should remain visible.
            raise ReleasePackageError("Unable to release the release package lock.") from exc


def _receipt_path(state_dir: Path, idempotency_key: str) -> Path:
    key_digest = hashlib.sha256(idempotency_key.encode("utf-8")).hexdigest()
    return state_dir / "receipts" / f"{key_digest}.json"


def _pending_path(state_dir: Path, idempotency_key: str) -> Path:
    key_digest = hashlib.sha256(idempotency_key.encode("utf-8")).hexdigest()
    return state_dir / "pending" / f"{key_digest}.json"


def _pending_journal(
    *,
    plan: Mapping[str, Any],
    idempotency_key: str,
    snapshot: SnapshotInputs,
    command: list[str],
    recorded_at: datetime,
) -> dict[str, Any]:
    if recorded_at.tzinfo is None:
        raise ReleasePackageError("Pending journal time must include a timezone.")
    payload: dict[str, Any] = {
        "schema_version": PENDING_SCHEMA,
        "status": "reconcile_required",
        "phase": "pending_image_import_and_backend_apply",
        "operation": plan["operation"],
        "deployment_scope": DEPLOYMENT_SCOPE,
        "deployed_service_set": plan["deployed_service_set"],
        "idempotency_key": idempotency_key,
        "recorded_at": recorded_at.astimezone(UTC).isoformat().replace("+00:00", "Z"),
        "plan_sha256": plan["plan_sha256"],
        "snapshot_payload_sha256": snapshot.manifest["snapshot_payload_sha256"],
        "tls_material_binding_sha256": plan["tls_material_binding_sha256"],
        "execution_env_sha256": snapshot.manifest["execution_env_sha256"],
        "docker_daemon_identity": plan["docker_daemon_identity"],
        "installable_manifest_sha256": plan["installable_manifest_sha256"],
        "trusted_build_receipt_sha256": plan["trusted_build_receipt_sha256"],
        "docker_import_plan_sha256": canonical_sha256(plan["docker_import_plan"]),
        "command_argv": command,
    }
    payload["journal_payload_sha256"] = canonical_sha256(payload)
    return payload


def _read_pending_journal(path: Path) -> dict[str, Any]:
    payload = _read_json(path, label="pending reconciliation journal")
    if payload.get("schema_version") != PENDING_SCHEMA:
        raise ReleasePackageError("Pending reconciliation journal has an unsupported schema.")
    expected_fields = {
        "schema_version",
        "status",
        "phase",
        "operation",
        "deployment_scope",
        "deployed_service_set",
        "idempotency_key",
        "recorded_at",
        "plan_sha256",
        "snapshot_payload_sha256",
        "tls_material_binding_sha256",
        "execution_env_sha256",
        "docker_daemon_identity",
        "installable_manifest_sha256",
        "trusted_build_receipt_sha256",
        "docker_import_plan_sha256",
        "command_argv",
        "journal_payload_sha256",
    }
    if set(payload) != expected_fields:
        raise ReleasePackageError("Pending reconciliation journal has unexpected or missing fields.")
    _verify_embedded_hash(
        payload,
        "journal_payload_sha256",
        label="pending reconciliation journal",
    )
    if payload.get("status") != "reconcile_required":
        raise ReleasePackageError("Pending journal has an invalid status.")
    for field in (
        "plan_sha256",
        "snapshot_payload_sha256",
        "tls_material_binding_sha256",
        "execution_env_sha256",
        "installable_manifest_sha256",
        "trusted_build_receipt_sha256",
        "docker_import_plan_sha256",
    ):
        value = payload.get(field)
        if not isinstance(value, str) or not SHA256_RE.fullmatch(value):
            raise ReleasePackageError(f"Pending reconciliation journal has an invalid {field}.")
    if path.read_bytes() != canonical_json_bytes(payload) + b"\n":
        raise ReleasePackageError("Pending reconciliation journal is not canonical JSON.")
    return payload


def _validate_receipt_payload(payload: Mapping[str, Any], *, label: str) -> None:
    expected_fields = {
        "schema_version",
        "operation",
        "status",
        "execution_mode",
        "deployment_scope",
        "deployed_service_set",
        "separately_deployed_artifacts",
        "identity_bound_image_keys",
        "release_completion_claimed",
        "release_id",
        "target_release_id",
        "idempotency_key",
        "recorded_at",
        "plan_sha256",
        "package_manifest_sha256",
        "release_manifest_sha256",
        "checksum_manifest_sha256",
        "image_identity_sha256",
        "installable_manifest_sha256",
        "trusted_build_receipt_sha256",
        "backup_proof_sha256",
        "compatibility_proof_sha256",
        "docker_daemon_identity",
        "production_preflight_summary_sha256",
        "compose_merge_gate_summary_sha256",
        "offline_execution_gate_summary_sha256",
        "snapshot_payload_sha256",
        "tls_material_binding_sha256",
        "execution_env_sha256",
        "docker_image_import_result",
        "post_apply_check",
        "offline",
        "pull_policy",
        "build_allowed",
        "database_downgrade_performed",
        "edge_or_dns_change_performed",
        "environment_overrides",
        "command_argv",
        "result",
        "receipt_payload_sha256",
    }
    if set(payload) != expected_fields:
        raise ReleasePackageError(f"{label} has unexpected or missing fields.")
    if payload.get("schema_version") != RECEIPT_SCHEMA:
        raise ReleasePackageError(f"{label} has an unsupported schema.")
    _verify_embedded_hash(payload, "receipt_payload_sha256", label=label)
    operation = payload.get("operation")
    if operation not in CONFIRMATIONS:
        raise ReleasePackageError(f"{label} has an invalid operation.")
    if payload.get("status") != "backend_verified":
        raise ReleasePackageError(f"{label} has an invalid status.")
    if payload.get("execution_mode") != "test_injected_runner":
        raise ReleasePackageError(f"{label} has an invalid execution mode.")
    if payload.get("deployment_scope") != DEPLOYMENT_SCOPE:
        raise ReleasePackageError(f"{label} does not declare the backend-only deployment scope.")
    expected_services = RUNTIME_SERVICES if operation in {"rollback", "forward"} else INSTALL_SERVICES
    if payload.get("deployed_service_set") != list(expected_services):
        raise ReleasePackageError(f"{label} has an invalid backend deployed_service_set.")
    if payload.get("release_completion_claimed") is not False:
        raise ReleasePackageError(f"{label} must not claim completion of the full release.")
    if payload.get("identity_bound_image_keys") != list(
        APPLICATION_IMAGE_KEYS + INFRASTRUCTURE_IMAGE_KEYS
    ):
        raise ReleasePackageError(f"{label} does not bind the complete production image inventory.")
    separate = payload.get("separately_deployed_artifacts")
    if not isinstance(separate, list) or len(separate) != len(SEPARATELY_DEPLOYED_IMAGE_KEYS):
        raise ReleasePackageError(f"{label} has an invalid separately deployed artifact inventory.")
    for expected_key, artifact in zip(SEPARATELY_DEPLOYED_IMAGE_KEYS, separate, strict=True):
        if (
            not isinstance(artifact, dict)
            or set(artifact)
            != {
                "image_key",
                "reference",
                "digest",
                "deployment_scope",
                "applied_by_this_operation",
            }
            or artifact.get("image_key") != expected_key
            or artifact.get("deployment_scope") != "separate_web_orchestrator"
            or artifact.get("applied_by_this_operation") is not False
            or not IMAGE_REFERENCE_RE.fullmatch(str(artifact.get("reference", "")))
            or not IMAGE_DIGEST_RE.fullmatch(str(artifact.get("digest", "")))
            or str(artifact.get("reference", "")).rsplit("@", 1)[-1]
            != artifact.get("digest")
        ):
            raise ReleasePackageError(f"{label} has an invalid separately deployed artifact record.")

    _validate_identifier(payload.get("release_id"), RELEASE_ID_RE, label=f"{label} release_id")
    _validate_identifier(
        payload.get("target_release_id"),
        RELEASE_ID_RE,
        label=f"{label} target_release_id",
    )
    _validate_identifier(
        payload.get("idempotency_key"),
        IDEMPOTENCY_RE,
        label=f"{label} idempotency_key",
    )
    recorded_at = payload.get("recorded_at")
    if not isinstance(recorded_at, str) or not recorded_at.endswith("Z"):
        raise ReleasePackageError(f"{label} has an invalid recorded_at timestamp.")
    try:
        parsed_time = datetime.fromisoformat(recorded_at[:-1] + "+00:00")
    except ValueError as exc:
        raise ReleasePackageError(f"{label} has an invalid recorded_at timestamp.") from exc
    if parsed_time.tzinfo is None or parsed_time.utcoffset() != UTC.utcoffset(parsed_time):
        raise ReleasePackageError(f"{label} recorded_at must be UTC.")

    for hash_field in (
        "plan_sha256",
        "package_manifest_sha256",
        "release_manifest_sha256",
        "checksum_manifest_sha256",
        "image_identity_sha256",
        "installable_manifest_sha256",
        "trusted_build_receipt_sha256",
        "production_preflight_summary_sha256",
        "compose_merge_gate_summary_sha256",
        "offline_execution_gate_summary_sha256",
        "snapshot_payload_sha256",
        "tls_material_binding_sha256",
        "execution_env_sha256",
    ):
        value = payload.get(hash_field)
        if not isinstance(value, str) or not SHA256_RE.fullmatch(value):
            raise ReleasePackageError(f"{label} has an invalid {hash_field}.")
    for optional_hash_field in ("backup_proof_sha256", "compatibility_proof_sha256"):
        value = payload.get(optional_hash_field)
        if value is not None and (
            not isinstance(value, str) or not SHA256_RE.fullmatch(value)
        ):
            raise ReleasePackageError(f"{label} has an invalid {optional_hash_field}.")

    daemon = payload.get("docker_daemon_identity")
    daemon_fields = {
        "docker_cli",
        "docker_host",
        "daemon_id",
        "daemon_name",
        "docker_root_dir",
        "server_version",
        "identity_sha256",
    }
    if not isinstance(daemon, dict) or set(daemon) != daemon_fields:
        raise ReleasePackageError(f"{label} has an invalid Docker daemon identity.")
    _verify_embedded_hash(daemon, "identity_sha256", label=f"{label} Docker daemon identity")
    if daemon.get("docker_cli") != str(DOCKER_CLI) or daemon.get("docker_host") != LOCAL_DOCKER_HOST:
        raise ReleasePackageError(f"{label} is not bound to the fixed local Docker target.")
    if any(
        not isinstance(daemon.get(key), str) or not str(daemon.get(key)).strip()
        for key in ("daemon_id", "daemon_name", "docker_root_dir", "server_version")
    ):
        raise ReleasePackageError(f"{label} has an incomplete Docker daemon identity.")

    import_result = payload.get("docker_image_import_result")
    if not isinstance(import_result, dict) or set(import_result) != {
        "schema_version",
        "status",
        "images",
        "summary_sha256",
    }:
        raise ReleasePackageError(f"{label} has an invalid Docker image import result.")
    _verify_embedded_hash(
        import_result,
        "summary_sha256",
        label=f"{label} Docker image import result",
    )
    imported_images = import_result.get("images")
    if (
        import_result.get("schema_version") != "yuqei.docker-image-import-result.v1"
        or import_result.get("status") != "verified"
        or not isinstance(imported_images, list)
        or len(imported_images) != len(IMPORT_ROLE_ENV_KEYS)
    ):
        raise ReleasePackageError(f"{label} has an invalid Docker image import result.")
    import_fields = {
        "sequence",
        "role",
        "archive_sha256",
        "loaded_image_id",
        "offline_compose_override",
        "load_stdout_sha256",
        "load_stderr_sha256",
    }
    for sequence, ((role, env_key), record) in enumerate(
        zip(IMPORT_ROLE_ENV_KEYS, imported_images, strict=True),
        start=1,
    ):
        if (
            not isinstance(record, dict)
            or set(record) != import_fields
            or record.get("sequence") != sequence
            or record.get("role") != role
            or not IMAGE_DIGEST_RE.fullmatch(str(record.get("loaded_image_id", "")))
            or record.get("offline_compose_override") != record.get("loaded_image_id")
            or any(
                not isinstance(record.get(field), str)
                or not SHA256_RE.fullmatch(str(record.get(field)))
                for field in ("archive_sha256", "load_stdout_sha256", "load_stderr_sha256")
            )
        ):
            raise ReleasePackageError(f"{label} has an invalid imported image record.")

    overrides = payload.get("environment_overrides")
    expected_override_keys = {
        *APPLICATION_IMAGE_KEYS,
        "AI_PLATFORM_RELEASE_IMAGE_DIGEST",
    }
    if operation == "install":
        expected_override_keys.update(BACKEND_INSTALL_INFRASTRUCTURE_IMAGE_KEYS)
    if not isinstance(overrides, dict) or set(overrides) != expected_override_keys:
        raise ReleasePackageError(f"{label} has an invalid environment override set.")
    for (_role, env_key), record in zip(IMPORT_ROLE_ENV_KEYS, imported_images, strict=True):
        if overrides.get(env_key) != record["loaded_image_id"]:
            raise ReleasePackageError(f"{label} does not bind {env_key} to its loaded image ID.")
    if not IMAGE_DIGEST_RE.fullmatch(str(overrides.get("AI_PLATFORM_RELEASE_IMAGE_DIGEST", ""))):
        raise ReleasePackageError(f"{label} has an invalid AI release image digest.")
    for key in BACKEND_INSTALL_INFRASTRUCTURE_IMAGE_KEYS if operation == "install" else ():
        if not IMAGE_REFERENCE_RE.fullmatch(str(overrides.get(key, ""))):
            raise ReleasePackageError(f"{label} has an invalid infrastructure image override.")

    command = payload.get("command_argv")
    if not isinstance(command, list) or any(not isinstance(item, str) for item in command):
        raise ReleasePackageError(f"{label} has an invalid command argv.")
    _assert_safe_compose_command(command, operation=operation)

    post_check = payload.get("post_apply_check")
    expected_post_fields = {
        "schema_version",
        "status",
        "deployment_scope",
        "operation",
        "target_release_id",
        "deployed_service_set",
        "daemon_identity_sha256",
        "post_check_payload_sha256",
    }
    if not isinstance(post_check, dict) or set(post_check) != expected_post_fields:
        raise ReleasePackageError(f"{label} has an invalid post-apply check.")
    _verify_embedded_hash(post_check, "post_check_payload_sha256", label=f"{label} post-apply check")
    if (
        post_check.get("schema_version") != POST_APPLY_SCHEMA
        or post_check.get("status") != "verified"
        or post_check.get("deployment_scope") != DEPLOYMENT_SCOPE
        or post_check.get("operation") != operation
        or post_check.get("target_release_id") != payload.get("target_release_id")
        or post_check.get("deployed_service_set") != payload.get("deployed_service_set")
        or post_check.get("daemon_identity_sha256") != daemon.get("identity_sha256")
    ):
        raise ReleasePackageError(f"{label} has an invalid post-apply check binding.")

    if (
        payload.get("offline") is not True
        or payload.get("pull_policy") != "never"
        or payload.get("build_allowed") is not False
        or payload.get("database_downgrade_performed") is not False
        or payload.get("edge_or_dns_change_performed") is not False
    ):
        raise ReleasePackageError(f"{label} violates offline or destructive-action boundaries.")
    result = payload.get("result")
    if (
        not isinstance(result, dict)
        or set(result) != {"returncode", "stdout_sha256", "stderr_sha256"}
        or result.get("returncode") != 0
        or any(
            not isinstance(result.get(field), str)
            or not SHA256_RE.fullmatch(str(result.get(field)))
            for field in ("stdout_sha256", "stderr_sha256")
        )
    ):
        raise ReleasePackageError(f"{label} has an invalid successful execution result.")


def _read_existing_receipt(path: Path) -> dict[str, Any]:
    if path.is_symlink():
        raise ReleasePackageError("Idempotency receipt must not be a symbolic link.")
    payload = _read_json(path, label="idempotency receipt")
    _validate_receipt_payload(payload, label="idempotency receipt")
    if path.read_bytes() != canonical_json_bytes(payload) + b"\n":
        raise ReleasePackageError("Existing idempotency receipt is not canonical JSON.")
    return payload


def _write_exclusive_canonical_json(path: Path, payload: Mapping[str, Any]) -> None:
    data = canonical_json_bytes(payload) + b"\n"
    try:
        descriptor = os.open(path, os.O_CREAT | os.O_EXCL | os.O_WRONLY, 0o600)
    except FileExistsError as exc:
        raise ReleasePackageError(f"Refusing to overwrite existing receipt: {path}") from exc
    try:
        with os.fdopen(descriptor, "wb") as handle:
            descriptor = -1
            handle.write(data)
            handle.flush()
            os.fsync(handle.fileno())
        _fsync_directory(path.parent)
    finally:
        if descriptor >= 0:
            os.close(descriptor)


def _output_bytes(value: object) -> bytes:
    if value is None:
        return b""
    if isinstance(value, bytes):
        return value
    return str(value).encode("utf-8", "replace")


def _run_application_image_imports(
    *,
    plan: Mapping[str, Any],
    snapshot: SnapshotInputs,
    runner: Callable[..., Any],
) -> tuple[bool, dict[str, Any]]:
    results: list[dict[str, Any]] = []
    runtime_environment = _fixed_runtime_environment()
    for item in plan["docker_import_plan"]:
        role = item["role"]
        archive = snapshot.docker_archives[role]
        if file_sha256(archive) != item["archive_sha256"]:
            return False, {"stage": "archive_recheck", "role": role}
        load_command = [
            str(DOCKER_CLI),
            "--host",
            LOCAL_DOCKER_HOST,
            "image",
            "load",
            "--input",
            str(archive),
        ]
        try:
            loaded = runner(
                load_command,
                check=False,
                capture_output=True,
                shell=False,
                env=runtime_environment,
                timeout=1800,
            )
        except (OSError, subprocess.TimeoutExpired) as exc:
            return False, {"stage": "docker_load", "role": role, "error": type(exc).__name__}
        loaded_stdout = _output_bytes(getattr(loaded, "stdout", b""))
        loaded_stderr = _output_bytes(getattr(loaded, "stderr", b""))
        if int(loaded.returncode) != 0:
            return False, {
                "stage": "docker_load",
                "role": role,
                "returncode": int(loaded.returncode),
                "stdout_sha256": hashlib.sha256(loaded_stdout).hexdigest(),
                "stderr_sha256": hashlib.sha256(loaded_stderr).hexdigest(),
            }
        inspect_command = [
            str(DOCKER_CLI),
            "--host",
            LOCAL_DOCKER_HOST,
            "image",
            "inspect",
            "--format",
            "{{.Id}}",
            item["expected_repo_tag"],
        ]
        try:
            inspected = runner(
                inspect_command,
                check=False,
                capture_output=True,
                shell=False,
                env=runtime_environment,
                timeout=60,
            )
        except (OSError, subprocess.TimeoutExpired) as exc:
            return False, {"stage": "docker_inspect", "role": role, "error": type(exc).__name__}
        observed_id = _output_bytes(getattr(inspected, "stdout", b"")).decode(
            "utf-8",
            errors="replace",
        ).strip()
        if int(inspected.returncode) != 0 or observed_id != item["expected_loaded_image_id"]:
            return False, {
                "stage": "docker_inspect",
                "role": role,
                "returncode": int(inspected.returncode),
                "expected_loaded_image_id": item["expected_loaded_image_id"],
                "observed_image_id_sha256": hashlib.sha256(
                    observed_id.encode("utf-8")
                ).hexdigest(),
            }
        results.append(
            {
                "sequence": item["sequence"],
                "role": role,
                "archive_sha256": item["archive_sha256"],
                "loaded_image_id": observed_id,
                "offline_compose_override": item["offline_compose_override"],
                "load_stdout_sha256": hashlib.sha256(loaded_stdout).hexdigest(),
                "load_stderr_sha256": hashlib.sha256(loaded_stderr).hexdigest(),
            }
        )
    summary: dict[str, Any] = {
        "schema_version": "yuqei.docker-image-import-result.v1",
        "status": "verified",
        "images": results,
    }
    summary["summary_sha256"] = canonical_sha256(summary)
    return True, summary


def _validate_post_apply_result(
    value: object,
    *,
    plan: Mapping[str, Any],
) -> dict[str, Any]:
    if not isinstance(value, dict):
        raise ReleasePackageError("Post-apply checker must return an object.")
    _verify_embedded_hash(value, "post_check_payload_sha256", label="post-apply check")
    expected = {
        "schema_version": POST_APPLY_SCHEMA,
        "status": "verified",
        "deployment_scope": DEPLOYMENT_SCOPE,
        "operation": plan["operation"],
        "target_release_id": plan["target_release_id"],
        "deployed_service_set": plan["deployed_service_set"],
        "daemon_identity_sha256": plan["docker_daemon_identity"]["identity_sha256"],
    }
    for key, expected_value in expected.items():
        if value.get(key) != expected_value:
            raise ReleasePackageError(f"Post-apply check has invalid {key}.")
    return dict(value)


def _receipt(
    *,
    plan: Mapping[str, Any],
    snapshot: SnapshotInputs,
    idempotency_key: str,
    command_argv: list[str],
    runtime_overrides: Mapping[str, str],
    image_import_result: Mapping[str, Any],
    post_apply_check: Mapping[str, Any],
    snapshot_offline_gate: Mapping[str, Any],
    returncode: int,
    stdout: bytes,
    stderr: bytes,
    recorded_at: datetime,
) -> dict[str, Any]:
    if recorded_at.tzinfo is None:
        raise ReleasePackageError("Receipt time must include a timezone.")
    payload: dict[str, Any] = {
        "schema_version": RECEIPT_SCHEMA,
        "operation": plan["operation"],
        "status": "backend_verified",
        "execution_mode": "test_injected_runner",
        "deployment_scope": plan["deployment_scope"],
        "deployed_service_set": plan["deployed_service_set"],
        "separately_deployed_artifacts": plan["separately_deployed_artifacts"],
        "identity_bound_image_keys": plan["identity_bound_image_keys"],
        "release_completion_claimed": False,
        "release_id": plan["release_id"],
        "target_release_id": plan["target_release_id"],
        "idempotency_key": idempotency_key,
        "recorded_at": recorded_at.astimezone(UTC).isoformat().replace("+00:00", "Z"),
        "plan_sha256": plan["plan_sha256"],
        "package_manifest_sha256": plan["package_manifest_sha256"],
        "release_manifest_sha256": plan["release_manifest_sha256"],
        "checksum_manifest_sha256": plan["checksum_manifest_sha256"],
        "image_identity_sha256": plan["image_identity_sha256"],
        "installable_manifest_sha256": plan["installable_manifest_sha256"],
        "trusted_build_receipt_sha256": plan["trusted_build_receipt_sha256"],
        "backup_proof_sha256": plan["backup_proof_sha256"],
        "compatibility_proof_sha256": plan["compatibility_proof_sha256"],
        "docker_daemon_identity": plan["docker_daemon_identity"],
        "production_preflight_summary_sha256": plan["production_preflight"][
            "summary_sha256"
        ],
        "compose_merge_gate_summary_sha256": plan["compose_merge_gate"]["summary_sha256"],
        "offline_execution_gate_summary_sha256": snapshot_offline_gate["summary_sha256"],
        "snapshot_payload_sha256": snapshot.manifest["snapshot_payload_sha256"],
        "tls_material_binding_sha256": plan["tls_material_binding_sha256"],
        "execution_env_sha256": snapshot.manifest["execution_env_sha256"],
        "docker_image_import_result": dict(image_import_result),
        "post_apply_check": dict(post_apply_check),
        "offline": True,
        "pull_policy": "never",
        "build_allowed": False,
        "database_downgrade_performed": False,
        "edge_or_dns_change_performed": False,
        "environment_overrides": dict(runtime_overrides),
        "command_argv": command_argv,
        "result": {
            "returncode": returncode,
            "stdout_sha256": hashlib.sha256(stdout).hexdigest(),
            "stderr_sha256": hashlib.sha256(stderr).hexdigest(),
        },
    }
    payload["receipt_payload_sha256"] = canonical_sha256(payload)
    return payload


def _require_apply_authorization(
    *,
    operation: str,
    confirmation: str,
    idempotency_key: str | None,
) -> str:
    expected = confirmation_for(operation)
    if confirmation != expected:
        raise ReleasePackageError(f"Apply confirmation must exactly equal: {expected}")
    if idempotency_key is None or not IDEMPOTENCY_RE.fullmatch(idempotency_key):
        raise ReleasePackageError("--idempotency-key is required and must use the safe identifier syntax.")
    return idempotency_key


def execute(
    *,
    package_dir: Path,
    expected_package_sha256: str,
    operation: str = "verify",
    env_file: Path | None = None,
    apply: bool = False,
    confirmation: str = "",
    idempotency_key: str | None = None,
    state_dir: Path = Path("/var/lib/yuqei/release-package"),
    repo_root: Path = REPO_ROOT,
    runner: Callable[..., Any] | None = None,
    inspection_runner: Callable[..., Any] = subprocess.run,
    post_apply_checker: Callable[..., Any] | None = None,
    now: datetime | None = None,
) -> ExecutionOutcome:
    package = validate_release_package(
        package_dir,
        expected_package_sha256=expected_package_sha256,
        repo_root=repo_root,
    )
    if operation == "verify":
        if apply:
            raise ReleasePackageError("The verify operation is read-only and does not accept --apply.")
        return ExecutionOutcome(verification_payload(package))
    if operation not in CONFIRMATIONS:
        raise ReleasePackageError("Operation must be verify, install, rollback, or forward.")
    if env_file is None:
        raise ReleasePackageError("--env-file is required for install, rollback, and forward plans.")
    safe_key: str | None = None
    if apply:
        safe_key = _require_apply_authorization(
            operation=operation,
            confirmation=confirmation,
            idempotency_key=idempotency_key,
        )
        if runner is None:
            blockers = ", ".join(_real_apply_blockers(operation))
            raise ReleasePackageError(
                "Real apply is fail-closed until all release blockers are resolved: "
                + blockers
            )
        if post_apply_checker is None:
            raise ReleasePackageError(
                "Injected apply tests require the fixed post-apply checker interface."
            )
        if operation == "rollback":
            raise ReleasePackageError(
                "Rollback apply is fail-closed because no approved rollback Docker import "
                "plan exists."
            )
    plan = build_plan(
        package,
        operation=operation,
        env_file=env_file,
        repo_root=repo_root,
        inspection_runner=inspection_runner,
    )
    if not apply:
        return ExecutionOutcome(plan)
    assert safe_key is not None  # Authorization above is required for every write path.

    state = _validate_state_target(state_dir, package_root=package.root)
    _prepare_secure_state_root(state, require_root_owner=False)
    with _exclusive_lock(state):
        locked_package = validate_release_package(
            package_dir,
            expected_package_sha256=expected_package_sha256,
            repo_root=repo_root,
        )
        locked_plan = build_plan(
            locked_package,
            operation=operation,
            env_file=env_file,
            repo_root=repo_root,
            inspection_runner=inspection_runner,
        )
        if locked_plan["plan_sha256"] != plan["plan_sha256"]:
            raise ReleasePackageError("Release package, env, or execution plan changed while acquiring the lock.")

        receipt_path = _receipt_path(state, safe_key)
        pending_path = _pending_path(state, safe_key)
        receipts_dir = receipt_path.parent
        pending_dir = pending_path.parent
        if receipts_dir.exists() and (receipts_dir.is_symlink() or not receipts_dir.is_dir()):
            raise ReleasePackageError("Receipt directory is unsafe.")
        if pending_dir.exists() and (pending_dir.is_symlink() or not pending_dir.is_dir()):
            raise ReleasePackageError("Pending journal directory is unsafe.")
        receipts_dir.mkdir(parents=True, exist_ok=True, mode=0o700)
        pending_dir.mkdir(parents=True, exist_ok=True, mode=0o700)
        _reject_symlink_components(receipts_dir, label="receipt directory")
        _reject_symlink_components(pending_dir, label="pending journal directory")
        if pending_path.exists() or pending_path.is_symlink():
            pending = _read_pending_journal(pending_path)
            if pending.get("idempotency_key") != safe_key:
                raise ReleasePackageError("Pending reconciliation journal is bound to another key.")
            return ExecutionOutcome(pending, exit_code=1, replayed=True)
        if receipt_path.exists() or receipt_path.is_symlink():
            existing = _read_existing_receipt(receipt_path)
            if existing.get("idempotency_key") != safe_key or existing.get("plan_sha256") != plan["plan_sha256"]:
                raise ReleasePackageError("Idempotency key is already bound to a different release operation.")
            return ExecutionOutcome(existing, exit_code=0, replayed=True)

        snapshot = _snapshot_inputs(
            state_dir=state,
            idempotency_key=safe_key,
            plan=plan,
            package=locked_package,
            repo_root=_validated_directory(repo_root, label="repository root"),
        )
        source_snapshot_evidence = _collect_runtime_evidence(
            env_file=snapshot.source_env_file,
            repo_root=snapshot.root,
            inspection_runner=inspection_runner,
        )
        if source_snapshot_evidence.env_sha256 != plan["env_file_sha256"]:
            raise ReleasePackageError("Source production env snapshot SHA-256 does not match the plan.")
        if source_snapshot_evidence.preflight_summary != plan["production_preflight"]:
            raise ReleasePackageError("Source production preflight summary changed.")
        if source_snapshot_evidence.daemon_identity != plan["docker_daemon_identity"]:
            raise ReleasePackageError("Docker daemon identity changed before apply.")

        execution_snapshot_evidence = _collect_runtime_evidence(
            env_file=snapshot.env_file,
            repo_root=snapshot.root,
            inspection_runner=inspection_runner,
        )
        if execution_snapshot_evidence.env_sha256 != snapshot.manifest["execution_env_sha256"]:
            raise ReleasePackageError("Execution env snapshot SHA-256 is inconsistent.")
        if execution_snapshot_evidence.preflight_summary != plan["production_preflight"]:
            raise ReleasePackageError("Execution TLS snapshot does not match the production preflight.")
        if execution_snapshot_evidence.daemon_identity != plan["docker_daemon_identity"]:
            raise ReleasePackageError("Docker daemon identity changed while preparing execution.")
        _verify_execution_snapshot(snapshot, plan)

        snapshot_offline_gate = _run_offline_execution_gate(
            env_file=snapshot.env_file,
            repo_root=snapshot.root,
            compose_files=snapshot.compose_files,
            project_name=plan["compose_project_name"],
            offline_application_overrides=plan["offline_application_overrides"],
            inspection_runner=inspection_runner,
        )
        runtime_overrides = dict(plan["environment_overrides"])
        runtime_overrides.update(plan["offline_application_overrides"])
        runtime_environment = _fixed_runtime_environment(runtime_overrides)
        command = _compose_command(
            operation=operation,
            project_name=plan["compose_project_name"],
            env_file=snapshot.env_file,
            repo_root=snapshot.root,
            compose_files=snapshot.compose_files,
        )
        _assert_safe_compose_command(command, operation=operation)

        journal = _pending_journal(
            plan=plan,
            idempotency_key=safe_key,
            snapshot=snapshot,
            command=command,
            recorded_at=now or datetime.now(UTC),
        )
        _write_exclusive_canonical_json(pending_path, journal)

        imported, import_result = _run_application_image_imports(
            plan=plan,
            snapshot=snapshot,
            runner=runner,
        )
        if not imported:
            return ExecutionOutcome(journal, exit_code=1)
        _verify_execution_snapshot(snapshot, plan)
        try:
            completed = runner(
                command,
                check=False,
                capture_output=True,
                shell=False,
                env=runtime_environment,
                timeout=1800,
            )
            returncode = int(completed.returncode)
            stdout = _output_bytes(getattr(completed, "stdout", b""))
            stderr = _output_bytes(getattr(completed, "stderr", b""))
        except (OSError, subprocess.TimeoutExpired) as exc:
            returncode = 127
            stdout = b""
            stderr = type(exc).__name__.encode("ascii", "replace")
        if returncode != 0:
            return ExecutionOutcome(journal, exit_code=1)

        post_daemon_identity = _docker_daemon_identity(
            runner=inspection_runner,
            repo_root=snapshot.root,
        )
        if post_daemon_identity != plan["docker_daemon_identity"]:
            return ExecutionOutcome(journal, exit_code=1)
        try:
            post_apply = _validate_post_apply_result(
                post_apply_checker(
                    plan=dict(plan),
                    command_argv=list(command),
                    runtime_environment=dict(runtime_environment),
                    daemon_identity=dict(post_daemon_identity),
                ),
                plan=plan,
            )
        except Exception:
            return ExecutionOutcome(journal, exit_code=1)
        receipt = _receipt(
            plan=plan,
            snapshot=snapshot,
            idempotency_key=safe_key,
            command_argv=command,
            runtime_overrides=runtime_overrides,
            image_import_result=import_result,
            post_apply_check=post_apply,
            snapshot_offline_gate=snapshot_offline_gate,
            returncode=returncode,
            stdout=stdout,
            stderr=stderr,
            recorded_at=now or datetime.now(UTC),
        )
        _write_exclusive_canonical_json(receipt_path, receipt)
        try:
            pending_path.unlink()
            _fsync_directory(pending_dir)
        except OSError as exc:
            raise ReleasePackageError(
                "Backend was verified but pending reconciliation journal could not be cleared."
            ) from exc
        return ExecutionOutcome(receipt, exit_code=0)


def build_parser() -> argparse.ArgumentParser:
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument("--package-dir", type=Path, required=True)
    parser.add_argument(
        "--expected-package-sha256",
        required=True,
        help="Externally approved raw SHA-256 of release-package.json.",
    )
    parser.add_argument(
        "--operation",
        choices=("verify", "install", "rollback", "forward"),
        default="verify",
        help="Defaults to read-only package verification; non-verify operations are plans without --apply.",
    )
    parser.add_argument("--env-file", type=Path, help="Absolute protected production env file.")
    parser.add_argument("--state-dir", type=Path, default=Path("/var/lib/yuqei/release-package"))
    parser.add_argument(
        "--apply",
        action="store_true",
        help=(
            "Request execution after authorization and all feature gates; "
            "the current production gate is closed."
        ),
    )
    parser.add_argument(
        "--confirmation",
        default="",
        help="Exact operation confirmation; see the dry-run plan or --help.",
    )
    parser.add_argument("--idempotency-key", default=None)
    return parser


def main(argv: list[str] | None = None) -> int:
    args = build_parser().parse_args(argv)
    try:
        outcome = execute(
            package_dir=args.package_dir,
            expected_package_sha256=args.expected_package_sha256,
            operation=args.operation,
            env_file=args.env_file,
            apply=args.apply,
            confirmation=args.confirmation,
            idempotency_key=args.idempotency_key,
            state_dir=args.state_dir,
        )
    except (ReleasePackageError, OSError) as exc:
        print(
            canonical_json_bytes(
                {"schema_version": RECEIPT_SCHEMA, "status": "rejected", "error": str(exc)}
            ).decode("utf-8"),
            file=sys.stderr,
        )
        return 2
    print(canonical_json_bytes(outcome.payload).decode("utf-8"))
    return outcome.exit_code


if __name__ == "__main__":
    raise SystemExit(main())