Merge pull request #232 from living-ip/feat/leoclean-iap-installer-runner

Add fail-closed GCP no-send IAP runner

docs/workstreams/gcp-leoclean/goals.md

@@ -24,38 +24,46 @@ until the GCP service has passed soak and restore testing.
 
 ## Current state
 
-- Current main/base: `3f09bb2b6c8ac3113fc1743529a03c0e9b6ea587`.
-- PRs #220, #221, #226, #227, #228, #229, and #230 are merged. Main therefore contains the
-  source-first V3 rebuild foundation, the PostgreSQL large-object boundary,
-  bounded Observatory prerequisite preservation, Cloud SQL provider-database
-  residual handling, and the manual exact-revision no-send publication
-  workflow plus the composable prerequisite/runtime ACL boundary and normalized
-  copied-artifact directory contract.
-- The exact-main GCP preflight passed for `3f09bb2`: the staging VM and private
-  PostgreSQL 16 Cloud SQL endpoint were reachable, the Artifact Registry
-  repository was immutable, and no public database attachment was present.
-- Manual publication run #3 passed for that exact revision. Artifact Registry
-  independently reports immutable manifest digest
-  `sha256:c326154acd0b9ae468449abe1c5fd398b6757f375a1a86fdb40e982c515a19a4`
+- Current main/base: `3d35a7fafb41b61a94cb75de9c231afdc08b7f2b`.
+- PRs #220, #221, #226, #227, #228, #229, #230, and #231 are merged. Main
+  therefore contains the source-first V3 rebuild foundation, the PostgreSQL
+  least-privilege boundary, the exact-revision publication workflow, the
+  hardened no-send installer with automatic transaction rollback, and the
+  container-bound `verify-database` command.
+- The exact-main GCP preflight passed for `3d35a7f`: `teleo-staging-1` and the
+  private PostgreSQL 16 Cloud SQL endpoint were reachable, Artifact Registry was
+  immutable, and neither the VM nor Cloud SQL had a public attachment.
+- Manual publication run #4 passed for that exact revision. The retained GitHub
+  artifact archive hashes to
+  `sha256:dbe7a34f2c03e2240de54c40afe0aefae00013dff51e17913140fa38f6e119ea`.
+  Its self-hashed publication journal records `fresh_push`, not reuse or
+  reconciliation, and binds release hash
+  `3e7b00d5ac12308cfc35e3b507258b659e6d95926f3c6241f67240d44105b72f`.
+- Artifact Registry independently reports immutable manifest digest
+  `sha256:c69781400eecd5d2a932b9c7c988ed460365b8bc3f592fe95e51e4c1edf7b230`
   and config digest
-  `sha256:5c4d5e1bcfc8cd112e5c999217fce064b5be718409708a154e9d58f007a521ad`,
-  with the expected revision and package labels. This is a valid intermediate
-  candidate, not an approved deployment candidate.
-- The original run artifact still needs to be retained and checked for its
-  non-reconstructible publication-outcome journal. Final registry state alone
-  cannot prove whether the run freshly pushed, reused, or reconciled a candidate.
-- The published image cannot yet perform a container-bound database proof. A
-  narrow follow-up is adding a hash-bound `verify-database` command that uses the
-  attached staging service-account metadata identity, the scoped secret, and the
-  existing least-privilege SQL verifier from inside the running container. The
-  next deployable image must be published from the merge revision containing
-  that verifier.
-- The GCP VM, private Cloud SQL instance, and scoped Secret Manager secret are
-  known to exist, but the scoped VM secret binding and runtime role have not been
-  activated for this service.
-- Live Cloud SQL provisioning of `leoclean_kb_runtime`, VM access to the scoped
-  secret, installation of the no-send service, effective running-process
-  identity, restart behavior, and rollback remain unproven.
+  `sha256:c5fc9e715bcbee7d4fcc512b0c38aa4acee589836fbc11dbe9c6c34194959438`,
+  with the expected revision, input, artifact, identity, runtime, and target
+  labels. The release descriptor and build/push receipt validate against current
+  package code.
+- A read-only IAM check confirms the scoped runtime secret exists with one
+  enabled version, grants `roles/secretmanager.secretAccessor` only to
+  `sa-teleo-staging-vm`, and the administrator secret does not grant that VM
+  access. The VM also has repository-level Artifact Registry reader and no
+  detected project-level role.
+- The native PostgreSQL `leoclean_kb_runtime` role and password match remain
+  unproven. Cloud SQL control-plane user listing is not authoritative for native
+  PostgreSQL roles; the container-bound permission verifier is the intended
+  proof.
+- Branch `feat/leoclean-iap-installer-runner` now contains the local review
+  candidate for the fixed-target IAP transfer/pull/install/restart/verify
+  runner and receipt-bound post-success rollback. It remains unmerged and has
+  not been used against GCP. The proportional suite passes 308 tests; the full
+  suite passes 3,013 tests with 3 skips when the repository's Python 3.11
+  environment is first on `PATH`.
+- Installation of the no-send service, effective running-process identity,
+  database permissions, restart behavior, and post-success rollback remain
+  unproven.
 - The source-first V3 reconstruction code is on main, but no canonical GCP
   promotion is implied by that merge.
 - Telegram still does not route to GCP. The VPS, production traffic, and
@@ -63,18 +71,22 @@ until the GCP service has passed soak and restore testing.
 
 ## Execution order
 
-1. Review and merge the container-bound database verifier. Run its disposable
-   PostgreSQL and OCI fail-closed tests, then publish one exact-main replacement
-   candidate and retain the complete publication artifact.
-2. Add the narrow IAP transfer/install/restart/rollback runner that consumes the
-   finalized release bundle, pulls only its immutable digest, invokes the existing
-   installer, and binds the live container proof to systemd/container identity.
-3. Provision `leoclean_kb_runtime`, grant only the staging VM access to the
-   scoped secret, and prove the Cloud SQL role's allowed reads and
-   function-only proposal staging plus denied writes and escalation.
-4. Install the exact immutable image as a separate no-send GCP service. Prove
-   the running process uses the scoped identity without administrator fallback,
-   then prove restart and rollback.
+1. Completed: merge the container-bound database verifier, run its disposable
+   PostgreSQL and OCI fail-closed tests, publish the exact-main candidate, retain
+   the complete artifact, and cross-check its receipts and registry digests.
+2. Local implementation complete; review and merge the narrow fixed-target IAP
+   transfer/install/restart/verify runner plus receipt-bound post-success
+   rollback. It consumes only the finalized release artifact and expected
+   hashes, pulls only the immutable digest, preserves ephemeral credential
+   cleanup, and emits sanitized self-hashed receipts.
+3. Prove or provision `leoclean_kb_runtime`. The VM already has access only to
+   the scoped secret; verify that secret matches the native role, allowed reads,
+   transaction-rolled-back function-only proposal staging, and denied writes or
+   escalation.
+4. Install the exact immutable image as a separate no-send GCP service. Bind
+   systemd, PID/invocation, full container ID, image/config digests, and the
+   container database receipt; restart once and then execute the reviewed
+   post-success rollback proof.
 5. Reconstruct or reconcile the V3 knowledge database into a separately
    identified canonical candidate and prove row/hash/source receipts before
    any promotion.
@@ -88,7 +100,7 @@ until the GCP service has passed soak and restore testing.
 
 ## Authority and rollback boundaries
 
-- Database role provisioning can be rolled back independently by refencing
+- Database role provisioning can be rolled back independently by revoking
   the runtime role and removing its scoped secret access.
 - Service deployment can be rolled back independently to the prior immutable
   image or stopped without changing canonical data.

ops/install_gcp_leoclean_nosend_release.py

@@ -4,6 +4,8 @@
 from __future__ import annotations
 
 import argparse
+import base64
+import binascii
 import contextlib
 import fcntl
 import hashlib
@@ -27,12 +29,15 @@ if str(CONTRACT_ROOT) not in sys.path:
 
 from ops import gcp_leoclean_nosend_package as package  # noqa: E402
 
-INSTALL_RECEIPT_SCHEMA = "livingip.leocleanNoSendInstallReceipt.v1"
+INSTALL_RECEIPT_SCHEMA = "livingip.leocleanNoSendInstallReceipt.v2"
+ROLLBACK_STATE_SCHEMA = "livingip.leocleanNoSendRollbackState.v1"
+ROLLBACK_RESULT_SCHEMA = "livingip.leocleanNoSendRollbackResult.v1"
 SERVICE = package.SERVICE
 SYSTEMCTL = "/usr/bin/systemctl"
 MAX_COMMAND_OUTPUT_BYTES = 1024 * 1024
 MAX_RELEASE_BYTES = 8 * 1024 * 1024
 MAX_SOURCE_BYTES = 8 * 1024 * 1024
+MAX_ROLLBACK_STATE_BYTES = 32 * 1024 * 1024
 HEX_64 = re.compile(r"^[0-9a-f]{64}$")
 HEX_40 = re.compile(r"^[0-9a-f]{40}$")
 HEX_32 = re.compile(r"^[0-9a-f]{32}$")
@@ -41,6 +46,7 @@ SOURCE_PATHS = (
     "ops/install_gcp_leoclean_nosend_release.py",
     "ops/leoclean_nosend_service_control.py",
 )
+MANAGED_FILE_ROLES = frozenset({"unit", "helper", "release", "receipt", "rollback_state"})
 SERVICE_EXEC_PROPERTIES = ("ExecStartPre", "ExecStart", "ExecStop")
 SERVICE_SECURITY_PROPERTIES = (
     "RuntimeDirectory",
@@ -134,6 +140,7 @@ class InstallPaths:
     docker_config: Path
     release: Path
     receipt: Path
+    rollback_state: Path
     lock: Path
     proc_root: Path
     dropin_directories: tuple[Path, ...]
@@ -154,6 +161,7 @@ class InstallPaths:
             docker_config=state / "docker-config",
             release=state / "release.json",
             receipt=state / "install-receipt.json",
+            rollback_state=state / "rollback-state.json",
             lock=rooted("/run/leoclean-gcp-nosend-installer.lock"),
             proc_root=proc_root or rooted("/proc"),
             dropin_directories=tuple(
@@ -173,6 +181,7 @@ class FileSnapshot:
     existed: bool
     content: bytes = b""
     mode: int = 0
+    identity: tuple[int, ...] | None = None
 
 
 @dataclass(frozen=True)
@@ -465,12 +474,74 @@ def _snapshot(path: Path) -> FileSnapshot:
         return FileSnapshot(False)
     except OSError as exc:
         raise InstallError("installed file could not be identified") from exc
-    _require(stat.S_ISREG(observed.st_mode), "installed path was not a regular file")
+    _require(
+        stat.S_ISREG(observed.st_mode) and observed.st_nlink == 1,
+        "installed path was not a regular file",
+    )
+    content = _read_stable_file(
+        path,
+        label="installed file",
+        max_bytes=MAX_ROLLBACK_STATE_BYTES,
+        expected_mode=stat.S_IMODE(observed.st_mode),
+    )
+    named = path.stat(follow_symlinks=False)
+    _require(
+        (named.st_dev, named.st_ino, named.st_mode, named.st_nlink)
+        == (observed.st_dev, observed.st_ino, observed.st_mode, observed.st_nlink),
+        "installed file changed while it was snapshotted",
+    )
+    return FileSnapshot(
+        True,
+        content,
+        stat.S_IMODE(observed.st_mode),
+        (
+            observed.st_dev,
+            observed.st_ino,
+            observed.st_mode,
+            observed.st_nlink,
+            observed.st_size,
+            observed.st_mtime_ns,
+            observed.st_ctime_ns,
+        ),
+    )
+
+
+def _snapshot_to_json(snapshot: FileSnapshot) -> dict[str, Any]:
+    if not snapshot.existed:
+        return {"existed": False}
+    _require(snapshot.identity is not None, "rollback snapshot identity was missing")
+    return {
+        "existed": True,
+        "mode": snapshot.mode,
+        "sha256": _sha256_bytes(snapshot.content),
+        "content_base64": base64.b64encode(snapshot.content).decode("ascii"),
+    }
+
+
+def _snapshot_from_json(value: Any, *, label: str) -> FileSnapshot:
+    _require(isinstance(value, dict), f"{label} was invalid")
+    if value == {"existed": False}:
+        return FileSnapshot(False)
+    _require(
+        set(value) == {"existed", "mode", "sha256", "content_base64"}
+        and value.get("existed") is True
+        and isinstance(value.get("mode"), int)
+        and not isinstance(value.get("mode"), bool)
+        and 0 <= value["mode"] <= 0o777
+        and isinstance(value.get("sha256"), str)
+        and HEX_64.fullmatch(value["sha256"]) is not None
+        and isinstance(value.get("content_base64"), str),
+        f"{label} was invalid",
+    )
     try:
-        content = path.read_bytes()
-    except OSError as exc:
-        raise InstallError("installed file could not be read") from exc
-    return FileSnapshot(True, content, stat.S_IMODE(observed.st_mode))
+        content = base64.b64decode(value["content_base64"], validate=True)
+    except (ValueError, binascii.Error) as exc:
+        raise InstallError(f"{label} was invalid") from exc
+    _require(
+        len(content) <= MAX_ROLLBACK_STATE_BYTES and _sha256_bytes(content) == value["sha256"],
+        f"{label} was invalid",
+    )
+    return FileSnapshot(True, content, value["mode"])
 
 
 def _ensure_directory(path: Path, mode: int, created: list[Path]) -> None:
@@ -573,19 +644,34 @@ def _publish_managed_file(
         raise
 
 
-def _restore_snapshot(path: Path, snapshot: FileSnapshot, installed_identity: tuple[int, int] | None) -> None:
+def _restore_snapshot(
+    path: Path,
+    snapshot: FileSnapshot,
+    installed_identity: tuple[int, ...] | None,
+) -> tuple[int, int] | None:
     if installed_identity is not None:
         observed = path.stat(follow_symlinks=False)
+        observed_identity = (
+            observed.st_dev,
+            observed.st_ino,
+            observed.st_mode,
+            observed.st_nlink,
+            observed.st_size,
+            observed.st_mtime_ns,
+            observed.st_ctime_ns,
+        )
         _require(
-            stat.S_ISREG(observed.st_mode) and (observed.st_dev, observed.st_ino) == installed_identity,
+            stat.S_ISREG(observed.st_mode)
+            and (observed.st_dev, observed.st_ino) == installed_identity[:2]
+            and (len(installed_identity) == 2 or observed_identity == installed_identity),
             "rollback refused a replaced installed file",
         )
     if snapshot.existed:
-        _atomic_write(path, snapshot.content, snapshot.mode)
-        return
+        return _atomic_write(path, snapshot.content, snapshot.mode)
     if path.exists() or path.is_symlink():
         path.unlink()
         _fsync_directory(path.parent)
+    return None
 
 
 def _dropins_absent(paths: InstallPaths) -> None:
@@ -979,27 +1065,174 @@ def _read_managed_release(paths: InstallPaths) -> dict[str, Any] | None:
     if not paths.release.exists() and not paths.release.is_symlink():
         return None
     try:
-        value = package.load_json(paths.release, "installed release descriptor")
+        raw = _read_stable_file(
+            paths.release,
+            label="installed release descriptor",
+            max_bytes=MAX_RELEASE_BYTES,
+            expected_mode=0o600,
+        )
+        value = _strict_json_bytes(raw, "installed release descriptor")
+        _require(isinstance(value, dict), "installed release descriptor was invalid")
         package.validate_release_descriptor(value)
-    except (OSError, package.PackageError) as exc:
+    except (OSError, InstallError, package.PackageError) as exc:
         raise InstallError("installed release descriptor was invalid") from exc
     return value
 
 
+def _managed_files(paths: InstallPaths) -> dict[str, tuple[Path, int]]:
+    return {
+        "unit": (paths.unit, 0o644),
+        "helper": (paths.helper, 0o755),
+        "release": (paths.release, 0o600),
+        "receipt": (paths.receipt, 0o600),
+        "rollback_state": (paths.rollback_state, 0o600),
+    }
+
+
+def _read_strict_json_file(path: Path, *, label: str, max_bytes: int, mode: int) -> tuple[dict[str, Any], bytes]:
+    raw = _read_stable_file(path, label=label, max_bytes=max_bytes, expected_mode=mode)
+    value = _strict_json_bytes(raw, label)
+    _require(isinstance(value, dict), f"{label} was invalid")
+    return value, raw
+
+
+def _validate_rollback_state(value: Mapping[str, Any], *, current_release_sha256: str) -> None:
+    _require(
+        set(value)
+        == {
+            "schema",
+            "current_release_sha256",
+            "previous_files",
+            "previous_directories",
+            "previous_service",
+            "sha256",
+        }
+        and value.get("schema") == ROLLBACK_STATE_SCHEMA
+        and value.get("current_release_sha256") == current_release_sha256,
+        "rollback state was invalid",
+    )
+    previous_files = value.get("previous_files")
+    _require(
+        isinstance(previous_files, dict) and set(previous_files) == MANAGED_FILE_ROLES,
+        "rollback state was invalid",
+    )
+    decoded = {
+        role: _snapshot_from_json(previous_files[role], label=f"rollback {role} snapshot")
+        for role in sorted(previous_files)
+    }
+    prior_presence = {role: snapshot.existed for role, snapshot in decoded.items()}
+    _require(
+        not any(prior_presence.values()) or all(prior_presence.values()),
+        "rollback prior managed file set was partial",
+    )
+    previous_directories = value.get("previous_directories")
+    _require(
+        isinstance(previous_directories, dict)
+        and set(previous_directories) == {"state_dir_existed", "docker_config_existed"}
+        and all(isinstance(item, bool) for item in previous_directories.values()),
+        "rollback state was invalid",
+    )
+    previous_service = value.get("previous_service")
+    _require(
+        isinstance(previous_service, dict)
+        and set(previous_service) == {"posture", "active_state", "sub_state", "n_restarts"}
+        and previous_service.get("posture") in {"absent", "inactive", "active"}
+        and previous_service.get("active_state") in {"inactive", "active"}
+        and previous_service.get("sub_state") in {"dead", "running"}
+        and isinstance(previous_service.get("n_restarts"), int)
+        and not isinstance(previous_service.get("n_restarts"), bool)
+        and previous_service["n_restarts"] >= 0,
+        "rollback state was invalid",
+    )
+    expected_state = {
+        "absent": ("inactive", "dead"),
+        "inactive": ("inactive", "dead"),
+        "active": ("active", "running"),
+    }[previous_service["posture"]]
+    _require(
+        (previous_service["active_state"], previous_service["sub_state"]) == expected_state
+        and ((previous_service["posture"] == "absent") == (not any(prior_presence.values()))),
+        "rollback state was invalid",
+    )
+    stable = {key: item for key, item in value.items() if key != "sha256"}
+    _require(value.get("sha256") == package.canonical_sha256(stable), "rollback state self-hash drifted")
+
+
+def _read_rollback_state(paths: InstallPaths, release_sha256: str) -> tuple[dict[str, Any], bytes]:
+    value, raw = _read_strict_json_file(
+        paths.rollback_state,
+        label="rollback state",
+        max_bytes=MAX_ROLLBACK_STATE_BYTES,
+        mode=0o600,
+    )
+    _validate_rollback_state(value, current_release_sha256=release_sha256)
+    return value, raw
+
+
+def _validate_install_receipt(
+    receipt: Mapping[str, Any],
+    *,
+    release: Mapping[str, Any],
+    paths: InstallPaths,
+    rollback_state_raw: bytes,
+) -> None:
+    installed_files = receipt.get("installed_files")
+    rollback = receipt.get("rollback")
+    _require(
+        receipt.get("schema") == INSTALL_RECEIPT_SCHEMA
+        and receipt.get("status") == "pass"
+        and receipt.get("release_sha256") == release["release_sha256"]
+        and isinstance(installed_files, dict)
+        and set(installed_files)
+        == {"unit_sha256", "helper_sha256", "release_descriptor_sha256", "rollback_state_sha256"}
+        and installed_files.get("unit_sha256")
+        == _sha256_bytes(
+            _read_stable_file(
+                paths.unit, label="installed service unit", max_bytes=MAX_RELEASE_BYTES, expected_mode=0o644
+            )
+        )
+        and installed_files.get("helper_sha256")
+        == _sha256_bytes(
+            _read_stable_file(
+                paths.helper, label="installed service helper", max_bytes=MAX_SOURCE_BYTES, expected_mode=0o755
+            )
+        )
+        and installed_files.get("release_descriptor_sha256")
+        == _sha256_bytes(
+            _read_stable_file(
+                paths.release,
+                label="installed release descriptor",
+                max_bytes=MAX_RELEASE_BYTES,
+                expected_mode=0o600,
+            )
+        )
+        and installed_files.get("rollback_state_sha256") == _sha256_bytes(rollback_state_raw)
+        and isinstance(rollback, dict)
+        and rollback.get("state_sha256") == _sha256_bytes(rollback_state_raw)
+        and rollback.get("state_path") == str(paths.rollback_state),
+        "installed receipt was invalid",
+    )
+    _require(isinstance(receipt.get("installer_source"), dict), "installed receipt source binding was invalid")
+    _validate_source_binding(receipt["installer_source"])
+    _require(
+        receipt.get("image")
+        == {
+            "reference": release["image"]["reference"],
+            "digest": release["image"]["digest"],
+            "config_digest": release["image"]["config_digest"],
+            "input_sha256": release["image"]["input_sha256"],
+        },
+        "installed receipt image binding was invalid",
+    )
+
+
 def _validate_managed_prior_state(paths: InstallPaths, helper_source: bytes) -> dict[str, Any] | None:
-    unit_exists = paths.unit.exists() or paths.unit.is_symlink()
-    helper_exists = paths.helper.exists() or paths.helper.is_symlink()
-    release_exists = paths.release.exists() or paths.release.is_symlink()
-    receipt_exists = paths.receipt.exists() or paths.receipt.is_symlink()
-    if not any((unit_exists, helper_exists, release_exists, receipt_exists)):
+    managed = _managed_files(paths)
+    presence = {role: path.exists() or path.is_symlink() for role, (path, _mode) in managed.items()}
+    if not any(presence.values()):
         return None
-    _require(all((unit_exists, helper_exists, release_exists, receipt_exists)), "managed install state was partial")
-    for path, mode in (
-        (paths.unit, 0o644),
-        (paths.helper, 0o755),
-        (paths.release, 0o600),
-        (paths.receipt, 0o600),
-    ):
+    _require(all(presence.values()), "managed install state was partial")
+    for path, mode in managed.values():
         observed = path.stat(follow_symlinks=False)
         _require(
             stat.S_ISREG(observed.st_mode)
@@ -1021,21 +1254,19 @@ def _validate_managed_prior_state(paths: InstallPaths, helper_source: bytes) ->
     _require(release is not None, "managed release state was absent")
     _require(paths.unit.read_bytes() == package.render_systemd_unit(release).encode(), "installed unit drifted")
     _require(paths.helper.is_file() and not paths.helper.is_symlink(), "installed helper was unsafe")
-    try:
-        receipt = package.load_json(paths.receipt, "install receipt")
-    except (OSError, package.PackageError) as exc:
-        raise InstallError("installed receipt was invalid") from exc
-    _require(
-        receipt.get("schema") == INSTALL_RECEIPT_SCHEMA
-        and receipt.get("status") == "pass"
-        and receipt.get("release_sha256") == release["release_sha256"],
-        "installed receipt was invalid",
+    rollback_state, rollback_state_raw = _read_rollback_state(paths, release["release_sha256"])
+    del rollback_state
+    receipt, _receipt_raw = _read_strict_json_file(
+        paths.receipt,
+        label="install receipt",
+        max_bytes=MAX_RELEASE_BYTES,
+        mode=0o600,
     )
-    _require(
-        isinstance(receipt.get("installed_files"), dict)
-        and receipt["installed_files"].get("unit_sha256") == _sha256_bytes(paths.unit.read_bytes())
-        and receipt["installed_files"].get("helper_sha256") == _sha256_bytes(paths.helper.read_bytes()),
-        "installed file receipt drifted",
+    _validate_install_receipt(
+        receipt,
+        release=release,
+        paths=paths,
+        rollback_state_raw=rollback_state_raw,
     )
     _require(bool(helper_source), "service-control helper source was empty")
     return release
@@ -1062,6 +1293,65 @@ def _cleanup_created_directories(created: list[Path]) -> bool:
     return complete
 
 
+def _directory_existed_safely(path: Path, mode: int) -> bool:
+    if not path.exists() and not path.is_symlink():
+        return False
+    observed = path.stat(follow_symlinks=False)
+    _require(
+        stat.S_ISDIR(observed.st_mode) and observed.st_uid == os.geteuid() and stat.S_IMODE(observed.st_mode) == mode,
+        "prior install directory posture was unsafe",
+    )
+    return True
+
+
+def _prior_directory_state(paths: InstallPaths) -> dict[str, bool]:
+    state_existed = _directory_existed_safely(paths.state_dir, 0o700)
+    docker_existed = _directory_existed_safely(paths.docker_config, 0o700)
+    if docker_existed:
+        _require(state_existed and not any(paths.docker_config.iterdir()), "prior Docker config posture was unsafe")
+    if state_existed:
+        allowed = {paths.docker_config.name} if docker_existed else set()
+        observed = {path.name for path in paths.state_dir.iterdir()}
+        managed_names = {paths.release.name, paths.receipt.name, paths.rollback_state.name}
+        _require(
+            observed <= allowed | managed_names,
+            "prior install state directory contained an unmanaged entry",
+        )
+    return {
+        "state_dir_existed": state_existed,
+        "docker_config_existed": docker_existed,
+    }
+
+
+def _build_rollback_state(
+    release_sha256: str,
+    snapshots: Mapping[Path, FileSnapshot],
+    paths: InstallPaths,
+    previous_directories: Mapping[str, bool],
+    prior_posture: ServicePosture,
+) -> dict[str, Any]:
+    path_to_role = {path: role for role, (path, _mode) in _managed_files(paths).items()}
+    _require(set(snapshots) == set(path_to_role), "rollback snapshot set was not exact")
+    stable = {
+        "schema": ROLLBACK_STATE_SCHEMA,
+        "current_release_sha256": release_sha256,
+        "previous_files": {
+            path_to_role[path]: _snapshot_to_json(snapshots[path])
+            for path in sorted(snapshots, key=lambda item: path_to_role[item])
+        },
+        "previous_directories": dict(previous_directories),
+        "previous_service": {
+            "posture": prior_posture.kind,
+            "active_state": prior_posture.state["ActiveState"],
+            "sub_state": prior_posture.state["SubState"],
+            "n_restarts": prior_posture.state["NRestarts"],
+        },
+    }
+    value = {**stable, "sha256": package.canonical_sha256(stable)}
+    _validate_rollback_state(value, current_release_sha256=release_sha256)
+    return value
+
+
 @contextlib.contextmanager
 def _install_lock(paths: InstallPaths):
     if paths.root == Path("/"):
@@ -1129,6 +1419,7 @@ def _install_release_transaction(
     )
     before = prior_posture.state
     prior_active = prior_posture.kind == "active"
+    previous_directories = _prior_directory_state(paths)
 
     created: list[Path] = []
     try:
@@ -1149,6 +1440,7 @@ def _install_release_transaction(
         paths.helper: _snapshot(paths.helper),
         paths.release: _snapshot(paths.release),
         paths.receipt: _snapshot(paths.receipt),
+        paths.rollback_state: _snapshot(paths.rollback_state),
     }
     installed_identities: dict[Path, tuple[int, int]] = {}
     attempted_paths: set[Path] = set()
@@ -1198,6 +1490,20 @@ def _install_release_transaction(
         release_bytes = _canonical_json(release)
         attempted_paths.add(paths.release)
         _publish_managed_file(paths.release, release_bytes, 0o600, installed_identities)
+        rollback_state = _build_rollback_state(
+            release["release_sha256"],
+            snapshots,
+            paths,
+            previous_directories,
+            prior_posture,
+        )
+        rollback_state_bytes = _canonical_json(rollback_state)
+        _require(
+            len(rollback_state_bytes) <= MAX_ROLLBACK_STATE_BYTES,
+            "rollback state exceeded the safety limit",
+        )
+        attempted_paths.add(paths.rollback_state)
+        _publish_managed_file(paths.rollback_state, rollback_state_bytes, 0o600, installed_identities)
         receipt = {
             "schema": INSTALL_RECEIPT_SCHEMA,
             "status": "pass",
@@ -1214,6 +1520,7 @@ def _install_release_transaction(
                 "unit_sha256": _sha256_bytes(unit_content),
                 "helper_sha256": _sha256_bytes(helper_content),
                 "release_descriptor_sha256": _sha256_bytes(release_bytes),
+                "rollback_state_sha256": _sha256_bytes(rollback_state_bytes),
             },
             "service_before": {
                 "load_state": before["LoadState"],
@@ -1228,6 +1535,8 @@ def _install_release_transaction(
                 "previous_service_active": prior_active,
                 "previous_service_posture": prior_posture.kind,
                 "automatic_on_failure": True,
+                "state_path": str(paths.rollback_state),
+                "state_sha256": _sha256_bytes(rollback_state_bytes),
             },
             "authority": {
                 "image_pulled": False,
@@ -1246,7 +1555,7 @@ def _install_release_transaction(
                 _systemctl("stop", runner)
             except InstallError:
                 rollback_errors.append("candidate_stop")
-        for path in (paths.receipt, paths.release, paths.unit, paths.helper):
+        for path in (paths.receipt, paths.rollback_state, paths.release, paths.unit, paths.helper):
             if path not in attempted_paths:
                 continue
             try:
@@ -1294,6 +1603,415 @@ def _install_release_transaction(
         raise InstallError("installation failed and was rolled back") from exc
 
 
+def _json_from_snapshot(snapshot: FileSnapshot, *, label: str) -> dict[str, Any]:
+    _require(snapshot.existed, f"{label} was absent")
+    value = _strict_json_bytes(snapshot.content, label)
+    _require(isinstance(value, dict), f"{label} was invalid")
+    return value
+
+
+def _validate_managed_snapshots(
+    snapshots: Mapping[str, FileSnapshot],
+    *,
+    paths: InstallPaths,
+) -> tuple[dict[str, Any], dict[str, Any], dict[str, Any]]:
+    expected_modes = {role: mode for role, (_path, mode) in _managed_files(paths).items()}
+    _require(set(snapshots) == set(expected_modes), "managed snapshot set was not exact")
+    _require(all(snapshot.existed for snapshot in snapshots.values()), "managed snapshot set was partial")
+    _require(
+        all(snapshots[role].mode == expected_modes[role] for role in expected_modes),
+        "managed snapshot mode drifted",
+    )
+    release = _json_from_snapshot(snapshots["release"], label="snapshot release descriptor")
+    try:
+        package.validate_release_descriptor(release)
+    except package.PackageError as exc:
+        raise InstallError("snapshot release descriptor was invalid") from exc
+    release_sha256 = release["release_sha256"]
+    _require(
+        snapshots["unit"].content == package.render_systemd_unit(release).encode("utf-8")
+        and snapshots["helper"].content.startswith(b"#!/usr/bin/python3 -I\n"),
+        "snapshot installed files differed from the release",
+    )
+    rollback_state = _json_from_snapshot(snapshots["rollback_state"], label="snapshot rollback state")
+    _validate_rollback_state(rollback_state, current_release_sha256=release_sha256)
+    receipt = _json_from_snapshot(snapshots["receipt"], label="snapshot install receipt")
+    installed_files = receipt.get("installed_files")
+    rollback = receipt.get("rollback")
+    _require(
+        receipt.get("schema") == INSTALL_RECEIPT_SCHEMA
+        and receipt.get("status") == "pass"
+        and receipt.get("release_sha256") == release_sha256
+        and isinstance(receipt.get("installer_source"), dict)
+        and isinstance(installed_files, dict)
+        and installed_files.get("unit_sha256") == _sha256_bytes(snapshots["unit"].content)
+        and installed_files.get("helper_sha256") == _sha256_bytes(snapshots["helper"].content)
+        and installed_files.get("release_descriptor_sha256") == _sha256_bytes(snapshots["release"].content)
+        and installed_files.get("rollback_state_sha256") == _sha256_bytes(snapshots["rollback_state"].content)
+        and isinstance(rollback, dict)
+        and rollback.get("state_path") == str(paths.rollback_state)
+        and rollback.get("state_sha256") == _sha256_bytes(snapshots["rollback_state"].content),
+        "snapshot install receipt was invalid",
+    )
+    _validate_source_binding(receipt["installer_source"])
+    return release, receipt, rollback_state
+
+
+def _current_managed_snapshots(paths: InstallPaths) -> dict[str, FileSnapshot]:
+    return {role: _snapshot(path) for role, (path, _mode) in _managed_files(paths).items()}
+
+
+def _previous_snapshots(rollback_state: Mapping[str, Any]) -> dict[str, FileSnapshot]:
+    previous_files = rollback_state["previous_files"]
+    _require(isinstance(previous_files, dict), "rollback prior file set was invalid")
+    return {
+        role: _snapshot_from_json(previous_files[role], label=f"rollback {role} snapshot")
+        for role in sorted(previous_files)
+    }
+
+
+def _prepare_explicit_rollback(
+    expected_release_sha256: str,
+    *,
+    paths: InstallPaths,
+    source_binding: Mapping[str, Any],
+    runner: Runner,
+    sleeper: Sleeper,
+) -> dict[str, Any]:
+    _require(HEX_64.fullmatch(expected_release_sha256) is not None, "expected release hash was invalid")
+    _bound_source_files(source_binding)
+    managed_release = _validate_managed_prior_state(paths, b"explicit rollback")
+    _require(managed_release is not None, "explicit rollback found no managed release")
+    current_snapshots = _current_managed_snapshots(paths)
+    current_release, current_receipt, rollback_state = _validate_managed_snapshots(
+        current_snapshots,
+        paths=paths,
+    )
+    _require(
+        current_release == managed_release and current_release["release_sha256"] == expected_release_sha256,
+        "installed release did not match the expected release hash",
+    )
+    previous_snapshots = _previous_snapshots(rollback_state)
+    previous_presence = [snapshot.existed for snapshot in previous_snapshots.values()]
+    previous_release: dict[str, Any] | None
+    if any(previous_presence):
+        _require(all(previous_presence), "rollback prior managed file set was partial")
+        previous_release, _previous_receipt, _previous_rollback = _validate_managed_snapshots(
+            previous_snapshots,
+            paths=paths,
+        )
+    else:
+        previous_release = None
+    current_posture = _stable_service_posture(
+        current_release,
+        runner=runner,
+        paths=paths,
+        sleeper=sleeper,
+    )
+    current_verification: dict[str, Any] | None = None
+    if current_posture.kind == "active":
+        current_verification = verify_running_release(
+            current_release,
+            runner=runner,
+            paths=paths,
+            sleeper=sleeper,
+        )
+        _require(
+            current_verification["main_pid"] == current_posture.state["MainPID"]
+            and current_verification["n_restarts"] == current_posture.state["NRestarts"]
+            and current_verification["invocation_id"] == current_posture.state["InvocationID"],
+            "installed service changed during rollback preparation",
+        )
+    previous_service = rollback_state["previous_service"]
+    _require(
+        (previous_release is None) == (previous_service["posture"] == "absent"),
+        "rollback prior release differed from its recorded posture",
+    )
+    return {
+        "current_release": current_release,
+        "current_receipt": current_receipt,
+        "current_snapshots": current_snapshots,
+        "current_posture": current_posture,
+        "current_verification": current_verification,
+        "previous_snapshots": previous_snapshots,
+        "previous_release": previous_release,
+        "previous_directories": rollback_state["previous_directories"],
+        "previous_service": previous_service,
+        "rollback_state_sha256": _sha256_bytes(current_snapshots["rollback_state"].content),
+    }
+
+
+def _build_rollback_plan(
+    expected_release_sha256: str,
+    prepared: Mapping[str, Any],
+    *,
+    paths: InstallPaths,
+    source_binding: Mapping[str, Any],
+) -> dict[str, Any]:
+    previous_release = prepared["previous_release"]
+    return {
+        "schema": ROLLBACK_RESULT_SCHEMA,
+        "status": "pass",
+        "mode": "dry_run",
+        "installer_source": dict(source_binding),
+        "expected_release_sha256": expected_release_sha256,
+        "installed_release_sha256": prepared["current_release"]["release_sha256"],
+        "installed_receipt_release_sha256": prepared["current_receipt"]["release_sha256"],
+        "rollback_state_sha256": prepared["rollback_state_sha256"],
+        "current_service_posture": prepared["current_posture"].kind,
+        "target_service_posture": prepared["previous_service"]["posture"],
+        "target_release_sha256": previous_release["release_sha256"] if previous_release is not None else None,
+        "target": {
+            "service": SERVICE,
+            "unit": str(paths.unit),
+            "helper": str(paths.helper),
+            "state_directory": str(paths.state_dir),
+        },
+        "authority": {
+            "execute_requires_root_and_execute_rollback": True,
+            "image_pulled": False,
+            "cloud_or_database_authority_changed": False,
+            "transport_authority_changed": False,
+        },
+    }
+
+
+def _remove_rollback_owned_directories(paths: InstallPaths, previous_directories: Mapping[str, bool]) -> None:
+    _require(not any(paths.docker_config.iterdir()), "rollback Docker config directory was not empty")
+    if not previous_directories["docker_config_existed"]:
+        paths.docker_config.rmdir()
+        _fsync_directory(paths.state_dir)
+    if not previous_directories["state_dir_existed"]:
+        _require(not any(paths.state_dir.iterdir()), "rollback state directory was not empty")
+        paths.state_dir.rmdir()
+        _fsync_directory(paths.state_dir.parent)
+
+
+def _assert_snapshot_identities(
+    snapshots: Mapping[str, FileSnapshot],
+    *,
+    paths: InstallPaths,
+) -> None:
+    for role, (path, expected_mode) in _managed_files(paths).items():
+        snapshot = snapshots[role]
+        _require(snapshot.existed and snapshot.identity is not None, "installed rollback file identity was missing")
+        try:
+            observed = path.stat(follow_symlinks=False)
+        except OSError as exc:
+            raise InstallError("installed rollback file identity drifted") from exc
+        observed_identity = (
+            observed.st_dev,
+            observed.st_ino,
+            observed.st_mode,
+            observed.st_nlink,
+            observed.st_size,
+            observed.st_mtime_ns,
+            observed.st_ctime_ns,
+        )
+        _require(
+            stat.S_ISREG(observed.st_mode)
+            and observed.st_nlink == 1
+            and stat.S_IMODE(observed.st_mode) == expected_mode
+            and observed_identity == snapshot.identity,
+            "installed rollback file identity drifted",
+        )
+
+
+def _execute_explicit_rollback(
+    expected_release_sha256: str,
+    prepared: Mapping[str, Any],
+    *,
+    paths: InstallPaths,
+    runner: Runner,
+    sleeper: Sleeper,
+) -> dict[str, Any]:
+    current_release = prepared["current_release"]
+    current_snapshots = prepared["current_snapshots"]
+    current_posture: ServicePosture = prepared["current_posture"]
+    previous_snapshots = prepared["previous_snapshots"]
+    previous_release = prepared["previous_release"]
+    previous_service = prepared["previous_service"]
+    role_paths = {role: path for role, (path, _mode) in _managed_files(paths).items()}
+    restored_identities: dict[Path, tuple[int, int] | None] = {}
+    mutated_paths: set[Path] = set()
+    files_mutated = False
+    service_stopped = False
+    target_may_be_active = False
+    try:
+        _assert_snapshot_identities(current_snapshots, paths=paths)
+        if current_posture.kind == "active":
+            fresh_verification = verify_running_release(
+                current_release,
+                runner=runner,
+                paths=paths,
+                sleeper=sleeper,
+            )
+            _require(
+                fresh_verification == prepared["current_verification"],
+                "service or container drifted before rollback stop",
+            )
+            before_stop = _service_state(runner)
+            _require(
+                before_stop["MainPID"] == current_posture.state["MainPID"]
+                and before_stop["NRestarts"] == current_posture.state["NRestarts"]
+                and before_stop["InvocationID"] == current_posture.state["InvocationID"]
+                and before_stop["ActiveState"] == "active"
+                and before_stop["SubState"] == "running",
+                "service PID or invocation drifted before rollback stop",
+            )
+            _systemctl("stop", runner)
+            service_stopped = True
+        stopped = _stable_service_posture(
+            current_release,
+            runner=runner,
+            paths=paths,
+            sleeper=sleeper,
+        )
+        _require(stopped.kind == "inactive", "installed service did not stop into a trusted posture")
+        for role in ("receipt", "rollback_state", "release", "unit", "helper"):
+            path = role_paths[role]
+            files_mutated = True
+            mutated_paths.add(path)
+            restored_identities[path] = _restore_snapshot(
+                path,
+                previous_snapshots[role],
+                current_snapshots[role].identity,
+            )
+        _systemctl("daemon-reload", runner)
+        if previous_service["posture"] == "active":
+            _require(previous_release is not None, "rollback active target release was missing")
+            target_may_be_active = True
+            _systemctl("restart", runner)
+            final_verification = verify_running_release(
+                previous_release,
+                runner=runner,
+                paths=paths,
+                sleeper=sleeper,
+            )
+            _require(
+                final_verification["n_restarts"] == previous_service["n_restarts"],
+                "rollback did not restore the recorded active posture",
+            )
+            final_posture = "active"
+        elif previous_service["posture"] == "inactive":
+            _require(previous_release is not None, "rollback inactive target release was missing")
+            final = _stable_service_posture(
+                previous_release,
+                runner=runner,
+                paths=paths,
+                sleeper=sleeper,
+            )
+            _require(
+                final.kind == "inactive" and final.state["NRestarts"] == previous_service["n_restarts"],
+                "rollback did not restore the recorded inactive posture",
+            )
+            final_verification = None
+            final_posture = "inactive"
+        else:
+            _require(previous_release is None, "rollback absent target unexpectedly had a release")
+            _systemctl("reset-failed", runner)
+            final = _stable_service_posture(
+                None,
+                runner=runner,
+                paths=paths,
+                sleeper=sleeper,
+            )
+            _require(final.kind == "absent", "rollback did not restore the absent posture")
+            _remove_rollback_owned_directories(paths, prepared["previous_directories"])
+            final_verification = None
+            final_posture = "absent"
+        return {
+            "schema": ROLLBACK_RESULT_SCHEMA,
+            "status": "pass",
+            "mode": "executed",
+            "rolled_back_release_sha256": expected_release_sha256,
+            "restored_release_sha256": previous_release["release_sha256"] if previous_release is not None else None,
+            "restored_service_posture": final_posture,
+            "verification": final_verification,
+            "authority": {
+                "image_pulled": False,
+                "cloud_or_database_authority_changed": False,
+                "transport_authority_changed": False,
+            },
+        }
+    except Exception as exc:
+        recovery_errors: list[str] = []
+        if target_may_be_active:
+            try:
+                _systemctl("stop", runner)
+            except InstallError:
+                recovery_errors.append("target_stop")
+        if files_mutated:
+            for role in ("receipt", "rollback_state", "release", "unit", "helper"):
+                path = role_paths[role]
+                if path not in mutated_paths:
+                    continue
+                try:
+                    if restored_identities.get(path) is None:
+                        _require(
+                            not path.exists() and not path.is_symlink(),
+                            "rollback recovery refused a replaced pathname",
+                        )
+                    _restore_snapshot(path, current_snapshots[role], restored_identities.get(path))
+                except (OSError, InstallError):
+                    recovery_errors.append("file_restore")
+            try:
+                _systemctl("daemon-reload", runner)
+            except InstallError:
+                recovery_errors.append("daemon_reload")
+        if service_stopped and current_posture.kind == "active":
+            try:
+                _systemctl("restart", runner)
+                verify_running_release(current_release, runner=runner, paths=paths, sleeper=sleeper)
+            except InstallError:
+                recovery_errors.append("current_restart")
+        if recovery_errors:
+            raise InstallError("explicit rollback failed and recovery was incomplete") from exc
+        if isinstance(exc, InstallError):
+            raise
+        raise InstallError("explicit rollback failed without changing the installed release") from exc
+
+
+def rollback_release(
+    expected_release_sha256: str,
+    *,
+    paths: InstallPaths,
+    source_binding: Mapping[str, Any],
+    runner: Runner = _default_runner,
+    sleeper: Sleeper = time.sleep,
+    execute_rollback: bool = False,
+) -> dict[str, Any]:
+    if not execute_rollback:
+        prepared = _prepare_explicit_rollback(
+            expected_release_sha256,
+            paths=paths,
+            source_binding=source_binding,
+            runner=runner,
+            sleeper=sleeper,
+        )
+        return _build_rollback_plan(
+            expected_release_sha256,
+            prepared,
+            paths=paths,
+            source_binding=source_binding,
+        )
+    with _install_lock(paths):
+        prepared = _prepare_explicit_rollback(
+            expected_release_sha256,
+            paths=paths,
+            source_binding=source_binding,
+            runner=runner,
+            sleeper=sleeper,
+        )
+        return _execute_explicit_rollback(
+            expected_release_sha256,
+            prepared,
+            paths=paths,
+            runner=runner,
+            sleeper=sleeper,
+        )
+
+
 def install_release(
     bundle: Path,
     *,
@@ -1330,23 +2048,45 @@ class _Parser(argparse.ArgumentParser):
 
 def parse_args(argv: list[str] | None = None) -> argparse.Namespace:
     parser = _Parser(description=__doc__, add_help=False)
-    parser.add_argument("--bundle", required=True, type=Path)
+    parser.add_argument("--bundle", type=Path)
     parser.add_argument("--execute-restart", action="store_true")
-    return parser.parse_args(argv)
+    parser.add_argument("--rollback-release")
+    parser.add_argument("--execute-rollback", action="store_true")
+    args = parser.parse_args(argv)
+    if args.rollback_release is None:
+        _require(args.bundle is not None and not args.execute_rollback, "invalid arguments")
+    else:
+        _require(
+            args.bundle is None and not args.execute_restart and HEX_64.fullmatch(args.rollback_release) is not None,
+            "invalid arguments",
+        )
+    _require(not (args.execute_restart and args.execute_rollback), "invalid arguments")
+    return args
 
 
 def main(argv: list[str] | None = None) -> int:
     try:
         args = parse_args(argv)
         root = Path("/")
-        if args.execute_restart:
-            _require(os.geteuid() == 0, "live installation requires root")
-        result = install_release(
-            args.bundle.absolute(),
-            paths=InstallPaths.under(root),
-            source_binding=reviewed_source_binding(),
-            execute_restart=args.execute_restart,
-        )
+        if args.execute_restart or args.execute_rollback:
+            _require(os.geteuid() == 0, "live installation or rollback requires root")
+        paths = InstallPaths.under(root)
+        binding = reviewed_source_binding()
+        if args.rollback_release is not None:
+            result = rollback_release(
+                args.rollback_release,
+                paths=paths,
+                source_binding=binding,
+                execute_rollback=args.execute_rollback,
+            )
+        else:
+            _require(args.bundle is not None, "invalid arguments")
+            result = install_release(
+                args.bundle.absolute(),
+                paths=paths,
+                source_binding=binding,
+                execute_restart=args.execute_restart,
+            )
         sys.stdout.write(json.dumps(result, allow_nan=False, separators=(",", ":"), sort_keys=True) + "\n")
         return 0
     except (InstallError, package.PackageError) as exc:

tests/test_gcp_leoclean_nosend_iap_runner.py

@@ -0,0 +1,779 @@
+from __future__ import annotations
+
+import io
+import json
+import os
+import stat
+import subprocess
+import tarfile
+from pathlib import Path
+from types import SimpleNamespace
+from typing import Any
+
+import pytest
+
+from ops import gcp_leoclean_nosend_iap_remote as remote
+from ops import run_gcp_leoclean_nosend_iap as local
+
+
+def completed(
+    argv: list[str],
+    *,
+    returncode: int = 0,
+    stdout: bytes = b"",
+    stderr: bytes = b"",
+) -> subprocess.CompletedProcess[bytes]:
+    return subprocess.CompletedProcess(argv, returncode, stdout, stderr)
+
+
+def private_directory(path: Path) -> Path:
+    path.mkdir(mode=0o700)
+    path.chmod(0o700)
+    return path
+
+
+def fixed_commands(tmp_path: Path, operation: str = "install") -> tuple[list[str], list[str], list[str]]:
+    request_id = "iap-abcdefghijkl"
+    archive_name, key_name = local.remote_names(request_id)
+    archive = tmp_path / archive_name
+    ssh_key = tmp_path / key_name
+    archive.write_bytes(b"archive")
+    archive.chmod(0o600)
+    return local.build_gcloud_argv(
+        operation=operation,
+        request_id=request_id,
+        runner_revision="a" * 40,
+        release_revision="b" * 40,
+        archive=archive,
+        archive_sha256="c" * 64,
+        ssh_key=ssh_key,
+        helper_sha256="d" * 64,
+    )
+
+
+def executable_plan(tmp_path: Path, operation: str) -> dict[str, Any]:
+    request_id = "iap-abcdefghijkl"
+    archive_name, key_name = local.remote_names(request_id)
+    archive = tmp_path / archive_name
+    ssh_key = tmp_path / key_name
+    archive.write_bytes(b"archive")
+    archive.chmod(0o600)
+    archive_sha256 = local._sha256_bytes(archive.read_bytes())
+    helper_sha256 = "c" * 64
+    commands = local.build_gcloud_argv(
+        operation=operation,
+        request_id=request_id,
+        runner_revision="a" * 40,
+        release_revision="b" * 40,
+        archive=archive,
+        archive_sha256=archive_sha256,
+        ssh_key=ssh_key,
+        helper_sha256=helper_sha256,
+    )
+    expected = {
+        "release_artifact_sha256": "d" * 64,
+        "release_sha256": "e" * 64,
+        "image_digest": f"sha256:{'f' * 64}",
+        "image_config_digest": f"sha256:{'1' * 64}",
+    }
+    return {
+        "schema": local.PLAN_SCHEMA,
+        "status": "pass",
+        "mode": "dry_run",
+        "operation": operation,
+        "request_id": "iap-abcdefghijkl",
+        "runner_revision": "a" * 40,
+        "release_revision": "b" * 40,
+        "target": {
+            "project": local.PROJECT,
+            "zone": local.ZONE,
+            "instance": local.INSTANCE,
+            "service": local.SERVICE,
+            "service_account": local.SERVICE_ACCOUNT,
+        },
+        "release": {
+            "release_sha256": expected["release_sha256"],
+            "image_reference": f"{local.package.IMAGE_REPOSITORY}@{expected['image_digest']}",
+            "image_digest": expected["image_digest"],
+            "image_config_digest": expected["image_config_digest"],
+        },
+        "release_artifact": {
+            "archive_sha256": expected["release_artifact_sha256"],
+            "archive_size": 1,
+            "members": [],
+        },
+        "expected": expected,
+        "request": {
+            "archive": str(archive),
+            "archive_sha256": archive_sha256,
+            "helper_sha256": helper_sha256,
+            "manifest_sha256": "d" * 64,
+            "ssh_key": str(ssh_key),
+        },
+        "actions": local._action_receipts(commands),
+        "authority": {
+            "gcp_mutation": operation in {"install", "rollback"},
+            "service_restart": operation in {"install", "rollback"},
+            "database_or_secret_provisioning": False,
+            "production_or_transport": False,
+            "vps": False,
+        },
+    }
+
+
+def test_gcloud_commands_are_fixed_iap_only_and_expire_keys(tmp_path: Path) -> None:
+    scp, ssh, cleanup = fixed_commands(tmp_path)
+
+    assert scp[:3] == [local.GCLOUD, "compute", "scp"]
+    assert ssh[:4] == [local.GCLOUD, "compute", "ssh", local.INSTANCE]
+    for command in (scp, ssh, cleanup):
+        assert f"--project={local.PROJECT}" in command
+        assert f"--zone={local.ZONE}" in command
+        assert "--tunnel-through-iap" in command
+        assert "--ssh-key-expire-after=5m" in command
+    encoded = json.dumps({"scp": scp, "ssh": ssh, "cleanup": cleanup})
+    assert local.INSTANCE in encoded
+    assert local.SERVICE not in encoded
+    assert "teleo-prod-1" not in encoded
+    assert "77.42.65.182" not in encoded
+    assert "telegram" not in encoded.casefold()
+    assert "sudo -- /usr/bin/python3 -I -c" in ssh[-1]
+    assert "/tmp/livingip-leoclean-nosend-remote" not in ssh[-1]
+
+
+def test_embedded_root_bootstrap_compiles() -> None:
+    compile(local.BOOTSTRAP_CODE, "<leoclean-iap-bootstrap>", "exec")
+
+
+def test_live_origin_main_readback_is_exact() -> None:
+    revision = "a" * 40
+    responses = iter(
+        [
+            f"{local.ORIGIN_URL}\n".encode(),
+            f"{revision}\trefs/heads/main\n".encode(),
+        ]
+    )
+
+    def runner(argv: list[str], **_kwargs: Any) -> subprocess.CompletedProcess[bytes]:
+        return completed(argv, stdout=next(responses))
+
+    local.validate_remote_main(revision, runner=runner)
+
+
+@pytest.mark.parametrize(
+    "responses",
+    [
+        [b"git@github.com:someone-else/teleo-infrastructure.git\n"],
+        [
+            f"{local.ORIGIN_URL}\n".encode(),
+            f"{'b' * 40}\trefs/heads/main\n".encode(),
+        ],
+    ],
+)
+def test_live_origin_main_readback_rejects_different_authority(responses: list[bytes]) -> None:
+    output = iter(responses)
+
+    def runner(argv: list[str], **_kwargs: Any) -> subprocess.CompletedProcess[bytes]:
+        return completed(argv, stdout=next(output))
+
+    with pytest.raises(local.IapRequestError):
+        local.validate_remote_main("a" * 40, runner=runner)
+
+
+@pytest.mark.parametrize("value", ["main", "a" * 39, "A" * 40, "a" * 41, "a" * 39 + ";"])
+def test_revision_input_is_exact(value: str, tmp_path: Path) -> None:
+    archive_name, key_name = local.remote_names("iap-abcdefghijkl")
+    with pytest.raises(local.IapRequestError, match="revision"):
+        local.build_gcloud_argv(
+            operation="install",
+            request_id="iap-abcdefghijkl",
+            runner_revision=value,
+            release_revision="b" * 40,
+            archive=tmp_path / archive_name,
+            archive_sha256="b" * 64,
+            ssh_key=tmp_path / key_name,
+            helper_sha256="c" * 64,
+        )
+
+
+@pytest.mark.parametrize("value", ["iap-short", "iap-abcdefghijkl;id", "../iap-abcdefghijkl", "iap-ABCDEFGHIJKL"])
+def test_request_id_rejects_shell_and_path_fragments(value: str) -> None:
+    with pytest.raises(local.IapRequestError, match="request ID"):
+        local.remote_names(value)
+
+
+def test_mutable_release_reference_is_rejected(monkeypatch: pytest.MonkeyPatch, tmp_path: Path) -> None:
+    release = {
+        "image": {"reference": f"{local.package.IMAGE_REPOSITORY}:latest"},
+        "image_input": {"runtime": {"teleo_git_head": "a" * 40}},
+    }
+    monkeypatch.setattr(local.installer, "validate_release_bundle", lambda _bundle: (release, b"unit"))
+
+    with pytest.raises(local.IapRequestError, match="immutable"):
+        local.validate_release(
+            tmp_path,
+            "a" * 40,
+            expected_release_sha256="b" * 64,
+            expected_image_digest=f"sha256:{'c' * 64}",
+            expected_image_config_digest=f"sha256:{'d' * 64}",
+        )
+
+
+def test_stable_reader_rejects_symlink(tmp_path: Path) -> None:
+    root = private_directory(tmp_path / "root")
+    target = root / "target"
+    target.write_text("safe", encoding="utf-8")
+    alias = root / "alias"
+    alias.symlink_to(target)
+
+    with pytest.raises(local.IapRequestError, match="unsafe"):
+        local._read_stable_file(alias.absolute(), root=root.absolute())
+
+
+def archive_manifest(payloads: dict[str, tuple[bytes, int]]) -> dict[str, Any]:
+    files = [
+        {"path": path, "sha256": local._sha256_bytes(content), "size": len(content), "mode": mode}
+        for path, (content, mode) in sorted(payloads.items())
+    ]
+    stable = {
+        "schema": remote.SCHEMA,
+        "operation": "verify",
+        "request_id": "iap-abcdefghijkl",
+        "runner_revision": "a" * 40,
+        "release_revision": "b" * 40,
+        "target": {
+            "project": remote.PROJECT,
+            "zone": remote.ZONE,
+            "instance": remote.INSTANCE,
+            "service": remote.SERVICE,
+            "service_account": remote.SERVICE_ACCOUNT,
+        },
+        "release": {
+            "release_sha256": "b" * 64,
+            "image_reference": f"{remote.IMAGE_REPOSITORY}@sha256:{'c' * 64}",
+            "image_digest": f"sha256:{'c' * 64}",
+            "image_config_digest": f"sha256:{'d' * 64}",
+        },
+        "installer_source": {"revision": "a" * 40, "files": [], "sha256": "e" * 64},
+        "release_artifact": {"archive_sha256": "f" * 64, "archive_size": 1, "members": []},
+        "files": files,
+    }
+    return {**stable, "manifest_sha256": remote._canonical_sha256(stable)}
+
+
+def exact_payloads() -> dict[str, tuple[bytes, int]]:
+    return {name: (f"payload:{name}\n".encode(), 0o644) for name in remote.EXPECTED_MEMBERS - {"request-manifest.json"}}
+
+
+def test_archive_is_deterministic_and_extracts_exact_members(tmp_path: Path) -> None:
+    first_root = private_directory(tmp_path / "first")
+    second_root = private_directory(tmp_path / "second")
+    extracted = private_directory(tmp_path / "extracted")
+    payloads = exact_payloads()
+    manifest = archive_manifest(payloads)
+
+    first_hash = local.build_archive(first_root / "request.tar.gz", manifest=manifest, payloads=payloads)
+    second_hash = local.build_archive(second_root / "request.tar.gz", manifest=manifest, payloads=payloads)
+    observed = remote.extract_archive(first_root / "request.tar.gz", extracted)
+
+    assert first_hash == second_hash
+    assert (first_root / "request.tar.gz").read_bytes() == (second_root / "request.tar.gz").read_bytes()
+    assert observed == manifest
+    assert {str(path.relative_to(extracted)) for path in extracted.rglob("*") if path.is_file()} == (
+        remote.EXPECTED_MEMBERS - {"request-manifest.json"}
+    )
+
+
+def write_unsafe_archive(path: Path, *, kind: str) -> None:
+    with tarfile.open(path, "w:gz") as archive:
+        for index, name in enumerate(sorted(remote.EXPECTED_MEMBERS)):
+            info = tarfile.TarInfo(name)
+            info.uid = 0
+            info.gid = 0
+            info.uname = "root"
+            info.gname = "root"
+            info.mtime = 0
+            info.mode = 0o600
+            content = b"{}"
+            if kind == "link" and index == 0:
+                info.type = tarfile.SYMTYPE
+                info.linkname = "/etc/passwd"
+                info.size = 0
+                archive.addfile(info)
+                continue
+            if kind == "oversize" and index == 0:
+                content = b"x" * (remote.MAX_MEMBER_BYTES + 1)
+            info.size = len(content)
+            archive.addfile(info, io.BytesIO(content))
+        if kind == "duplicate":
+            info = tarfile.TarInfo("request-manifest.json")
+            info.size = 2
+            info.mode = 0o600
+            info.uid = info.gid = info.mtime = 0
+            info.uname = info.gname = "root"
+            archive.addfile(info, io.BytesIO(b"{}"))
+
+
+@pytest.mark.parametrize("kind", ["link", "oversize", "duplicate"])
+def test_archive_rejects_links_oversize_and_duplicate_members(tmp_path: Path, kind: str) -> None:
+    path = tmp_path / f"{kind}.tar.gz"
+    destination = private_directory(tmp_path / kind)
+    write_unsafe_archive(path, kind=kind)
+
+    with pytest.raises(remote.RemoteError):
+        remote.extract_archive(path, destination)
+
+
+def test_archive_output_requires_private_directory(tmp_path: Path) -> None:
+    output = tmp_path / "public"
+    output.mkdir(mode=0o755)
+    output.chmod(0o755)
+    with pytest.raises(local.IapRequestError, match="private"):
+        local.build_archive(output / "request.tar.gz", manifest={}, payloads={})
+
+
+@pytest.mark.parametrize(
+    ("field", "value"),
+    [
+        ("expected_artifact_sha256", "not-a-hash"),
+        ("expected_release_sha256", "a" * 63),
+        ("expected_image_digest", "latest"),
+        ("expected_image_config_digest", "sha256:" + "A" * 64),
+    ],
+)
+def test_expected_release_inputs_are_exact(field: str, value: str) -> None:
+    arguments = {
+        "expected_artifact_sha256": "a" * 64,
+        "expected_release_sha256": "b" * 64,
+        "expected_image_digest": f"sha256:{'c' * 64}",
+        "expected_image_config_digest": f"sha256:{'d' * 64}",
+    }
+    arguments[field] = value
+    with pytest.raises(local.IapRequestError, match="expected"):
+        local._validate_expected_bindings(**arguments)
+
+
+def test_release_and_original_artifact_require_independent_expected_values(
+    monkeypatch: pytest.MonkeyPatch,
+    tmp_path: Path,
+) -> None:
+    release = {
+        "release_sha256": "a" * 64,
+        "image": {
+            "reference": f"{local.package.IMAGE_REPOSITORY}@sha256:{'b' * 64}",
+            "digest": f"sha256:{'b' * 64}",
+            "config_digest": f"sha256:{'c' * 64}",
+        },
+        "image_input": {"runtime": {"teleo_git_head": "d" * 40}},
+    }
+    monkeypatch.setattr(local.installer, "validate_release_bundle", lambda _bundle: (release, b"unit"))
+    with pytest.raises(local.IapRequestError, match="independently supplied"):
+        local.validate_release(
+            tmp_path,
+            "d" * 40,
+            expected_release_sha256="e" * 64,
+            expected_image_digest=f"sha256:{'b' * 64}",
+            expected_image_config_digest=f"sha256:{'c' * 64}",
+        )
+
+    artifact = tmp_path / "artifact.zip"
+    artifact.write_bytes(b"not-the-approved-artifact")
+    artifact.chmod(0o600)
+    with pytest.raises(local.IapRequestError, match="independently supplied"):
+        local.validate_release_artifact(
+            artifact,
+            bundle=tmp_path,
+            release=release,
+            expected_artifact_sha256="f" * 64,
+        )
+
+
+def test_dry_run_plan_records_only_bounded_action_hashes(tmp_path: Path) -> None:
+    plan = executable_plan(tmp_path, "install")
+    encoded = json.dumps(plan, sort_keys=True)
+
+    assert [item["label"] for item in plan["actions"]] == [
+        "upload_request",
+        "execute_remote_request",
+        "cleanup_failed_transfer",
+    ]
+    assert all(set(item) == {"label", "argv_sha256"} for item in plan["actions"])
+    assert local.BOOTSTRAP_CODE not in encoded
+    assert "--command=" not in encoded
+    assert "sudo --" not in encoded
+
+
+def test_gcloud_environment_selects_absolute_compatible_python(
+    monkeypatch: pytest.MonkeyPatch,
+    tmp_path: Path,
+) -> None:
+    home = private_directory(tmp_path / "home")
+    config = private_directory(home / "gcloud")
+    python = tmp_path / "python311"
+    python.write_text("placeholder", encoding="utf-8")
+    python.chmod(0o700)
+    monkeypatch.setenv("HOME", str(home))
+    monkeypatch.setenv("CLOUDSDK_CONFIG", str(config))
+    monkeypatch.setenv("CLOUDSDK_PYTHON", str(python))
+
+    environment = local.gcloud_environment(
+        probe_runner=lambda argv, **_kwargs: completed(argv, stdout=b"Python 3.11.9\n")
+    )
+
+    assert environment["CLOUDSDK_PYTHON"] == str(python.resolve())
+    assert environment["CLOUDSDK_CONFIG"] == str(config.resolve())
+
+
+def test_local_result_receipt_is_private_and_self_hashed(tmp_path: Path) -> None:
+    plan = executable_plan(tmp_path, "verify")
+    result = {"bounded": "remote-result"}
+
+    path, receipt = local._write_result_receipt(plan, result)
+    stable = {key: value for key, value in receipt.items() if key != "receipt_sha256"}
+
+    assert stat.S_IMODE(path.stat().st_mode) == 0o600
+    assert receipt["receipt_sha256"] == local.package.canonical_sha256(stable)
+    assert json.loads(path.read_text(encoding="utf-8")) == receipt
+
+
+def test_execute_failure_does_not_forward_command_output(monkeypatch: pytest.MonkeyPatch, tmp_path: Path) -> None:
+    plan = executable_plan(tmp_path, "install")
+    monkeypatch.setattr(local, "validate_remote_main", lambda *_args, **_kwargs: None)
+
+    def runner(argv: list[str], **_kwargs: Any) -> subprocess.CompletedProcess[bytes]:
+        return completed(argv, returncode=1, stderr=b"password=extremely-secret")
+
+    with pytest.raises(local.IapRequestError) as caught:
+        local.execute_plan(plan, runner=runner, gcloud_env={"PATH": "/usr/bin"})
+    assert "extremely-secret" not in str(caught.value)
+    assert "password" not in str(caught.value).casefold()
+
+
+def test_execute_rejects_unsanitized_remote_result(monkeypatch: pytest.MonkeyPatch, tmp_path: Path) -> None:
+    plan = executable_plan(tmp_path, "verify")
+    monkeypatch.setattr(local, "validate_remote_main", lambda *_args, **_kwargs: None)
+    calls = 0
+
+    def runner(argv: list[str], **_kwargs: Any) -> subprocess.CompletedProcess[bytes]:
+        nonlocal calls
+        calls += 1
+        if calls == 1:
+            return completed(argv)
+        payload = {
+            "schema": local.REMOTE_RESULT_SCHEMA,
+            "status": "pass",
+            "operation": "verify",
+            "request_id": "iap-abcdefghijkl",
+            "access_token": "not-allowed",
+        }
+        return completed(argv, stdout=json.dumps(payload).encode())
+
+    with pytest.raises(local.IapRequestError) as caught:
+        local.execute_plan(plan, runner=runner, gcloud_env={"PATH": "/usr/bin"})
+    assert "not-allowed" not in str(caught.value)
+
+
+def test_execute_rederives_commands_and_rejects_plan_tampering(tmp_path: Path) -> None:
+    plan = executable_plan(tmp_path, "verify")
+    plan["actions"][1]["argv_sha256"] = "0" * 64
+
+    with pytest.raises(local.IapRequestError, match="fixed IAP plan"):
+        local.execute_plan(plan, runner=lambda argv, **_kwargs: completed(argv))
+
+
+@pytest.mark.parametrize("fail_pull", [False, True])
+def test_ephemeral_registry_credentials_are_always_removed(tmp_path: Path, fail_pull: bool) -> None:
+    calls: list[tuple[list[str], bytes | None]] = []
+
+    def runner(argv: list[str], **kwargs: Any) -> subprocess.CompletedProcess[bytes]:
+        input_bytes = kwargs.get("input")
+        calls.append((argv, input_bytes))
+        if argv[0] == "/usr/bin/curl":
+            return completed(argv, stdout=b'{"access_token":"abcdefghijklmnop","token_type":"Bearer"}')
+        if "pull" in argv and fail_pull:
+            return completed(argv, returncode=1, stderr=b"registry said secret")
+        return completed(argv)
+
+    reference = f"{remote.IMAGE_REPOSITORY}@sha256:{'a' * 64}"
+    if fail_pull:
+        with pytest.raises(remote.RemoteError, match="immutable image pull"):
+            remote.pull_immutable_image(reference, tmp_path, runner=runner)
+    else:
+        remote.pull_immutable_image(reference, tmp_path, runner=runner)
+    assert not any(path.name.startswith("docker-auth-") for path in tmp_path.iterdir())
+    assert all("abcdefghijklmnop" not in " ".join(argv) for argv, _input in calls)
+    assert any(input_bytes == b"abcdefghijklmnop\n" for _argv, input_bytes in calls)
+    assert any("logout" in argv for argv, _input in calls)
+
+
+def test_metadata_target_binds_project_zone_instance_and_service_account() -> None:
+    values = {
+        "instance/service-accounts/default/email": remote.SERVICE_ACCOUNT,
+        "project/project-id": remote.PROJECT,
+        "instance/name": remote.INSTANCE,
+        "instance/zone": f"projects/123/zones/{remote.ZONE}",
+    }
+
+    def runner(argv: list[str], **_kwargs: Any) -> subprocess.CompletedProcess[bytes]:
+        path = argv[-1].removeprefix(f"{remote.METADATA}/")
+        return completed(argv, stdout=values[path].encode())
+
+    remote.verify_service_account(runner=runner)
+
+    values["instance/name"] = "wrong-instance"
+    with pytest.raises(remote.RemoteError, match="target or service account"):
+        remote.verify_service_account(runner=runner)
+
+
+def test_outer_lock_and_reused_request_id_fail_closed(tmp_path: Path) -> None:
+    lock = tmp_path / "runner.lock"
+    first = remote._acquire_lock(lock, expected_uid=os.geteuid())
+    try:
+        with pytest.raises(remote.RemoteError, match="another request"):
+            remote._acquire_lock(lock, expected_uid=os.geteuid())
+    finally:
+        os.close(first)
+
+    root = tmp_path / "requests"
+    claimed = remote._claim_request_directory(
+        root,
+        "iap-abcdefghijkl",
+        expected_uid=os.geteuid(),
+        trusted_parent=tmp_path,
+    )
+    assert stat.S_IMODE(claimed.stat().st_mode) == 0o700
+    with pytest.raises(remote.RemoteError, match="already used"):
+        remote._claim_request_directory(
+            root,
+            "iap-abcdefghijkl",
+            expected_uid=os.geteuid(),
+            trusted_parent=tmp_path,
+        )
+
+
+def test_rollback_uses_only_explicit_receipt_bound_interface(tmp_path: Path) -> None:
+    observed: dict[str, Any] = {}
+
+    def rollback(release_sha256: str, **kwargs: Any) -> dict[str, Any]:
+        observed.update({"release_sha256": release_sha256, **kwargs})
+        return {"schema": "rollback", "status": "pass"}
+
+    installer = SimpleNamespace(rollback_release=rollback)
+    result = remote._execute_explicit_rollback(
+        installer_module=installer,
+        release_sha256="a" * 64,
+        paths=object(),
+        source_binding={"revision": "b" * 40},
+        runner=lambda argv, **_kwargs: completed(argv),
+    )
+
+    assert result["status"] == "pass"
+    assert observed["release_sha256"] == "a" * 64
+    assert observed["execute_rollback"] is True
+    assert "install_release" not in observed
+
+
+def test_rollback_fails_when_explicit_interface_is_absent(tmp_path: Path) -> None:
+    with pytest.raises(remote.RemoteError, match="lacked the explicit rollback"):
+        remote._execute_explicit_rollback(
+            installer_module=SimpleNamespace(),
+            release_sha256="a" * 64,
+            paths=object(),
+            source_binding={},
+            runner=lambda argv, **_kwargs: completed(argv),
+        )
+
+
+@pytest.mark.parametrize("failure_point", ["first_database", "restart", "second_database"])
+def test_post_install_proof_failure_executes_explicit_rollback(
+    monkeypatch: pytest.MonkeyPatch,
+    tmp_path: Path,
+    failure_point: str,
+) -> None:
+    release = {
+        "release_sha256": "a" * 64,
+        "image": {
+            "reference": f"{remote.IMAGE_REPOSITORY}@sha256:{'b' * 64}",
+            "digest": f"sha256:{'b' * 64}",
+            "config_digest": f"sha256:{'c' * 64}",
+        },
+        "image_input": {"runtime": {"teleo_git_head": "d" * 40}},
+    }
+    manifest = {
+        "operation": "install",
+        "request_id": "iap-abcdefghijkl",
+        "runner_revision": "e" * 40,
+        "release_revision": "d" * 40,
+        "manifest_sha256": "f" * 64,
+        "release": {
+            "release_sha256": release["release_sha256"],
+            "image_reference": release["image"]["reference"],
+            "image_digest": release["image"]["digest"],
+            "image_config_digest": release["image"]["config_digest"],
+        },
+        "release_artifact": {"archive_sha256": "1" * 64},
+        "installer_source": {"revision": "e" * 40},
+    }
+    verifications = iter(
+        [
+            {
+                "main_pid": 10,
+                "invocation_id": "2" * 32,
+                "container_id": "3" * 64,
+                "n_restarts": 0,
+            },
+            {
+                "main_pid": 11,
+                "invocation_id": "4" * 32,
+                "container_id": "5" * 64,
+                "n_restarts": 0,
+            },
+        ]
+    )
+    rollbacks: list[str] = []
+
+    class Paths:
+        @staticmethod
+        def under(_root: Path) -> object:
+            return object()
+
+    installer_module = SimpleNamespace(
+        SYSTEMCTL="/usr/bin/systemctl",
+        InstallPaths=Paths,
+        validate_release_bundle=lambda _bundle: (release, b"unit"),
+        _validate_source_binding=lambda _binding: None,
+        install_release=lambda *_args, **_kwargs: {"status": "pass"},
+        verify_running_release=lambda *_args, **_kwargs: next(verifications),
+        rollback_release=lambda release_sha256, **_kwargs: (
+            rollbacks.append(release_sha256)
+            or {
+                "schema": "livingip.leocleanNoSendRollbackResult.v1",
+                "status": "pass",
+                "mode": "executed",
+                "rolled_back_release_sha256": release_sha256,
+                "restored_release_sha256": None,
+                "restored_service_posture": "absent",
+                "verification": None,
+            }
+        ),
+    )
+    package_module = SimpleNamespace(
+        render_systemd_unit=lambda _release: "unit",
+        canonical_sha256=local.package.canonical_sha256,
+    )
+    monkeypatch.setattr(remote, "validate_original_release_artifact", lambda *_args, **_kwargs: None)
+    monkeypatch.setattr(remote, "_load_contract_modules", lambda _root: (package_module, installer_module))
+    monkeypatch.setattr(remote, "pull_immutable_image", lambda *_args, **_kwargs: None)
+    database_calls = 0
+
+    def verify_database(**_kwargs: Any) -> dict[str, Any]:
+        nonlocal database_calls
+        database_calls += 1
+        if failure_point == "first_database" and database_calls == 1:
+            raise remote.RemoteError("database failed")
+        if failure_point == "second_database" and database_calls == 2:
+            raise remote.RemoteError("database failed")
+        return {"status": "pass"}
+
+    monkeypatch.setattr(remote, "_verify_database", verify_database)
+
+    def runner(argv: list[str], **_kwargs: Any) -> subprocess.CompletedProcess[bytes]:
+        if failure_point == "restart" and argv[:2] == ["/usr/bin/systemctl", "restart"]:
+            return completed(argv, returncode=1)
+        return completed(argv)
+
+    with pytest.raises(remote.RemoteError, match="install_failed_rolled_back"):
+        remote.execute_request(
+            manifest,
+            tmp_path,
+            tmp_path,
+            "6" * 64,
+            "7" * 64,
+            runner=runner,
+        )
+    assert rollbacks == [release["release_sha256"]]
+
+
+@pytest.mark.parametrize(
+    ("posture", "restored_release", "verification"),
+    [
+        (
+            "active",
+            "e" * 64,
+            {
+                "active_state": "active",
+                "sub_state": "running",
+                "container_id": "f" * 64,
+            },
+        ),
+        ("inactive", "e" * 64, None),
+        ("absent", None, None),
+    ],
+)
+def test_rollback_result_accepts_exact_recorded_postures(
+    posture: str,
+    restored_release: str | None,
+    verification: dict[str, Any] | None,
+) -> None:
+    manifest = {
+        "request_id": "iap-abcdefghijkl",
+        "runner_revision": "a" * 40,
+        "release_revision": "d" * 40,
+        "manifest_sha256": "1" * 64,
+        "release_artifact": {"archive_sha256": "2" * 64},
+    }
+    release = {
+        "release_sha256": "b" * 64,
+        "image": {"digest": f"sha256:{'c' * 64}", "config_digest": f"sha256:{'d' * 64}"},
+    }
+    rollback = {
+        "schema": "livingip.leocleanNoSendRollbackResult.v1",
+        "status": "pass",
+        "mode": "executed",
+        "rolled_back_release_sha256": "b" * 64,
+        "restored_release_sha256": restored_release,
+        "restored_service_posture": posture,
+        "verification": verification,
+    }
+
+    result = remote._rollback_result(
+        manifest=manifest,
+        release=release,
+        rollback=rollback,
+        archive_sha256="3" * 64,
+        helper_sha256="4" * 64,
+    )
+
+    assert result["service"]["restored_posture"] == posture
+    assert result["database"] == {"status": "not_run_after_explicit_rollback"}
+
+
+def test_absent_rollback_rejects_invented_database_or_runtime_verification() -> None:
+    manifest = {
+        "request_id": "iap-abcdefghijkl",
+        "runner_revision": "a" * 40,
+        "release_revision": "d" * 40,
+        "manifest_sha256": "1" * 64,
+        "release_artifact": {"archive_sha256": "2" * 64},
+    }
+    release = {
+        "release_sha256": "b" * 64,
+        "image": {"digest": f"sha256:{'c' * 64}", "config_digest": f"sha256:{'d' * 64}"},
+    }
+    rollback = {
+        "schema": "livingip.leocleanNoSendRollbackResult.v1",
+        "status": "pass",
+        "mode": "executed",
+        "rolled_back_release_sha256": "b" * 64,
+        "restored_release_sha256": None,
+        "restored_service_posture": "absent",
+        "verification": {"active_state": "active"},
+    }
+    with pytest.raises(remote.RemoteError, match="absent rollback"):
+        remote._rollback_result(
+            manifest=manifest,
+            release=release,
+            rollback=rollback,
+            archive_sha256="3" * 64,
+            helper_sha256="4" * 64,
+        )

tests/test_gcp_leoclean_nosend_installer.py

@@ -337,6 +337,20 @@ def test_alternate_repo_root_cannot_bind_the_executing_installer(tmp_path: Path)
         installer.parse_args(["--bundle", str(tmp_path), "--repo-root", str(tmp_path)])
 
 
+def test_explicit_rollback_cli_is_dry_run_unless_execute_flag_is_separate() -> None:
+    release_sha256 = "a" * 64
+
+    dry_run = installer.parse_args(["--rollback-release", release_sha256])
+    execute = installer.parse_args(["--rollback-release", release_sha256, "--execute-rollback"])
+
+    assert dry_run.execute_rollback is False
+    assert execute.execute_rollback is True
+    with pytest.raises(installer.InstallError, match="invalid arguments"):
+        installer.parse_args(["--execute-rollback"])
+    with pytest.raises(installer.InstallError, match="invalid arguments"):
+        installer.parse_args(["--bundle", "/tmp/release", "--rollback-release", release_sha256])
+
+
 def test_rehashed_alternate_source_bytes_cannot_bypass_actual_source_binding(tmp_path: Path) -> None:
     bundle, _release, _image = release_bundle(tmp_path, "2")
     paths = install_paths(tmp_path)
@@ -372,6 +386,7 @@ def test_disposable_root_install_is_digest_bound_idempotent_and_exact(tmp_path:
     )
 
     assert receipt["status"] == "pass"
+    assert receipt["schema"] == "livingip.leocleanNoSendInstallReceipt.v2"
     assert receipt["release_sha256"] == release["release_sha256"]
     assert host.restart_calls == 1
     assert paths.unit.read_text(encoding="utf-8") == package.render_systemd_unit(release)
@@ -379,6 +394,11 @@ def test_disposable_root_install_is_digest_bound_idempotent_and_exact(tmp_path:
     assert paths.helper.stat().st_mode & 0o777 == 0o755
     assert paths.release.stat().st_mode & 0o777 == 0o600
     assert paths.receipt.stat().st_mode & 0o777 == 0o600
+    assert paths.rollback_state.stat().st_mode & 0o777 == 0o600
+    rollback_state = json.loads(paths.rollback_state.read_text(encoding="utf-8"))
+    assert rollback_state["schema"] == installer.ROLLBACK_STATE_SCHEMA
+    assert rollback_state["current_release_sha256"] == release["release_sha256"]
+    assert receipt["rollback"]["state_sha256"] == package.sha256_file(paths.rollback_state)
     assert not list(paths.docker_config.iterdir())
     assert not list(paths.unit.parent.glob(f".{paths.unit.name}.*.tmp"))
 
@@ -962,6 +982,331 @@ def test_effective_process_environment_override_is_rejected_and_rolled_back(tmp_
     assert not paths.state_dir.exists()
 
 
+def test_explicit_rollback_is_dry_run_by_default_and_removes_first_install(tmp_path: Path) -> None:
+    bundle, release, image = release_bundle(tmp_path, "2")
+    paths = install_paths(tmp_path)
+    host = FakeHost(paths)
+    host.add_release(release, image)
+    installer.install_release(
+        bundle,
+        paths=paths,
+        source_binding=source_binding(),
+        runner=host,
+        sleeper=lambda _seconds: None,
+        execute_restart=True,
+    )
+    before_calls = list(host.calls)
+    before_stops = host.stop_calls
+    before_restarts = host.restart_calls
+
+    plan = installer.rollback_release(
+        str(release["release_sha256"]),
+        paths=paths,
+        source_binding=source_binding(),
+        runner=host,
+        sleeper=lambda _seconds: None,
+    )
+
+    assert plan["mode"] == "dry_run"
+    assert plan["installed_release_sha256"] == release["release_sha256"]
+    assert plan["target_release_sha256"] is None
+    assert plan["target_service_posture"] == "absent"
+    assert host.stop_calls == before_stops
+    assert host.restart_calls == before_restarts
+    assert host.calls != before_calls
+    assert host.active
+
+    with pytest.raises(installer.InstallError, match="did not match the expected release hash"):
+        installer.rollback_release(
+            "f" * 64,
+            paths=paths,
+            source_binding=source_binding(),
+            runner=host,
+            sleeper=lambda _seconds: None,
+            execute_rollback=True,
+        )
+    assert host.stop_calls == before_stops
+
+    result = installer.rollback_release(
+        str(release["release_sha256"]),
+        paths=paths,
+        source_binding=source_binding(),
+        runner=host,
+        sleeper=lambda _seconds: None,
+        execute_rollback=True,
+    )
+
+    assert result == {
+        "schema": installer.ROLLBACK_RESULT_SCHEMA,
+        "status": "pass",
+        "mode": "executed",
+        "rolled_back_release_sha256": release["release_sha256"],
+        "restored_release_sha256": None,
+        "restored_service_posture": "absent",
+        "verification": None,
+        "authority": {
+            "image_pulled": False,
+            "cloud_or_database_authority_changed": False,
+            "transport_authority_changed": False,
+        },
+    }
+    assert not host.active
+    assert host.loaded_release is None
+    assert not paths.unit.exists()
+    assert not paths.helper.exists()
+    assert not paths.state_dir.exists()
+
+
+def test_explicit_rollback_restores_exact_previous_active_release_and_history(tmp_path: Path) -> None:
+    first_bundle, first_release, first_image = release_bundle(tmp_path, "2")
+    second_bundle, second_release, second_image = release_bundle(tmp_path, "4")
+    paths = install_paths(tmp_path)
+    host = FakeHost(paths)
+    host.add_release(first_release, first_image)
+    host.add_release(second_release, second_image)
+    installer.install_release(
+        first_bundle,
+        paths=paths,
+        source_binding=source_binding(),
+        runner=host,
+        sleeper=lambda _seconds: None,
+        execute_restart=True,
+    )
+    first_files = {role: path.read_bytes() for role, (path, _mode) in installer._managed_files(paths).items()}
+    installer.install_release(
+        second_bundle,
+        paths=paths,
+        source_binding=source_binding(),
+        runner=host,
+        sleeper=lambda _seconds: None,
+        execute_restart=True,
+    )
+
+    result = installer.rollback_release(
+        str(second_release["release_sha256"]),
+        paths=paths,
+        source_binding=source_binding(),
+        runner=host,
+        sleeper=lambda _seconds: None,
+        execute_rollback=True,
+    )
+
+    assert result["restored_release_sha256"] == first_release["release_sha256"]
+    assert result["restored_service_posture"] == "active"
+    assert host.current_release == first_release
+    assert {role: path.read_bytes() for role, (path, _mode) in installer._managed_files(paths).items()} == first_files
+
+    nested = installer.rollback_release(
+        str(first_release["release_sha256"]),
+        paths=paths,
+        source_binding=source_binding(),
+        runner=host,
+        sleeper=lambda _seconds: None,
+        execute_rollback=True,
+    )
+    assert nested["restored_service_posture"] == "absent"
+    assert not paths.state_dir.exists()
+
+
+def test_explicit_rollback_restores_previous_inactive_posture(tmp_path: Path) -> None:
+    first_bundle, first_release, first_image = release_bundle(tmp_path, "2")
+    second_bundle, second_release, second_image = release_bundle(tmp_path, "4")
+    paths = install_paths(tmp_path)
+    host = FakeHost(paths)
+    host.add_release(first_release, first_image)
+    host.add_release(second_release, second_image)
+    installer.install_release(
+        first_bundle,
+        paths=paths,
+        source_binding=source_binding(),
+        runner=host,
+        sleeper=lambda _seconds: None,
+        execute_restart=True,
+    )
+    host._stop()
+    installer.install_release(
+        second_bundle,
+        paths=paths,
+        source_binding=source_binding(),
+        runner=host,
+        sleeper=lambda _seconds: None,
+        execute_restart=True,
+    )
+    restarts = host.restart_calls
+
+    result = installer.rollback_release(
+        str(second_release["release_sha256"]),
+        paths=paths,
+        source_binding=source_binding(),
+        runner=host,
+        sleeper=lambda _seconds: None,
+        execute_rollback=True,
+    )
+
+    assert result["restored_release_sha256"] == first_release["release_sha256"]
+    assert result["restored_service_posture"] == "inactive"
+    assert not host.active
+    assert host.loaded_release == first_release
+    assert host.restart_calls == restarts
+
+
+def test_failed_explicit_rollback_recovers_current_release(tmp_path: Path) -> None:
+    first_bundle, first_release, first_image = release_bundle(tmp_path, "2")
+    second_bundle, second_release, second_image = release_bundle(tmp_path, "4")
+    paths = install_paths(tmp_path)
+    host = FakeHost(paths)
+    host.add_release(first_release, first_image)
+    host.add_release(second_release, second_image)
+    installer.install_release(
+        first_bundle,
+        paths=paths,
+        source_binding=source_binding(),
+        runner=host,
+        sleeper=lambda _seconds: None,
+        execute_restart=True,
+    )
+    installer.install_release(
+        second_bundle,
+        paths=paths,
+        source_binding=source_binding(),
+        runner=host,
+        sleeper=lambda _seconds: None,
+        execute_restart=True,
+    )
+    second_files = {role: path.read_bytes() for role, (path, _mode) in installer._managed_files(paths).items()}
+    fail_next_reload = True
+
+    def runner(arguments: list[str]) -> subprocess.CompletedProcess[bytes]:
+        nonlocal fail_next_reload
+        if arguments[:2] == [installer.SYSTEMCTL, "daemon-reload"] and fail_next_reload:
+            fail_next_reload = False
+            return subprocess.CompletedProcess(arguments, 1, stdout=b"", stderr=b"injected")
+        return host(arguments)
+
+    with pytest.raises(installer.InstallError, match="systemd daemon-reload failed"):
+        installer.rollback_release(
+            str(second_release["release_sha256"]),
+            paths=paths,
+            source_binding=source_binding(),
+            runner=runner,
+            sleeper=lambda _seconds: None,
+            execute_rollback=True,
+        )
+
+    assert host.active
+    assert host.current_release == second_release
+    assert {role: path.read_bytes() for role, (path, _mode) in installer._managed_files(paths).items()} == second_files
+
+
+def test_explicit_rollback_rejects_receipt_drift_before_service_stop(tmp_path: Path) -> None:
+    bundle, release, image = release_bundle(tmp_path, "2")
+    paths = install_paths(tmp_path)
+    host = FakeHost(paths)
+    host.add_release(release, image)
+    installer.install_release(
+        bundle,
+        paths=paths,
+        source_binding=source_binding(),
+        runner=host,
+        sleeper=lambda _seconds: None,
+        execute_restart=True,
+    )
+    receipt = json.loads(paths.receipt.read_text(encoding="utf-8"))
+    receipt["release_sha256"] = "f" * 64
+    paths.receipt.write_text(json.dumps(receipt), encoding="utf-8")
+    paths.receipt.chmod(0o600)
+    stop_calls = host.stop_calls
+
+    with pytest.raises(installer.InstallError, match=r"(installed|snapshot install) receipt was invalid"):
+        installer.rollback_release(
+            str(release["release_sha256"]),
+            paths=paths,
+            source_binding=source_binding(),
+            runner=host,
+            sleeper=lambda _seconds: None,
+            execute_rollback=True,
+        )
+
+    assert host.stop_calls == stop_calls
+    assert host.active
+
+
+def test_explicit_rollback_rejects_path_inode_drift_before_service_stop(tmp_path: Path) -> None:
+    bundle, release, image = release_bundle(tmp_path, "2")
+    paths = install_paths(tmp_path)
+    host = FakeHost(paths)
+    host.add_release(release, image)
+    installer.install_release(
+        bundle,
+        paths=paths,
+        source_binding=source_binding(),
+        runner=host,
+        sleeper=lambda _seconds: None,
+        execute_restart=True,
+    )
+    prepared = installer._prepare_explicit_rollback(
+        str(release["release_sha256"]),
+        paths=paths,
+        source_binding=source_binding(),
+        runner=host,
+        sleeper=lambda _seconds: None,
+    )
+    unit = paths.unit.read_bytes()
+    paths.unit.unlink()
+    paths.unit.write_bytes(unit)
+    paths.unit.chmod(0o644)
+    stop_calls = host.stop_calls
+
+    with pytest.raises(installer.InstallError, match="file identity drifted"):
+        installer._execute_explicit_rollback(
+            str(release["release_sha256"]),
+            prepared,
+            paths=paths,
+            runner=host,
+            sleeper=lambda _seconds: None,
+        )
+
+    assert host.stop_calls == stop_calls
+    assert host.active
+
+
+def test_explicit_rollback_rejects_pid_drift_before_service_stop(tmp_path: Path) -> None:
+    bundle, release, image = release_bundle(tmp_path, "2")
+    paths = install_paths(tmp_path)
+    host = FakeHost(paths)
+    host.add_release(release, image)
+    installer.install_release(
+        bundle,
+        paths=paths,
+        source_binding=source_binding(),
+        runner=host,
+        sleeper=lambda _seconds: None,
+        execute_restart=True,
+    )
+    prepared = installer._prepare_explicit_rollback(
+        str(release["release_sha256"]),
+        paths=paths,
+        source_binding=source_binding(),
+        runner=host,
+        sleeper=lambda _seconds: None,
+    )
+    host.main_pid += 1
+    host._write_process_environment()
+    stop_calls = host.stop_calls
+
+    with pytest.raises(installer.InstallError, match=r"changed during verification|drifted before rollback stop"):
+        installer._execute_explicit_rollback(
+            str(release["release_sha256"]),
+            prepared,
+            paths=paths,
+            runner=host,
+            sleeper=lambda _seconds: None,
+        )
+
+    assert host.stop_calls == stop_calls
+    assert host.active
+
+
 def test_control_helper_stops_only_trusted_full_container_id(tmp_path: Path) -> None:
     cid_parent = tmp_path / "runtime"
     cid_parent.mkdir(mode=0o700)