Merge pull request #241 from living-ip/fix/iap-missing-receipt

Preserve no-send IAP failure receipts

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

@@ -344,6 +344,33 @@ contact GCP or Cloud SQL, change Telegram traffic, touch the VPS, approve a
 proposal, or promote production. Those remain later, separately authorized
 boundaries.
 
+## IAP result and failure receipts
+
+The fixed-target IAP runner treats the remote `result.json`, the remote process
+exit status/stdout, the IAP transport, and the private local result receipt as
+one bounded result path. A successful operation is accepted only when the
+structured result matches the exact request, revision, artifact, image, config
+digest, service, database proof, and no-send authority contract.
+
+Post-install failures are returned with allowlisted `failure_stage` and
+`failure_reason` values. If explicit rollback passes, its exact restored
+posture is recorded. If rollback cannot be proven, the remote helper stops the
+staging service only after exact-binding the currently loaded unit, `ExecStop`,
+supervising process, and container to the reviewed release. It then proves
+`inactive/dead`, `MainPID=0`, and absence of the named candidate container
+before reporting a fail-closed posture. Drift blocks the stop; a failure to
+prove that posture is reported as unknown and is never converted into success.
+
+If IAP returns no output, malformed output, a contract mismatch, or a transport
+interruption, the local runner writes a private self-hashed failure receipt
+with the bounded stage/reason, transfer-cleanup result, and `remote_state:
+unknown`. Command stdout, stderr, tokens, passwords, and secret values are not
+copied into a receipt or exception. That local-only receipt records requested
+authority, not unproven remote outcomes. Operators must not infer a successful
+or inert remote service from a local transport-failure receipt. Remote
+operation failures and local transport failures use distinct outer receipt
+schemas; the successful result-receipt v1 contract is unchanged.
+
 ## Offline evidence and live gate
 
 Offline tests prove strict receipt/descriptor/inspection schemas, exact

ops/gcp_leoclean_nosend_iap_remote.py

@@ -16,6 +16,7 @@ import subprocess
 import sys
 import tarfile
 import tempfile
+import time
 import zipfile
 from collections.abc import Callable, Mapping
 from pathlib import Path
@@ -24,6 +25,7 @@ from typing import Any
 SCHEMA = "livingip.leocleanNoSendIapRequest.v1"
 RESULT_SCHEMA = "livingip.leocleanNoSendIapRemoteResult.v1"
 FAILURE_RESULT_SCHEMA = "livingip.leocleanNoSendIapFailure.v1"
+OPERATION_FAILURE_RESULT_SCHEMA = "livingip.leocleanNoSendIapOperationalFailure.v1"
 PROJECT = "teleo-501523"
 ZONE = "europe-west6-a"
 INSTANCE = "teleo-staging-1"
@@ -48,6 +50,17 @@ MAX_MEMBER_BYTES = 2 * 1024 * 1024
 MAX_MANIFEST_BYTES = 128 * 1024
 MAX_COMMAND_OUTPUT_BYTES = 1024 * 1024
 MAX_TOKEN_BYTES = 64 * 1024
+POST_INSTALL_FAILURES = frozenset(
+    {
+        ("install_receipt", "install_receipt_invalid"),
+        ("service_before_restart", "service_verification_failed"),
+        ("database_before_restart", "database_verification_failed"),
+        ("service_restart", "service_restart_failed"),
+        ("service_after_restart", "service_verification_failed"),
+        ("restart_identity", "restart_identity_mismatch"),
+        ("database_after_restart", "database_verification_failed"),
+    }
+)
 EXPECTED_MEMBERS = {
     "request-manifest.json",
     "release-artifact.zip",
@@ -852,6 +865,236 @@ def _is_verification_failure(installer_module: Any, failure: Exception) -> bool:
     return isinstance(failure_type, type) and isinstance(failure, failure_type)
 
 
+def _post_install_failure_coordinates(
+    installer_module: Any,
+    failure: Exception,
+    *,
+    fallback_stage: str,
+    fallback_reason: str,
+) -> tuple[str, str]:
+    if _is_verification_failure(installer_module, failure):
+        failure_stage = getattr(failure, "failure_stage", None)
+        failure_reason = getattr(failure, "failure_reason", None)
+        _require(
+            isinstance(failure_stage, str)
+            and isinstance(failure_reason, str)
+            and (failure_stage, failure_reason) in installer_module.VERIFICATION_FAILURES,
+            "verification failure was not allowlisted",
+        )
+        return failure_stage, failure_reason
+    _require(
+        (fallback_stage, fallback_reason) in POST_INSTALL_FAILURES,
+        "post-install failure was not allowlisted",
+    )
+    return fallback_stage, fallback_reason
+
+
+def _rollback_proven_service(posture: str) -> dict[str, Any]:
+    _require(posture in {"active", "inactive", "absent"}, "rollback posture was invalid")
+    return {
+        "status": "not_required",
+        "service_posture": posture,
+        "active_state": "not_observed",
+        "sub_state": "not_observed",
+        "main_pid": None,
+        "n_restarts": None,
+        "candidate_container": "not_observed",
+    }
+
+
+def _fail_closed_inert_service(
+    *,
+    installer_module: Any,
+    release: Mapping[str, Any],
+    paths: Any,
+    runner: Runner,
+) -> dict[str, Any]:
+    def contract_runner(argv: list[str]) -> subprocess.CompletedProcess[bytes]:
+        return _run(
+            argv,
+            runner=runner,
+            label="fail-closed service command",
+            require_success=False,
+        )
+
+    try:
+        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 fail-closed service controls",
+        )
+        before = stable_service_posture(
+            release,
+            runner=contract_runner,
+            paths=paths,
+            sleeper=time.sleep,
+        )
+        _require(before.kind in {"active", "inactive"}, "managed service posture was not trusted")
+        if before.kind == "active":
+            verification = verify_running_release(
+                release,
+                runner=contract_runner,
+                paths=paths,
+                sleeper=time.sleep,
+            )
+            _require(
+                verification.get("main_pid") == before.state.get("MainPID")
+                and verification.get("invocation_id") == before.state.get("InvocationID")
+                and verification.get("n_restarts") == before.state.get("NRestarts"),
+                "running service identity changed before fail-closed stop",
+            )
+            systemctl("stop", contract_runner)
+        after = stable_service_posture(
+            release,
+            runner=contract_runner,
+            paths=paths,
+            sleeper=time.sleep,
+        )
+        _require(
+            after.kind == "inactive"
+            and after.state.get("ActiveState") == "inactive"
+            and after.state.get("SubState") == "dead"
+            and after.state.get("MainPID") == 0
+            and isinstance(after.state.get("NRestarts"), int)
+            and not isinstance(after.state.get("NRestarts"), bool)
+            and after.state["NRestarts"] >= 0,
+            "service did not stop into a stable inert posture",
+        )
+        require_candidate_absent(
+            runner=contract_runner,
+            paths=paths,
+            sleeper=time.sleep,
+        )
+        return {
+            "status": "pass",
+            "service_posture": "inactive",
+            "active_state": "inactive",
+            "sub_state": "dead",
+            "main_pid": 0,
+            "n_restarts": after.state["NRestarts"],
+            "candidate_container": "absent",
+        }
+    except Exception:
+        return {
+            "status": "fail",
+            "service_posture": "unknown",
+            "active_state": "unknown",
+            "sub_state": "unknown",
+            "main_pid": None,
+            "n_restarts": None,
+            "candidate_container": "unknown",
+        }
+
+
+def _operational_failure_result(
+    *,
+    manifest: Mapping[str, Any],
+    release: Mapping[str, Any],
+    failure_stage: str,
+    failure_reason: str,
+    rollback: Mapping[str, str],
+    fail_closed: Mapping[str, Any],
+    archive_sha256: str,
+    helper_sha256: str,
+    installer_module: Any,
+) -> dict[str, Any]:
+    _require(
+        manifest.get("operation") == "install"
+        and set(rollback) == {"status", "mode", "restored_posture"}
+        and (
+            (failure_stage, failure_reason) in POST_INSTALL_FAILURES
+            or (failure_stage, failure_reason) in installer_module.VERIFICATION_FAILURES
+        ),
+        "post-install failure was not allowlisted",
+    )
+    rollback_passed = (
+        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",
+    }
+    verification_pair = (failure_stage, failure_reason) in installer_module.VERIFICATION_FAILURES
+    _require(
+        (rollback_passed and not verification_pair) or rollback_failed,
+        "post-install rollback summary was invalid",
+    )
+    if rollback_passed:
+        _require(
+            fail_closed == _rollback_proven_service(str(rollback["restored_posture"])),
+            "post-install restored posture summary was invalid",
+        )
+    else:
+        _require(
+            fail_closed
+            in (
+                {
+                    "status": "fail",
+                    "service_posture": "unknown",
+                    "active_state": "unknown",
+                    "sub_state": "unknown",
+                    "main_pid": None,
+                    "n_restarts": None,
+                    "candidate_container": "unknown",
+                },
+                {
+                    "status": "pass",
+                    "service_posture": "inactive",
+                    "active_state": "inactive",
+                    "sub_state": "dead",
+                    "main_pid": 0,
+                    "n_restarts": fail_closed.get("n_restarts"),
+                    "candidate_container": "absent",
+                },
+            )
+            and (
+                fail_closed.get("status") != "pass"
+                or (
+                    isinstance(fail_closed.get("n_restarts"), int)
+                    and not isinstance(fail_closed.get("n_restarts"), bool)
+                    and fail_closed["n_restarts"] >= 0
+                )
+            ),
+            "post-install fail-closed summary was invalid",
+        )
+    return {
+        "schema": OPERATION_FAILURE_RESULT_SCHEMA,
+        "status": "fail",
+        "operation": manifest["operation"],
+        "request_id": manifest["request_id"],
+        "runner_revision": manifest["runner_revision"],
+        "release_revision": manifest["release_revision"],
+        "release_sha256": release["release_sha256"],
+        "image_digest": release["image"]["digest"],
+        "image_config_digest": release["image"]["config_digest"],
+        "release_artifact_sha256": manifest["release_artifact"]["archive_sha256"],
+        "request_binding": {
+            "archive_sha256": archive_sha256,
+            "helper_sha256": helper_sha256,
+            "manifest_sha256": manifest["manifest_sha256"],
+        },
+        "failure_stage": failure_stage,
+        "failure_reason": failure_reason,
+        "rollback": dict(rollback),
+        "fail_closed": dict(fail_closed),
+        "authority": {
+            "sending_enabled": False,
+            "production_traffic_changed": False,
+            "vps_touched": False,
+            "canonical_proposal_approved": False,
+        },
+    }
+
+
 def execute_request(
     manifest: Mapping[str, Any],
     payload_root: Path,
@@ -919,7 +1162,6 @@ def execute_request(
                     installer_module=installer_module,
                 )
             ) from exc
-        _require(install_receipt.get("status") == "pass", "installation did not pass")
     elif operation == "rollback":
         rollback = _execute_explicit_rollback(
             installer_module=installer_module,
@@ -945,12 +1187,19 @@ def execute_request(
         )
 
     if operation == "install":
+        failure_stage = "install_receipt"
+        failure_reason = "install_receipt_invalid"
         try:
+            _require(install_receipt.get("status") == "pass", "installation did not pass")
+            failure_stage = "service_before_restart"
+            failure_reason = "service_verification_failed"
             before = installer_module.verify_running_release(
                 release,
                 runner=contract_runner,
                 paths=paths,
             )
+            failure_stage = "database_before_restart"
+            failure_reason = "database_verification_failed"
             before_database = _verify_database(
                 container_id=before["container_id"],
                 run_id=f"{manifest['request_id']}-pre",
@@ -959,16 +1208,22 @@ def execute_request(
                 package_module=package_module,
                 runner=runner,
             )
+            failure_stage = "service_restart"
+            failure_reason = "service_restart_failed"
             _run(
                 [installer_module.SYSTEMCTL, "restart", SERVICE],
                 runner=runner,
                 label="fixed service restart",
             )
+            failure_stage = "service_after_restart"
+            failure_reason = "service_verification_failed"
             after = installer_module.verify_running_release(
                 release,
                 runner=contract_runner,
                 paths=paths,
             )
+            failure_stage = "restart_identity"
+            failure_reason = "restart_identity_mismatch"
             _require(
                 before["main_pid"] != after["main_pid"]
                 and before["invocation_id"] != after["invocation_id"]
@@ -976,6 +1231,8 @@ def execute_request(
                 and after["n_restarts"] >= before["n_restarts"],
                 "post-install restart did not create a new exact service/container identity",
             )
+            failure_stage = "database_after_restart"
+            failure_reason = "database_verification_failed"
             after_database = _verify_database(
                 container_id=after["container_id"],
                 run_id=f"{manifest['request_id']}-post",
@@ -985,6 +1242,12 @@ def execute_request(
                 runner=runner,
             )
         except Exception as exc:
+            bounded_stage, bounded_reason = _post_install_failure_coordinates(
+                installer_module,
+                exc,
+                fallback_stage=failure_stage,
+                fallback_reason=failure_reason,
+            )
             try:
                 recovery = _execute_explicit_rollback(
                     installer_module=installer_module,
@@ -1000,8 +1263,30 @@ def execute_request(
                     archive_sha256=archive_sha256,
                     helper_sha256=helper_sha256,
                 )
-            except Exception as rollback_exc:
-                raise RemoteError("install_failed_rollback_incomplete") from rollback_exc
+            except Exception:
+                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=bounded_stage,
+                        failure_reason=bounded_reason,
+                        rollback={
+                            "status": "fail",
+                            "mode": "explicit",
+                            "restored_posture": "unknown",
+                        },
+                        fail_closed=fail_closed,
+                        archive_sha256=archive_sha256,
+                        helper_sha256=helper_sha256,
+                        installer_module=installer_module,
+                    )
+                ) from exc
             if _is_verification_failure(installer_module, exc):
                 raise RemoteOperationFailure(
                     _verification_failure_result(
@@ -1018,7 +1303,24 @@ def execute_request(
                         installer_module=installer_module,
                     )
                 ) from exc
-            raise RemoteError("install_failed_rolled_back") from exc
+            restored_posture = str(rollback_result["service"]["restored_posture"])
+            raise RemoteOperationFailure(
+                _operational_failure_result(
+                    manifest=manifest,
+                    release=release,
+                    failure_stage=bounded_stage,
+                    failure_reason=bounded_reason,
+                    rollback={
+                        "status": "pass",
+                        "mode": "explicit",
+                        "restored_posture": restored_posture,
+                    },
+                    fail_closed=_rollback_proven_service(restored_posture),
+                    archive_sha256=archive_sha256,
+                    helper_sha256=helper_sha256,
+                    installer_module=installer_module,
+                )
+            ) from exc
         service = {
             "name": SERVICE,
             "before_restart": before,
@@ -1089,6 +1391,35 @@ def execute_request(
     }
 
 
+def _execute_and_persist_result(
+    manifest: Mapping[str, Any],
+    payload_root: Path,
+    request_dir: Path,
+    archive_sha256: str,
+    helper_sha256: str,
+    *,
+    runner: Runner,
+) -> dict[str, Any]:
+    try:
+        result = execute_request(
+            manifest,
+            payload_root,
+            request_dir,
+            archive_sha256,
+            helper_sha256,
+            runner=runner,
+        )
+    except RemoteOperationFailure as exc:
+        result = exc.result
+    result_path = request_dir / "result.json"
+    _copy_into_request(
+        result_path,
+        (json.dumps(result, allow_nan=False, indent=2, sort_keys=True) + "\n").encode("utf-8"),
+        0o600,
+    )
+    return result
+
+
 def receive_request(
     operation: str,
     request_id: str,
@@ -1154,24 +1485,14 @@ def receive_request(
                 release_revision=release_revision,
             )
             verify_service_account(runner=runner)
-            try:
-                result = execute_request(
-                    manifest,
-                    payload_root,
-                    request_dir,
-                    archive_sha256,
-                    helper_sha256,
-                    runner=runner,
-                )
-            except RemoteOperationFailure as exc:
-                result = exc.result
-            result_path = request_dir / "result.json"
-            _copy_into_request(
-                result_path,
-                (json.dumps(result, allow_nan=False, indent=2, sort_keys=True) + "\n").encode("utf-8"),
-                0o600,
+            return _execute_and_persist_result(
+                manifest,
+                payload_root,
+                request_dir,
+                archive_sha256,
+                helper_sha256,
+                runner=runner,
             )
-            return result
         finally:
             os.close(lock_descriptor)
     finally:

ops/run_gcp_leoclean_nosend_iap.py

@@ -33,7 +33,10 @@ SCHEMA = "livingip.leocleanNoSendIapRequest.v1"
 PLAN_SCHEMA = "livingip.leocleanNoSendIapPlan.v1"
 REMOTE_RESULT_SCHEMA = "livingip.leocleanNoSendIapRemoteResult.v1"
 REMOTE_FAILURE_SCHEMA = "livingip.leocleanNoSendIapFailure.v1"
+REMOTE_OPERATION_FAILURE_SCHEMA = "livingip.leocleanNoSendIapOperationalFailure.v1"
 RESULT_RECEIPT_SCHEMA = "livingip.leocleanNoSendIapResultReceipt.v1"
+FAILURE_RECEIPT_SCHEMA = "livingip.leocleanNoSendIapFailureReceipt.v1"
+LOCAL_FAILURE_RECEIPT_SCHEMA = "livingip.leocleanNoSendIapLocalFailureReceipt.v1"
 EXECUTION_SCHEMA = "livingip.leocleanNoSendIapExecution.v1"
 PROJECT = package.PROJECT
 ZONE = package.ZONE
@@ -78,6 +81,26 @@ FORBIDDEN_OUTPUT_MARKERS = (
     "77.42.65.182",
     "teleo-prod-1",
 )
+POST_INSTALL_FAILURES = frozenset(
+    {
+        ("install_receipt", "install_receipt_invalid"),
+        ("service_before_restart", "service_verification_failed"),
+        ("database_before_restart", "database_verification_failed"),
+        ("service_restart", "service_restart_failed"),
+        ("service_after_restart", "service_verification_failed"),
+        ("restart_identity", "restart_identity_mismatch"),
+        ("database_after_restart", "database_verification_failed"),
+    }
+)
+LOCAL_FAILURES = frozenset(
+    {
+        ("iap_transport", "upload_failed"),
+        ("iap_transport", "remote_command_interrupted"),
+        ("remote_receipt", "absent"),
+        ("remote_receipt", "malformed"),
+        ("remote_receipt", "contract_mismatch"),
+    }
+)
 BOOTSTRAP_CODE = textwrap.dedent(
     r"""
     import fcntl, hashlib, io, json, os, pwd, re, stat, sys, tarfile
@@ -203,7 +226,7 @@ def _strict_json(raw: bytes, label: str) -> dict[str, Any]:
             object_pairs_hook=reject_duplicates,
             parse_constant=lambda _value: (_ for _ in ()).throw(ValueError("constant")),
         )
-    except (UnicodeError, ValueError, json.JSONDecodeError) as exc:
+    except (UnicodeError, ValueError, json.JSONDecodeError, RecursionError) as exc:
         raise IapRequestError(f"{label} was invalid JSON") from exc
     _require(isinstance(value, dict), f"{label} was not a JSON object")
     return value
@@ -717,6 +740,17 @@ def _action_receipts(commands: tuple[list[str], list[str], list[str]]) -> list[d
     ]
 
 
+def _expected_authority(operation: str) -> dict[str, bool]:
+    _require(operation in OPERATIONS, "operation was invalid")
+    return {
+        "gcp_mutation": operation in {"install", "rollback"},
+        "service_restart": operation in {"install", "rollback"},
+        "database_or_secret_provisioning": False,
+        "production_or_transport": False,
+        "vps": False,
+    }
+
+
 def prepare_request(
     operation: str,
     request_id: str,
@@ -833,13 +867,7 @@ def prepare_request(
             "ssh_key": str(ssh_key),
         },
         "actions": actions,
-        "authority": {
-            "gcp_mutation": operation in {"install", "rollback"},
-            "service_restart": operation in {"install", "rollback"},
-            "database_or_secret_provisioning": False,
-            "production_or_transport": False,
-            "vps": False,
-        },
+        "authority": _expected_authority(operation),
     }
 
 
@@ -1028,33 +1056,114 @@ def validate_remote_result(plan: Mapping[str, Any], result: Mapping[str, Any]) -
 
 
 def validate_remote_failure(plan: Mapping[str, Any], result: Mapping[str, Any]) -> None:
-    _require(
-        set(result)
-        == {
-            "schema",
-            "status",
-            "operation",
-            "request_id",
-            "runner_revision",
-            "release_revision",
-            "release_sha256",
-            "image_digest",
-            "image_config_digest",
-            "release_artifact_sha256",
-            "request_binding",
-            "failure_stage",
-            "failure_reason",
-            "rollback",
-            "authority",
-        },
-        "remote failure fields were not exact",
-    )
     request = plan["request"]
     release = plan["release"]
     rollback = result.get("rollback")
+    common_fields = {
+        "schema",
+        "status",
+        "operation",
+        "request_id",
+        "runner_revision",
+        "release_revision",
+        "release_sha256",
+        "image_digest",
+        "image_config_digest",
+        "release_artifact_sha256",
+        "request_binding",
+        "failure_stage",
+        "failure_reason",
+        "rollback",
+        "authority",
+    }
+    schema = result.get("schema")
+    if schema == REMOTE_FAILURE_SCHEMA:
+        _require(set(result) == common_fields, "remote failure fields were not exact")
+        failure_contract = (
+            (result.get("failure_stage"), result.get("failure_reason")) in installer.VERIFICATION_FAILURES
+            and rollback
+            in (
+                {"status": "pass", "mode": "installer_automatic", "restored_posture": "prior"},
+                {"status": "pass", "mode": "explicit", "restored_posture": "active"},
+                {"status": "pass", "mode": "explicit", "restored_posture": "inactive"},
+                {"status": "pass", "mode": "explicit", "restored_posture": "absent"},
+                {"status": "not_run", "mode": "not_applicable", "restored_posture": "unchanged"},
+            )
+        )
+    else:
+        _require(
+            schema == REMOTE_OPERATION_FAILURE_SCHEMA and set(result) == common_fields | {"fail_closed"},
+            "remote failure fields were not exact",
+        )
+        failure_pair = (result.get("failure_stage"), result.get("failure_reason"))
+        rollback_passed = (
+            isinstance(rollback, dict)
+            and set(rollback) == {"status", "mode", "restored_posture"}
+            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")
+        if rollback_passed:
+            fail_closed_contract = fail_closed == {
+                "status": "not_required",
+                "service_posture": rollback["restored_posture"],
+                "active_state": "not_observed",
+                "sub_state": "not_observed",
+                "main_pid": None,
+                "n_restarts": None,
+                "candidate_container": "not_observed",
+            }
+        else:
+            fail_closed_contract = (
+                fail_closed
+                == {
+                    "status": "fail",
+                    "service_posture": "unknown",
+                    "active_state": "unknown",
+                    "sub_state": "unknown",
+                    "main_pid": None,
+                    "n_restarts": None,
+                    "candidate_container": "unknown",
+                }
+                or (
+                    isinstance(fail_closed, dict)
+                    and set(fail_closed)
+                    == {
+                        "status",
+                        "service_posture",
+                        "active_state",
+                        "sub_state",
+                        "main_pid",
+                        "n_restarts",
+                        "candidate_container",
+                    }
+                    and fail_closed.get("status") == "pass"
+                    and fail_closed.get("service_posture") == "inactive"
+                    and fail_closed.get("active_state") == "inactive"
+                    and fail_closed.get("sub_state") == "dead"
+                    and fail_closed.get("main_pid") == 0
+                    and isinstance(fail_closed.get("n_restarts"), int)
+                    and not isinstance(fail_closed.get("n_restarts"), bool)
+                    and fail_closed["n_restarts"] >= 0
+                    and fail_closed.get("candidate_container") == "absent"
+                )
+            )
+        failure_contract = (
+            plan["operation"] == "install"
+            and (
+                (failure_pair in POST_INSTALL_FAILURES and (rollback_passed or rollback_failed))
+                or (failure_pair in installer.VERIFICATION_FAILURES and rollback_failed)
+            )
+            and fail_closed_contract
+        )
     _require(
-        result.get("schema") == REMOTE_FAILURE_SCHEMA
-        and result.get("status") == "fail"
+        result.get("status") == "fail"
         and result.get("operation") == plan["operation"]
         and result.get("request_id") == plan["request_id"]
         and result.get("runner_revision") == plan["runner_revision"]
@@ -1069,15 +1178,7 @@ def validate_remote_failure(plan: Mapping[str, Any], result: Mapping[str, Any])
             "helper_sha256": request["helper_sha256"],
             "manifest_sha256": request["manifest_sha256"],
         }
-        and (result.get("failure_stage"), result.get("failure_reason")) in installer.VERIFICATION_FAILURES
-        and rollback
-        in (
-            {"status": "pass", "mode": "installer_automatic", "restored_posture": "prior"},
-            {"status": "pass", "mode": "explicit", "restored_posture": "active"},
-            {"status": "pass", "mode": "explicit", "restored_posture": "inactive"},
-            {"status": "pass", "mode": "explicit", "restored_posture": "absent"},
-            {"status": "not_run", "mode": "not_applicable", "restored_posture": "unchanged"},
-        )
+        and failure_contract
         and result.get("authority")
         == {
             "sending_enabled": False,
@@ -1089,33 +1190,8 @@ def validate_remote_failure(plan: Mapping[str, Any], result: Mapping[str, Any])
     )
 
 
-def _write_result_receipt(
-    plan: Mapping[str, Any],
-    result: Mapping[str, Any],
-    *,
-    status: str = "pass",
-) -> tuple[Path, dict[str, Any]]:
-    _require(status in {"pass", "fail"} and result.get("status") == status, "local result status was invalid")
-    archive = Path(str(plan["request"]["archive"]))
-    output = archive.parent / f"livingip-leoclean-nosend-result-{plan['request_id']}.json"
+def _publish_local_receipt(output: Path, stable: Mapping[str, Any]) -> dict[str, Any]:
     _require(not output.exists() and not output.is_symlink(), "local result receipt already existed")
-    stable = {
-        "schema": RESULT_RECEIPT_SCHEMA,
-        "status": status,
-        "operation": plan["operation"],
-        "request_id": plan["request_id"],
-        "runner_revision": plan["runner_revision"],
-        "release_revision": plan["release_revision"],
-        "request": {
-            "archive_sha256": plan["request"]["archive_sha256"],
-            "helper_sha256": plan["request"]["helper_sha256"],
-            "manifest_sha256": plan["request"]["manifest_sha256"],
-        },
-        "release": dict(plan["release"]),
-        "release_artifact": dict(plan["release_artifact"]),
-        "expected": dict(plan["expected"]),
-        "remote_result": dict(result),
-    }
     receipt = {**stable, "receipt_sha256": package.canonical_sha256(stable)}
     content = _canonical_json(receipt)
     descriptor, temporary_name = tempfile.mkstemp(prefix=".iap-result-", dir=output.parent)
@@ -1142,6 +1218,74 @@ def _write_result_receipt(
                 temporary.unlink()
         except FileNotFoundError:
             pass
+    return receipt
+
+
+def _write_result_receipt(
+    plan: Mapping[str, Any],
+    result: Mapping[str, Any],
+    *,
+    status: str = "pass",
+) -> tuple[Path, dict[str, Any]]:
+    _require(status in {"pass", "fail"} and result.get("status") == status, "local result status was invalid")
+    archive = Path(str(plan["request"]["archive"]))
+    output = archive.parent / f"livingip-leoclean-nosend-result-{plan['request_id']}.json"
+    stable = {
+        "schema": RESULT_RECEIPT_SCHEMA if status == "pass" else FAILURE_RECEIPT_SCHEMA,
+        "status": status,
+        "operation": plan["operation"],
+        "request_id": plan["request_id"],
+        "runner_revision": plan["runner_revision"],
+        "release_revision": plan["release_revision"],
+        "request": {
+            "archive_sha256": plan["request"]["archive_sha256"],
+            "helper_sha256": plan["request"]["helper_sha256"],
+            "manifest_sha256": plan["request"]["manifest_sha256"],
+        },
+        "release": dict(plan["release"]),
+        "release_artifact": dict(plan["release_artifact"]),
+        "expected": dict(plan["expected"]),
+        "remote_result": dict(result),
+    }
+    receipt = _publish_local_receipt(output, stable)
+    return output, receipt
+
+
+def _write_local_failure_receipt(
+    plan: Mapping[str, Any],
+    *,
+    failure_stage: str,
+    failure_reason: str,
+    cleanup_status: str,
+) -> tuple[Path, dict[str, Any]]:
+    _require(
+        (failure_stage, failure_reason) in LOCAL_FAILURES and cleanup_status in {"pass", "fail"},
+        "local failure classification was invalid",
+    )
+    archive = Path(str(plan["request"]["archive"]))
+    output = archive.parent / f"livingip-leoclean-nosend-result-{plan['request_id']}.json"
+    stable = {
+        "schema": LOCAL_FAILURE_RECEIPT_SCHEMA,
+        "status": "fail",
+        "operation": plan["operation"],
+        "request_id": plan["request_id"],
+        "runner_revision": plan["runner_revision"],
+        "release_revision": plan["release_revision"],
+        "request": {
+            "archive_sha256": plan["request"]["archive_sha256"],
+            "helper_sha256": plan["request"]["helper_sha256"],
+            "manifest_sha256": plan["request"]["manifest_sha256"],
+        },
+        "release": dict(plan["release"]),
+        "release_artifact": dict(plan["release_artifact"]),
+        "expected": dict(plan["expected"]),
+        "failure_stage": failure_stage,
+        "failure_reason": failure_reason,
+        "cleanup": {"status": cleanup_status, "mode": "failed_transfer"},
+        "remote_state": "unknown",
+        "requested_authority": dict(plan["authority"]),
+    }
+    receipt = _publish_local_receipt(output, stable)
     return output, receipt
 
 
@@ -1177,6 +1321,7 @@ def execute_plan(
     )
     operation = plan.get("operation")
     _require(operation in {"install", "verify", "rollback"}, "operation was not executable")
+    _require(plan.get("authority") == _expected_authority(operation), "request authority was not exact")
     request_id = plan.get("request_id")
     runner_revision = plan.get("runner_revision")
     release_revision = plan.get("release_revision")
@@ -1261,9 +1406,12 @@ def execute_plan(
     scp, ssh, cleanup = expected_commands
     validate_remote_main(runner_revision, runner=git_runner)
     environment = dict(gcloud_env or gcloud_environment())
+    failure_stage = "iap_transport"
+    failure_reason = "upload_failed"
     try:
         try:
             _run_bounded(list(scp), runner=runner, label="IAP upload", environment=environment)
+            failure_reason = "remote_command_interrupted"
             completed = _run_bounded(
                 list(ssh),
                 runner=runner,
@@ -1271,8 +1419,17 @@ def execute_plan(
                 environment=environment,
                 accepted_returncodes=(0, 65),
             )
+            failure_stage = "remote_receipt"
+            if not completed.stdout:
+                failure_reason = "absent"
+                raise IapRequestError("remote result was absent")
+            failure_reason = "malformed"
             result = _strict_json(completed.stdout, "remote result")
-            encoded = json.dumps(result, allow_nan=False, sort_keys=True).casefold()
+            try:
+                encoded = json.dumps(result, allow_nan=False, sort_keys=True).casefold()
+            except (TypeError, ValueError, RecursionError) as exc:
+                raise IapRequestError("remote result could not be normalized") from exc
+            failure_reason = "contract_mismatch"
             _require(
                 not any(marker in encoded for marker in FORBIDDEN_OUTPUT_MARKERS),
                 "remote result was not sanitized",
@@ -1282,6 +1439,7 @@ def execute_plan(
             else:
                 validate_remote_result(plan, result)
         except IapRequestError as exc:
+            cleanup_status = "pass"
             try:
                 _run_bounded(
                     list(cleanup),
@@ -1289,9 +1447,19 @@ def execute_plan(
                     label="IAP failed-transfer cleanup",
                     environment=environment,
                 )
-            except IapRequestError as cleanup_exc:
-                raise IapRequestError("IAP operation failed and remote upload cleanup was unproven") from cleanup_exc
-            raise exc
+            except IapRequestError:
+                cleanup_status = "fail"
+            receipt_path, _receipt = _write_local_failure_receipt(
+                plan,
+                failure_stage=failure_stage,
+                failure_reason=failure_reason,
+                cleanup_status=cleanup_status,
+            )
+            raise IapRequestError(
+                "IAP operation failed closed "
+                f"at {failure_stage}/{failure_reason}; "
+                f"sanitized receipt: {receipt_path}"
+            ) from exc
     finally:
         _cleanup_local_ssh_key(ssh_key)
     if completed.returncode == 65:

tests/test_gcp_leoclean_nosend_iap_runner.py

@@ -159,6 +159,106 @@ def verifier_failure_result(
     }
 
 
+def operational_failure_result(plan: dict[str, Any]) -> dict[str, Any]:
+    result = verifier_failure_result(
+        plan,
+        failure_stage="database_before_restart",
+        failure_reason="database_verification_failed",
+    )
+    result["schema"] = local.REMOTE_OPERATION_FAILURE_SCHEMA
+    result["rollback"] = {
+        "status": "fail",
+        "mode": "explicit",
+        "restored_posture": "unknown",
+    }
+    result["fail_closed"] = {
+        "status": "pass",
+        "service_posture": "inactive",
+        "active_state": "inactive",
+        "sub_state": "dead",
+        "main_pid": 0,
+        "n_restarts": 0,
+        "candidate_container": "absent",
+    }
+    return result
+
+
+def successful_remote_result(plan: dict[str, Any]) -> dict[str, Any]:
+    def verification(main_pid: int, invocation: str, container: str) -> dict[str, Any]:
+        return {
+            "active_state": "active",
+            "sub_state": "running",
+            "main_pid": main_pid,
+            "n_restarts": 0,
+            "invocation_id": invocation * 32,
+            "supervising_process": {"status": "pass"},
+            "container_id": container * 64,
+            "container_pid": main_pid + 1,
+            "health": "healthy",
+            "validated_process_environment_fields": ["PATH"],
+            "dropins": "absent",
+            "environment_files": "absent",
+        }
+
+    def database(name: str, receipt_hash: str) -> dict[str, Any]:
+        return {
+            "schema": "livingip.leocleanNoSendContainerDatabaseVerification.v1",
+            "status": "pass",
+            "receipt_sha256": receipt_hash * 64,
+            "receipt_path": f"/var/lib/livingip/leoclean-nosend-iap/requests/{plan['request_id']}/{name}",
+            "proof": {
+                "proposal_staging": "function_only",
+                "proposal_transaction": "rolled_back",
+                "durable_proposal_created": False,
+                "canonical_direct_writes": "denied",
+                "stage_direct_writes": "denied",
+                "role_escalation": "denied",
+            },
+        }
+
+    before = verification(101, "2", "3")
+    if plan["operation"] == "install":
+        after = verification(103, "5", "6")
+        service_result = {
+            "name": local.SERVICE,
+            "before_restart": before,
+            "after_restart": after,
+        }
+        database_result = {
+            "status": "pass",
+            "before_restart": database("database-before-restart.json", "4"),
+            "after_restart": database("database-after-restart.json", "7"),
+        }
+    else:
+        service_result = {"name": local.SERVICE, "verification": before}
+        database_result = database("database-verify.json", "4")
+    return {
+        "schema": local.REMOTE_RESULT_SCHEMA,
+        "status": "pass",
+        "operation": plan["operation"],
+        "request_id": plan["request_id"],
+        "runner_revision": plan["runner_revision"],
+        "release_revision": plan["release_revision"],
+        "release_sha256": plan["release"]["release_sha256"],
+        "image_digest": plan["release"]["image_digest"],
+        "image_config_digest": plan["release"]["image_config_digest"],
+        "release_artifact_sha256": plan["release_artifact"]["archive_sha256"],
+        "request_binding": {
+            "archive_sha256": plan["request"]["archive_sha256"],
+            "helper_sha256": plan["request"]["helper_sha256"],
+            "manifest_sha256": plan["request"]["manifest_sha256"],
+        },
+        "service": service_result,
+        "database": database_result,
+        "authority": {
+            "sending_enabled": False,
+            "production_traffic_changed": False,
+            "vps_touched": False,
+            "canonical_proposal_approved": False,
+        },
+    }
+
+
 def test_gcloud_commands_are_fixed_iap_only_and_expire_keys(tmp_path: Path) -> None:
     scp, ssh, cleanup = fixed_commands(tmp_path)
 
@@ -183,6 +283,10 @@ def test_embedded_root_bootstrap_compiles() -> None:
     compile(local.BOOTSTRAP_CODE, "<leoclean-iap-bootstrap>", "exec")
 
 
+def test_remote_and_local_post_install_failure_contracts_match() -> None:
+    assert remote.POST_INSTALL_FAILURES == local.POST_INSTALL_FAILURES
+
+
 def test_live_origin_main_readback_is_exact() -> None:
     revision = "a" * 40
     responses = iter(
@@ -499,6 +603,125 @@ def test_local_result_receipt_is_private_and_self_hashed(tmp_path: Path) -> None
     assert json.loads(path.read_text(encoding="utf-8")) == receipt
 
 
+def test_execute_success_returns_private_receipt(
+    monkeypatch: pytest.MonkeyPatch,
+    tmp_path: Path,
+) -> None:
+    plan = executable_plan(tmp_path, "install")
+    remote_result = successful_remote_result(plan)
+    monkeypatch.setattr(local, "validate_remote_main", lambda *_args, **_kwargs: None)
+    calls = 0
+
+    def runner(argv: list[str], **_kwargs: Any) -> subprocess.CompletedProcess[bytes]:
+        nonlocal calls
+        calls += 1
+        if calls == 1:
+            return completed(argv)
+        return completed(argv, stdout=json.dumps(remote_result).encode())
+
+    execution = local.execute_plan(plan, runner=runner, gcloud_env={"PATH": "/usr/bin"})
+    receipt_path = Path(execution["receipt_path"])
+    receipt = json.loads(receipt_path.read_text(encoding="utf-8"))
+
+    assert calls == 2
+    assert execution["status"] == "pass"
+    assert stat.S_IMODE(receipt_path.stat().st_mode) == 0o600
+    assert receipt["remote_result"] == remote_result
+    assert receipt["receipt_sha256"] == execution["receipt_sha256"]
+
+
+@pytest.mark.parametrize("remote_returncode", [0, 65], ids=["success", "failure"])
+def test_execute_remote_without_output_records_absent_receipt(
+    monkeypatch: pytest.MonkeyPatch,
+    tmp_path: Path,
+    remote_returncode: int,
+) -> None:
+    plan = executable_plan(tmp_path, "verify")
+    monkeypatch.setattr(local, "validate_remote_main", lambda *_args, **_kwargs: None)
+    calls = 0
+
+    def runner(argv: list[str], **_kwargs: Any) -> subprocess.CompletedProcess[bytes]:
+        nonlocal calls
+        calls += 1
+        return completed(argv, returncode=remote_returncode if calls == 2 else 0)
+
+    with pytest.raises(local.IapRequestError, match="remote_receipt/absent"):
+        local.execute_plan(plan, runner=runner, gcloud_env={"PATH": "/usr/bin"})
+
+    receipt_path = tmp_path / "livingip-leoclean-nosend-result-iap-abcdefghijkl.json"
+    receipt = json.loads(receipt_path.read_text(encoding="utf-8"))
+    assert calls == 3
+    assert receipt["schema"] == local.LOCAL_FAILURE_RECEIPT_SCHEMA
+    assert (receipt["failure_stage"], receipt["failure_reason"]) == ("remote_receipt", "absent")
+    assert receipt["cleanup"] == {"status": "pass", "mode": "failed_transfer"}
+    assert receipt["remote_state"] == "unknown"
+
+
+def test_execute_transport_interruption_records_bounded_failure(
+    monkeypatch: pytest.MonkeyPatch,
+    tmp_path: Path,
+) -> None:
+    plan = executable_plan(tmp_path, "verify")
+    monkeypatch.setattr(local, "validate_remote_main", lambda *_args, **_kwargs: None)
+    calls = 0
+
+    def runner(argv: list[str], **_kwargs: Any) -> subprocess.CompletedProcess[bytes]:
+        nonlocal calls
+        calls += 1
+        if calls == 2:
+            raise subprocess.TimeoutExpired(argv, 300, output=b"access_token=hidden")
+        return completed(argv)
+
+    with pytest.raises(local.IapRequestError, match="iap_transport/remote_command_interrupted") as caught:
+        local.execute_plan(plan, runner=runner, gcloud_env={"PATH": "/usr/bin"})
+
+    receipt_path = tmp_path / "livingip-leoclean-nosend-result-iap-abcdefghijkl.json"
+    receipt_text = receipt_path.read_text(encoding="utf-8")
+    receipt = json.loads(receipt_text)
+    assert calls == 3
+    assert (receipt["failure_stage"], receipt["failure_reason"]) == (
+        "iap_transport",
+        "remote_command_interrupted",
+    )
+    assert "hidden" not in receipt_text
+    assert "access_token" not in receipt_text
+    assert "hidden" not in str(caught.value)
+
+
+@pytest.mark.parametrize(
+    "payload",
+    [
+        b"{not-json",
+        (b"[" * 1100) + b"0" + (b"]" * 1100),
+    ],
+    ids=["invalid-json", "recursive-json"],
+)
+def test_execute_malformed_remote_receipt_fails_closed(
+    monkeypatch: pytest.MonkeyPatch,
+    tmp_path: Path,
+    payload: bytes,
+) -> None:
+    plan = executable_plan(tmp_path, "verify")
+    monkeypatch.setattr(local, "validate_remote_main", lambda *_args, **_kwargs: None)
+    calls = 0
+
+    def runner(argv: list[str], **_kwargs: Any) -> subprocess.CompletedProcess[bytes]:
+        nonlocal calls
+        calls += 1
+        if calls == 2:
+            return completed(argv, stdout=payload)
+        return completed(argv)
+
+    with pytest.raises(local.IapRequestError, match="remote_receipt/malformed"):
+        local.execute_plan(plan, runner=runner, gcloud_env={"PATH": "/usr/bin"})
+
+    receipt_path = tmp_path / "livingip-leoclean-nosend-result-iap-abcdefghijkl.json"
+    receipt = json.loads(receipt_path.read_text(encoding="utf-8"))
+    assert calls == 3
+    assert (receipt["failure_stage"], receipt["failure_reason"]) == ("remote_receipt", "malformed")
+    assert receipt["remote_state"] == "unknown"
+
+
 def test_execute_records_sanitized_verifier_failure(
     monkeypatch: pytest.MonkeyPatch,
     tmp_path: Path,
@@ -532,6 +755,94 @@ def test_execute_records_sanitized_verifier_failure(
     assert "remote_operation_failed" not in str(caught.value)
 
 
+def test_execute_records_operational_failure_and_inert_state(
+    monkeypatch: pytest.MonkeyPatch,
+    tmp_path: Path,
+) -> None:
+    plan = executable_plan(tmp_path, "install")
+    failure = operational_failure_result(plan)
+    monkeypatch.setattr(local, "validate_remote_main", lambda *_args, **_kwargs: None)
+    calls = 0
+
+    def runner(argv: list[str], **_kwargs: Any) -> subprocess.CompletedProcess[bytes]:
+        nonlocal calls
+        calls += 1
+        if calls == 1:
+            return completed(argv)
+        return completed(argv, returncode=65, stdout=json.dumps(failure).encode())
+
+    with pytest.raises(
+        local.IapRequestError,
+        match="database_before_restart/database_verification_failed",
+    ):
+        local.execute_plan(plan, runner=runner, gcloud_env={"PATH": "/usr/bin"})
+
+    receipt_path = tmp_path / "livingip-leoclean-nosend-result-iap-abcdefghijkl.json"
+    receipt = json.loads(receipt_path.read_text(encoding="utf-8"))
+    assert calls == 2
+    assert receipt["remote_result"] == failure
+    assert receipt["remote_result"]["fail_closed"]["service_posture"] == "inactive"
+
+
+def test_operational_failure_persists_and_propagates_to_local_receipt(
+    monkeypatch: pytest.MonkeyPatch,
+    tmp_path: Path,
+    capsys: pytest.CaptureFixture[str],
+) -> None:
+    plan = executable_plan(tmp_path, "install")
+    failure = operational_failure_result(plan)
+    request_dir = private_directory(tmp_path / "remote-request")
+    payload_root = private_directory(tmp_path / "remote-payload")
+
+    def fail_execute(*_args: Any, **_kwargs: Any) -> dict[str, Any]:
+        raise remote.RemoteOperationFailure(failure)
+
+    monkeypatch.setattr(remote, "execute_request", fail_execute)
+    persisted = remote._execute_and_persist_result(
+        plan,
+        payload_root,
+        request_dir,
+        plan["request"]["archive_sha256"],
+        plan["request"]["helper_sha256"],
+        runner=lambda argv, **_kwargs: completed(argv),
+    )
+    remote_receipt_path = request_dir / "result.json"
+    assert persisted == failure
+    assert stat.S_IMODE(remote_receipt_path.stat().st_mode) == 0o600
+    assert json.loads(remote_receipt_path.read_text(encoding="utf-8")) == failure
+
+    monkeypatch.setattr(remote, "receive_request", lambda *_args: persisted)
+    previous_umask = os.umask(0o077)
+    try:
+        returncode = remote.main(["install", "request", "runner", "release", "archive", "helper"])
+    finally:
+        os.umask(previous_umask)
+    captured = capsys.readouterr()
+    assert returncode == 65
+    assert captured.err == ""
+    assert json.loads(captured.out) == failure
+
+    monkeypatch.setattr(local, "validate_remote_main", lambda *_args, **_kwargs: None)
+    calls = 0
+
+    def local_runner(argv: list[str], **_kwargs: Any) -> subprocess.CompletedProcess[bytes]:
+        nonlocal calls
+        calls += 1
+        if calls == 1:
+            return completed(argv)
+        return completed(argv, returncode=65, stdout=captured.out.encode())
+
+    with pytest.raises(local.IapRequestError, match="database_before_restart/database_verification_failed"):
+        local.execute_plan(plan, runner=local_runner, gcloud_env={"PATH": "/usr/bin"})
+
+    local_receipt_path = tmp_path / "livingip-leoclean-nosend-result-iap-abcdefghijkl.json"
+    local_receipt = json.loads(local_receipt_path.read_text(encoding="utf-8"))
+    stable = {key: value for key, value in local_receipt.items() if key != "receipt_sha256"}
+    assert local_receipt["schema"] == local.FAILURE_RECEIPT_SCHEMA
+    assert local_receipt["remote_result"] == failure
+    assert local_receipt["receipt_sha256"] == local.package.canonical_sha256(stable)
+
+
 def test_remote_failure_rejects_unallowlisted_reason(tmp_path: Path) -> None:
     plan = executable_plan(tmp_path, "install")
     failure = verifier_failure_result(plan)
@@ -541,6 +852,15 @@ def test_remote_failure_rejects_unallowlisted_reason(tmp_path: Path) -> None:
         local.validate_remote_failure(plan, failure)
 
 
+def test_operational_failure_rejects_extra_rollback_fields(tmp_path: Path) -> None:
+    plan = executable_plan(tmp_path, "install")
+    failure = operational_failure_result(plan)
+    failure["rollback"]["credential"] = "not-allowed"
+
+    with pytest.raises(local.IapRequestError, match="did not match"):
+        local.validate_remote_failure(plan, failure)
+
+
 def test_remote_main_emits_structured_failure_on_stdout(
     monkeypatch: pytest.MonkeyPatch,
     tmp_path: Path,
@@ -570,8 +890,16 @@ def test_execute_failure_does_not_forward_command_output(monkeypatch: pytest.Mon
 
     with pytest.raises(local.IapRequestError) as caught:
         local.execute_plan(plan, runner=runner, gcloud_env={"PATH": "/usr/bin"})
+    receipt_path = tmp_path / "livingip-leoclean-nosend-result-iap-abcdefghijkl.json"
+    receipt_text = receipt_path.read_text(encoding="utf-8")
+    receipt = json.loads(receipt_text)
     assert "extremely-secret" not in str(caught.value)
     assert "password" not in str(caught.value).casefold()
+    assert "extremely-secret" not in receipt_text
+    assert "password" not in receipt_text.casefold()
+    assert receipt["cleanup"] == {"status": "fail", "mode": "failed_transfer"}
+    assert "authority" not in receipt
+    assert receipt["requested_authority"] == plan["authority"]
 
 
 def test_execute_rejects_unsanitized_remote_result(monkeypatch: pytest.MonkeyPatch, tmp_path: Path) -> None:
@@ -595,7 +923,16 @@ def test_execute_rejects_unsanitized_remote_result(monkeypatch: pytest.MonkeyPat
 
     with pytest.raises(local.IapRequestError) as caught:
         local.execute_plan(plan, runner=runner, gcloud_env={"PATH": "/usr/bin"})
+    receipt_path = tmp_path / "livingip-leoclean-nosend-result-iap-abcdefghijkl.json"
+    receipt_text = receipt_path.read_text(encoding="utf-8")
+    receipt = json.loads(receipt_text)
     assert "not-allowed" not in str(caught.value)
+    assert "not-allowed" not in receipt_text
+    assert (receipt["failure_stage"], receipt["failure_reason"]) == (
+        "remote_receipt",
+        "contract_mismatch",
+    )
+    assert receipt["cleanup"] == {"status": "pass", "mode": "failed_transfer"}
 
 
 def test_execute_rederives_commands_and_rejects_plan_tampering(tmp_path: Path) -> None:
@@ -606,6 +943,22 @@ def test_execute_rederives_commands_and_rejects_plan_tampering(tmp_path: Path) -
         local.execute_plan(plan, runner=lambda argv, **_kwargs: completed(argv))
 
 
+def test_execute_rejects_authority_tampering_before_dispatch(tmp_path: Path) -> None:
+    plan = executable_plan(tmp_path, "verify")
+    plan["authority"]["credential"] = "not-allowed"
+    calls: list[list[str]] = []
+
+    with pytest.raises(local.IapRequestError, match="request authority was not exact") as caught:
+        local.execute_plan(
+            plan,
+            runner=lambda argv, **_kwargs: (calls.append(argv) or completed(argv)),
+        )
+
+    assert calls == []
+    assert "not-allowed" not in str(caught.value)
+    assert not (tmp_path / "livingip-leoclean-nosend-result-iap-abcdefghijkl.json").exists()
+
+
 @pytest.mark.parametrize("fail_pull", [False, True])
 def test_ephemeral_registry_credentials_are_always_removed(tmp_path: Path, fail_pull: bool) -> None:
     calls: list[tuple[list[str], bytes | None]] = []
@@ -838,6 +1191,7 @@ def test_post_install_proof_failure_executes_explicit_rollback(
 
     installer_module = SimpleNamespace(
         SYSTEMCTL="/usr/bin/systemctl",
+        VERIFICATION_FAILURES=local.installer.VERIFICATION_FAILURES,
         InstallPaths=Paths,
         validate_release_bundle=lambda _bundle: (release, b"unit"),
         _validate_source_binding=lambda _binding: None,
@@ -881,7 +1235,7 @@ def test_post_install_proof_failure_executes_explicit_rollback(
             return completed(argv, returncode=1)
         return completed(argv)
 
-    with pytest.raises(remote.RemoteError, match="install_failed_rolled_back"):
+    with pytest.raises(remote.RemoteOperationFailure) as caught:
         remote.execute_request(
             manifest,
             tmp_path,
@@ -891,6 +1245,180 @@ def test_post_install_proof_failure_executes_explicit_rollback(
             runner=runner,
         )
     assert rollbacks == [release["release_sha256"]]
+    expected_failure = {
+        "first_database": ("database_before_restart", "database_verification_failed"),
+        "restart": ("service_restart", "service_restart_failed"),
+        "second_database": ("database_after_restart", "database_verification_failed"),
+    }[failure_point]
+    result = caught.value.result
+    assert result["schema"] == remote.OPERATION_FAILURE_RESULT_SCHEMA
+    assert (result["failure_stage"], result["failure_reason"]) == expected_failure
+    assert result["rollback"] == {
+        "status": "pass",
+        "mode": "explicit",
+        "restored_posture": "absent",
+    }
+    assert result["fail_closed"] == {
+        "status": "not_required",
+        "service_posture": "absent",
+        "active_state": "not_observed",
+        "sub_state": "not_observed",
+        "main_pid": None,
+        "n_restarts": None,
+        "candidate_container": "not_observed",
+    }
+
+
+def test_incomplete_rollback_stops_service_and_reports_inert_posture(
+    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},
+    }
+    stops: list[str] = []
+    absence_checks: list[bool] = []
+
+    class Paths:
+        @staticmethod
+        def under(_root: Path) -> object:
+            return object()
+
+    postures = iter(
+        [
+            SimpleNamespace(
+                kind="active",
+                state={
+                    "ActiveState": "active",
+                    "SubState": "running",
+                    "MainPID": 10,
+                    "NRestarts": 0,
+                    "InvocationID": "2" * 32,
+                },
+            ),
+            SimpleNamespace(
+                kind="inactive",
+                state={
+                    "ActiveState": "inactive",
+                    "SubState": "dead",
+                    "MainPID": 0,
+                    "NRestarts": 0,
+                    "InvocationID": "",
+                },
+            ),
+        ]
+    )
+    installer_module = SimpleNamespace(
+        SYSTEMCTL="/usr/bin/systemctl",
+        InstallPaths=Paths,
+        validate_release_bundle=lambda _bundle: (release, b"unit"),
+        _validate_source_binding=lambda _binding: None,
+        install_release=lambda *_args, **_kwargs: {"status": "pass"},
+        verify_running_release=lambda *_args, **_kwargs: {
+            "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")),
+        _systemctl=lambda action, _runner: stops.append(action),
+        _stable_service_posture=lambda *_args, **_kwargs: next(postures),
+        _require_candidate_absent=lambda **_kwargs: absence_checks.append(True),
+        VERIFICATION_FAILURES=frozenset(),
+    )
+    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")),
+    )
+    monkeypatch.setattr(remote.time, "sleep", lambda _seconds: 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 stops == ["stop"]
+    assert absence_checks == [True]
+    assert result["rollback"] == {
+        "status": "fail",
+        "mode": "explicit",
+        "restored_posture": "unknown",
+    }
+    assert result["fail_closed"] == {
+        "status": "pass",
+        "service_posture": "inactive",
+        "active_state": "inactive",
+        "sub_state": "dead",
+        "main_pid": 0,
+        "n_restarts": 0,
+        "candidate_container": "absent",
+    }
+    assert result["authority"]["sending_enabled"] is False
+
+
+def test_fail_closed_stop_refuses_untrusted_unit_or_dropin() -> None:
+    stop_calls: list[str] = []
+    installer_module = SimpleNamespace(
+        _systemctl=lambda action, _runner: stop_calls.append(action),
+        _stable_service_posture=lambda *_args, **_kwargs: (_ for _ in ()).throw(
+            local.installer.InstallError("unit or drop-in drift")
+        ),
+        verify_running_release=lambda *_args, **_kwargs: {"status": "unexpected"},
+        _require_candidate_absent=lambda **_kwargs: None,
+    )
+
+    result = remote._fail_closed_inert_service(
+        installer_module=installer_module,
+        release={"release_sha256": "a" * 64},
+        paths=object(),
+        runner=lambda argv, **_kwargs: completed(argv),
+    )
+
+    assert stop_calls == []
+    assert result == {
+        "status": "fail",
+        "service_posture": "unknown",
+        "active_state": "unknown",
+        "sub_state": "unknown",
+        "main_pid": None,
+        "n_restarts": None,
+        "candidate_container": "unknown",
+    }
 
 
 @pytest.mark.parametrize(