Merge pull request #206 from living-ip/feat/leoclean-image-inspection

Bind GCP staging release to an immutable image digest

docs/gcp-leoclean-nosend-service-package.md

@@ -136,7 +136,7 @@ remain permitted because they cannot enter this image context.
 Use the exact values in `build-args.json` as Docker build arguments. The image
 build itself revalidates that those values equal `image-input.json`.
 
-## Prepare one digest-bound release candidate
+## Finalize one inspected digest-bound release
 
 After an authorized build and Artifact Registry push returns the immutable
 image digest:
@@ -146,24 +146,66 @@ python3 ops/gcp_leoclean_nosend_package.py finalize \
   --image-input /tmp/leoclean-nosend-image-context/image-input.json \
   --image-reference \
     europe-west6-docker.pkg.dev/teleo-501523/teleo/leoclean-nosend-staging@sha256:<64-hex-digest> \
-  --output-release /tmp/leoclean-nosend-release.json \
-  --output-unit /tmp/leoclean-gcp-nosend.service
+  --expected-revision <reviewed-40-hex-git-revision> \
+  --expected-input-sha256 <reviewed-image-input-sha256> \
+  --expected-config-digest sha256:<reviewed-build-config-digest> \
+  --docker-binary /usr/bin/docker \
+  --docker-host unix:///var/run/docker.sock \
+  --docker-config /home/operator/.docker \
+  --output-bundle /tmp/leoclean-nosend-release
 ```
 
-This command validates the supplied digest syntax, target repository, expected
-descriptor labels, and generated unit. It does **not** inspect Artifact
-Registry and therefore does not attest that the remote digest actually has the
-declared platform, labels, input hash, or source revision. The candidate unit
-must not be installed until a separate read-only registry-inspection receipt
-binds that exact digest to `linux/amd64`, its immutable config digest, the
-required labels, the image-input hash, and the reviewed source revision.
+This command uses Docker to pull the exact Artifact Registry digest for
+`linux/amd64`, inspects the immutable local image, and refuses to create either
+bundle file unless the requested repository digest is present and the platform,
+config digest, complete required label set, runtime-bearing image configuration,
+image-input hash, and separately supplied reviewed source revision all match.
+The config digest is a separate, manually reviewed trust input from the
+authorized build output; copying it from this command's own registry inspection
+would defeat the independent binding. This repository does not yet produce a
+structured build-receipt artifact, so this slice must not be described as
+proving build-receipt provenance. That producer belongs to the separately
+reviewed build/push slice before live finalization.
+The normalized inspection is embedded in strict release schema
+`livingip.leocleanNoSendRelease.v2`; unit generation revalidates it. Release v1
+has no deployed consumer and is intentionally rejected; regenerate any local
+v1 artifact. For this Linux-only target, inspection normalizes the deprecated
+Windows-only Docker `ArgsEscaped` field while still requiring the exact
+entrypoint and command arrays.
+
+The output parent must already exist as an absolute, non-symlink directory.
+The output is one atomically published directory containing `release.json` and
+`leoclean-gcp-nosend.service`. Both files are rendered, strictly JSON-decoded,
+revalidated, and fsynced in a sibling temporary directory before a
+platform-native no-replace directory rename. An existing or concurrently
+created destination is never overwritten. The staged directory and parent are
+fsynced; a post-publication durability failure removes only the same published
+inode and re-fsyncs the parent. Any failure leaves no partial final bundle.
+
+The Docker binary, config directory, and local `unix://` daemon endpoint are
+explicit inputs; inherited `DOCKER_*` overrides are removed. The pull does not
+start a container, install a unit, restart a service, or contact Cloud SQL.
+Registry authentication and this pull require a separate live-read approval.
+While holding an advisory lock on the explicit Docker config directory, the
+command pulls and records the dynamically observed config ID. It intentionally
+retains the exact repository-digest reference: Docker's image cache belongs to
+the daemon, so a config-directory lock cannot prove that another client did not
+start relying on the same immutable reference after the precheck. Repeated runs
+re-pull and re-inspect the same digest without deleting shared cache state.
+Content-addressable layers and the verified digest reference may remain. A
+failed pull or inspection returns a bounded error without forwarding Docker
+stderr.
 
 Tags, alternate repositories, alternate projects/instances/service accounts,
-extra descriptor fields, and mismatched descriptor labels are rejected.
+extra descriptor fields, wrong-architecture images, missing repository-digest
+bindings, mismatched image labels, duplicate JSON keys, and non-finite JSON
+values are rejected.
 
 ## Offline evidence and live gate
 
-Offline tests prove descriptor strictness, artifact and synthetic identity binding,
+Offline tests prove descriptor and inspection strictness, explicit local Docker
+authority, non-destructive cache reuse, atomic bundle publication, artifact and
+synthetic identity binding,
 digest-only images, pinned runtime versions, a secret-free context, the
 capability-drop contract, no published ports, no Docker-socket mount, no
 environment file, and no production/VPS/Telegram target. CI also builds exactly
@@ -180,9 +222,9 @@ records whether `linux/amd64` ran natively or through cross-platform emulation.
 Pinned base images and build cache may remain.
 
 The entrypoint command rejection is not a real Hermes tool invocation, and the
-bridge-network smoke does not prove general network-egress denial. These tests
-also do not prove canonical identity, Cloud SQL permissions, Artifact Registry
-digest contents,
+bridge-network smoke does not prove general network-egress denial. The offline
+inspection tests use exact Docker JSON fixtures; they do not prove canonical
+identity, Cloud SQL permissions, or the live Artifact Registry digest contents,
 or that `teleo-staging-1` currently has a compatible container
 engine, enough disk, the expected architecture, the expected service account,
 or an unused unit name. Those are the next read-only live preflight. If the
@@ -191,17 +233,22 @@ stop instead of falling back to the production sync script or a mutable host
 virtual environment.
 
 Publishing the draft branch and stacked PR is only a review handoff. The next
-deployment-evidence slice must add immutable Artifact Registry inspection and
-make installation fail closed without its receipt. Live installation, Artifact
-Registry publication, service restart, IAM/secret
-changes, Cloud SQL access, and proposal staging remain separate approvals.
+deployment-evidence step must execute one authorized remote registry read plus
+the retained local exact-reference cache mutation against the staging digest.
+A later installer slice must prove installation and rollback consume the same
+receipt. Live installation, Artifact Registry publication, service restart,
+IAM/secret changes, Cloud SQL access, and proposal staging remain separate
+approvals.
 
 ## Rollback and sunset
 
-The rollback boundary is one release descriptor, one image digest, and the
-dedicated no-send unit. Rollback restores the prior descriptor/unit and prior
-digest; it must never restart or edit the production Leo, Telegram, or VPS
-services.
+This slice's rollback boundary is removal of the unpublished atomic output
+bundle. The verified repository-digest reference and content-addressable layers
+may remain in the local daemon cache so the finalizer never deletes shared cache
+state it cannot exclusively own. It installs no unit and changes no running
+service. Restoring a prior descriptor, unit, and digest
+belongs to the later installer slice, which must never restart or edit the
+production Leo, Telegram, or VPS services.
 
 The fresh tmpfs profile intentionally has no session continuity. That is
 appropriate for isolated gateway/database and fresh-turn parity proof, but it

docs/workstreams/gcp-leoclean/goals.md

@@ -20,39 +20,59 @@ cut over safely and retire the VPS after soak.
 - Complete: PR #179 least-privilege PostgreSQL authority is merged on main.
 - Complete: the revision-bound, structurally no-send runtime from PR #183 is
   merged into `main` at `2d88c9765bd9b43641d1efb04a1118c5ce0bce39`.
-- Rebuilt from current `main`: PR #182 contains only the four functional
-  no-send service-package commits. The two stale proof-only commits were not
-  replayed; this file is the single replacement evidence update.
+- Complete: PR #182 head
+  `3e59845454f5315e955734db7fe598ec4ba4cf70` was merged with merge commit
+  `94a8160ceeb6821fd113f4c843f024d5de99939a` after all five checks passed.
+- Complete: PR #202 remained the exact reviewed three-file QEMU/PID-1
+  correction at `ffe170007d0e20d77a5208263e5b33fc999139fc` after retargeting
+  to `main`, then merged with merge commit
+  `383996a43beebef57cf84a2ced53213e28a41dca` after all five checks passed.
 - PR #193 is source material only. It must not be merged wholesale; only the
   required identity-binding and OCI-lifecycle corrections belong in the narrow
   #182/#183 stack.
-- Complete in the rebuilt code candidate: the narrow package correction binds
-  every packaged and synthetic-identity source byte to its reviewed Git blob;
-  container health binds the packaged/live config plus the exact sealed
-  no-send tool and plugin surface; and the disposable image smoke binds the
-  bootstrap user, entrypoint, command, healthcheck, and stop signal.
-- Separate follow-up: candidate release finalization does not inspect Artifact
-  Registry. A narrow deployment-evidence slice must bind the immutable remote
-  digest, `linux/amd64` manifest/config, labels, input hash, and source revision
-  before any unit can be installed.
-- Current gate: obtain a clean exact-head CI run and exact-revision human
-  review. Rebuilding or publishing the branch does not authorize merge or
-  deployment.
+- Complete through PR #182 head `3e59845`: the narrow package
+  correction binds every packaged and synthetic-identity source byte to its
+  reviewed Git blob; container health binds the packaged/live config plus the
+  exact sealed no-send tool and plugin surface; and the disposable image smoke
+  also binds the implicit root bootstrap user, entrypoint, command, healthcheck,
+  and stop signal. The focused package/runtime/identity/OCI suite passed 204
+  tests, the repository suite passed 2,338 with 3 skips, and one exact-revision
+  synthetic `linux/amd64` OCI lifecycle proved UID/GID `65532`, zero runtime
+  capabilities, no-send health, and complete candidate cleanup on the arm64 Mac
+  Docker daemon. No GCP or live service was contacted.
+- In progress from exact current-main base
+  `383996a43beebef57cf84a2ced53213e28a41dca`: make release finalization
+  pull and inspect the exact immutable staging digest and refuse release/unit
+  output unless the repository binding, `linux/amd64` platform, config digest,
+  complete labels, runtime configuration, input hash, and source revision all
+  match. Trust-bound JSON is strict, release publication is atomic and
+  no-clobber, and failure rollback re-fsyncs its output parent. This slice does
+  not build, push, install, or deploy an image and has not accessed Artifact
+  Registry or GCP.
+- Current gate: finish exact-revision offline tests, then stop for separate
+  branch-publication and human-review approval. Remote registry inspection is
+  a later, separately authorized live-read gate.
 - Not yet live-proven: staging VM/container engine, IAM/Secret Manager, Cloud
   SQL effective identity, restart behavior, or model/database parity.
 
 ## Remaining order
 
-1. Obtain exact-revision human review for the no-send service-package stack.
-2. Run the read-only `teleo-staging-1` host/engine/service-account preflight.
-3. Provision the scoped Cloud SQL role/secret and deploy the separate staging
+1. Finish, publish, and obtain exact-revision human review for immutable image
+   inspection and release finalization.
+2. Add the structured build/push receipt producer that supplies the independently
+   reviewed manifest and config digests consumed by finalization.
+3. Add the separate fail-closed installer/restart/verification/rollback slice
+   that consumes the exact inspection-bound release bundle.
+4. Run the read-only Artifact Registry and
+   `teleo-staging-1` host/engine/service-account preflight.
+5. Provision the scoped Cloud SQL role/secret and deploy the separate staging
    service.
-4. Prove effective identity, reads, function-only proposal staging, denied
+6. Prove effective identity, reads, function-only proposal staging, denied
    writes/escalation, receipts, restart stability, and rollback.
-5. Run behavioral and database parity backtesting.
-6. Connect the protected Observatory to canonical claims, proposals, pgvector
+7. Run behavioral and database parity backtesting.
+8. Connect the protected Observatory to canonical claims, proposals, pgvector
    projection, and runtime receipts.
-7. Cut over, soak, restore-test, revoke VPS credentials, and retire the VPS.
+9. Cut over, soak, restore-test, revoke VPS credentials, and retire the VPS.
 
 ## Current live-action boundary
 

ops/gcp_leoclean_nosend_package.py

@@ -4,14 +4,19 @@
 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
@@ -27,7 +32,8 @@ if str(CONTRACT_ROOT) not in sys.path:
 from scripts import leo_identity_manifest as strict_identity_manifest  # noqa: E402
 
 IMAGE_INPUT_SCHEMA = "livingip.leocleanNoSendImageInput.v2"
-RELEASE_SCHEMA = "livingip.leocleanNoSendRelease.v1"
+IMAGE_INSPECTION_SCHEMA = "livingip.leocleanNoSendImageInspection.v1"
+RELEASE_SCHEMA = "livingip.leocleanNoSendRelease.v2"
 ARTIFACT_MANIFEST_SCHEMA = "livingip.leocleanNoSendArtifactManifest.v1"
 ARTIFACT_RECEIPT_SCHEMA = "livingip.leocleanNoSendArtifactVerification.v1"
 IDENTITY_MANIFEST_SCHEMA = "livingip.leoIdentityManifest.v1"
@@ -66,6 +72,57 @@ RUNTIME_ROOT = "/opt/livingip/leoclean-nosend"
 PROFILE_ROOT = "/run/leoclean-profile"
 RUNTIME_UID = 65532
 RUNTIME_GID = 65532
+MAX_DOCKER_INSPECTION_BYTES = 1024 * 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}$")
@@ -132,21 +189,57 @@ def sha256_file(path: Path) -> str:
 
 
 def canonical_sha256(value: Any) -> str:
-    encoded = json.dumps(value, ensure_ascii=True, separators=(",", ":"), sort_keys=True).encode("utf-8")
+    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:
-        value = json.loads(path.read_text(encoding="utf-8"))
-    except (OSError, json.JSONDecodeError) as exc:
+        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:
-    path.write_text(json.dumps(value, indent=2, sort_keys=True) + "\n", encoding="utf-8")
+    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)
 
 
@@ -1278,8 +1371,456 @@ def prepare_context(
         raise
 
 
-def build_release_descriptor(image_input: dict[str, Any], image_reference: str) -> dict[str, Any]:
+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 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_")}
+
+
+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,
+        timeout=timeout,
+        env=_docker_environment(),
+    )
+
+
+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],
+    image_reference: str,
+    *,
+    expected_revision: str,
+    expected_input_sha256: str,
+    expected_config_digest: str,
+    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)
+    _require(
+        IMAGE_REFERENCE.fullmatch(image_reference) is not None,
+        "inspection image must be the exact staging repository at a sha256 digest",
+    )
+    _require(_is_revision(expected_revision), "expected reviewed revision is invalid")
+    _require(_is_sha256(expected_input_sha256), "expected reviewed input SHA-256 is invalid")
+    _require(
+        isinstance(expected_config_digest, str)
+        and expected_config_digest.startswith("sha256:")
+        and _is_sha256(expected_config_digest.removeprefix("sha256:")),
+        "expected reviewed config digest is invalid",
+    )
+    _require(
+        image_input["runtime"]["teleo_git_head"] == expected_revision,
+        "image input differs from the expected reviewed revision",
+    )
+    _require(
+        image_input["input_sha256"] == expected_input_sha256,
+        "image input differs from the expected reviewed input SHA-256",
+    )
+    _validate_docker_authority(docker_binary, docker_host, docker_config)
+    with _DockerCacheLock(docker_config):
+        try:
+            _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 expected reviewed config digest",
+            )
+        except (OSError, subprocess.SubprocessError, PackageError) as exc:
+            raise PackageError("immutable registry image inspection failed") from exc
+    return inspection
+
+
+def build_release_descriptor(
+    image_input: dict[str, Any],
+    inspection: dict[str, Any],
+) -> dict[str, Any]:
+    validate_image_input(image_input)
+    validate_image_inspection(inspection, 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)}"
@@ -1290,10 +1831,12 @@ def build_release_descriptor(image_input: dict[str, Any], image_reference: str)
         "image": {
             "reference": image_reference,
             "digest": digest,
+            "config_digest": inspection["config_digest"],
             "platform": PLATFORM,
             "labels": labels,
             "input_sha256": image_input["input_sha256"],
         },
+        "inspection": inspection,
         "image_input": image_input,
     }
     value = {**stable, "release_sha256": canonical_sha256(stable)}
@@ -1316,7 +1859,7 @@ def expected_image_labels(image_input: dict[str, Any]) -> dict[str, str]:
 
 def validate_release_descriptor(value: dict[str, Any]) -> None:
     _require(
-        set(value) == {"schema", "target", "image", "image_input", "release_sha256"},
+        set(value) == {"schema", "target", "image", "inspection", "image_input", "release_sha256"},
         "release descriptor fields are not exact",
     )
     _require(value.get("schema") == RELEASE_SCHEMA, "release descriptor schema is invalid")
@@ -1326,13 +1869,19 @@ def validate_release_descriptor(value: dict[str, Any]) -> None:
     validate_image_input(image_input)
     image = value.get("image")
     _require(
-        isinstance(image, dict) and set(image) == {"reference", "digest", "platform", "labels", "input_sha256"},
+        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")
+    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(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")
@@ -1419,6 +1968,146 @@ WantedBy=multi-user.target
     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 _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(
@@ -1551,8 +2240,13 @@ def _parse_args(argv: list[str] | None = None) -> argparse.Namespace:
     finalize = subparsers.add_parser("finalize")
     finalize.add_argument("--image-input", type=Path, required=True)
     finalize.add_argument("--image-reference", required=True)
-    finalize.add_argument("--output-release", type=Path, required=True)
-    finalize.add_argument("--output-unit", type=Path, required=True)
+    finalize.add_argument("--expected-revision", required=True)
+    finalize.add_argument("--expected-input-sha256", required=True)
+    finalize.add_argument("--expected-config-digest", 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)
@@ -1625,13 +2319,18 @@ def main(argv: list[str] | None = None) -> int:
             write_json(args.output, result, mode=0o600)
         elif args.command == "finalize":
             image_input = load_json(args.image_input, "image input")
-            result = build_release_descriptor(image_input, args.image_reference)
-            for path in (args.output_release, args.output_unit):
-                _require(not path.exists() and not path.is_symlink(), "finalize output already exists")
-                path.parent.mkdir(parents=True, exist_ok=True)
-            write_json(args.output_release, result)
-            args.output_unit.write_text(render_systemd_unit(result), encoding="utf-8")
-            args.output_unit.chmod(0o644)
+            inspection = inspect_registry_image(
+                image_input,
+                args.image_reference,
+                expected_revision=args.expected_revision,
+                expected_input_sha256=args.expected_input_sha256,
+                expected_config_digest=args.expected_config_digest,
+                docker_binary=args.docker_binary,
+                docker_host=args.docker_host,
+                docker_config=args.docker_config,
+            )
+            result = build_release_descriptor(image_input, inspection)
+            publish_finalize_bundle(args.output_bundle, result)
         else:
             result = load_json(args.release, "release descriptor")
             validate_release_descriptor(result)