Merge pull request #182 from living-ip/feat/leoclean-nosend-gcp-service

Package leoclean no-send GCP staging service

.github/workflows/ci.yml

@@ -42,6 +42,7 @@ jobs:
             lib/evaluate.py \
             lib/llm.py \
             lib/post_extract.py \
+            docker/leoclean-nosend/entrypoint.py \
             ops/apply_gcp_iam_split.py \
             ops/attest_gcp_reasoning_compute.py \
             ops/capture_vps_canonical_postgres_snapshot.py \
@@ -50,6 +51,7 @@ jobs:
             ops/apply_gcp_runtime_baseline.py \
             ops/check_gcp_service_communications.py \
             ops/derive_leoclean_runtime_query_contract.py \
+            ops/gcp_leoclean_nosend_package.py \
             ops/plan_gcp_iam_split.py \
             ops/private_receipt_io.py \
             ops/run_local_canonical_postgres_rebuild.py \
@@ -63,12 +65,18 @@ jobs:
             ops/verify_postgres_parity_manifest.py \
             ops/verify_vps_canonical_snapshot_delta.py \
             telegram/approvals.py \
+            hermes-agent/patches/enforce_leoclean_learning_policy.py \
             hermes-agent/leoclean-bin/kb_tool.py \
             hermes-agent/leoclean-bin/cloudsql_memory_tool.py \
+            hermes-agent/runtime/leoclean-nosend/bootstrap.py \
+            hermes-agent/runtime/leoclean-nosend/profile-sources/cloudsql_memory_runtime.py \
+            hermes-agent/runtime/leoclean-nosend/profile-sources/leo_db_context.py \
+            hermes-agent/runtime/leoclean-nosend/profile-sources/operational_contracts.py \
             scripts/check_crabbox_ci_contract.py \
             scripts/check_llm_refinement_contract.py \
             scripts/build_working_leo_m3taversal_outcome_sandbox.py \
             scripts/leo_behavior_manifest.py \
+            scripts/leo_identity_manifest.py \
             scripts/leo_tool_trace.py \
             scripts/compile_leoclean_nosend_runtime.py \
             scripts/replay_decision_engine_eval.py \
@@ -80,6 +88,7 @@ jobs:
             scripts/run_leo_m3taversal_oos_handler_suite.py \
             scripts/verify_leo_db_first_oos_canary.py \
             scripts/run_leo_clone_bound_handler_checkpoint.py \
+            scripts/run_gcp_leoclean_nosend_oci_smoke.py \
             scripts/working_leo_m3taversal_oos_benchmark.py \
             scripts/working_leo_open_ended_benchmark.py \
             scripts/verify_leoclean_nosend_runtime.py \
@@ -94,6 +103,8 @@ jobs:
             tests/test_gcp_infra_execute_canary.py \
             tests/test_gcp_infra_readiness_checker.py \
             tests/test_gcp_leoclean_runtime_permissions.py \
+            tests/test_gcp_leoclean_nosend_package.py \
+            tests/test_gcp_leoclean_nosend_oci_smoke.py \
             tests/test_gcp_leoclean_service_environment.py \
             tests/test_gcp_runtime_baseline_apply.py \
             tests/test_gcp_service_communications.py \
@@ -213,23 +224,55 @@ jobs:
   leoclean-nosend-runtime:
     name: Leoclean no-send runtime
     runs-on: ubuntu-latest
-    timeout-minutes: 20
+    timeout-minutes: 45
     steps:
       - uses: actions/checkout@v4
       - uses: actions/setup-python@v5
         with:
           python-version: "3.11.9"
       - name: Install exact uv toolchain
-        run: python -m pip install 'uv==0.9.30'
+        run: python -m pip install 'uv==0.9.30' 'PyYAML==6.0.3'
+      - name: Pull digest-pinned OCI bases
+        run: |
+          docker pull --platform=linux/amd64 "ghcr.io/astral-sh/uv:0.9.30@sha256:538e0b39736e7feae937a65983e49d2ab75e1559d35041f9878b7b7e51de91e4"
+          docker pull --platform=linux/amd64 "docker.io/library/python:3.11.9-slim-bookworm@sha256:8fb099199b9f2d70342674bd9dbccd3ed03a258f26bbd1d556822c6dfc60c317"
+          docker pull --platform=linux/amd64 "docker.io/library/postgres:16.14-bookworm@sha256:92620daddcd947f8d5ab5ba66e848702fe443d87fed30c4cea8e389fd78dfc55"
       - name: Compile and exercise release artifact
+        env:
+          LEOCLEAN_NOSEND_ARTIFACT_OUTPUT: ${{ runner.temp }}/leoclean-nosend-artifact
         run: scripts/run_leoclean_nosend_runtime_canary.sh
+      - name: Build and exercise one disposable OCI image
+        run: |
+          python scripts/run_gcp_leoclean_nosend_oci_smoke.py \
+            --artifact "${RUNNER_TEMP}/leoclean-nosend-artifact" \
+            --runtime-receipt .crabbox-results/leoclean-nosend-runtime.json \
+            --output .crabbox-results/gcp-leoclean-nosend-oci-smoke.json
       - name: Upload no-send runtime receipt
         if: always()
         uses: actions/upload-artifact@v4
         with:
           name: teleo-infrastructure-leoclean-nosend-runtime
-          path: .crabbox-results/leoclean-nosend-runtime.json
+          path: |
+            .crabbox-results/leoclean-nosend-runtime.json
+            .crabbox-results/gcp-leoclean-nosend-oci-smoke.json
           if-no-files-found: error
+      - name: Remove interrupted no-send OCI smokes
+        if: always()
+        run: |
+          container_inventory="$(docker ps -aq --filter 'label=livingip.leoclean-nosend-smoke')"
+          if [[ -n "${container_inventory}" ]]; then
+            mapfile -t containers <<<"${container_inventory}"
+            docker rm --force "${containers[@]}"
+          fi
+          image_inventory="$(docker images -q --filter 'label=livingip.leoclean-nosend-smoke')"
+          if [[ -n "${image_inventory}" ]]; then
+            mapfile -t images <<<"${image_inventory}"
+            docker image rm --force "${images[@]}"
+          fi
+          remaining_containers="$(docker ps -aq --filter 'label=livingip.leoclean-nosend-smoke')"
+          remaining_images="$(docker images -q --filter 'label=livingip.leoclean-nosend-smoke')"
+          test -z "${remaining_containers}"
+          test -z "${remaining_images}"
 
   phase1b-local-proof:
     name: Phase 1B local proof

docker/leoclean-nosend/Dockerfile

@@ -0,0 +1,79 @@
+FROM ghcr.io/astral-sh/uv:0.9.30@sha256:538e0b39736e7feae937a65983e49d2ab75e1559d35041f9878b7b7e51de91e4 AS uv
+
+FROM docker.io/library/python:3.11.9-slim-bookworm@sha256:8fb099199b9f2d70342674bd9dbccd3ed03a258f26bbd1d556822c6dfc60c317 AS python-build
+
+COPY --from=uv /uv /usr/local/bin/uv
+COPY artifact/hermes-agent /opt/livingip/leoclean-nosend/artifact/hermes-agent
+
+RUN /usr/local/bin/uv venv \
+      --python /usr/local/bin/python3.11 \
+      --no-project \
+      /opt/livingip/leoclean-nosend/venv \
+  && VIRTUAL_ENV=/opt/livingip/leoclean-nosend/venv /usr/local/bin/uv sync \
+      --directory /opt/livingip/leoclean-nosend/artifact/hermes-agent \
+      --active \
+      --frozen \
+      --no-dev \
+      --extra cron \
+      --no-editable \
+  && /usr/local/bin/uv pip check --python /opt/livingip/leoclean-nosend/venv/bin/python
+
+FROM docker.io/library/postgres:16.14-bookworm@sha256:92620daddcd947f8d5ab5ba66e848702fe443d87fed30c4cea8e389fd78dfc55
+
+ARG TELEO_REVISION
+ARG ARTIFACT_SHA256
+ARG IDENTITY_SHA256
+ARG INPUT_SHA256
+
+LABEL org.opencontainers.image.source="https://github.com/living-ip/teleo-infrastructure" \
+      org.opencontainers.image.revision="${TELEO_REVISION}" \
+      livingip.artifact.sha256="${ARTIFACT_SHA256}" \
+      livingip.identity.sha256="${IDENTITY_SHA256}" \
+      livingip.input.sha256="${INPUT_SHA256}" \
+      livingip.runtime="leoclean-nosend" \
+      livingip.target="gcp-staging"
+
+COPY --from=python-build /usr/local /usr/local
+COPY --from=python-build /opt/livingip/leoclean-nosend/venv /opt/livingip/leoclean-nosend/venv
+COPY artifact /opt/livingip/leoclean-nosend/artifact
+COPY identity /opt/livingip/leoclean-nosend/identity
+COPY identity-sources /opt/livingip/leoclean-nosend/identity-sources
+COPY scripts /opt/livingip/leoclean-nosend/scripts
+COPY entrypoint.py /opt/livingip/leoclean-nosend/entrypoint.py
+COPY package_contract.py /opt/livingip/leoclean-nosend/package_contract.py
+COPY artifact-receipt.json /opt/livingip/leoclean-nosend/metadata/artifact-receipt.json
+COPY image-input.json /opt/livingip/leoclean-nosend/metadata/image-input.json
+COPY Dockerfile /opt/livingip/leoclean-nosend/metadata/Dockerfile
+COPY cloudsql-server-ca.pem /usr/local/libexec/livingip/leoclean-kb/cloudsql-server-ca.pem
+
+RUN ln -s /usr/local/bin/python3.11 /usr/bin/python3 \
+  && mkdir -p /usr/local/libexec/livingip/leoclean-kb/gcloud-config \
+  && chmod 0755 /usr/local/libexec/livingip/leoclean-kb \
+  && chmod 0555 /usr/local/libexec/livingip/leoclean-kb/gcloud-config \
+  && chmod 0444 /usr/local/libexec/livingip/leoclean-kb/cloudsql-server-ca.pem \
+  && chmod 0555 /opt/livingip/leoclean-nosend/entrypoint.py \
+  && chmod 0444 /opt/livingip/leoclean-nosend/package_contract.py \
+  && find /opt/livingip/leoclean-nosend/identity -type f -exec chmod 0444 {} + \
+  && case "$TELEO_REVISION" in (*[!0-9a-f]*|'') exit 65;; esac \
+  && test "${#TELEO_REVISION}" -eq 40 \
+  && for digest in "$ARTIFACT_SHA256" "$IDENTITY_SHA256" "$INPUT_SHA256"; do \
+       case "$digest" in (*[!0-9a-f]*|'') exit 65;; esac; \
+       test "${#digest}" -eq 64; \
+     done \
+  && test "$(/usr/local/bin/python3.11 -c 'import platform; print(platform.python_version())')" = "3.11.9" \
+  && /usr/bin/psql --version | grep -Eq '^psql \(PostgreSQL\) 16[.]14([[:space:]]|$)' \
+  && test "$(sha256sum /usr/local/libexec/livingip/leoclean-kb/cloudsql-server-ca.pem | cut -d ' ' -f 1)" = "80701e768f0e1f6b9d621aa0b53f6e851daaa276c6d9a8e51a300fbc015539cb" \
+  && /opt/livingip/leoclean-nosend/venv/bin/python /opt/livingip/leoclean-nosend/entrypoint.py \
+       verify-build "$TELEO_REVISION" "$ARTIFACT_SHA256" "$IDENTITY_SHA256" "$INPUT_SHA256"
+
+ENV PYTHONDONTWRITEBYTECODE=1 \
+    PYTHONNOUSERSITE=1 \
+    PYTHONSAFEPATH=1
+
+STOPSIGNAL SIGTERM
+
+HEALTHCHECK --interval=15s --timeout=10s --start-period=30s --retries=4 \
+  CMD ["/opt/livingip/leoclean-nosend/venv/bin/python", "/opt/livingip/leoclean-nosend/entrypoint.py", "health"]
+
+ENTRYPOINT ["/opt/livingip/leoclean-nosend/venv/bin/python", "/opt/livingip/leoclean-nosend/entrypoint.py"]
+CMD ["service"]

docker/leoclean-nosend/entrypoint.py

@@ -0,0 +1,301 @@
+#!/usr/bin/env python3
+"""Validate and launch the immutable leoclean no-send container."""
+
+from __future__ import annotations
+
+import ctypes
+import importlib.util
+import json
+import os
+import platform
+import shutil
+import stat
+import subprocess
+import sys
+from pathlib import Path
+from typing import Any
+
+
+def _load_adjacent_contract() -> Any:
+    path = Path(__file__).resolve().with_name("package_contract.py")
+    spec = importlib.util.spec_from_file_location("livingip_leoclean_nosend_package_contract", path)
+    if spec is None or spec.loader is None:
+        raise ImportError("the adjacent package contract cannot be loaded")
+    module = importlib.util.module_from_spec(spec)
+    spec.loader.exec_module(module)
+    return module
+
+
+contract = _load_adjacent_contract()
+
+ROOT = Path(contract.RUNTIME_ROOT)
+ARTIFACT = ROOT / "artifact"
+IDENTITY = ROOT / "identity"
+IDENTITY_SOURCES = ROOT / "identity-sources"
+PROFILE = Path(contract.PROFILE_ROOT)
+RECEIPT = ROOT / "metadata" / "artifact-receipt.json"
+IMAGE_INPUT = ROOT / "metadata" / "image-input.json"
+DOCKERFILE = ROOT / "metadata" / "Dockerfile"
+ENTRYPOINT = ROOT / "entrypoint.py"
+PACKAGE_CONTRACT = ROOT / "package_contract.py"
+CA = Path(contract.CA_DESTINATION)
+READINESS = PROFILE / "state" / "readiness.json"
+PSQL = Path("/usr/bin/psql")
+RUNTIME_PYTHON = ROOT / "venv" / "bin" / "python"
+BOOTSTRAP = ARTIFACT / "bootstrap.py"
+STATE_DIRECTORIES = ("cache", "cron", "logs", "memories", "platforms", "sessions", "state")
+
+
+class EntrypointError(RuntimeError):
+    """The installed image or live process violates its release contract."""
+
+
+def _require(condition: bool, message: str) -> None:
+    if not condition:
+        raise EntrypointError(message)
+
+
+def _validate_installed() -> dict[str, Any]:
+    image_input = contract.load_json(IMAGE_INPUT, "installed image input")
+    contract.validate_installed_runtime_identity(
+        image_input,
+        artifact=ARTIFACT,
+        receipt=RECEIPT,
+        identity=IDENTITY,
+        contract_source_root=ROOT,
+        identity_source_root=IDENTITY_SOURCES,
+    )
+    _require(contract.sha256_file(CA) == contract.CA_SHA256, "installed Cloud SQL CA drifted")
+    _require(platform.python_version() == contract.PYTHON_VERSION, "installed Python version drifted")
+    _require(Path("/usr/bin/python3").resolve() == Path("/usr/local/bin/python3.11"), "system Python binding drifted")
+    psql_version = subprocess.run(
+        [str(PSQL), "--version"],
+        check=True,
+        capture_output=True,
+        text=True,
+        timeout=10,
+    ).stdout.strip()
+    _require(
+        psql_version.startswith(f"psql (PostgreSQL) {contract.POSTGRES_VERSION}"), "installed psql version drifted"
+    )
+    bindings = image_input["source_bindings"]
+    _require(contract.sha256_file(DOCKERFILE) == bindings["Dockerfile"], "installed Dockerfile binding drifted")
+    _require(contract.sha256_file(ENTRYPOINT) == bindings["entrypoint.py"], "installed entrypoint binding drifted")
+    _require(
+        contract.sha256_file(PACKAGE_CONTRACT) == bindings["package_contract.py"],
+        "installed package contract binding drifted",
+    )
+    for relative in contract.IDENTITY_CONTRACT_SOURCES:
+        installed = ROOT / relative
+        _require(
+            contract.sha256_file(installed) == bindings[relative],
+            f"installed identity contract source drifted: {relative}",
+        )
+    return image_input
+
+
+def _prepare_profile() -> None:
+    _require(os.geteuid() == 0, "profile preparation requires the capability-limited container bootstrap user")
+    _require(PROFILE.is_dir() and not PROFILE.is_symlink(), "runtime profile tmpfs is missing or unsafe")
+    _require(not any(PROFILE.iterdir()), "runtime profile tmpfs is not empty")
+    template = ARTIFACT / "profile-template"
+    for source in template.iterdir():
+        destination = PROFILE / source.name
+        if source.is_dir():
+            shutil.copytree(source, destination, symlinks=False)
+        else:
+            shutil.copy2(source, destination, follow_symlinks=False)
+    for name in contract.RUNTIME_IDENTITY_FILES:
+        shutil.copy2(IDENTITY / name, PROFILE / name, follow_symlinks=False)
+        (PROFILE / name).chmod(0o600)
+    for name in STATE_DIRECTORIES:
+        path = PROFILE / name
+        path.mkdir(mode=0o700)
+    profile_entries = sorted(
+        PROFILE.rglob("*"),
+        key=lambda path: len(path.relative_to(PROFILE).parts),
+        reverse=True,
+    )
+    for path in profile_entries:
+        _require(not path.is_symlink(), "prepared runtime profile contains a symlink")
+    for path in profile_entries:
+        os.chown(path, contract.RUNTIME_UID, contract.RUNTIME_GID)
+    runtime_identity = PROFILE / "bin" / "gcp-runtime-identity.json"
+    os.chown(runtime_identity, 0, 0)
+    runtime_identity.chmod(0o644)
+    os.chown(PROFILE, contract.RUNTIME_UID, contract.RUNTIME_GID)
+
+
+def _drop_privileges() -> None:
+    if os.geteuid() == contract.RUNTIME_UID and os.getegid() == contract.RUNTIME_GID:
+        return
+    _require(os.geteuid() == 0, "container bootstrap user is unexpected")
+    libc = ctypes.CDLL(None, use_errno=True)
+    pr_capbset_drop = 24
+    for capability in (0, 6, 7, 8):  # CHOWN, SETGID, SETUID, SETPCAP (last)
+        if libc.prctl(pr_capbset_drop, capability, 0, 0, 0) != 0:
+            raise EntrypointError("container bootstrap capability drop failed")
+    os.setgroups([])
+    os.setgid(contract.RUNTIME_GID)
+    os.setuid(contract.RUNTIME_UID)
+    _require(os.geteuid() == contract.RUNTIME_UID and os.getegid() == contract.RUNTIME_GID, "privilege drop failed")
+
+
+def _runtime_environment() -> dict[str, str]:
+    return {
+        "HOME": str(PROFILE / "state"),
+        "HERMES_HOME": str(PROFILE),
+        "LANG": "C",
+        "LC_ALL": "C",
+        "PATH": "/usr/local/sbin:/usr/local/bin:/usr/sbin:/usr/bin:/sbin:/bin",
+        "PYTHONDONTWRITEBYTECODE": "1",
+        "PYTHONNOUSERSITE": "1",
+        "PYTHONSAFEPATH": "1",
+    }
+
+
+def _exec_runtime(command: str) -> None:
+    _prepare_profile()
+    _drop_privileges()
+    argv = [str(RUNTIME_PYTHON), str(BOOTSTRAP), command, "--profile", str(PROFILE)]
+    if command == "service":
+        argv.extend(["--readiness-file", str(READINESS)])
+    os.execve(str(RUNTIME_PYTHON), argv, _runtime_environment())
+
+
+def _validate_zero_capabilities() -> None:
+    status = Path("/proc/1/status").read_text(encoding="utf-8")
+    fields = {}
+    for line in status.splitlines():
+        key, separator, value = line.partition(":")
+        if separator:
+            fields[key] = value.strip()
+    for name in ("CapInh", "CapPrm", "CapEff", "CapBnd", "CapAmb"):
+        _require(fields.get(name) == "0000000000000000", f"runtime process retains {name}")
+    _require(fields.get("NoNewPrivs") == "1", "runtime process does not enforce no-new-privileges")
+    _require(
+        fields.get("Uid")
+        == f"{contract.RUNTIME_UID}\t{contract.RUNTIME_UID}\t{contract.RUNTIME_UID}\t{contract.RUNTIME_UID}",
+        "runtime uid drifted",
+    )
+    _require(
+        fields.get("Gid")
+        == f"{contract.RUNTIME_GID}\t{contract.RUNTIME_GID}\t{contract.RUNTIME_GID}\t{contract.RUNTIME_GID}",
+        "runtime gid drifted",
+    )
+
+
+def _validate_readiness_behavior(image_input: dict[str, Any], value: dict[str, Any]) -> None:
+    expected_config_sha256 = image_input["runtime"]["config_sha256"]
+    live_config = PROFILE / "config.yaml"
+    _require(live_config.is_file() and not live_config.is_symlink(), "runtime profile config is missing or unsafe")
+    _require(value.get("config_sha256") == expected_config_sha256, "runtime readiness config binding drifted")
+    _require(contract.sha256_file(live_config) == expected_config_sha256, "runtime profile config drifted")
+    _require(value.get("tool_surface") == contract.NO_SEND_TOOL_SURFACE, "runtime tool surface drifted")
+
+
+def _health(image_input: dict[str, Any]) -> dict[str, Any]:
+    _drop_privileges()
+    _validate_zero_capabilities()
+    _require(READINESS.is_file() and not READINESS.is_symlink(), "runtime readiness receipt is missing or unsafe")
+    _require(stat.S_IMODE(READINESS.stat().st_mode) == 0o600, "runtime readiness receipt mode drifted")
+    value = contract.load_json(READINESS, "runtime readiness receipt")
+    _require(
+        set(value)
+        == {
+            "schema",
+            "status",
+            "process_id",
+            "runtime_mode",
+            "config_sha256",
+            "gateway_adapter_count",
+            "connected_platforms",
+            "tool_surface",
+            "identity_sha256",
+        },
+        "runtime readiness fields are not exact",
+    )
+    _require(value.get("schema") == "livingip.leocleanNoSendServiceReadiness.v1", "runtime readiness schema drifted")
+    _require(
+        value.get("status") == "ready" and value.get("process_id") == 1, "runtime readiness process binding drifted"
+    )
+    _require(value.get("runtime_mode") == "gcp_staging_no_send", "runtime readiness mode drifted")
+    _require(
+        value.get("gateway_adapter_count") == 0 and value.get("connected_platforms") == [],
+        "runtime has a transport adapter",
+    )
+    _validate_readiness_behavior(image_input, value)
+    expected_identity = {name: contract.sha256_file(PROFILE / name) for name in contract.RUNTIME_IDENTITY_FILES}
+    _require(value.get("identity_sha256") == expected_identity, "runtime readiness identity binding drifted")
+    cmdline = Path("/proc/1/cmdline").read_bytes().split(b"\0")
+    expected_native = [
+        str(RUNTIME_PYTHON).encode(),
+        str(BOOTSTRAP).encode(),
+        b"service",
+        b"--profile",
+        str(PROFILE).encode(),
+        b"--readiness-file",
+        str(READINESS).encode(),
+        b"",
+    ]
+    expected_binfmt = [expected_native[0], *expected_native]
+    _require(cmdline in (expected_native, expected_binfmt), "runtime process command line drifted")
+    return {
+        "schema": "livingip.leocleanNoSendContainerHealth.v1",
+        "status": "pass",
+        "process_id": 1,
+        "gateway_adapter_count": 0,
+        "send_message_present": False,
+    }
+
+
+def _verification_result(image_input: dict[str, Any]) -> dict[str, Any]:
+    return {
+        "schema": "livingip.leocleanNoSendContainerVerification.v1",
+        "status": "pass",
+        "teleo_git_head": image_input["runtime"]["teleo_git_head"],
+        "artifact_sha256": image_input["runtime"]["artifact_sha256"],
+        "identity_sha256": image_input["identity"]["bundle_sha256"],
+        "input_sha256": image_input["input_sha256"],
+        "python_version": contract.PYTHON_VERSION,
+        "postgres_version": contract.POSTGRES_VERSION,
+        "cloudsql_ca_sha256": contract.CA_SHA256,
+    }
+
+
+def main(argv: list[str] | None = None) -> int:
+    arguments = argv if argv is not None else sys.argv[1:]
+    arguments = arguments or ["service"]
+    command = arguments[0]
+    try:
+        image_input = _validate_installed()
+        if command in {"probe", "service"}:
+            _require(len(arguments) == 1, "runtime command arguments are not exact")
+            _exec_runtime(command)
+        if command == "health":
+            _require(len(arguments) == 1, "health command arguments are not exact")
+            result = _health(image_input)
+        elif command == "verify":
+            _require(len(arguments) == 1, "verify command arguments are not exact")
+            result = _verification_result(image_input)
+        elif command == "verify-build":
+            _require(len(arguments) == 5, "build verification arguments are not exact")
+            expected = (
+                image_input["runtime"]["teleo_git_head"],
+                image_input["runtime"]["artifact_sha256"],
+                image_input["identity"]["bundle_sha256"],
+                image_input["input_sha256"],
+            )
+            _require(tuple(arguments[1:]) == expected, "image build arguments drifted from image input")
+            result = _verification_result(image_input)
+        else:
+            raise EntrypointError("unsupported container command")
+        print(json.dumps(result, separators=(",", ":"), sort_keys=True))
+        return 0
+    except (EntrypointError, contract.PackageError, OSError, subprocess.SubprocessError):
+        print(json.dumps({"status": "fail", "error": "container_contract_failed"}, sort_keys=True), file=sys.stderr)
+        return 65
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())

docs/gcp-leoclean-nosend-runtime.md

@@ -72,7 +72,13 @@ The canary:
 6. writes `.crabbox-results/leoclean-nosend-runtime.json` only after release
    verification passes.
 
-GitHub Actions runs this same wrapper and retains the receipt as a CI artifact.
+GitHub Actions runs this same wrapper, retains the verified artifact for the
+next job step, and then builds and exercises one disposable `linux/amd64` OCI
+image with a synthetic noncanonical identity. The OCI smoke proves the built-in
+zero-adapter/no-`send_message` health response and exact fail-closed entrypoint
+dispatch for an unsupported `send-message` command, then verifies cleanup. It
+does not invoke a real Hermes send tool or prove general network-egress denial.
+Both bounded receipts are retained as CI artifacts.
 
 ## Claim ceiling and next slices
 
@@ -94,3 +100,7 @@ Those remain separate rollback boundaries:
    projection, and runtime receipts; and
 6. cut over only after soak and restore testing, then revoke the VPS credentials
    and retire the VPS.
+
+The independently reversible OCI/systemd packaging contract for step 2 is
+documented in `docs/gcp-leoclean-nosend-service-package.md`. Its offline proof
+does not count as a live deployment or satisfy steps 1, 3, 4, 5, or 6.

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

@@ -0,0 +1,211 @@
+# GCP leoclean no-send service package
+
+## Outcome
+
+This slice packages the revision-bound Hermes/leoclean no-send artifact and one
+explicit identity bundle as an immutable OCI image for a separate
+`leoclean-gcp-nosend.service` on `teleo-staging-1`.
+
+It is target-architecture staging work, not a production cutover. The image is
+the atomic compatibility boundary because Python, the locked Hermes virtual
+environment, PostgreSQL client, Cloud SQL CA, runtime artifact, identity files,
+and container entrypoint must move and roll back together.
+
+## Exact target and authority boundary
+
+- project: `teleo-501523`
+- zone: `europe-west6-a`
+- instance: `teleo-staging-1`
+- service account:
+  `sa-teleo-staging-vm@teleo-501523.iam.gserviceaccount.com`
+- service: `leoclean-gcp-nosend.service`
+- platform: `linux/amd64`
+- image repository:
+  `europe-west6-docker.pkg.dev/teleo-501523/teleo/leoclean-nosend-staging`
+
+The package adds only a deployable, digest-addressed staging service surface.
+It does not add IAM, Secret Manager, database-role, proposal-approval,
+canonical-apply, transport, production, or VPS authority.
+
+The service has no provider credential in this slice. It can prove the gateway
+and database-tool boundary, but paid model calls require a separately reviewed
+internal no-send ingress and credential-delivery design during the parity
+slice.
+
+## Immutable closure
+
+The dedicated Dockerfile uses only digest-pinned base images:
+
+- Python `3.11.9`;
+- uv `0.9.30`;
+- PostgreSQL client `16.14` on Bookworm.
+
+There is no `apt-get`, floating `pip install`, editable install, or image tag in
+the closure. The final image also binds:
+
+- the exact Teleo Git revision and release-pass artifact receipt;
+- Hermes commit and source-tree hash;
+- the artifact content and manifest hashes;
+- the exact uv lock;
+- the synthetic identity manifest, lock, `SOUL.md`, structured `identity.json`,
+  original source hashes, and bundle hashes;
+- the exact no-send model, skills, tool registry, proposal-only database policy,
+  and runtime-probe receipt that produced the identity runtime contract;
+- the Cloud SQL server CA hash;
+- the Dockerfile, entrypoint, and package-contract hashes.
+
+The runtime image starts with a capability-limited bootstrap that creates a
+fresh tmpfs profile, then drops all groups, UID/GID to `65532`, and all Linux
+capability bounding sets before executing Hermes. The gateway readiness receipt
+is written only after the real zero-adapter runner starts and its sealed tool
+registry is rechecked.
+
+The PostgreSQL base image's inherited data-volume path is explicitly shadowed
+by a tiny read-only, non-executable, mode-`0000` tmpfs. This prevents Docker
+from exposing an unused writable database directory to the runtime.
+
+## Prepare a build context
+
+First compile and release-verify the no-send artifact from the exact clean
+revision, retaining it only after verification passes:
+
+```bash
+LEOCLEAN_NOSEND_ARTIFACT_OUTPUT=/tmp/leoclean-nosend-artifact \
+  scripts/run_leoclean_nosend_runtime_canary.sh
+```
+
+Derive the identity runtime contract from that artifact and receipt, generate a
+synthetic/noncanonical manifest from the committed fixture sources, and compile
+the two generated views:
+
+```bash
+python3 ops/gcp_leoclean_nosend_package.py identity-contract \
+  --artifact /tmp/leoclean-nosend-artifact \
+  --receipt .crabbox-results/leoclean-nosend-runtime.json \
+  --output /tmp/leoclean-nosend-identity-contract.json
+
+python3 ops/gcp_leoclean_nosend_package.py generate-identity-manifest \
+  --artifact /tmp/leoclean-nosend-artifact \
+  --receipt .crabbox-results/leoclean-nosend-runtime.json \
+  --database-fingerprint fixtures/working-leo/leo-identity-v1/leo-database-fingerprint-v1.json \
+  --constitution fixtures/working-leo/leo-identity-v1/leo-constitution-v1.json \
+  --database-identity fixtures/working-leo/leo-identity-v1/leo-database-identity-v1.json \
+  --identity-source-root . \
+  --output /tmp/leoclean-nosend-identity-manifest.json
+
+python3 ops/gcp_leoclean_nosend_package.py compile-identity \
+  --artifact /tmp/leoclean-nosend-artifact \
+  --receipt .crabbox-results/leoclean-nosend-runtime.json \
+  --identity-manifest /tmp/leoclean-nosend-identity-manifest.json \
+  --identity-source-root . \
+  --output /tmp/leoclean-nosend-identity
+```
+
+The compiled identity directory contains exactly:
+
+- `SOUL.md`
+- `identity.json`
+- `identity-lock.json`
+- `identity-manifest.json`
+
+Then run:
+
+```bash
+python3 ops/gcp_leoclean_nosend_package.py prepare \
+  --artifact /tmp/leoclean-nosend-artifact \
+  --receipt .crabbox-results/leoclean-nosend-runtime.json \
+  --identity /tmp/leoclean-nosend-identity \
+  --identity-source-root . \
+  --output /tmp/leoclean-nosend-image-context
+```
+
+The command fails closed on a development receipt, revision drift, content or
+mode drift, unsafe links, an inconsistent identity lock, source or generated
+view drift, a runtime/model/skills/permission mismatch, any canonical-authority
+claim, a changed CA, a non-staging target, or an existing output directory. The
+resulting context is secret-free. Direct canonical and staging-table writes
+remain denied; proposal staging remains available only through
+`kb_stage.stage_leoclean_proposal(text,text,text,text,jsonb)`.
+
+The Git gate is closure-specific: the Dockerfile, entrypoint, package contract,
+Cloud SQL CA, identity validators, and manifest-derived synthetic identity
+sources must be tracked regular files whose bytes, modes, staged state, and
+SHA-256 values match the exact `HEAD` commit. Unrelated user worktree changes
+remain permitted because they cannot enter this image context.
+
+Use the exact values in `build-args.json` as Docker build arguments. The image
+build itself revalidates that those values equal `image-input.json`.
+
+## Prepare one digest-bound release candidate
+
+After an authorized build and Artifact Registry push returns the immutable
+image digest:
+
+```bash
+python3 ops/gcp_leoclean_nosend_package.py finalize \
+  --image-input /tmp/leoclean-nosend-image-context/image-input.json \
+  --image-reference \
+    europe-west6-docker.pkg.dev/teleo-501523/teleo/leoclean-nosend-staging@sha256:<64-hex-digest> \
+  --output-release /tmp/leoclean-nosend-release.json \
+  --output-unit /tmp/leoclean-gcp-nosend.service
+```
+
+This command validates the supplied digest syntax, target repository, expected
+descriptor labels, and generated unit. It does **not** inspect Artifact
+Registry and therefore does not attest that the remote digest actually has the
+declared platform, labels, input hash, or source revision. The candidate unit
+must not be installed until a separate read-only registry-inspection receipt
+binds that exact digest to `linux/amd64`, its immutable config digest, the
+required labels, the image-input hash, and the reviewed source revision.
+
+Tags, alternate repositories, alternate projects/instances/service accounts,
+extra descriptor fields, and mismatched descriptor labels are rejected.
+
+## Offline evidence and live gate
+
+Offline tests prove descriptor strictness, artifact and synthetic identity binding,
+digest-only images, pinned runtime versions, a secret-free context, the
+capability-drop contract, no published ports, no Docker-socket mount, no
+environment file, and no production/VPS/Telegram target. CI also builds exactly
+one `linux/amd64` image, starts it with the reviewed restrictions, requires the
+built-in zero-adapter/no-`send_message` health response, checks that an
+unsupported `send-message` entrypoint command fails with the exact redacted
+error, and binds the running container to the dynamically resolved image ID.
+Before startup, the smoke also requires the built image to declare the implicit
+root bootstrap user, exact Python entrypoint, `service` command, healthcheck
+timings, and `SIGTERM` stop signal. The successful health command then proves
+PID 1 dropped to UID/GID `65532` with zero Linux capabilities. Finally, the
+smoke removes the candidate container, tag, and visible image ID. The receipt
+records whether `linux/amd64` ran natively or through cross-platform emulation.
+Pinned base images and build cache may remain.
+
+The entrypoint command rejection is not a real Hermes tool invocation, and the
+bridge-network smoke does not prove general network-egress denial. These tests
+also do not prove canonical identity, Cloud SQL permissions, 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.
+
+Publishing the draft branch and stacked PR is only a review handoff. The next
+deployment-evidence slice must add immutable Artifact Registry inspection and
+make installation fail closed without its receipt. Live installation, Artifact
+Registry publication, service restart, IAM/secret
+changes, Cloud SQL access, and proposal staging remain separate approvals.
+
+## Rollback and sunset
+
+The rollback boundary is one release descriptor, one image digest, and the
+dedicated no-send unit. Rollback restores the prior descriptor/unit and prior
+digest; it must never restart or edit the production Leo, Telegram, or VPS
+services.
+
+The fresh tmpfs profile intentionally has no session continuity. That is
+appropriate for isolated gateway/database and fresh-turn parity proof, but it
+is not the production state design. Before traffic cutover, a separate reviewed
+slice must bind durable noncanonical session state without making it identity or
+knowledge authority. The tmpfs-only state contract sunsets when that cutover
+candidate is proven.

docs/workstreams/gcp-leoclean/goals.md

@@ -0,0 +1,61 @@
+# GCP leoclean migration goal
+
+## Objective
+
+Deploy and prove Leo's least-privilege Hermes/leoclean runtime in GCP staging
+against canonical Cloud SQL, establish behavioral and database parity, connect
+the protected Observatory to canonical/proposal/vector/runtime receipts, then
+cut over safely and retire the VPS after soak.
+
+## Architecture baseline
+
+- Hermes with the `leoclean` profile remains Leo's intended runtime.
+- Cloud SQL is canonical PostgreSQL authority.
+- The protected Observatory is a separate frontend surface.
+- The VPS remains untouched until an explicitly approved, parity-proven
+  cutover and retirement sequence.
+
+## Current state
+
+- Complete: PR #179 least-privilege PostgreSQL authority is merged on main.
+- Complete: the revision-bound, structurally no-send runtime from PR #183 is
+  merged into `main` at `2d88c9765bd9b43641d1efb04a1118c5ce0bce39`.
+- Rebuilt from current `main`: PR #182 contains only the four functional
+  no-send service-package commits. The two stale proof-only commits were not
+  replayed; this file is the single replacement evidence update.
+- PR #193 is source material only. It must not be merged wholesale; only the
+  required identity-binding and OCI-lifecycle corrections belong in the narrow
+  #182/#183 stack.
+- Complete in the rebuilt code candidate: the narrow package correction binds
+  every packaged and synthetic-identity source byte to its reviewed Git blob;
+  container health binds the packaged/live config plus the exact sealed
+  no-send tool and plugin surface; and the disposable image smoke binds the
+  bootstrap user, entrypoint, command, healthcheck, and stop signal.
+- Separate follow-up: candidate release finalization does not inspect Artifact
+  Registry. A narrow deployment-evidence slice must bind the immutable remote
+  digest, `linux/amd64` manifest/config, labels, input hash, and source revision
+  before any unit can be installed.
+- Current gate: obtain a clean exact-head CI run and exact-revision human
+  review. Rebuilding or publishing the branch does not authorize merge or
+  deployment.
+- Not yet live-proven: staging VM/container engine, IAM/Secret Manager, Cloud
+  SQL effective identity, restart behavior, or model/database parity.
+
+## Remaining order
+
+1. Obtain exact-revision human review for the no-send service-package stack.
+2. Run the read-only `teleo-staging-1` host/engine/service-account preflight.
+3. Provision the scoped Cloud SQL role/secret and deploy the separate staging
+   service.
+4. Prove effective identity, reads, function-only proposal staging, denied
+   writes/escalation, receipts, restart stability, and rollback.
+5. Run behavioral and database parity backtesting.
+6. Connect the protected Observatory to canonical claims, proposals, pgvector
+   projection, and runtime receipts.
+7. Cut over, soak, restore-test, revoke VPS credentials, and retire the VPS.
+
+## Current live-action boundary
+
+No merge, Artifact Registry push, GCP/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.

hermes-agent/runtime/leoclean-nosend/bootstrap.py

@@ -15,9 +15,12 @@ import signal
 import stat
 import subprocess
 import sys
+import tempfile
 from pathlib import Path
 from typing import Any
 
+READINESS_SCHEMA = "livingip.leocleanNoSendServiceReadiness.v1"
+
 
 class NoSendContractError(RuntimeError):
     """The runtime does not satisfy its no-send contract."""
@@ -738,7 +741,58 @@ def probe(profile: Path, *, template: bool) -> dict[str, Any]:
     }
 
 
-async def _run_service(profile: Path) -> bool:
+def _write_readiness(
+    profile: Path,
+    readiness_file: Path,
+    contract: dict[str, Any],
+    runner: Any,
+    tool_surface: dict[str, Any],
+) -> None:
+    state = profile / "state"
+    _require(state.is_dir() and not state.is_symlink(), "service state directory is missing or unsafe")
+    expected_parent = state.resolve(strict=True)
+    _require(
+        readiness_file.parent.resolve(strict=True) == expected_parent, "readiness file must be inside profile state"
+    )
+    _require(readiness_file.name == "readiness.json", "readiness file name is not exact")
+    _require(not readiness_file.is_symlink(), "readiness file cannot be a symlink")
+    identity_hashes = {}
+    for name in ("SOUL.md", "identity-lock.json", "identity-manifest.json"):
+        path = profile / name
+        _require(path.is_file() and not path.is_symlink(), f"runtime identity entry is unsafe: {name}")
+        identity_hashes[name] = _sha256(path)
+    payload = {
+        "schema": READINESS_SCHEMA,
+        "status": "ready",
+        "process_id": os.getpid(),
+        "runtime_mode": contract["runtime_mode"],
+        "config_sha256": _sha256(profile / "config.yaml"),
+        "gateway_adapter_count": len(runner.adapters),
+        "connected_platforms": [],
+        "tool_surface": tool_surface,
+        "identity_sha256": identity_hashes,
+    }
+    encoded = (json.dumps(payload, separators=(",", ":"), sort_keys=True) + "\n").encode("utf-8")
+    descriptor = None
+    temporary = None
+    try:
+        descriptor, temporary = tempfile.mkstemp(prefix=".readiness.", dir=expected_parent)
+        os.fchmod(descriptor, 0o600)
+        with os.fdopen(descriptor, "wb", closefd=True) as handle:
+            descriptor = None
+            handle.write(encoded)
+            handle.flush()
+            os.fsync(handle.fileno())
+        os.replace(temporary, readiness_file)
+        temporary = None
+    finally:
+        if descriptor is not None:
+            os.close(descriptor)
+        if temporary is not None:
+            Path(temporary).unlink(missing_ok=True)
+
+
+async def _run_service(profile: Path, readiness_file: Path | None = None) -> bool:
     contract, _config, _surface, _effective_policy = _prepare(profile, template=False)
     from gateway.config import GatewayConfig
     from gateway.run import GatewayRunner
@@ -749,7 +803,7 @@ async def _run_service(profile: Path) -> bool:
     _require(getattr(runner, "_smart_model_routing", None) == {}, "service retained smart-model routing")
     _assert_tool_registry(profile, contract)
     importlib.import_module("model_tools")
-    _assert_tool_registry(profile, contract)
+    tool_surface = _assert_tool_registry(profile, contract)
 
     loop = asyncio.get_running_loop()
     stop_tasks: set[asyncio.Task[Any]] = set()
@@ -766,9 +820,15 @@ async def _run_service(profile: Path) -> bool:
             pass
     started = await runner.start()
     _require(started, "Hermes no-send gateway did not start")
-    _assert_tool_registry(profile, contract)
-    await runner.wait_for_shutdown()
-    return True
+    tool_surface = _assert_tool_registry(profile, contract)
+    if readiness_file is not None:
+        _write_readiness(profile, readiness_file, contract, runner, tool_surface)
+    try:
+        await runner.wait_for_shutdown()
+        return True
+    finally:
+        if readiness_file is not None:
+            readiness_file.unlink(missing_ok=True)
 
 
 def main() -> int:
@@ -779,12 +839,15 @@ def main() -> int:
         child.add_argument("--profile", required=True, type=Path)
         if name == "probe":
             child.add_argument("--template", action="store_true")
+        else:
+            child.add_argument("--readiness-file", type=Path)
     args = parser.parse_args()
     try:
         if args.command == "probe":
             print(json.dumps(probe(args.profile.resolve(), template=args.template), sort_keys=True))
             return 0
-        return 0 if asyncio.run(_run_service(args.profile.resolve())) else 1
+        readiness_file = args.readiness_file.absolute() if args.readiness_file else None
+        return 0 if asyncio.run(_run_service(args.profile.resolve(), readiness_file)) else 1
     except NoSendContractError as exc:
         print(json.dumps({"status": "fail", "error": str(exc)}, sort_keys=True), file=sys.stderr)
         return 65

hermes-agent/runtime/leoclean-nosend/runtime-lock.json

@@ -34,7 +34,7 @@
     {
       "source": "hermes-agent/runtime/leoclean-nosend/profile-template/config.yaml",
       "target": "config.yaml",
-      "mode": "0600"
+      "mode": "0444"
     },
     {
       "source": "hermes-agent/leoclean-bin/teleo-kb-gcp",

scripts/compile_leoclean_nosend_runtime.py

@@ -46,11 +46,7 @@ def _canonical_sha256(value: Any) -> str:
 
 
 def _is_sha256(value: Any) -> bool:
-    return (
-        isinstance(value, str)
-        and len(value) == 64
-        and all(character in "0123456789abcdef" for character in value)
-    )
+    return isinstance(value, str) and len(value) == 64 and all(character in "0123456789abcdef" for character in value)
 
 
 def _safe_relative(value: Any, label: str) -> str:
@@ -72,7 +68,10 @@ def _load_json(path: Path, label: str) -> dict[str, Any]:
 
 def _validate_lock(value: dict[str, Any]) -> None:
     _require(value.get("schema") == LOCK_SCHEMA, "unsupported runtime lock schema")
-    _require(set(value) == {"schema", "hermes", "base_images", "patches", "profile_sources"}, "runtime lock fields are invalid")
+    _require(
+        set(value) == {"schema", "hermes", "base_images", "patches", "profile_sources"},
+        "runtime lock fields are invalid",
+    )
     hermes = value.get("hermes")
     _require(isinstance(hermes, dict), "Hermes lock is missing")
     _require(
@@ -90,7 +89,10 @@ def _validate_lock(value: dict[str, Any]) -> None:
         "Hermes lock fields are invalid",
     )
     commit = hermes.get("commit")
-    _require(isinstance(commit, str) and len(commit) == 40 and all(ch in "0123456789abcdef" for ch in commit), "Hermes commit must be an exact SHA")
+    _require(
+        isinstance(commit, str) and len(commit) == 40 and all(ch in "0123456789abcdef" for ch in commit),
+        "Hermes commit must be an exact SHA",
+    )
     _require(hermes.get("repository") == "https://github.com/NousResearch/hermes-agent", "Hermes repository is invalid")
     _require(
         hermes.get("archive_url") == f"https://codeload.github.com/NousResearch/hermes-agent/tar.gz/{commit}",
@@ -108,7 +110,10 @@ def _validate_lock(value: dict[str, Any]) -> None:
     targets: set[str] = set()
     for patch in patches:
         _require(isinstance(patch, dict), "patch lock entry must be an object")
-        _require(set(patch) == {"source", "source_sha256", "target", "before_sha256", "after_sha256"}, "patch lock fields are invalid")
+        _require(
+            set(patch) == {"source", "source_sha256", "target", "before_sha256", "after_sha256"},
+            "patch lock fields are invalid",
+        )
         _safe_relative(patch["source"], "patch source")
         target = _safe_relative(patch["target"], "patch target")
         _require(isinstance(target, str) and target not in targets, "patch targets must be unique")
@@ -119,12 +124,15 @@ def _validate_lock(value: dict[str, Any]) -> None:
     _require(isinstance(sources, list) and bool(sources), "profile sources are missing")
     profile_targets: set[str] = set()
     for source in sources:
-        _require(isinstance(source, dict) and set(source) == {"source", "target", "mode"}, "profile source fields are invalid")
+        _require(
+            isinstance(source, dict) and set(source) == {"source", "target", "mode"},
+            "profile source fields are invalid",
+        )
         _safe_relative(source["source"], "profile source")
         target = _safe_relative(source["target"], "profile target")
         _require(target not in profile_targets, "duplicate profile target")
         profile_targets.add(target)
-        _require(source["mode"] in {"0600", "0644", "0755"}, "profile mode is invalid")
+        _require(source["mode"] in {"0444", "0600", "0644", "0755"}, "profile mode is invalid")
 
 
 def _download(url: str, target: Path, timeout: float) -> None:
@@ -190,7 +198,9 @@ def _apply_patches(repo_root: Path, hermes_root: Path, lock: dict[str, Any]) ->
         patcher = _bound_file(repo_root, entry["source"], f"managed patcher {entry['source']}")
         target = _bound_file(hermes_root, entry["target"], f"Hermes patch target {entry['target']}")
         _require(_sha256(patcher) == entry["source_sha256"], f"managed patcher drifted: {entry['source']}")
-        _require(_sha256(target) == entry["before_sha256"], f"Hermes patch target drifted before apply: {entry['target']}")
+        _require(
+            _sha256(target) == entry["before_sha256"], f"Hermes patch target drifted before apply: {entry['target']}"
+        )
         module = _load_patcher(patcher)
         result, code = module.install(target, check=False)
         _require(code == 0, f"managed patch failed: {entry['source']}")
@@ -219,6 +229,7 @@ def _copy_profile(repo_root: Path, artifact: Path, lock: dict[str, Any]) -> None
         target.parent.mkdir(parents=True, exist_ok=True)
         shutil.copyfile(source, target)
         target.chmod(int(entry["mode"], 8))
+    profile.chmod(0o555)
 
 
 def _tree_manifest(root: Path, *, excluded: frozenset[str] = frozenset()) -> dict[str, Any]:
@@ -304,8 +315,12 @@ def compile_artifact(
         extracted = staging / "extracted"
         extracted.mkdir()
         hermes_source = _safe_extract(archive_path, extracted, lock["hermes"]["archive_root"])
-        _require(_sha256(hermes_source / "pyproject.toml") == lock["hermes"]["pyproject_sha256"], "Hermes pyproject drifted")
-        _require(_sha256(hermes_source / "uv.lock") == lock["hermes"]["uv_lock_sha256"], "Hermes dependency lock drifted")
+        _require(
+            _sha256(hermes_source / "pyproject.toml") == lock["hermes"]["pyproject_sha256"], "Hermes pyproject drifted"
+        )
+        _require(
+            _sha256(hermes_source / "uv.lock") == lock["hermes"]["uv_lock_sha256"], "Hermes dependency lock drifted"
+        )
         patch_receipts = _apply_patches(repo_root, hermes_source, lock)
         patched_source = _tree_manifest(hermes_source)
         _require(

scripts/leo_identity_manifest.py

@@ -4,6 +4,7 @@
 from __future__ import annotations
 
 import argparse
+import copy
 import json
 import re
 import sys
@@ -35,6 +36,24 @@ IDENTITY_TABLES = frozenset(
         "public.claims",
     }
 )
+LEOCLEAN_NO_SEND_TOOL_SURFACE = {
+    "toolset": "livingip-leoclean-nosend",
+    "registry_sealed": True,
+    "send_message_present": False,
+    "terminal_handler": "livingip.restricted_teleo_kb.v1",
+    "terminal_restricted_to": "profile/bin/teleo-kb",
+    "tools": ["skill_view", "skills_list", "terminal"],
+    "plugins": [
+        {
+            "name": "leo-db-context",
+            "source": "user",
+            "tools": [],
+            "hooks": ["post_llm_call", "pre_llm_call"],
+            "enabled": True,
+            "error": None,
+        }
+    ],
+}
 
 
 def expected_runtime_policy() -> dict[str, Any]:
@@ -45,6 +64,35 @@ def expected_runtime_policy() -> dict[str, Any]:
     }
 
 
+def expected_no_send_runtime_policy() -> dict[str, Any]:
+    return {
+        "schema": "livingip.leocleanNoSendPermissionReceipt.v1",
+        "messaging_send": {
+            "capability": "absent",
+            "transport_authority": "absent",
+            "gateway_adapters": "none",
+            "send_message_tool": "absent",
+        },
+        "general_network_egress": {
+            "container_mode": "bridge",
+            "denial_enforced": False,
+            "denial_tested": False,
+        },
+        "database": {
+            "direct_canonical_writes": "denied",
+            "direct_stage_table_writes": "denied",
+            "proposal_staging": "function_only",
+            "proposal_function": "kb_stage.stage_leoclean_proposal(text,text,text,text,jsonb)",
+            "live_effective_permissions_proven": False,
+        },
+        "allowed_tools": ["skill_view", "skills_list", "terminal"],
+    }
+
+
+def _valid_runtime_policy(value: Any) -> bool:
+    return value in (expected_runtime_policy(), expected_no_send_runtime_policy())
+
+
 def expected_session_boundary() -> dict[str, Any]:
     return {
         "classification": "temporary_noncanonical",
@@ -426,16 +474,239 @@ def render_identity_views(
     }
 
 
+def identity_runtime_inputs_from_contract(value: dict[str, Any]) -> dict[str, Any]:
+    """Adapt one package-validated synthetic runtime contract to identity inputs."""
+
+    _require(
+        set(value) == {"schema", "runtime", "model", "skills", "permissions", "behavior_inputs", "contract_sha256"},
+        "identity runtime contract fields are invalid",
+    )
+    _require(
+        value.get("schema") == "livingip.leocleanNoSendIdentityRuntimeContract.v1",
+        "identity runtime contract schema is invalid",
+    )
+    stable = {key: item for key, item in value.items() if key != "contract_sha256"}
+    _require(
+        value.get("contract_sha256") == behavior.canonical_sha256(stable),
+        "identity runtime contract hash drift detected",
+    )
+    for section in ("runtime", "model", "skills", "permissions", "behavior_inputs"):
+        _require(isinstance(value.get(section), dict), f"identity runtime contract {section} is missing")
+    runtime = value["runtime"]
+    _require(
+        set(runtime)
+        == {
+            "identity_runtime_sha256",
+            "hermes_git_head",
+            "hermes_source_sha256",
+            "teleo_git_head",
+            "teleo_source_sha256",
+            "python",
+        },
+        "identity runtime contract runtime fields are invalid",
+    )
+    teleo_git_head = runtime.get("teleo_git_head")
+    _require(bool(HEX_40.fullmatch(str(teleo_git_head or ""))), "identity runtime contract Teleo revision is invalid")
+    _require(bool(HEX_40.fullmatch(str(runtime.get("hermes_git_head") or ""))), "identity runtime Hermes revision is invalid")
+    for field in ("identity_runtime_sha256", "hermes_source_sha256", "teleo_source_sha256"):
+        _require(_is_sha256(runtime.get(field)), f"identity runtime contract {field} is invalid")
+    python_runtime = runtime.get("python")
+    _require(
+        isinstance(python_runtime, dict)
+        and set(python_runtime)
+        == {
+            "implementation",
+            "python_version",
+            "abi_tag",
+            "runtime_distributions",
+            "runtime_distributions_sha256",
+        }
+        and python_runtime.get("implementation") == "CPython"
+        and bool(python_runtime.get("python_version"))
+        and bool(python_runtime.get("abi_tag"))
+        and isinstance(python_runtime.get("runtime_distributions"), dict)
+        and bool(python_runtime["runtime_distributions"])
+        and python_runtime.get("runtime_distributions_sha256")
+        == behavior.canonical_sha256(python_runtime["runtime_distributions"]),
+        "identity runtime Python binding is invalid",
+    )
+    model = value["model"]
+    _require(
+        set(model)
+        == {
+            "provider",
+            "default_model",
+            "smart_routing",
+            "smart_routing_model",
+            "gateway_smart_model_routing",
+            "model_section_sha256",
+            "hosted_weights",
+            "actual_model_required_in_turn_receipt",
+        }
+        and model.get("provider") == "openrouter"
+        and model.get("default_model") == "anthropic/claude-sonnet-4-6"
+        and model.get("smart_routing") is False
+        and model.get("smart_routing_model") is None
+        and model.get("gateway_smart_model_routing") is False
+        and _is_sha256(model.get("model_section_sha256"))
+        and model.get("hosted_weights") == "external_provider_managed_not_locally_hashable"
+        and model.get("actual_model_required_in_turn_receipt") is True,
+        "identity runtime model binding is invalid",
+    )
+    skills = value["skills"]
+    _require(
+        set(skills) == {"content_sha256", "file_count"}
+        and _is_sha256(skills.get("content_sha256"))
+        and isinstance(skills.get("file_count"), int)
+        and not isinstance(skills.get("file_count"), bool)
+        and skills["file_count"] > 0,
+        "identity runtime skills binding is invalid",
+    )
+    permissions = value["permissions"]
+    _require(
+        set(permissions)
+        == {
+            "identity_config_sha256",
+            "gateway_permissions_sha256",
+            "tool_permissions_sha256",
+            "runtime_policy",
+            "authorization_decision_required_in_runtime_receipt",
+        },
+        "identity runtime permission fields are invalid",
+    )
+    for field in ("identity_config_sha256", "gateway_permissions_sha256", "tool_permissions_sha256"):
+        _require(_is_sha256(permissions.get(field)), f"identity runtime permission {field} is invalid")
+    _require(
+        permissions.get("runtime_policy") == expected_no_send_runtime_policy()
+        and permissions.get("authorization_decision_required_in_runtime_receipt") is True,
+        "identity runtime contract must use the exact no-send permission policy",
+    )
+    behavior_inputs = value["behavior_inputs"]
+    _require(
+        set(behavior_inputs)
+        == {
+            "artifact_sha256",
+            "artifact_receipt_sha256",
+            "config_sha256",
+            "runtime_contract_sha256",
+            "uv_lock_sha256",
+            "identity_contract_sources",
+            "effective_policy",
+            "tool_surface",
+        },
+        "identity runtime behavior inputs are invalid",
+    )
+    for field in ("artifact_sha256", "artifact_receipt_sha256", "config_sha256", "runtime_contract_sha256", "uv_lock_sha256"):
+        _require(_is_sha256(behavior_inputs.get(field)), f"identity runtime behavior input {field} is invalid")
+    source_bindings = behavior_inputs.get("identity_contract_sources")
+    _require(
+        isinstance(source_bindings, dict)
+        and set(source_bindings) == {"scripts/leo_behavior_manifest.py", "scripts/leo_identity_manifest.py"}
+        and all(_is_sha256(item) for item in source_bindings.values())
+        and runtime.get("teleo_source_sha256") == behavior.canonical_sha256(source_bindings),
+        "identity runtime contract source bindings are invalid",
+    )
+    _require(
+        behavior_inputs.get("effective_policy")
+        == {
+            "cheap_model_route": None,
+            "memory_enabled": False,
+            "memory_nudge_interval": 0,
+            "skill_creation_nudge_interval": 0,
+            "smart_model_routing": False,
+            "user_profile_enabled": False,
+        },
+        "identity runtime effective policy is invalid",
+    )
+    _require(
+        behavior_inputs.get("tool_surface") == LEOCLEAN_NO_SEND_TOOL_SURFACE,
+        "identity runtime tool surface is invalid",
+    )
+    runtime_payload = {
+        "schema": value["schema"],
+        "runtime": {key: item for key, item in runtime.items() if key != "identity_runtime_sha256"},
+        "model": model,
+        "skills": skills,
+        "permissions": permissions,
+        "behavior_inputs": behavior_inputs,
+    }
+    _require(
+        runtime.get("identity_runtime_sha256") == behavior.canonical_sha256(runtime_payload),
+        "identity runtime contract runtime hash drift detected",
+    )
+    return {
+        "source_commit": teleo_git_head,
+        "runtime": copy.deepcopy(runtime),
+        "model": copy.deepcopy(value["model"]),
+        "skills": copy.deepcopy(value["skills"]),
+        "permissions": copy.deepcopy(value["permissions"]),
+    }
+
+
 def build_identity_manifest(
     *,
-    behavior_manifest: dict[str, Any],
+    behavior_manifest: dict[str, Any] | None = None,
+    identity_runtime_contract: dict[str, Any] | None = None,
     database_fingerprint_path: Path,
     constitution_path: Path,
     database_identity_path: Path,
     source_root: Path,
     source_commit: str | None = None,
 ) -> dict[str, Any]:
-    validate_behavior_manifest(behavior_manifest)
+    _require(
+        (behavior_manifest is None) != (identity_runtime_contract is None),
+        "exactly one behavior manifest or identity runtime contract is required",
+    )
+    if identity_runtime_contract is not None:
+        runtime_inputs = identity_runtime_inputs_from_contract(identity_runtime_contract)
+        contract_commit = runtime_inputs["source_commit"]
+        _require(
+            source_commit is None or source_commit == contract_commit,
+            "source commit does not match runtime contract",
+        )
+    else:
+        _require(isinstance(behavior_manifest, dict), "behavior manifest is required")
+        validate_behavior_manifest(behavior_manifest)
+        runtime_commit = behavior_manifest["teleo_infrastructure_runtime"]["git_head"]
+        source_commit = source_commit or runtime_commit
+        _require(bool(HEX_40.fullmatch(str(source_commit))), "source commit is invalid")
+        _require(source_commit == runtime_commit, "source commit does not match the behavior manifest")
+        model = behavior_manifest["model_runtime"]
+        components = behavior_manifest["components"]
+        runtime_inputs = {
+            "source_commit": source_commit,
+            "runtime": {
+                "identity_runtime_sha256": behavior_manifest["identity_runtime_sha256"],
+                "hermes_git_head": behavior_manifest["hermes_runtime"]["git_head"],
+                "hermes_source_sha256": behavior_manifest["hermes_runtime"]["source_tree"]["sha256"],
+                "teleo_git_head": runtime_commit,
+                "teleo_source_sha256": behavior_manifest["teleo_infrastructure_runtime"]["identity_source_tree"][
+                    "sha256"
+                ],
+                "python": behavior_manifest["python_runtime"],
+            },
+            "model": {
+                "provider": model["provider"],
+                "default_model": model["default_model"],
+                "smart_routing": model.get("smart_routing"),
+                "smart_routing_model": model.get("smart_routing_model"),
+                "gateway_smart_model_routing": model.get("gateway_smart_model_routing"),
+                "model_section_sha256": model["model_section_sha256"],
+                "hosted_weights": "external_provider_managed_not_locally_hashable",
+                "actual_model_required_in_turn_receipt": True,
+            },
+            "skills": {
+                "content_sha256": components["procedural_skills"]["content"]["sha256"],
+                "file_count": components["procedural_skills"]["content"]["file_count"],
+            },
+            "permissions": {
+                "identity_config_sha256": model["identity_config_sha256"],
+                "gateway_permissions_sha256": model["gateway_permissions_sha256"],
+                "tool_permissions_sha256": model["tool_permissions_sha256"],
+                "runtime_policy": model["identity_runtime_policy"],
+                "authorization_decision_required_in_runtime_receipt": True,
+            },
+        }
     database_fingerprint = load_json(database_fingerprint_path, "database fingerprint")
     validate_database_fingerprint(database_fingerprint)
     constitution = load_json(constitution_path, "constitution source")
@@ -448,46 +719,10 @@ def build_identity_manifest(
         database_identity,
         fingerprint=database_fingerprint,
     )
-    runtime_commit = behavior_manifest["teleo_infrastructure_runtime"]["git_head"]
-    source_commit = source_commit or runtime_commit
-    _require(bool(HEX_40.fullmatch(str(source_commit))), "source commit is invalid")
-    _require(source_commit == runtime_commit, "source commit does not match the behavior manifest")
-
-    model = behavior_manifest["model_runtime"]
-    components = behavior_manifest["components"]
     marker = database_fingerprint["read_consistency"]["before"]
     core = {
         "identity_name": "Leo",
-        "source_commit": source_commit,
-        "runtime": {
-            "identity_runtime_sha256": behavior_manifest["identity_runtime_sha256"],
-            "hermes_git_head": behavior_manifest["hermes_runtime"]["git_head"],
-            "hermes_source_sha256": behavior_manifest["hermes_runtime"]["source_tree"]["sha256"],
-            "teleo_git_head": runtime_commit,
-            "teleo_source_sha256": behavior_manifest["teleo_infrastructure_runtime"]["identity_source_tree"]["sha256"],
-            "python": behavior_manifest["python_runtime"],
-        },
-        "model": {
-            "provider": model["provider"],
-            "default_model": model["default_model"],
-            "smart_routing": model.get("smart_routing"),
-            "smart_routing_model": model.get("smart_routing_model"),
-            "gateway_smart_model_routing": model.get("gateway_smart_model_routing"),
-            "model_section_sha256": model["model_section_sha256"],
-            "hosted_weights": "external_provider_managed_not_locally_hashable",
-            "actual_model_required_in_turn_receipt": True,
-        },
-        "skills": {
-            "content_sha256": components["procedural_skills"]["content"]["sha256"],
-            "file_count": components["procedural_skills"]["content"]["file_count"],
-        },
-        "permissions": {
-            "identity_config_sha256": model["identity_config_sha256"],
-            "gateway_permissions_sha256": model["gateway_permissions_sha256"],
-            "tool_permissions_sha256": model["tool_permissions_sha256"],
-            "runtime_policy": model["identity_runtime_policy"],
-            "authorization_decision_required_in_runtime_receipt": True,
-        },
+        **runtime_inputs,
         "canonical_database": {
             "database": marker["database"],
             "database_user": marker["database_user"],
@@ -547,10 +782,20 @@ def build_identity_manifest(
             for name, content in sorted(views.items())
         },
     }
-    return {**stable, "manifest_sha256": behavior.canonical_sha256(stable)}
+    manifest = {**stable, "manifest_sha256": behavior.canonical_sha256(stable)}
+    if identity_runtime_contract is not None:
+        validate_identity_manifest(
+            manifest,
+            expected_runtime_policy_value=identity_runtime_contract["permissions"]["runtime_policy"],
+        )
+    return manifest
 
 
-def validate_identity_manifest(value: dict[str, Any]) -> None:
+def validate_identity_manifest(
+    value: dict[str, Any],
+    *,
+    expected_runtime_policy_value: dict[str, Any] | None = None,
+) -> None:
     _require(value.get("schema") == SCHEMA, "unsupported identity manifest schema")
     expected = value.get("manifest_sha256")
     _require(_is_sha256(expected), "identity manifest sha256 is invalid")
@@ -656,9 +901,18 @@ def validate_identity_manifest(value: dict[str, Any]) -> None:
     )
     for field in ("identity_config_sha256", "gateway_permissions_sha256", "tool_permissions_sha256"):
         _require(_is_sha256(permissions.get(field)), f"permission binding {field} is invalid")
+    runtime_policy = (
+        expected_runtime_policy_value if expected_runtime_policy_value is not None else expected_runtime_policy()
+    )
+    _require(_valid_runtime_policy(runtime_policy), "expected identity runtime policy is invalid")
+    policy_error = (
+        "identity runtime policy must be the exact local readback policy"
+        if expected_runtime_policy_value is None
+        else "identity runtime policy differs from the exact policy expected by this validator"
+    )
     _require(
-        permissions.get("runtime_policy") == expected_runtime_policy(),
-        "identity runtime policy must be the exact local readback policy",
+        permissions.get("runtime_policy") == runtime_policy,
+        policy_error,
     )
     _require(
         permissions.get("authorization_decision_required_in_runtime_receipt") is True,
@@ -765,18 +1019,49 @@ def validate_identity_manifest(value: dict[str, Any]) -> None:
         )
 
 
-def verify_bound_sources(value: dict[str, Any], source_root: Path) -> dict[str, str]:
-    validate_identity_manifest(value)
+def _bound_source_path(source_root: Path, repo_path: str, *, label: str) -> Path:
+    _require(source_root.is_dir() and not source_root.is_symlink(), "identity source root is missing or unsafe")
+    relative = Path(repo_path)
+    _require(
+        not relative.is_absolute() and relative.parts and ".." not in relative.parts,
+        f"bound {label} path is unsafe",
+    )
+    try:
+        resolved_root = source_root.resolve(strict=True)
+    except OSError as exc:
+        raise IdentityManifestError("identity source root cannot be resolved") from exc
+    candidate = source_root
+    for part in relative.parts:
+        candidate = candidate / part
+        _require(not candidate.is_symlink(), f"bound {label} has a symlink component")
+    try:
+        resolved = candidate.resolve(strict=True)
+        resolved.relative_to(resolved_root)
+    except (OSError, ValueError) as exc:
+        raise IdentityManifestError(f"bound {label} escapes the identity source root") from exc
+    _require(resolved.is_file(), f"bound {label} is missing")
+    return resolved
+
+
+def verify_bound_sources(
+    value: dict[str, Any],
+    source_root: Path,
+    *,
+    expected_runtime_policy_value: dict[str, Any] | None = None,
+) -> dict[str, str]:
+    validate_identity_manifest(value, expected_runtime_policy_value=expected_runtime_policy_value)
     core = value["identity_inputs"]
     bindings = {
         "database fingerprint": core["canonical_database"]["source"],
         "constitution source": core["identity_sources"]["static_constitution"],
         "database identity source": core["identity_sources"]["database_derived_identity"],
     }
-    paths = {label: source_root / binding["repo_path"] for label, binding in bindings.items()}
+    paths = {
+        label: _bound_source_path(source_root, binding["repo_path"], label=label)
+        for label, binding in bindings.items()
+    }
     for label, binding in bindings.items():
         path = paths[label]
-        _require(path.is_file(), f"bound {label} is missing")
         if binding.get("content_sha256"):
             _require(behavior.file_sha256(path) == binding["content_sha256"], f"bound {label} content drift detected")
     fingerprint = load_json(paths["database fingerprint"], "database fingerprint")

scripts/run_gcp_leoclean_nosend_oci_smoke.py

@@ -0,0 +1,420 @@
+#!/usr/bin/env python3
+"""Build and exercise one disposable synthetic leoclean no-send OCI image."""
+
+from __future__ import annotations
+
+import argparse
+import json
+import platform as host_platform
+import shutil
+import subprocess
+import sys
+import tempfile
+import time
+import uuid
+from pathlib import Path
+from typing import Any
+
+if __package__ in {None, ""}:
+    sys.path.insert(0, str(Path(__file__).resolve().parents[1]))
+
+from ops import gcp_leoclean_nosend_package as package
+
+SCHEMA = "livingip.leocleanNoSendOciSmoke.v1"
+SUCCESS_CLAIM_CEILING = (
+    "One disposable image built from a synthetic noncanonical identity started, reached its built-in no-send health "
+    "contract, rejected an unsupported send-message entrypoint command, and was removed. This does not prove a real "
+    "Hermes tool-call denial, general network-egress denial, canonical identity, Cloud SQL permissions, Artifact "
+    "Registry publication, GCP/systemd behavior, model parity, Telegram isolation in a live service, or production "
+    "readiness. Only the candidate container, tag, and visible image ID are required absent; digest-pinned base images "
+    "and build cache may remain."
+)
+FAILURE_CLAIM_CEILING = (
+    "This failed receipt makes no affirmative image-build, startup, health, command-rejection, or cleanup claim. "
+    "Cleanup is passing only when Docker inventory queries succeed and show the candidate container, tag, and visible "
+    "image ID absent; digest-pinned base images and build cache may remain."
+)
+ENTRYPOINT = "/opt/livingip/leoclean-nosend/entrypoint.py"
+RUNTIME_PYTHON = "/opt/livingip/leoclean-nosend/venv/bin/python"
+EXPECTED_ENTRYPOINT = [RUNTIME_PYTHON, ENTRYPOINT]
+EXPECTED_COMMAND = ["service"]
+EXPECTED_HEALTHCHECK = {
+    "Test": ["CMD", RUNTIME_PYTHON, ENTRYPOINT, "health"],
+    "Interval": 15_000_000_000,
+    "Timeout": 10_000_000_000,
+    "StartPeriod": 30_000_000_000,
+    "Retries": 4,
+}
+EXPECTED_STOP_SIGNAL = "SIGTERM"
+
+
+class SmokeError(RuntimeError):
+    """The disposable image did not satisfy its bounded smoke contract."""
+
+
+def _require(condition: bool, message: str) -> None:
+    if not condition:
+        raise SmokeError(message)
+
+
+def _run(
+    arguments: list[str],
+    *,
+    check: bool = True,
+    timeout: float = 120.0,
+) -> subprocess.CompletedProcess[str]:
+    try:
+        result = subprocess.run(
+            arguments,
+            check=False,
+            capture_output=True,
+            text=True,
+            timeout=timeout,
+        )
+    except (OSError, subprocess.TimeoutExpired) as exc:
+        raise SmokeError(f"command unavailable or timed out: {arguments[0]}") from exc
+    if check and result.returncode != 0:
+        raise SmokeError(f"command failed: {arguments[0]} {arguments[1] if len(arguments) > 1 else ''}".strip())
+    return result
+
+
+def _docker_json(arguments: list[str], *, timeout: float = 120.0) -> Any:
+    result = _run(["docker", *arguments], timeout=timeout)
+    try:
+        return json.loads(result.stdout)
+    except json.JSONDecodeError as exc:
+        raise SmokeError("Docker did not return JSON") from exc
+
+
+def _validate_image_runtime_configuration(image: dict[str, Any]) -> dict[str, Any]:
+    config = image.get("Config")
+    _require(isinstance(config, dict), "smoke image runtime configuration is missing")
+    user = config.get("User")
+    _require(user in (None, ""), "smoke image must begin as the root bootstrap user")
+    _require(config.get("Entrypoint") == EXPECTED_ENTRYPOINT, "smoke image entrypoint drifted")
+    _require(config.get("Cmd") == EXPECTED_COMMAND, "smoke image command drifted")
+    _require(config.get("StopSignal") == EXPECTED_STOP_SIGNAL, "smoke image stop signal drifted")
+    healthcheck = config.get("Healthcheck")
+    _require(isinstance(healthcheck, dict), "smoke image healthcheck is missing")
+    normalized_healthcheck = dict(healthcheck)
+    start_interval = normalized_healthcheck.pop("StartInterval", None)
+    _require(
+        start_interval is None or (type(start_interval) is int and start_interval == 0),
+        "smoke image healthcheck start interval drifted",
+    )
+    _require(normalized_healthcheck == EXPECTED_HEALTHCHECK, "smoke image healthcheck drifted")
+    return {
+        "bootstrap_user": "root",
+        "entrypoint": EXPECTED_ENTRYPOINT,
+        "command": EXPECTED_COMMAND,
+        "healthcheck": {**EXPECTED_HEALTHCHECK, "StartInterval": 0},
+        "stop_signal": EXPECTED_STOP_SIGNAL,
+    }
+
+
+def _write_receipt(path: Path, value: dict[str, Any]) -> None:
+    path.parent.mkdir(parents=True, exist_ok=True)
+    temporary = path.with_name(f".{path.name}.{uuid.uuid4().hex}.tmp")
+    temporary.write_text(json.dumps(value, indent=2, sort_keys=True) + "\n", encoding="utf-8")
+    temporary.replace(path)
+
+
+def _build_command(*, context: Path, tag: str, smoke_label: str, build_args: dict[str, str]) -> list[str]:
+    command = [
+        "docker",
+        "build",
+        "--platform",
+        package.PLATFORM,
+        "--pull=false",
+        "--tag",
+        tag,
+        "--label",
+        smoke_label,
+    ]
+    for name, value in sorted(build_args.items()):
+        command.extend(["--build-arg", f"{name}={value}"])
+    command.append(str(context))
+    return command
+
+
+def _run_command(*, tag: str, name: str, smoke_label: str) -> list[str]:
+    return [
+        "docker",
+        "run",
+        "--detach",
+        "--rm",
+        f"--name={name}",
+        "--pull=never",
+        "--read-only",
+        "--cap-drop=ALL",
+        "--cap-add=CHOWN",
+        "--cap-add=SETGID",
+        "--cap-add=SETUID",
+        "--cap-add=SETPCAP",
+        "--security-opt=no-new-privileges:true",
+        "--pids-limit=512",
+        "--network=bridge",
+        "--stop-timeout=30",
+        f"--tmpfs={package.PROFILE_ROOT}:rw,noexec,nosuid,nodev,size=64m,uid=0,gid=0,mode=0700",
+        f"--tmpfs=/tmp:rw,noexec,nosuid,nodev,size=64m,uid={package.RUNTIME_UID},gid={package.RUNTIME_GID},mode=0700",
+        "--tmpfs=/var/lib/postgresql/data:ro,noexec,nosuid,nodev,size=64k,uid=0,gid=0,mode=0000",
+        "--label",
+        smoke_label,
+        tag,
+        "service",
+    ]
+
+
+def _wait_healthy(name: str, *, expected_image_id: str, timeout: float = 180.0) -> dict[str, Any]:
+    deadline = time.monotonic() + timeout
+    while time.monotonic() < deadline:
+        result = _run(["docker", "container", "inspect", name], check=False, timeout=15)
+        if result.returncode != 0:
+            raise SmokeError("smoke container disappeared before health verification")
+        try:
+            inspected = json.loads(result.stdout)[0]
+        except (IndexError, KeyError, json.JSONDecodeError) as exc:
+            raise SmokeError("smoke container inspection is invalid") from exc
+        state = inspected.get("State") or {}
+        _require(inspected.get("Image") == expected_image_id, "running container image ID differs from the built image")
+        status = state.get("Status")
+        health = (state.get("Health") or {}).get("Status")
+        if health == "healthy":
+            return {"container_status": status, "health_status": health, "image_id": inspected["Image"]}
+        if status in {"dead", "exited", "removing"} or health == "unhealthy":
+            raise SmokeError(f"smoke container failed before health: {status}/{health}")
+        time.sleep(1)
+    raise SmokeError("smoke container did not become healthy before the deadline")
+
+
+def _verify_health(name: str) -> dict[str, Any]:
+    result = _run(
+        [
+            "docker",
+            "exec",
+            "--user",
+            f"{package.RUNTIME_UID}:{package.RUNTIME_GID}",
+            name,
+            RUNTIME_PYTHON,
+            ENTRYPOINT,
+            "health",
+        ],
+        timeout=30,
+    )
+    try:
+        value = json.loads(result.stdout)
+    except json.JSONDecodeError as exc:
+        raise SmokeError("container health command did not return JSON") from exc
+    _require(
+        value
+        == {
+            "schema": "livingip.leocleanNoSendContainerHealth.v1",
+            "status": "pass",
+            "process_id": 1,
+            "gateway_adapter_count": 0,
+            "send_message_present": False,
+        },
+        "container health response drifted",
+    )
+    return value
+
+
+def _verify_unsupported_send(name: str) -> dict[str, Any]:
+    result = _run(
+        [
+            "docker",
+            "exec",
+            "--user",
+            f"{package.RUNTIME_UID}:{package.RUNTIME_GID}",
+            name,
+            RUNTIME_PYTHON,
+            ENTRYPOINT,
+            "send-message",
+        ],
+        check=False,
+        timeout=30,
+    )
+    _require(result.returncode == 65, "unsupported send-message command did not fail closed")
+    _require(result.stdout == "", "unsupported send-message command emitted stdout")
+    try:
+        error = json.loads(result.stderr)
+    except json.JSONDecodeError as exc:
+        raise SmokeError("unsupported send-message command did not return redacted JSON") from exc
+    _require(
+        error == {"status": "fail", "error": "container_contract_failed"},
+        "unsupported send-message failure was not the exact redacted contract error",
+    )
+    return {"exit_code": result.returncode, "stdout_empty": True, "redacted_error": error}
+
+
+def _cleanup(*, name: str, tag: str, image_id: str | None, smoke_label: str) -> dict[str, Any]:
+    _run(["docker", "rm", "--force", name], check=False, timeout=45)
+    _run(["docker", "image", "rm", "--force", tag], check=False, timeout=90)
+    labeled = _run(
+        ["docker", "ps", "-aq", "--filter", f"label={smoke_label}"],
+        timeout=15,
+    ).stdout.split()
+    named = _run(["docker", "ps", "-aq", "--filter", f"name=^{name}$"], timeout=15).stdout.split()
+    tagged = _run(
+        ["docker", "image", "ls", "--quiet", "--filter", f"reference={tag}"], timeout=15
+    ).stdout.split()
+    labeled_images = _run(
+        ["docker", "image", "ls", "--quiet", "--filter", f"label={smoke_label}"], timeout=15
+    ).stdout.split()
+    visible_image_ids = _run(["docker", "image", "ls", "--no-trunc", "--quiet"], timeout=15).stdout.split()
+    result = {
+        "labeled_containers_absent": labeled == [],
+        "named_container_absent": named == [],
+        "image_tag_absent": tagged == [],
+        "labeled_images_absent": labeled_images == [],
+        "candidate_image_id_absent": image_id is None or image_id not in visible_image_ids,
+        "docker_inventory_queries_succeeded": True,
+    }
+    result["status"] = "pass" if all(result.values()) else "fail"
+    return result
+
+
+def run_smoke(*, repo_root: Path, artifact: Path, runtime_receipt: Path, output: Path) -> tuple[dict[str, Any], int]:
+    smoke_id = uuid.uuid4().hex
+    tag = f"livingip-leoclean-nosend-smoke:{smoke_id}"
+    name = f"livingip-leoclean-nosend-smoke-{smoke_id[:12]}"
+    smoke_label = f"livingip.leoclean-nosend-smoke={smoke_id}"
+    image_id: str | None = None
+    report: dict[str, Any] = {
+        "schema": SCHEMA,
+        "status": "fail",
+        "claim_ceiling": FAILURE_CLAIM_CEILING,
+        "smoke_id": smoke_id,
+        "platform": package.PLATFORM,
+        "synthetic_identity": True,
+        "canonical_authority_granted": False,
+        "build_count": 1,
+        "general_network_egress_denial_tested": False,
+        "real_hermes_send_tool_invocation_tested": False,
+    }
+    exit_code = 65
+    try:
+        _require(shutil.which("docker") is not None, "Docker is unavailable")
+        docker_server_arch = _docker_json(["version", "--format", "{{json .Server.Arch}}"], timeout=30)
+        _require(isinstance(docker_server_arch, str) and bool(docker_server_arch), "Docker server architecture is invalid")
+        execution_mode = "native" if docker_server_arch in {"amd64", "x86_64"} else "emulated_or_cross_platform"
+        _require(repo_root.is_dir() and not repo_root.is_symlink(), "repository root is missing or unsafe")
+        revision = package._git_head(repo_root)
+        runtime = package.validate_artifact(artifact, runtime_receipt, revision)
+        contract = package.build_expected_identity_contract(
+            artifact=artifact,
+            receipt_path=runtime_receipt,
+            runtime=runtime,
+            source_revision=revision,
+            source_root=repo_root,
+        )
+        fixture = repo_root / "fixtures" / "working-leo" / "leo-identity-v1"
+        manifest = package.strict_identity_manifest.build_identity_manifest(
+            identity_runtime_contract=contract,
+            database_fingerprint_path=fixture / "leo-database-fingerprint-v1.json",
+            constitution_path=fixture / "leo-constitution-v1.json",
+            database_identity_path=fixture / "leo-database-identity-v1.json",
+            source_root=repo_root,
+        )
+        with tempfile.TemporaryDirectory(prefix="livingip-leoclean-oci-smoke-") as temporary:
+            temporary_root = Path(temporary)
+            identity = temporary_root / "identity"
+            package.compile_identity_bundle(
+                manifest=manifest,
+                expected_identity_contract=contract,
+                source_root=repo_root,
+                output=identity,
+            )
+            context = temporary_root / "context"
+            image_input = package.prepare_context(
+                repo_root=repo_root,
+                artifact=artifact,
+                receipt=runtime_receipt,
+                identity=identity,
+                identity_source_root=repo_root,
+                output=context,
+                source_revision=revision,
+            )
+            build_args = package.load_json(context / "build-args.json", "build arguments")
+            _run(_build_command(context=context, tag=tag, smoke_label=smoke_label, build_args=build_args), timeout=1200)
+            inspected = _docker_json(["image", "inspect", tag], timeout=30)
+            _require(isinstance(inspected, list) and len(inspected) == 1, "smoke image inspection is not exact")
+            image = inspected[0]
+            image_id = image.get("Id")
+            _require(isinstance(image_id, str) and image_id.startswith("sha256:"), "smoke image ID is invalid")
+            _require(image.get("Os") == "linux" and image.get("Architecture") == "amd64", "smoke image platform drifted")
+            labels = (image.get("Config") or {}).get("Labels") or {}
+            for key, value in package.expected_image_labels(image_input).items():
+                _require(labels.get(key) == value, f"smoke image label drifted: {key}")
+            _require(labels.get("livingip.leoclean-nosend-smoke") == smoke_id, "smoke image label is missing")
+            runtime_configuration = _validate_image_runtime_configuration(image)
+            _run(_run_command(tag=tag, name=name, smoke_label=smoke_label), timeout=60)
+            health_wait = _wait_healthy(name, expected_image_id=image_id)
+            health = _verify_health(name)
+            unsupported = _verify_unsupported_send(name)
+            _run(["docker", "stop", "--time=30", name], timeout=60)
+            deadline = time.monotonic() + 60
+            while time.monotonic() < deadline:
+                if _run(["docker", "container", "inspect", name], check=False, timeout=15).returncode != 0:
+                    break
+                time.sleep(0.5)
+            else:
+                raise SmokeError("smoke container was not removed after stop")
+            report.update(
+                {
+                    "teleo_git_head": revision,
+                    "artifact_sha256": runtime["artifact_sha256"],
+                    "identity_sha256": image_input["identity"]["bundle_sha256"],
+                    "input_sha256": image_input["input_sha256"],
+                    "image_id": image_id,
+                    "docker_server_architecture": docker_server_arch,
+                    "host_machine": host_platform.machine(),
+                    "target_execution_mode": execution_mode,
+                    "image_runtime_configuration": runtime_configuration,
+                    "health_wait": health_wait,
+                    "health": health,
+                    "unsupported_send_message": unsupported,
+                }
+            )
+            exit_code = 0
+    except (SmokeError, package.PackageError, OSError, ValueError) as exc:
+        report["error"] = str(exc)
+    finally:
+        try:
+            cleanup = _cleanup(name=name, tag=tag, image_id=image_id, smoke_label=smoke_label)
+        except SmokeError as exc:
+            cleanup = {"status": "fail", "error": str(exc)}
+        report["cleanup"] = cleanup
+        if cleanup.get("status") != "pass":
+            exit_code = 65
+            report["error"] = "disposable OCI cleanup did not pass"
+        report["status"] = "pass" if exit_code == 0 else "fail"
+        report["claim_ceiling"] = SUCCESS_CLAIM_CEILING if exit_code == 0 else FAILURE_CLAIM_CEILING
+        _write_receipt(output, report)
+    return report, exit_code
+
+
+def _parse_args(argv: list[str] | None = None) -> argparse.Namespace:
+    parser = argparse.ArgumentParser(description=__doc__)
+    parser.add_argument("--repo-root", type=Path, default=Path(__file__).resolve().parents[1])
+    parser.add_argument("--artifact", type=Path, required=True)
+    parser.add_argument("--runtime-receipt", type=Path, required=True)
+    parser.add_argument("--output", type=Path, required=True)
+    return parser.parse_args(argv)
+
+
+def main(argv: list[str] | None = None) -> int:
+    args = _parse_args(argv)
+    report, exit_code = run_smoke(
+        repo_root=args.repo_root.resolve(),
+        artifact=args.artifact.resolve(),
+        runtime_receipt=args.runtime_receipt.resolve(),
+        output=args.output.absolute(),
+    )
+    stream = sys.stdout if exit_code == 0 else sys.stderr
+    print(json.dumps({"schema": SCHEMA, "status": report["status"]}, sort_keys=True), file=stream)
+    return exit_code
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())

scripts/run_leoclean_nosend_runtime_canary.sh

@@ -6,6 +6,8 @@ results_dir="${LEOCLEAN_NOSEND_RESULTS_DIR:-${repo_root}/.crabbox-results}"
 receipt="${LEOCLEAN_NOSEND_RECEIPT:-${results_dir}/leoclean-nosend-runtime.json}"
 uv_bin="${UV_BIN:-$(command -v uv || true)}"
 python_bin="${PYTHON_BIN:-python3}"
+artifact_output="${LEOCLEAN_NOSEND_ARTIFACT_OUTPUT:-}"
+artifact_output_partial=""
 
 if [[ -z "$uv_bin" || ! -x "$uv_bin" ]]; then
   echo "uv 0.9.30 is required" >&2
@@ -22,6 +24,11 @@ fi
 
 temporary="$(mktemp -d "${TMPDIR:-/tmp}/livingip-nosend-runtime.XXXXXX")"
 cleanup() {
+  if [[ -n "$artifact_output_partial" ]]; then
+    chmod -R u+w "$artifact_output_partial" 2>/dev/null || true
+    rm -rf "$artifact_output_partial"
+  fi
+  chmod -R u+w "$temporary" 2>/dev/null || true
   rm -rf "$temporary"
 }
 trap cleanup EXIT INT TERM
@@ -82,3 +89,19 @@ if receipt.get("status") != "pass" or receipt.get("verification_mode") != "relea
     raise SystemExit("no-send runtime receipt did not pass in release mode")
 print(json.dumps({"artifact_sha256": receipt["artifact_sha256"], "status": "pass"}, sort_keys=True))
 PY
+
+if [[ -n "$artifact_output" ]]; then
+  if [[ -e "$artifact_output" || -L "$artifact_output" ]]; then
+    echo "retained artifact output already exists: $artifact_output" >&2
+    exit 65
+  fi
+  mkdir -p "$(dirname "$artifact_output")"
+  artifact_output_partial="${artifact_output}.partial.$$"
+  if [[ -e "$artifact_output_partial" || -L "$artifact_output_partial" ]]; then
+    echo "retained artifact partial output already exists: $artifact_output_partial" >&2
+    exit 65
+  fi
+  cp -pR "$artifact" "$artifact_output_partial"
+  mv "$artifact_output_partial" "$artifact_output"
+  artifact_output_partial=""
+fi

tests/test_gcp_leoclean_nosend_oci_smoke.py

@@ -0,0 +1,219 @@
+from __future__ import annotations
+
+import copy
+import json
+import subprocess
+from pathlib import Path
+
+import pytest
+
+from scripts import run_gcp_leoclean_nosend_oci_smoke as smoke
+
+
+def _runtime_image(*, user: str | None = "", start_interval: int | None = 0) -> dict:
+    healthcheck = {
+        **smoke.EXPECTED_HEALTHCHECK,
+        "StartInterval": start_interval,
+    }
+    return {
+        "Config": {
+            "User": user,
+            "Entrypoint": list(smoke.EXPECTED_ENTRYPOINT),
+            "Cmd": list(smoke.EXPECTED_COMMAND),
+            "Healthcheck": healthcheck,
+            "StopSignal": smoke.EXPECTED_STOP_SIGNAL,
+        }
+    }
+
+
+@pytest.mark.parametrize("user", [None, ""])
+@pytest.mark.parametrize("start_interval", [None, 0])
+def test_runtime_configuration_accepts_portable_unset_values(user: str | None, start_interval: int | None) -> None:
+    receipt = smoke._validate_image_runtime_configuration(
+        _runtime_image(user=user, start_interval=start_interval)
+    )
+
+    assert receipt["bootstrap_user"] == "root"
+    assert receipt["entrypoint"] == smoke.EXPECTED_ENTRYPOINT
+    assert receipt["command"] == ["service"]
+    assert receipt["stop_signal"] == "SIGTERM"
+
+
+@pytest.mark.parametrize(
+    ("field", "value", "message"),
+    [
+        ("User", "65532:65532", "root bootstrap user"),
+        ("Entrypoint", ["/bin/sh"], "entrypoint drifted"),
+        ("Cmd", ["probe"], "command drifted"),
+        ("StopSignal", "SIGKILL", "stop signal drifted"),
+    ],
+)
+def test_runtime_configuration_rejects_authoritative_field_drift(field: str, value: object, message: str) -> None:
+    image = _runtime_image()
+    image["Config"][field] = value
+
+    with pytest.raises(smoke.SmokeError, match=message):
+        smoke._validate_image_runtime_configuration(image)
+
+
+@pytest.mark.parametrize(
+    ("field", "value"),
+    [
+        ("Test", ["CMD-SHELL", "true"]),
+        ("Interval", 1),
+        ("Timeout", 1),
+        ("StartPeriod", 1),
+        ("Retries", 1),
+        ("StartInterval", 1),
+        ("StartInterval", False),
+        ("Unexpected", 1),
+    ],
+)
+def test_runtime_configuration_rejects_healthcheck_drift(field: str, value: object) -> None:
+    image = _runtime_image()
+    image["Config"]["Healthcheck"] = copy.deepcopy(image["Config"]["Healthcheck"])
+    image["Config"]["Healthcheck"][field] = value
+
+    with pytest.raises(smoke.SmokeError, match="healthcheck"):
+        smoke._validate_image_runtime_configuration(image)
+
+
+def test_build_command_is_one_platform_pinned_local_build(tmp_path: Path) -> None:
+    command = smoke._build_command(
+        context=tmp_path,
+        tag="livingip-leoclean-nosend-smoke:test",
+        smoke_label="livingip.leoclean-nosend-smoke=test",
+        build_args={"TELEO_REVISION": "1" * 40, "INPUT_SHA256": "2" * 64},
+    )
+
+    assert command[:6] == ["docker", "build", "--platform", "linux/amd64", "--pull=false", "--tag"]
+    assert command.count("build") == 1
+    assert "--push" not in command
+    assert "--load" not in command
+    assert command[-1] == str(tmp_path)
+    assert command.index("INPUT_SHA256=" + "2" * 64) < command.index("TELEO_REVISION=" + "1" * 40)
+
+
+def test_run_command_has_no_port_secret_or_host_mount_surface() -> None:
+    command = smoke._run_command(
+        tag="livingip-leoclean-nosend-smoke:test",
+        name="livingip-leoclean-nosend-smoke-test",
+        smoke_label="livingip.leoclean-nosend-smoke=test",
+    )
+    encoded = " ".join(command).casefold()
+
+    for required in (
+        "--read-only",
+        "--cap-drop=all",
+        "--security-opt=no-new-privileges:true",
+        "--network=bridge",
+        "--pids-limit=512",
+        "--pull=never",
+    ):
+        assert required in encoded
+    assert "--publish" not in encoded
+    assert " -p " not in encoded
+    assert "--env" not in encoded
+    assert "--volume" not in encoded
+    assert "docker.sock" not in encoded
+    assert "telegram" not in encoded
+
+
+def test_unsupported_send_requires_exact_redacted_fail_closed_result(monkeypatch: pytest.MonkeyPatch) -> None:
+    result = subprocess.CompletedProcess(
+        args=["docker"],
+        returncode=65,
+        stdout="",
+        stderr=json.dumps({"status": "fail", "error": "container_contract_failed"}) + "\n",
+    )
+    monkeypatch.setattr(smoke, "_run", lambda *_args, **_kwargs: result)
+
+    receipt = smoke._verify_unsupported_send("fixture")
+
+    assert receipt == {
+        "exit_code": 65,
+        "stdout_empty": True,
+        "redacted_error": {"status": "fail", "error": "container_contract_failed"},
+    }
+
+
+def test_claim_ceiling_does_not_overstate_entrypoint_dispatch_probe() -> None:
+    assert "does not prove a real Hermes tool-call denial" in smoke.SUCCESS_CLAIM_CEILING
+    assert "general network-egress denial" in smoke.SUCCESS_CLAIM_CEILING
+    assert "Cloud SQL permissions" in smoke.SUCCESS_CLAIM_CEILING
+    assert "production readiness" in smoke.SUCCESS_CLAIM_CEILING
+    assert "base images and build cache may remain" in smoke.SUCCESS_CLAIM_CEILING
+    assert "makes no affirmative image-build" in smoke.FAILURE_CLAIM_CEILING
+
+
+def test_health_wait_rejects_container_running_a_different_image(monkeypatch: pytest.MonkeyPatch) -> None:
+    inspected = [{"Image": "sha256:wrong", "State": {"Status": "running", "Health": {"Status": "healthy"}}}]
+    result = subprocess.CompletedProcess(args=["docker"], returncode=0, stdout=json.dumps(inspected), stderr="")
+    monkeypatch.setattr(smoke, "_run", lambda *_args, **_kwargs: result)
+
+    with pytest.raises(smoke.SmokeError, match="image ID differs"):
+        smoke._wait_healthy("fixture", expected_image_id="sha256:expected", timeout=0.1)
+
+
+def test_cleanup_fails_when_docker_inventory_is_indeterminate(monkeypatch: pytest.MonkeyPatch) -> None:
+    calls = 0
+
+    def fake_run(*_args, **_kwargs):
+        nonlocal calls
+        calls += 1
+        if calls <= 2:
+            return subprocess.CompletedProcess(args=["docker"], returncode=1, stdout="", stderr="not found")
+        raise smoke.SmokeError("Docker inventory unavailable")
+
+    monkeypatch.setattr(smoke, "_run", fake_run)
+
+    with pytest.raises(smoke.SmokeError, match="inventory unavailable"):
+        smoke._cleanup(
+            name="fixture",
+            tag="livingip-leoclean-nosend-smoke:fixture",
+            image_id="sha256:" + "1" * 64,
+            smoke_label="livingip.leoclean-nosend-smoke=fixture",
+        )
+
+
+def test_failed_receipt_never_carries_the_success_claim(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
+    monkeypatch.setattr(smoke.shutil, "which", lambda _name: None)
+    monkeypatch.setattr(smoke, "_cleanup", lambda **_kwargs: {"status": "pass"})
+    output = tmp_path / "receipt.json"
+
+    report, exit_code = smoke.run_smoke(
+        repo_root=tmp_path,
+        artifact=tmp_path / "artifact",
+        runtime_receipt=tmp_path / "runtime.json",
+        output=output,
+    )
+
+    assert exit_code == 65
+    assert report["status"] == "fail"
+    assert report["claim_ceiling"] == smoke.FAILURE_CLAIM_CEILING
+    assert smoke.SUCCESS_CLAIM_CEILING not in output.read_text(encoding="utf-8")
+
+
+def test_runtime_canary_can_retain_only_a_successfully_verified_artifact() -> None:
+    source = (Path(__file__).resolve().parents[1] / "scripts" / "run_leoclean_nosend_runtime_canary.sh").read_text(
+        encoding="utf-8"
+    )
+
+    verification = source.index('receipt.get("status") != "pass"')
+    retention = source.index('if [[ -n "$artifact_output" ]]')
+    assert verification < retention
+    assert "retained artifact output already exists" in source
+    assert 'cp -pR "$artifact" "$artifact_output_partial"' in source
+    assert 'mv "$artifact_output_partial" "$artifact_output"' in source
+
+
+def test_ci_preloads_the_exact_target_architecture_and_checks_inventory() -> None:
+    workflow = (Path(__file__).resolve().parents[1] / ".github" / "workflows" / "ci.yml").read_text(
+        encoding="utf-8"
+    )
+
+    assert workflow.count("docker pull --platform=linux/amd64") == 3
+    assert "container_inventory=\"$(docker ps" in workflow
+    assert "image_inventory=\"$(docker images" in workflow
+    assert "remaining_containers=\"$(docker ps" in workflow
+    assert "remaining_images=\"$(docker images" in workflow

tests/test_gcp_leoclean_nosend_package.py

@@ -0,0 +1,911 @@
+"""Offline contracts for the isolated GCP leoclean no-send service image."""
+
+from __future__ import annotations
+
+import copy
+import importlib.util
+import json
+import os
+import shutil
+import subprocess
+import sys
+from pathlib import Path
+
+import pytest
+
+from ops import gcp_leoclean_nosend_package as package
+
+ROOT = Path(__file__).resolve().parents[1]
+REVISION = package._git_commit(ROOT)
+IMAGE_DIGEST = "2" * 64
+CLAIM_CEILING = "Offline package proof only; live GCP identity, database access, restart, and parity remain unproven."
+
+
+def write_json(path: Path, value: object, mode: int = 0o600) -> None:
+    path.write_text(json.dumps(value, indent=2, sort_keys=True) + "\n", encoding="utf-8")
+    path.chmod(mode)
+
+
+def git_source_fixture(tmp_path: Path, files: dict[str, str]) -> tuple[Path, str, dict[str, str]]:
+    root = tmp_path / "source-repository"
+    root.mkdir()
+    subprocess.run(["git", "init", "--quiet", str(root)], check=True)
+    for relative, content in files.items():
+        path = root / relative
+        path.parent.mkdir(parents=True, exist_ok=True)
+        path.write_text(content, encoding="utf-8")
+    subprocess.run(["git", "-C", str(root), "add", "--", *sorted(files)], check=True)
+    subprocess.run(
+        [
+            "git",
+            "-C",
+            str(root),
+            "-c",
+            "user.name=fwazb",
+            "-c",
+            "user.email=fawaz.butt95@gmail.com",
+            "commit",
+            "--quiet",
+            "-m",
+            "Create source fixture",
+        ],
+        check=True,
+    )
+    revision = package._git_commit(root)
+    expected = {relative: package.sha256_file(root / relative) for relative in files}
+    return root, revision, expected
+
+
+def load_entrypoint_module(tmp_path: Path):
+    image_root = tmp_path / "entrypoint-image"
+    image_root.mkdir()
+    entrypoint_path = image_root / "entrypoint.py"
+    shutil.copy2(ROOT / "docker" / "leoclean-nosend" / "entrypoint.py", entrypoint_path)
+    shutil.copy2(ROOT / "ops" / "gcp_leoclean_nosend_package.py", image_root / "package_contract.py")
+    scripts = image_root / "scripts"
+    scripts.mkdir()
+    for relative in package.IDENTITY_CONTRACT_SOURCES:
+        shutil.copy2(ROOT / relative, image_root / relative)
+    spec = importlib.util.spec_from_file_location(f"leoclean_entrypoint_{tmp_path.name}", entrypoint_path)
+    assert spec is not None and spec.loader is not None
+    module = importlib.util.module_from_spec(spec)
+    spec.loader.exec_module(module)
+    return module
+
+
+def artifact_fixture(tmp_path: Path) -> tuple[Path, Path]:
+    artifact = tmp_path / "artifact"
+    artifact.mkdir()
+    (artifact / "bootstrap.py").write_text("#!/usr/bin/env python3\n", encoding="utf-8")
+    (artifact / "bootstrap.py").chmod(0o755)
+    hermes = artifact / "hermes-agent"
+    hermes.mkdir()
+    (hermes / "uv.lock").write_text(
+        'version = 1\n\n[[package]]\nname = "pyyaml"\nversion = "6.0.3"\n',
+        encoding="utf-8",
+    )
+    uv_lock_sha256 = package.sha256_file(hermes / "uv.lock")
+    write_json(
+        artifact / "runtime-lock.json",
+        {
+            "base_images": {
+                "python": package.PYTHON_IMAGE.removeprefix("docker.io/library/"),
+                "uv": package.UV_IMAGE,
+            },
+            "hermes": {
+                "commit": package.HERMES_COMMIT,
+                "uv_lock_sha256": uv_lock_sha256,
+                "patched_source_tree_sha256": "4" * 64,
+            },
+        },
+        mode=0o644,
+    )
+    profile = artifact / "profile-template"
+    profile.mkdir()
+    (profile / "config.yaml").write_text(
+        """model:
+  provider: openrouter
+  default: anthropic/claude-sonnet-4-6
+  max_tokens: 8192
+smart_model_routing: false
+agent:
+  max_turns: 90
+memory:
+  enabled: false
+  memory_enabled: false
+  user_profile_enabled: false
+  nudge_interval: 0
+skills:
+  creation_nudge_interval: 0
+platform_toolsets:
+  api_server: [livingip-leoclean-nosend]
+known_plugin_toolsets:
+  api_server: []
+mcp_servers: {}
+display:
+  tool_progress: false
+""",
+        encoding="utf-8",
+    )
+    (profile / "config.yaml").chmod(0o600)
+    skills = profile / "skills" / "teleo-kb-bridge"
+    skills.mkdir(parents=True)
+    (skills / "SKILL.md").write_text("# fixture skill\n", encoding="utf-8")
+    write_json(
+        artifact / "runtime-contract.json",
+        {
+            "schema": "livingip.leocleanNoSendRuntimeContract.v1",
+            "transport_authority": "absent",
+            "toolset": {"allowed_tools": ["skills_list", "skill_view", "terminal"]},
+            "database": {
+                "direct_canonical_writes": "denied",
+                "direct_stage_table_writes": "denied",
+                "proposal_staging": "function_only",
+                "proposal_function": package.PROPOSAL_FUNCTION,
+            },
+        },
+        mode=0o644,
+    )
+    content = package.tree_manifest(artifact)
+    manifest = {
+        "schema": package.ARTIFACT_MANIFEST_SCHEMA,
+        "teleo_git_head": REVISION,
+        "hermes_commit": package.HERMES_COMMIT,
+        "claim_ceiling": CLAIM_CEILING,
+        "content": content,
+    }
+    write_json(artifact / "artifact-manifest.json", manifest, mode=0o644)
+    receipt = tmp_path / "receipt.json"
+    write_json(
+        receipt,
+        {
+            "schema": package.ARTIFACT_RECEIPT_SCHEMA,
+            "status": "pass",
+            "verification_mode": "release",
+            "teleo_git_head": REVISION,
+            "artifact_sha256": content["sha256"],
+            "hermes_commit": package.HERMES_COMMIT,
+            "hermes_source_sha256": "4" * 64,
+            "config_sha256": package.sha256_file(profile / "config.yaml"),
+            "runtime_probe": runtime_probe_fixture(package.sha256_file(profile / "config.yaml")),
+            "claim_ceiling": CLAIM_CEILING,
+        },
+        mode=0o644,
+    )
+    return artifact, receipt
+
+
+def runtime_surface_fixture() -> dict[str, object]:
+    return copy.deepcopy(package.NO_SEND_TOOL_SURFACE)
+
+
+def runtime_probe_fixture(config_sha256: str) -> dict[str, object]:
+    surface = runtime_surface_fixture()
+    return {
+        "schema": "livingip.leocleanNoSendRuntimeProbe.v1",
+        "status": "pass",
+        "config_sha256": config_sha256,
+        "gateway_adapter_count": 0,
+        "connected_platforms": [],
+        "python_version": package.PYTHON_VERSION,
+        "model": package.DEFAULT_MODEL,
+        "provider_credential_names_present": ["OPENROUTER_API_KEY"],
+        "provider_credential_sentinel_redaction": "pass",
+        "effective_policy": {
+            "cheap_model_route": None,
+            "memory_enabled": False,
+            "memory_nudge_interval": 0,
+            "skill_creation_nudge_interval": 0,
+            "smart_model_routing": False,
+            "user_profile_enabled": False,
+        },
+        "tool_surface": surface,
+        "lifecycle": {
+            "started": True,
+            "stopped": True,
+            "post_start_registry": surface,
+            "post_discovery_registry": surface,
+        },
+    }
+
+
+def test_git_source_closure_accepts_exact_revision_and_ignores_unrelated_dirty_files(tmp_path: Path) -> None:
+    root, revision, expected = git_source_fixture(tmp_path, {"package/source.py": "reviewed = True\n"})
+    (root / "unrelated.txt").write_text("not in the image closure\n", encoding="utf-8")
+
+    package._verify_git_source_closure(root, revision, expected)
+
+
+@pytest.mark.parametrize("mutation", ["unstaged", "staged", "deleted", "mode"])
+def test_git_source_closure_rejects_dirty_required_source(tmp_path: Path, mutation: str) -> None:
+    root, revision, expected = git_source_fixture(tmp_path, {"package/source.py": "reviewed = True\n"})
+    source = root / "package" / "source.py"
+    if mutation == "deleted":
+        source.unlink()
+    elif mutation == "mode":
+        source.chmod(0o755)
+    else:
+        source.write_text("reviewed = False\n", encoding="utf-8")
+        if mutation == "staged":
+            subprocess.run(["git", "-C", str(root), "add", "--", "package/source.py"], check=True)
+
+    with pytest.raises(package.PackageError, match="closure is dirty"):
+        package._verify_git_source_closure(root, revision, expected)
+
+
+def test_git_source_closure_rejects_untracked_required_source(tmp_path: Path) -> None:
+    root, revision, expected = git_source_fixture(tmp_path, {"package/source.py": "reviewed = True\n"})
+    untracked = root / "package" / "validator.py"
+    untracked.write_text("unreviewed = True\n", encoding="utf-8")
+    expected["package/validator.py"] = package.sha256_file(untracked)
+
+    with pytest.raises(package.PackageError):
+        package._verify_git_source_closure(root, revision, expected)
+
+
+def test_git_source_closure_rejects_head_and_hash_mismatch(tmp_path: Path) -> None:
+    root, revision, expected = git_source_fixture(tmp_path, {"package/source.py": "reviewed = True\n"})
+    (root / "unrelated.txt").write_text("second commit\n", encoding="utf-8")
+    subprocess.run(["git", "-C", str(root), "add", "unrelated.txt"], check=True)
+    subprocess.run(
+        [
+            "git",
+            "-C",
+            str(root),
+            "-c",
+            "user.name=fwazb",
+            "-c",
+            "user.email=fawaz.butt95@gmail.com",
+            "commit",
+            "--quiet",
+            "-m",
+            "Advance source fixture",
+        ],
+        check=True,
+    )
+
+    with pytest.raises(package.PackageError, match="reviewed revision"):
+        package._verify_git_source_closure(root, revision, expected)
+
+    current = package._git_commit(root)
+    wrong_hash = {"package/source.py": "9" * 64}
+    with pytest.raises(package.PackageError, match="package binding"):
+        package._verify_git_source_closure(root, current, wrong_hash)
+
+
+def identity_fixture(tmp_path: Path, artifact: Path, receipt: Path) -> tuple[Path, dict[str, object]]:
+    runtime = package.validate_artifact(artifact, receipt, REVISION)
+    contract = package.build_expected_identity_contract(
+        artifact=artifact,
+        receipt_path=receipt,
+        runtime=runtime,
+        source_revision=REVISION,
+        source_root=ROOT,
+    )
+    fixture = ROOT / "fixtures" / "working-leo" / "leo-identity-v1"
+    manifest = package.strict_identity_manifest.build_identity_manifest(
+        identity_runtime_contract=contract,
+        database_fingerprint_path=fixture / "leo-database-fingerprint-v1.json",
+        constitution_path=fixture / "leo-constitution-v1.json",
+        database_identity_path=fixture / "leo-database-identity-v1.json",
+        source_root=ROOT,
+    )
+    identity = tmp_path / "identity"
+    package.compile_identity_bundle(
+        manifest=manifest,
+        expected_identity_contract=contract,
+        source_root=ROOT,
+        output=identity,
+    )
+    return identity, contract
+
+
+def fully_rehash_identity_bundle(identity: Path) -> None:
+    manifest_path = identity / "identity-manifest.json"
+    manifest = json.loads(manifest_path.read_text(encoding="utf-8"))
+    inputs_hash = package.canonical_sha256(manifest["identity_inputs"])
+    manifest["identity_inputs_sha256"] = inputs_hash
+    fixture = ROOT / "fixtures" / "working-leo" / "leo-identity-v1"
+    fingerprint = package.strict_identity_manifest.load_json(
+        fixture / "leo-database-fingerprint-v1.json", "database fingerprint"
+    )
+    rules = package.strict_identity_manifest.canonical_rules(
+        package.strict_identity_manifest.load_json(fixture / "leo-constitution-v1.json", "constitution")
+    )
+    database_identity = package.strict_identity_manifest.load_json(
+        fixture / "leo-database-identity-v1.json", "database identity"
+    )
+    records = package.strict_identity_manifest.canonical_database_records(
+        database_identity,
+        fingerprint_sha256=fingerprint["fingerprint_sha256"],
+    )
+    provenance = package.strict_identity_manifest.database_identity_provenance(
+        database_identity,
+        fingerprint=fingerprint,
+    )
+    rendered = package.strict_identity_manifest.render_identity_views(
+        identity_inputs_sha256=inputs_hash,
+        database_fingerprint_sha256=fingerprint["fingerprint_sha256"],
+        database_provenance=provenance,
+        rules=rules,
+        records=records,
+    )
+    for name, content in rendered.items():
+        (identity / name).write_text(content, encoding="utf-8")
+        (identity / name).chmod(0o600)
+        manifest["compiled_views"][name]["sha256"] = package.sha256_file(identity / name)
+        manifest["compiled_views"][name]["generated_from_identity_inputs_sha256"] = inputs_hash
+    stable = {key: value for key, value in manifest.items() if key != "manifest_sha256"}
+    manifest["manifest_sha256"] = package.canonical_sha256(stable)
+    write_json(manifest_path, manifest)
+    lock_path = identity / "identity-lock.json"
+    lock = json.loads(lock_path.read_text(encoding="utf-8"))
+    lock["manifest_sha256"] = manifest["manifest_sha256"]
+    lock["identity_inputs_sha256"] = inputs_hash
+    lock["compiled_views"] = {
+        name: {"sha256": package.sha256_file(identity / name)} for name in sorted(rendered)
+    }
+    write_json(lock_path, lock)
+
+
+def fully_rehash_identity_contract(contract: dict[str, object]) -> None:
+    runtime = contract["runtime"]
+    assert isinstance(runtime, dict)
+    runtime["identity_runtime_sha256"] = package.canonical_sha256(package._identity_runtime_payload(contract))
+    stable = {key: value for key, value in contract.items() if key != "contract_sha256"}
+    contract["contract_sha256"] = package.canonical_sha256(stable)
+
+
+def refresh_image_identity_binding(image_input: dict[str, object], identity: Path) -> None:
+    binding = image_input["identity"]
+    assert isinstance(binding, dict)
+    manifest = json.loads((identity / "identity-manifest.json").read_text(encoding="utf-8"))
+    binding.update(
+        {
+            "bundle_sha256": package.identity_bundle_manifest(identity)["sha256"],
+            "manifest_sha256": manifest["manifest_sha256"],
+            "identity_inputs_sha256": manifest["identity_inputs_sha256"],
+            "identity_lock_sha256": package.sha256_file(identity / "identity-lock.json"),
+            "soul_sha256": package.sha256_file(identity / "SOUL.md"),
+            "identity_view_sha256": package.sha256_file(identity / "identity.json"),
+        }
+    )
+    stable = {key: value for key, value in image_input.items() if key != "input_sha256"}
+    image_input["input_sha256"] = package.canonical_sha256(stable)
+
+
+def image_input_fixture(tmp_path: Path) -> dict[str, object]:
+    artifact, receipt = artifact_fixture(tmp_path)
+    identity, _contract = identity_fixture(tmp_path, artifact, receipt)
+    return package.build_image_input(
+        artifact=artifact,
+        receipt=receipt,
+        identity=identity,
+        ca=ROOT / "ops" / "gcp-teleo-pgvector-standby-server-ca.pem",
+        source_revision=REVISION,
+        dockerfile=ROOT / "docker" / "leoclean-nosend" / "Dockerfile",
+        entrypoint=ROOT / "docker" / "leoclean-nosend" / "entrypoint.py",
+        package_contract=ROOT / "ops" / "gcp_leoclean_nosend_package.py",
+        contract_source_root=ROOT,
+        identity_source_root=ROOT,
+    )
+
+
+def test_image_input_binds_exact_runtime_identity_and_staging_target(tmp_path: Path) -> None:
+    value = image_input_fixture(tmp_path)
+    package.validate_image_input(value)
+
+    assert value["target"] == {
+        "project": "teleo-501523",
+        "zone": "europe-west6-a",
+        "instance": "teleo-staging-1",
+        "service": "leoclean-gcp-nosend.service",
+        "service_account": "sa-teleo-staging-vm@teleo-501523.iam.gserviceaccount.com",
+        "platform": "linux/amd64",
+    }
+    assert value["runtime"]["python_version"] == "3.11.9"
+    assert value["runtime"]["postgres_version"] == "16.14"
+    assert value["identity"]["canonical_authority_granted"] is False
+    database_policy = value["identity"]["runtime_contract"]["permissions"]["runtime_policy"]["database"]
+    assert database_policy == {
+        "direct_canonical_writes": "denied",
+        "direct_stage_table_writes": "denied",
+        "proposal_staging": "function_only",
+        "proposal_function": "kb_stage.stage_leoclean_proposal(text,text,text,text,jsonb)",
+        "live_effective_permissions_proven": False,
+    }
+    assert value["base_images"]["postgres"].endswith(
+        "@sha256:92620daddcd947f8d5ab5ba66e848702fe443d87fed30c4cea8e389fd78dfc55"
+    )
+
+
+def test_artifact_requires_release_receipt_exact_revision_and_unchanged_content(tmp_path: Path) -> None:
+    artifact, receipt = artifact_fixture(tmp_path)
+    package.validate_artifact(artifact, receipt, REVISION)
+
+    development = json.loads(receipt.read_text(encoding="utf-8"))
+    development["verification_mode"] = "development"
+    write_json(receipt, development, mode=0o644)
+    with pytest.raises(package.PackageError, match="not release"):
+        package.validate_artifact(artifact, receipt, REVISION)
+
+    development["verification_mode"] = "release"
+    write_json(receipt, development, mode=0o644)
+    (artifact / "bootstrap.py").write_text("mutated\n", encoding="utf-8")
+    (artifact / "bootstrap.py").chmod(0o755)
+    with pytest.raises(package.PackageError, match="content manifest drifted"):
+        package.validate_artifact(artifact, receipt, REVISION)
+
+
+def test_artifact_binds_runtime_probe_to_the_exact_packaged_config(tmp_path: Path) -> None:
+    artifact, receipt = artifact_fixture(tmp_path)
+    value = json.loads(receipt.read_text(encoding="utf-8"))
+    value["runtime_probe"]["config_sha256"] = "9" * 64
+    write_json(receipt, value, mode=0o644)
+
+    with pytest.raises(package.PackageError, match="runtime probe config differs"):
+        package.validate_artifact(artifact, receipt, REVISION)
+
+
+def test_identity_bundle_rejects_drift_extra_files_and_false_authority(tmp_path: Path) -> None:
+    artifact, receipt = artifact_fixture(tmp_path)
+    identity, contract = identity_fixture(tmp_path, artifact, receipt)
+    validate = lambda root: package.validate_identity_bundle(  # noqa: E731
+        root,
+        expected_revision=REVISION,
+        expected_identity_contract=contract,
+        source_root=ROOT,
+    )
+    validate(identity)
+
+    (identity / "SOUL.md").write_text("drift\n", encoding="utf-8")
+    (identity / "SOUL.md").chmod(0o600)
+    with pytest.raises(package.PackageError, match="view derivation drifted"):
+        validate(identity)
+
+    identity, _contract = identity_fixture(tmp_path / "second", artifact, receipt)
+    (identity / "extra.txt").write_text("extra\n", encoding="utf-8")
+    with pytest.raises(package.PackageError, match="entry set"):
+        validate(identity)
+
+    identity, _contract = identity_fixture(tmp_path / "third", artifact, receipt)
+    manifest_path = identity / "identity-manifest.json"
+    manifest = json.loads(manifest_path.read_text(encoding="utf-8"))
+    manifest["identity_inputs"]["canonical_database"] = {
+        "authority": "synthetic_noncanonical_fixture",
+        "canonical_authority_granted": True,
+    }
+    stable = {key: value for key, value in manifest.items() if key != "manifest_sha256"}
+    manifest["manifest_sha256"] = package.canonical_sha256(stable)
+    write_json(manifest_path, manifest)
+    lock_path = identity / "identity-lock.json"
+    lock = json.loads(lock_path.read_text(encoding="utf-8"))
+    lock["manifest_sha256"] = manifest["manifest_sha256"]
+    write_json(lock_path, lock)
+    with pytest.raises(package.PackageError, match="semantic/source validation"):
+        validate(identity)
+
+
+@pytest.mark.parametrize(
+    ("path", "value"),
+    [
+        (("source_commit",), "2" * 40),
+        (("runtime", "hermes_source_sha256"), "7" * 64),
+        (("model", "default_model"), "attacker/model"),
+        (("skills", "content_sha256"), "8" * 64),
+        (("permissions", "runtime_policy", "general_network_egress", "denial_tested"), True),
+    ],
+)
+def test_fully_rehashed_identity_runtime_mutations_fail_closed(
+    tmp_path: Path, path: tuple[str, ...], value: object
+) -> None:
+    artifact, receipt = artifact_fixture(tmp_path)
+    identity, contract = identity_fixture(tmp_path, artifact, receipt)
+    manifest_path = identity / "identity-manifest.json"
+    manifest = json.loads(manifest_path.read_text(encoding="utf-8"))
+    target = manifest["identity_inputs"]
+    for key in path[:-1]:
+        target = target[key]
+    target[path[-1]] = value
+    if path == ("source_commit",):
+        manifest["identity_inputs"]["runtime"]["teleo_git_head"] = value
+    write_json(manifest_path, manifest)
+    fully_rehash_identity_bundle(identity)
+
+    with pytest.raises(package.PackageError):
+        package.validate_identity_bundle(
+            identity,
+            expected_revision=REVISION,
+            expected_identity_contract=contract,
+            source_root=ROOT,
+        )
+
+
+def test_custom_no_send_policy_requires_explicit_validator_authority(tmp_path: Path) -> None:
+    artifact, receipt = artifact_fixture(tmp_path)
+    identity, contract = identity_fixture(tmp_path, artifact, receipt)
+    manifest = json.loads((identity / "identity-manifest.json").read_text(encoding="utf-8"))
+
+    with pytest.raises(package.strict_identity_manifest.IdentityManifestError, match="exact local readback policy"):
+        package.strict_identity_manifest.validate_identity_manifest(manifest)
+    package.strict_identity_manifest.validate_identity_manifest(
+        manifest,
+        expected_runtime_policy_value=contract["permissions"]["runtime_policy"],
+    )
+    with pytest.raises(package.strict_identity_manifest.IdentityManifestError, match="expected identity runtime policy"):
+        package.strict_identity_manifest.validate_identity_manifest(manifest, expected_runtime_policy_value={})
+
+
+@pytest.mark.parametrize("mutation", ["legacy_policy", "malformed_behavior_inputs"])
+def test_runtime_contract_adapter_rejects_weaker_or_malformed_contracts(tmp_path: Path, mutation: str) -> None:
+    artifact, receipt = artifact_fixture(tmp_path)
+    _identity, contract = identity_fixture(tmp_path, artifact, receipt)
+    mutated = copy.deepcopy(contract)
+    if mutation == "legacy_policy":
+        mutated["permissions"]["runtime_policy"] = package.strict_identity_manifest.expected_runtime_policy()
+    else:
+        del mutated["behavior_inputs"]["tool_surface"]
+    fully_rehash_identity_contract(mutated)
+
+    with pytest.raises(package.strict_identity_manifest.IdentityManifestError):
+        package.strict_identity_manifest.identity_runtime_inputs_from_contract(mutated)
+
+
+def test_identity_source_content_drift_fails_closed(tmp_path: Path) -> None:
+    artifact, receipt = artifact_fixture(tmp_path)
+    identity, contract = identity_fixture(tmp_path, artifact, receipt)
+    source_root = tmp_path / "source-root"
+    fixture_relative = Path("fixtures/working-leo/leo-identity-v1")
+    shutil.copytree(ROOT / fixture_relative, source_root / fixture_relative)
+    package.validate_identity_bundle(
+        identity,
+        expected_revision=REVISION,
+        expected_identity_contract=contract,
+        source_root=source_root,
+    )
+    constitution = source_root / fixture_relative / "leo-constitution-v1.json"
+    constitution.write_text(constitution.read_text(encoding="utf-8") + "\n", encoding="utf-8")
+
+    with pytest.raises(package.PackageError, match="semantic/source validation"):
+        package.validate_identity_bundle(
+            identity,
+            expected_revision=REVISION,
+            expected_identity_contract=contract,
+            source_root=source_root,
+        )
+
+
+def test_identity_source_intermediate_symlink_fails_closed(tmp_path: Path) -> None:
+    artifact, receipt = artifact_fixture(tmp_path)
+    identity, contract = identity_fixture(tmp_path, artifact, receipt)
+    outside = tmp_path / "outside"
+    shutil.copytree(ROOT / "fixtures", outside / "fixtures")
+    source_root = tmp_path / "source-root"
+    source_root.mkdir()
+    (source_root / "fixtures").symlink_to(outside / "fixtures", target_is_directory=True)
+
+    with pytest.raises(package.PackageError, match="semantic/source validation"):
+        package.validate_identity_bundle(
+            identity,
+            expected_revision=REVISION,
+            expected_identity_contract=contract,
+            source_root=source_root,
+        )
+
+
+@pytest.mark.parametrize("mutation", ["skills", "behavior_artifact_and_uv"])
+def test_installed_validation_rederives_fully_rehashed_identity_contract(tmp_path: Path, mutation: str) -> None:
+    artifact, receipt = artifact_fixture(tmp_path)
+    identity, _contract = identity_fixture(tmp_path, artifact, receipt)
+    image_input = package.build_image_input(
+        artifact=artifact,
+        receipt=receipt,
+        identity=identity,
+        ca=ROOT / "ops" / "gcp-teleo-pgvector-standby-server-ca.pem",
+        source_revision=REVISION,
+        dockerfile=ROOT / "docker" / "leoclean-nosend" / "Dockerfile",
+        entrypoint=ROOT / "docker" / "leoclean-nosend" / "entrypoint.py",
+        package_contract=ROOT / "ops" / "gcp_leoclean_nosend_package.py",
+        contract_source_root=ROOT,
+        identity_source_root=ROOT,
+    )
+    embedded = image_input["identity"]["runtime_contract"]
+    manifest_path = identity / "identity-manifest.json"
+    manifest = json.loads(manifest_path.read_text(encoding="utf-8"))
+    if mutation == "skills":
+        embedded["skills"] = {"content_sha256": "8" * 64, "file_count": 999}
+        manifest["identity_inputs"]["skills"] = copy.deepcopy(embedded["skills"])
+    else:
+        embedded["behavior_inputs"]["artifact_sha256"] = "8" * 64
+        embedded["behavior_inputs"]["uv_lock_sha256"] = "9" * 64
+        image_input["runtime"]["artifact_sha256"] = "8" * 64
+        image_input["runtime"]["uv_lock_sha256"] = "9" * 64
+    fully_rehash_identity_contract(embedded)
+    manifest["identity_inputs"]["runtime"]["identity_runtime_sha256"] = embedded["runtime"][
+        "identity_runtime_sha256"
+    ]
+    write_json(manifest_path, manifest)
+    fully_rehash_identity_bundle(identity)
+    refresh_image_identity_binding(image_input, identity)
+
+    package.validate_image_input(image_input)
+    with pytest.raises(package.PackageError):
+        package.validate_installed_runtime_identity(
+            image_input,
+            artifact=artifact,
+            receipt=receipt,
+            identity=identity,
+            contract_source_root=ROOT,
+            identity_source_root=ROOT,
+        )
+
+
+def test_release_requires_exact_digest_repository_and_strict_schema(tmp_path: Path) -> None:
+    image_input = image_input_fixture(tmp_path)
+    reference = f"{package.IMAGE_REPOSITORY}@sha256:{IMAGE_DIGEST}"
+    release = package.build_release_descriptor(image_input, reference)
+    package.validate_release_descriptor(release)
+
+    assert release["image"]["reference"] == reference
+    assert release["image"]["digest"] == f"sha256:{IMAGE_DIGEST}"
+    assert release["image"]["platform"] == "linux/amd64"
+
+    with pytest.raises(package.PackageError, match="sha256 digest"):
+        package.build_release_descriptor(image_input, f"{package.IMAGE_REPOSITORY}:latest")
+    with pytest.raises(package.PackageError, match="staging repository"):
+        package.build_release_descriptor(
+            image_input,
+            f"europe-west6-docker.pkg.dev/teleo-501523/teleo/other@sha256:{IMAGE_DIGEST}",
+        )
+
+    extra = copy.deepcopy(release)
+    extra["unexpected"] = True
+    with pytest.raises(package.PackageError, match="fields are not exact"):
+        package.validate_release_descriptor(extra)
+
+    wrong_label = copy.deepcopy(release)
+    wrong_label["image"]["labels"]["livingip.artifact.sha256"] = "9" * 64
+    stable = {key: value for key, value in wrong_label.items() if key != "release_sha256"}
+    wrong_label["release_sha256"] = package.canonical_sha256(stable)
+    with pytest.raises(package.PackageError, match="labels are not exact"):
+        package.validate_release_descriptor(wrong_label)
+
+
+def test_rendered_unit_is_isolated_digest_pinned_and_has_no_credential_or_transport_surface(tmp_path: Path) -> None:
+    image_input = image_input_fixture(tmp_path)
+    release = package.build_release_descriptor(
+        image_input,
+        f"{package.IMAGE_REPOSITORY}@sha256:{IMAGE_DIGEST}",
+    )
+    unit = package.render_systemd_unit(release)
+
+    assert "leoclean-gcp-nosend" in unit
+    assert f"@sha256:{IMAGE_DIGEST}" in unit
+    assert "--pull=never" in unit
+    assert "--read-only" in unit
+    assert "--cap-drop=ALL" in unit
+    assert "--cap-add=CHOWN" in unit
+    assert "--cap-add=SETGID" in unit
+    assert "--cap-add=SETUID" in unit
+    assert "--cap-add=SETPCAP" in unit
+    assert "--network=bridge" in unit
+    assert "--tmpfs=/run/leoclean-profile:" in unit
+    assert "--tmpfs=/var/lib/postgresql/data:ro,noexec,nosuid,nodev,size=64k,uid=0,gid=0,mode=0000" in unit
+    assert "NoNewPrivileges=yes" in unit
+    assert "EnvironmentFile" not in unit
+    assert "LoadCredential" not in unit
+    assert "docker.sock" not in unit
+    assert "--publish" not in unit
+    assert "teleo-prod-1" not in unit
+    assert "77.42.65.182" not in unit
+    assert "telegram" not in unit.casefold()
+
+
+def test_prepare_context_is_secret_free_normalized_and_idempotently_refuses_overwrite(tmp_path: Path) -> None:
+    artifact, receipt = artifact_fixture(tmp_path)
+    identity, _contract = identity_fixture(tmp_path, artifact, receipt)
+    output = tmp_path / "context"
+    result = package.prepare_context(
+        repo_root=ROOT,
+        artifact=artifact,
+        receipt=receipt,
+        identity=identity,
+        identity_source_root=ROOT,
+        output=output,
+        source_revision=REVISION,
+    )
+
+    package.validate_image_input(result)
+    assert sorted(path.name for path in output.iterdir()) == [
+        "Dockerfile",
+        "artifact",
+        "artifact-receipt.json",
+        "build-args.json",
+        "cloudsql-server-ca.pem",
+        "entrypoint.py",
+        "identity",
+        "identity-sources",
+        "image-input.json",
+        "package_contract.py",
+        "scripts",
+    ]
+    assert all((output / "identity" / name).stat().st_mode & 0o777 == 0o600 for name in package.IDENTITY_FILES)
+    encoded = (output / "image-input.json").read_text(encoding="utf-8").casefold()
+    assert "telegram_bot_token" not in encoded
+    assert "postgres-password" not in encoded
+
+    with pytest.raises(package.PackageError, match="already exists"):
+        package.prepare_context(
+            repo_root=ROOT,
+            artifact=artifact,
+            receipt=receipt,
+            identity=identity,
+            identity_source_root=ROOT,
+            output=output,
+            source_revision=REVISION,
+        )
+
+
+def test_dockerfile_has_only_digest_pinned_bases_and_no_floating_package_install() -> None:
+    dockerfile = (ROOT / "docker" / "leoclean-nosend" / "Dockerfile").read_text(encoding="utf-8")
+    from_lines = [line for line in dockerfile.splitlines() if line.startswith("FROM ")]
+
+    assert len(from_lines) == 3
+    assert all("@sha256:" in line for line in from_lines)
+    assert "# syntax=" not in dockerfile
+    assert "postgres:16.14-bookworm@sha256:92620d" in dockerfile
+    assert "python:3.11.9-slim-bookworm@sha256:8fb099" in dockerfile
+    assert "uv:0.9.30@sha256:538e0b" in dockerfile
+    assert "apt-get" not in dockerfile
+    assert "pip install" not in dockerfile
+    assert "uv sync" in dockerfile
+    assert "--frozen" in dockerfile
+    assert "--no-editable" in dockerfile
+    assert "HEALTHCHECK" in dockerfile
+    assert "verify-build" in dockerfile
+    assert "identity -type f -exec chmod 0444" in dockerfile
+
+
+def test_entrypoint_prepares_exact_profile_drops_privileges_and_health_checks_pid_one() -> None:
+    source = (ROOT / "docker" / "leoclean-nosend" / "entrypoint.py").read_text(encoding="utf-8")
+
+    assert "os.setgroups([])" in source
+    assert "os.setgid(contract.RUNTIME_GID)" in source
+    assert "os.setuid(contract.RUNTIME_UID)" in source
+    assert "pr_capbset_drop = 24" in source
+    assert "reverse=True" in source
+    assert source.index("os.chown(runtime_identity, 0, 0)") < source.index(
+        "os.chown(PROFILE, contract.RUNTIME_UID, contract.RUNTIME_GID)"
+    )
+    assert 'value.get("process_id") == 1' in source
+    assert "expected_binfmt = [expected_native[0], *expected_native]" in source
+    assert "cmdline in (expected_native, expected_binfmt)" in source
+    assert "contract.NO_SEND_TOOL_SURFACE" in source
+    assert 'image_input["runtime"]["config_sha256"]' in source
+    assert "gateway_adapter_count" in source
+    assert "os.execve" in source
+    assert "OPENROUTER_API_KEY" not in source
+    assert "TELEGRAM_BOT_TOKEN" not in source
+
+
+def test_entrypoint_readiness_binds_packaged_and_live_config(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
+    entrypoint = load_entrypoint_module(tmp_path)
+    profile = tmp_path / "profile"
+    profile.mkdir()
+    config = profile / "config.yaml"
+    config.write_text("model: reviewed\n", encoding="utf-8")
+    expected = entrypoint.contract.sha256_file(config)
+    image_input = {"runtime": {"config_sha256": expected}}
+    readiness = {
+        "config_sha256": expected,
+        "tool_surface": copy.deepcopy(entrypoint.contract.NO_SEND_TOOL_SURFACE),
+    }
+    monkeypatch.setattr(entrypoint, "PROFILE", profile)
+
+    entrypoint._validate_readiness_behavior(image_input, readiness)
+
+    wrong_receipt = copy.deepcopy(readiness)
+    wrong_receipt["config_sha256"] = "9" * 64
+    with pytest.raises(entrypoint.EntrypointError, match="readiness config"):
+        entrypoint._validate_readiness_behavior(image_input, wrong_receipt)
+
+    config.write_text("model: drifted\n", encoding="utf-8")
+    with pytest.raises(entrypoint.EntrypointError, match="profile config drifted"):
+        entrypoint._validate_readiness_behavior(image_input, readiness)
+
+    drift_hash = entrypoint.contract.sha256_file(config)
+    both_drifted = {**readiness, "config_sha256": drift_hash}
+    with pytest.raises(entrypoint.EntrypointError, match="readiness config"):
+        entrypoint._validate_readiness_behavior(image_input, both_drifted)
+
+    config.unlink()
+    with pytest.raises(entrypoint.EntrypointError, match="missing or unsafe"):
+        entrypoint._validate_readiness_behavior(image_input, readiness)
+
+
+@pytest.mark.parametrize(
+    ("mutation", "value"),
+    [
+        ("extra", True),
+        ("registry_sealed", False),
+        ("send_message_present", True),
+        ("terminal_handler", "unrestricted"),
+        ("terminal_restricted_to", "shell"),
+        ("toolset", "default"),
+        ("tools", ["skill_view", "skills_list", "terminal", "send_message"]),
+        ("plugins", []),
+    ],
+)
+def test_entrypoint_readiness_rejects_exact_tool_surface_drift(
+    tmp_path: Path,
+    monkeypatch: pytest.MonkeyPatch,
+    mutation: str,
+    value: object,
+) -> None:
+    entrypoint = load_entrypoint_module(tmp_path)
+    profile = tmp_path / "profile"
+    profile.mkdir()
+    config = profile / "config.yaml"
+    config.write_text("model: reviewed\n", encoding="utf-8")
+    expected = entrypoint.contract.sha256_file(config)
+    image_input = {"runtime": {"config_sha256": expected}}
+    surface = copy.deepcopy(entrypoint.contract.NO_SEND_TOOL_SURFACE)
+    surface[mutation] = value
+    readiness = {"config_sha256": expected, "tool_surface": surface}
+    monkeypatch.setattr(entrypoint, "PROFILE", profile)
+
+    with pytest.raises(entrypoint.EntrypointError, match="tool surface drifted"):
+        entrypoint._validate_readiness_behavior(image_input, readiness)
+
+
+def test_entrypoint_readiness_rejects_plugin_field_drift(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
+    entrypoint = load_entrypoint_module(tmp_path)
+    profile = tmp_path / "profile"
+    profile.mkdir()
+    config = profile / "config.yaml"
+    config.write_text("model: reviewed\n", encoding="utf-8")
+    expected = entrypoint.contract.sha256_file(config)
+    monkeypatch.setattr(entrypoint, "PROFILE", profile)
+
+    for field, value in (
+        ("source", "project"),
+        ("tools", ["send_message"]),
+        ("hooks", ["pre_llm_call"]),
+        ("enabled", False),
+        ("error", "load failed"),
+    ):
+        surface = copy.deepcopy(entrypoint.contract.NO_SEND_TOOL_SURFACE)
+        surface["plugins"][0][field] = value
+        readiness = {"config_sha256": expected, "tool_surface": surface}
+        with pytest.raises(entrypoint.EntrypointError, match="tool surface drifted"):
+            entrypoint._validate_readiness_behavior(
+                {"runtime": {"config_sha256": expected}},
+                readiness,
+            )
+
+
+def test_entrypoint_loads_adjacent_contract_with_python_safe_path(tmp_path: Path) -> None:
+    image_root = tmp_path / "image"
+    image_root.mkdir()
+    entrypoint = image_root / "entrypoint.py"
+    shutil.copy2(ROOT / "docker" / "leoclean-nosend" / "entrypoint.py", entrypoint)
+    shutil.copy2(ROOT / "ops" / "gcp_leoclean_nosend_package.py", image_root / "package_contract.py")
+    scripts = image_root / "scripts"
+    scripts.mkdir()
+    for relative in package.IDENTITY_CONTRACT_SOURCES:
+        shutil.copy2(ROOT / relative, image_root / relative)
+    environment = {**os.environ, "PYTHONSAFEPATH": "1"}
+
+    result = subprocess.run(
+        [sys.executable, "-P", str(entrypoint), "verify"],
+        cwd=tmp_path,
+        env=environment,
+        check=False,
+        capture_output=True,
+        text=True,
+        timeout=10,
+    )
+
+    assert result.returncode == 65
+    assert json.loads(result.stderr) == {"error": "container_contract_failed", "status": "fail"}
+    assert "Traceback" not in result.stderr

tests/test_leoclean_nosend_runtime.py

@@ -7,6 +7,7 @@ import copy
 import importlib.util
 import io
 import json
+import os
 import re
 import shutil
 import subprocess
@@ -52,6 +53,7 @@ def copy_profile(tmp_path: Path, bootstrap) -> Path:
     bootstrap._runtime_root = lambda: runtime_root
     profile = tmp_path / "profile"
     shutil.copytree(runtime_root / "profile-template", profile)
+    profile.chmod(0o700)
     return profile
 
 
@@ -100,6 +102,19 @@ def test_committed_lock_and_contract_are_exactly_pinned() -> None:
     }
 
 
+def test_compiled_immutable_profile_is_readable_after_privilege_drop(tmp_path: Path) -> None:
+    runtime_root = tmp_path / "runtime"
+    runtime_root.mkdir()
+
+    compiler._copy_profile(ROOT, runtime_root, LOCK)
+
+    profile = runtime_root / "profile-template"
+    assert profile.stat().st_mode & 0o777 == 0o555
+    assert (profile / "config.yaml").stat().st_mode & 0o777 == 0o444
+    assert all(path.stat().st_mode & 0o004 for path in profile.rglob("*") if path.is_file())
+    assert all(path.stat().st_mode & 0o001 for path in profile.rglob("*") if path.is_dir())
+
+
 def test_profile_sources_keep_gcp_adapters_out_of_vps_deploy_paths() -> None:
     by_target = {entry["target"]: entry for entry in LOCK["profile_sources"]}
     assert by_target["bin/cloudsql_memory_tool.py"]["source"] == (
@@ -219,6 +234,67 @@ def test_profile_template_and_runtime_state_are_exact(tmp_path: Path) -> None:
     bootstrap._validate_profile(profile, CONTRACT, template=False)
 
 
+def test_service_readiness_is_atomic_sanitized_and_identity_bound(tmp_path: Path) -> None:
+    bootstrap = load_bootstrap()
+    profile = copy_profile(tmp_path, bootstrap)
+    for name in CONTRACT["profile"]["required_runtime_entries"]:
+        (profile / name).write_text(f"{name}\n", encoding="utf-8")
+    state = profile / "state"
+    state.mkdir()
+    readiness = state / "readiness.json"
+    tool_surface = {
+        "registry_sealed": True,
+        "send_message_present": False,
+        "tools": ["skill_view", "skills_list", "terminal"],
+    }
+
+    bootstrap._write_readiness(
+        profile,
+        readiness,
+        CONTRACT,
+        SimpleNamespace(adapters={}),
+        tool_surface,
+    )
+
+    payload = json.loads(readiness.read_text(encoding="utf-8"))
+    assert payload == {
+        "schema": bootstrap.READINESS_SCHEMA,
+        "status": "ready",
+        "process_id": os.getpid(),
+        "runtime_mode": "gcp_staging_no_send",
+        "config_sha256": bootstrap._sha256(profile / "config.yaml"),
+        "gateway_adapter_count": 0,
+        "connected_platforms": [],
+        "tool_surface": tool_surface,
+        "identity_sha256": {
+            name: bootstrap._sha256(profile / name)
+            for name in ("SOUL.md", "identity-lock.json", "identity-manifest.json")
+        },
+    }
+    assert readiness.stat().st_mode & 0o777 == 0o600
+    assert str(profile) not in readiness.read_text(encoding="utf-8")
+
+
+def test_service_readiness_rejects_escape_and_symlink(tmp_path: Path) -> None:
+    bootstrap = load_bootstrap()
+    profile = copy_profile(tmp_path, bootstrap)
+    for name in CONTRACT["profile"]["required_runtime_entries"]:
+        (profile / name).write_text("{}\n", encoding="utf-8")
+    state = profile / "state"
+    state.mkdir()
+    runner = SimpleNamespace(adapters={})
+
+    with pytest.raises(bootstrap.NoSendContractError, match="inside profile state"):
+        bootstrap._write_readiness(profile, tmp_path / "readiness.json", CONTRACT, runner, {})
+
+    target = state / "target.json"
+    target.write_text("{}\n", encoding="utf-8")
+    linked = state / "readiness.json"
+    linked.symlink_to(target)
+    with pytest.raises(bootstrap.NoSendContractError, match="symlink"):
+        bootstrap._write_readiness(profile, linked, CONTRACT, runner, {})
+
+
 @pytest.mark.parametrize(
     ("section", "field", "unsafe"),
     [
@@ -240,6 +316,7 @@ def test_profile_rejects_learning_policy_drift(
     bootstrap = load_bootstrap()
     profile = copy_profile(tmp_path, bootstrap)
     config_path = profile / "config.yaml"
+    config_path.chmod(0o600)
     config = yaml.safe_load(config_path.read_text(encoding="utf-8"))
     config[section][field] = unsafe
     config_path.write_text(yaml.safe_dump(config, sort_keys=False), encoding="utf-8")
@@ -254,6 +331,7 @@ def test_profile_rejects_smart_model_routing(tmp_path: Path) -> None:
     bootstrap = load_bootstrap()
     profile = copy_profile(tmp_path, bootstrap)
     config_path = profile / "config.yaml"
+    config_path.chmod(0o600)
     config = yaml.safe_load(config_path.read_text(encoding="utf-8"))
     config["smart_model_routing"] = {
         "enabled": True,
@@ -750,6 +828,7 @@ def test_ci_runs_the_release_canary_and_retains_its_receipt() -> None:
     assert "--development" not in wrapper
     assert "--allow-dirty" not in wrapper
     assert 'rm -f "$receipt"' in wrapper
+    assert 'chmod -R u+w "$temporary"' in wrapper
     assert 'repeat_artifact="$temporary/artifact-repeat"' in wrapper
     assert "first != second" in wrapper
     assert "compile-primary.json" in wrapper