Report no-send verifier startup failures

ops/gcp_leoclean_nosend_iap_remote.py

@@ -23,6 +23,7 @@ from typing import Any
 
 SCHEMA = "livingip.leocleanNoSendIapRequest.v1"
 RESULT_SCHEMA = "livingip.leocleanNoSendIapRemoteResult.v1"
+FAILURE_RESULT_SCHEMA = "livingip.leocleanNoSendIapFailure.v1"
 PROJECT = "teleo-501523"
 ZONE = "europe-west6-a"
 INSTANCE = "teleo-staging-1"
@@ -77,6 +78,14 @@ class RemoteError(RuntimeError):
     """A remote request failed closed without exposing command output."""
 
 
+class RemoteOperationFailure(RemoteError):
+    """A sanitized, request-bound failure that can be returned to the caller."""
+
+    def __init__(self, result: Mapping[str, Any]) -> None:
+        super().__init__("remote_operation_failed")
+        self.result = dict(result)
+
+
 def _require(condition: bool, message: str) -> None:
     if not condition:
         raise RemoteError(message)
@@ -781,6 +790,68 @@ def _rollback_result(
     }
 
 
+def _verification_failure_result(
+    *,
+    manifest: Mapping[str, Any],
+    release: Mapping[str, Any],
+    failure: Exception,
+    rollback: Mapping[str, str],
+    archive_sha256: str,
+    helper_sha256: str,
+    installer_module: Any,
+) -> dict[str, Any]:
+    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",
+    )
+    _require(
+        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"},
+        ),
+        "verification rollback summary was invalid",
+    )
+    return {
+        "schema": 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),
+        "authority": {
+            "sending_enabled": False,
+            "production_traffic_changed": False,
+            "vps_touched": False,
+            "canonical_proposal_approved": False,
+        },
+    }
+
+
+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)
+
+
 def execute_request(
     manifest: Mapping[str, Any],
     payload_root: Path,
@@ -821,18 +892,33 @@ def execute_request(
     paths = installer_module.InstallPaths.under(Path("/"))
     if operation == "install":
         pull_immutable_image(release["image"]["reference"], request_dir, runner=runner)
-        install_receipt = installer_module.install_release(
+        try:
-            bundle,
+            install_receipt = installer_module.install_release(
-            paths=paths,
+                bundle,
-            source_binding=source_binding,
+                paths=paths,
-            runner=lambda argv: _run(
+                source_binding=source_binding,
-                argv,
+                runner=lambda argv: _run(
-                runner=runner,
+                    argv,
-                label="installer command",
+                    runner=runner,
-                require_success=False,
+                    label="installer command",
-            ),
+                    require_success=False,
-            execute_restart=True,
+                ),
-        )
+                execute_restart=True,
+            )
+        except Exception as exc:
+            if not _is_verification_failure(installer_module, exc):
+                raise
+            raise RemoteOperationFailure(
+                _verification_failure_result(
+                    manifest=manifest,
+                    release=release,
+                    failure=exc,
+                    rollback={"status": "pass", "mode": "installer_automatic", "restored_posture": "prior"},
+                    archive_sha256=archive_sha256,
+                    helper_sha256=helper_sha256,
+                    installer_module=installer_module,
+                )
+            ) from exc
         _require(install_receipt.get("status") == "pass", "installation did not pass")
     elif operation == "rollback":
         rollback = _execute_explicit_rollback(
@@ -907,7 +993,7 @@ def execute_request(
                     source_binding=source_binding,
                     runner=runner,
                 )
-                _rollback_result(
+                rollback_result = _rollback_result(
                     manifest=manifest,
                     release=release,
                     rollback=recovery,
@@ -916,6 +1002,22 @@ def execute_request(
                 )
             except Exception as rollback_exc:
                 raise RemoteError("install_failed_rollback_incomplete") from rollback_exc
+            if _is_verification_failure(installer_module, exc):
+                raise RemoteOperationFailure(
+                    _verification_failure_result(
+                        manifest=manifest,
+                        release=release,
+                        failure=exc,
+                        rollback={
+                            "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
             raise RemoteError("install_failed_rolled_back") from exc
         service = {
             "name": SERVICE,
@@ -928,11 +1030,26 @@ def execute_request(
             "after_restart": after_database,
         }
     else:
-        verification = installer_module.verify_running_release(
+        try:
-            release,
+            verification = installer_module.verify_running_release(
-            runner=contract_runner,
+                release,
-            paths=paths,
+                runner=contract_runner,
-        )
+                paths=paths,
+            )
+        except Exception as exc:
+            if not _is_verification_failure(installer_module, exc):
+                raise
+            raise RemoteOperationFailure(
+                _verification_failure_result(
+                    manifest=manifest,
+                    release=release,
+                    failure=exc,
+                    rollback={"status": "not_run", "mode": "not_applicable", "restored_posture": "unchanged"},
+                    archive_sha256=archive_sha256,
+                    helper_sha256=helper_sha256,
+                    installer_module=installer_module,
+                )
+            ) from exc
         database = _verify_database(
             container_id=verification["container_id"],
             run_id=f"{manifest['request_id']}-verify",
@@ -1037,14 +1154,17 @@ def receive_request(
                 release_revision=release_revision,
             )
             verify_service_account(runner=runner)
-            result = execute_request(
+            try:
-                manifest,
+                result = execute_request(
-                payload_root,
+                    manifest,
-                request_dir,
+                    payload_root,
-                archive_sha256,
+                    request_dir,
-                helper_sha256,
+                    archive_sha256,
-                runner=runner,
+                    helper_sha256,
-            )
+                    runner=runner,
+                )
+            except RemoteOperationFailure as exc:
+                result = exc.result
             result_path = request_dir / "result.json"
             _copy_into_request(
                 result_path,
@@ -1066,7 +1186,7 @@ def main(argv: list[str] | None = None) -> int:
         _require(len(arguments) == 6, "remote arguments were not exact")
         result = receive_request(*arguments)
         sys.stdout.write(json.dumps(result, allow_nan=False, separators=(",", ":"), sort_keys=True) + "\n")
-        return 0
+        return 0 if result.get("status") == "pass" else 65
     except RemoteError as exc:
         error = str(exc)
         if error not in {"install_failed_rolled_back", "install_failed_rollback_incomplete"}:

ops/install_gcp_leoclean_nosend_release.py

@@ -110,6 +110,26 @@ FORBIDDEN_PROCESS_PREFIXES = (
     "LD_",
     "PYTHON",
 )
+VERIFICATION_FAILURES = frozenset(
+    {
+        ("service_activation", "service_readback_invalid"),
+        ("service_activation", "service_not_active"),
+        ("invocation_identity", "invocation_id_invalid"),
+        ("unit_dropins", "dropin_contract_mismatch"),
+        ("unit_configuration", "unit_contract_mismatch"),
+        ("startup_identity", "service_identity_changed"),
+        ("startup_identity", "container_id_invalid"),
+        ("process_environment", "process_environment_mismatch"),
+        ("supervising_process", "supervising_process_mismatch"),
+        ("container_startup", "container_contract_mismatch"),
+        ("container_startup", "container_not_healthy"),
+        ("image_identity", "image_contract_mismatch"),
+        ("stable_container", "container_stability_mismatch"),
+        ("stable_unit_configuration", "unit_contract_drift"),
+        ("stable_supervising_process", "supervising_process_drift"),
+        ("stable_service_identity", "service_identity_drift"),
+    }
+)
 
 Runner = Callable[[list[str]], subprocess.CompletedProcess[bytes]]
 Sleeper = Callable[[float], None]
@@ -119,6 +139,22 @@ class InstallError(RuntimeError):
     """A bounded installer failure safe to report without command output."""
 
 
+class VerificationError(InstallError):
+    """A bounded verifier failure carrying only allowlisted diagnostic codes."""
+
+    def __init__(
+        self,
+        failure_stage: str,
+        failure_reason: str,
+        message: str = "release verification failed",
+    ) -> 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
+
+
 class AtomicWriteError(InstallError):
     """An atomic publication failure retaining any published inode identity."""
 
@@ -132,6 +168,29 @@ def _require(condition: bool, message: str) -> None:
         raise InstallError(message)
 
 
+def _verification_require(
+    condition: bool,
+    failure_stage: str,
+    failure_reason: str,
+    message: str = "release verification failed",
+) -> None:
+    if not condition:
+        raise VerificationError(failure_stage, failure_reason, message)
+
+
+def _verification_check(
+    failure_stage: str,
+    failure_reason: str,
+    check: Callable[[], Any],
+) -> Any:
+    try:
+        return check()
+    except VerificationError:
+        raise
+    except InstallError as exc:
+        raise VerificationError(failure_stage, failure_reason, str(exc)) from exc
+
+
 @dataclass(frozen=True)
 class InstallPaths:
     root: Path
@@ -1060,80 +1119,180 @@ def verify_running_release(
 ) -> dict[str, Any]:
     first: dict[str, Any] | None = None
     for _attempt in range(attempts):
-        observed = _service_state(runner)
+        observed = _verification_check(
+            "service_activation",
+            "service_readback_invalid",
+            lambda: _service_state(runner),
+        )
         if observed["ActiveState"] == "active" and observed["SubState"] == "running" and observed["MainPID"] > 1:
             first = observed
             break
         sleeper(0.25)
-    _require(first is not None, "service did not become active within the bounded wait")
+    _verification_require(
-    _require(HEX_32.fullmatch(str(first["InvocationID"])) is not None, "service invocation ID was invalid")
+        first is not None,
-    _dropins_absent(paths)
+        "service_activation",
-    _validate_unit_configuration(first, paths, release)
+        "service_not_active",
-    validated_environment_fields = _validate_process_environment(first["MainPID"], paths)
+        "service did not become active within the bounded wait",
-    supervising_process = _validate_supervising_process(
-        first["MainPID"],
-        paths,
-        release,
-        str(first["ControlGroup"]),
     )
+    _verification_require(
+        HEX_32.fullmatch(str(first["InvocationID"])) is not None,
+        "invocation_identity",
+        "invocation_id_invalid",
+        "service invocation ID was invalid",
+    )
+    _verification_check(
+        "unit_dropins",
+        "dropin_contract_mismatch",
+        lambda: _dropins_absent(paths),
+    )
+    _verification_check(
+        "unit_configuration",
+        "unit_contract_mismatch",
+        lambda: _validate_unit_configuration(first, paths, release),
+    )
+    validated_environment_fields: tuple[str, ...] | None = None
+    supervising_process: dict[str, Any] | None = None
     container: dict[str, Any] | None = None
     for _attempt in range(attempts):
-        current = _service_state(runner)
+        current = _verification_check(
-        _require(
+            "startup_identity",
+            "service_identity_changed",
+            lambda: _service_state(runner),
+        )
+        _verification_require(
             current["MainPID"] == first["MainPID"]
             and current["NRestarts"] == first["NRestarts"]
             and current["InvocationID"] == first["InvocationID"]
             and current["ControlGroup"] == first["ControlGroup"]
             and current["ActiveState"] == "active"
             and current["SubState"] == "running",
+            "startup_identity",
+            "service_identity_changed",
             "service changed during startup verification",
         )
-        container_id = _read_startup_container_id(paths)
+        container_id = _verification_check(
+            "startup_identity",
+            "container_id_invalid",
+            lambda: _read_startup_container_id(paths),
+        )
         if container_id is None:
             sleeper(0.25)
             continue
-        first_container = _docker_inspect(
+        if validated_environment_fields is None:
-            ["container", "inspect", package.CONTAINER_NAME],
+            validated_environment_fields = _verification_check(
-            runner,
+                "process_environment",
-            paths,
+                "process_environment_mismatch",
-            "running container inspection",
+                lambda current=current: _validate_process_environment(current["MainPID"], paths),
+            )
+            supervising_process = _verification_check(
+                "supervising_process",
+                "supervising_process_mismatch",
+                lambda current=current: _validate_supervising_process(
+                    current["MainPID"],
+                    paths,
+                    release,
+                    str(current["ControlGroup"]),
+                ),
+            )
+        first_container = _verification_check(
+            "container_startup",
+            "container_contract_mismatch",
+            lambda: _docker_inspect(
+                ["container", "inspect", package.CONTAINER_NAME],
+                runner,
+                paths,
+                "running container inspection",
+            ),
         )
         if first_container is None:
             sleeper(0.25)
             continue
-        container = _validate_starting_container(
+        container = _verification_check(
-            first_container,
+            "container_startup",
-            release,
+            "container_contract_mismatch",
-            expected_container_id=container_id,
+            lambda first_container=first_container, container_id=container_id: _validate_starting_container(
+                first_container,
+                release,
+                expected_container_id=container_id,
+            ),
         )
         if container is not None:
             break
         sleeper(0.25)
-    _require(container is not None, "container did not become healthy within the bounded wait")
+    _verification_require(
-    _validate_local_image(release, runner, paths)
+        container is not None,
-    second_container = _docker_inspect(
+        "container_startup",
-        ["container", "inspect", container["container_id"]],
+        "container_not_healthy",
-        runner,
+        "container did not become healthy within the bounded wait",
-        paths,
-        "stable container inspection",
     )
-    _require(second_container is not None, "running container exited during verification")
+    _verification_require(
-    stable_container = _validate_starting_container(
+        validated_environment_fields is not None and supervising_process is not None,
-        second_container,
+        "startup_identity",
-        release,
+        "container_id_invalid",
-        expected_container_id=container["container_id"],
+        "container identity was not established before verification",
     )
-    _require(stable_container is not None, "running container changed during verification")
+    _verification_check(
-    second = _service_state(runner)
+        "image_identity",
-    _validate_unit_configuration(second, paths, release)
+        "image_contract_mismatch",
-    stable_supervising_process = _validate_supervising_process(
+        lambda: _validate_local_image(release, runner, paths),
-        second["MainPID"],
-        paths,
-        release,
-        str(second["ControlGroup"]),
     )
-    stable_container_id = _read_startup_container_id(paths)
+    second_container = _verification_check(
-    _require(
+        "stable_container",
+        "container_stability_mismatch",
+        lambda: _docker_inspect(
+            ["container", "inspect", container["container_id"]],
+            runner,
+            paths,
+            "stable container inspection",
+        ),
+    )
+    _verification_require(
+        second_container is not None,
+        "stable_container",
+        "container_stability_mismatch",
+        "running container exited during verification",
+    )
+    stable_container = _verification_check(
+        "stable_container",
+        "container_stability_mismatch",
+        lambda: _validate_starting_container(
+            second_container,
+            release,
+            expected_container_id=container["container_id"],
+        ),
+    )
+    _verification_require(
+        stable_container is not None,
+        "stable_container",
+        "container_stability_mismatch",
+        "running container changed during verification",
+    )
+    second = _verification_check(
+        "stable_service_identity",
+        "service_identity_drift",
+        lambda: _service_state(runner),
+    )
+    _verification_check(
+        "stable_unit_configuration",
+        "unit_contract_drift",
+        lambda: _validate_unit_configuration(second, paths, release),
+    )
+    stable_supervising_process = _verification_check(
+        "stable_supervising_process",
+        "supervising_process_drift",
+        lambda: _validate_supervising_process(
+            second["MainPID"],
+            paths,
+            release,
+            str(second["ControlGroup"]),
+        ),
+    )
+    stable_container_id = _verification_check(
+        "stable_service_identity",
+        "service_identity_drift",
+        lambda: _read_startup_container_id(paths),
+    )
+    _verification_require(
         first["MainPID"] == second["MainPID"]
         and first["NRestarts"] == second["NRestarts"]
         and first["InvocationID"] == second["InvocationID"]
@@ -1143,6 +1302,8 @@ def verify_running_release(
         and container == stable_container
         and stable_container_id == container["container_id"]
         and supervising_process == stable_supervising_process,
+        "stable_service_identity",
+        "service_identity_drift",
         "service or container changed during verification",
     )
     return {

ops/run_gcp_leoclean_nosend_iap.py

@@ -32,6 +32,7 @@ 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_FAILURE_SCHEMA = "livingip.leocleanNoSendIapFailure.v1"
 RESULT_RECEIPT_SCHEMA = "livingip.leocleanNoSendIapResultReceipt.v1"
 EXECUTION_SCHEMA = "livingip.leocleanNoSendIapExecution.v1"
 PROJECT = package.PROJECT
@@ -215,6 +216,7 @@ def _run_bounded(
     label: str,
     input_bytes: bytes | None = None,
     environment: Mapping[str, str] | None = None,
+    accepted_returncodes: tuple[int, ...] = (0,),
 ) -> subprocess.CompletedProcess[bytes]:
     try:
         completed = runner(
@@ -232,7 +234,7 @@ def _run_bounded(
         len(completed.stdout) <= MAX_COMMAND_OUTPUT_BYTES and len(completed.stderr) <= MAX_COMMAND_OUTPUT_BYTES,
         f"{label} output exceeded the safety limit",
     )
-    _require(completed.returncode == 0, f"{label} failed")
+    _require(completed.returncode in accepted_returncodes, f"{label} failed")
     return completed
 
 
@@ -1025,13 +1027,81 @@ def validate_remote_result(plan: Mapping[str, Any], result: Mapping[str, Any]) -
             )
 
 
-def _write_result_receipt(plan: Mapping[str, Any], result: Mapping[str, Any]) -> tuple[Path, dict[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")
+    _require(
+        result.get("schema") == REMOTE_FAILURE_SCHEMA
+        and 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"]
+        and result.get("release_revision") == plan["release_revision"]
+        and result.get("release_sha256") == release["release_sha256"]
+        and result.get("image_digest") == release["image_digest"]
+        and result.get("image_config_digest") == release["image_config_digest"]
+        and result.get("release_artifact_sha256") == plan["release_artifact"]["archive_sha256"]
+        and result.get("request_binding")
+        == {
+            "archive_sha256": request["archive_sha256"],
+            "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 result.get("authority")
+        == {
+            "sending_enabled": False,
+            "production_traffic_changed": False,
+            "vps_touched": False,
+            "canonical_proposal_approved": False,
+        },
+        "remote failure did not match the request",
+    )
+
+
+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"
     _require(not output.exists() and not output.is_symlink(), "local result receipt already existed")
     stable = {
         "schema": RESULT_RECEIPT_SCHEMA,
-        "status": "pass",
+        "status": status,
         "operation": plan["operation"],
         "request_id": plan["request_id"],
         "runner_revision": plan["runner_revision"],
@@ -1194,7 +1264,23 @@ def execute_plan(
     try:
         try:
             _run_bounded(list(scp), runner=runner, label="IAP upload", environment=environment)
-            completed = _run_bounded(list(ssh), runner=runner, label="IAP remote operation", environment=environment)
+            completed = _run_bounded(
+                list(ssh),
+                runner=runner,
+                label="IAP remote operation",
+                environment=environment,
+                accepted_returncodes=(0, 65),
+            )
+            result = _strict_json(completed.stdout, "remote result")
+            encoded = json.dumps(result, allow_nan=False, sort_keys=True).casefold()
+            _require(
+                not any(marker in encoded for marker in FORBIDDEN_OUTPUT_MARKERS),
+                "remote result was not sanitized",
+            )
+            if completed.returncode == 65:
+                validate_remote_failure(plan, result)
+            else:
+                validate_remote_result(plan, result)
         except IapRequestError as exc:
             try:
                 _run_bounded(
@@ -1208,10 +1294,13 @@ def execute_plan(
             raise exc
     finally:
         _cleanup_local_ssh_key(ssh_key)
-    result = _strict_json(completed.stdout, "remote result")
+    if completed.returncode == 65:
-    validate_remote_result(plan, result)
+        receipt_path, _receipt = _write_result_receipt(plan, result, status="fail")
-    encoded = json.dumps(result, allow_nan=False, sort_keys=True).casefold()
+        raise IapRequestError(
-    _require(not any(marker in encoded for marker in FORBIDDEN_OUTPUT_MARKERS), "remote result was not sanitized")
+            "IAP verifier failed closed "
+            f"at {result['failure_stage']}/{result['failure_reason']}; "
+            f"sanitized receipt: {receipt_path}"
+        )
     receipt_path, receipt = _write_result_receipt(plan, result)
     return {
         "schema": EXECUTION_SCHEMA,

tests/test_gcp_leoclean_nosend_iap_runner.py

@@ -121,6 +121,44 @@ def executable_plan(tmp_path: Path, operation: str) -> dict[str, Any]:
     }
 
 
+def verifier_failure_result(
+    plan: dict[str, Any],
+    *,
+    failure_stage: str = "process_environment",
+    failure_reason: str = "process_environment_mismatch",
+) -> dict[str, Any]:
+    return {
+        "schema": local.REMOTE_FAILURE_SCHEMA,
+        "status": "fail",
+        "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"],
+        },
+        "failure_stage": failure_stage,
+        "failure_reason": failure_reason,
+        "rollback": {
+            "status": "pass",
+            "mode": "installer_automatic",
+            "restored_posture": "prior",
+        },
+        "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)
 
@@ -451,7 +489,7 @@ def test_gcloud_environment_selects_absolute_compatible_python(
 
 def test_local_result_receipt_is_private_and_self_hashed(tmp_path: Path) -> None:
     plan = executable_plan(tmp_path, "verify")
-    result = {"bounded": "remote-result"}
+    result = {"status": "pass", "bounded": "remote-result"}
 
     path, receipt = local._write_result_receipt(plan, result)
     stable = {key: value for key, value in receipt.items() if key != "receipt_sha256"}
@@ -461,6 +499,64 @@ 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_records_sanitized_verifier_failure(
+    monkeypatch: pytest.MonkeyPatch,
+    tmp_path: Path,
+) -> None:
+    plan = executable_plan(tmp_path, "install")
+    failure = verifier_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(),
+            stderr=b'{"error":"remote_operation_failed","status":"fail"}',
+        )
+
+    with pytest.raises(local.IapRequestError, match="process_environment/process_environment_mismatch") 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 = json.loads(receipt_path.read_text(encoding="utf-8"))
+    assert calls == 2
+    assert stat.S_IMODE(receipt_path.stat().st_mode) == 0o600
+    assert receipt["status"] == "fail"
+    assert receipt["remote_result"] == failure
+    assert "remote_operation_failed" not in str(caught.value)
+
+
+def test_remote_failure_rejects_unallowlisted_reason(tmp_path: Path) -> None:
+    plan = executable_plan(tmp_path, "install")
+    failure = verifier_failure_result(plan)
+    failure["failure_reason"] = "password=not-allowlisted"
+
+    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,
+    capsys: pytest.CaptureFixture[str],
+) -> None:
+    failure = verifier_failure_result(executable_plan(tmp_path, "install"))
+    monkeypatch.setattr(remote, "receive_request", lambda *_args: failure)
+
+    returncode = remote.main(["install", "request", "runner", "release", "archive", "helper"])
+
+    captured = capsys.readouterr()
+    assert returncode == 65
+    assert json.loads(captured.out) == failure
+    assert captured.err == ""
+
+
 def test_execute_failure_does_not_forward_command_output(monkeypatch: pytest.MonkeyPatch, tmp_path: Path) -> None:
     plan = executable_plan(tmp_path, "install")
     monkeypatch.setattr(local, "validate_remote_main", lambda *_args, **_kwargs: None)
@@ -609,6 +705,80 @@ def test_rollback_fails_when_explicit_interface_is_absent(tmp_path: Path) -> Non
         )
 
 
+def test_install_verifier_failure_returns_bounded_automatic_rollback_result(
+    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},
+    }
+
+    class Paths:
+        @staticmethod
+        def under(_root: Path) -> object:
+            return object()
+
+    def fail_install(*_args: Any, **_kwargs: Any) -> dict[str, Any]:
+        raise local.installer.VerificationError(
+            "process_environment",
+            "process_environment_mismatch",
+        )
+
+    installer_module = SimpleNamespace(
+        VerificationError=local.installer.VerificationError,
+        VERIFICATION_FAILURES=local.installer.VERIFICATION_FAILURES,
+        InstallPaths=Paths,
+        validate_release_bundle=lambda _bundle: (release, b"unit"),
+        _validate_source_binding=lambda _binding: None,
+        install_release=fail_install,
+    )
+    package_module = SimpleNamespace(render_systemd_unit=lambda _release: "unit")
+    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 result["failure_stage"] == "process_environment"
+    assert result["failure_reason"] == "process_environment_mismatch"
+    assert result["rollback"] == {
+        "status": "pass",
+        "mode": "installer_automatic",
+        "restored_posture": "prior",
+    }
+    assert result["authority"]["sending_enabled"] is False
+
+
 @pytest.mark.parametrize("failure_point", ["first_database", "restart", "second_database"])
 def test_post_install_proof_failure_executes_explicit_rollback(
     monkeypatch: pytest.MonkeyPatch,

tests/test_gcp_leoclean_nosend_installer.py

@@ -522,6 +522,78 @@ def test_first_install_waits_for_complete_cid_and_healthy_container(tmp_path: Pa
     assert host.stop_calls == 0
 
 
+def test_first_install_defers_process_identity_until_release_cid_exists(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)
+    delayed_container_id: str | None = None
+    process_identity_restored = False
+
+    def runner(arguments: list[str]) -> subprocess.CompletedProcess[bytes]:
+        nonlocal delayed_container_id
+        result = host(arguments)
+        if arguments[:2] == [installer.SYSTEMCTL, "restart"]:
+            assert host.container is not None
+            delayed_container_id = str(host.container["Id"])
+            paths.cidfile.unlink()
+            process_root = paths.proc_root / str(host.main_pid)
+            (process_root / "environ").unlink()
+            (process_root / "exe").unlink()
+            (process_root / "cmdline").unlink()
+            (process_root / "cgroup").unlink()
+        return result
+
+    def sleeper(_seconds: float) -> None:
+        nonlocal process_identity_restored
+        if host.active and delayed_container_id is not None and not process_identity_restored:
+            host._write_process_environment()
+            paths.cidfile.write_text(delayed_container_id, encoding="ascii")
+            paths.cidfile.chmod(0o600)
+            process_identity_restored = True
+
+    receipt = installer.install_release(
+        bundle,
+        paths=paths,
+        source_binding=source_binding(),
+        runner=runner,
+        sleeper=sleeper,
+        execute_restart=True,
+    )
+
+    assert receipt["status"] == "pass"
+    assert process_identity_restored
+    assert host.stop_calls == 0
+
+
+def test_process_environment_failure_has_bounded_stage_and_reason(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,
+    )
+    process_root = paths.proc_root / str(host.main_pid)
+    (process_root / "environ").write_bytes(b"PATH=/unexpected\0")
+
+    with pytest.raises(installer.VerificationError) as caught:
+        installer.verify_running_release(
+            release,
+            runner=host,
+            paths=paths,
+            sleeper=lambda _seconds: None,
+        )
+
+    assert caught.value.failure_stage == "process_environment"
+    assert caught.value.failure_reason == "process_environment_mismatch"
+
+
 def test_first_install_health_timeout_rolls_back_to_absent(tmp_path: Path) -> None:
     bundle, release, image = release_bundle(tmp_path, "2")
     paths = install_paths(tmp_path)