Merge pull request #242 from living-ip/fix/leoclean-install-convergence

Converge leoclean no-send installation state

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

@@ -324,9 +324,22 @@ disappears it rechecks that the canonical name is still unbound before removing
 the CID file. It never removes a container, uses a mutable name as stop
 authority, or invokes `--force`.
 
-An already installed release is verified without a restart. During an upgrade,
+Install is convergent: it means ensuring the requested exact release is
-the installer accepts only a stable two-readback active/running or inactive/dead
+installed, active, and verified. An exact active release is verified as a
-prior posture. Before trusting the old `ExecStop`, it exact-binds effective
+no-op. An exact inactive release receives one controlled `start`, while a
+missing or different release uses the replacement transaction. The persisted
+install history is not rewritten by exact-release convergence. New replacement
+history uses `livingip.leocleanNoSendInstallReceipt.v3`; existing
+`livingip.leocleanNoSendInstallReceipt.v2` history remains readable. Each
+invocation returns a separate
+`livingip.leocleanNoSendInstallExecution.v1` result whose transition records
+`verified_noop`, `controlled_start`, `installed`, or `replaced`, the prior
+posture and restart counter, the service action, and whether managed files were
+replaced.
+
+During an upgrade, the installer accepts only a stable two-readback
+active/running or inactive/dead prior posture. Before trusting the old
+`ExecStop`, it exact-binds effective
 `ExecStartPre`, `ExecStart`, `ExecStop`, runtime-directory and systemd security
 properties to the release and verifies the supervising PID's executable,
 command line and cgroup through `/proc`. It then stops the prior service through
@@ -339,6 +352,13 @@ two readbacks. On failure it stops the candidate, restores the exact prior
 files, modes and active/inactive/absent posture, and reloads systemd. An
 incomplete rollback fails closed and is reported as such.
 
+The IAP result and bounded failure contracts use the v2 result schemas. A
+post-install failure restores the transition's proven prior posture without
+consuming historical rollback for an exact-release no-op or controlled start.
+Replacement failures use only the receipt-bound explicit rollback. Malformed
+or missing transition data forbids historical rollback and instead contains
+the service fail closed with a sanitized failure stage and reason.
+
 This installer does not enable the service, provision IAM, roles or secrets,
 contact GCP or Cloud SQL, change Telegram traffic, touch the VPS, approve a
 proposal, or promote production. Those remain later, separately authorized

ops/gcp_leoclean_nosend_iap_remote.py

@@ -23,9 +23,9 @@ from pathlib import Path
 from typing import Any
 
 SCHEMA = "livingip.leocleanNoSendIapRequest.v1"
-RESULT_SCHEMA = "livingip.leocleanNoSendIapRemoteResult.v1"
+RESULT_SCHEMA = "livingip.leocleanNoSendIapRemoteResult.v2"
-FAILURE_RESULT_SCHEMA = "livingip.leocleanNoSendIapFailure.v1"
+FAILURE_RESULT_SCHEMA = "livingip.leocleanNoSendIapFailure.v2"
-OPERATION_FAILURE_RESULT_SCHEMA = "livingip.leocleanNoSendIapOperationalFailure.v1"
+OPERATION_FAILURE_RESULT_SCHEMA = "livingip.leocleanNoSendIapOperationalFailure.v2"
 PROJECT = "teleo-501523"
 ZONE = "europe-west6-a"
 INSTANCE = "teleo-staging-1"
@@ -99,6 +99,164 @@ class RemoteOperationFailure(RemoteError):
         self.result = dict(result)
 
 
+class PostureRecoveryFailure(RemoteError):
+    """A posture recovery failed after a bounded amount of service action."""
+
+    def __init__(self, *, executed: bool) -> None:
+        super().__init__("posture_recovery_failed")
+        self.executed = executed
+
+
+class ExplicitRollbackFailure(RemoteError):
+    """An explicit rollback failed before or after entering its mutation phase."""
+
+    def __init__(self, *, executed: bool) -> None:
+        super().__init__("explicit_rollback_failed")
+        self.executed = executed
+
+
+def _validate_installer_transition(value: Any) -> dict[str, Any]:
+    _require(
+        isinstance(value, dict)
+        and set(value)
+        == {
+            "outcome",
+            "prior_posture",
+            "prior_n_restarts",
+            "final_posture",
+            "service_action",
+            "managed_files_replaced",
+        },
+        "installer transition was invalid",
+    )
+    outcome = value.get("outcome")
+    prior_posture = value.get("prior_posture")
+    prior_n_restarts = value.get("prior_n_restarts")
+    service_action = value.get("service_action")
+    replaced = value.get("managed_files_replaced")
+    _require(
+        isinstance(prior_n_restarts, int)
+        and not isinstance(prior_n_restarts, bool)
+        and prior_n_restarts >= 0
+        and value.get("final_posture") == "active"
+        and (
+            (outcome, prior_posture, service_action, replaced)
+            == ("verified_noop", "active", "none", False)
+            or (outcome, prior_posture, service_action, replaced)
+            == ("controlled_start", "inactive", "start", False)
+            or (outcome, prior_posture, service_action, replaced)
+            == ("installed", "absent", "restart", True)
+            or (
+                outcome == "replaced"
+                and prior_posture in {"active", "inactive"}
+                and service_action == "restart"
+                and replaced is True
+            )
+        ),
+        "installer transition was invalid",
+    )
+    return dict(value)
+
+
+def _validate_install_execution_result(
+    value: Any,
+    *,
+    release: Mapping[str, Any],
+    installer_module: Any,
+) -> dict[str, Any]:
+    _require(
+        isinstance(value, dict)
+        and set(value)
+        == {
+            "schema",
+            "status",
+            "outcome",
+            "release_sha256",
+            "persisted_receipt",
+            "transition",
+            "verification",
+        },
+        "install execution result was invalid",
+    )
+    transition = _validate_installer_transition(value.get("transition"))
+    persisted_receipt = value.get("persisted_receipt")
+    outcome = value.get("outcome")
+    _require(
+        value.get("schema") == installer_module.INSTALL_EXECUTION_RESULT_SCHEMA
+        and value.get("status") == "pass"
+        and outcome == transition["outcome"]
+        and value.get("release_sha256") == release["release_sha256"]
+        and isinstance(persisted_receipt, dict)
+        and set(persisted_receipt) == {"schema", "sha256"}
+        and persisted_receipt.get("schema")
+        in {
+            installer_module.LEGACY_INSTALL_RECEIPT_SCHEMA,
+            installer_module.INSTALL_RECEIPT_SCHEMA,
+        }
+        and isinstance(persisted_receipt.get("sha256"), str)
+        and HEX_64.fullmatch(persisted_receipt["sha256"]) is not None
+        and isinstance(value.get("verification"), dict)
+        and (
+            outcome in {"verified_noop", "controlled_start"}
+            or (
+                outcome in {"installed", "replaced"}
+                and persisted_receipt["schema"] == installer_module.INSTALL_RECEIPT_SCHEMA
+            )
+        ),
+        "install execution result was invalid",
+    )
+    return transition
+
+
+def _rollback_summary(
+    *,
+    status: str,
+    mode: str,
+    executed: bool,
+    restored_posture: str,
+    prior_posture_preserved: bool,
+) -> dict[str, Any]:
+    value = {
+        "status": status,
+        "mode": mode,
+        "executed": executed,
+        "restored_posture": restored_posture,
+        "prior_posture_preserved": prior_posture_preserved,
+    }
+    _require(
+        (
+            status == "pass"
+            and (
+                (
+                    mode in {"installer_automatic", "explicit"}
+                    and executed is True
+                    and restored_posture in {"active", "inactive", "absent"}
+                )
+                or (
+                    mode == "posture_recovery"
+                    and restored_posture in {"active", "inactive"}
+                )
+            )
+            and prior_posture_preserved is True
+        )
+        or (
+            status == "not_run"
+            and mode == "not_applicable"
+            and executed is False
+            and restored_posture == "unchanged"
+            and prior_posture_preserved is True
+        )
+        or (
+            status == "fail"
+            and mode in {"explicit", "posture_recovery"}
+            and restored_posture == "unknown"
+            and prior_posture_preserved is False
+        ),
+        "rollback summary was invalid",
+    )
+    return value
+
+
 def _require(condition: bool, message: str) -> None:
     if not condition:
         raise RemoteError(message)
@@ -716,20 +874,43 @@ def _execute_explicit_rollback(
 
     rollback = getattr(installer_module, "rollback_release", None)
     _require(callable(rollback), "reviewed installer lacked the explicit rollback interface")
-    result = rollback(
+    executed = False
-        release_sha256,
+
-        paths=paths,
+    def contract_runner(argv: list[str]) -> subprocess.CompletedProcess[bytes]:
-        source_binding=source_binding,
+        return _run(
-        runner=lambda argv: _run(
             argv,
             runner=runner,
             label="rollback command",
             require_success=False,
-        ),
+        )
+
+    try:
+        plan = rollback(
+            release_sha256,
+            paths=paths,
+            source_binding=source_binding,
+            runner=contract_runner,
+            execute_rollback=False,
+        )
+        _require(
+            isinstance(plan, dict)
+            and plan.get("status") == "pass"
+            and plan.get("mode") == "dry_run"
+            and plan.get("expected_release_sha256") == release_sha256,
+            "explicit rollback preflight did not pass",
+        )
+        executed = True
+        result = rollback(
+            release_sha256,
+            paths=paths,
+            source_binding=source_binding,
+            runner=contract_runner,
             execute_rollback=True,
         )
         _require(isinstance(result, dict) and result.get("status") == "pass", "explicit rollback did not pass")
         return result
+    except Exception as exc:
+        raise ExplicitRollbackFailure(executed=executed) from exc
 
 
 def _rollback_result(
@@ -808,7 +989,7 @@ def _verification_failure_result(
     manifest: Mapping[str, Any],
     release: Mapping[str, Any],
     failure: Exception,
-    rollback: Mapping[str, str],
+    rollback: Mapping[str, Any],
     archive_sha256: str,
     helper_sha256: str,
     installer_module: Any,
@@ -822,13 +1003,13 @@ def _verification_failure_result(
         "verification failure was not allowlisted",
     )
     _require(
-        rollback
+        dict(rollback)
-        in (
+        == _rollback_summary(
-            {"status": "pass", "mode": "installer_automatic", "restored_posture": "prior"},
+            status=str(rollback.get("status")),
-            {"status": "pass", "mode": "explicit", "restored_posture": "active"},
+            mode=str(rollback.get("mode")),
-            {"status": "pass", "mode": "explicit", "restored_posture": "inactive"},
+            executed=rollback.get("executed") is True,
-            {"status": "pass", "mode": "explicit", "restored_posture": "absent"},
+            restored_posture=str(rollback.get("restored_posture")),
-            {"status": "not_run", "mode": "not_applicable", "restored_posture": "unchanged"},
+            prior_posture_preserved=rollback.get("prior_posture_preserved") is True,
         ),
         "verification rollback summary was invalid",
     )
@@ -860,6 +1041,25 @@ def _verification_failure_result(
     }
 
 
+def _installer_failure_rollback(failure: Exception) -> dict[str, Any]:
+    rollback_executed = getattr(failure, "rollback_executed", None)
+    restored_posture = getattr(failure, "restored_posture", None)
+    prior_posture_preserved = getattr(failure, "prior_posture_preserved", None)
+    _require(
+        isinstance(rollback_executed, bool)
+        and restored_posture in {"active", "inactive", "absent"}
+        and prior_posture_preserved is True,
+        "installer failure lacked a proven recovery posture",
+    )
+    return _rollback_summary(
+        status="pass" if rollback_executed else "not_run",
+        mode="installer_automatic" if rollback_executed else "not_applicable",
+        executed=rollback_executed,
+        restored_posture=str(restored_posture) if rollback_executed else "unchanged",
+        prior_posture_preserved=True,
+    )
+
+
 def _is_verification_failure(installer_module: Any, failure: Exception) -> bool:
     failure_type = getattr(installer_module, "VerificationError", None)
     return isinstance(failure_type, type) and isinstance(failure, failure_type)
@@ -899,6 +1099,7 @@ def _rollback_proven_service(posture: str) -> dict[str, Any]:
         "main_pid": None,
         "n_restarts": None,
         "candidate_container": "not_observed",
+        "service_action_executed": False,
     }
 
 
@@ -909,6 +1110,8 @@ def _fail_closed_inert_service(
     paths: Any,
     runner: Runner,
 ) -> dict[str, Any]:
+    service_action_executed = False
+
     def contract_runner(argv: list[str]) -> subprocess.CompletedProcess[bytes]:
         return _run(
             argv,
@@ -949,6 +1152,7 @@ def _fail_closed_inert_service(
                 and verification.get("n_restarts") == before.state.get("NRestarts"),
                 "running service identity changed before fail-closed stop",
             )
+            service_action_executed = True
             systemctl("stop", contract_runner)
         after = stable_service_posture(
             release,
@@ -979,6 +1183,7 @@ def _fail_closed_inert_service(
             "main_pid": 0,
             "n_restarts": after.state["NRestarts"],
             "candidate_container": "absent",
+            "service_action_executed": service_action_executed,
         }
     except Exception:
         return {
@@ -989,16 +1194,128 @@ def _fail_closed_inert_service(
             "main_pid": None,
             "n_restarts": None,
             "candidate_container": "unknown",
+            "service_action_executed": service_action_executed,
         }
 
 
+def _restore_transition_posture(
+    *,
+    installer_module: Any,
+    release: Mapping[str, Any],
+    transition: Mapping[str, Any],
+    paths: Any,
+    runner: Runner,
+) -> dict[str, Any]:
+    checked = _validate_installer_transition(transition)
+    desired = str(checked["prior_posture"])
+    _require(
+        checked["outcome"] in {"verified_noop", "controlled_start"}
+        and desired in {"active", "inactive"},
+        "installer transition required release rollback",
+    )
+
+    def contract_runner(argv: list[str]) -> subprocess.CompletedProcess[bytes]:
+        return _run(
+            argv,
+            runner=runner,
+            label="prior posture recovery command",
+            require_success=False,
+        )
+
+    systemctl = getattr(installer_module, "_systemctl", None)
+    stable_service_posture = getattr(installer_module, "_stable_service_posture", None)
+    verify_running_release = getattr(installer_module, "verify_running_release", None)
+    require_candidate_absent = getattr(installer_module, "_require_candidate_absent", None)
+    _require(
+        callable(systemctl)
+        and callable(stable_service_posture)
+        and callable(verify_running_release)
+        and callable(require_candidate_absent),
+        "reviewed installer lacked posture recovery controls",
+    )
+    executed = False
+    try:
+        observed = stable_service_posture(
+            release,
+            runner=contract_runner,
+            paths=paths,
+            sleeper=time.sleep,
+        )
+        _require(observed.kind in {"active", "inactive"}, "managed service posture was not trusted")
+        if desired == "inactive":
+            if observed.kind == "active":
+                verification = verify_running_release(
+                    release,
+                    runner=contract_runner,
+                    paths=paths,
+                    sleeper=time.sleep,
+                )
+                _require(
+                    verification.get("main_pid") == observed.state.get("MainPID")
+                    and verification.get("invocation_id") == observed.state.get("InvocationID")
+                    and verification.get("n_restarts") == observed.state.get("NRestarts"),
+                    "running service identity changed before posture recovery stop",
+                )
+                executed = True
+                systemctl("stop", contract_runner)
+            final = stable_service_posture(
+                release,
+                runner=contract_runner,
+                paths=paths,
+                sleeper=time.sleep,
+            )
+            _require(
+                final.kind == "inactive"
+                and final.state.get("NRestarts") == checked["prior_n_restarts"],
+                "prior inactive posture was not restored",
+            )
+            require_candidate_absent(runner=contract_runner, paths=paths, sleeper=time.sleep)
+        else:
+            if observed.kind == "inactive":
+                require_candidate_absent(runner=contract_runner, paths=paths, sleeper=time.sleep)
+                executed = True
+                systemctl("start", contract_runner)
+            final = stable_service_posture(
+                release,
+                runner=contract_runner,
+                paths=paths,
+                sleeper=time.sleep,
+            )
+            _require(
+                final.kind == "active"
+                and final.state.get("NRestarts") == checked["prior_n_restarts"],
+                "prior active posture was not restored",
+            )
+            verification = verify_running_release(
+                release,
+                runner=contract_runner,
+                paths=paths,
+                sleeper=time.sleep,
+            )
+            _require(
+                verification.get("main_pid") == final.state.get("MainPID")
+                and verification.get("invocation_id") == final.state.get("InvocationID")
+                and verification.get("n_restarts") == checked["prior_n_restarts"],
+                "restored active service identity was unstable",
+            )
+    except Exception as exc:
+        raise PostureRecoveryFailure(executed=executed) from exc
+    return _rollback_summary(
+        status="pass",
+        mode="posture_recovery",
+        executed=executed,
+        restored_posture=desired,
+        prior_posture_preserved=True,
+    )
+
+
 def _operational_failure_result(
     *,
     manifest: Mapping[str, Any],
     release: Mapping[str, Any],
     failure_stage: str,
     failure_reason: str,
-    rollback: Mapping[str, str],
+    rollback: Mapping[str, Any],
     fail_closed: Mapping[str, Any],
     archive_sha256: str,
     helper_sha256: str,
@@ -1006,7 +1323,8 @@ def _operational_failure_result(
 ) -> dict[str, Any]:
     _require(
         manifest.get("operation") == "install"
-        and set(rollback) == {"status", "mode", "restored_posture"}
+        and set(rollback)
+        == {"status", "mode", "executed", "restored_posture", "prior_posture_preserved"}
         and (
             (failure_stage, failure_reason) in POST_INSTALL_FAILURES
             or (failure_stage, failure_reason) in installer_module.VERIFICATION_FAILURES
@@ -1015,14 +1333,22 @@ def _operational_failure_result(
     )
     rollback_passed = (
         rollback.get("status") == "pass"
-        and rollback.get("mode") == "explicit"
+        and rollback.get("mode") in {"explicit", "posture_recovery"}
+        and isinstance(rollback.get("executed"), bool)
         and rollback.get("restored_posture") in {"active", "inactive", "absent"}
+        and (
+            rollback.get("mode") == "explicit"
+            or rollback.get("restored_posture") in {"active", "inactive"}
+        )
+        and rollback.get("prior_posture_preserved") is True
+    )
+    rollback_failed = (
+        rollback.get("status") == "fail"
+        and rollback.get("mode") in {"explicit", "posture_recovery"}
+        and isinstance(rollback.get("executed"), bool)
+        and rollback.get("restored_posture") == "unknown"
+        and rollback.get("prior_posture_preserved") is False
     )
-    rollback_failed = rollback == {
-        "status": "fail",
-        "mode": "explicit",
-        "restored_posture": "unknown",
-    }
     verification_pair = (failure_stage, failure_reason) in installer_module.VERIFICATION_FAILURES
     _require(
         (rollback_passed and not verification_pair) or rollback_failed,
@@ -1045,6 +1371,7 @@ def _operational_failure_result(
                     "main_pid": None,
                     "n_restarts": None,
                     "candidate_container": "unknown",
+                    "service_action_executed": fail_closed.get("service_action_executed"),
                 },
                 {
                     "status": "pass",
@@ -1054,6 +1381,7 @@ def _operational_failure_result(
                     "main_pid": 0,
                     "n_restarts": fail_closed.get("n_restarts"),
                     "candidate_container": "absent",
+                    "service_action_executed": fail_closed.get("service_action_executed"),
                 },
             )
             and (
@@ -1063,7 +1391,8 @@ def _operational_failure_result(
                     and not isinstance(fail_closed.get("n_restarts"), bool)
                     and fail_closed["n_restarts"] >= 0
                 )
-            ),
+            )
+            and isinstance(fail_closed.get("service_action_executed"), bool),
             "post-install fail-closed summary was invalid",
         )
     return {
@@ -1156,7 +1485,7 @@ def execute_request(
                     manifest=manifest,
                     release=release,
                     failure=exc,
-                    rollback={"status": "pass", "mode": "installer_automatic", "restored_posture": "prior"},
+                    rollback=_installer_failure_rollback(exc),
                     archive_sha256=archive_sha256,
                     helper_sha256=helper_sha256,
                     installer_module=installer_module,
@@ -1189,8 +1518,13 @@ def execute_request(
     if operation == "install":
         failure_stage = "install_receipt"
         failure_reason = "install_receipt_invalid"
+        installer_transition: dict[str, Any] | None = None
         try:
-            _require(install_receipt.get("status") == "pass", "installation did not pass")
+            installer_transition = _validate_install_execution_result(
+                install_receipt,
+                release=release,
+                installer_module=installer_module,
+            )
             failure_stage = "service_before_restart"
             failure_reason = "service_verification_failed"
             before = installer_module.verify_running_release(
@@ -1248,6 +1582,37 @@ def execute_request(
                 fallback_stage=failure_stage,
                 fallback_reason=failure_reason,
             )
+            if installer_transition is None:
+                fail_closed = _fail_closed_inert_service(
+                    installer_module=installer_module,
+                    release=release,
+                    paths=paths,
+                    runner=runner,
+                )
+                raise RemoteOperationFailure(
+                    _operational_failure_result(
+                        manifest=manifest,
+                        release=release,
+                        failure_stage="install_receipt",
+                        failure_reason="install_receipt_invalid",
+                        rollback=_rollback_summary(
+                            status="fail",
+                            mode="posture_recovery",
+                            executed=bool(fail_closed["service_action_executed"]),
+                            restored_posture="unknown",
+                            prior_posture_preserved=False,
+                        ),
+                        fail_closed=fail_closed,
+                        archive_sha256=archive_sha256,
+                        helper_sha256=helper_sha256,
+                        installer_module=installer_module,
+                    )
+                ) from exc
+            release_changing = installer_transition["outcome"] in {"installed", "replaced"}
+            recovery_mode = "explicit" if release_changing else "posture_recovery"
+            recovery_executed = False
+            try:
+                if release_changing:
                     try:
                         recovery = _execute_explicit_rollback(
                             installer_module=installer_module,
@@ -1256,6 +1621,10 @@ def execute_request(
                             source_binding=source_binding,
                             runner=runner,
                         )
+                    except ExplicitRollbackFailure as rollback_exc:
+                        recovery_executed = rollback_exc.executed
+                        raise
+                    recovery_executed = True
                     rollback_result = _rollback_result(
                         manifest=manifest,
                         release=release,
@@ -1263,6 +1632,26 @@ def execute_request(
                         archive_sha256=archive_sha256,
                         helper_sha256=helper_sha256,
                     )
+                    rollback_summary = _rollback_summary(
+                        status="pass",
+                        mode="explicit",
+                        executed=True,
+                        restored_posture=str(rollback_result["service"]["restored_posture"]),
+                        prior_posture_preserved=True,
+                    )
+                else:
+                    try:
+                        rollback_summary = _restore_transition_posture(
+                            installer_module=installer_module,
+                            release=release,
+                            transition=installer_transition,
+                            paths=paths,
+                            runner=runner,
+                        )
+                    except PostureRecoveryFailure as posture_exc:
+                        recovery_executed = posture_exc.executed
+                        raise
+                    recovery_executed = bool(rollback_summary["executed"])
             except Exception:
                 fail_closed = _fail_closed_inert_service(
                     installer_module=installer_module,
@@ -1276,11 +1665,13 @@ def execute_request(
                         release=release,
                         failure_stage=bounded_stage,
                         failure_reason=bounded_reason,
-                        rollback={
+                        rollback=_rollback_summary(
-                            "status": "fail",
+                            status="fail",
-                            "mode": "explicit",
+                            mode=recovery_mode,
-                            "restored_posture": "unknown",
+                            executed=recovery_executed,
-                        },
+                            restored_posture="unknown",
+                            prior_posture_preserved=False,
+                        ),
                         fail_closed=fail_closed,
                         archive_sha256=archive_sha256,
                         helper_sha256=helper_sha256,
@@ -1293,28 +1684,20 @@ def execute_request(
                         manifest=manifest,
                         release=release,
                         failure=exc,
-                        rollback={
+                        rollback=rollback_summary,
-                            "status": "pass",
-                            "mode": "explicit",
-                            "restored_posture": rollback_result["service"]["restored_posture"],
-                        },
                         archive_sha256=archive_sha256,
                         helper_sha256=helper_sha256,
                         installer_module=installer_module,
                     )
                 ) from exc
-            restored_posture = str(rollback_result["service"]["restored_posture"])
+            restored_posture = str(rollback_summary["restored_posture"])
             raise RemoteOperationFailure(
                 _operational_failure_result(
                     manifest=manifest,
                     release=release,
                     failure_stage=bounded_stage,
                     failure_reason=bounded_reason,
-                    rollback={
+                    rollback=rollback_summary,
-                        "status": "pass",
-                        "mode": "explicit",
-                        "restored_posture": restored_posture,
-                    },
                     fail_closed=_rollback_proven_service(restored_posture),
                     archive_sha256=archive_sha256,
                     helper_sha256=helper_sha256,
@@ -1323,6 +1706,7 @@ def execute_request(
             ) from exc
         service = {
             "name": SERVICE,
+            "installer_transition": installer_transition,
             "before_restart": before,
             "after_restart": after,
         }
@@ -1346,7 +1730,13 @@ def execute_request(
                     manifest=manifest,
                     release=release,
                     failure=exc,
-                    rollback={"status": "not_run", "mode": "not_applicable", "restored_posture": "unchanged"},
+                    rollback=_rollback_summary(
+                        status="not_run",
+                        mode="not_applicable",
+                        executed=False,
+                        restored_posture="unchanged",
+                        prior_posture_preserved=True,
+                    ),
                     archive_sha256=archive_sha256,
                     helper_sha256=helper_sha256,
                     installer_module=installer_module,

ops/install_gcp_leoclean_nosend_release.py

@@ -30,7 +30,9 @@ if str(CONTRACT_ROOT) not in sys.path:
 from ops import gcp_leoclean_nosend_package as package  # noqa: E402
 from ops import leoclean_nosend_service_control as service_control  # noqa: E402
 
-INSTALL_RECEIPT_SCHEMA = "livingip.leocleanNoSendInstallReceipt.v2"
+LEGACY_INSTALL_RECEIPT_SCHEMA = "livingip.leocleanNoSendInstallReceipt.v2"
+INSTALL_RECEIPT_SCHEMA = "livingip.leocleanNoSendInstallReceipt.v3"
+INSTALL_EXECUTION_RESULT_SCHEMA = "livingip.leocleanNoSendInstallExecution.v1"
 ROLLBACK_STATE_SCHEMA = "livingip.leocleanNoSendRollbackState.v1"
 ROLLBACK_RESULT_SCHEMA = "livingip.leocleanNoSendRollbackResult.v1"
 SERVICE = package.SERVICE
@@ -171,12 +173,19 @@ class VerificationError(InstallError):
         failure_stage: str,
         failure_reason: str,
         message: str = "release verification failed",
+        *,
+        rollback_executed: bool | None = None,
+        restored_posture: str | None = None,
+        prior_posture_preserved: bool | None = None,
     ) -> None:
         if (failure_stage, failure_reason) not in VERIFICATION_FAILURES:
             raise ValueError("verification failure code was not allowlisted")
         super().__init__(message)
         self.failure_stage = failure_stage
         self.failure_reason = failure_reason
+        self.rollback_executed = rollback_executed
+        self.restored_posture = restored_posture
+        self.prior_posture_preserved = prior_posture_preserved
 
 
 class AtomicWriteError(InstallError):
@@ -284,6 +293,93 @@ class ServicePosture:
     state: dict[str, Any]
 
 
+def _install_transition(
+    outcome: str,
+    *,
+    prior_posture: str,
+    prior_n_restarts: int,
+    service_action: str,
+    managed_files_replaced: bool,
+) -> dict[str, Any]:
+    _require(
+        outcome in {"verified_noop", "controlled_start", "installed", "replaced"}
+        and prior_posture in {"absent", "inactive", "active"}
+        and isinstance(prior_n_restarts, int)
+        and not isinstance(prior_n_restarts, bool)
+        and prior_n_restarts >= 0
+        and service_action in {"none", "start", "restart"}
+        and (
+            (outcome == "verified_noop" and prior_posture == "active" and service_action == "none")
+            or (outcome == "controlled_start" and prior_posture == "inactive" and service_action == "start")
+            or (outcome == "installed" and prior_posture == "absent" and service_action == "restart")
+            or (outcome == "replaced" and prior_posture in {"inactive", "active"} and service_action == "restart")
+        )
+        and managed_files_replaced == (outcome in {"installed", "replaced"}),
+        "installer transition was invalid",
+    )
+    return {
+        "outcome": outcome,
+        "prior_posture": prior_posture,
+        "prior_n_restarts": prior_n_restarts,
+        "final_posture": "active",
+        "service_action": service_action,
+        "managed_files_replaced": managed_files_replaced,
+    }
+
+
+def _install_execution_result(
+    *,
+    outcome: str,
+    release: Mapping[str, Any],
+    receipt: Mapping[str, Any],
+    receipt_raw: bytes,
+    transition: Mapping[str, Any],
+    verification: Mapping[str, Any],
+) -> dict[str, Any]:
+    _require(
+        outcome in {"verified_noop", "controlled_start", "installed", "replaced"}
+        and receipt.get("status") == "pass"
+        and receipt.get("release_sha256") == release["release_sha256"]
+        and (
+            (
+                outcome in {"verified_noop", "controlled_start"}
+                and receipt.get("schema") in {LEGACY_INSTALL_RECEIPT_SCHEMA, INSTALL_RECEIPT_SCHEMA}
+            )
+            or (
+                outcome in {"installed", "replaced"}
+                and receipt.get("schema") == INSTALL_RECEIPT_SCHEMA
+                and receipt.get("outcome") == outcome
+            )
+        ),
+        "exact-release execution result was invalid",
+    )
+    return {
+        "schema": INSTALL_EXECUTION_RESULT_SCHEMA,
+        "status": "pass",
+        "outcome": outcome,
+        "release_sha256": release["release_sha256"],
+        "persisted_receipt": {
+            "schema": receipt["schema"],
+            "sha256": _sha256_bytes(receipt_raw),
+        },
+        "transition": dict(transition),
+        "verification": dict(verification),
+    }
+
+
+def _attach_recovery(
+    failure: Exception,
+    *,
+    rollback_executed: bool,
+    restored_posture: str,
+    prior_posture_preserved: bool,
+) -> None:
+    if isinstance(failure, VerificationError):
+        failure.rollback_executed = rollback_executed
+        failure.restored_posture = restored_posture
+        failure.prior_posture_preserved = prior_posture_preserved
+
+
 def _sha256_bytes(value: bytes) -> str:
     return hashlib.sha256(value).hexdigest()
 
@@ -1558,17 +1654,228 @@ def _read_rollback_state(paths: InstallPaths, release_sha256: str) -> tuple[dict
     return value, raw
 
 
+def _validate_current_install_receipt_contract(
+    receipt: Mapping[str, Any],
+    *,
+    release: Mapping[str, Any],
+    rollback_state: Mapping[str, Any],
+) -> None:
+    _require(
+        set(receipt)
+        == {
+            "schema",
+            "status",
+            "outcome",
+            "transition",
+            "installer_source",
+            "release_sha256",
+            "image",
+            "installed_files",
+            "service_before",
+            "verification",
+            "rollback",
+            "authority",
+        },
+        "installed receipt fields were not exact",
+    )
+    rollback = receipt.get("rollback")
+    _require(
+        isinstance(rollback, dict)
+        and set(rollback)
+        == {
+            "previous_release_present",
+            "previous_service_active",
+            "previous_service_posture",
+            "automatic_on_failure",
+            "state_path",
+            "state_sha256",
+        }
+        and isinstance(rollback.get("previous_release_present"), bool)
+        and isinstance(rollback.get("previous_service_active"), bool)
+        and isinstance(rollback.get("automatic_on_failure"), bool)
+        and rollback["automatic_on_failure"] is True,
+        "installed receipt rollback fields were invalid",
+    )
+    previous_files = rollback_state["previous_files"]
+    previous_service = rollback_state["previous_service"]
+    previous_release_present = all(
+        _snapshot_from_json(previous_files[role], label=f"rollback {role} snapshot").existed
+        for role in sorted(previous_files)
+    )
+    prior_posture = previous_service["posture"]
+    prior_n_restarts = previous_service["n_restarts"]
+    outcome = receipt.get("outcome")
+    _require(
+        (
+            outcome == "installed"
+            and prior_posture == "absent"
+            and previous_release_present is False
+        )
+        or (
+            outcome == "replaced"
+            and prior_posture in {"active", "inactive"}
+            and previous_release_present is True
+        ),
+        "installed receipt outcome differed from rollback state",
+    )
+    _require(
+        rollback["previous_release_present"] is previous_release_present
+        and rollback["previous_service_active"] is (prior_posture == "active")
+        and rollback["previous_service_posture"] == prior_posture,
+        "installed receipt prior posture differed from rollback state",
+    )
+    service_before = receipt.get("service_before")
+    _require(
+        isinstance(service_before, dict)
+        and set(service_before)
+        == {
+            "load_state",
+            "active_state",
+            "sub_state",
+            "main_pid",
+            "n_restarts",
+        }
+        and isinstance(service_before.get("main_pid"), int)
+        and not isinstance(service_before.get("main_pid"), bool)
+        and service_before["main_pid"] >= 0
+        and isinstance(service_before.get("n_restarts"), int)
+        and not isinstance(service_before.get("n_restarts"), bool)
+        and service_before["n_restarts"] >= 0
+        and service_before["active_state"] == previous_service["active_state"]
+        and service_before["sub_state"] == previous_service["sub_state"]
+        and service_before["n_restarts"] == prior_n_restarts,
+        "installed receipt service posture differed from rollback state",
+    )
+    _require(
+        (
+            prior_posture == "absent"
+            and service_before["load_state"] == "not-found"
+            and service_before["main_pid"] == 0
+        )
+        or (
+            prior_posture == "inactive"
+            and service_before["load_state"] == "loaded"
+            and service_before["main_pid"] == 0
+        )
+        or (
+            prior_posture == "active"
+            and service_before["load_state"] == "loaded"
+            and service_before["main_pid"] > 1
+        ),
+        "installed receipt load or process posture differed from rollback state",
+    )
+    _require(
+        receipt.get("transition")
+        == _install_transition(
+            str(outcome),
+            prior_posture=prior_posture,
+            prior_n_restarts=prior_n_restarts,
+            service_action="restart",
+            managed_files_replaced=True,
+        ),
+        "installed receipt transition was invalid",
+    )
+    verification = receipt.get("verification")
+    _require(
+        isinstance(verification, dict)
+        and set(verification)
+        == {
+            "active_state",
+            "sub_state",
+            "main_pid",
+            "n_restarts",
+            "invocation_id",
+            "supervising_process",
+            "container_id",
+            "container_pid",
+            "health",
+            "validated_process_environment_fields",
+            "dropins",
+            "environment_files",
+        }
+        and verification.get("active_state") == "active"
+        and verification.get("sub_state") == "running"
+        and isinstance(verification.get("main_pid"), int)
+        and not isinstance(verification.get("main_pid"), bool)
+        and verification["main_pid"] > 1
+        and isinstance(verification.get("n_restarts"), int)
+        and not isinstance(verification.get("n_restarts"), bool)
+        and verification["n_restarts"] >= 0
+        and isinstance(verification.get("invocation_id"), str)
+        and HEX_32.fullmatch(verification["invocation_id"]) is not None
+        and isinstance(verification.get("container_id"), str)
+        and HEX_64.fullmatch(verification["container_id"]) is not None
+        and isinstance(verification.get("container_pid"), int)
+        and not isinstance(verification.get("container_pid"), bool)
+        and verification["container_pid"] > 1
+        and verification.get("health") == "healthy"
+        and verification.get("dropins") == "absent"
+        and verification.get("environment_files") == "absent",
+        "installed receipt verification was invalid",
+    )
+    supervising_process = verification["supervising_process"]
+    environment_fields = verification["validated_process_environment_fields"]
+    expected_start = shlex.split(_rendered_service_fields(release)["ExecStart"][0])
+    expected_executable = expected_start[0]
+    expected_cmdline = b"\0".join(item.encode("utf-8") for item in expected_start) + b"\0"
+    _require(
+        isinstance(supervising_process, dict)
+        and set(supervising_process) == {"executable", "cmdline_sha256", "cgroup"}
+        and isinstance(supervising_process.get("executable"), str)
+        and supervising_process["executable"] in {expected_executable, str(Path(expected_executable).resolve())}
+        and isinstance(supervising_process.get("cmdline_sha256"), str)
+        and supervising_process["cmdline_sha256"] == _sha256_bytes(expected_cmdline)
+        and supervising_process.get("cgroup") == f"/system.slice/{SERVICE}"
+        and isinstance(environment_fields, list)
+        and all(isinstance(field, str) and field for field in environment_fields)
+        and environment_fields == sorted(set(environment_fields))
+        and "PATH" in environment_fields,
+        "installed receipt verification detail was invalid",
+    )
+    _require(
+        receipt.get("authority")
+        == {
+            "image_pulled": False,
+            "cloud_or_database_authority_changed": False,
+            "transport_authority_changed": False,
+        },
+        "installed receipt authority was invalid",
+    )
+    installed_files = receipt.get("installed_files")
+    _require(
+        isinstance(installed_files, dict)
+        and set(installed_files)
+        == {
+            "unit_sha256",
+            "helper_sha256",
+            "release_descriptor_sha256",
+            "rollback_state_sha256",
+        },
+        "installed receipt file fields were not exact",
+    )
+
+
 def _validate_install_receipt(
     receipt: Mapping[str, Any],
     *,
     release: Mapping[str, Any],
     paths: InstallPaths,
+    rollback_state: Mapping[str, Any],
     rollback_state_raw: bytes,
 ) -> None:
     installed_files = receipt.get("installed_files")
     rollback = receipt.get("rollback")
+    schema = receipt.get("schema")
+    legacy_result_valid = schema == LEGACY_INSTALL_RECEIPT_SCHEMA and receipt.get("result") == "installed"
+    if schema == INSTALL_RECEIPT_SCHEMA:
+        _validate_current_install_receipt_contract(
+            receipt,
+            release=release,
+            rollback_state=rollback_state,
+        )
+    current_outcome_valid = schema == INSTALL_RECEIPT_SCHEMA
     _require(
-        receipt.get("schema") == INSTALL_RECEIPT_SCHEMA
+        (legacy_result_valid or current_outcome_valid)
         and receipt.get("status") == "pass"
         and receipt.get("release_sha256") == release["release_sha256"]
         and isinstance(installed_files, dict)
@@ -1644,7 +1951,6 @@ def _validate_managed_prior_state(paths: InstallPaths, helper_source: bytes) ->
     _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")
     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",
@@ -1655,6 +1961,7 @@ def _validate_managed_prior_state(paths: InstallPaths, helper_source: bytes) ->
         receipt,
         release=release,
         paths=paths,
+        rollback_state=rollback_state,
         rollback_state_raw=rollback_state_raw,
     )
     _require(bool(helper_source), "service-control helper source was empty")
@@ -1665,7 +1972,7 @@ def _systemctl(action: str, runner: Runner) -> None:
     if action == "daemon-reload":
         argv = [SYSTEMCTL, "daemon-reload"]
     else:
-        _require(action in {"restart", "stop", "reset-failed"}, "systemd action was not allowed")
+        _require(action in {"start", "restart", "stop", "reset-failed"}, "systemd action was not allowed")
         argv = [SYSTEMCTL, action, SERVICE]
     _run_checked(argv, runner, label=f"systemd {action}")
 
@@ -1817,11 +2124,26 @@ def _install_release_transaction(
         _ensure_directory(paths.state_dir, 0o700, created)
         _ensure_directory(paths.docker_config, 0o700, created)
         _require(not any(paths.docker_config.iterdir()), "service Docker config directory was not empty")
+        try:
             _validate_local_image(release, runner, paths)
+        except VerificationError:
+            raise
+        except InstallError as exc:
+            raise VerificationError(
+                "image_identity",
+                "image_contract_mismatch",
+                str(exc),
+            ) from exc
     except Exception as exc:
         cleanup_complete = _cleanup_created_directories(created)
         if not cleanup_complete:
             raise InstallError("preflight failed and cleanup was incomplete") from exc
+        _attach_recovery(
+            exc,
+            rollback_executed=False,
+            restored_posture=prior_posture.kind,
+            prior_posture_preserved=True,
+        )
         raise
 
     snapshots = {
@@ -1841,19 +2163,133 @@ def _install_release_transaction(
         and snapshots[paths.helper].content == helper_content
     )
     if exact_already_installed:
+        exact_snapshots = _current_managed_snapshots(paths)
+        exact_release, existing_receipt, _exact_rollback = _validate_managed_snapshots(
+            exact_snapshots,
+            paths=paths,
+        )
+        _require(exact_release == release, "exact managed release changed before convergence")
+        receipt_raw = exact_snapshots["receipt"].content
+        if prior_posture.kind == "active":
+            try:
                 verified = verify_running_release(release, runner=runner, paths=paths, sleeper=sleeper)
-        existing_receipt = package.load_json(paths.receipt, "install receipt")
+                _verification_require(
-        return {**existing_receipt, "result": "already_installed", "verification": verified}
+                    verified["main_pid"] == before["MainPID"]
+                    and verified["n_restarts"] == before["NRestarts"]
+                    and verified["invocation_id"] == before["InvocationID"],
+                    "startup_identity",
+                    "service_identity_changed",
+                    "active exact release changed during verification",
+                )
+                _assert_snapshot_identities(exact_snapshots, paths=paths)
+            except Exception as exc:
+                _attach_recovery(
+                    exc,
+                    rollback_executed=False,
+                    restored_posture="active",
+                    prior_posture_preserved=True,
+                )
+                raise
+            transition = _install_transition(
+                "verified_noop",
+                prior_posture="active",
+                prior_n_restarts=before["NRestarts"],
+                service_action="none",
+                managed_files_replaced=False,
+            )
+            return _install_execution_result(
+                outcome="verified_noop",
+                release=release,
+                receipt=existing_receipt,
+                receipt_raw=receipt_raw,
+                transition=transition,
+                verification=verified,
+            )
+
+        _require(prior_posture.kind == "inactive", "exact managed release posture was invalid")
+        start_attempted = False
+        try:
+            _require_candidate_absent(runner=runner, paths=paths, sleeper=sleeper)
+            start_attempted = True
+            try:
+                _systemctl("start", runner)
+            except InstallError as exc:
+                raise VerificationError(
+                    "service_activation",
+                    "service_not_active",
+                    "controlled service start failed",
+                ) from exc
+            verified = verify_running_release(release, runner=runner, paths=paths, sleeper=sleeper)
+            _verification_require(
+                verified["n_restarts"] == before["NRestarts"],
+                "startup_identity",
+                "service_identity_changed",
+                "controlled service start changed the restart counter",
+            )
+            _assert_snapshot_identities(exact_snapshots, paths=paths)
+            transition = _install_transition(
+                "controlled_start",
+                prior_posture="inactive",
+                prior_n_restarts=before["NRestarts"],
+                service_action="start",
+                managed_files_replaced=False,
+            )
+            return _install_execution_result(
+                outcome="controlled_start",
+                release=release,
+                receipt=existing_receipt,
+                receipt_raw=receipt_raw,
+                transition=transition,
+                verification=verified,
+            )
+        except Exception as exc:
+            try:
+                if start_attempted:
+                    _systemctl("stop", runner)
+                _require_candidate_absent(runner=runner, paths=paths, sleeper=sleeper)
+                restored_posture = _stable_service_posture(
+                    release,
+                    runner=runner,
+                    paths=paths,
+                    sleeper=sleeper,
+                )
+                _require(
+                    restored_posture.kind == "inactive"
+                    and restored_posture.state["NRestarts"] == before["NRestarts"],
+                    "controlled start rollback did not restore the prior inactive posture",
+                )
+            except InstallError as recovery_exc:
+                raise InstallError("controlled service start failed and cleanup was incomplete") from recovery_exc
+            _attach_recovery(
+                exc,
+                rollback_executed=start_attempted,
+                restored_posture="inactive",
+                prior_posture_preserved=True,
+            )
+            if isinstance(exc, InstallError):
+                raise
+            raise InstallError("controlled service start failed and prior posture was restored") from exc
 
     if prior_active:
         _require(prior_release is not None, "active managed service had no prior release")
+        try:
             prior_verification = verify_running_release(prior_release, runner=runner, paths=paths, sleeper=sleeper)
-        _require(
+            _verification_require(
                 prior_verification["main_pid"] == before["MainPID"]
                 and prior_verification["n_restarts"] == before["NRestarts"]
                 and prior_verification["invocation_id"] == before["InvocationID"],
+                "startup_identity",
+                "service_identity_changed",
                 "active prior service changed before the trusted stop",
             )
+        except Exception as exc:
+            _attach_recovery(
+                exc,
+                rollback_executed=False,
+                restored_posture="active",
+                prior_posture_preserved=True,
+            )
+            raise
 
     candidate_may_be_active = False
     try:
@@ -1893,10 +2329,19 @@ def _install_release_transaction(
         )
         attempted_paths.add(paths.rollback_state)
         _publish_managed_file(paths.rollback_state, rollback_state_bytes, 0o600, installed_identities)
+        outcome = "installed" if prior_release is None else "replaced"
+        transition = _install_transition(
+            outcome,
+            prior_posture=prior_posture.kind,
+            prior_n_restarts=before["NRestarts"],
+            service_action="restart",
+            managed_files_replaced=True,
+        )
         receipt = {
             "schema": INSTALL_RECEIPT_SCHEMA,
             "status": "pass",
-            "result": "installed",
+            "outcome": outcome,
+            "transition": transition,
             "installer_source": dict(source_binding),
             "release_sha256": release["release_sha256"],
             "image": {
@@ -1936,7 +2381,14 @@ def _install_release_transaction(
         receipt_bytes = _canonical_json(receipt)
         attempted_paths.add(paths.receipt)
         _publish_managed_file(paths.receipt, receipt_bytes, 0o600, installed_identities)
-        return receipt
+        return _install_execution_result(
+            outcome=outcome,
+            release=release,
+            receipt=receipt,
+            receipt_raw=receipt_bytes,
+            transition=transition,
+            verification=verified,
+        )
     except Exception as exc:
         rollback_errors: list[str] = []
         if candidate_may_be_active:
@@ -1992,6 +2444,12 @@ def _install_release_transaction(
             rollback_errors.append("directory_cleanup")
         if rollback_errors:
             raise InstallError("installation failed and automatic rollback was incomplete") from exc
+        _attach_recovery(
+            exc,
+            rollback_executed=True,
+            restored_posture=prior_posture.kind,
+            prior_posture_preserved=True,
+        )
         if isinstance(exc, InstallError):
             raise
         raise InstallError("installation failed and was rolled back") from exc
@@ -2032,8 +2490,17 @@ def _validate_managed_snapshots(
     receipt = _json_from_snapshot(snapshots["receipt"], label="snapshot install receipt")
     installed_files = receipt.get("installed_files")
     rollback = receipt.get("rollback")
+    schema = receipt.get("schema")
+    legacy_result_valid = schema == LEGACY_INSTALL_RECEIPT_SCHEMA and receipt.get("result") == "installed"
+    if schema == INSTALL_RECEIPT_SCHEMA:
+        _validate_current_install_receipt_contract(
+            receipt,
+            release=release,
+            rollback_state=rollback_state,
+        )
+    current_outcome_valid = schema == INSTALL_RECEIPT_SCHEMA
     _require(
-        receipt.get("schema") == INSTALL_RECEIPT_SCHEMA
+        (legacy_result_valid or current_outcome_valid)
         and receipt.get("status") == "pass"
         and receipt.get("release_sha256") == release_sha256
         and isinstance(receipt.get("installer_source"), dict)

ops/run_gcp_leoclean_nosend_iap.py

@@ -31,9 +31,9 @@ from ops import install_gcp_leoclean_nosend_release as installer  # noqa: E402
 
 SCHEMA = "livingip.leocleanNoSendIapRequest.v1"
 PLAN_SCHEMA = "livingip.leocleanNoSendIapPlan.v1"
-REMOTE_RESULT_SCHEMA = "livingip.leocleanNoSendIapRemoteResult.v1"
+REMOTE_RESULT_SCHEMA = "livingip.leocleanNoSendIapRemoteResult.v2"
-REMOTE_FAILURE_SCHEMA = "livingip.leocleanNoSendIapFailure.v1"
+REMOTE_FAILURE_SCHEMA = "livingip.leocleanNoSendIapFailure.v2"
-REMOTE_OPERATION_FAILURE_SCHEMA = "livingip.leocleanNoSendIapOperationalFailure.v1"
+REMOTE_OPERATION_FAILURE_SCHEMA = "livingip.leocleanNoSendIapOperationalFailure.v2"
 RESULT_RECEIPT_SCHEMA = "livingip.leocleanNoSendIapResultReceipt.v1"
 FAILURE_RECEIPT_SCHEMA = "livingip.leocleanNoSendIapFailureReceipt.v1"
 LOCAL_FAILURE_RECEIPT_SCHEMA = "livingip.leocleanNoSendIapLocalFailureReceipt.v1"
@@ -944,6 +944,113 @@ def _validate_database_summary(value: Any, request_id: str, expected_name: str)
     )
 
 
+def _validate_installer_transition(value: Any) -> None:
+    _require(
+        isinstance(value, dict)
+        and set(value)
+        == {
+            "outcome",
+            "prior_posture",
+            "prior_n_restarts",
+            "final_posture",
+            "service_action",
+            "managed_files_replaced",
+        }
+        and isinstance(value.get("prior_n_restarts"), int)
+        and not isinstance(value.get("prior_n_restarts"), bool)
+        and value["prior_n_restarts"] >= 0
+        and value.get("final_posture") == "active"
+        and (
+            (
+                value.get("outcome"),
+                value.get("prior_posture"),
+                value.get("service_action"),
+                value.get("managed_files_replaced"),
+            )
+            in {
+                ("verified_noop", "active", "none", False),
+                ("controlled_start", "inactive", "start", False),
+                ("installed", "absent", "restart", True),
+            }
+            or (
+                value.get("outcome") == "replaced"
+                and value.get("prior_posture") in {"active", "inactive"}
+                and value.get("service_action") == "restart"
+                and value.get("managed_files_replaced") is True
+            )
+        ),
+        "installer transition was invalid",
+    )
+
+
+def _validate_rollback_summary(value: Any) -> str:
+    _require(
+        isinstance(value, dict)
+        and set(value)
+        == {
+            "status",
+            "mode",
+            "executed",
+            "restored_posture",
+            "prior_posture_preserved",
+        },
+        "remote rollback summary was invalid",
+    )
+    installer_automatic = (
+        value.get("status") == "pass"
+        and value.get("mode") == "installer_automatic"
+        and value.get("executed") is True
+        and value.get("restored_posture") in {"active", "inactive", "absent"}
+        and value.get("prior_posture_preserved") is True
+    )
+    explicit = (
+        value.get("status") == "pass"
+        and value.get("mode") == "explicit"
+        and value.get("executed") is True
+        and value.get("restored_posture") in {"active", "inactive", "absent"}
+        and value.get("prior_posture_preserved") is True
+    )
+    posture_recovery = (
+        value.get("status") == "pass"
+        and value.get("mode") == "posture_recovery"
+        and isinstance(value.get("executed"), bool)
+        and value.get("restored_posture") in {"active", "inactive"}
+        and value.get("prior_posture_preserved") is True
+    )
+    not_run = value == {
+        "status": "not_run",
+        "mode": "not_applicable",
+        "executed": False,
+        "restored_posture": "unchanged",
+        "prior_posture_preserved": True,
+    }
+    explicit_failed = (
+        value.get("status") == "fail"
+        and value.get("mode") == "explicit"
+        and isinstance(value.get("executed"), bool)
+        and value.get("restored_posture") == "unknown"
+        and value.get("prior_posture_preserved") is False
+    )
+    posture_recovery_failed = (
+        value.get("status") == "fail"
+        and value.get("mode") == "posture_recovery"
+        and isinstance(value.get("executed"), bool)
+        and value.get("restored_posture") == "unknown"
+        and value.get("prior_posture_preserved") is False
+    )
+    kinds = {
+        "installer_automatic": installer_automatic,
+        "explicit": explicit,
+        "posture_recovery": posture_recovery,
+        "not_run": not_run,
+        "explicit_failed": explicit_failed,
+        "posture_recovery_failed": posture_recovery_failed,
+    }
+    matches = [kind for kind, matched in kinds.items() if matched]
+    _require(len(matches) == 1, "remote rollback summary was invalid")
+    return matches[0]
+
+
 def validate_remote_result(plan: Mapping[str, Any], result: Mapping[str, Any]) -> None:
     _require(
         set(result)
@@ -999,10 +1106,12 @@ def validate_remote_result(plan: Mapping[str, Any], result: Mapping[str, Any]) -
     database = result["database"]
     if operation == "install":
         _require(
-            isinstance(service, dict) and set(service) == {"name", "before_restart", "after_restart"},
+            isinstance(service, dict)
+            and set(service) == {"name", "installer_transition", "before_restart", "after_restart"},
             "install service result was invalid",
         )
         _require(service["name"] == SERVICE, "install service target differed")
+        _validate_installer_transition(service["installer_transition"])
         _validate_verification(service["before_restart"], "pre-restart verification")
         _validate_verification(service["after_restart"], "post-restart verification")
         before = service["before_restart"]
@@ -1079,15 +1188,21 @@ def validate_remote_failure(plan: Mapping[str, Any], result: Mapping[str, Any])
     schema = result.get("schema")
     if schema == REMOTE_FAILURE_SCHEMA:
         _require(set(result) == common_fields, "remote failure fields were not exact")
+        rollback_kind = _validate_rollback_summary(rollback)
         failure_contract = (
             (result.get("failure_stage"), result.get("failure_reason")) in installer.VERIFICATION_FAILURES
-            and rollback
+            and (
-            in (
+                (
-                {"status": "pass", "mode": "installer_automatic", "restored_posture": "prior"},
+                    plan["operation"] == "install"
-                {"status": "pass", "mode": "explicit", "restored_posture": "active"},
+                    and rollback_kind
-                {"status": "pass", "mode": "explicit", "restored_posture": "inactive"},
+                    in {
-                {"status": "pass", "mode": "explicit", "restored_posture": "absent"},
+                        "installer_automatic",
-                {"status": "not_run", "mode": "not_applicable", "restored_posture": "unchanged"},
+                        "explicit",
+                        "posture_recovery",
+                        "not_run",
+                    }
+                )
+                or (plan["operation"] == "verify" and rollback_kind == "not_run")
             )
         )
     else:
@@ -1096,19 +1211,11 @@ def validate_remote_failure(plan: Mapping[str, Any], result: Mapping[str, Any])
             "remote failure fields were not exact",
         )
         failure_pair = (result.get("failure_stage"), result.get("failure_reason"))
-        rollback_passed = (
+        rollback_kind = _validate_rollback_summary(rollback)
-            isinstance(rollback, dict)
+        rollback_passed = rollback_kind in {"explicit", "posture_recovery"}
-            and set(rollback) == {"status", "mode", "restored_posture"}
+        rollback_failed = rollback_kind in {"explicit_failed", "posture_recovery_failed"}
-            and rollback.get("status") == "pass"
-            and rollback.get("mode") == "explicit"
-            and rollback.get("restored_posture") in {"active", "inactive", "absent"}
-        )
-        rollback_failed = rollback == {
-            "status": "fail",
-            "mode": "explicit",
-            "restored_posture": "unknown",
-        }
         fail_closed = result.get("fail_closed")
+        _require(isinstance(fail_closed, dict), "remote fail-closed summary was invalid")
         if rollback_passed:
             fail_closed_contract = fail_closed == {
                 "status": "not_required",
@@ -1118,6 +1225,7 @@ def validate_remote_failure(plan: Mapping[str, Any], result: Mapping[str, Any])
                 "main_pid": None,
                 "n_restarts": None,
                 "candidate_container": "not_observed",
+                "service_action_executed": False,
             }
         else:
             fail_closed_contract = (
@@ -1130,6 +1238,7 @@ def validate_remote_failure(plan: Mapping[str, Any], result: Mapping[str, Any])
                     "main_pid": None,
                     "n_restarts": None,
                     "candidate_container": "unknown",
+                    "service_action_executed": fail_closed.get("service_action_executed"),
                 }
                 or (
                     isinstance(fail_closed, dict)
@@ -1142,6 +1251,7 @@ def validate_remote_failure(plan: Mapping[str, Any], result: Mapping[str, Any])
                         "main_pid",
                         "n_restarts",
                         "candidate_container",
+                        "service_action_executed",
                     }
                     and fail_closed.get("status") == "pass"
                     and fail_closed.get("service_posture") == "inactive"
@@ -1152,8 +1262,14 @@ def validate_remote_failure(plan: Mapping[str, Any], result: Mapping[str, Any])
                     and not isinstance(fail_closed.get("n_restarts"), bool)
                     and fail_closed["n_restarts"] >= 0
                     and fail_closed.get("candidate_container") == "absent"
+                    and isinstance(fail_closed.get("service_action_executed"), bool)
                 )
             )
+            if isinstance(fail_closed, dict):
+                _require(
+                    isinstance(fail_closed.get("service_action_executed"), bool),
+                    "remote fail-closed action summary was invalid",
+                )
         failure_contract = (
             plan["operation"] == "install"
             and (

tests/test_gcp_leoclean_nosend_iap_runner.py

@@ -26,6 +26,24 @@ def completed(
     return subprocess.CompletedProcess(argv, returncode, stdout, stderr)
 
 
+def install_execution_result(
+    release: dict[str, Any],
+    transition: dict[str, Any],
+) -> dict[str, Any]:
+    return {
+        "schema": local.installer.INSTALL_EXECUTION_RESULT_SCHEMA,
+        "status": "pass",
+        "outcome": transition.get("outcome"),
+        "release_sha256": release["release_sha256"],
+        "persisted_receipt": {
+            "schema": local.installer.INSTALL_RECEIPT_SCHEMA,
+            "sha256": "9" * 64,
+        },
+        "transition": transition,
+        "verification": {},
+    }
+
+
 def private_directory(path: Path) -> Path:
     path.mkdir(mode=0o700)
     path.chmod(0o700)
@@ -148,7 +166,9 @@ def verifier_failure_result(
         "rollback": {
             "status": "pass",
             "mode": "installer_automatic",
-            "restored_posture": "prior",
+            "executed": True,
+            "restored_posture": "absent",
+            "prior_posture_preserved": True,
         },
         "authority": {
             "sending_enabled": False,
@@ -169,7 +189,9 @@ def operational_failure_result(plan: dict[str, Any]) -> dict[str, Any]:
     result["rollback"] = {
         "status": "fail",
         "mode": "explicit",
+        "executed": True,
         "restored_posture": "unknown",
+        "prior_posture_preserved": False,
     }
     result["fail_closed"] = {
         "status": "pass",
@@ -179,6 +201,7 @@ def operational_failure_result(plan: dict[str, Any]) -> dict[str, Any]:
         "main_pid": 0,
         "n_restarts": 0,
         "candidate_container": "absent",
+        "service_action_executed": True,
     }
     return result
 
@@ -221,6 +244,14 @@ def successful_remote_result(plan: dict[str, Any]) -> dict[str, Any]:
         after = verification(103, "5", "6")
         service_result = {
             "name": local.SERVICE,
+            "installer_transition": {
+                "outcome": "installed",
+                "prior_posture": "absent",
+                "prior_n_restarts": 0,
+                "final_posture": "active",
+                "service_action": "restart",
+                "managed_files_replaced": True,
+            },
             "before_restart": before,
             "after_restart": after,
         }
@@ -857,7 +888,62 @@ def test_operational_failure_rejects_extra_rollback_fields(tmp_path: Path) -> No
     failure = operational_failure_result(plan)
     failure["rollback"]["credential"] = "not-allowed"
 
-    with pytest.raises(local.IapRequestError, match="did not match"):
+    with pytest.raises(local.IapRequestError, match="rollback summary was invalid"):
+        local.validate_remote_failure(plan, failure)
+
+
+@pytest.mark.parametrize(
+    "rollback",
+    [
+        {
+            "status": "not_run",
+            "mode": "not_applicable",
+            "executed": False,
+            "restored_posture": "unchanged",
+            "prior_posture_preserved": True,
+        },
+        {
+            "status": "pass",
+            "mode": "installer_automatic",
+            "executed": True,
+            "restored_posture": "inactive",
+            "prior_posture_preserved": True,
+        },
+        {
+            "status": "pass",
+            "mode": "posture_recovery",
+            "executed": False,
+            "restored_posture": "absent",
+            "prior_posture_preserved": True,
+        },
+    ],
+)
+def test_operational_failure_rejects_impossible_rollback_modes(
+    tmp_path: Path,
+    rollback: dict[str, Any],
+) -> None:
+    plan = executable_plan(tmp_path, "install")
+    failure = operational_failure_result(plan)
+    failure["rollback"] = rollback
+
+    with pytest.raises(local.IapRequestError, match=r"remote failure did not match|rollback summary was invalid"):
+        local.validate_remote_failure(plan, failure)
+
+
+def test_verify_failure_rejects_mutating_rollback_summary(tmp_path: Path) -> None:
+    plan = executable_plan(tmp_path, "verify")
+    failure = verifier_failure_result(plan)
+
+    with pytest.raises(local.IapRequestError, match="remote failure did not match"):
+        local.validate_remote_failure(plan, failure)
+
+    failure["rollback"] = {
+        "status": "not_run",
+        "mode": "not_applicable",
+        "executed": False,
+        "restored_posture": "unchanged",
+        "prior_posture_preserved": True,
+    }
     local.validate_remote_failure(plan, failure)
 
 
@@ -1030,10 +1116,17 @@ def test_outer_lock_and_reused_request_id_fail_closed(tmp_path: Path) -> None:
 
 
 def test_rollback_uses_only_explicit_receipt_bound_interface(tmp_path: Path) -> None:
-    observed: dict[str, Any] = {}
+    observed: list[dict[str, Any]] = []
 
     def rollback(release_sha256: str, **kwargs: Any) -> dict[str, Any]:
-        observed.update({"release_sha256": release_sha256, **kwargs})
+        observed.append({"release_sha256": release_sha256, **kwargs})
+        if kwargs["execute_rollback"] is False:
+            return {
+                "schema": "rollback",
+                "status": "pass",
+                "mode": "dry_run",
+                "expected_release_sha256": release_sha256,
+            }
         return {"schema": "rollback", "status": "pass"}
 
     installer = SimpleNamespace(rollback_release=rollback)
@@ -1046,9 +1139,9 @@ def test_rollback_uses_only_explicit_receipt_bound_interface(tmp_path: Path) ->
     )
 
     assert result["status"] == "pass"
-    assert observed["release_sha256"] == "a" * 64
+    assert [call["release_sha256"] for call in observed] == ["a" * 64, "a" * 64]
-    assert observed["execute_rollback"] is True
+    assert [call["execute_rollback"] for call in observed] == [False, True]
-    assert "install_release" not in observed
+    assert all("install_release" not in call for call in observed)
 
 
 def test_rollback_fails_when_explicit_interface_is_absent(tmp_path: Path) -> None:
@@ -1062,6 +1155,26 @@ def test_rollback_fails_when_explicit_interface_is_absent(tmp_path: Path) -> Non
         )
 
 
+def test_explicit_rollback_preflight_failure_reports_no_execution() -> None:
+    calls: list[bool] = []
+
+    def rollback(_release_sha256: str, **kwargs: Any) -> dict[str, Any]:
+        calls.append(kwargs["execute_rollback"])
+        return {"schema": "rollback", "status": "fail", "mode": "dry_run"}
+
+    with pytest.raises(remote.ExplicitRollbackFailure) as caught:
+        remote._execute_explicit_rollback(
+            installer_module=SimpleNamespace(rollback_release=rollback),
+            release_sha256="a" * 64,
+            paths=object(),
+            source_binding={"revision": "b" * 40},
+            runner=lambda argv, **_kwargs: completed(argv),
+        )
+
+    assert caught.value.executed is False
+    assert calls == [False]
+
+
 def test_install_verifier_failure_returns_bounded_automatic_rollback_result(
     monkeypatch: pytest.MonkeyPatch,
     tmp_path: Path,
@@ -1097,10 +1210,14 @@ def test_install_verifier_failure_returns_bounded_automatic_rollback_result(
             return object()
 
     def fail_install(*_args: Any, **_kwargs: Any) -> dict[str, Any]:
-        raise local.installer.VerificationError(
+        failure = local.installer.VerificationError(
             "process_environment",
             "process_environment_mismatch",
         )
+        failure.rollback_executed = True
+        failure.restored_posture = "absent"
+        failure.prior_posture_preserved = True
+        raise failure
 
     installer_module = SimpleNamespace(
         VerificationError=local.installer.VerificationError,
@@ -1131,7 +1248,9 @@ def test_install_verifier_failure_returns_bounded_automatic_rollback_result(
     assert result["rollback"] == {
         "status": "pass",
         "mode": "installer_automatic",
-        "restored_posture": "prior",
+        "executed": True,
+        "restored_posture": "absent",
+        "prior_posture_preserved": True,
     }
     assert result["authority"]["sending_enabled"] is False
 
@@ -1189,17 +1308,16 @@ def test_post_install_proof_failure_executes_explicit_rollback(
         def under(_root: Path) -> object:
             return object()
 
-    installer_module = SimpleNamespace(
+    def rollback_release(release_sha256: str, **kwargs: Any) -> dict[str, Any]:
-        SYSTEMCTL="/usr/bin/systemctl",
+        if kwargs["execute_rollback"] is False:
-        VERIFICATION_FAILURES=local.installer.VERIFICATION_FAILURES,
+            return {
-        InstallPaths=Paths,
+                "schema": "livingip.leocleanNoSendRollbackResult.v1",
-        validate_release_bundle=lambda _bundle: (release, b"unit"),
+                "status": "pass",
-        _validate_source_binding=lambda _binding: None,
+                "mode": "dry_run",
-        install_release=lambda *_args, **_kwargs: {"status": "pass"},
+                "expected_release_sha256": release_sha256,
-        verify_running_release=lambda *_args, **_kwargs: next(verifications),
+            }
-        rollback_release=lambda release_sha256, **_kwargs: (
         rollbacks.append(release_sha256)
-            or {
+        return {
             "schema": "livingip.leocleanNoSendRollbackResult.v1",
             "status": "pass",
             "mode": "executed",
@@ -1208,7 +1326,29 @@ def test_post_install_proof_failure_executes_explicit_rollback(
             "restored_service_posture": "absent",
             "verification": None,
         }
+
+    installer_module = SimpleNamespace(
+        SYSTEMCTL="/usr/bin/systemctl",
+        VERIFICATION_FAILURES=local.installer.VERIFICATION_FAILURES,
+        INSTALL_EXECUTION_RESULT_SCHEMA=local.installer.INSTALL_EXECUTION_RESULT_SCHEMA,
+        LEGACY_INSTALL_RECEIPT_SCHEMA=local.installer.LEGACY_INSTALL_RECEIPT_SCHEMA,
+        INSTALL_RECEIPT_SCHEMA=local.installer.INSTALL_RECEIPT_SCHEMA,
+        InstallPaths=Paths,
+        validate_release_bundle=lambda _bundle: (release, b"unit"),
+        _validate_source_binding=lambda _binding: None,
+        install_release=lambda *_args, **_kwargs: install_execution_result(
+            release,
+            {
+                "outcome": "installed",
+                "prior_posture": "absent",
+                "prior_n_restarts": 0,
+                "final_posture": "active",
+                "service_action": "restart",
+                "managed_files_replaced": True,
+            },
         ),
+        verify_running_release=lambda *_args, **_kwargs: next(verifications),
+        rollback_release=rollback_release,
     )
     package_module = SimpleNamespace(
         render_systemd_unit=lambda _release: "unit",
@@ -1256,7 +1396,9 @@ def test_post_install_proof_failure_executes_explicit_rollback(
     assert result["rollback"] == {
         "status": "pass",
         "mode": "explicit",
+        "executed": True,
         "restored_posture": "absent",
+        "prior_posture_preserved": True,
     }
     assert result["fail_closed"] == {
         "status": "not_required",
@@ -1266,9 +1408,413 @@ def test_post_install_proof_failure_executes_explicit_rollback(
         "main_pid": None,
         "n_restarts": None,
         "candidate_container": "not_observed",
+        "service_action_executed": False,
     }
 
 
+@pytest.mark.parametrize(
+    ("outcome", "prior_posture", "service_action", "expected_actions", "expected_executed"),
+    [
+        ("verified_noop", "active", "none", [], False),
+        ("controlled_start", "inactive", "start", ["stop"], True),
+    ],
+)
+def test_post_install_failure_preserves_exact_release_posture_without_consuming_history(
+    monkeypatch: pytest.MonkeyPatch,
+    tmp_path: Path,
+    outcome: str,
+    prior_posture: str,
+    service_action: str,
+    expected_actions: list[str],
+    expected_executed: bool,
+) -> 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},
+    }
+    verification = {
+        "main_pid": 10,
+        "invocation_id": "2" * 32,
+        "container_id": "3" * 64,
+        "n_restarts": 0,
+    }
+    actions: list[str] = []
+    postures = [
+        SimpleNamespace(
+            kind="active",
+            state={"MainPID": 10, "InvocationID": "2" * 32, "NRestarts": 0},
+        )
+    ]
+    if prior_posture == "active":
+        postures.append(
+            SimpleNamespace(
+                kind="active",
+                state={"MainPID": 10, "InvocationID": "2" * 32, "NRestarts": 0},
+            )
+        )
+    else:
+        postures.append(
+            SimpleNamespace(
+                kind="inactive",
+                state={"MainPID": 0, "InvocationID": "", "NRestarts": 0},
+            )
+        )
+    posture_iter = iter(postures)
+
+    class Paths:
+        @staticmethod
+        def under(_root: Path) -> object:
+            return object()
+
+    installer_module = SimpleNamespace(
+        SYSTEMCTL="/usr/bin/systemctl",
+        VERIFICATION_FAILURES=local.installer.VERIFICATION_FAILURES,
+        INSTALL_EXECUTION_RESULT_SCHEMA=local.installer.INSTALL_EXECUTION_RESULT_SCHEMA,
+        LEGACY_INSTALL_RECEIPT_SCHEMA=local.installer.LEGACY_INSTALL_RECEIPT_SCHEMA,
+        INSTALL_RECEIPT_SCHEMA=local.installer.INSTALL_RECEIPT_SCHEMA,
+        InstallPaths=Paths,
+        validate_release_bundle=lambda _bundle: (release, b"unit"),
+        _validate_source_binding=lambda _binding: None,
+        install_release=lambda *_args, **_kwargs: install_execution_result(
+            release,
+            {
+                "outcome": outcome,
+                "prior_posture": prior_posture,
+                "prior_n_restarts": 0,
+                "final_posture": "active",
+                "service_action": service_action,
+                "managed_files_replaced": False,
+            },
+        ),
+        verify_running_release=lambda *_args, **_kwargs: verification,
+        rollback_release=lambda *_args, **_kwargs: (_ for _ in ()).throw(
+            AssertionError("historical rollback must not be consumed")
+        ),
+        _systemctl=lambda action, _runner: actions.append(action),
+        _stable_service_posture=lambda *_args, **_kwargs: next(posture_iter),
+        _require_candidate_absent=lambda **_kwargs: 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)
+    monkeypatch.setattr(
+        remote,
+        "_verify_database",
+        lambda **_kwargs: (_ for _ in ()).throw(remote.RemoteError("database failed")),
+    )
+
+    with pytest.raises(remote.RemoteOperationFailure) as caught:
+        remote.execute_request(
+            manifest,
+            tmp_path,
+            tmp_path,
+            "6" * 64,
+            "7" * 64,
+            runner=lambda argv, **_kwargs: completed(argv),
+        )
+
+    result = caught.value.result
+    assert actions == expected_actions
+    assert result["rollback"] == {
+        "status": "pass",
+        "mode": "posture_recovery",
+        "executed": expected_executed,
+        "restored_posture": prior_posture,
+        "prior_posture_preserved": True,
+    }
+
+
+def test_invalid_installer_transition_fails_closed_with_structured_receipt(
+    monkeypatch: pytest.MonkeyPatch,
+    tmp_path: Path,
+) -> 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},
+    }
+    actions: list[str] = []
+    historical_rollbacks: list[bool] = []
+    postures = iter(
+        [
+            SimpleNamespace(
+                kind="active",
+                state={"MainPID": 10, "InvocationID": "2" * 32, "NRestarts": 0},
+            ),
+            SimpleNamespace(
+                kind="inactive",
+                state={
+                    "ActiveState": "inactive",
+                    "SubState": "dead",
+                    "MainPID": 0,
+                    "InvocationID": "",
+                    "NRestarts": 0,
+                },
+            ),
+        ]
+    )
+
+    class Paths:
+        @staticmethod
+        def under(_root: Path) -> object:
+            return object()
+
+    installer_module = SimpleNamespace(
+        SYSTEMCTL="/usr/bin/systemctl",
+        VERIFICATION_FAILURES=local.installer.VERIFICATION_FAILURES,
+        INSTALL_EXECUTION_RESULT_SCHEMA=local.installer.INSTALL_EXECUTION_RESULT_SCHEMA,
+        LEGACY_INSTALL_RECEIPT_SCHEMA=local.installer.LEGACY_INSTALL_RECEIPT_SCHEMA,
+        INSTALL_RECEIPT_SCHEMA=local.installer.INSTALL_RECEIPT_SCHEMA,
+        InstallPaths=Paths,
+        validate_release_bundle=lambda _bundle: (release, b"unit"),
+        _validate_source_binding=lambda _binding: None,
+        install_release=lambda *_args, **_kwargs: install_execution_result(
+            release,
+            {"outcome": "controlled_start"},
+        ),
+        verify_running_release=lambda *_args, **_kwargs: {
+            "main_pid": 10,
+            "invocation_id": "2" * 32,
+            "container_id": "3" * 64,
+            "n_restarts": 0,
+        },
+        rollback_release=lambda *_args, **_kwargs: historical_rollbacks.append(True),
+        _systemctl=lambda action, _runner: actions.append(action),
+        _stable_service_posture=lambda *_args, **_kwargs: next(postures),
+        _require_candidate_absent=lambda **_kwargs: 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)
+
+    with pytest.raises(remote.RemoteOperationFailure) as caught:
+        remote.execute_request(
+            manifest,
+            tmp_path,
+            tmp_path,
+            "6" * 64,
+            "7" * 64,
+            runner=lambda argv, **_kwargs: completed(argv),
+        )
+
+    result = caught.value.result
+    assert historical_rollbacks == []
+    assert actions == ["stop"]
+    assert (result["failure_stage"], result["failure_reason"]) == (
+        "install_receipt",
+        "install_receipt_invalid",
+    )
+    assert result["rollback"] == {
+        "status": "fail",
+        "mode": "posture_recovery",
+        "executed": True,
+        "restored_posture": "unknown",
+        "prior_posture_preserved": False,
+    }
+    assert result["fail_closed"]["status"] == "pass"
+    assert result["fail_closed"]["service_action_executed"] is True
+
+
+def test_posture_recovery_refuses_identity_drift_before_stop(
+    monkeypatch: pytest.MonkeyPatch,
+    tmp_path: Path,
+) -> 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},
+    }
+    actions: list[str] = []
+    postures = iter(
+        [
+            SimpleNamespace(kind="active", state={"MainPID": 10, "InvocationID": "2" * 32, "NRestarts": 0}),
+            SimpleNamespace(kind="active", state={"MainPID": 10, "InvocationID": "2" * 32, "NRestarts": 0}),
+        ]
+    )
+    verifications = iter(
+        [
+            {"main_pid": 10, "invocation_id": "2" * 32, "container_id": "3" * 64, "n_restarts": 0},
+            {"main_pid": 99, "invocation_id": "9" * 32, "container_id": "8" * 64, "n_restarts": 0},
+            {"main_pid": 99, "invocation_id": "9" * 32, "container_id": "8" * 64, "n_restarts": 0},
+        ]
+    )
+
+    class Paths:
+        @staticmethod
+        def under(_root: Path) -> object:
+            return object()
+
+    installer_module = SimpleNamespace(
+        SYSTEMCTL="/usr/bin/systemctl",
+        VERIFICATION_FAILURES=local.installer.VERIFICATION_FAILURES,
+        INSTALL_EXECUTION_RESULT_SCHEMA=local.installer.INSTALL_EXECUTION_RESULT_SCHEMA,
+        LEGACY_INSTALL_RECEIPT_SCHEMA=local.installer.LEGACY_INSTALL_RECEIPT_SCHEMA,
+        INSTALL_RECEIPT_SCHEMA=local.installer.INSTALL_RECEIPT_SCHEMA,
+        InstallPaths=Paths,
+        validate_release_bundle=lambda _bundle: (release, b"unit"),
+        _validate_source_binding=lambda _binding: None,
+        install_release=lambda *_args, **_kwargs: install_execution_result(
+            release,
+            {
+                "outcome": "controlled_start",
+                "prior_posture": "inactive",
+                "prior_n_restarts": 0,
+                "final_posture": "active",
+                "service_action": "start",
+                "managed_files_replaced": False,
+            },
+        ),
+        verify_running_release=lambda *_args, **_kwargs: next(verifications),
+        rollback_release=lambda *_args, **_kwargs: (_ for _ in ()).throw(
+            AssertionError("historical rollback must not be consumed")
+        ),
+        _systemctl=lambda action, _runner: actions.append(action),
+        _stable_service_posture=lambda *_args, **_kwargs: next(postures),
+        _require_candidate_absent=lambda **_kwargs: 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)
+    monkeypatch.setattr(
+        remote,
+        "_verify_database",
+        lambda **_kwargs: (_ for _ in ()).throw(remote.RemoteError("database failed")),
+    )
+
+    with pytest.raises(remote.RemoteOperationFailure) as caught:
+        remote.execute_request(
+            manifest,
+            tmp_path,
+            tmp_path,
+            "6" * 64,
+            "7" * 64,
+            runner=lambda argv, **_kwargs: completed(argv),
+        )
+
+    result = caught.value.result
+    assert actions == []
+    assert result["rollback"]["executed"] is False
+    assert result["rollback"]["mode"] == "posture_recovery"
+    assert result["fail_closed"]["status"] == "fail"
+    assert result["fail_closed"]["service_action_executed"] is False
+
+
+def test_posture_recovery_rejects_restart_counter_drift_after_fixed_restart() -> None:
+    actions: list[str] = []
+    postures = iter(
+        [
+            SimpleNamespace(
+                kind="active",
+                state={"MainPID": 10, "InvocationID": "2" * 32, "NRestarts": 1},
+            ),
+            SimpleNamespace(
+                kind="inactive",
+                state={"MainPID": 0, "InvocationID": "", "NRestarts": 1},
+            ),
+        ]
+    )
+    installer_module = SimpleNamespace(
+        _systemctl=lambda action, _runner: actions.append(action),
+        _stable_service_posture=lambda *_args, **_kwargs: next(postures),
+        verify_running_release=lambda *_args, **_kwargs: {
+            "main_pid": 10,
+            "invocation_id": "2" * 32,
+            "container_id": "3" * 64,
+            "n_restarts": 1,
+        },
+        _require_candidate_absent=lambda **_kwargs: None,
+    )
+
+    with pytest.raises(remote.PostureRecoveryFailure) as caught:
+        remote._restore_transition_posture(
+            installer_module=installer_module,
+            release={"release_sha256": "a" * 64},
+            transition={
+                "outcome": "controlled_start",
+                "prior_posture": "inactive",
+                "prior_n_restarts": 0,
+                "final_posture": "active",
+                "service_action": "start",
+                "managed_files_replaced": False,
+            },
+            paths=object(),
+            runner=lambda argv, **_kwargs: completed(argv),
+        )
+
+    assert caught.value.executed is True
+    assert actions == ["stop"]
+
+
 def test_incomplete_rollback_stops_service_and_reports_inert_posture(
     monkeypatch: pytest.MonkeyPatch,
     tmp_path: Path,
@@ -1329,19 +1875,43 @@ def test_incomplete_rollback_stops_service_and_reports_inert_posture(
             ),
         ]
     )
+
+    def fail_rollback(release_sha256: str, **kwargs: Any) -> dict[str, Any]:
+        if kwargs["execute_rollback"] is False:
+            return {
+                "schema": "livingip.leocleanNoSendRollbackResult.v1",
+                "status": "pass",
+                "mode": "dry_run",
+                "expected_release_sha256": release_sha256,
+            }
+        raise RuntimeError("rollback failed")
+
     installer_module = SimpleNamespace(
         SYSTEMCTL="/usr/bin/systemctl",
+        INSTALL_EXECUTION_RESULT_SCHEMA=local.installer.INSTALL_EXECUTION_RESULT_SCHEMA,
+        LEGACY_INSTALL_RECEIPT_SCHEMA=local.installer.LEGACY_INSTALL_RECEIPT_SCHEMA,
+        INSTALL_RECEIPT_SCHEMA=local.installer.INSTALL_RECEIPT_SCHEMA,
         InstallPaths=Paths,
         validate_release_bundle=lambda _bundle: (release, b"unit"),
         _validate_source_binding=lambda _binding: None,
-        install_release=lambda *_args, **_kwargs: {"status": "pass"},
+        install_release=lambda *_args, **_kwargs: install_execution_result(
+            release,
+            {
+                "outcome": "installed",
+                "prior_posture": "absent",
+                "prior_n_restarts": 0,
+                "final_posture": "active",
+                "service_action": "restart",
+                "managed_files_replaced": True,
+            },
+        ),
         verify_running_release=lambda *_args, **_kwargs: {
             "main_pid": 10,
             "invocation_id": "2" * 32,
             "container_id": "3" * 64,
             "n_restarts": 0,
         },
-        rollback_release=lambda *_args, **_kwargs: (_ for _ in ()).throw(RuntimeError("rollback failed")),
+        rollback_release=fail_rollback,
         _systemctl=lambda action, _runner: stops.append(action),
         _stable_service_posture=lambda *_args, **_kwargs: next(postures),
         _require_candidate_absent=lambda **_kwargs: absence_checks.append(True),
@@ -1377,7 +1947,9 @@ def test_incomplete_rollback_stops_service_and_reports_inert_posture(
     assert result["rollback"] == {
         "status": "fail",
         "mode": "explicit",
+        "executed": True,
         "restored_posture": "unknown",
+        "prior_posture_preserved": False,
     }
     assert result["fail_closed"] == {
         "status": "pass",
@@ -1387,6 +1959,7 @@ def test_incomplete_rollback_stops_service_and_reports_inert_posture(
         "main_pid": 0,
         "n_restarts": 0,
         "candidate_container": "absent",
+        "service_action_executed": True,
     }
     assert result["authority"]["sending_enabled"] is False
 
@@ -1418,6 +1991,7 @@ def test_fail_closed_stop_refuses_untrusted_unit_or_dropin() -> None:
         "main_pid": None,
         "n_restarts": None,
         "candidate_container": "unknown",
+        "service_action_executed": False,
     }
 
 

tests/test_gcp_leoclean_nosend_installer.py

@@ -110,6 +110,7 @@ class FakeHost:
         self.current_release: dict[str, object] | None = None
         self.loaded_release: dict[str, object] | None = None
         self.generation = 0
+        self.start_calls = 0
         self.restart_calls = 0
         self.stop_calls = 0
         self.fail_dropin_release: str | None = None
@@ -157,7 +158,7 @@ class FakeHost:
         (directory / "cmdline").write_bytes(b"\0".join(item.encode() for item in command) + b"\0")
         (directory / "cgroup").write_text(f"0::/system.slice/{package.SERVICE}\n", encoding="utf-8")
 
-    def _start(self) -> None:
+    def _start(self, action: str = "restart") -> None:
         release = self.loaded_release
         assert release is not None
         image = release["image"]
@@ -166,6 +167,9 @@ class FakeHost:
         runtime = inspection["runtime_config"]
         assert isinstance(runtime, dict)
         self.generation += 1
+        if action == "start":
+            self.start_calls += 1
+        else:
             self.restart_calls += 1
         self.active = True
         self.main_pid = 1000 + self.generation
@@ -274,6 +278,9 @@ class FakeHost:
             if action == "restart":
                 self._start()
                 return subprocess.CompletedProcess(arguments, 0, stdout=b"", stderr=b"")
+            if action == "start":
+                self._start("start")
+                return subprocess.CompletedProcess(arguments, 0, stdout=b"", stderr=b"")
             if action == "stop":
                 self._stop()
                 return subprocess.CompletedProcess(arguments, 0, stdout=b"", stderr=b"")
@@ -397,7 +404,7 @@ def test_disposable_root_install_is_digest_bound_idempotent_and_exact(tmp_path:
     host = FakeHost(paths)
     host.add_release(release, image)
 
-    receipt = installer.install_release(
+    result = installer.install_release(
         bundle,
         paths=paths,
         source_binding=source_binding(),
@@ -405,10 +412,26 @@ def test_disposable_root_install_is_digest_bound_idempotent_and_exact(tmp_path:
         sleeper=lambda _seconds: None,
         execute_restart=True,
     )
+    receipt = json.loads(paths.receipt.read_text(encoding="utf-8"))
 
-    assert receipt["status"] == "pass"
+    assert result["status"] == "pass"
-    assert receipt["schema"] == "livingip.leocleanNoSendInstallReceipt.v2"
+    assert result["schema"] == installer.INSTALL_EXECUTION_RESULT_SCHEMA
-    assert receipt["release_sha256"] == release["release_sha256"]
+    assert result["outcome"] == "installed"
+    assert result["transition"] == {
+        "outcome": "installed",
+        "prior_posture": "absent",
+        "prior_n_restarts": 0,
+        "final_posture": "active",
+        "service_action": "restart",
+        "managed_files_replaced": True,
+    }
+    assert result["release_sha256"] == release["release_sha256"]
+    assert result["persisted_receipt"] == {
+        "schema": installer.INSTALL_RECEIPT_SCHEMA,
+        "sha256": package.sha256_file(paths.receipt),
+    }
+    assert receipt["schema"] == installer.INSTALL_RECEIPT_SCHEMA
+    assert receipt["outcome"] == "installed"
     assert host.restart_calls == 1
     assert paths.unit.read_text(encoding="utf-8") == package.render_systemd_unit(release)
     assert paths.unit.stat().st_mode & 0o777 == 0o644
@@ -423,6 +446,12 @@ def test_disposable_root_install_is_digest_bound_idempotent_and_exact(tmp_path:
     assert not list(paths.docker_config.iterdir())
     assert not list(paths.unit.parent.glob(f".{paths.unit.name}.*.tmp"))
 
+    managed_before = {
+        role: path.read_bytes()
+        for role, (path, _mode) in installer._managed_files(paths).items()
+    }
+    identity_before = (host.main_pid, host.n_restarts, host.invocation)
+    calls_before = len(host.calls)
     repeated = installer.install_release(
         bundle,
         paths=paths,
@@ -432,8 +461,31 @@ def test_disposable_root_install_is_digest_bound_idempotent_and_exact(tmp_path:
         execute_restart=True,
     )
 
-    assert repeated["result"] == "already_installed"
+    assert repeated["outcome"] == "verified_noop"
+    assert repeated["schema"] == installer.INSTALL_EXECUTION_RESULT_SCHEMA
+    assert repeated["persisted_receipt"] == {
+        "schema": installer.INSTALL_RECEIPT_SCHEMA,
+        "sha256": package.sha256_file(paths.receipt),
+    }
+    assert repeated["transition"] == {
+        "outcome": "verified_noop",
+        "prior_posture": "active",
+        "prior_n_restarts": 0,
+        "final_posture": "active",
+        "service_action": "none",
+        "managed_files_replaced": False,
+    }
     assert host.restart_calls == 1
+    assert host.start_calls == 0
+    assert (host.main_pid, host.n_restarts, host.invocation) == identity_before
+    assert {
+        role: path.read_bytes()
+        for role, (path, _mode) in installer._managed_files(paths).items()
+    } == managed_before
+    assert not any(
+        call[0] == installer.SYSTEMCTL and call[1] in {"start", "restart", "stop", "daemon-reload"}
+        for call in host.calls[calls_before:]
+    )
 
 
 def test_first_install_accepts_production_shaped_systemd_not_found_defaults(tmp_path: Path) -> None:
@@ -465,10 +517,391 @@ def test_first_install_accepts_production_shaped_systemd_not_found_defaults(tmp_
         execute_restart=True,
     )
 
-    assert repeated["result"] == "already_installed"
+    assert repeated["outcome"] == "verified_noop"
     assert host.restart_calls == 1
 
 
+def test_exact_inactive_release_converges_with_one_controlled_start_and_preserves_history(
+    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,
+    )
+    host._stop()
+    managed_before = {
+        role: path.read_bytes()
+        for role, (path, _mode) in installer._managed_files(paths).items()
+    }
+    calls_before = len(host.calls)
+    restart_calls = host.restart_calls
+    stop_calls = host.stop_calls
+
+    result = installer.install_release(
+        bundle,
+        paths=paths,
+        source_binding=source_binding(),
+        runner=host,
+        sleeper=lambda _seconds: None,
+        execute_restart=True,
+    )
+
+    assert result["outcome"] == "controlled_start"
+    assert result["schema"] == installer.INSTALL_EXECUTION_RESULT_SCHEMA
+    assert result["transition"] == {
+        "outcome": "controlled_start",
+        "prior_posture": "inactive",
+        "prior_n_restarts": 0,
+        "final_posture": "active",
+        "service_action": "start",
+        "managed_files_replaced": False,
+    }
+    assert host.active
+    assert host.start_calls == 1
+    assert host.restart_calls == restart_calls
+    assert host.stop_calls == stop_calls
+    assert {
+        role: path.read_bytes()
+        for role, (path, _mode) in installer._managed_files(paths).items()
+    } == managed_before
+    assert [
+        call[1]
+        for call in host.calls[calls_before:]
+        if call[0] == installer.SYSTEMCTL and call[1] in {"start", "restart", "stop", "daemon-reload"}
+    ] == ["start"]
+
+
+@pytest.mark.parametrize(
+    "role",
+    ["unit", "helper", "release", "receipt", "rollback_state"],
+)
+def test_each_missing_managed_file_is_rejected_as_partial_before_mutation(
+    tmp_path: Path,
+    role: str,
+) -> 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,
+    )
+    managed_path = installer._managed_files(paths)[role][0]
+    managed_path.unlink()
+    calls_before = len(host.calls)
+
+    with pytest.raises(installer.InstallError, match="managed install state was partial"):
+        installer.install_release(
+            bundle,
+            paths=paths,
+            source_binding=source_binding(),
+            runner=host,
+            sleeper=lambda _seconds: None,
+            execute_restart=True,
+        )
+
+    assert not any(
+        call[0] == installer.SYSTEMCTL and call[1] in {"start", "restart", "stop", "daemon-reload"}
+        for call in host.calls[calls_before:]
+    )
+
+
+@pytest.mark.parametrize(
+    "mutation",
+    [
+        "outer_extra",
+        "rollback_extra",
+        "previous_release_integer",
+        "previous_active_integer",
+        "automatic_integer",
+        "service_restart_drift",
+        "rollback_posture_drift",
+        "load_state_drift",
+        "main_pid_drift",
+        "verification_forged",
+        "authority_forged",
+        "invocation_integer",
+        "container_integer",
+        "cmdline_integer",
+        "environment_nonstring",
+    ],
+)
+def test_current_install_receipt_is_exact_and_bound_to_rollback_state(
+    tmp_path: Path,
+    mutation: str,
+) -> 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"))
+    if mutation == "outer_extra":
+        receipt["unexpected"] = "field"
+    elif mutation == "rollback_extra":
+        receipt["rollback"]["unexpected"] = "field"
+    elif mutation == "previous_release_integer":
+        receipt["rollback"]["previous_release_present"] = 0
+    elif mutation == "previous_active_integer":
+        receipt["rollback"]["previous_service_active"] = 0
+    elif mutation == "automatic_integer":
+        receipt["rollback"]["automatic_on_failure"] = 1
+    elif mutation == "service_restart_drift":
+        receipt["service_before"]["n_restarts"] += 1
+    elif mutation == "rollback_posture_drift":
+        receipt["rollback"]["previous_service_posture"] = "active"
+    elif mutation == "load_state_drift":
+        receipt["service_before"]["load_state"] = "loaded"
+    elif mutation == "main_pid_drift":
+        receipt["service_before"]["main_pid"] = 12345
+    elif mutation == "verification_forged":
+        receipt["verification"] = {"status": "forged"}
+    elif mutation == "authority_forged":
+        receipt["authority"] = {"cloud_or_database_authority_changed": True}
+    elif mutation == "invocation_integer":
+        receipt["verification"]["invocation_id"] = int("1" * 32)
+    elif mutation == "container_integer":
+        receipt["verification"]["container_id"] = int("2" * 64)
+    elif mutation == "cmdline_integer":
+        receipt["verification"]["supervising_process"]["cmdline_sha256"] = int("3" * 64)
+    else:
+        receipt["verification"]["validated_process_environment_fields"] = [{}]
+    paths.receipt.write_text(json.dumps(receipt, sort_keys=True) + "\n", encoding="utf-8")
+    paths.receipt.chmod(0o600)
+    stop_calls = host.stop_calls
+
+    with pytest.raises(installer.InstallError, match="installed receipt"):
+        installer.install_release(
+            bundle,
+            paths=paths,
+            source_binding=source_binding(),
+            runner=host,
+            sleeper=lambda _seconds: None,
+            execute_restart=True,
+        )
+    with pytest.raises(installer.InstallError, match="installed receipt"):
+        installer._validate_managed_snapshots(
+            installer._current_managed_snapshots(paths),
+            paths=paths,
+        )
+
+    assert host.stop_calls == stop_calls
+
+
+def test_legacy_v2_install_receipt_remains_readable_without_rewrite(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,
+    )
+    legacy = json.loads(paths.receipt.read_text(encoding="utf-8"))
+    legacy["schema"] = installer.LEGACY_INSTALL_RECEIPT_SCHEMA
+    legacy["result"] = "installed"
+    legacy.pop("outcome")
+    legacy.pop("transition")
+    paths.receipt.write_text(json.dumps(legacy, indent=2, sort_keys=True) + "\n", encoding="utf-8")
+    paths.receipt.chmod(0o600)
+    receipt_before = paths.receipt.read_bytes()
+
+    result = installer.install_release(
+        bundle,
+        paths=paths,
+        source_binding=source_binding(),
+        runner=host,
+        sleeper=lambda _seconds: None,
+        execute_restart=True,
+    )
+
+    assert result["outcome"] == "verified_noop"
+    assert paths.receipt.read_bytes() == receipt_before
+
+
+@pytest.mark.parametrize("failure_mode", ["start_command", "post_start_verification"])
+def test_exact_inactive_start_failure_restores_inactive_without_rewriting_history(
+    tmp_path: Path,
+    failure_mode: str,
+) -> 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,
+    )
+    host._stop()
+    managed_before = {
+        role: path.read_bytes()
+        for role, (path, _mode) in installer._managed_files(paths).items()
+    }
+    if failure_mode == "post_start_verification":
+        host.fail_dropin_release = str(release["release_sha256"])
+
+    def runner(arguments: list[str]) -> subprocess.CompletedProcess[bytes]:
+        completed = host(arguments)
+        if failure_mode == "start_command" and arguments[:2] == [installer.SYSTEMCTL, "start"]:
+            return subprocess.CompletedProcess(arguments, 1, stdout=b"", stderr=b"injected")
+        return completed
+
+    with pytest.raises(installer.VerificationError) as caught:
+        installer.install_release(
+            bundle,
+            paths=paths,
+            source_binding=source_binding(),
+            runner=runner,
+            sleeper=lambda _seconds: None,
+            execute_restart=True,
+        )
+
+    assert caught.value.rollback_executed is True
+    assert caught.value.restored_posture == "inactive"
+    assert caught.value.prior_posture_preserved is True
+    assert not host.active
+    assert host.container is None
+    assert host.start_calls == 1
+    assert {
+        role: path.read_bytes()
+        for role, (path, _mode) in installer._managed_files(paths).items()
+    } == managed_before
+
+
+@pytest.mark.parametrize("cleanup_failure", ["stop_command", "restart_counter_drift"])
+def test_exact_inactive_cleanup_failure_never_claims_prior_posture_preserved(
+    tmp_path: Path,
+    cleanup_failure: str,
+) -> 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,
+    )
+    host._stop()
+    host.fail_dropin_release = str(release["release_sha256"])
+    managed_before = {
+        role: path.read_bytes()
+        for role, (path, _mode) in installer._managed_files(paths).items()
+    }
+
+    def runner(arguments: list[str]) -> subprocess.CompletedProcess[bytes]:
+        completed = host(arguments)
+        if arguments[:2] == [installer.SYSTEMCTL, "stop"]:
+            if cleanup_failure == "stop_command":
+                return subprocess.CompletedProcess(arguments, 1, stdout=b"", stderr=b"injected")
+            host.n_restarts += 1
+        return completed
+
+    with pytest.raises(installer.InstallError, match="cleanup was incomplete") as caught:
+        installer.install_release(
+            bundle,
+            paths=paths,
+            source_binding=source_binding(),
+            runner=runner,
+            sleeper=lambda _seconds: None,
+            execute_restart=True,
+        )
+
+    assert getattr(caught.value, "prior_posture_preserved", None) is not True
+    assert not host.active
+    assert {
+        role: path.read_bytes()
+        for role, (path, _mode) in installer._managed_files(paths).items()
+    } == managed_before
+
+
+@pytest.mark.parametrize("prior_posture", ["active", "inactive"])
+def test_exact_release_rejects_receipt_replacement_race(
+    tmp_path: Path,
+    prior_posture: str,
+) -> 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,
+    )
+    if prior_posture == "inactive":
+        host._stop()
+    replaced = False
+    service_show_calls = 0
+
+    def replace_receipt() -> None:
+        nonlocal replaced
+        if replaced:
+            return
+        replaced = True
+        paths.receipt.unlink()
+        paths.receipt.write_text('{"schema":"forged"}\n', encoding="utf-8")
+        paths.receipt.chmod(0o600)
+
+    def runner(arguments: list[str]) -> subprocess.CompletedProcess[bytes]:
+        nonlocal service_show_calls
+        completed = host(arguments)
+        if prior_posture == "active" and arguments[:2] == [installer.SYSTEMCTL, "show"]:
+            service_show_calls += 1
+            if service_show_calls == 3:
+                replace_receipt()
+        if prior_posture == "inactive" and arguments[:2] == [installer.SYSTEMCTL, "start"]:
+            replace_receipt()
+        return completed
+
+    with pytest.raises(installer.InstallError, match="file identity drifted"):
+        installer.install_release(
+            bundle,
+            paths=paths,
+            source_binding=source_binding(),
+            runner=runner,
+            sleeper=lambda _seconds: None,
+            execute_restart=True,
+        )
+
+    assert replaced
+    assert host.active is (prior_posture == "active")
+
+
 def test_first_install_waits_for_complete_cid_and_healthy_container(tmp_path: Path) -> None:
     bundle, release, image = release_bundle(tmp_path, "2")
     paths = install_paths(tmp_path)
@@ -1135,7 +1568,94 @@ def test_upgrade_stops_verified_prior_release_before_unit_replacement(tmp_path:
         execute_restart=True,
     )
     start = len(host.calls)
-    receipt = installer.install_release(
+    result = installer.install_release(
+        second_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"))
+
+    actions = [call[1] for call in host.calls[start:] if call[0] == installer.SYSTEMCTL]
+    assert actions.index("stop") < actions.index("daemon-reload") < actions.index("restart")
+    assert result["release_sha256"] == second_release["release_sha256"]
+    assert result["outcome"] == "replaced"
+    assert result["transition"] == {
+        "outcome": "replaced",
+        "prior_posture": "active",
+        "prior_n_restarts": 0,
+        "final_posture": "active",
+        "service_action": "restart",
+        "managed_files_replaced": True,
+    }
+    assert receipt["rollback"]["previous_service_posture"] == "active"
+    assert host.current_release == second_release
+    assert host.stop_calls == 1
+
+
+def test_inactive_prior_release_is_replaced_without_claiming_active_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()
+    stop_calls = host.stop_calls
+
+    result = installer.install_release(
+        second_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"))
+
+    assert result["outcome"] == "replaced"
+    assert result["transition"]["prior_posture"] == "inactive"
+    assert result["transition"]["prior_n_restarts"] == 0
+    assert receipt["rollback"]["previous_release_present"] is True
+    assert receipt["rollback"]["previous_service_active"] is False
+    assert receipt["rollback"]["previous_service_posture"] == "inactive"
+    assert host.stop_calls == stop_calls
+    assert host.active
+
+
+def test_active_prior_verification_failure_reports_unchanged_recovery_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,
+    )
+    process_root = paths.proc_root / str(host.main_pid)
+    (process_root / "environ").write_bytes(b"PATH=/unexpected\0")
+    stop_calls = host.stop_calls
+
+    with pytest.raises(installer.VerificationError) as caught:
+        installer.install_release(
             second_bundle,
             paths=paths,
             source_binding=source_binding(),
@@ -1144,11 +1664,11 @@ def test_upgrade_stops_verified_prior_release_before_unit_replacement(tmp_path:
             execute_restart=True,
         )
 
-    actions = [call[1] for call in host.calls[start:] if call[0] == installer.SYSTEMCTL]
+    assert caught.value.rollback_executed is False
-    assert actions.index("stop") < actions.index("daemon-reload") < actions.index("restart")
+    assert caught.value.restored_posture == "active"
-    assert receipt["release_sha256"] == second_release["release_sha256"]
+    assert caught.value.prior_posture_preserved is True
-    assert host.current_release == second_release
+    assert host.stop_calls == stop_calls
-    assert host.stop_calls == 1
+    assert host.active
 
 
 def test_stale_loaded_unit_is_rejected_before_trusting_old_exec_stop(tmp_path: Path) -> None:
@@ -1327,6 +1847,45 @@ def test_transitional_prior_service_posture_is_rejected_before_mutation(tmp_path
     assert host.restart_calls == 1
 
 
+def test_failed_prior_service_posture_is_rejected_before_mutation(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.property_overrides.update(
+        {
+            "ActiveState": "failed",
+            "SubState": "failed",
+            "MainPID": "0",
+            "InvocationID": "",
+            "ControlGroup": "",
+        }
+    )
+
+    with pytest.raises(installer.InstallError, match="transitional or failed"):
+        installer.install_release(
+            second_bundle,
+            paths=paths,
+            source_binding=source_binding(),
+            runner=host,
+            sleeper=lambda _seconds: None,
+            execute_restart=True,
+        )
+
+    assert host.stop_calls == 0
+    assert host.restart_calls == 1
+
+
 def test_changing_prior_service_posture_fails_two_readback_gate(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")
@@ -1366,7 +1925,7 @@ def test_wrong_local_image_is_rejected_without_service_restart(tmp_path: Path) -
     host = FakeHost(paths)
     host.add_release(release, image)
 
-    with pytest.raises(installer.InstallError, match=r"finalized (release|inspection)"):
+    with pytest.raises(installer.VerificationError, match=r"finalized (release|inspection)") as caught:
         installer.install_release(
             bundle,
             paths=paths,
@@ -1376,6 +1935,9 @@ def test_wrong_local_image_is_rejected_without_service_restart(tmp_path: Path) -
             execute_restart=True,
         )
 
+    assert caught.value.rollback_executed is False
+    assert caught.value.restored_posture == "absent"
+    assert caught.value.prior_posture_preserved is True
     assert host.restart_calls == 0
     assert not paths.unit.exists()
     assert not paths.state_dir.exists()