Add GCP no-send staging preflight

.github/workflows/ci.yml

@@ -46,6 +46,7 @@ jobs:
             ops/apply_gcp_iam_split.py \
             ops/attest_gcp_reasoning_compute.py \
             ops/capture_vps_canonical_postgres_snapshot.py \
+            ops/check_gcp_leoclean_nosend_preflight.py \
             ops/check_gcp_infra_readiness.py \
             ops/run_gcp_infra_execute_canary.py \
             ops/apply_gcp_runtime_baseline.py \
@@ -102,6 +103,7 @@ jobs:
             tests/test_capture_vps_canonical_postgres_snapshot.py \
             tests/test_gcp_infra_execute_canary.py \
             tests/test_gcp_infra_readiness_checker.py \
+            tests/test_gcp_leoclean_nosend_preflight.py \
             tests/test_gcp_leoclean_runtime_permissions.py \
             tests/test_gcp_leoclean_nosend_package.py \
             tests/test_gcp_leoclean_nosend_oci_smoke.py \

docs/gcp-leoclean-nosend-preflight.md

@@ -0,0 +1,57 @@
+# GCP leoclean no-send staging preflight
+
+## Outcome
+
+This slice checks whether the existing `teleo-staging-1` target is ready for
+the separately reviewed no-send service package. It creates one revision-bound
+receipt and does not deploy the service.
+
+The checker validates the exact project, VM, machine type, attached service
+account, exact boot-disk attachment and architecture, private-only VM
+VPC/subnet attachment, Private Google Access, immutable Docker Artifact
+Registry repository, exact private PostgreSQL 16 Cloud SQL endpoint, host
+architecture, Docker server, Docker storage floor, unused service/container
+names, and private TCP/5432 reachability from the VM.
+
+## Authority boundary
+
+The control-plane commands are `describe` operations. The bounded remote probe
+runs through IAP/SSH and reads metadata, Docker, disk, systemd, container-name,
+and TCP state. It does not read Secret Manager, use a database credential, run
+a database query, install or pull software, create a container, write a unit,
+or start/restart a service.
+
+`gcloud compute ssh` may register or refresh an SSH key. The checker therefore
+uses a maximum five-minute key lifetime and records that caveat explicitly; it
+does not claim that the access path or cloud credential metadata is
+mutation-free. The remote command itself does not change host runtime state.
+
+## Run
+
+Run from the exact clean, tracked source revision with an already authenticated Google
+Cloud account that has the reviewed IAP/OS Login read path:
+
+```bash
+python3 ops/check_gcp_leoclean_nosend_preflight.py \
+  --output /tmp/leoclean-nosend-gcp-preflight.json
+```
+
+The output file is created with mode `0600`. Authentication, permission, and
+connection failures emit only a bounded category; command stderr is not copied
+into the receipt. The checker rejects an alternate repository root, an output
+path inside the source tree, untracked critical inputs, and source drift during
+the live observation.
+
+## Claim ceiling and rollback
+
+A passing receipt proves only the listed control-plane observations and
+host-network private TCP reachability at one exact, clean Git revision. It does
+not prove Docker-bridge reachability, Artifact Registry pull authorization,
+secret access, database authentication, scoped PostgreSQL permissions, image
+publication, deployment, restart, rollback, model behavior, parity, traffic,
+or production readiness.
+
+This slice changes no repository-independent service state. Its local rollback
+boundary is deletion of the branch/worktree and any generated receipt. Any
+time-bounded SSH key created by the access path expires after at most five
+minutes.

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

@@ -275,10 +275,11 @@ Cloud SQL permissions, a real Artifact Registry push, the registry's tag
 policy, or the live Artifact Registry digest contents,
 or that `teleo-staging-1` currently has a compatible container
 engine, enough disk, the expected architecture, the expected service account,
-or an unused unit name. Those are the next read-only live preflight. If the
-engine is absent, installation is separate infrastructure work; the deploy must
-stop instead of falling back to the production sync script or a mutable host
-virtual environment.
+or an unused unit name. Those are checked by the separate
+`ops/check_gcp_leoclean_nosend_preflight.py` observation receipt. If the engine
+is absent, installation is separate infrastructure work; the deploy must stop
+instead of falling back to the production sync script or a mutable host virtual
+environment.
 
 Publishing the branch and stacked PR is only a review handoff. After both the
 digest-inspection package and this receipt producer receive exact-head human

docs/workstreams/gcp-leoclean/goals.md

@@ -60,13 +60,18 @@ cut over safely and retire the VPS after soak.
   commit.
 - Current exact-head validation: 206 focused package/OCI tests and the full
   repository suite at 2,480 passed with 3 expected skips; Ruff, Python compile,
-  and diff checks pass. Two independent advisory exact-head reviews approved
+  and diff checks pass. Three independent advisory exact-head reviews approved
   the receipt deletion/identity boundary. Tests mock Docker and perform no
   registry push.
-- Current gate: publish and review the final #209-based #210 head, allow PR
-  #209 to correct PR #206's feature branch, rerun/review corrected #206, and
-  only then merge/retarget the narrow stack. Registry access remains a later,
-  explicitly authorized live gate.
+- Current gate: human exact-head review remains pending on #206 and #210. PR
+  #209 must first correct #206's feature branch; corrected #206 must be
+  re-reviewed and merged before #210 is retargeted to `main` for its own human
+  review and merge.
+- In progress offline: a separate GCP staging preflight rebuilt on the final
+  #210 head. It checks exact control-plane and host prerequisites without
+  reading secrets, logging into PostgreSQL, deploying, or restarting. Its
+  bounded IAP/SSH path uses a fixed remote command and a maximum five-minute
+  key lifetime, but no live run is currently authorized.
 - Not yet live-proven: staging VM/container engine, IAM/Secret Manager, Cloud
   SQL effective identity, restart behavior, or model/database parity.
 
@@ -76,13 +81,13 @@ cut over safely and retire the VPS after soak.
    human review, and merge the corrected PR #206.
 2. Update, review, retarget, and merge draft PR #210's structured build/push
    receipt producer and receipt-bound release-v3 finalizer.
-3. Under explicit live authorization, verify the staging repository/tag policy,
-   build and push one candidate, publish its receipt, and independently finalize
-   the exact digest.
-4. Add the separate fail-closed installer/restart/verification/rollback slice
+3. Review and merge the bounded staging preflight; under separate live
+   authorization, run its control-plane and IAP/SSH observations and verify the
+   staging repository/tag policy.
+4. Under explicit live authorization, build and push one candidate, publish
+   its receipt, and independently finalize the exact digest.
+5. Add the separate fail-closed installer/restart/verification/rollback slice
    that consumes the exact inspection-bound release bundle.
-5. Run the read-only `teleo-staging-1`
-   host/engine/service-account preflight.
 6. Provision the scoped Cloud SQL role/secret and deploy the separate staging
    service.
 7. Prove effective identity, reads, function-only proposal staging, denied
@@ -95,6 +100,8 @@ cut over safely and retire the VPS after soak.
 ## Current live-action boundary
 
 No merge without explicit human approval, Artifact Registry push,
-GCP/Cloud SQL/Secret Manager mutation,
+GCP control-plane or IAP/SSH access, Cloud SQL/Secret Manager mutation,
 service deployment/restart, Telegram change, production promotion, canonical
 proposal apply, or VPS access is part of the current offline corrective slice.
+Any later authorization for the bounded preflight must explicitly accept that
+the SSH access path may register a key with a maximum five-minute lifetime.

ops/check_gcp_leoclean_nosend_preflight.py

@@ -0,0 +1,627 @@
+#!/usr/bin/env python3
+"""Read-only control-plane and host preflight for leoclean GCP staging."""
+
+from __future__ import annotations
+
+import argparse
+import ast
+import hashlib
+import ipaddress
+import json
+import os
+import re
+import shlex
+import subprocess
+import sys
+from collections.abc import Callable
+from datetime import UTC, datetime
+from pathlib import Path
+from typing import Any
+
+if __package__:
+    from ops import gcp_leoclean_nosend_package as package
+else:
+    import gcp_leoclean_nosend_package as package
+
+SCHEMA = "livingip.leocleanNoSendGcpPreflight.v1"
+SOURCE_ROOT = Path(__file__).resolve().parents[1]
+RUNTIME_CLOUDSQL_SOURCE = SOURCE_ROOT / "hermes-agent" / "leoclean-bin" / "cloudsql_memory_tool.py"
+CRITICAL_TRACKED_PATHS = (
+    "ops/check_gcp_leoclean_nosend_preflight.py",
+    "ops/gcp_leoclean_nosend_package.py",
+    "hermes-agent/leoclean-bin/cloudsql_memory_tool.py",
+)
+REGION = package.ZONE.rsplit("-", 1)[0]
+MACHINE_TYPE = "e2-standard-2"
+MINIMUM_BOOT_DISK_GB = 80
+MINIMUM_DOCKER_FREE_BYTES = 10 * 1024**3
+MINIMUM_DOCKER_MAJOR = 24
+SSH_KEY_LIFETIME = "5m"
+NETWORK = "teleo-staging-net"
+SUBNET = "teleo-staging-subnet"
+SUBNET_CIDR = ipaddress.ip_network("10.70.0.0/24")
+CLOUDSQL_INSTANCE = "teleo-pgvector-standby"
+ARTIFACT_REPOSITORY = "teleo"
+
+Runner = Callable[[list[str]], subprocess.CompletedProcess[str]]
+HEX_40 = re.compile(r"^[0-9a-f]{40}$")
+VERSION = re.compile(r"^(\d+)\.(\d+)(?:\.(\d+))?(?:[-+].*)?$")
+SAFE_VALUE = re.compile(r"^[A-Za-z0-9@._/+:-]{1,512}$")
+
+
+class PreflightError(RuntimeError):
+    """A prerequisite is missing or the live readback is unavailable."""
+
+    def __init__(self, category: str, message: str) -> None:
+        super().__init__(message)
+        self.category = category
+
+
+def _default_runner(argv: list[str]) -> subprocess.CompletedProcess[str]:
+    environment = os.environ.copy()
+    environment["CLOUDSDK_PYTHON"] = sys.executable
+    return subprocess.run(argv, text=True, capture_output=True, check=False, timeout=60, env=environment)
+
+
+def _assert_read_only_command(argv: list[str]) -> None:
+    allowed_prefixes = (
+        ("gcloud", "projects", "describe"),
+        ("gcloud", "compute", "instances", "describe"),
+        ("gcloud", "compute", "disks", "describe"),
+        ("gcloud", "compute", "networks", "subnets", "describe"),
+        ("gcloud", "sql", "instances", "describe"),
+        ("gcloud", "artifacts", "repositories", "describe"),
+        ("gcloud", "compute", "ssh"),
+    )
+    if not any(tuple(argv[: len(prefix)]) == prefix for prefix in allowed_prefixes):
+        raise PreflightError("unsafe_command", "preflight attempted an unreviewed command")
+    if argv[:3] == ["gcloud", "compute", "ssh"]:
+        remote = [item.removeprefix("--command=") for item in argv if item.startswith("--command=")]
+        if len(remote) != 1:
+            raise PreflightError("unsafe_command", "preflight SSH command is not exact")
+        private_ip = re.search(r"/dev/tcp/([0-9.]+)/5432", remote[0])
+        if private_ip is None or argv != _ssh_argv(private_ip.group(1)):
+            raise PreflightError("unsafe_command", "preflight SSH command is not exact")
+    lowered = " ".join(argv).casefold()
+    for marker in (
+        " add-iam-policy-binding",
+        " remove-iam-policy-binding",
+        " secrets versions access",
+        " compute scp",
+        " docker run",
+        " docker pull",
+        " docker push",
+        " docker rm",
+        " systemctl start",
+        " systemctl stop",
+        " systemctl restart",
+        " systemctl enable",
+        " systemctl disable",
+        " systemctl daemon-reload",
+    ):
+        if marker in lowered:
+            raise PreflightError("unsafe_command", "preflight command contains a mutation marker")
+
+
+def _classify_command_failure(completed: subprocess.CompletedProcess[str]) -> PreflightError:
+    combined = f"{completed.stdout}\n{completed.stderr}".casefold()
+    if "reauthentication failed" in combined or "gcloud auth login" in combined:
+        return PreflightError("authentication_required", "Google Cloud authentication requires interactive refresh")
+    if "permission_denied" in combined or "does not have permission" in combined or "forbidden" in combined:
+        return PreflightError("permission_denied", "the active Google Cloud principal lacks read-only preflight access")
+    if "connection timed out" in combined or "failed to connect" in combined:
+        return PreflightError("connection_unavailable", "the bounded IAP/SSH readback was unavailable")
+    return PreflightError("command_failed", f"a read-only preflight command exited {completed.returncode}")
+
+
+def _invoke(argv: list[str], runner: Runner) -> str:
+    _assert_read_only_command(argv)
+    try:
+        completed = runner(argv)
+    except (OSError, subprocess.SubprocessError) as exc:
+        raise PreflightError("command_unavailable", f"cannot run read-only preflight: {type(exc).__name__}") from exc
+    if completed.returncode != 0:
+        raise _classify_command_failure(completed)
+    return completed.stdout
+
+
+def _json(argv: list[str], runner: Runner) -> dict[str, Any]:
+    try:
+        value = json.loads(_invoke(argv, runner))
+    except json.JSONDecodeError as exc:
+        raise PreflightError("invalid_readback", "a control-plane readback was not valid JSON") from exc
+    if not isinstance(value, dict):
+        raise PreflightError("invalid_readback", "a control-plane readback was not a JSON object")
+    return value
+
+
+def _require(condition: bool, message: str) -> None:
+    if not condition:
+        raise PreflightError("prerequisite_mismatch", message)
+
+
+def _suffix(value: object) -> str:
+    return str(value or "").rstrip("/").rsplit("/", 1)[-1]
+
+
+def _runtime_string_constant(name: str) -> str:
+    try:
+        tree = ast.parse(RUNTIME_CLOUDSQL_SOURCE.read_text(encoding="utf-8"))
+    except (OSError, SyntaxError) as exc:
+        raise PreflightError("invalid_runtime_contract", "cannot read the tracked runtime database contract") from exc
+    for node in tree.body:
+        if not isinstance(node, ast.Assign) or len(node.targets) != 1:
+            continue
+        target = node.targets[0]
+        if (
+            isinstance(target, ast.Name)
+            and target.id == name
+            and isinstance(node.value, ast.Constant)
+            and isinstance(node.value.value, str)
+        ):
+            return node.value.value
+    raise PreflightError("invalid_runtime_contract", f"runtime database contract lacks {name}")
+
+
+def _project_readback(runner: Runner) -> dict[str, Any]:
+    value = _json(
+        [
+            "gcloud",
+            "projects",
+            "describe",
+            package.PROJECT,
+            "--format=json(projectId,lifecycleState)",
+        ],
+        runner,
+    )
+    _require(value == {"lifecycleState": "ACTIVE", "projectId": package.PROJECT}, "GCP project identity is not exact")
+    return {"project_id": package.PROJECT, "lifecycle_state": "ACTIVE"}
+
+
+def _instance_readback(runner: Runner) -> dict[str, Any]:
+    value = _json(
+        [
+            "gcloud",
+            "compute",
+            "instances",
+            "describe",
+            package.INSTANCE,
+            "--project",
+            package.PROJECT,
+            "--zone",
+            package.ZONE,
+            "--format=json(name,status,zone,machineType,cpuPlatform,serviceAccounts.email,disks,networkInterfaces,tags.items)",
+        ],
+        runner,
+    )
+    accounts = value.get("serviceAccounts")
+    emails = [item.get("email") for item in accounts] if isinstance(accounts, list) else []
+    _require(value.get("name") == package.INSTANCE, "staging instance name drifted")
+    _require(value.get("status") == "RUNNING", "staging instance is not running")
+    _require(_suffix(value.get("zone")) == package.ZONE, "staging instance zone drifted")
+    _require(_suffix(value.get("machineType")) == MACHINE_TYPE, "staging instance machine type drifted")
+    _require(emails == [package.SERVICE_ACCOUNT], "staging instance service account drifted")
+    disks = value.get("disks")
+    _require(isinstance(disks, list) and len(disks) == 1, "staging instance disk attachment drifted")
+    _require(disks[0].get("boot") is True, "staging instance boot-disk attachment drifted")
+    _require(_suffix(disks[0].get("source")) == package.INSTANCE, "staging instance boot disk drifted")
+    interfaces = value.get("networkInterfaces")
+    _require(isinstance(interfaces, list) and len(interfaces) == 1, "staging instance network interface drifted")
+    interface = interfaces[0]
+    _require(_suffix(interface.get("network")) == NETWORK, "staging instance network drifted")
+    _require(_suffix(interface.get("subnetwork")) == SUBNET, "staging instance subnet drifted")
+    _require(not interface.get("accessConfigs"), "staging instance has a public network address")
+    _require(not interface.get("ipv6AccessConfigs"), "staging instance has public IPv6 access configuration")
+    _require(not interface.get("externalIpv6"), "staging instance has a public IPv6 address")
+    _require(interface.get("stackType") == "IPV4_ONLY", "staging instance IP stack drifted")
+    try:
+        network_ip = ipaddress.ip_address(str(interface.get("networkIP") or ""))
+    except ValueError as exc:
+        raise PreflightError("invalid_readback", "staging instance private address is invalid") from exc
+    _require(network_ip in SUBNET_CIDR, "staging instance private address is outside the reviewed subnet")
+    tags = value.get("tags")
+    tag_items = tags.get("items") if isinstance(tags, dict) else None
+    _require(tag_items == ["teleo-staging-ssh"], "staging instance network tag drifted")
+    return {
+        "name": package.INSTANCE,
+        "status": "RUNNING",
+        "zone": package.ZONE,
+        "machine_type": MACHINE_TYPE,
+        "cpu_platform": str(value.get("cpuPlatform") or "unknown"),
+        "service_account": package.SERVICE_ACCOUNT,
+    }
+
+
+def _disk_readback(runner: Runner) -> dict[str, Any]:
+    value = _json(
+        [
+            "gcloud",
+            "compute",
+            "disks",
+            "describe",
+            package.INSTANCE,
+            "--project",
+            package.PROJECT,
+            "--zone",
+            package.ZONE,
+            "--format=json(name,status,sizeGb,zone,architecture,users)",
+        ],
+        runner,
+    )
+    try:
+        size_gb = int(value.get("sizeGb"))
+    except (TypeError, ValueError) as exc:
+        raise PreflightError("invalid_readback", "boot disk size was not an integer") from exc
+    _require(value.get("name") == package.INSTANCE, "staging boot disk name drifted")
+    _require(value.get("status") == "READY", "staging boot disk is not ready")
+    _require(_suffix(value.get("zone")) == package.ZONE, "staging boot disk zone drifted")
+    _require(size_gb >= MINIMUM_BOOT_DISK_GB, "staging boot disk is below the reviewed minimum")
+    _require(value.get("architecture") == "X86_64", "staging boot disk architecture drifted")
+    users = value.get("users")
+    _require(
+        isinstance(users, list) and len(users) == 1 and _suffix(users[0]) == package.INSTANCE,
+        "staging boot disk attachment drifted",
+    )
+    return {
+        "name": package.INSTANCE,
+        "status": "READY",
+        "size_gb": size_gb,
+        "minimum_size_gb": MINIMUM_BOOT_DISK_GB,
+        "architecture": str(value.get("architecture") or "unspecified"),
+    }
+
+
+def _subnet_readback(runner: Runner) -> dict[str, Any]:
+    value = _json(
+        [
+            "gcloud",
+            "compute",
+            "networks",
+            "subnets",
+            "describe",
+            SUBNET,
+            "--project",
+            package.PROJECT,
+            "--region",
+            REGION,
+            "--format=json(name,region,network,ipCidrRange,privateIpGoogleAccess)",
+        ],
+        runner,
+    )
+    _require(value.get("name") == SUBNET, "staging subnet name drifted")
+    _require(_suffix(value.get("region")) == REGION, "staging subnet region drifted")
+    _require(_suffix(value.get("network")) == NETWORK, "staging subnet network drifted")
+    try:
+        subnet_cidr = ipaddress.ip_network(str(value.get("ipCidrRange") or ""), strict=True)
+    except ValueError as exc:
+        raise PreflightError("invalid_readback", "staging subnet CIDR is invalid") from exc
+    _require(subnet_cidr == SUBNET_CIDR, "staging subnet CIDR drifted")
+    _require(value.get("privateIpGoogleAccess") is True, "staging subnet lacks Private Google Access")
+    return {
+        "name": SUBNET,
+        "region": REGION,
+        "network": NETWORK,
+        "cidr": str(SUBNET_CIDR),
+        "private_google_access": True,
+    }
+
+
+def _artifact_readback(runner: Runner) -> dict[str, Any]:
+    value = _json(
+        [
+            "gcloud",
+            "artifacts",
+            "repositories",
+            "describe",
+            ARTIFACT_REPOSITORY,
+            "--project",
+            package.PROJECT,
+            "--location",
+            REGION,
+            "--format=json(name,format,mode,dockerConfig.immutableTags)",
+        ],
+        runner,
+    )
+    _require(_suffix(value.get("name")) == ARTIFACT_REPOSITORY, "Artifact Registry repository drifted")
+    _require(value.get("format") == "DOCKER", "Artifact Registry repository is not Docker")
+    _require(value.get("mode") == "STANDARD_REPOSITORY", "Artifact Registry repository mode drifted")
+    docker_config = value.get("dockerConfig")
+    _require(isinstance(docker_config, dict), "Artifact Registry Docker policy is missing")
+    _require(docker_config.get("immutableTags") is True, "Artifact Registry tags are not immutable")
+    return {"name": ARTIFACT_REPOSITORY, "region": REGION, "format": "DOCKER", "immutable_tags": True}
+
+
+def _cloudsql_readback(runner: Runner) -> tuple[dict[str, Any], str]:
+    value = _json(
+        [
+            "gcloud",
+            "sql",
+            "instances",
+            "describe",
+            CLOUDSQL_INSTANCE,
+            "--project",
+            package.PROJECT,
+            "--format=json(name,state,region,databaseVersion,ipAddresses,settings.ipConfiguration)",
+        ],
+        runner,
+    )
+    _require(value.get("name") == CLOUDSQL_INSTANCE, "Cloud SQL instance name drifted")
+    _require(value.get("state") == "RUNNABLE", "Cloud SQL instance is not runnable")
+    _require(value.get("region") == REGION, "Cloud SQL region drifted")
+    _require(value.get("databaseVersion") == "POSTGRES_16", "Cloud SQL is not exactly PostgreSQL 16")
+    settings = value.get("settings")
+    ip_config = settings.get("ipConfiguration") if isinstance(settings, dict) else None
+    _require(isinstance(ip_config, dict), "Cloud SQL IP configuration is missing")
+    _require(ip_config.get("ipv4Enabled") is False, "Cloud SQL public IPv4 is enabled")
+    _require(_suffix(ip_config.get("privateNetwork")) == NETWORK, "Cloud SQL private network drifted")
+    _require(ip_config.get("sslMode") == "ENCRYPTED_ONLY", "Cloud SQL TLS mode is incompatible with the runtime")
+    addresses = value.get("ipAddresses")
+    _require(isinstance(addresses, list) and len(addresses) == 1, "Cloud SQL address set is not private-only")
+    _require(addresses[0].get("type") == "PRIVATE", "Cloud SQL exposes a non-private address")
+    private_ips = [addresses[0].get("ipAddress")]
+    _require(isinstance(private_ips[0], str), "Cloud SQL private address is not exact")
+    try:
+        private_ip = str(ipaddress.ip_address(private_ips[0]))
+    except ValueError as exc:
+        raise PreflightError("invalid_readback", "Cloud SQL private address is invalid") from exc
+    _require(ipaddress.ip_address(private_ip).is_private, "Cloud SQL address is not private")
+    _require(
+        private_ip == _runtime_string_constant("RUNTIME_CLOUDSQL_HOST"),
+        "Cloud SQL private address drifted from the runtime contract",
+    )
+    return (
+        {
+            "name": CLOUDSQL_INSTANCE,
+            "state": "RUNNABLE",
+            "region": REGION,
+            "database_version": str(value.get("databaseVersion")),
+            "public_ipv4_enabled": False,
+            "private_network": NETWORK,
+            "ssl_mode": ip_config.get("sslMode"),
+            "private_ip_sha256": hashlib.sha256(private_ip.encode("ascii")).hexdigest(),
+        },
+        private_ip,
+    )
+
+
+def _host_script(private_ip: str) -> str:
+    _require(str(ipaddress.ip_address(private_ip)) == private_ip, "Cloud SQL private address is invalid")
+    metadata = "http://metadata.google.internal/computeMetadata/v1"
+    return f"""set -euo pipefail
+export LC_ALL=C
+export PATH=/usr/local/sbin:/usr/local/bin:/usr/sbin:/usr/bin:/sbin:/bin
+metadata() {{ curl -q --noproxy '*' --proto '=http' --max-time 5 --fail --silent --show-error --header 'Metadata-Flavor: Google' \"{metadata}/$1\"; }}
+architecture=\"$(uname -m)\"
+project=\"$(metadata project/project-id)\"
+instance=\"$(metadata instance/name)\"
+service_account=\"$(metadata instance/service-accounts/default/email)\"
+test -x /usr/bin/docker
+docker_server_version=\"$(sudo -n /usr/bin/docker version --format '{{{{.Server.Version}}}}')\"
+docker_server_os=\"$(sudo -n /usr/bin/docker version --format '{{{{.Server.Os}}}}')\"
+docker_server_arch=\"$(sudo -n /usr/bin/docker version --format '{{{{.Server.Arch}}}}')\"
+docker_root=\"$(sudo -n /usr/bin/docker info --format '{{{{.DockerRootDir}}}}')\"
+docker_root_free_bytes=\"$(sudo -n /usr/bin/df -B1 --output=avail \"$docker_root\" | tail -n 1 | tr -d ' ')\"
+unit_load_state=\"$(systemctl show {shlex.quote(package.SERVICE)} --property=LoadState --value)\"
+container_collision_count=\"$(sudo -n /usr/bin/docker ps -aq --filter 'name=^/{package.CONTAINER_NAME}$' | wc -l | tr -d ' ')\"
+if timeout 5 bash -c 'exec 3<>/dev/tcp/{private_ip}/5432; exec 3<&-; exec 3>&-'; then cloudsql_tcp=reachable; else cloudsql_tcp=unreachable; fi
+printf 'architecture=%s\n' \"$architecture\"
+printf 'project=%s\n' \"$project\"
+printf 'instance=%s\n' \"$instance\"
+printf 'service_account=%s\n' \"$service_account\"
+printf 'docker_server_version=%s\n' \"$docker_server_version\"
+printf 'docker_server_os=%s\n' \"$docker_server_os\"
+printf 'docker_server_arch=%s\n' \"$docker_server_arch\"
+printf 'docker_root_free_bytes=%s\n' \"$docker_root_free_bytes\"
+printf 'unit_load_state=%s\n' \"$unit_load_state\"
+printf 'container_collision_count=%s\n' \"$container_collision_count\"
+printf 'cloudsql_tcp=%s\n' \"$cloudsql_tcp\"
+"""
+
+
+def _ssh_argv(private_ip: str) -> list[str]:
+    return [
+        "gcloud",
+        "compute",
+        "ssh",
+        package.INSTANCE,
+        "--project",
+        package.PROJECT,
+        "--zone",
+        package.ZONE,
+        "--tunnel-through-iap",
+        "--quiet",
+        f"--ssh-key-expire-after={SSH_KEY_LIFETIME}",
+        f"--command={_host_script(private_ip)}",
+    ]
+
+
+def _parse_host_output(output: str) -> dict[str, str]:
+    values: dict[str, str] = {}
+    for line in output.splitlines():
+        key, separator, value = line.partition("=")
+        if not separator or key in values or not re.fullmatch(r"[a-z_]+", key) or not SAFE_VALUE.fullmatch(value):
+            raise PreflightError("invalid_readback", "host preflight output was not bounded key/value data")
+        values[key] = value
+    expected = {
+        "architecture",
+        "project",
+        "instance",
+        "service_account",
+        "docker_server_version",
+        "docker_server_os",
+        "docker_server_arch",
+        "docker_root_free_bytes",
+        "unit_load_state",
+        "container_collision_count",
+        "cloudsql_tcp",
+    }
+    _require(set(values) == expected, "host preflight fields are not exact")
+    return values
+
+
+def _host_readback(runner: Runner, private_ip: str) -> dict[str, Any]:
+    output = _invoke(_ssh_argv(private_ip), runner)
+    value = _parse_host_output(output)
+    version = VERSION.fullmatch(value["docker_server_version"])
+    _require(value["architecture"] == "x86_64", "staging host architecture is not x86_64")
+    _require(value["project"] == package.PROJECT, "host metadata project drifted")
+    _require(value["instance"] == package.INSTANCE, "host metadata instance drifted")
+    _require(value["service_account"] == package.SERVICE_ACCOUNT, "host metadata service account drifted")
+    _require(version is not None and int(version.group(1)) >= MINIMUM_DOCKER_MAJOR, "Docker server is below the reviewed minimum")
+    _require(value["docker_server_os"] == "linux", "Docker server OS is not Linux")
+    _require(value["docker_server_arch"] == "amd64", "Docker server architecture is not amd64")
+    try:
+        free_bytes = int(value["docker_root_free_bytes"])
+        collision_count = int(value["container_collision_count"])
+    except ValueError as exc:
+        raise PreflightError("invalid_readback", "host numeric field was invalid") from exc
+    _require(free_bytes >= MINIMUM_DOCKER_FREE_BYTES, "Docker storage has insufficient free space")
+    _require(value["unit_load_state"] == "not-found", "staging service unit name already exists")
+    _require(collision_count == 0, "staging container name already exists")
+    _require(value["cloudsql_tcp"] == "reachable", "Cloud SQL private TCP/5432 is not reachable from staging")
+    return {
+        "architecture": "x86_64",
+        "metadata_identity": {
+            "project": package.PROJECT,
+            "instance": package.INSTANCE,
+            "service_account": package.SERVICE_ACCOUNT,
+        },
+        "docker": {
+            "server_version": value["docker_server_version"],
+            "server_os": "linux",
+            "server_architecture": "amd64",
+            "minimum_major": MINIMUM_DOCKER_MAJOR,
+            "root_free_bytes": free_bytes,
+            "minimum_free_bytes": MINIMUM_DOCKER_FREE_BYTES,
+        },
+        "service_unit_load_state": "not-found",
+        "container_collision_count": 0,
+        "cloudsql_private_tcp_5432": "reachable",
+    }
+
+
+def build_preflight(*, runner: Runner = _default_runner, source_revision: str) -> dict[str, Any]:
+    _require(HEX_40.fullmatch(source_revision) is not None, "source revision must be an exact commit")
+    cloudsql, private_ip = _cloudsql_readback(runner)
+    payload = {
+        "schema": SCHEMA,
+        "status": "pass",
+        "observed_at_utc": datetime.now(UTC).isoformat(),
+        "source_revision": source_revision,
+        "target": {
+            "project": package.PROJECT,
+            "zone": package.ZONE,
+            "instance": package.INSTANCE,
+            "service": package.SERVICE,
+            "container": package.CONTAINER_NAME,
+            "platform": package.PLATFORM,
+        },
+        "control_plane": {
+            "project": _project_readback(runner),
+            "instance": _instance_readback(runner),
+            "boot_disk": _disk_readback(runner),
+            "subnet": _subnet_readback(runner),
+            "artifact_repository": _artifact_readback(runner),
+            "cloudsql": cloudsql,
+        },
+        "host": _host_readback(runner, private_ip),
+        "authority": {
+            "mode": "bounded_observation_preflight",
+            "secret_values_read": False,
+            "database_credentials_used": False,
+            "database_queries_run": False,
+            "control_plane_observations_before_ssh_are_describe_only": True,
+            "remote_host_runtime_mutations": False,
+            "ssh_access_may_register_time_bounded_key": True,
+            "ssh_key_maximum_lifetime": SSH_KEY_LIFETIME,
+        },
+        "claim_ceiling": (
+            "This receipt proves only the listed control-plane observations and host-network private TCP "
+            "reachability. It does not prove Docker-bridge reachability, Artifact Registry pull authorization, "
+            "secret access, database authentication, scoped PostgreSQL permissions, image publication, deployment, "
+            "restart, rollback, model behavior, parity, traffic, or production readiness. The IAP/SSH access path "
+            "may register a time-bounded SSH key; the remote command does not change host runtime state."
+        ),
+    }
+    return payload
+
+
+def _validated_repo_root(repo_root: Path) -> Path:
+    resolved = repo_root.expanduser().resolve()
+    if resolved != SOURCE_ROOT:
+        raise PreflightError("invalid_source_root", "preflight must run against the repository containing this script")
+    return resolved
+
+
+def _source_revision(
+    repo_root: Path,
+    *,
+    critical_paths: tuple[str, ...] = CRITICAL_TRACKED_PATHS,
+) -> str:
+    try:
+        status = subprocess.run(
+            ["git", "-C", str(repo_root), "status", "--porcelain=v1", "--untracked-files=normal"],
+            text=True,
+            capture_output=True,
+            check=False,
+            timeout=10,
+        )
+        revision = subprocess.run(
+            ["git", "-C", str(repo_root), "rev-parse", "HEAD"],
+            text=True,
+            capture_output=True,
+            check=False,
+            timeout=10,
+        )
+        tracked = subprocess.run(
+            ["git", "-C", str(repo_root), "ls-files", "--error-unmatch", "--", *critical_paths],
+            text=True,
+            capture_output=True,
+            check=False,
+            timeout=10,
+        )
+    except (OSError, subprocess.SubprocessError) as exc:
+        raise PreflightError("local_preflight_unavailable", "cannot inspect the source revision") from exc
+    if status.returncode != 0 or revision.returncode != 0 or tracked.returncode != 0:
+        raise PreflightError("local_preflight_unavailable", "cannot inspect the source revision")
+    if status.stdout.strip():
+        raise PreflightError("dirty_source_tree", "preflight requires a clean tracked source tree")
+    source_revision = revision.stdout.strip()
+    _require(HEX_40.fullmatch(source_revision) is not None, "source revision must be an exact commit")
+    return source_revision
+
+
+def main() -> int:
+    parser = argparse.ArgumentParser(description=__doc__)
+    parser.add_argument("--repo-root", type=Path, default=Path(__file__).resolve().parents[1])
+    parser.add_argument("--output", type=Path)
+    args = parser.parse_args()
+    output_path: Path | None = None
+    try:
+        repo_root = _validated_repo_root(args.repo_root)
+        if args.output:
+            output_path = args.output.expanduser().resolve()
+            try:
+                output_path.relative_to(repo_root)
+            except ValueError:
+                pass
+            else:
+                raise PreflightError("invalid_output_path", "preflight receipt must be written outside the source tree")
+        source_revision = _source_revision(repo_root)
+        payload = build_preflight(source_revision=source_revision)
+        if _source_revision(repo_root) != source_revision:
+            raise PreflightError("source_revision_changed", "source revision changed during preflight")
+    except (PreflightError, subprocess.SubprocessError, OSError) as exc:
+        category = exc.category if isinstance(exc, PreflightError) else "local_preflight_unavailable"
+        payload = {"schema": SCHEMA, "status": "blocked", "category": category}
+        print(json.dumps(payload, separators=(",", ":"), sort_keys=True))
+        return 69 if category in {"authentication_required", "permission_denied", "connection_unavailable"} else 1
+    if output_path:
+        output_path.parent.mkdir(parents=True, exist_ok=True)
+        package.write_json(output_path, payload, mode=0o600)
+    print(json.dumps(payload, indent=2, sort_keys=True))
+    return 0
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())

tests/test_gcp_leoclean_nosend_preflight.py

@@ -0,0 +1,459 @@
+"""Contracts for the read-only leoclean GCP staging preflight."""
+
+from __future__ import annotations
+
+import copy
+import hashlib
+import json
+import subprocess
+import sys
+from collections.abc import Callable
+from pathlib import Path
+
+import pytest
+
+from ops import check_gcp_leoclean_nosend_preflight as preflight
+
+REVISION = "a" * 40
+PRIVATE_IP = "10.61.0.3"
+
+
+def live_fixtures() -> dict[str, object]:
+    return {
+        "project": {"lifecycleState": "ACTIVE", "projectId": "teleo-501523"},
+        "instance": {
+            "name": "teleo-staging-1",
+            "status": "RUNNING",
+            "zone": "https://www.googleapis.com/compute/v1/projects/teleo-501523/zones/europe-west6-a",
+            "machineType": (
+                "https://www.googleapis.com/compute/v1/projects/teleo-501523/zones/europe-west6-a/"
+                "machineTypes/e2-standard-2"
+            ),
+            "cpuPlatform": "Intel Broadwell",
+            "serviceAccounts": [
+                {"email": "sa-teleo-staging-vm@teleo-501523.iam.gserviceaccount.com"}
+            ],
+            "disks": [
+                {
+                    "boot": True,
+                    "source": (
+                        "https://www.googleapis.com/compute/v1/projects/teleo-501523/zones/europe-west6-a/"
+                        "disks/teleo-staging-1"
+                    ),
+                }
+            ],
+            "networkInterfaces": [
+                {
+                    "network": (
+                        "https://www.googleapis.com/compute/v1/projects/teleo-501523/global/networks/teleo-staging-net"
+                    ),
+                    "subnetwork": (
+                        "https://www.googleapis.com/compute/v1/projects/teleo-501523/regions/europe-west6/"
+                        "subnetworks/teleo-staging-subnet"
+                    ),
+                    "networkIP": "10.70.0.2",
+                    "accessConfigs": [],
+                    "ipv6AccessConfigs": [],
+                    "stackType": "IPV4_ONLY",
+                }
+            ],
+            "tags": {"items": ["teleo-staging-ssh"]},
+        },
+        "disk": {
+            "name": "teleo-staging-1",
+            "status": "READY",
+            "sizeGb": "80",
+            "zone": "https://www.googleapis.com/compute/v1/projects/teleo-501523/zones/europe-west6-a",
+            "architecture": "X86_64",
+            "users": [
+                "https://www.googleapis.com/compute/v1/projects/teleo-501523/zones/europe-west6-a/"
+                "instances/teleo-staging-1"
+            ],
+        },
+        "subnet": {
+            "name": "teleo-staging-subnet",
+            "region": "https://www.googleapis.com/compute/v1/projects/teleo-501523/regions/europe-west6",
+            "network": "https://www.googleapis.com/compute/v1/projects/teleo-501523/global/networks/teleo-staging-net",
+            "ipCidrRange": "10.70.0.0/24",
+            "privateIpGoogleAccess": True,
+        },
+        "artifact": {
+            "name": "projects/teleo-501523/locations/europe-west6/repositories/teleo",
+            "format": "DOCKER",
+            "mode": "STANDARD_REPOSITORY",
+            "dockerConfig": {"immutableTags": True},
+        },
+        "cloudsql": {
+            "name": "teleo-pgvector-standby",
+            "state": "RUNNABLE",
+            "region": "europe-west6",
+            "databaseVersion": "POSTGRES_16",
+            "ipAddresses": [{"type": "PRIVATE", "ipAddress": PRIVATE_IP}],
+            "settings": {
+                "ipConfiguration": {
+                    "ipv4Enabled": False,
+                    "privateNetwork": "projects/teleo-501523/global/networks/teleo-staging-net",
+                    "sslMode": "ENCRYPTED_ONLY",
+                }
+            },
+        },
+        "host": {
+            "architecture": "x86_64",
+            "project": "teleo-501523",
+            "instance": "teleo-staging-1",
+            "service_account": "sa-teleo-staging-vm@teleo-501523.iam.gserviceaccount.com",
+            "docker_server_version": "29.4.0",
+            "docker_server_os": "linux",
+            "docker_server_arch": "amd64",
+            "docker_root_free_bytes": str(20 * 1024**3),
+            "unit_load_state": "not-found",
+            "container_collision_count": "0",
+            "cloudsql_tcp": "reachable",
+        },
+    }
+
+
+def fixture_runner(
+    values: dict[str, object],
+    calls: list[list[str]] | None = None,
+) -> Callable[[list[str]], subprocess.CompletedProcess[str]]:
+    def run(argv: list[str]) -> subprocess.CompletedProcess[str]:
+        if calls is not None:
+            calls.append(argv)
+        if argv[:3] == ["gcloud", "projects", "describe"]:
+            value: object = values["project"]
+        elif argv[:4] == ["gcloud", "compute", "instances", "describe"]:
+            value = values["instance"]
+        elif argv[:4] == ["gcloud", "compute", "disks", "describe"]:
+            value = values["disk"]
+        elif argv[:5] == ["gcloud", "compute", "networks", "subnets", "describe"]:
+            value = values["subnet"]
+        elif argv[:4] == ["gcloud", "artifacts", "repositories", "describe"]:
+            value = values["artifact"]
+        elif argv[:4] == ["gcloud", "sql", "instances", "describe"]:
+            value = values["cloudsql"]
+        elif argv[:3] == ["gcloud", "compute", "ssh"]:
+            host = values["host"]
+            assert isinstance(host, dict)
+            output = "".join(f"{key}={item}\n" for key, item in host.items())
+            return subprocess.CompletedProcess(argv, 0, stdout=output, stderr="")
+        else:  # pragma: no cover - the assertion shows unexpected command drift
+            raise AssertionError(argv)
+        return subprocess.CompletedProcess(argv, 0, stdout=json.dumps(value), stderr="")
+
+    return run
+
+
+def test_exact_read_only_preflight_passes_without_exposing_private_ip() -> None:
+    calls: list[list[str]] = []
+    payload = preflight.build_preflight(runner=fixture_runner(live_fixtures(), calls), source_revision=REVISION)
+
+    assert payload["status"] == "pass"
+    assert payload["source_revision"] == REVISION
+    assert payload["target"] == {
+        "project": "teleo-501523",
+        "zone": "europe-west6-a",
+        "instance": "teleo-staging-1",
+        "service": "leoclean-gcp-nosend.service",
+        "container": "livingip-leoclean-nosend",
+        "platform": "linux/amd64",
+    }
+    assert payload["authority"] == {
+        "mode": "bounded_observation_preflight",
+        "secret_values_read": False,
+        "database_credentials_used": False,
+        "database_queries_run": False,
+        "control_plane_observations_before_ssh_are_describe_only": True,
+        "remote_host_runtime_mutations": False,
+        "ssh_access_may_register_time_bounded_key": True,
+        "ssh_key_maximum_lifetime": "5m",
+    }
+    assert payload["host"]["cloudsql_private_tcp_5432"] == "reachable"
+    assert payload["control_plane"]["cloudsql"]["private_ip_sha256"] == hashlib.sha256(
+        PRIVATE_IP.encode("ascii")
+    ).hexdigest()
+    assert PRIVATE_IP not in json.dumps(payload)
+    assert len(calls) == 7
+    assert calls[:-1] == [
+        [
+            "gcloud",
+            "sql",
+            "instances",
+            "describe",
+            "teleo-pgvector-standby",
+            "--project",
+            "teleo-501523",
+            "--format=json(name,state,region,databaseVersion,ipAddresses,settings.ipConfiguration)",
+        ],
+        [
+            "gcloud",
+            "projects",
+            "describe",
+            "teleo-501523",
+            "--format=json(projectId,lifecycleState)",
+        ],
+        [
+            "gcloud",
+            "compute",
+            "instances",
+            "describe",
+            "teleo-staging-1",
+            "--project",
+            "teleo-501523",
+            "--zone",
+            "europe-west6-a",
+            "--format=json(name,status,zone,machineType,cpuPlatform,serviceAccounts.email,disks,networkInterfaces,tags.items)",
+        ],
+        [
+            "gcloud",
+            "compute",
+            "disks",
+            "describe",
+            "teleo-staging-1",
+            "--project",
+            "teleo-501523",
+            "--zone",
+            "europe-west6-a",
+            "--format=json(name,status,sizeGb,zone,architecture,users)",
+        ],
+        [
+            "gcloud",
+            "compute",
+            "networks",
+            "subnets",
+            "describe",
+            "teleo-staging-subnet",
+            "--project",
+            "teleo-501523",
+            "--region",
+            "europe-west6",
+            "--format=json(name,region,network,ipCidrRange,privateIpGoogleAccess)",
+        ],
+        [
+            "gcloud",
+            "artifacts",
+            "repositories",
+            "describe",
+            "teleo",
+            "--project",
+            "teleo-501523",
+            "--location",
+            "europe-west6",
+            "--format=json(name,format,mode,dockerConfig.immutableTags)",
+        ],
+    ]
+    ssh = calls[-1]
+    assert ssh == preflight._ssh_argv(PRIVATE_IP)
+    assert "--tunnel-through-iap" in ssh
+    assert "--ssh-key-expire-after=5m" in ssh
+    remote = next(item.removeprefix("--command=") for item in ssh if item.startswith("--command="))
+    assert remote.startswith("set -euo pipefail\n")
+    for forbidden in (
+        "docker run",
+        "docker pull",
+        "docker rm",
+        "secretmanager",
+        "secrets versions access",
+        "systemctl start",
+        "systemctl restart",
+        "systemctl enable",
+    ):
+        assert forbidden not in remote.casefold()
+
+
+@pytest.mark.parametrize(
+    ("path", "field", "unsafe", "message"),
+    [
+        (("instance", "serviceAccounts"), None, [{"email": "postgres-admin@example.com"}], "service account"),
+        (("disk",), "sizeGb", "40", "boot disk"),
+        (("disk",), "architecture", "ARM64", "architecture"),
+        (("disk",), "users", [], "attachment"),
+        (("instance", "networkInterfaces"), None, [], "network interface"),
+        (("instance", "networkInterfaces", 0), "accessConfigs", [{"natIP": "203.0.113.10"}], "public"),
+        (("instance", "networkInterfaces", 0), "ipv6AccessConfigs", [{"externalIpv6": "2001:db8::1"}], "IPv6"),
+        (("instance", "networkInterfaces", 0), "externalIpv6", "2001:db8::1", "IPv6"),
+        (("instance", "networkInterfaces", 0), "networkIP", "10.99.0.2", "reviewed subnet"),
+        (("subnet",), "privateIpGoogleAccess", False, "Private Google Access"),
+        (("subnet",), "ipCidrRange", "10.99.0.0/24", "CIDR drifted"),
+        (("artifact", "dockerConfig"), "immutableTags", False, "not immutable"),
+        (("cloudsql", "settings", "ipConfiguration"), "ipv4Enabled", True, "public IPv4"),
+        (("cloudsql",), "databaseVersion", "POSTGRES_160", "exactly PostgreSQL 16"),
+        (("host",), "container_collision_count", "1", "container name"),
+        (("host",), "unit_load_state", "loaded", "unit name"),
+        (("host",), "docker_root_free_bytes", "1024", "free space"),
+        (("host",), "cloudsql_tcp", "unreachable", "TCP/5432"),
+    ],
+)
+def test_preflight_fails_closed_on_prerequisite_drift(
+    path: tuple[str | int, ...],
+    field: str | None,
+    unsafe: object,
+    message: str,
+) -> None:
+    values = copy.deepcopy(live_fixtures())
+    target: object = values
+    for part in path:
+        if isinstance(part, int):
+            assert isinstance(target, list)
+            target = target[part]
+        else:
+            assert isinstance(target, dict)
+            target = target[part]
+    if field is None:
+        assert isinstance(target, list)
+        parent: object = values
+        for part in path[:-1]:
+            if isinstance(part, int):
+                assert isinstance(parent, list)
+                parent = parent[part]
+            else:
+                assert isinstance(parent, dict)
+                parent = parent[part]
+        assert isinstance(parent, dict)
+        last = path[-1]
+        assert isinstance(last, str)
+        parent[last] = unsafe
+    else:
+        assert isinstance(target, dict)
+        target[field] = unsafe
+
+    with pytest.raises(preflight.PreflightError, match=message):
+        preflight.build_preflight(runner=fixture_runner(values), source_revision=REVISION)
+
+
+def test_authentication_failure_is_classified_without_echoing_error() -> None:
+    sentinel = "sensitive-browser-session-sentinel"
+
+    def fail(argv: list[str]) -> subprocess.CompletedProcess[str]:
+        return subprocess.CompletedProcess(
+            argv,
+            1,
+            stdout="",
+            stderr=f"Reauthentication failed. {sentinel}. Run gcloud auth login.",
+        )
+
+    with pytest.raises(preflight.PreflightError) as caught:
+        preflight.build_preflight(runner=fail, source_revision=REVISION)
+
+    assert caught.value.category == "authentication_required"
+    assert sentinel not in str(caught.value)
+
+
+@pytest.mark.parametrize(
+    ("mutation", "message"),
+    [
+        ({"ipAddresses": [{"type": "PRIVATE", "ipAddress": "10.99.0.2"}]}, "address drifted"),
+        (
+            {
+                "ipAddresses": [
+                    {"type": "PRIVATE", "ipAddress": PRIVATE_IP},
+                    {"type": "PRIMARY", "ipAddress": "203.0.113.10"},
+                ]
+            },
+            "private-only",
+        ),
+        (
+            {"settings": {"ipConfiguration": {"sslMode": "TRUSTED_CLIENT_CERTIFICATE_REQUIRED"}}},
+            "TLS mode",
+        ),
+    ],
+)
+def test_cloudsql_endpoint_and_tls_are_exact(mutation: dict[str, object], message: str) -> None:
+    values = copy.deepcopy(live_fixtures())
+    cloudsql = values["cloudsql"]
+    assert isinstance(cloudsql, dict)
+    if "settings" in mutation:
+        settings = cloudsql["settings"]
+        assert isinstance(settings, dict)
+        ip_config = settings["ipConfiguration"]
+        assert isinstance(ip_config, dict)
+        changed = mutation["settings"]
+        assert isinstance(changed, dict)
+        changed_ip_config = changed["ipConfiguration"]
+        assert isinstance(changed_ip_config, dict)
+        ip_config.update(changed_ip_config)
+    else:
+        cloudsql.update(mutation)
+    with pytest.raises(preflight.PreflightError, match=message):
+        preflight.build_preflight(runner=fixture_runner(values), source_revision=REVISION)
+
+
+def test_host_output_rejects_extra_duplicate_or_unbounded_fields() -> None:
+    valid = live_fixtures()["host"]
+    assert isinstance(valid, dict)
+    encoded = "".join(f"{key}={item}\n" for key, item in valid.items())
+
+    for unsafe in (
+        encoded + "unexpected=value\n",
+        encoded + "project=duplicate\n",
+        encoded.replace("docker_server_os=linux", "docker_server_os=linux value with spaces"),
+    ):
+        with pytest.raises(preflight.PreflightError, match="host preflight"):
+            preflight._parse_host_output(unsafe)
+
+
+def test_command_gate_rejects_mutating_or_unreviewed_gcloud_calls() -> None:
+    for command in (
+        ["gcloud", "compute", "instances", "start", "teleo-staging-1"],
+        ["gcloud", "secrets", "versions", "access", "latest"],
+        ["gcloud", "compute", "scp", "file", "teleo-staging-1:/tmp/file"],
+        ["gcloud", "compute", "ssh", "teleo-staging-1", "--command=sudo systemctl restart x"],
+        ["gcloud", "compute", "ssh", "teleo-staging-1", "--command=touch /tmp/preflight-was-here"],
+        ["bash", "-c", "true"],
+    ):
+        with pytest.raises(preflight.PreflightError):
+            preflight._assert_read_only_command(command)
+
+
+def test_source_revision_requires_clean_tracked_commit(tmp_path: Path) -> None:
+    subprocess.run(["git", "init", "--quiet", str(tmp_path)], check=True)
+    tracked = tmp_path / "tracked.txt"
+    tracked.write_text("committed\n", encoding="utf-8")
+    subprocess.run(["git", "-C", str(tmp_path), "add", "tracked.txt"], check=True)
+    subprocess.run(
+        [
+            "git",
+            "-C",
+            str(tmp_path),
+            "-c",
+            "user.name=Preflight Test",
+            "-c",
+            "user.email=preflight@example.invalid",
+            "commit",
+            "--quiet",
+            "-m",
+            "fixture",
+        ],
+        check=True,
+    )
+    revision = subprocess.run(
+        ["git", "-C", str(tmp_path), "rev-parse", "HEAD"],
+        text=True,
+        capture_output=True,
+        check=True,
+    ).stdout.strip()
+    assert preflight._source_revision(tmp_path, critical_paths=("tracked.txt",)) == revision
+
+    (tmp_path / "untracked.txt").write_text("drift\n", encoding="utf-8")
+    with pytest.raises(preflight.PreflightError, match="clean tracked"):
+        preflight._source_revision(tmp_path, critical_paths=("tracked.txt",))
+
+
+def test_preflight_rejects_an_unrelated_clean_repository(tmp_path: Path) -> None:
+    with pytest.raises(preflight.PreflightError, match="containing this script"):
+        preflight._validated_repo_root(tmp_path)
+
+
+def test_direct_cli_imports_then_rejects_an_unrelated_repository(tmp_path: Path) -> None:
+    completed = subprocess.run(
+        [sys.executable, str(preflight.SOURCE_ROOT / "ops" / "check_gcp_leoclean_nosend_preflight.py"), "--repo-root", str(tmp_path)],
+        text=True,
+        capture_output=True,
+        check=False,
+    )
+    assert completed.returncode == 1
+    assert json.loads(completed.stdout) == {
+        "category": "invalid_source_root",
+        "schema": "livingip.leocleanNoSendGcpPreflight.v1",
+        "status": "blocked",
+    }