#!/usr/bin/env python3
"""Build and validate the immutable GCP leoclean no-send service package."""

from __future__ import annotations

import argparse
import ctypes
import errno
import fcntl
import hashlib
import json
import os
import re
import shutil
import stat
import subprocess
import sys
import tomllib
import uuid
from collections.abc import Mapping
from pathlib import Path
from typing import Any

CONTRACT_ROOT = (
    Path(__file__).resolve().parents[1]
    if Path(__file__).resolve().parent.name == "ops"
    else Path(__file__).resolve().parent
)
if str(CONTRACT_ROOT) not in sys.path:
    sys.path.insert(0, str(CONTRACT_ROOT))

from scripts import leo_identity_manifest as strict_identity_manifest  # noqa: E402

IMAGE_INPUT_SCHEMA = "livingip.leocleanNoSendImageInput.v2"
IMAGE_INSPECTION_SCHEMA = "livingip.leocleanNoSendImageInspection.v1"
BUILD_PUSH_RECEIPT_SCHEMA = "livingip.leocleanNoSendBuildPushReceipt.v1"
RELEASE_SCHEMA = "livingip.leocleanNoSendRelease.v3"
ARTIFACT_MANIFEST_SCHEMA = "livingip.leocleanNoSendArtifactManifest.v1"
ARTIFACT_RECEIPT_SCHEMA = "livingip.leocleanNoSendArtifactVerification.v1"
IDENTITY_MANIFEST_SCHEMA = "livingip.leoIdentityManifest.v1"
IDENTITY_LOCK_SCHEMA = "livingip.leoIdentityProfileLock.v1"
IDENTITY_RUNTIME_CONTRACT_SCHEMA = "livingip.leocleanNoSendIdentityRuntimeContract.v1"
COMPILED_IDENTITY_SCHEMA = "livingip.leocleanNoSendCompiledIdentity.v1"
PERMISSION_RECEIPT_SCHEMA = "livingip.leocleanNoSendPermissionReceipt.v1"

PROJECT = "teleo-501523"
ZONE = "europe-west6-a"
INSTANCE = "teleo-staging-1"
SERVICE = "leoclean-gcp-nosend.service"
SERVICE_ACCOUNT = "sa-teleo-staging-vm@teleo-501523.iam.gserviceaccount.com"
PLATFORM = "linux/amd64"
IMAGE_REPOSITORY = "europe-west6-docker.pkg.dev/teleo-501523/teleo/leoclean-nosend-staging"
CONTAINER_NAME = "livingip-leoclean-nosend"
LOCAL_BUILD_REPOSITORY = "livingip-leoclean-nosend-build"

HERMES_COMMIT = "b2f477a30b3c05d0f383c543af98496ae8a96070"
PYTHON_VERSION = "3.11.9"
PYYAML_VERSION = "6.0.3"
UV_VERSION = "0.9.30"
POSTGRES_VERSION = "16.14"
MODEL_PROVIDER = "openrouter"
DEFAULT_MODEL = "anthropic/claude-sonnet-4-6"
PYTHON_IMAGE = (
    "docker.io/library/python:3.11.9-slim-bookworm@"
    "sha256:8fb099199b9f2d70342674bd9dbccd3ed03a258f26bbd1d556822c6dfc60c317"
)
UV_IMAGE = "ghcr.io/astral-sh/uv:0.9.30@sha256:538e0b39736e7feae937a65983e49d2ab75e1559d35041f9878b7b7e51de91e4"
POSTGRES_IMAGE = (
    "docker.io/library/postgres:16.14-bookworm@sha256:92620daddcd947f8d5ab5ba66e848702fe443d87fed30c4cea8e389fd78dfc55"
)
CA_SHA256 = "80701e768f0e1f6b9d621aa0b53f6e851daaa276c6d9a8e51a300fbc015539cb"
CA_DESTINATION = "/usr/local/libexec/livingip/leoclean-kb/cloudsql-server-ca.pem"
RUNTIME_ROOT = "/opt/livingip/leoclean-nosend"
PROFILE_ROOT = "/run/leoclean-profile"
RUNTIME_UID = 65532
RUNTIME_GID = 65532
MAX_DOCKER_INSPECTION_BYTES = 1024 * 1024
MAX_DOCKER_COMMAND_OUTPUT_BYTES = 1024 * 1024
MAX_BUILD_PUSH_RECEIPT_BYTES = 256 * 1024
IMAGE_RUNTIME_CONFIG = {
    "user": None,
    "environment": [
        "PATH=/usr/local/sbin:/usr/local/bin:/usr/sbin:/usr/bin:/sbin:/bin:/usr/lib/postgresql/16/bin",
        "GOSU_VERSION=1.19",
        "LANG=en_US.utf8",
        "PG_MAJOR=16",
        "PG_VERSION=16.14-1.pgdg12+1",
        "PGDATA=/var/lib/postgresql/data",
        "PYTHONDONTWRITEBYTECODE=1",
        "PYTHONNOUSERSITE=1",
        "PYTHONSAFEPATH=1",
    ],
    "working_dir": "",
    "exposed_ports": {"5432/tcp": {}},
    "volumes": {"/var/lib/postgresql/data": {}},
    "shell": None,
    "on_build": None,
    "args_escaped": False,
    "stop_timeout": None,
    "tty": False,
    "open_stdin": False,
    "stdin_once": False,
    "attach_stdin": False,
    "attach_stdout": False,
    "attach_stderr": False,
    "network_disabled": False,
    "hostname": "",
    "domainname": "",
    "mac_address": "",
    "entrypoint": [
        "/opt/livingip/leoclean-nosend/venv/bin/python",
        "/opt/livingip/leoclean-nosend/entrypoint.py",
    ],
    "cmd": ["service"],
    "healthcheck": {
        "test": [
            "CMD",
            "/opt/livingip/leoclean-nosend/venv/bin/python",
            "/opt/livingip/leoclean-nosend/entrypoint.py",
            "health",
        ],
        "interval_ns": 15_000_000_000,
        "timeout_ns": 10_000_000_000,
        "start_period_ns": 30_000_000_000,
        "start_interval_ns": 0,
        "retries": 4,
    },
    "stop_signal": "SIGTERM",
}

HEX_40 = re.compile(r"^[0-9a-f]{40}$")
HEX_64 = re.compile(r"^[0-9a-f]{64}$")
IMAGE_REFERENCE = re.compile(rf"^{re.escape(IMAGE_REPOSITORY)}@sha256:([0-9a-f]{{64}})$")
FORBIDDEN_DESCRIPTOR_MARKERS = (
    "77.42.65.182",
    "gcp-teleo-pgvector-standby-postgres-password",
    "leoclean-gateway.service",
    "sa-teleo-prod-vm",
    "telegram_bot_token",
    "teleo-prod-1",
)
PROPOSAL_FUNCTION = "kb_stage.stage_leoclean_proposal(text,text,text,text,jsonb)"
IDENTITY_FILES = ("SOUL.md", "identity.json", "identity-lock.json", "identity-manifest.json")
RUNTIME_IDENTITY_FILES = ("SOUL.md", "identity-lock.json", "identity-manifest.json")
IDENTITY_CONTRACT_SOURCES = (
    "scripts/leo_behavior_manifest.py",
    "scripts/leo_identity_manifest.py",
)
PACKAGE_SOURCE_PATHS = (
    "docker/leoclean-nosend/Dockerfile",
    "docker/leoclean-nosend/entrypoint.py",
    "ops/gcp-teleo-pgvector-standby-server-ca.pem",
    "ops/gcp_leoclean_nosend_package.py",
    *IDENTITY_CONTRACT_SOURCES,
)
NO_SEND_TOOL_SURFACE = strict_identity_manifest.LEOCLEAN_NO_SEND_TOOL_SURFACE
SYNTHETIC_IDENTITY_SOURCE_LABELS = (
    "database fingerprint",
    "constitution source",
    "database identity source",
)


class PackageError(RuntimeError):
    """The image or service package does not satisfy the staging contract."""


def _require(condition: bool, message: str) -> None:
    if not condition:
        raise PackageError(message)


def pid_one_command_line_is_exact(cmdline: list[bytes], expected_native: list[bytes]) -> bool:
    """Accept the pinned native argv and exact x86_64 binfmt wrappers only."""
    if not expected_native or expected_native[-1] != b"":
        return False
    expected_binfmt = [expected_native[0], *expected_native]
    accepted = (
        expected_native,
        expected_binfmt,
        [b"/usr/bin/qemu-x86_64", *expected_binfmt],
        [b"/usr/bin/qemu-x86_64-static", *expected_binfmt],
    )
    return cmdline in accepted


def sha256_file(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 canonical_sha256(value: Any) -> str:
    try:
        encoded = json.dumps(
            value,
            allow_nan=False,
            ensure_ascii=True,
            separators=(",", ":"),
            sort_keys=True,
        ).encode("utf-8")
    except (TypeError, ValueError) as exc:
        raise PackageError("cannot canonically encode non-finite or unsupported JSON") from exc
    return hashlib.sha256(encoded).hexdigest()


def _strict_json_loads(encoded: str, label: str) -> Any:
    def reject_duplicate_keys(pairs: list[tuple[str, Any]]) -> dict[str, Any]:
        value: dict[str, Any] = {}
        for key, item in pairs:
            if key in value:
                raise ValueError("duplicate JSON key")
            value[key] = item
        return value

    def reject_constant(_value: str) -> Any:
        raise ValueError("invalid JSON constant")

    try:
        return json.loads(
            encoded,
            object_pairs_hook=reject_duplicate_keys,
            parse_constant=reject_constant,
        )
    except (json.JSONDecodeError, ValueError) as exc:
        raise PackageError(f"{label} is invalid JSON") from exc


def load_json(path: Path, label: str) -> dict[str, Any]:
    try:
        encoded = path.read_text(encoding="utf-8")
    except (OSError, UnicodeError) as exc:
        raise PackageError(f"cannot read {label}: {type(exc).__name__}") from exc
    value = _strict_json_loads(encoded, label)
    _require(isinstance(value, dict), f"{label} must be a JSON object")
    return value


def write_json(path: Path, value: Mapping[str, Any], mode: int = 0o644) -> None:
    try:
        encoded = json.dumps(value, allow_nan=False, indent=2, sort_keys=True) + "\n"
    except (TypeError, ValueError) as exc:
        raise PackageError("cannot encode non-finite or unsupported JSON") from exc
    path.write_text(encoded, encoding="utf-8")
    path.chmod(mode)


def _is_sha256(value: Any) -> bool:
    return isinstance(value, str) and bool(HEX_64.fullmatch(value))


def _is_revision(value: Any) -> bool:
    return isinstance(value, str) and bool(HEX_40.fullmatch(value))


def _safe_tree(root: Path, label: str) -> None:
    _require(root.is_dir() and not root.is_symlink(), f"{label} root is missing or unsafe")
    for path in root.rglob("*"):
        _require(not path.is_symlink(), f"{label} contains a symlink")
        _require(path.is_dir() or path.is_file(), f"{label} contains an unsupported entry")


def _parent_directory_paths(file_paths: set[str]) -> set[str]:
    directories: set[str] = set()
    for relative in file_paths:
        path = Path(relative)
        _require(not path.is_absolute() and path.parts and ".." not in path.parts, "tree file path is unsafe")
        parent = path.parent
        while parent != Path("."):
            directories.add(parent.as_posix())
            parent = parent.parent
    return directories


def _normalize_prepared_directory_modes(root: Path) -> None:
    root.chmod(0o700)
    for path in root.rglob("*"):
        if not path.is_dir():
            continue
        relative = path.relative_to(root).as_posix()
        path.chmod(0o700 if relative == "identity" else 0o755)


def tree_manifest(root: Path, *, excluded: frozenset[str] = frozenset()) -> dict[str, Any]:
    _safe_tree(root, "tree")
    files: list[dict[str, Any]] = []
    for path in sorted(root.rglob("*"), key=lambda item: item.relative_to(root).as_posix()):
        relative = path.relative_to(root).as_posix()
        if relative in excluded or path.is_dir():
            continue
        mode = stat.S_IMODE(path.stat().st_mode)
        files.append(
            {
                "path": relative,
                "bytes": path.stat().st_size,
                "mode": f"0o{mode:03o}",
                "sha256": sha256_file(path),
            }
        )
    stable = {"files": files, "file_count": len(files), "total_bytes": sum(item["bytes"] for item in files)}
    return {**stable, "sha256": canonical_sha256(stable)}


def identity_bundle_manifest(root: Path) -> dict[str, Any]:
    _safe_tree(root, "identity bundle")
    actual = sorted(path.name for path in root.iterdir())
    _require(actual == sorted(IDENTITY_FILES), "identity bundle entry set is not exact")
    files = []
    for name in IDENTITY_FILES:
        path = root / name
        _require(path.is_file() and not path.is_symlink(), f"identity file is unsafe: {name}")
        _require(path.stat().st_size <= 1024 * 1024, f"identity file is too large: {name}")
        _require(not stat.S_IMODE(path.stat().st_mode) & 0o022, f"identity file is group/world writable: {name}")
        files.append({"path": name, "bytes": path.stat().st_size, "sha256": sha256_file(path)})
    stable = {"files": files, "file_count": len(files), "total_bytes": sum(item["bytes"] for item in files)}
    return {**stable, "sha256": canonical_sha256(stable)}


def no_send_permission_policy() -> dict[str, Any]:
    return {
        "schema": PERMISSION_RECEIPT_SCHEMA,
        "messaging_send": {
            "capability": "absent",
            "transport_authority": "absent",
            "gateway_adapters": "none",
            "send_message_tool": "absent",
        },
        "general_network_egress": {
            "container_mode": "bridge",
            "denial_enforced": False,
            "denial_tested": False,
        },
        "database": {
            "direct_canonical_writes": "denied",
            "direct_stage_table_writes": "denied",
            "proposal_staging": "function_only",
            "proposal_function": PROPOSAL_FUNCTION,
            "live_effective_permissions_proven": False,
        },
        "allowed_tools": list(NO_SEND_TOOL_SURFACE["tools"]),
    }


def _runtime_surface(value: Any, label: str) -> dict[str, Any]:
    _require(isinstance(value, dict), f"{label} is missing")
    _require(value == NO_SEND_TOOL_SURFACE, f"{label} is not the exact sealed no-send tool surface")
    return dict(value)


def _validate_runtime_probe(
    receipt: dict[str, Any],
    *,
    expected_config_sha256: str | None = None,
) -> tuple[dict[str, Any], dict[str, Any]]:
    probe = receipt.get("runtime_probe")
    _require(
        isinstance(probe, dict)
        and probe.get("schema") == "livingip.leocleanNoSendRuntimeProbe.v1"
        and probe.get("status") == "pass",
        "runtime probe is missing or did not pass",
    )
    probe_config_sha256 = probe.get("config_sha256")
    _require(_is_sha256(probe_config_sha256), "runtime probe config hash is invalid")
    _require(
        probe_config_sha256 == receipt.get("config_sha256"),
        "runtime probe config differs from the runtime receipt",
    )
    if expected_config_sha256 is not None:
        _require(_is_sha256(expected_config_sha256), "expected runtime config hash is invalid")
        _require(
            probe_config_sha256 == expected_config_sha256,
            "runtime probe config differs from the packaged config",
        )
    _require(probe.get("gateway_adapter_count") == 0, "runtime probe found a gateway adapter")
    _require(probe.get("connected_platforms") == [], "runtime probe found a connected platform")
    _require(probe.get("python_version") == PYTHON_VERSION, "runtime probe Python version drifted")
    _require(probe.get("model") == DEFAULT_MODEL, "runtime probe model drifted")
    _require(
        probe.get("provider_credential_names_present") == ["OPENROUTER_API_KEY"]
        and probe.get("provider_credential_sentinel_redaction") == "pass",
        "runtime probe provider-credential boundary drifted",
    )
    effective_policy = probe.get("effective_policy")
    _require(
        effective_policy
        == {
            "cheap_model_route": None,
            "memory_enabled": False,
            "memory_nudge_interval": 0,
            "skill_creation_nudge_interval": 0,
            "smart_model_routing": False,
            "user_profile_enabled": False,
        },
        "runtime probe learning/routing policy drifted",
    )
    surface = _runtime_surface(probe.get("tool_surface"), "runtime probe tool surface")
    lifecycle = probe.get("lifecycle")
    _require(
        isinstance(lifecycle, dict) and lifecycle.get("started") is True and lifecycle.get("stopped") is True,
        "runtime probe lifecycle was not exercised",
    )
    for phase in ("post_start_registry", "post_discovery_registry"):
        _require(_runtime_surface(lifecycle.get(phase), f"runtime probe {phase}") == surface, f"{phase} drifted")
    return effective_policy, surface


def _locked_python_runtime(artifact: Path, *, expected_uv_lock_sha256: str) -> dict[str, Any]:
    uv_lock_path = artifact / "hermes-agent" / "uv.lock"
    _require(uv_lock_path.is_file() and not uv_lock_path.is_symlink(), "artifact uv.lock is missing or unsafe")
    _require(sha256_file(uv_lock_path) == expected_uv_lock_sha256, "artifact uv.lock differs from the runtime lock")
    try:
        lock = tomllib.loads(uv_lock_path.read_text(encoding="utf-8"))
    except (OSError, UnicodeDecodeError, tomllib.TOMLDecodeError) as exc:
        raise PackageError(f"cannot read artifact uv.lock: {type(exc).__name__}") from exc
    packages = lock.get("package")
    _require(isinstance(packages, list), "artifact uv.lock package set is missing")
    matches = [item for item in packages if isinstance(item, dict) and str(item.get("name", "")).casefold() == "pyyaml"]
    _require(len(matches) == 1 and matches[0].get("version") == PYYAML_VERSION, "artifact PyYAML lock drifted")
    distributions = {"PyYAML": PYYAML_VERSION}
    return {
        "implementation": "CPython",
        "python_version": PYTHON_VERSION,
        "abi_tag": "cpython-311",
        "runtime_distributions": distributions,
        "runtime_distributions_sha256": canonical_sha256(distributions),
    }


def _identity_contract_source_bindings(source_root: Path) -> dict[str, str]:
    bindings: dict[str, str] = {}
    for relative in IDENTITY_CONTRACT_SOURCES:
        path = _safe_bound_file(source_root, relative, label=f"identity contract source: {relative}")
        bindings[relative] = sha256_file(path)
    return bindings


def _identity_runtime_payload(contract: Mapping[str, Any]) -> dict[str, Any]:
    runtime = contract.get("runtime") or {}
    return {
        "schema": contract.get("schema"),
        "runtime": {key: value for key, value in runtime.items() if key != "identity_runtime_sha256"},
        "model": contract.get("model"),
        "skills": contract.get("skills"),
        "permissions": contract.get("permissions"),
        "behavior_inputs": contract.get("behavior_inputs"),
    }


def validate_expected_identity_contract(value: dict[str, Any], *, expected_revision: str) -> None:
    _require(_is_revision(expected_revision), "expected identity revision is invalid")
    _require(
        set(value) == {"schema", "runtime", "model", "skills", "permissions", "behavior_inputs", "contract_sha256"},
        "expected identity contract fields are not exact",
    )
    _require(value.get("schema") == IDENTITY_RUNTIME_CONTRACT_SCHEMA, "expected identity contract schema is invalid")
    stable = {key: item for key, item in value.items() if key != "contract_sha256"}
    _require(value.get("contract_sha256") == canonical_sha256(stable), "expected identity contract hash drifted")
    runtime = value.get("runtime")
    _require(
        isinstance(runtime, dict)
        and set(runtime)
        == {
            "identity_runtime_sha256",
            "hermes_git_head",
            "hermes_source_sha256",
            "teleo_git_head",
            "teleo_source_sha256",
            "python",
        },
        "expected identity runtime fields are not exact",
    )
    _require(
        runtime.get("identity_runtime_sha256") == canonical_sha256(_identity_runtime_payload(value)),
        "expected identity runtime hash drifted",
    )
    _require(runtime.get("hermes_git_head") == HERMES_COMMIT, "expected identity Hermes revision drifted")
    _require(runtime.get("teleo_git_head") == expected_revision, "expected identity Teleo revision drifted")
    for field in ("identity_runtime_sha256", "hermes_source_sha256", "teleo_source_sha256"):
        _require(_is_sha256(runtime.get(field)), f"expected identity runtime {field} is invalid")
    distributions = {"PyYAML": PYYAML_VERSION}
    _require(
        runtime.get("python")
        == {
            "implementation": "CPython",
            "python_version": PYTHON_VERSION,
            "abi_tag": "cpython-311",
            "runtime_distributions": distributions,
            "runtime_distributions_sha256": canonical_sha256(distributions),
        },
        "expected identity Python runtime drifted",
    )
    model = value.get("model")
    _require(
        isinstance(model, dict)
        and set(model)
        == {
            "provider",
            "default_model",
            "smart_routing",
            "smart_routing_model",
            "gateway_smart_model_routing",
            "model_section_sha256",
            "hosted_weights",
            "actual_model_required_in_turn_receipt",
        },
        "expected identity model fields are not exact",
    )
    _require(
        model.get("provider") == MODEL_PROVIDER
        and model.get("default_model") == DEFAULT_MODEL
        and model.get("smart_routing") is False
        and model.get("smart_routing_model") is None
        and model.get("gateway_smart_model_routing") is False
        and _is_sha256(model.get("model_section_sha256"))
        and model.get("hosted_weights") == "external_provider_managed_not_locally_hashable"
        and model.get("actual_model_required_in_turn_receipt") is True,
        "expected identity model binding drifted",
    )
    skills = value.get("skills")
    _require(
        isinstance(skills, dict)
        and set(skills) == {"content_sha256", "file_count"}
        and _is_sha256(skills.get("content_sha256"))
        and isinstance(skills.get("file_count"), int)
        and not isinstance(skills.get("file_count"), bool)
        and skills["file_count"] > 0,
        "expected identity skills binding is invalid",
    )
    permissions = value.get("permissions")
    _require(
        isinstance(permissions, dict)
        and set(permissions)
        == {
            "identity_config_sha256",
            "gateway_permissions_sha256",
            "tool_permissions_sha256",
            "runtime_policy",
            "authorization_decision_required_in_runtime_receipt",
        },
        "expected identity permission fields are not exact",
    )
    for field in ("identity_config_sha256", "gateway_permissions_sha256", "tool_permissions_sha256"):
        _require(_is_sha256(permissions.get(field)), f"expected identity permission {field} is invalid")
    _require(
        permissions.get("runtime_policy") == no_send_permission_policy()
        and permissions.get("authorization_decision_required_in_runtime_receipt") is True,
        "expected identity permission policy drifted",
    )
    behavior_inputs = value.get("behavior_inputs")
    _require(
        isinstance(behavior_inputs, dict)
        and set(behavior_inputs)
        == {
            "artifact_sha256",
            "artifact_receipt_sha256",
            "config_sha256",
            "runtime_contract_sha256",
            "uv_lock_sha256",
            "identity_contract_sources",
            "effective_policy",
            "tool_surface",
        },
        "expected identity behavior inputs are not exact",
    )
    for field in (
        "artifact_sha256",
        "artifact_receipt_sha256",
        "config_sha256",
        "runtime_contract_sha256",
        "uv_lock_sha256",
    ):
        _require(_is_sha256(behavior_inputs.get(field)), f"expected identity behavior input {field} is invalid")
    source_bindings = behavior_inputs.get("identity_contract_sources")
    _require(
        isinstance(source_bindings, dict)
        and set(source_bindings) == set(IDENTITY_CONTRACT_SOURCES)
        and all(_is_sha256(item) for item in source_bindings.values())
        and runtime.get("teleo_source_sha256") == canonical_sha256(source_bindings),
        "expected identity contract source binding drifted",
    )
    _require(
        behavior_inputs.get("tool_surface") == NO_SEND_TOOL_SURFACE,
        "expected identity tool surface drifted",
    )
    _require(
        behavior_inputs.get("effective_policy")
        == {
            "cheap_model_route": None,
            "memory_enabled": False,
            "memory_nudge_interval": 0,
            "skill_creation_nudge_interval": 0,
            "smart_model_routing": False,
            "user_profile_enabled": False,
        },
        "expected identity effective policy drifted",
    )


def build_expected_identity_contract(
    *,
    artifact: Path,
    receipt_path: Path,
    runtime: Mapping[str, Any],
    source_revision: str,
    source_root: Path,
) -> dict[str, Any]:
    receipt = load_json(receipt_path, "artifact verification receipt")
    effective_policy, tool_surface = _validate_runtime_probe(
        receipt,
        expected_config_sha256=sha256_file(artifact / "profile-template" / "config.yaml"),
    )
    profile = artifact / "profile-template"
    config_path = profile / "config.yaml"
    config = strict_identity_manifest.behavior.safe_model_config(config_path)
    _require(config.get("status") == "observed", "packaged identity model config was not observed")
    _require(
        config.get("config_sha256") == receipt.get("config_sha256"), "packaged config differs from runtime receipt"
    )
    _require(
        config.get("provider") == MODEL_PROVIDER and config.get("default_model") == DEFAULT_MODEL,
        "packaged provider/default model drifted",
    )
    _require(
        config.get("gateway_smart_model_routing") is False
        and config.get("smart_routing") in (None, False)
        and config.get("smart_routing_model") is None,
        "packaged smart routing is not disabled",
    )
    runtime_contract_path = artifact / "runtime-contract.json"
    runtime_contract = load_json(runtime_contract_path, "artifact runtime contract")
    _require(runtime_contract.get("transport_authority") == "absent", "runtime transport authority is not absent")
    declared_tools = (runtime_contract.get("toolset") or {}).get("allowed_tools")
    _require(
        isinstance(declared_tools, list) and sorted(declared_tools) == tool_surface["tools"],
        "runtime contract and probed tool surface differ",
    )
    database = runtime_contract.get("database") or {}
    _require(
        database.get("direct_canonical_writes") == "denied"
        and database.get("direct_stage_table_writes") == "denied"
        and database.get("proposal_staging") == "function_only"
        and database.get("proposal_function") == PROPOSAL_FUNCTION,
        "runtime database proposal/write boundary drifted",
    )
    skills = tree_manifest(profile / "skills")
    _require(skills["file_count"] > 0, "packaged identity skills are empty")
    source_bindings = _identity_contract_source_bindings(source_root)
    permissions = {
        "identity_config_sha256": config["identity_config_sha256"],
        "gateway_permissions_sha256": config["gateway_permissions_sha256"],
        "tool_permissions_sha256": config["tool_permissions_sha256"],
        "runtime_policy": no_send_permission_policy(),
        "authorization_decision_required_in_runtime_receipt": True,
    }
    behavior_inputs = {
        "artifact_sha256": runtime["artifact_sha256"],
        "artifact_receipt_sha256": runtime["artifact_receipt_sha256"],
        "config_sha256": config["config_sha256"],
        "runtime_contract_sha256": sha256_file(runtime_contract_path),
        "uv_lock_sha256": runtime["uv_lock_sha256"],
        "identity_contract_sources": source_bindings,
        "effective_policy": effective_policy,
        "tool_surface": tool_surface,
    }
    runtime_binding: dict[str, Any] = {
        "hermes_git_head": runtime["hermes_commit"],
        "hermes_source_sha256": runtime["hermes_source_sha256"],
        "teleo_git_head": source_revision,
        "teleo_source_sha256": canonical_sha256(source_bindings),
        "python": _locked_python_runtime(artifact, expected_uv_lock_sha256=str(runtime["uv_lock_sha256"])),
    }
    core: dict[str, Any] = {
        "schema": IDENTITY_RUNTIME_CONTRACT_SCHEMA,
        "runtime": runtime_binding,
        "model": {
            "provider": MODEL_PROVIDER,
            "default_model": DEFAULT_MODEL,
            "smart_routing": False,
            "smart_routing_model": None,
            "gateway_smart_model_routing": False,
            "model_section_sha256": config["model_section_sha256"],
            "hosted_weights": "external_provider_managed_not_locally_hashable",
            "actual_model_required_in_turn_receipt": True,
        },
        "skills": {"content_sha256": skills["sha256"], "file_count": skills["file_count"]},
        "permissions": permissions,
        "behavior_inputs": behavior_inputs,
    }
    runtime_binding["identity_runtime_sha256"] = canonical_sha256(_identity_runtime_payload(core))
    value = {**core, "contract_sha256": canonical_sha256(core)}
    validate_expected_identity_contract(value, expected_revision=source_revision)
    return value


def _identity_source_paths(manifest: dict[str, Any]) -> tuple[str, str, str]:
    inputs = manifest.get("identity_inputs")
    database = inputs.get("canonical_database") if isinstance(inputs, dict) else None
    sources = inputs.get("identity_sources") if isinstance(inputs, dict) else None
    raw_paths = (
        (database or {}).get("source", {}).get("repo_path") if isinstance(database, dict) else None,
        (sources or {}).get("static_constitution", {}).get("repo_path") if isinstance(sources, dict) else None,
        (sources or {}).get("database_derived_identity", {}).get("repo_path") if isinstance(sources, dict) else None,
    )
    paths: list[str] = []
    for label, raw in zip(SYNTHETIC_IDENTITY_SOURCE_LABELS, raw_paths, strict=True):
        _require(isinstance(raw, str) and bool(raw), f"identity {label} path is missing")
        relative = Path(raw)
        _require(not relative.is_absolute() and ".." not in relative.parts, f"identity {label} path is unsafe")
        paths.append(relative.as_posix())
    _require(len(set(paths)) == len(paths), "identity source paths are not distinct")
    return tuple(paths)  # type: ignore[return-value]


def _safe_bound_file(source_root: Path, relative: str, *, label: str) -> Path:
    """Resolve one repository-relative file without following any nested symlink."""

    _require(source_root.is_dir() and not source_root.is_symlink(), "identity source root is missing or unsafe")
    relative_path = Path(relative)
    _require(
        not relative_path.is_absolute() and relative_path.parts and ".." not in relative_path.parts,
        f"{label} path is unsafe",
    )
    try:
        resolved_root = source_root.resolve(strict=True)
    except OSError as exc:
        raise PackageError("identity source root cannot be resolved") from exc
    candidate = source_root
    for part in relative_path.parts:
        candidate = candidate / part
        _require(not candidate.is_symlink(), f"{label} has a symlink component")
    try:
        resolved = candidate.resolve(strict=True)
        resolved.relative_to(resolved_root)
    except (OSError, ValueError) as exc:
        raise PackageError(f"{label} escapes the identity source root") from exc
    _require(resolved.is_file(), f"{label} is missing or unsafe")
    return resolved


def identity_source_manifest(manifest: dict[str, Any], source_root: Path) -> dict[str, Any]:
    _require(source_root.is_dir() and not source_root.is_symlink(), "identity source root is missing or unsafe")
    files = []
    for relative in sorted(_identity_source_paths(manifest)):
        path = _safe_bound_file(source_root, relative, label=f"identity source: {relative}")
        _require(path.stat().st_size <= 1024 * 1024, f"identity source is too large: {relative}")
        files.append({"path": relative, "bytes": path.stat().st_size, "sha256": sha256_file(path)})
    stable = {"files": files, "file_count": len(files), "total_bytes": sum(item["bytes"] for item in files)}
    return {**stable, "sha256": canonical_sha256(stable)}


def compile_identity_bundle(
    *,
    manifest: dict[str, Any],
    expected_identity_contract: dict[str, Any],
    source_root: Path,
    output: Path,
) -> dict[str, Any]:
    expected_revision = expected_identity_contract.get("runtime", {}).get("teleo_git_head")
    _require(_is_revision(expected_revision), "compiled identity revision is invalid")
    validate_expected_identity_contract(expected_identity_contract, expected_revision=expected_revision)
    try:
        views = strict_identity_manifest.verify_bound_sources(
            manifest,
            source_root,
            expected_runtime_policy_value=expected_identity_contract["permissions"]["runtime_policy"],
        )
    except strict_identity_manifest.IdentityManifestError as exc:
        raise PackageError("compiled identity manifest failed strict semantic validation") from exc
    _require(set(views) == {"SOUL.md", "identity.json"}, "compiled identity views are incomplete")
    _require(not output.exists() and not output.is_symlink(), "compiled identity output already exists")
    output.mkdir(parents=True, mode=0o700)
    try:
        for name, content in views.items():
            path = output / name
            path.write_text(content, encoding="utf-8")
            path.chmod(0o600)
        write_json(output / "identity-manifest.json", manifest, mode=0o600)
        write_json(
            output / "identity-lock.json",
            {
                "schema": IDENTITY_LOCK_SCHEMA,
                "manifest_sha256": manifest["manifest_sha256"],
                "identity_inputs_sha256": manifest["identity_inputs_sha256"],
                "compiled_views": {name: {"sha256": sha256_file(output / name)} for name in sorted(views)},
                "session_boundary": manifest["identity_inputs"]["session_boundary"],
            },
            mode=0o600,
        )
        return validate_identity_bundle(
            output,
            expected_revision=expected_revision,
            expected_identity_contract=expected_identity_contract,
            source_root=source_root,
        )
    except BaseException:
        shutil.rmtree(output, ignore_errors=True)
        raise


def build_synthetic_identity_manifest(
    *,
    artifact: Path,
    receipt: Path,
    source_revision: str,
    contract_source_root: Path,
    identity_source_root: Path,
    database_fingerprint: Path,
    constitution: Path,
    database_identity: Path,
) -> dict[str, Any]:
    """Generate one synthetic identity only through the package's strict runtime contract."""

    runtime = validate_artifact(artifact, receipt, source_revision)
    expected_contract = build_expected_identity_contract(
        artifact=artifact,
        receipt_path=receipt,
        runtime=runtime,
        source_revision=source_revision,
        source_root=contract_source_root,
    )
    try:
        manifest = strict_identity_manifest.build_identity_manifest(
            identity_runtime_contract=expected_contract,
            database_fingerprint_path=database_fingerprint,
            constitution_path=constitution,
            database_identity_path=database_identity,
            source_root=identity_source_root,
        )
        strict_identity_manifest.verify_bound_sources(
            manifest,
            identity_source_root,
            expected_runtime_policy_value=expected_contract["permissions"]["runtime_policy"],
        )
    except strict_identity_manifest.IdentityManifestError as exc:
        raise PackageError("synthetic identity manifest failed strict package validation") from exc
    return manifest


def validate_identity_bundle(
    root: Path,
    *,
    expected_revision: str,
    expected_identity_contract: dict[str, Any],
    source_root: Path,
) -> dict[str, Any]:
    validate_expected_identity_contract(expected_identity_contract, expected_revision=expected_revision)
    bundle = identity_bundle_manifest(root)
    manifest = load_json(root / "identity-manifest.json", "identity manifest")
    lock = load_json(root / "identity-lock.json", "identity lock")
    _require(
        set(manifest) == {"schema", "identity_inputs", "identity_inputs_sha256", "compiled_views", "manifest_sha256"},
        "identity manifest fields are not exact",
    )
    _require(manifest.get("schema") == IDENTITY_MANIFEST_SCHEMA, "identity manifest schema is invalid")
    try:
        rendered = strict_identity_manifest.verify_bound_sources(
            manifest,
            source_root,
            expected_runtime_policy_value=expected_identity_contract["permissions"]["runtime_policy"],
        )
    except strict_identity_manifest.IdentityManifestError as exc:
        raise PackageError("identity manifest failed strict semantic/source validation") from exc
    inputs = manifest["identity_inputs"]
    runtime = inputs.get("runtime")
    _require(
        inputs.get("source_commit") == (runtime or {}).get("teleo_git_head") == expected_revision,
        "identity source/runtime revision differs from package revision",
    )
    for section in ("runtime", "model", "skills", "permissions"):
        _require(
            inputs.get(section) == expected_identity_contract[section],
            f"identity {section} differs from the expected package runtime contract",
        )
    views = manifest.get("compiled_views")
    _require(isinstance(views, dict) and set(views) == set(rendered), "compiled identity views are incomplete")
    for name, content in rendered.items():
        _require(
            (root / name).read_text(encoding="utf-8") == content, f"compiled identity view derivation drifted: {name}"
        )
        _require(views[name].get("sha256") == sha256_file(root / name), f"compiled identity view hash drifted: {name}")
    _require(
        set(lock) == {"schema", "manifest_sha256", "identity_inputs_sha256", "compiled_views", "session_boundary"},
        "identity lock fields are not exact",
    )
    _require(lock.get("schema") == IDENTITY_LOCK_SCHEMA, "identity lock schema is invalid")
    _require(lock.get("manifest_sha256") == manifest["manifest_sha256"], "identity lock manifest binding drifted")
    _require(
        lock.get("identity_inputs_sha256") == manifest["identity_inputs_sha256"], "identity lock input binding drifted"
    )
    expected_views = {name: {"sha256": sha256_file(root / name)} for name in sorted(rendered)}
    _require(lock.get("compiled_views") == expected_views, "identity lock compiled-view binding drifted")
    _require(lock.get("session_boundary") == inputs.get("session_boundary"), "identity lock session binding drifted")
    database = inputs.get("canonical_database")
    _require(isinstance(database, dict), "identity database authority is missing")
    _require(
        database.get("authority") == "synthetic_noncanonical_fixture"
        and database.get("canonical_authority_granted") is False,
        "this package accepts only synthetic noncanonical identity",
    )
    sources = inputs.get("identity_sources")
    provenance = (
        (sources or {}).get("database_derived_identity", {}).get("provenance") if isinstance(sources, dict) else None
    )
    _require(
        provenance
        == {
            "mode": "synthetic_noncanonical_fixture",
            "canonical_authority_granted": False,
            "same_transaction_as_fingerprint": False,
            "export_receipt_sha256": None,
            "authority": "synthetic_noncanonical_fixture",
        },
        "synthetic identity provenance binding drifted",
    )
    for field in ("fingerprint_sha256", "structure_sha256", "table_rows_sha256"):
        _require(_is_sha256(database.get(field)), f"identity database {field} is invalid")
    for field in ("database", "database_user", "system_identifier"):
        _require(
            isinstance(database.get(field), str) and bool(database[field]), f"identity database {field} is missing"
        )
    _require(
        isinstance(database.get("table_count"), int)
        and not isinstance(database["table_count"], bool)
        and database["table_count"] > 0,
        "identity database table_count is invalid",
    )
    _require(
        isinstance(database.get("total_rows"), int)
        and not isinstance(database["total_rows"], bool)
        and database["total_rows"] >= 0,
        "identity database total_rows is invalid",
    )
    source_manifest = identity_source_manifest(manifest, source_root)
    return {
        "bundle_sha256": bundle["sha256"],
        "manifest_sha256": manifest["manifest_sha256"],
        "identity_inputs_sha256": manifest["identity_inputs_sha256"],
        "identity_lock_sha256": sha256_file(root / "identity-lock.json"),
        "soul_sha256": sha256_file(root / "SOUL.md"),
        "identity_view_sha256": sha256_file(root / "identity.json"),
        "identity_sources_sha256": source_manifest["sha256"],
        **{
            field: database[field]
            for field in (
                "database",
                "database_user",
                "system_identifier",
                "fingerprint_sha256",
                "structure_sha256",
                "table_rows_sha256",
                "table_count",
                "total_rows",
            )
        },
        "runtime_contract": expected_identity_contract,
        "authority": "synthetic_noncanonical_fixture",
        "canonical_authority_granted": False,
    }


def validate_artifact(artifact: Path, receipt_path: Path, expected_revision: str) -> dict[str, Any]:
    _require(_is_revision(expected_revision), "expected Teleo revision is invalid")
    _safe_tree(artifact, "runtime artifact")
    manifest = load_json(artifact / "artifact-manifest.json", "artifact manifest")
    receipt = load_json(receipt_path, "artifact verification receipt")
    _require(manifest.get("schema") == ARTIFACT_MANIFEST_SCHEMA, "artifact manifest schema is invalid")
    _require(receipt.get("schema") == ARTIFACT_RECEIPT_SCHEMA, "artifact receipt schema is invalid")
    _require(receipt.get("status") == "pass", "artifact receipt did not pass")
    _require(receipt.get("verification_mode") == "release", "artifact receipt is not release evidence")
    _require(manifest.get("teleo_git_head") == expected_revision, "artifact revision differs from source revision")
    _require(receipt.get("teleo_git_head") == expected_revision, "receipt revision differs from source revision")
    content = tree_manifest(artifact, excluded=frozenset({"artifact-manifest.json"}))
    _require(manifest.get("content") == content, "artifact content manifest drifted")
    _require(receipt.get("artifact_sha256") == content["sha256"], "artifact receipt content hash drifted")
    _require(manifest.get("hermes_commit") == HERMES_COMMIT, "artifact Hermes commit drifted")
    _require(receipt.get("hermes_commit") == HERMES_COMMIT, "receipt Hermes commit drifted")
    _require(_is_sha256(receipt.get("hermes_source_sha256")), "receipt Hermes source hash is invalid")
    config_path = artifact / "profile-template" / "config.yaml"
    _require(
        config_path.is_file()
        and not config_path.is_symlink()
        and receipt.get("config_sha256") == sha256_file(config_path),
        "artifact config differs from the runtime receipt",
    )
    _validate_runtime_probe(receipt, expected_config_sha256=sha256_file(config_path))
    runtime_lock = load_json(artifact / "runtime-lock.json", "artifact runtime lock")
    _require(
        runtime_lock.get("base_images") == {"python": PYTHON_IMAGE.removeprefix("docker.io/library/"), "uv": UV_IMAGE},
        "artifact base-image lock drifted",
    )
    hermes = runtime_lock.get("hermes")
    _require(isinstance(hermes, dict) and hermes.get("commit") == HERMES_COMMIT, "artifact Hermes lock drifted")
    _require(
        hermes.get("patched_source_tree_sha256") == receipt["hermes_source_sha256"],
        "artifact Hermes source differs from the runtime lock",
    )
    _require(_is_sha256(hermes.get("uv_lock_sha256")), "artifact uv lock hash is invalid")
    claim_ceiling = manifest.get("claim_ceiling")
    _require(isinstance(claim_ceiling, str) and claim_ceiling == receipt.get("claim_ceiling"), "claim ceiling drifted")
    return {
        "teleo_git_head": expected_revision,
        "artifact_sha256": content["sha256"],
        "artifact_manifest_sha256": sha256_file(artifact / "artifact-manifest.json"),
        "artifact_receipt_sha256": sha256_file(receipt_path),
        "hermes_commit": HERMES_COMMIT,
        "hermes_source_sha256": receipt.get("hermes_source_sha256"),
        "uv_lock_sha256": hermes.get("uv_lock_sha256"),
        "config_sha256": receipt.get("config_sha256"),
        "runtime_contract_sha256": sha256_file(artifact / "runtime-contract.json"),
        "claim_ceiling": claim_ceiling,
    }


def _target() -> dict[str, str]:
    return {
        "project": PROJECT,
        "zone": ZONE,
        "instance": INSTANCE,
        "service": SERVICE,
        "service_account": SERVICE_ACCOUNT,
        "platform": PLATFORM,
    }


def _base_images() -> dict[str, str]:
    return {"python": PYTHON_IMAGE, "uv": UV_IMAGE, "postgres": POSTGRES_IMAGE}


def _assert_no_forbidden_descriptor_markers(value: Mapping[str, Any]) -> None:
    encoded = json.dumps(value, ensure_ascii=True, separators=(",", ":"), sort_keys=True).casefold()
    for marker in FORBIDDEN_DESCRIPTOR_MARKERS:
        _require(
            marker.casefold() not in encoded, "release descriptor contains a forbidden production/transport marker"
        )


def validate_image_input(value: dict[str, Any]) -> None:
    _require(
        set(value)
        == {
            "schema",
            "target",
            "runtime",
            "identity",
            "base_images",
            "source_bindings",
            "claim_ceiling",
            "input_sha256",
        },
        "image input fields are not exact",
    )
    _require(value.get("schema") == IMAGE_INPUT_SCHEMA, "image input schema is invalid")
    _require(value.get("target") == _target(), "image input target is not exact GCP staging")
    _require(value.get("base_images") == _base_images(), "image input base-image closure drifted")
    runtime = value.get("runtime")
    _require(isinstance(runtime, dict), "image runtime binding is missing")
    _require(
        set(runtime)
        == {
            "teleo_git_head",
            "artifact_sha256",
            "artifact_manifest_sha256",
            "artifact_receipt_sha256",
            "hermes_commit",
            "hermes_source_sha256",
            "uv_lock_sha256",
            "config_sha256",
            "runtime_contract_sha256",
            "python_version",
            "uv_version",
            "postgres_version",
            "cloudsql_ca_sha256",
        },
        "image runtime binding fields are not exact",
    )
    _require(_is_revision(runtime.get("teleo_git_head")), "image Teleo revision is invalid")
    for field in (
        "artifact_sha256",
        "artifact_manifest_sha256",
        "artifact_receipt_sha256",
        "hermes_source_sha256",
        "uv_lock_sha256",
        "config_sha256",
        "runtime_contract_sha256",
        "cloudsql_ca_sha256",
    ):
        _require(_is_sha256(runtime.get(field)), f"image runtime {field} is invalid")
    _require(runtime.get("hermes_commit") == HERMES_COMMIT, "image Hermes commit drifted")
    _require(runtime.get("python_version") == PYTHON_VERSION, "image Python version drifted")
    _require(runtime.get("uv_version") == UV_VERSION, "image uv version drifted")
    _require(runtime.get("postgres_version") == POSTGRES_VERSION, "image PostgreSQL version drifted")
    _require(runtime.get("cloudsql_ca_sha256") == CA_SHA256, "image Cloud SQL CA drifted")
    identity = value.get("identity")
    _require(
        isinstance(identity, dict)
        and set(identity)
        == {
            "bundle_sha256",
            "manifest_sha256",
            "identity_inputs_sha256",
            "identity_lock_sha256",
            "soul_sha256",
            "identity_view_sha256",
            "identity_sources_sha256",
            "database",
            "database_user",
            "system_identifier",
            "fingerprint_sha256",
            "structure_sha256",
            "table_rows_sha256",
            "table_count",
            "total_rows",
            "runtime_contract",
            "authority",
            "canonical_authority_granted",
        },
        "image identity binding fields are not exact",
    )
    for field in (
        "bundle_sha256",
        "manifest_sha256",
        "identity_inputs_sha256",
        "identity_lock_sha256",
        "soul_sha256",
        "identity_view_sha256",
        "identity_sources_sha256",
        "fingerprint_sha256",
        "structure_sha256",
        "table_rows_sha256",
    ):
        _require(_is_sha256(identity.get(field)), f"image identity {field} is invalid")
    _require(
        identity.get("authority") == "synthetic_noncanonical_fixture"
        and identity.get("canonical_authority_granted") is False,
        "image identity must remain synthetic and noncanonical",
    )
    for field in ("database", "database_user", "system_identifier"):
        _require(isinstance(identity.get(field), str) and bool(identity[field]), f"image identity {field} is missing")
    _require(
        isinstance(identity.get("table_count"), int)
        and not isinstance(identity["table_count"], bool)
        and identity["table_count"] > 0
        and isinstance(identity.get("total_rows"), int)
        and not isinstance(identity["total_rows"], bool)
        and identity["total_rows"] >= 0,
        "image identity database counts are invalid",
    )
    identity_contract = identity.get("runtime_contract")
    _require(isinstance(identity_contract, dict), "image expected identity runtime contract is missing")
    validate_expected_identity_contract(identity_contract, expected_revision=runtime["teleo_git_head"])
    _require(
        identity_contract["runtime"]["hermes_git_head"] == runtime["hermes_commit"]
        and identity_contract["runtime"]["hermes_source_sha256"] == runtime["hermes_source_sha256"]
        and identity_contract["runtime"]["teleo_git_head"] == runtime["teleo_git_head"]
        and identity_contract["runtime"]["python"]["python_version"] == runtime["python_version"],
        "image identity contract differs from the packaged runtime",
    )
    behavior_inputs = identity_contract["behavior_inputs"]
    _require(
        behavior_inputs["artifact_sha256"] == runtime["artifact_sha256"]
        and behavior_inputs["artifact_receipt_sha256"] == runtime["artifact_receipt_sha256"]
        and behavior_inputs["config_sha256"] == runtime["config_sha256"]
        and behavior_inputs["runtime_contract_sha256"] == runtime["runtime_contract_sha256"]
        and behavior_inputs["uv_lock_sha256"] == runtime["uv_lock_sha256"],
        "image identity behavior inputs differ from the packaged runtime",
    )
    bindings = value.get("source_bindings")
    _require(
        isinstance(bindings, dict)
        and set(bindings)
        == {
            "Dockerfile",
            "entrypoint.py",
            "package_contract.py",
            "scripts/leo_behavior_manifest.py",
            "scripts/leo_identity_manifest.py",
        }
        and all(_is_sha256(item) for item in bindings.values()),
        "image packaging source bindings are invalid",
    )
    _require(
        identity_contract["behavior_inputs"]["identity_contract_sources"]
        == {
            "scripts/leo_behavior_manifest.py": bindings["scripts/leo_behavior_manifest.py"],
            "scripts/leo_identity_manifest.py": bindings["scripts/leo_identity_manifest.py"],
        },
        "image identity contract differs from the packaged identity validators",
    )
    _require(
        isinstance(value.get("claim_ceiling"), str) and bool(value["claim_ceiling"]), "image claim ceiling is missing"
    )
    stable = {key: item for key, item in value.items() if key != "input_sha256"}
    _require(value.get("input_sha256") == canonical_sha256(stable), "image input self-hash drifted")
    _assert_no_forbidden_descriptor_markers(value)


def build_image_input(
    *,
    artifact: Path,
    receipt: Path,
    identity: Path,
    ca: Path,
    source_revision: str,
    dockerfile: Path,
    entrypoint: Path,
    package_contract: Path,
    contract_source_root: Path,
    identity_source_root: Path,
) -> dict[str, Any]:
    _require(sha256_file(ca) == CA_SHA256, "Cloud SQL CA does not match the reviewed certificate")
    runtime = validate_artifact(artifact, receipt, source_revision)
    runtime.update(
        {
            "python_version": PYTHON_VERSION,
            "uv_version": UV_VERSION,
            "postgres_version": POSTGRES_VERSION,
            "cloudsql_ca_sha256": CA_SHA256,
        }
    )
    expected_identity_contract = build_expected_identity_contract(
        artifact=artifact,
        receipt_path=receipt,
        runtime=runtime,
        source_revision=source_revision,
        source_root=contract_source_root,
    )
    identity_binding = validate_identity_bundle(
        identity,
        expected_revision=source_revision,
        expected_identity_contract=expected_identity_contract,
        source_root=identity_source_root,
    )
    identity_contract_sources = _identity_contract_source_bindings(contract_source_root)
    identity_manifest = load_json(identity / "identity-manifest.json", "identity manifest")
    identity_sources = identity_source_manifest(identity_manifest, identity_source_root)
    _require(
        contract_source_root.resolve() == identity_source_root.resolve(),
        "synthetic identity sources must use the reviewed package repository",
    )
    fixed_source_hashes = {
        "docker/leoclean-nosend/Dockerfile": sha256_file(dockerfile),
        "docker/leoclean-nosend/entrypoint.py": sha256_file(entrypoint),
        "ops/gcp-teleo-pgvector-standby-server-ca.pem": sha256_file(ca),
        "ops/gcp_leoclean_nosend_package.py": sha256_file(package_contract),
        **identity_contract_sources,
    }
    reviewed_source_hashes = {
        **fixed_source_hashes,
        **{str(entry["path"]): str(entry["sha256"]) for entry in identity_sources["files"]},
    }
    _verify_git_source_closure(contract_source_root, source_revision, reviewed_source_hashes)
    stable = {
        "schema": IMAGE_INPUT_SCHEMA,
        "target": _target(),
        "runtime": {key: runtime[key] for key in runtime if key != "claim_ceiling"},
        "identity": identity_binding,
        "base_images": _base_images(),
        "source_bindings": {
            "Dockerfile": fixed_source_hashes["docker/leoclean-nosend/Dockerfile"],
            "entrypoint.py": fixed_source_hashes["docker/leoclean-nosend/entrypoint.py"],
            "package_contract.py": fixed_source_hashes["ops/gcp_leoclean_nosend_package.py"],
            **identity_contract_sources,
        },
        "claim_ceiling": runtime["claim_ceiling"],
    }
    value = {**stable, "input_sha256": canonical_sha256(stable)}
    validate_image_input(value)
    return value


def validate_installed_runtime_identity(
    image_input: dict[str, Any],
    *,
    artifact: Path,
    receipt: Path,
    identity: Path,
    contract_source_root: Path,
    identity_source_root: Path,
) -> tuple[dict[str, Any], dict[str, Any]]:
    """Re-derive installed runtime and identity authority instead of trusting the descriptor."""

    validate_image_input(image_input)
    revision = image_input["runtime"]["teleo_git_head"]
    runtime = validate_artifact(artifact, receipt, revision)
    for field in (
        "teleo_git_head",
        "artifact_sha256",
        "artifact_manifest_sha256",
        "artifact_receipt_sha256",
        "hermes_commit",
        "hermes_source_sha256",
        "uv_lock_sha256",
        "config_sha256",
        "runtime_contract_sha256",
    ):
        _require(runtime.get(field) == image_input["runtime"].get(field), f"installed runtime binding drifted: {field}")
    expected_identity_contract = build_expected_identity_contract(
        artifact=artifact,
        receipt_path=receipt,
        runtime=runtime,
        source_revision=revision,
        source_root=contract_source_root,
    )
    _require(
        expected_identity_contract == image_input["identity"]["runtime_contract"],
        "installed identity runtime contract differs from its installed sources",
    )
    identity_binding = validate_identity_bundle(
        identity,
        expected_revision=revision,
        expected_identity_contract=expected_identity_contract,
        source_root=identity_source_root,
    )
    _require(identity_binding == image_input["identity"], "installed identity bundle drifted")
    return runtime, identity_binding


def _copy_identity_sources(manifest: dict[str, Any], source_root: Path, output: Path) -> None:
    expected = identity_source_manifest(manifest, source_root)
    output.mkdir(parents=True, mode=0o755)
    for entry in expected["files"]:
        relative = entry["path"]
        destination = output / relative
        destination.parent.mkdir(parents=True, exist_ok=True)
        shutil.copyfile(_safe_bound_file(source_root, relative, label=f"identity source: {relative}"), destination)
        destination.chmod(0o644)
    copied = identity_source_manifest(manifest, output)
    _require(copied == expected, "copied identity sources differ from their reviewed inputs")


def prepare_context(
    *,
    repo_root: Path,
    artifact: Path,
    receipt: Path,
    identity: Path,
    identity_source_root: Path,
    output: Path,
    source_revision: str,
) -> dict[str, Any]:
    _require(not output.exists() and not output.is_symlink(), "output context already exists")
    dockerfile = repo_root / "docker" / "leoclean-nosend" / "Dockerfile"
    entrypoint = repo_root / "docker" / "leoclean-nosend" / "entrypoint.py"
    package_contract = repo_root / "ops" / "gcp_leoclean_nosend_package.py"
    ca = repo_root / "ops" / "gcp-teleo-pgvector-standby-server-ca.pem"
    for path, label in (
        (dockerfile, "Dockerfile"),
        (entrypoint, "entrypoint"),
        (package_contract, "package contract"),
        (ca, "Cloud SQL CA"),
    ):
        _require(path.is_file() and not path.is_symlink(), f"{label} source is missing or unsafe")
    image_input = build_image_input(
        artifact=artifact,
        receipt=receipt,
        identity=identity,
        ca=ca,
        source_revision=source_revision,
        dockerfile=dockerfile,
        entrypoint=entrypoint,
        package_contract=package_contract,
        contract_source_root=repo_root,
        identity_source_root=identity_source_root,
    )
    identity_manifest = load_json(identity / "identity-manifest.json", "identity manifest")
    output.mkdir(parents=True, mode=0o700)
    try:
        shutil.copytree(artifact, output / "artifact", symlinks=False)
        shutil.copytree(identity, output / "identity", symlinks=False)
        for name in IDENTITY_FILES:
            (output / "identity" / name).chmod(0o600)
        _copy_identity_sources(identity_manifest, identity_source_root, output / "identity-sources")
        shutil.copyfile(receipt, output / "artifact-receipt.json")
        shutil.copyfile(ca, output / "cloudsql-server-ca.pem")
        shutil.copyfile(dockerfile, output / "Dockerfile")
        shutil.copyfile(entrypoint, output / "entrypoint.py")
        shutil.copyfile(package_contract, output / "package_contract.py")
        scripts = output / "scripts"
        scripts.mkdir(mode=0o755)
        for relative in IDENTITY_CONTRACT_SOURCES:
            destination = output / relative
            shutil.copyfile(repo_root / relative, destination)
            destination.chmod(0o644)
        for name in (
            "artifact-receipt.json",
            "cloudsql-server-ca.pem",
            "Dockerfile",
            "entrypoint.py",
            "package_contract.py",
        ):
            (output / name).chmod(0o644)
        (output / "artifact" / "artifact-manifest.json").chmod(0o644)
        write_json(output / "image-input.json", image_input)
        write_json(
            output / "build-args.json",
            {
                "ARTIFACT_SHA256": image_input["runtime"]["artifact_sha256"],
                "IDENTITY_SHA256": image_input["identity"]["bundle_sha256"],
                "INPUT_SHA256": image_input["input_sha256"],
                "TELEO_REVISION": source_revision,
            },
        )
        _normalize_prepared_directory_modes(output)
        _safe_tree(output, "prepared image context")
        return image_input
    except BaseException:
        shutil.rmtree(output, ignore_errors=True)
        raise


def validate_prepared_image_context(context: Path) -> tuple[dict[str, Any], dict[str, str]]:
    """Revalidate every file that can enter the reviewed image build context."""

    _require(context.is_absolute(), "prepared image context must be absolute")
    _safe_tree(context, "prepared image context")
    _require(
        {path.name for path in context.iterdir()}
        == {
            "Dockerfile",
            "artifact",
            "artifact-receipt.json",
            "build-args.json",
            "cloudsql-server-ca.pem",
            "entrypoint.py",
            "identity",
            "identity-sources",
            "image-input.json",
            "package_contract.py",
            "scripts",
        },
        "prepared image context entry set is not exact",
    )
    image_input = load_json(context / "image-input.json", "image input")
    validate_image_input(image_input)
    expected_build_args = {
        "ARTIFACT_SHA256": image_input["runtime"]["artifact_sha256"],
        "IDENTITY_SHA256": image_input["identity"]["bundle_sha256"],
        "INPUT_SHA256": image_input["input_sha256"],
        "TELEO_REVISION": image_input["runtime"]["teleo_git_head"],
    }
    _require(
        load_json(context / "build-args.json", "build arguments") == expected_build_args,
        "prepared image build arguments drifted",
    )
    bindings = image_input["source_bindings"]
    for relative, expected in (
        ("Dockerfile", bindings["Dockerfile"]),
        ("entrypoint.py", bindings["entrypoint.py"]),
        ("package_contract.py", bindings["package_contract.py"]),
        ("scripts/leo_behavior_manifest.py", bindings["scripts/leo_behavior_manifest.py"]),
        ("scripts/leo_identity_manifest.py", bindings["scripts/leo_identity_manifest.py"]),
        ("cloudsql-server-ca.pem", image_input["runtime"]["cloudsql_ca_sha256"]),
        ("artifact-receipt.json", image_input["runtime"]["artifact_receipt_sha256"]),
    ):
        _require(
            sha256_file(_safe_bound_file(context, relative, label=f"prepared image source: {relative}")) == expected,
            f"prepared image source drifted: {relative}",
        )
    runtime, identity = validate_installed_runtime_identity(
        image_input,
        artifact=context / "artifact",
        receipt=context / "artifact-receipt.json",
        identity=context / "identity",
        contract_source_root=context,
        identity_source_root=context / "identity-sources",
    )
    _require(runtime["artifact_sha256"] == image_input["runtime"]["artifact_sha256"], "prepared artifact drifted")
    _require(identity == image_input["identity"], "prepared identity drifted")
    identity_manifest = load_json(context / "identity" / "identity-manifest.json", "identity manifest")
    expected_identity_paths = set(_identity_source_paths(identity_manifest))
    actual_identity_paths = {
        path.relative_to(context / "identity-sources").as_posix()
        for path in (context / "identity-sources").rglob("*")
        if path.is_file()
    }
    _require(actual_identity_paths == expected_identity_paths, "prepared identity source entry set drifted")
    actual_script_paths = {
        path.relative_to(context).as_posix()
        for path in (context / "scripts").rglob("*")
        if path.is_file()
    }
    _require(
        actual_script_paths == set(IDENTITY_CONTRACT_SOURCES),
        "prepared identity validator entry set drifted",
    )
    _require(
        identity_source_manifest(identity_manifest, context / "identity-sources")["sha256"]
        == image_input["identity"]["identity_sources_sha256"],
        "prepared identity sources drifted",
    )
    artifact_manifest = load_json(context / "artifact" / "artifact-manifest.json", "artifact manifest")
    artifact_content = artifact_manifest.get("content")
    artifact_files = artifact_content.get("files") if isinstance(artifact_content, dict) else None
    _require(
        isinstance(artifact_files, list)
        and all(
            isinstance(entry, dict)
            and isinstance(entry.get("path"), str)
            and isinstance(entry.get("mode"), str)
            for entry in artifact_files
        ),
        "prepared artifact file manifest is invalid",
    )
    expected_file_modes: dict[str, int] = {
        "Dockerfile": 0o644,
        "artifact/artifact-manifest.json": 0o644,
        "artifact-receipt.json": 0o644,
        "build-args.json": 0o644,
        "cloudsql-server-ca.pem": 0o644,
        "entrypoint.py": 0o644,
        "image-input.json": 0o644,
        "package_contract.py": 0o644,
    }
    for entry in artifact_files:
        raw_mode = entry["mode"]
        _require(bool(re.fullmatch(r"0o[0-7]{3,4}", raw_mode)), "prepared artifact file mode is invalid")
        expected_file_modes[f"artifact/{entry['path']}"] = int(raw_mode.removeprefix("0o"), 8)
    expected_file_modes.update({f"identity/{name}": 0o600 for name in IDENTITY_FILES})
    expected_file_modes.update(
        {f"identity-sources/{relative}": 0o644 for relative in expected_identity_paths}
    )
    expected_file_modes.update({relative: 0o644 for relative in IDENTITY_CONTRACT_SOURCES})
    actual_files = {path.relative_to(context).as_posix() for path in context.rglob("*") if path.is_file()}
    _require(actual_files == set(expected_file_modes), "prepared image file entry set drifted")
    for relative, expected_mode in expected_file_modes.items():
        _require(
            stat.S_IMODE((context / relative).stat().st_mode) == expected_mode,
            f"prepared image file mode drifted: {relative}",
        )
    expected_directories = {"artifact", "identity", "identity-sources", "scripts"}
    expected_directories.update(
        f"artifact/{relative}"
        for relative in _parent_directory_paths({entry["path"] for entry in artifact_files})
    )
    expected_directories.update(
        _parent_directory_paths({f"identity-sources/{relative}" for relative in expected_identity_paths})
    )
    expected_directories.update(_parent_directory_paths(set(IDENTITY_CONTRACT_SOURCES)))
    actual_directories = {
        path.relative_to(context).as_posix() for path in context.rglob("*") if path.is_dir()
    }
    _require(actual_directories == expected_directories, "prepared image directory entry set drifted")
    _require(stat.S_IMODE(context.stat().st_mode) == 0o700, "prepared image context mode drifted")
    for relative in actual_directories:
        expected_mode = 0o700 if relative == "identity" else 0o755
        _require(
            stat.S_IMODE((context / relative).stat().st_mode) == expected_mode,
            f"prepared image directory mode drifted: {relative}",
        )
    return image_input, expected_build_args


def validate_reviewed_context_source(
    repo_root: Path,
    context: Path,
    image_input: dict[str, Any],
) -> None:
    """Bind the prepared context to tracked bytes at the exact reviewed checkout."""

    validate_image_input(image_input)
    revision = image_input["runtime"]["teleo_git_head"]
    bindings = image_input["source_bindings"]
    expected_source_sha256 = {
        "docker/leoclean-nosend/Dockerfile": bindings["Dockerfile"],
        "docker/leoclean-nosend/entrypoint.py": bindings["entrypoint.py"],
        "ops/gcp-teleo-pgvector-standby-server-ca.pem": image_input["runtime"]["cloudsql_ca_sha256"],
        "ops/gcp_leoclean_nosend_package.py": bindings["package_contract.py"],
        "scripts/leo_behavior_manifest.py": bindings["scripts/leo_behavior_manifest.py"],
        "scripts/leo_identity_manifest.py": bindings["scripts/leo_identity_manifest.py"],
    }
    _verify_git_source_closure(repo_root, revision, expected_source_sha256)
    identity_manifest = load_json(context / "identity" / "identity-manifest.json", "identity manifest")
    reviewed_identity_sources = identity_source_manifest(identity_manifest, repo_root)
    prepared_identity_sources = identity_source_manifest(identity_manifest, context / "identity-sources")
    _require(
        reviewed_identity_sources == prepared_identity_sources
        and reviewed_identity_sources["sha256"] == image_input["identity"]["identity_sources_sha256"],
        "prepared identity sources differ from the reviewed checkout",
    )


def validate_image_runtime_configuration(inspected: dict[str, Any]) -> dict[str, Any]:
    """Validate the runtime-bearing Docker config shared by local and registry inspection."""

    config = inspected.get("Config")
    healthcheck = config.get("Healthcheck") if isinstance(config, dict) else None
    raw_start_interval = healthcheck.get("StartInterval") if isinstance(healthcheck, dict) else None
    raw_args_escaped = config.get("ArgsEscaped") if isinstance(config, dict) else None
    expected_healthcheck_keys = {"Test", "Interval", "Timeout", "StartPeriod", "Retries"}
    allowed_healthcheck_keys = (expected_healthcheck_keys, expected_healthcheck_keys | {"StartInterval"})
    boolean_defaults = {
        "Tty": False,
        "OpenStdin": False,
        "StdinOnce": False,
        "AttachStdin": False,
        "AttachStdout": False,
        "AttachStderr": False,
        "NetworkDisabled": False,
    }
    string_defaults = {"WorkingDir": "", "Hostname": "", "Domainname": "", "MacAddress": ""}
    allowed_config_keys = {
        "ArgsEscaped",
        "AttachStderr",
        "AttachStdin",
        "AttachStdout",
        "Cmd",
        "Domainname",
        "Entrypoint",
        "Env",
        "ExposedPorts",
        "Healthcheck",
        "Hostname",
        "Image",
        "Labels",
        "MacAddress",
        "NetworkDisabled",
        "OnBuild",
        "OpenStdin",
        "Shell",
        "StdinOnce",
        "StopSignal",
        "StopTimeout",
        "Tty",
        "User",
        "Volumes",
        "WorkingDir",
    }
    _require(
        isinstance(config, dict)
        and isinstance(healthcheck, dict)
        and set(config) <= allowed_config_keys
        and config.get("User") in (None, "")
        and config.get("Image") in (None, "")
        # ArgsEscaped is a deprecated, Windows-only compatibility field. Linux
        # builders may omit it or serialize either boolean; exact argv arrays
        # below remain authoritative for this linux/amd64 image.
        and (raw_args_escaped is None or type(raw_args_escaped) is bool)
        and all(type(config.get(name, default)) is bool for name, default in boolean_defaults.items())
        and all(config.get(name, default) is default for name, default in boolean_defaults.items())
        and all(isinstance(config.get(name, default), str) for name, default in string_defaults.items())
        and all(config.get(name, default) == default for name, default in string_defaults.items())
        and config.get("StopTimeout") is None
        and set(healthcheck) in allowed_healthcheck_keys
        and all(type(healthcheck.get(name)) is int for name in ("Interval", "Timeout", "StartPeriod", "Retries"))
        and (
            raw_start_interval is None
            or (
                isinstance(raw_start_interval, int)
                and not isinstance(raw_start_interval, bool)
                and raw_start_interval == 0
            )
        ),
        "inspected image runtime config is invalid",
    )
    start_interval = 0
    runtime_config = {
        "user": None,
        "environment": config.get("Env"),
        "working_dir": config.get("WorkingDir", ""),
        "exposed_ports": config.get("ExposedPorts"),
        "volumes": config.get("Volumes"),
        "shell": config.get("Shell"),
        "on_build": config.get("OnBuild"),
        "args_escaped": False,
        "stop_timeout": config.get("StopTimeout"),
        "tty": config.get("Tty", False),
        "open_stdin": config.get("OpenStdin", False),
        "stdin_once": config.get("StdinOnce", False),
        "attach_stdin": config.get("AttachStdin", False),
        "attach_stdout": config.get("AttachStdout", False),
        "attach_stderr": config.get("AttachStderr", False),
        "network_disabled": config.get("NetworkDisabled", False),
        "hostname": config.get("Hostname", ""),
        "domainname": config.get("Domainname", ""),
        "mac_address": config.get("MacAddress", ""),
        "entrypoint": config.get("Entrypoint"),
        "cmd": config.get("Cmd"),
        "healthcheck": {
            "test": healthcheck.get("Test"),
            "interval_ns": healthcheck.get("Interval"),
            "timeout_ns": healthcheck.get("Timeout"),
            "start_period_ns": healthcheck.get("StartPeriod"),
            "start_interval_ns": start_interval,
            "retries": healthcheck.get("Retries"),
        },
        "stop_signal": config.get("StopSignal"),
    }
    _require(runtime_config == IMAGE_RUNTIME_CONFIG, "inspected image runtime config is invalid")
    return runtime_config


def validate_image_configuration(image_input: dict[str, Any], inspected: dict[str, Any]) -> dict[str, Any]:
    """Validate and normalize the immutable runtime-bearing image configuration."""

    validate_image_input(image_input)
    config_digest = inspected.get("Id")
    _require(
        isinstance(config_digest, str)
        and config_digest.startswith("sha256:")
        and _is_sha256(config_digest.removeprefix("sha256:")),
        "inspected image config digest is invalid",
    )
    variant = inspected.get("Variant")
    _require(
        inspected.get("Os") == "linux"
        and inspected.get("Architecture") == "amd64"
        and (variant is None or variant == ""),
        "inspected image platform is not linux/amd64",
    )
    config = inspected.get("Config")
    labels = config.get("Labels") if isinstance(config, dict) else None
    expected_labels = expected_image_labels(image_input)
    _require(
        isinstance(labels, dict)
        and all(isinstance(key, str) and isinstance(item, str) for key, item in labels.items())
        and all(labels.get(key) == item for key, item in expected_labels.items())
        and not any(key.startswith("livingip.") and key not in expected_labels for key in labels),
        "inspected image labels are not exact",
    )
    runtime_config = validate_image_runtime_configuration(inspected)
    return {
        "config_digest": config_digest,
        "platform": PLATFORM,
        "labels": expected_labels,
        "runtime_config": runtime_config,
    }


def validate_build_push_receipt(value: dict[str, Any], *, image_input: dict[str, Any]) -> None:
    """Validate the local-build to registry-manifest handoff consumed by finalization."""

    validate_image_input(image_input)
    _require(
        set(value)
        == {
            "schema",
            "source_revision",
            "input_sha256",
            "local_image",
            "registry",
            "receipt_sha256",
        },
        "build/push receipt fields are not exact",
    )
    _require(value.get("schema") == BUILD_PUSH_RECEIPT_SCHEMA, "build/push receipt schema is invalid")
    _require(
        value.get("source_revision") == image_input["runtime"]["teleo_git_head"],
        "build/push receipt source revision drifted",
    )
    _require(
        value.get("input_sha256") == image_input["input_sha256"],
        "build/push receipt image-input binding drifted",
    )
    local_image = value.get("local_image")
    _require(
        isinstance(local_image, dict) and set(local_image) == {"config_digest", "platform", "labels", "runtime_config"},
        "build/push receipt local image fields are not exact",
    )
    config_digest = local_image.get("config_digest")
    _require(
        isinstance(config_digest, str)
        and config_digest.startswith("sha256:")
        and _is_sha256(config_digest.removeprefix("sha256:")),
        "build/push receipt config digest is invalid",
    )
    _require(local_image.get("platform") == PLATFORM, "build/push receipt platform is not exact")
    _require(
        local_image.get("labels") == expected_image_labels(image_input),
        "build/push receipt labels are not exact",
    )
    _require(
        local_image.get("runtime_config") == IMAGE_RUNTIME_CONFIG,
        "build/push receipt runtime config is not exact",
    )
    registry = value.get("registry")
    _require(
        isinstance(registry, dict) and set(registry) == {"repository", "manifest_digest"},
        "build/push receipt registry fields are not exact",
    )
    _require(registry.get("repository") == IMAGE_REPOSITORY, "build/push receipt repository is not exact")
    manifest_digest = registry.get("manifest_digest")
    _require(
        isinstance(manifest_digest, str)
        and manifest_digest.startswith("sha256:")
        and _is_sha256(manifest_digest.removeprefix("sha256:")),
        "build/push receipt manifest digest is invalid",
    )
    stable = {key: item for key, item in value.items() if key != "receipt_sha256"}
    _require(value.get("receipt_sha256") == canonical_sha256(stable), "build/push receipt self-hash drifted")
    _assert_no_forbidden_descriptor_markers(value)


def build_build_push_receipt(
    image_input: dict[str, Any],
    *,
    post_push_image: dict[str, Any],
) -> dict[str, Any]:
    validate_image_input(image_input)
    observed = validate_image_configuration(image_input, post_push_image)
    repository_digests = _target_repository_digests(post_push_image)
    _require(len(repository_digests) == 1, "staging push did not produce one exact repository digest")
    match = IMAGE_REFERENCE.fullmatch(repository_digests[0])
    _require(match is not None, "staging push repository digest is invalid")
    stable = {
        "schema": BUILD_PUSH_RECEIPT_SCHEMA,
        "source_revision": image_input["runtime"]["teleo_git_head"],
        "input_sha256": image_input["input_sha256"],
        "local_image": {
            "config_digest": observed["config_digest"],
            "platform": observed["platform"],
            "labels": observed["labels"],
            "runtime_config": observed["runtime_config"],
        },
        "registry": {
            "repository": IMAGE_REPOSITORY,
            "manifest_digest": f"sha256:{match.group(1)}",
        },
    }
    receipt = {**stable, "receipt_sha256": canonical_sha256(stable)}
    validate_build_push_receipt(receipt, image_input=image_input)
    return receipt


def build_image_inspection(
    image_input: dict[str, Any],
    image_reference: str,
    inspected: dict[str, Any],
) -> dict[str, Any]:
    """Normalize one Docker inspection after an exact digest pull."""

    validate_image_input(image_input)
    match = IMAGE_REFERENCE.fullmatch(image_reference)
    _require(match is not None, "inspected image must be the exact staging repository at a sha256 digest")
    digest = f"sha256:{match.group(1)}"
    observed = validate_image_configuration(image_input, inspected)
    repository_digests = inspected.get("RepoDigests")
    _require(
        isinstance(repository_digests, list)
        and all(isinstance(item, str) for item in repository_digests)
        and len(repository_digests) == len(set(repository_digests))
        and image_reference in repository_digests,
        "inspected image is not bound to the requested repository digest",
    )
    value = {
        "schema": IMAGE_INSPECTION_SCHEMA,
        "status": "pass",
        "reference": image_reference,
        "digest": digest,
        "config_digest": observed["config_digest"],
        "platform": PLATFORM,
        "labels": observed["labels"],
        "runtime_config": observed["runtime_config"],
        "input_sha256": image_input["input_sha256"],
        "source_revision": image_input["runtime"]["teleo_git_head"],
    }
    validate_image_inspection(value, image_input=image_input)
    return value


def validate_image_inspection(value: dict[str, Any], *, image_input: dict[str, Any]) -> None:
    validate_image_input(image_input)
    _require(
        set(value)
        == {
            "schema",
            "status",
            "reference",
            "digest",
            "config_digest",
            "platform",
            "labels",
            "runtime_config",
            "input_sha256",
            "source_revision",
        },
        "image inspection fields are not exact",
    )
    _require(
        value.get("schema") == IMAGE_INSPECTION_SCHEMA and value.get("status") == "pass",
        "image inspection did not pass",
    )
    reference = value.get("reference")
    match = IMAGE_REFERENCE.fullmatch(reference) if isinstance(reference, str) else None
    _require(match is not None, "image inspection reference is not the exact staging digest")
    _require(value.get("digest") == f"sha256:{match.group(1)}", "image inspection digest differs from its reference")
    config_digest = value.get("config_digest")
    _require(
        isinstance(config_digest, str)
        and config_digest.startswith("sha256:")
        and _is_sha256(config_digest.removeprefix("sha256:")),
        "image inspection config digest is invalid",
    )
    _require(value.get("platform") == PLATFORM, "image inspection platform is not exact")
    _require(value.get("labels") == expected_image_labels(image_input), "image inspection labels are not exact")
    _require(value.get("runtime_config") == IMAGE_RUNTIME_CONFIG, "image inspection runtime config is not exact")
    _require(value.get("input_sha256") == image_input["input_sha256"], "image inspection input binding drifted")
    _require(
        value.get("source_revision") == image_input["runtime"]["teleo_git_head"],
        "image inspection source revision drifted",
    )
    _assert_no_forbidden_descriptor_markers(value)


def _load_docker_inspection(raw: bytes) -> Any:
    _require(len(raw) <= MAX_DOCKER_INSPECTION_BYTES, "registry image inspection output is too large")
    try:
        encoded = raw.decode("utf-8")
        return _strict_json_loads(encoded, "registry image inspection output")
    except (UnicodeDecodeError, PackageError) as exc:
        raise PackageError("registry image inspection output is invalid") from exc


def _docker_environment() -> dict[str, str]:
    return {
        name: value for name, value in os.environ.items() if not name.startswith(("DOCKER_", "BUILDKIT_", "BUILDX_"))
    }


def _validate_docker_authority(docker_binary: Path, docker_host: str, docker_config: Path) -> None:
    _require(docker_binary.is_absolute(), "Docker binary must be absolute")
    _require("\x00" not in str(docker_binary) and "\n" not in str(docker_binary), "Docker binary is invalid")
    _require(docker_host.startswith("unix://"), "Docker host must be a local Unix endpoint")
    socket_path = Path(docker_host.removeprefix("unix://"))
    _require(
        socket_path.is_absolute()
        and ".." not in socket_path.parts
        and "\x00" not in docker_host
        and "\n" not in docker_host,
        "Docker host must be a local Unix endpoint",
    )
    _require(
        docker_config.is_absolute() and docker_config.is_dir() and not docker_config.is_symlink(),
        "Docker config directory is missing or unsafe",
    )


class _DockerCacheLock:
    def __init__(self, docker_config: Path) -> None:
        self._docker_config = docker_config
        self._descriptor: int | None = None

    def __enter__(self) -> None:
        flags = os.O_RDONLY
        if hasattr(os, "O_DIRECTORY"):
            flags |= os.O_DIRECTORY
        if hasattr(os, "O_NOFOLLOW"):
            flags |= os.O_NOFOLLOW
        try:
            descriptor = os.open(self._docker_config, flags)
            fcntl.flock(descriptor, fcntl.LOCK_EX | fcntl.LOCK_NB)
        except OSError as exc:
            if "descriptor" in locals():
                os.close(descriptor)
            raise PackageError("Docker cache authority is already in use or unsafe") from exc
        self._descriptor = descriptor

    def __exit__(self, _kind: object, _value: object, _traceback: object) -> None:
        if self._descriptor is None:
            return
        try:
            fcntl.flock(self._descriptor, fcntl.LOCK_UN)
        finally:
            os.close(self._descriptor)
            self._descriptor = None


def _run_docker(
    docker_binary: Path,
    docker_host: str,
    docker_config: Path,
    arguments: list[str],
    *,
    check: bool,
    timeout: int,
) -> subprocess.CompletedProcess[bytes]:
    _validate_docker_authority(docker_binary, docker_host, docker_config)
    return subprocess.run(
        [str(docker_binary), "--host", docker_host, "--config", str(docker_config), *arguments],
        check=check,
        capture_output=True,
        stdin=subprocess.DEVNULL,
        timeout=timeout,
        env=_docker_environment(),
    )


def _bounded_docker_output(result: subprocess.CompletedProcess[bytes], label: str) -> None:
    _require(
        len(result.stdout) <= MAX_DOCKER_COMMAND_OUTPUT_BYTES and len(result.stderr) <= MAX_DOCKER_COMMAND_OUTPUT_BYTES,
        f"{label} output is too large",
    )


def _inspect_local_image(
    docker_binary: Path,
    docker_host: str,
    docker_config: Path,
    image_reference: str,
) -> dict[str, Any]:
    result = _run_docker(
        docker_binary,
        docker_host,
        docker_config,
        ["image", "inspect", image_reference],
        check=True,
        timeout=60,
    )
    _bounded_docker_output(result, "local image inspection")
    inspected = _load_docker_inspection(result.stdout)
    _require(isinstance(inspected, list) and len(inspected) == 1, "local image inspection is not exact")
    _require(isinstance(inspected[0], dict), "local image inspection payload is invalid")
    return inspected[0]


def _local_tag_config_id(
    docker_binary: Path,
    docker_host: str,
    docker_config: Path,
    *,
    repository: str,
    tag: str,
) -> str | None:
    _require(
        repository in {LOCAL_BUILD_REPOSITORY, IMAGE_REPOSITORY}
        and isinstance(tag, str)
        and bool(re.fullmatch(r"[a-z0-9][a-z0-9._-]{0,127}", tag)),
        "local image tag is invalid",
    )
    reference = f"{repository}:{tag}"
    inventory = _run_docker(
        docker_binary,
        docker_host,
        docker_config,
        ["image", "ls", "--no-trunc", "--format", "{{json .}}", repository],
        check=True,
        timeout=60,
    )
    _bounded_docker_output(inventory, "local image inventory")
    matching_ids: set[str] = set()
    for raw_line in inventory.stdout.splitlines():
        if not raw_line.strip():
            continue
        try:
            row = _load_docker_inspection(raw_line)
        except PackageError as exc:
            raise PackageError("local image inventory output is invalid") from exc
        _require(isinstance(row, dict), "local image inventory row is invalid")
        if row.get("Repository") != repository or row.get("Tag") != tag:
            continue
        config_id = row.get("ID")
        _require(
            isinstance(config_id, str)
            and config_id.startswith("sha256:")
            and _is_sha256(config_id.removeprefix("sha256:")),
            "local image inventory config digest is invalid",
        )
        matching_ids.add(config_id)
    _require(len(matching_ids) <= 1, "local image inventory is ambiguous")
    if not matching_ids:
        return None
    inventory_config_id = next(iter(matching_ids))
    inspected = _inspect_local_image(docker_binary, docker_host, docker_config, reference)
    config_id = inspected.get("Id")
    _require(config_id == inventory_config_id, "local image inventory and inspection disagree")
    return inventory_config_id


def _target_repository_digests(inspected: dict[str, Any]) -> list[str]:
    repository_digests = inspected.get("RepoDigests")
    _require(
        repository_digests is None
        or (
            isinstance(repository_digests, list)
            and all(isinstance(item, str) for item in repository_digests)
            and len(repository_digests) == len(set(repository_digests))
        ),
        "local image repository digests are invalid",
    )
    matching = sorted(item for item in (repository_digests or []) if item.startswith(f"{IMAGE_REPOSITORY}@"))
    _require(
        all(IMAGE_REFERENCE.fullmatch(item) is not None for item in matching),
        "local image staging repository digest is invalid",
    )
    return matching


def build_and_push_staging_image(
    repo_root: Path,
    context: Path,
    *,
    docker_binary: Path,
    docker_host: str,
    docker_config: Path,
    output_receipt: Path,
    execute_staging_push: bool,
) -> dict[str, Any]:
    """Build one reviewed context, push one content-bound candidate, and receipt the result."""

    _require(execute_staging_push is True, "staging image push requires explicit execution authorization")
    image_input, build_args = validate_prepared_image_context(context)
    validate_reviewed_context_source(repo_root, context, image_input)
    _validate_docker_authority(docker_binary, docker_host, docker_config)
    _validate_receipt_output_path(output_receipt)
    _require(
        context != output_receipt and context not in output_receipt.parents,
        "build/push receipt must be outside the image context",
    )

    build_tag = f"input-{image_input['input_sha256']}"
    build_reference = f"{LOCAL_BUILD_REPOSITORY}:{build_tag}"
    try:
        with _DockerCacheLock(docker_config):
            _require(
                _local_tag_config_id(
                    docker_binary,
                    docker_host,
                    docker_config,
                    repository=LOCAL_BUILD_REPOSITORY,
                    tag=build_tag,
                )
                is None,
                "local build tag must be absent before build",
            )
            command = [
                "build",
                "--quiet",
                "--platform",
                PLATFORM,
                "--pull=false",
                "--file",
                str(context / "Dockerfile"),
                "--tag",
                build_reference,
            ]
            for name in sorted(build_args):
                command.extend(["--build-arg", f"{name}={build_args[name]}"])
            command.append(str(context))
            built = _run_docker(
                docker_binary,
                docker_host,
                docker_config,
                command,
                check=True,
                timeout=1800,
            )
            _bounded_docker_output(built, "staging image build")
            built_image = _inspect_local_image(docker_binary, docker_host, docker_config, build_reference)
            observed = validate_image_configuration(image_input, built_image)
            build_config_id = observed["config_digest"]
            _require(
                _local_tag_config_id(
                    docker_binary,
                    docker_host,
                    docker_config,
                    repository=LOCAL_BUILD_REPOSITORY,
                    tag=build_tag,
                )
                == build_config_id,
                "local build tag differs from the inspected image",
            )
            _require(
                _target_repository_digests(built_image) == [],
                "local build image already has a staging repository digest",
            )
            post_build_input, post_build_args = validate_prepared_image_context(context)
            _require(
                post_build_input == image_input and post_build_args == build_args,
                "prepared image context changed during build",
            )
            validate_reviewed_context_source(repo_root, context, post_build_input)

            candidate_tag = f"candidate-{build_config_id.removeprefix('sha256:')}"
            candidate_reference = f"{IMAGE_REPOSITORY}:{candidate_tag}"
            _require(
                _local_tag_config_id(
                    docker_binary,
                    docker_host,
                    docker_config,
                    repository=IMAGE_REPOSITORY,
                    tag=candidate_tag,
                )
                is None,
                "content-bound staging candidate tag already exists locally",
            )
            tagged = _run_docker(
                docker_binary,
                docker_host,
                docker_config,
                ["image", "tag", build_reference, candidate_reference],
                check=True,
                timeout=60,
            )
            _bounded_docker_output(tagged, "staging candidate tag")
            _require(
                _local_tag_config_id(
                    docker_binary,
                    docker_host,
                    docker_config,
                    repository=IMAGE_REPOSITORY,
                    tag=candidate_tag,
                )
                == build_config_id,
                "staging candidate tag differs from the built image",
            )
            pre_push_image = _inspect_local_image(
                docker_binary,
                docker_host,
                docker_config,
                candidate_reference,
            )
            _require(
                validate_image_configuration(image_input, pre_push_image) == observed,
                "staging candidate configuration drifted before push",
            )
            _require(
                _target_repository_digests(pre_push_image) == [],
                "staging candidate already has a repository digest before push",
            )
            pushed = _run_docker(
                docker_binary,
                docker_host,
                docker_config,
                ["image", "push", "--quiet", candidate_reference],
                check=True,
                timeout=1800,
            )
            _bounded_docker_output(pushed, "staging image push")
            post_push_image = _inspect_local_image(
                docker_binary,
                docker_host,
                docker_config,
                candidate_reference,
            )
            _require(
                validate_image_configuration(image_input, post_push_image) == observed,
                "staging candidate configuration drifted across push",
            )
            repository_digests = _target_repository_digests(post_push_image)
            _require(len(repository_digests) == 1, "staging push did not produce one exact repository digest")
            registry_reference = repository_digests[0]
            match = IMAGE_REFERENCE.fullmatch(registry_reference)
            _require(match is not None, "staging push repository digest is invalid")
            receipt = build_build_push_receipt(image_input, post_push_image=post_push_image)
    except (OSError, subprocess.SubprocessError, PackageError) as exc:
        raise PackageError("staging image build/push failed") from exc
    publish_build_push_receipt(output_receipt, receipt, image_input=image_input)
    return receipt


def _local_image_reference_config_id(
    docker_binary: Path,
    docker_host: str,
    docker_config: Path,
    image_reference: str,
) -> str | None:
    inventory = _run_docker(
        docker_binary,
        docker_host,
        docker_config,
        ["image", "ls", "--digests", "--no-trunc", "--format", "{{json .}}", IMAGE_REPOSITORY],
        check=True,
        timeout=60,
    )
    _require(len(inventory.stdout) <= MAX_DOCKER_INSPECTION_BYTES, "local image inventory output is too large")
    match = IMAGE_REFERENCE.fullmatch(image_reference)
    _require(match is not None, "local image reference is invalid")
    expected_digest = f"sha256:{match.group(1)}"
    matching_ids: set[str] = set()
    for raw_line in inventory.stdout.splitlines():
        if not raw_line.strip():
            continue
        try:
            row = _load_docker_inspection(raw_line)
        except PackageError as exc:
            raise PackageError("local image inventory output is invalid") from exc
        _require(isinstance(row, dict), "local image inventory row is invalid")
        if row.get("Repository") != IMAGE_REPOSITORY or row.get("Digest") != expected_digest:
            continue
        config_id = row.get("ID")
        _require(
            isinstance(config_id, str)
            and config_id.startswith("sha256:")
            and _is_sha256(config_id.removeprefix("sha256:")),
            "local image inventory config digest is invalid",
        )
        matching_ids.add(config_id)
    _require(len(matching_ids) <= 1, "local image inventory is ambiguous")
    if not matching_ids:
        return None
    inventory_config_id = next(iter(matching_ids))
    result = _run_docker(
        docker_binary,
        docker_host,
        docker_config,
        ["image", "inspect", image_reference],
        check=True,
        timeout=60,
    )
    inspected = _load_docker_inspection(result.stdout)
    _require(isinstance(inspected, list) and len(inspected) == 1, "local image reference inspection is not exact")
    _require(isinstance(inspected[0], dict), "local image reference inspection payload is invalid")
    config_id = inspected[0].get("Id")
    _require(
        isinstance(config_id, str)
        and config_id.startswith("sha256:")
        and _is_sha256(config_id.removeprefix("sha256:")),
        "local image reference config digest is invalid",
    )
    _require(config_id == inventory_config_id, "local image inventory and inspection disagree")
    return config_id


def inspect_registry_image(
    image_input: dict[str, Any],
    build_push_receipt: dict[str, Any],
    *,
    docker_binary: Path,
    docker_host: str,
    docker_config: Path,
) -> dict[str, Any]:
    """Pull and inspect one immutable Artifact Registry image without starting it."""

    validate_image_input(image_input)
    validate_build_push_receipt(build_push_receipt, image_input=image_input)
    image_reference = (
        f"{build_push_receipt['registry']['repository']}@{build_push_receipt['registry']['manifest_digest']}"
    )
    expected_config_digest = build_push_receipt["local_image"]["config_digest"]
    _validate_docker_authority(docker_binary, docker_host, docker_config)
    inspection: dict[str, Any] | None = None
    try:
        with _DockerCacheLock(docker_config):
            _run_docker(
                docker_binary,
                docker_host,
                docker_config,
                ["pull", "--quiet", "--platform", PLATFORM, image_reference],
                check=True,
                timeout=600,
            )
            pulled_config_id = _local_image_reference_config_id(
                docker_binary,
                docker_host,
                docker_config,
                image_reference,
            )
            _require(pulled_config_id is not None, "pulled registry image reference is missing")
            result = _run_docker(
                docker_binary,
                docker_host,
                docker_config,
                ["image", "inspect", image_reference],
                check=True,
                timeout=60,
            )
            inspected = _load_docker_inspection(result.stdout)
            _require(isinstance(inspected, list) and len(inspected) == 1, "registry image inspection is not exact")
            _require(isinstance(inspected[0], dict), "registry image inspection payload is invalid")
            inspection = build_image_inspection(image_input, image_reference, inspected[0])
            _require(
                inspection["config_digest"] == expected_config_digest == pulled_config_id,
                "inspected image differs from the receipted build config digest",
            )
    except (OSError, subprocess.SubprocessError, PackageError) as exc:
        raise PackageError("immutable registry image inspection failed") from exc
    _require(inspection is not None, "immutable registry image inspection failed")
    return inspection


def build_release_descriptor(
    image_input: dict[str, Any],
    inspection: dict[str, Any],
    build_push_receipt: dict[str, Any],
) -> dict[str, Any]:
    validate_image_input(image_input)
    validate_image_inspection(inspection, image_input=image_input)
    validate_build_push_receipt(build_push_receipt, image_input=image_input)
    image_reference = inspection["reference"]
    match = IMAGE_REFERENCE.fullmatch(image_reference)
    _require(match is not None, "release image must be the exact staging repository at a sha256 digest")
    digest = f"sha256:{match.group(1)}"
    labels = expected_image_labels(image_input)
    stable = {
        "schema": RELEASE_SCHEMA,
        "target": _target(),
        "image": {
            "reference": image_reference,
            "digest": digest,
            "config_digest": inspection["config_digest"],
            "platform": PLATFORM,
            "labels": labels,
            "input_sha256": image_input["input_sha256"],
        },
        "build_push_receipt": build_push_receipt,
        "inspection": inspection,
        "image_input": image_input,
    }
    value = {**stable, "release_sha256": canonical_sha256(stable)}
    validate_release_descriptor(value)
    return value


def expected_image_labels(image_input: dict[str, Any]) -> dict[str, str]:
    validate_image_input(image_input)
    return {
        "livingip.artifact.sha256": image_input["runtime"]["artifact_sha256"],
        "livingip.identity.sha256": image_input["identity"]["bundle_sha256"],
        "livingip.input.sha256": image_input["input_sha256"],
        "livingip.runtime": "leoclean-nosend",
        "livingip.target": "gcp-staging",
        "org.opencontainers.image.revision": image_input["runtime"]["teleo_git_head"],
        "org.opencontainers.image.source": "https://github.com/living-ip/teleo-infrastructure",
    }


def validate_release_descriptor(value: dict[str, Any]) -> None:
    _require(
        set(value)
        == {
            "schema",
            "target",
            "image",
            "build_push_receipt",
            "inspection",
            "image_input",
            "release_sha256",
        },
        "release descriptor fields are not exact",
    )
    _require(value.get("schema") == RELEASE_SCHEMA, "release descriptor schema is invalid")
    _require(value.get("target") == _target(), "release target is not exact GCP staging")
    image_input = value.get("image_input")
    _require(isinstance(image_input, dict), "release image input is missing")
    validate_image_input(image_input)
    image = value.get("image")
    _require(
        isinstance(image, dict)
        and set(image) == {"reference", "digest", "config_digest", "platform", "labels", "input_sha256"},
        "release image fields are not exact",
    )
    reference = image.get("reference")
    match = IMAGE_REFERENCE.fullmatch(reference) if isinstance(reference, str) else None
    _require(match is not None, "release image is not digest-pinned in the staging repository")
    _require(image.get("digest") == f"sha256:{match.group(1)}", "release image digest differs from its reference")
    build_push_receipt = value.get("build_push_receipt")
    _require(isinstance(build_push_receipt, dict), "release build/push receipt is missing")
    validate_build_push_receipt(build_push_receipt, image_input=image_input)
    receipt_reference = (
        f"{build_push_receipt['registry']['repository']}@{build_push_receipt['registry']['manifest_digest']}"
    )
    _require(
        receipt_reference == reference,
        "release image differs from its build/push receipt",
    )
    _require(
        build_push_receipt["local_image"]["config_digest"] == image.get("config_digest"),
        "release config digest differs from its build/push receipt",
    )
    inspection = value.get("inspection")
    _require(isinstance(inspection, dict), "release image inspection is missing")
    validate_image_inspection(inspection, image_input=image_input)
    _require(inspection.get("reference") == reference, "release image differs from its inspection")
    _require(image.get("config_digest") == inspection.get("config_digest"), "release config digest drifted")
    _require(
        build_push_receipt["local_image"]["platform"] == inspection.get("platform"),
        "release platform differs from its build/push receipt",
    )
    _require(
        build_push_receipt["local_image"]["labels"] == inspection.get("labels"),
        "release labels differ from its build/push receipt",
    )
    _require(
        build_push_receipt["local_image"]["runtime_config"] == inspection.get("runtime_config"),
        "release runtime config differs from its build/push receipt",
    )
    _require(image.get("platform") == PLATFORM, "release platform is not exact")
    _require(image.get("input_sha256") == image_input["input_sha256"], "release image input binding drifted")
    _require(image.get("labels") == expected_image_labels(image_input), "release image labels are not exact")
    stable = {key: item for key, item in value.items() if key != "release_sha256"}
    _require(value.get("release_sha256") == canonical_sha256(stable), "release descriptor self-hash drifted")
    _assert_no_forbidden_descriptor_markers(value)


def render_systemd_unit(release: dict[str, Any]) -> str:
    validate_release_descriptor(release)
    image = release["image"]["reference"]
    release_sha = release["release_sha256"]
    profile_tmpfs = f"{PROFILE_ROOT}:rw,noexec,nosuid,nodev,size=64m,uid=0,gid=0,mode=0700"
    temp_tmpfs = f"/tmp:rw,noexec,nosuid,nodev,size=64m,uid={RUNTIME_UID},gid={RUNTIME_GID},mode=0700"
    postgres_data_tmpfs = "/var/lib/postgresql/data:ro,noexec,nosuid,nodev,size=64k,uid=0,gid=0,mode=0000"
    exec_start = " ".join(
        [
            "/usr/bin/docker run",
            "--rm",
            f"--name={CONTAINER_NAME}",
            "--pull=never",
            "--read-only",
            "--cap-drop=ALL",
            "--cap-add=CHOWN",
            "--cap-add=SETGID",
            "--cap-add=SETUID",
            "--cap-add=SETPCAP",
            "--security-opt=no-new-privileges:true",
            "--pids-limit=512",
            "--network=bridge",
            "--stop-timeout=30",
            f"--tmpfs={profile_tmpfs}",
            f"--tmpfs={temp_tmpfs}",
            f"--tmpfs={postgres_data_tmpfs}",
            f"--label=livingip.release.sha256={release_sha}",
            image,
            "service",
        ]
    )
    unit = f"""[Unit]
Description=LivingIP leoclean-gcp-nosend staging
Documentation=https://github.com/living-ip/teleo-infrastructure
After=docker.service network-online.target
Requires=docker.service
Wants=network-online.target

[Service]
Type=simple
ExecStart={exec_start}
ExecStop=/usr/bin/docker stop --time=30 {CONTAINER_NAME}
ExecStopPost=-/usr/bin/docker rm --force {CONTAINER_NAME}
Restart=on-failure
RestartSec=5s
TimeoutStartSec=120s
TimeoutStopSec=45s
UMask=0077
NoNewPrivileges=yes
CapabilityBoundingSet=
AmbientCapabilities=
PrivateDevices=yes
PrivateTmp=yes
ProtectSystem=strict
ProtectHome=yes
ProtectKernelTunables=yes
ProtectKernelModules=yes
ProtectKernelLogs=yes
ProtectControlGroups=yes
LockPersonality=yes
RestrictSUIDSGID=yes
RestrictRealtime=yes
RestrictAddressFamilies=AF_UNIX AF_INET AF_INET6
SystemCallArchitectures=native

[Install]
WantedBy=multi-user.target
"""
    _require("--publish" not in unit and " -p " not in unit, "service unit publishes a port")
    _require("docker.sock" not in unit, "service unit mounts the Docker socket")
    _require(
        "EnvironmentFile" not in unit and "LoadCredential" not in unit,
        "service unit adds an unreviewed credential source",
    )
    _require("telegram" not in unit.casefold(), "service unit contains a messaging marker")
    return unit


def _fsync_path(path: Path, *, directory: bool = False) -> None:
    flags = os.O_RDONLY
    if directory and hasattr(os, "O_DIRECTORY"):
        flags |= os.O_DIRECTORY
    if hasattr(os, "O_NOFOLLOW"):
        flags |= os.O_NOFOLLOW
    descriptor = os.open(path, flags)
    try:
        mode = os.fstat(descriptor).st_mode
        if directory and not stat.S_ISDIR(mode):
            raise OSError(errno.ENOTDIR, "fsync target is not a directory", str(path))
        if not directory and not stat.S_ISREG(mode):
            raise OSError(errno.EINVAL, "fsync target is not a regular file", str(path))
        os.fsync(descriptor)
    finally:
        os.close(descriptor)


def _rename_noreplace(source: Path, destination: Path) -> None:
    """Atomically publish one directory without replacing an existing path."""

    library = ctypes.CDLL(None, use_errno=True)
    source_bytes = os.fsencode(source)
    destination_bytes = os.fsencode(destination)
    if sys.platform.startswith("linux"):
        try:
            rename = library.renameat2
        except AttributeError as exc:
            raise OSError(errno.ENOSYS, "renameat2 is unavailable") from exc
        rename.argtypes = [ctypes.c_int, ctypes.c_char_p, ctypes.c_int, ctypes.c_char_p, ctypes.c_uint]
        rename.restype = ctypes.c_int
        result = rename(-100, source_bytes, -100, destination_bytes, 1)
    elif sys.platform == "darwin":
        try:
            rename = library.renamex_np
        except AttributeError as exc:
            raise OSError(errno.ENOSYS, "renamex_np is unavailable") from exc
        rename.argtypes = [ctypes.c_char_p, ctypes.c_char_p, ctypes.c_uint]
        rename.restype = ctypes.c_int
        result = rename(source_bytes, destination_bytes, 0x00000004)
    else:
        raise OSError(errno.ENOSYS, "atomic no-replace rename is unsupported")
    if result != 0:
        error = ctypes.get_errno()
        if error == errno.EEXIST:
            raise FileExistsError(error, "finalize output bundle already exists", str(destination))
        raise OSError(error, os.strerror(error), str(destination))


def _open_private_operator_directory(directory: Path, label: str) -> int:
    _require(directory.is_absolute(), f"{label} must be absolute")
    flags = os.O_RDONLY
    if hasattr(os, "O_DIRECTORY"):
        flags |= os.O_DIRECTORY
    if hasattr(os, "O_NOFOLLOW"):
        flags |= os.O_NOFOLLOW
    try:
        descriptor = os.open(directory, flags)
    except OSError as exc:
        raise PackageError(f"{label} is missing or unsafe") from exc
    try:
        observed = os.fstat(descriptor)
        named = os.stat(directory, follow_symlinks=False)
        _require(
            stat.S_ISDIR(observed.st_mode)
            and (observed.st_dev, observed.st_ino) == (named.st_dev, named.st_ino)
            and observed.st_uid == os.geteuid()
            and observed.st_mode & 0o022 == 0,
            f"{label} must be an operator-owned directory without group/other write access",
        )
    except (OSError, PackageError):
        os.close(descriptor)
        raise
    return descriptor


def _directory_entry_stat(directory_descriptor: int, name: str) -> os.stat_result | None:
    try:
        return os.stat(name, dir_fd=directory_descriptor, follow_symlinks=False)
    except FileNotFoundError:
        return None


def _fsync_directory_descriptor(directory_descriptor: int) -> None:
    observed = os.fstat(directory_descriptor)
    if not stat.S_ISDIR(observed.st_mode):
        raise OSError(errno.ENOTDIR, "fsync target is not a directory")
    os.fsync(directory_descriptor)


def _validate_receipt_output_path(output_receipt: Path) -> None:
    _require(
        output_receipt.is_absolute() and output_receipt.name not in {"", ".", ".."},
        "build/push receipt output path is invalid",
    )
    parent_descriptor = _open_private_operator_directory(
        output_receipt.parent,
        "build/push receipt parent",
    )
    try:
        _require(
            _directory_entry_stat(parent_descriptor, output_receipt.name) is None,
            "build/push receipt output already exists",
        )
    finally:
        os.close(parent_descriptor)


def _verify_receipt_entry(
    directory_descriptor: int,
    name: str,
    expected_identity: tuple[int, int],
    *,
    expected_size: int,
    expected_links: int,
    label: str,
) -> os.stat_result:
    try:
        observed = os.stat(name, dir_fd=directory_descriptor, follow_symlinks=False)
    except OSError as exc:
        raise PackageError(f"{label} cannot identify output") from exc
    _require(
        stat.S_ISREG(observed.st_mode)
        and (observed.st_dev, observed.st_ino) == expected_identity
        and observed.st_uid == os.geteuid()
        and stat.S_IMODE(observed.st_mode) == 0o600
        and observed.st_size == expected_size
        and observed.st_nlink == expected_links,
        f"{label} identity or posture drifted",
    )
    return observed


def load_private_json(path: Path, label: str) -> dict[str, Any]:
    """Read one authority-bearing JSON file through a private dirfd and stable inode."""

    _require(path.is_absolute() and path.name not in {"", ".", ".."}, f"{label} path is invalid")
    parent_descriptor = _open_private_operator_directory(path.parent, f"{label} parent")
    descriptor: int | None = None
    try:
        flags = os.O_RDONLY
        if hasattr(os, "O_NOFOLLOW"):
            flags |= os.O_NOFOLLOW
        descriptor = os.open(path.name, flags, dir_fd=parent_descriptor)
        before = os.fstat(descriptor)
        _require(
            stat.S_ISREG(before.st_mode)
            and before.st_uid == os.geteuid()
            and before.st_mode & 0o022 == 0
            and before.st_nlink == 2
            and before.st_size <= MAX_BUILD_PUSH_RECEIPT_BYTES,
            f"{label} file posture is unsafe",
        )
        chunks: list[bytes] = []
        remaining = MAX_BUILD_PUSH_RECEIPT_BYTES + 1
        while remaining:
            chunk = os.read(descriptor, min(65536, remaining))
            if not chunk:
                break
            chunks.append(chunk)
            remaining -= len(chunk)
        payload = b"".join(chunks)
        _require(len(payload) <= MAX_BUILD_PUSH_RECEIPT_BYTES, f"{label} is too large")
        after = os.fstat(descriptor)
        named = os.stat(path.name, dir_fd=parent_descriptor, follow_symlinks=False)
        _require(
            (before.st_dev, before.st_ino, before.st_size, before.st_mtime_ns, before.st_ctime_ns)
            == (after.st_dev, after.st_ino, after.st_size, after.st_mtime_ns, after.st_ctime_ns)
            and (after.st_dev, after.st_ino) == (named.st_dev, named.st_ino)
            and len(payload) == after.st_size,
            f"{label} changed while it was read",
        )
    except (OSError, UnicodeError) as exc:
        raise PackageError(f"cannot read {label}: {type(exc).__name__}") from exc
    finally:
        if descriptor is not None:
            os.close(descriptor)
        os.close(parent_descriptor)
    try:
        encoded = payload.decode("utf-8")
    except UnicodeError as exc:
        raise PackageError(f"cannot read {label}: {type(exc).__name__}") from exc
    value = _strict_json_loads(encoded, label)
    _require(isinstance(value, dict), f"{label} must be a JSON object")
    return value


def publish_build_push_receipt(
    output_receipt: Path,
    receipt: dict[str, Any],
    *,
    image_input: dict[str, Any],
) -> None:
    """Atomically publish one validated, self-hashed build/push receipt."""

    validate_build_push_receipt(receipt, image_input=image_input)
    _validate_receipt_output_path(output_receipt)
    try:
        payload = (json.dumps(receipt, allow_nan=False, indent=2, sort_keys=True) + "\n").encode("utf-8")
    except (TypeError, ValueError) as exc:
        raise PackageError("cannot encode build/push receipt JSON") from exc
    staged_name = f".{output_receipt.name}.{uuid.uuid4().hex}.tmp"
    parent_descriptor = _open_private_operator_directory(
        output_receipt.parent,
        "build/push receipt parent",
    )
    descriptor: int | None = None
    identity: tuple[int, int] | None = None
    try:
        _require(
            _directory_entry_stat(parent_descriptor, output_receipt.name) is None,
            "build/push receipt output already exists",
        )
        flags = os.O_WRONLY | os.O_CREAT | os.O_EXCL
        if hasattr(os, "O_NOFOLLOW"):
            flags |= os.O_NOFOLLOW
        descriptor = os.open(staged_name, flags, 0o600, dir_fd=parent_descriptor)
        observed = os.fstat(descriptor)
        identity = (observed.st_dev, observed.st_ino)
        os.fchmod(descriptor, 0o600)
        with os.fdopen(descriptor, "wb", closefd=False) as handle:
            handle.write(payload)
            handle.flush()
        os.fsync(descriptor)
        _verify_receipt_entry(
            parent_descriptor,
            staged_name,
            identity,
            expected_size=len(payload),
            expected_links=1,
            label="staged build/push receipt",
        )
        staged_value = _strict_json_loads(payload.decode("utf-8"), "staged build/push receipt")
        _require(isinstance(staged_value, dict), "staged build/push receipt must be a JSON object")
        validate_build_push_receipt(staged_value, image_input=image_input)
        os.link(
            staged_name,
            output_receipt.name,
            src_dir_fd=parent_descriptor,
            dst_dir_fd=parent_descriptor,
            follow_symlinks=False,
        )
        _verify_receipt_entry(
            parent_descriptor,
            output_receipt.name,
            identity,
            expected_size=len(payload),
            expected_links=2,
            label="published build/push receipt",
        )
        _verify_receipt_entry(
            parent_descriptor,
            staged_name,
            identity,
            expected_size=len(payload),
            expected_links=2,
            label="retained build/push receipt link",
        )
        _fsync_directory_descriptor(parent_descriptor)
    except (OSError, PackageError) as exc:
        if isinstance(exc, PackageError):
            raise
        raise PackageError("cannot publish build/push receipt") from exc
    finally:
        if descriptor is not None:
            os.close(descriptor)
        os.close(parent_descriptor)


def _remove_published_bundle(output_bundle: Path, expected_identity: tuple[int, int]) -> None:
    try:
        observed = output_bundle.stat(follow_symlinks=False)
    except OSError as exc:
        raise PackageError("finalize bundle rollback cannot identify published output") from exc
    _require(
        stat.S_ISDIR(observed.st_mode) and (observed.st_dev, observed.st_ino) == expected_identity,
        "finalize bundle rollback refused to remove a replaced output",
    )
    try:
        shutil.rmtree(output_bundle)
        _fsync_path(output_bundle.parent, directory=True)
    except OSError as exc:
        raise PackageError("finalize bundle rollback failed") from exc
    _require(
        not output_bundle.exists() and not output_bundle.is_symlink(),
        "finalize bundle rollback did not pass",
    )


def _remove_staged_bundle(staged_bundle: Path, expected_identity: tuple[int, int]) -> None:
    try:
        observed = staged_bundle.stat(follow_symlinks=False)
    except OSError as exc:
        raise PackageError("finalize temporary cleanup cannot identify staged output") from exc
    _require(
        stat.S_ISDIR(observed.st_mode) and (observed.st_dev, observed.st_ino) == expected_identity,
        "finalize temporary cleanup refused to remove a replaced output",
    )
    try:
        shutil.rmtree(staged_bundle)
        _fsync_path(staged_bundle.parent, directory=True)
    except OSError as exc:
        raise PackageError("finalize temporary cleanup failed") from exc
    _require(
        not staged_bundle.exists() and not staged_bundle.is_symlink(),
        "finalize temporary cleanup did not pass",
    )


def publish_finalize_bundle(output_bundle: Path, release: dict[str, Any]) -> None:
    """Publish release.json and the unit through one directory rename."""

    validate_release_descriptor(release)
    _require(not output_bundle.exists() and not output_bundle.is_symlink(), "finalize output bundle already exists")
    try:
        release_payload = json.dumps(release, allow_nan=False, indent=2, sort_keys=True) + "\n"
    except (TypeError, ValueError) as exc:
        raise PackageError("cannot encode finalized release JSON") from exc
    unit_payload = render_systemd_unit(release)
    staged_bundle = output_bundle.with_name(f".{output_bundle.name}.{uuid.uuid4().hex}.tmp")
    staged_created = False
    published = False
    staged_identity: tuple[int, int] | None = None
    try:
        _require(
            output_bundle.is_absolute() and output_bundle.parent.is_dir() and not output_bundle.parent.is_symlink(),
            "finalize output parent must be an existing safe absolute directory",
        )
        staged_bundle.mkdir(mode=0o700)
        staged_created = True
        staged_stat = staged_bundle.stat(follow_symlinks=False)
        staged_identity = (staged_stat.st_dev, staged_stat.st_ino)
        staged_release = staged_bundle / "release.json"
        staged_unit = staged_bundle / SERVICE
        staged_release.write_text(release_payload, encoding="utf-8")
        staged_release.chmod(0o644)
        staged_unit.write_text(unit_payload, encoding="utf-8")
        staged_unit.chmod(0o644)
        validate_release_descriptor(load_json(staged_release, "staged release descriptor"))
        _require(staged_unit.read_text(encoding="utf-8") == render_systemd_unit(release), "staged unit drifted")
        _fsync_path(staged_release)
        _fsync_path(staged_unit)
        staged_bundle.chmod(0o755)
        _fsync_path(staged_bundle, directory=True)
        _rename_noreplace(staged_bundle, output_bundle)
        published = True
        _fsync_path(output_bundle.parent, directory=True)
    except OSError as exc:
        if published:
            _require(staged_identity is not None, "finalize bundle rollback identity is missing")
            _remove_published_bundle(output_bundle, staged_identity)
        raise PackageError("cannot publish finalized bundle") from exc
    finally:
        if staged_created and (staged_bundle.exists() or staged_bundle.is_symlink()):
            _require(staged_identity is not None, "finalize temporary cleanup identity is missing")
            _remove_staged_bundle(staged_bundle, staged_identity)


def _git_commit(repo_root: Path) -> str:
    try:
        revision = subprocess.run(
            ["git", "-C", str(repo_root), "rev-parse", "--verify", "HEAD^{commit}"],
            check=True,
            capture_output=True,
            text=True,
        ).stdout.strip()
    except (OSError, subprocess.CalledProcessError) as exc:
        raise PackageError("cannot determine source revision") from exc
    _require(_is_revision(revision), "source revision is invalid")
    return revision


def _verify_git_source_closure(
    repo_root: Path,
    revision: str,
    expected_sha256: Mapping[str, str],
) -> None:
    """Require every executable package input to equal its reviewed Git blob and mode."""

    _require(_is_revision(revision), "source revision is invalid")
    _require(repo_root.is_dir() and not repo_root.is_symlink(), "source repository root is missing or unsafe")
    try:
        top_level = subprocess.run(
            ["git", "-C", str(repo_root), "rev-parse", "--show-toplevel"],
            check=True,
            capture_output=True,
            text=True,
        ).stdout.strip()
    except (OSError, subprocess.CalledProcessError) as exc:
        raise PackageError("cannot determine source repository root") from exc
    _require(Path(top_level).resolve() == repo_root.resolve(), "source root is not the Git repository root")
    _require(_git_commit(repo_root) == revision, "source checkout differs from the reviewed revision")
    relative_paths = sorted(expected_sha256)
    _require(relative_paths and len(relative_paths) == len(expected_sha256), "reviewed source closure is invalid")
    try:
        dirty = subprocess.run(
            [
                "git",
                "-C",
                str(repo_root),
                "status",
                "--porcelain=v1",
                "-z",
                "--untracked-files=all",
                "--",
                *relative_paths,
            ],
            check=True,
            capture_output=True,
        ).stdout
    except (OSError, subprocess.CalledProcessError) as exc:
        raise PackageError("cannot inspect reviewed source closure") from exc
    _require(not dirty, "reviewed source closure is dirty")
    for relative in relative_paths:
        _require(_is_sha256(expected_sha256[relative]), f"reviewed source hash is invalid: {relative}")
        _require(":" not in relative and "\x00" not in relative, "source path is unsafe")
        path = _safe_bound_file(repo_root, relative, label=f"reviewed source: {relative}")
        try:
            entry = subprocess.run(
                ["git", "-C", str(repo_root), "ls-tree", "-z", revision, "--", relative],
                check=True,
                capture_output=True,
            ).stdout
            _require(entry.endswith(b"\x00") and entry.count(b"\x00") == 1, f"source is not tracked: {relative}")
            metadata, tracked_path = entry[:-1].split(b"\t", 1)
            mode, object_type, object_id = metadata.split(b" ", 2)
            _require(
                tracked_path.decode("utf-8") == relative and object_type == b"blob" and mode in {b"100644", b"100755"},
                f"reviewed source entry is invalid: {relative}",
            )
            committed = subprocess.run(
                ["git", "-C", str(repo_root), "cat-file", "blob", object_id.decode("ascii")],
                check=True,
                capture_output=True,
            ).stdout
        except (OSError, subprocess.CalledProcessError, UnicodeDecodeError, ValueError) as exc:
            raise PackageError(f"cannot verify reviewed source: {relative}") from exc
        _require(path.read_bytes() == committed, f"source differs from reviewed revision: {relative}")
        _require(
            hashlib.sha256(committed).hexdigest() == expected_sha256[relative],
            f"source hash differs from package binding: {relative}",
        )
        executable = bool(stat.S_IMODE(path.stat().st_mode) & 0o111)
        _require(executable == (mode == b"100755"), f"source mode differs from reviewed revision: {relative}")


def _git_head(repo_root: Path) -> str:
    revision = _git_commit(repo_root)
    expected = {
        relative: sha256_file(_safe_bound_file(repo_root, relative, label=f"reviewed source: {relative}"))
        for relative in PACKAGE_SOURCE_PATHS
    }
    _verify_git_source_closure(repo_root, revision, expected)
    return revision


def _parse_args(argv: list[str] | None = None) -> argparse.Namespace:
    parser = argparse.ArgumentParser(description=__doc__)
    subparsers = parser.add_subparsers(dest="command", required=True)
    prepare = subparsers.add_parser("prepare")
    prepare.add_argument("--repo-root", type=Path, default=Path(__file__).resolve().parents[1])
    prepare.add_argument("--artifact", type=Path, required=True)
    prepare.add_argument("--receipt", type=Path, required=True)
    prepare.add_argument("--identity", type=Path, required=True)
    prepare.add_argument("--identity-source-root", type=Path, required=True)
    prepare.add_argument("--output", type=Path, required=True)
    compile_identity = subparsers.add_parser("compile-identity")
    compile_identity.add_argument("--repo-root", type=Path, default=Path(__file__).resolve().parents[1])
    compile_identity.add_argument("--artifact", type=Path, required=True)
    compile_identity.add_argument("--receipt", type=Path, required=True)
    compile_identity.add_argument("--identity-manifest", type=Path, required=True)
    compile_identity.add_argument("--identity-source-root", type=Path, required=True)
    compile_identity.add_argument("--output", type=Path, required=True)
    identity_contract = subparsers.add_parser("identity-contract")
    identity_contract.add_argument("--repo-root", type=Path, default=Path(__file__).resolve().parents[1])
    identity_contract.add_argument("--artifact", type=Path, required=True)
    identity_contract.add_argument("--receipt", type=Path, required=True)
    identity_contract.add_argument("--output", type=Path, required=True)
    generate_identity = subparsers.add_parser("generate-identity-manifest")
    generate_identity.add_argument("--repo-root", type=Path, default=Path(__file__).resolve().parents[1])
    generate_identity.add_argument("--artifact", type=Path, required=True)
    generate_identity.add_argument("--receipt", type=Path, required=True)
    generate_identity.add_argument("--database-fingerprint", type=Path, required=True)
    generate_identity.add_argument("--constitution", type=Path, required=True)
    generate_identity.add_argument("--database-identity", type=Path, required=True)
    generate_identity.add_argument("--identity-source-root", type=Path, required=True)
    generate_identity.add_argument("--output", type=Path, required=True)
    build_push = subparsers.add_parser("build-push")
    build_push.add_argument("--repo-root", type=Path, default=Path(__file__).resolve().parents[1])
    build_push.add_argument("--image-context", type=Path, required=True)
    build_push.add_argument("--docker-binary", type=Path, required=True)
    build_push.add_argument("--docker-host", required=True)
    build_push.add_argument("--docker-config", type=Path, required=True)
    build_push.add_argument("--output-receipt", type=Path, required=True)
    build_push.add_argument("--execute-staging-push", action="store_true")
    finalize = subparsers.add_parser("finalize")
    finalize.add_argument("--image-input", type=Path, required=True)
    finalize.add_argument("--build-push-receipt", type=Path, required=True)
    finalize.add_argument("--docker-binary", type=Path, required=True)
    finalize.add_argument("--docker-host", required=True)
    finalize.add_argument("--docker-config", type=Path, required=True)
    finalize.add_argument("--output-bundle", type=Path, required=True)
    validate = subparsers.add_parser("validate")
    validate.add_argument("--release", type=Path, required=True)
    return parser.parse_args(argv)


def main(argv: list[str] | None = None) -> int:
    try:
        args = _parse_args(argv)
        if args.command == "prepare":
            repo_root = args.repo_root.resolve()
            result = prepare_context(
                repo_root=repo_root,
                artifact=args.artifact.resolve(),
                receipt=args.receipt.resolve(),
                identity=args.identity.resolve(),
                identity_source_root=args.identity_source_root.resolve(),
                output=args.output.absolute(),
                source_revision=_git_head(repo_root),
            )
        elif args.command == "compile-identity":
            repo_root = args.repo_root.resolve()
            revision = _git_head(repo_root)
            artifact = args.artifact.resolve()
            receipt = args.receipt.resolve()
            runtime = validate_artifact(artifact, receipt, revision)
            contract = build_expected_identity_contract(
                artifact=artifact,
                receipt_path=receipt,
                runtime=runtime,
                source_revision=revision,
                source_root=repo_root,
            )
            binding = compile_identity_bundle(
                manifest=load_json(args.identity_manifest.resolve(), "identity manifest"),
                expected_identity_contract=contract,
                source_root=args.identity_source_root.resolve(),
                output=args.output.absolute(),
            )
            result = {"schema": COMPILED_IDENTITY_SCHEMA, **binding}
        elif args.command == "identity-contract":
            repo_root = args.repo_root.resolve()
            revision = _git_head(repo_root)
            artifact = args.artifact.resolve()
            receipt = args.receipt.resolve()
            runtime = validate_artifact(artifact, receipt, revision)
            result = build_expected_identity_contract(
                artifact=artifact,
                receipt_path=receipt,
                runtime=runtime,
                source_revision=revision,
                source_root=repo_root,
            )
            _require(not args.output.exists() and not args.output.is_symlink(), "identity contract output exists")
            args.output.parent.mkdir(parents=True, exist_ok=True)
            write_json(args.output, result)
        elif args.command == "generate-identity-manifest":
            repo_root = args.repo_root.resolve()
            result = build_synthetic_identity_manifest(
                artifact=args.artifact.resolve(),
                receipt=args.receipt.resolve(),
                source_revision=_git_head(repo_root),
                contract_source_root=repo_root,
                identity_source_root=args.identity_source_root.resolve(),
                database_fingerprint=args.database_fingerprint.resolve(),
                constitution=args.constitution.resolve(),
                database_identity=args.database_identity.resolve(),
            )
            _require(not args.output.exists() and not args.output.is_symlink(), "identity manifest output exists")
            args.output.parent.mkdir(parents=True, exist_ok=True)
            write_json(args.output, result, mode=0o600)
        elif args.command == "build-push":
            result = build_and_push_staging_image(
                args.repo_root.resolve(),
                args.image_context.absolute(),
                docker_binary=args.docker_binary,
                docker_host=args.docker_host,
                docker_config=args.docker_config,
                output_receipt=args.output_receipt.absolute(),
                execute_staging_push=args.execute_staging_push,
            )
        elif args.command == "finalize":
            image_input = load_json(args.image_input, "image input")
            build_push_receipt = load_private_json(
                args.build_push_receipt.absolute(),
                "build/push receipt",
            )
            inspection = inspect_registry_image(
                image_input,
                build_push_receipt,
                docker_binary=args.docker_binary,
                docker_host=args.docker_host,
                docker_config=args.docker_config,
            )
            result = build_release_descriptor(image_input, inspection, build_push_receipt)
            publish_finalize_bundle(args.output_bundle, result)
        else:
            result = load_json(args.release, "release descriptor")
            validate_release_descriptor(result)
        print(json.dumps({"status": "pass", "schema": result["schema"]}, sort_keys=True))
        return 0
    except PackageError as exc:
        print(json.dumps({"status": "fail", "error": str(exc)}, sort_keys=True), file=sys.stderr)
        return 65


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