Merge final Cloud SQL boundary into source-first V3 integration

deploy/sync-gcp-leoclean-runtime.sh

@@ -857,12 +857,33 @@ expected = {
     "current_tier": "T3_live_readonly",
     "mode": "live_private_gcp_staging",
     "required_tier": "T3_live_readonly",
+    "schema_version": 2,
     "status": "pass",
 }
 mismatched = sorted(key for key, value in expected.items() if receipt.get(key) != value)
+expected_safety = {
+    "arbitrary_database_writes_denied": False,
+    "canonical_relation_writes_denied": True,
+    "direct_stage_table_writes_denied": True,
+    "production_eligible": False,
+    "proposal_staging": "function_only",
+    "public_large_object_mutation_residual": "present",
+    "staging_only": True,
+}
+safety = receipt.get("safety")
+if not isinstance(safety, dict):
+    mismatched.append("safety")
+else:
+    mismatched.extend(
+        f"safety.{key}" for key, value in expected_safety.items() if safety.get(key) != value
+    )
 if mismatched:
     raise SystemExit("permission receipt contract mismatch: " + ",".join(mismatched))
-print(json.dumps({"permission_receipt": "validated", "validated_fields": sorted(expected)}, sort_keys=True))
+print(json.dumps({
+    "permission_receipt": "validated",
+    "validated_fields": sorted(expected),
+    "validated_safety_fields": sorted(expected_safety),
+}, sort_keys=True))
 PY
 }
 

docs/workstreams/gcp-leoclean/goals.md

@@ -1,135 +1,78 @@
-# GCP leoclean migration goal
+# GCP-native Leo goal
 
-## Objective
+## Outcome
 
-Deploy and prove Leo's least-privilege Hermes/leoclean runtime in GCP staging
+Run Leo's Hermes `leoclean` runtime as a fresh GCP service, backed by the V3
-against canonical Cloud SQL, establish behavioral and database parity, connect
+LivingIP knowledge base in canonical Cloud SQL, and route Telegram directly to
-the protected Observatory to canonical/proposal/vector/runtime receipts, then
+that verified GCP service.
-cut over safely and retire the VPS after soak.
 
-## Architecture baseline
+This is not a VPS lift-and-shift. The existing VPS is never copied into or run
+inside GCP. It remains untouched as a temporary comparison and rollback source
+until the GCP service has passed soak and restore testing.
 
-- Hermes with the `leoclean` profile remains Leo's intended runtime.
+## Target architecture
-- Cloud SQL is canonical PostgreSQL authority.
+
-- The protected Observatory is a separate frontend surface.
+- An immutable `leoclean` image is pulled from Artifact Registry by digest.
-- The VPS remains untouched until an explicitly approved, parity-proven
+- A dedicated GCP service runs Hermes with the `leoclean` profile.
-  cutover and retirement sequence.
+- Cloud SQL PostgreSQL 16 is the canonical V3 knowledge authority.
+- Secret Manager supplies only the scoped `leoclean_kb_runtime` credential.
+- Leo reads canonical knowledge and stages proposals only through
+  `kb_stage.stage_leoclean_proposal(...)`.
+- Telegram delivery terminates at the verified GCP service.
+- The protected Observatory remains a separate follow-up surface over
+  canonical, proposal, vector, and runtime receipts.
 
 ## Current state
 
-- Complete: PR #179 least-privilege PostgreSQL authority is merged on main.
+- Current main/base: `832f0c3f47abe3eded61f4be585fde3c4f00b91a`.
-- Complete: the revision-bound, structurally no-send runtime from PR #183 is
+- The GCP VM, private Cloud SQL instance, immutable image, and scoped Secret
-  merged into `main` at `2d88c9765bd9b43641d1efb04a1118c5ce0bce39`.
+  Manager secret exist.
-- Complete: PR #182 head
+- The Cloud SQL-compatible least-privilege correction is implemented on
-  `3e59845454f5315e955734db7fe598ec4ba4cf70` was merged with merge commit
+  `fix/cloudsql-large-object-boundary` and is awaiting publication and human
-  `94a8160ceeb6821fd113f4c843f024d5de99939a` after all five checks passed.
+  review.
-- Complete: PR #202 remained the exact reviewed three-file QEMU/PID-1
+- Exact local evidence: 2,601 repository tests passed with 2 expected skips;
-  correction at `ffe170007d0e20d77a5208263e5b33fc999139fc` after retargeting
+  the digest-pinned, network-disabled PostgreSQL 16.14 lifecycle canary passed
-  to `main`, then merged with merge commit
+  2 tests; Ruff, Bash syntax, and `git diff --check` passed.
-  `383996a43beebef57cf84a2ced53213e28a41dca` after all five checks passed.
+- The `leoclean_kb_runtime` role has not yet been successfully provisioned in
-- PR #193 is source material only. It must not be merged wholesale; only the
+  Cloud SQL.
-  required identity-binding and OCI-lifecycle corrections belong in the narrow
+- The VM service account has not yet been granted access to the scoped secret.
-  #182/#183 stack.
+- The no-send GCP service is not installed or running.
-- Complete through PR #182 head `3e59845`: the narrow package
+- V3 knowledge reconstruction and promotion have not yet been reconciled from
-  correction binds every packaged and synthetic-identity source byte to its
+  PR #220 into canonical Cloud SQL.
-  reviewed Git blob; container health binds the packaged/live config plus the
+- Telegram still does not route to GCP.
-  exact sealed no-send tool and plugin surface; and the disposable image smoke
-  also binds the implicit root bootstrap user, entrypoint, command, healthcheck,
-  and stop signal. The focused package/runtime/identity/OCI suite passed 204
-  tests, the repository suite passed 2,338 with 3 skips, and one exact-revision
-  synthetic `linux/amd64` OCI lifecycle proved UID/GID `65532`, zero runtime
-  capabilities, no-send health, and complete candidate cleanup on the arm64 Mac
-  Docker daemon. No GCP or live service was contacted.
-- PR #206 at exact head `e42d635523fcb90ecdd504b0f0290df7015991e7`
-  makes release finalization pull and inspect the exact immutable staging
-  digest. Its five checks pass, but it must not merge unchanged: PR #209's
-  final repair at `d2ad4f7c964ff70f01be65ad57290b1346153f95` is the required
-  corrective layer that prevents finalization from deleting daemon-global
-  digest references a Docker-config lock cannot exclusively own.
-- Published as draft PR #210: one structured, self-hashed build/push receipt
-  producer. It derives the content-bound candidate tag, manifest digest, and
-  config binding from validated pre/post-push daemon state; finalization
-  consumes that receipt instead of caller-supplied digest/config values. The
-  branch has been rebuilt directly on PR #209's final repair and carries the
-  same retained-digest rule through both build/push and finalization. No
-  daemon-global build tag, candidate tag, immutable digest reference, or
-  receipt pathname is removed automatically. The receipt is published and
-  consumed only through a private operator-owned directory with a retained
-  two-link, stable-inode contract; remote candidates are retained. The exact
-  #210 head belongs in the mutable PR evidence, not recursively inside its own
-  commit.
-- Current exact-head validation: 206 focused package/OCI tests and the full
-  repository suite at 2,480 passed with 3 expected skips; Ruff, Python compile,
-  and diff checks pass. Three independent advisory exact-head reviews approved
-  the receipt deletion/identity boundary. Tests mock Docker and perform no
-  registry push.
-- Current gate: human exact-head review remains pending on #206 and #210. PR
-  #209 must first correct #206's feature branch; corrected #206 must be
-  re-reviewed and merged before #210 is retargeted to `main` for its own human
-  review and merge.
-- In progress offline: a separate GCP staging preflight rebuilt on the final
-  #210 head. It checks exact control-plane and host prerequisites without
-  reading secrets, logging into PostgreSQL, deploying, or restarting. Its
-  bounded IAP/SSH path uses a fixed remote command and a maximum five-minute
-  key lifetime. Receipt publication now requires a pre-existing private
-  operator-owned directory, creates one mode-`0600` no-clobber/no-symlink
-  staged inode through a held directory descriptor, fsyncs it before a
-  no-clobber canonical hard link, and retains both links without automatic
-  cleanup. The host probe proves Docker loaded/active/running before its first
-  client call and unchanged service state, `MainPID`, and restart count after
-  the last, preventing socket activation from changing runtime state. No live
-  run is currently authorized.
-- Artifact Registry cleanup policy is not an additional hard gate for the
-  package: immutable tags prevent deletion of the retained tagged candidate.
-  The post-push/finalization path must still re-prove the exact retained
-  candidate tag-to-digest binding immediately before deployment.
-- Current offline evidence for the preflight extraction: 250 focused
-  preflight/package/OCI tests pass; Ruff, Python compilation, formatting, and
-  diff checks pass. The repository suite produced 2,504 passes and 5 expected
-  skips inside the desktop sandbox; all 18 sandbox-denied Docker, loopback, and
-  Swift-cache cases passed in a separately authorized 59-test local rerun,
-  reconciling the suite to 2,522 passes and 5 skips. No cloud or registry
-  endpoint was contacted.
-- Historical only: the superseded preflight branch records an earlier live
-  observation of `e2-standard-4` on `teleo-staging-europe-west6`
-  (`10.60.0.0/20`) with an external IPv4 attachment. That observation is not
-  current-revision proof and has not been refreshed in this slice. The checker
-  intentionally fails while any public attachment remains.
-- The older provisioning baseline names `e2-standard-2` and
-  `teleo-staging-subnet`; reconciling that declarative baseline is a separate
-  infrastructure change, not part of this preflight implementation.
-- Not yet live-proven: staging VM/container engine, IAM/Secret Manager, Cloud
-  SQL effective identity, restart behavior, or model/database parity.
 
-## Remaining order
+## Execution order
 
-1. Merge PR #209 into PR #206's feature branch, rerun exact-head checks, obtain
+1. Publish and review the Cloud SQL authority correction.
-   human review, and merge the corrected PR #206.
+2. Provision `leoclean_kb_runtime`, grant scoped-secret access, install the
-2. Update, review, retarget, and merge draft PR #210's structured build/push
+   immutable no-send service, and prove identity, permissions, restart, and
-   receipt producer and receipt-bound release-v3 finalizer.
+   rollback.
-3. Review and merge the bounded staging preflight; under separate live
+3. Extract the required V3 reconstruction changes from PR #220, rebuild or
-   authorization, run its control-plane and IAP/SSH observations and verify the
+   migrate into a separately identified canonical candidate database, and
-   staging repository/tag policy.
+   prove row/hash/source receipts before promotion.
-4. Under explicit live authorization, build and push one candidate, publish
+4. Bind the GCP runtime to the verified V3 canonical database and run database
-   its receipt, and independently finalize the exact digest.
+   and behavioral parity checks without Telegram sends.
-5. Review and merge the offline fail-closed
+5. Switch Telegram delivery to the GCP service, prove real response receipts,
-   installer/restart/verification/rollback slice that consumes the exact
+   soak, restart, and restore behavior, and retain an immediate rollback path.
-   inspection-bound release bundle; authorize its first staging execution
+6. Revoke obsolete runtime credentials after soak. VPS destruction remains a
-   separately.
+   separate explicit decision.
-6. Provision the scoped Cloud SQL role/secret and deploy the separate staging
-   service.
-7. Prove effective identity, reads, function-only proposal staging, denied
-   writes/escalation, receipts, restart stability, and rollback.
-8. Run behavioral and database parity backtesting.
-9. Connect the protected Observatory to canonical claims, proposals, pgvector
-   projection, and runtime receipts.
-10. Cut over, soak, restore-test, revoke VPS credentials, and retire the VPS.
 
-## Current live-action boundary
+## Authority and rollback boundaries
 
-No merge without explicit human approval, Artifact Registry push,
+- Database role provisioning can be rolled back independently by refencing
-GCP control-plane or IAP/SSH access, Cloud SQL/Secret Manager mutation,
+  the runtime role and removing its scoped secret access.
-service deployment/restart, Telegram change, production promotion, canonical
+- Service deployment can be rolled back independently to the prior immutable
-proposal apply, or VPS access is part of the current offline corrective slice.
+  image or stopped without changing canonical data.
-Any later authorization for the bounded preflight must explicitly accept that
+- V3 database promotion requires a pre-promotion backup and exact reconciliation
-the SSH access path may register a key with a maximum five-minute lifetime.
+  receipt; it does not imply Telegram cutover.
+- Telegram routing is the final reversible switch and happens only after the
+  no-send service, V3 data, restart, and rollback proofs pass.
+- Canonical proposal approval, Observatory deployment, vector-backend choice,
+  and irreversible VPS deletion are not part of the current deployment slice.
+
+## Human gates
+
+Database/security review, GCP role and IAM mutation, V3 canonical promotion,
+and Telegram cutover are distinct critical gates. Evidence from tests or agent
+review is advisory; the exact revision and live receipts must be presented to
+the human development lead at each gate.

hermes-agent/leoclean-bin/cloudsql_memory_tool.py

@@ -170,6 +170,57 @@ def runtime_cloudsdk_config() -> str:
 
 
 CLONE_READ_ONLY_PGOPTIONS = "-c default_transaction_read_only=on"
+RUNTIME_STANDARD_STRINGS_PGOPTIONS = "-c standard_conforming_strings=on"
+RUNTIME_READ_ONLY_PGOPTIONS = (
+    f"{RUNTIME_STANDARD_STRINGS_PGOPTIONS} {CLONE_READ_ONLY_PGOPTIONS}"
+)
+RUNTIME_SQL_READ_CAPABILITY = "read"
+RUNTIME_SQL_STAGE_PROPOSAL_CAPABILITY = "stage-proposal"
+RUNTIME_SQL_CAPABILITIES = frozenset(
+    {RUNTIME_SQL_READ_CAPABILITY, RUNTIME_SQL_STAGE_PROPOSAL_CAPABILITY}
+)
+MAX_RUNTIME_SQL_BYTES = 262_144
+RUNTIME_SQL_FORBIDDEN_KEYWORD = re.compile(
+    r"\b(?:alter|analyze|begin|call|checkpoint|cluster|comment|commit|copy|create|"
+    r"deallocate|delete|discard|do|drop|execute|grant|insert|into|listen|load|lock|merge|"
+    r"notify|prepare|refresh|reindex|release|reset|revoke|rollback|savepoint|security|"
+    r"set|start|truncate|update|vacuum)\b",
+    re.IGNORECASE,
+)
+RUNTIME_SQL_LARGE_OBJECT_MUTATOR = re.compile(
+    r"(?<![a-z0-9_])(?:(?:pg_catalog)\s*\.\s*)?"
+    r"(?:lo_creat|lo_create|lo_export|lo_from_bytea|lo_import|lo_open|lo_put|"
+    r"lo_truncate64|lo_truncate|lo_unlink|lowrite)\s*\(",
+    re.IGNORECASE,
+)
+RUNTIME_SQL_FORBIDDEN_FUNCTION = re.compile(
+    r"(?<![a-z0-9_])(?:(?:pg_catalog)\s*\.\s*)?"
+    r"(?:set_config|nextval|setval|pg_notify|"
+    r"pg_(?:try_)?advisory_(?:xact_)?lock(?:_shared)?|"
+    r"pg_advisory_unlock(?:_all|_shared)?|pg_cancel_backend|"
+    r"pg_terminate_backend|pg_logical_emit_message)\s*\(",
+    re.IGNORECASE,
+)
+RUNTIME_SQL_APPLICATION_FUNCTION = re.compile(
+    r"(?<![a-z0-9_])(public|kb_stage)\s*\.\s*([a-z_][a-z0-9_]*)\s*\(",
+    re.IGNORECASE,
+)
+RUNTIME_SQL_FUNCTION_CALL = re.compile(
+    r"(?<![a-z0-9_.])(?:(?P<schema>[a-z_][a-z0-9_]*)\s*\.\s*)?"
+    r"(?P<name>[a-z_][a-z0-9_]*)\s*\(",
+    re.IGNORECASE,
+)
+RUNTIME_SQL_NONFUNCTION_PAREN_KEYWORDS = frozenset(
+    {"as", "case", "exists", "filter", "in", "over", "select", "values", "when", "where", "with"}
+)
+RUNTIME_SQL_STAGE_PROPOSAL_FUNCTION = ("kb_stage", "stage_leoclean_proposal")
+RUNTIME_SQL_STAGE_PROPOSAL_ALLOWED_FUNCTIONS = frozenset(
+    {
+        (None, "jsonb_build_object"),
+        (None, "nullif"),
+        RUNTIME_SQL_STAGE_PROPOSAL_FUNCTION,
+    }
+)
 RECEIPTED_READ_COMMANDS = frozenset(
     {
         "search",
@@ -596,7 +647,180 @@ def password(args: argparse.Namespace) -> str:
     raise SystemExit("Clone credential mode requires an inherited test credential.")
 
 
-def run_psql(args: argparse.Namespace, sql: str, db: str | None = None) -> str:
+def _runtime_sql_code(sql: str) -> str:
+    """Return SQL code with literals/comments removed, rejecting inexact syntax."""
+
+    if not isinstance(sql, str) or not sql or "\x00" in sql:
+        raise SystemExit("Runtime SQL was rejected: expected a non-empty text statement.")
+    if "\r" in sql:
+        raise SystemExit("Runtime SQL was rejected: carriage returns are not allowed.")
+    if len(sql.encode("utf-8")) > MAX_RUNTIME_SQL_BYTES:
+        raise SystemExit("Runtime SQL was rejected: statement exceeds the reviewed size bound.")
+
+    code: list[str] = []
+    index = 0
+    block_depth = 0
+    state = "code"
+    while index < len(sql):
+        char = sql[index]
+        next_char = sql[index + 1] if index + 1 < len(sql) else ""
+
+        if state == "line-comment":
+            if char == "\n":
+                code.append("\n")
+                state = "code"
+            else:
+                code.append(" ")
+            index += 1
+            continue
+
+        if state == "block-comment":
+            if char == "/" and next_char == "*":
+                block_depth += 1
+                code.extend((" ", " "))
+                index += 2
+            elif char == "*" and next_char == "/":
+                block_depth -= 1
+                code.extend((" ", " "))
+                index += 2
+                if block_depth == 0:
+                    state = "code"
+            else:
+                code.append("\n" if char == "\n" else " ")
+                index += 1
+            continue
+
+        if state in {"single-quote", "escape-string"}:
+            if state == "escape-string" and char == "\\":
+                code.append(" ")
+                if next_char:
+                    code.append("\n" if next_char == "\n" else " ")
+                    index += 2
+                else:
+                    index += 1
+            elif char == "'" and next_char == "'":
+                code.extend((" ", " "))
+                index += 2
+            elif char == "'":
+                code.append(" ")
+                state = "code"
+                index += 1
+            else:
+                code.append("\n" if char == "\n" else " ")
+                index += 1
+            continue
+
+        if state == "double-quote":
+            if char == '"' and next_char == '"':
+                code.append("_")
+                index += 2
+            elif char == '"':
+                code.append(" ")
+                state = "code"
+                index += 1
+            else:
+                code.append(char if (char.isalnum() or char == "_") else "_")
+                index += 1
+            continue
+
+        if char == "-" and next_char == "-":
+            code.extend((" ", " "))
+            state = "line-comment"
+            index += 2
+        elif char == "/" and next_char == "*":
+            code.extend((" ", " "))
+            block_depth = 1
+            state = "block-comment"
+            index += 2
+        elif char == "'":
+            escape_string = index > 0 and sql[index - 1] in "eE" and (
+                index == 1 or not (sql[index - 2].isalnum() or sql[index - 2] == "_")
+            )
+            code.append(" ")
+            state = "escape-string" if escape_string else "single-quote"
+            index += 1
+        elif char == '"':
+            if index >= 2 and sql[index - 2 : index].lower() == "u&":
+                raise SystemExit("Runtime SQL was rejected: Unicode-escape identifiers are not allowed.")
+            code.append(" ")
+            state = "double-quote"
+            index += 1
+        elif char == "\\":
+            raise SystemExit("Runtime SQL was rejected: psql meta-commands are not allowed.")
+        elif char == "$" and re.match(r"\$(?:[a-z_][a-z0-9_]*)?\$", sql[index:], re.IGNORECASE):
+            raise SystemExit("Runtime SQL was rejected: dollar-quoted bodies are not allowed.")
+        else:
+            code.append(char)
+            index += 1
+
+    if state == "block-comment":
+        raise SystemExit("Runtime SQL was rejected: unterminated block comment.")
+    if state in {"single-quote", "escape-string", "double-quote"}:
+        raise SystemExit("Runtime SQL was rejected: unterminated quoted value.")
+    return "".join(code)
+
+
+def validate_runtime_sql(sql: str, *, capability: str = RUNTIME_SQL_READ_CAPABILITY) -> None:
+    """Fail closed unless rendered SQL matches one reviewed runtime capability."""
+
+    if capability not in RUNTIME_SQL_CAPABILITIES:
+        raise SystemExit("Runtime SQL was rejected: unknown execution capability.")
+    code = _runtime_sql_code(sql)
+    if code.count(";") > 1:
+        raise SystemExit("Runtime SQL was rejected: exactly one statement is required.")
+    if ";" in code:
+        statement, remainder = code.split(";", 1)
+        if remainder.strip():
+            raise SystemExit("Runtime SQL was rejected: multiple statements are not allowed.")
+    else:
+        statement = code
+    first_token = re.match(r"\s*([a-z_][a-z0-9_]*)", statement, re.IGNORECASE)
+    if first_token is None or first_token.group(1).lower() not in {"select", "with"}:
+        raise SystemExit("Runtime SQL was rejected: only SELECT or WITH statements are allowed.")
+    if RUNTIME_SQL_FORBIDDEN_KEYWORD.search(statement):
+        raise SystemExit("Runtime SQL was rejected: mutating or session-control SQL is not allowed.")
+    if RUNTIME_SQL_LARGE_OBJECT_MUTATOR.search(statement):
+        raise SystemExit("Runtime SQL was rejected: large-object mutators are not allowed.")
+    if RUNTIME_SQL_FORBIDDEN_FUNCTION.search(statement):
+        raise SystemExit("Runtime SQL was rejected: side-effecting system functions are not allowed.")
+
+    application_functions = {
+        (match.group(1).lower(), match.group(2).lower())
+        for match in RUNTIME_SQL_APPLICATION_FUNCTION.finditer(statement)
+    }
+    if capability == RUNTIME_SQL_READ_CAPABILITY:
+        if application_functions:
+            raise SystemExit("Runtime SQL was rejected: application functions are not allowed for reads.")
+    else:
+        function_calls = [
+            (
+                match.group("schema").lower() if match.group("schema") else None,
+                match.group("name").lower(),
+            )
+            for match in RUNTIME_SQL_FUNCTION_CALL.finditer(statement)
+            if not (
+                match.group("schema") is None
+                and match.group("name").lower() in RUNTIME_SQL_NONFUNCTION_PAREN_KEYWORDS
+            )
+        ]
+        if function_calls.count(RUNTIME_SQL_STAGE_PROPOSAL_FUNCTION) != 1 or any(
+            function not in RUNTIME_SQL_STAGE_PROPOSAL_ALLOWED_FUNCTIONS for function in function_calls
+        ):
+            raise SystemExit(
+                "Runtime SQL was rejected: proposal execution exceeds the reviewed staging function surface."
+            )
+
+
+def run_psql(
+    args: argparse.Namespace,
+    sql: str,
+    db: str | None = None,
+    *,
+    capability: str = RUNTIME_SQL_READ_CAPABILITY,
+) -> str:
+    validate_runtime_sql(sql, capability=capability)
+    if capability == RUNTIME_SQL_STAGE_PROPOSAL_CAPABILITY and args.credential_mode != "runtime":
+        raise SystemExit("Proposal execution requires the scoped runtime credential mode.")
     credential = password(args)
     if args.credential_mode == "runtime":
         env = dict(RUNTIME_PSQL_ENV_BASE)
@@ -612,8 +836,11 @@ def run_psql(args: argparse.Namespace, sql: str, db: str | None = None) -> str:
             "PGPASSWORD": credential,
         }
     )
-    if args.credential_mode == "clone-readonly":
+    env["PGOPTIONS"] = (
-        env["PGOPTIONS"] = CLONE_READ_ONLY_PGOPTIONS
+        RUNTIME_READ_ONLY_PGOPTIONS
+        if capability == RUNTIME_SQL_READ_CAPABILITY
+        else RUNTIME_STANDARD_STRINGS_PGOPTIONS
+    )
     result = subprocess.run(
         [RUNTIME_PSQL_BIN, "-X", "-At", "-q", "-v", "ON_ERROR_STOP=1"],
         text=True,
@@ -627,9 +854,15 @@ def run_psql(args: argparse.Namespace, sql: str, db: str | None = None) -> str:
     return result.stdout
 
 
-def psql_json_lines(args: argparse.Namespace, sql: str, db: str | None = None) -> list[dict[str, Any]]:
+def psql_json_lines(
+    args: argparse.Namespace,
+    sql: str,
+    db: str | None = None,
+    *,
+    capability: str = RUNTIME_SQL_READ_CAPABILITY,
+) -> list[dict[str, Any]]:
     rows: list[dict[str, Any]] = []
-    for line in run_psql(args, sql, db=db).splitlines():
+    for line in run_psql(args, sql, db=db, capability=capability).splitlines():
         line = line.strip()
         if line:
             rows.append(json.loads(line))
@@ -1440,7 +1673,12 @@ select jsonb_build_object(
 )::text
 from proposer;
 """
-    rows = psql_json_lines(args, sql, db=args.canonical_db)
+    rows = psql_json_lines(
+        args,
+        sql,
+        db=args.canonical_db,
+        capability=RUNTIME_SQL_STAGE_PROPOSAL_CAPABILITY,
+    )
     if not rows:
         raise SystemExit("No core-change proposal inserted.")
     return rows[0]
@@ -1513,7 +1751,12 @@ select jsonb_build_object(
 )::text
 from staged;
 """
-    rows = psql_json_lines(args, sql, db=args.canonical_db)
+    rows = psql_json_lines(
+        args,
+        sql,
+        db=args.canonical_db,
+        capability=RUNTIME_SQL_STAGE_PROPOSAL_CAPABILITY,
+    )
     if not rows:
         raise SystemExit("No edge proposal inserted. Check that both claim ids exist and edge_type is valid.")
     return rows[0]

ops/derive_leoclean_runtime_query_contract.py

@@ -12,35 +12,58 @@ from __future__ import annotations
 
 import argparse
 import ast
+import hashlib
 import json
 import re
 from collections import defaultdict
 from pathlib import Path
 
 CANONICAL_RELATION = re.compile(
-    r"\b(?:from|join)\s+(public|kb_stage)\.([a-z_][a-z0-9_]*)",
+    r"\b(?:from|join)\s+(?:only\s+)?(public|kb_stage)\s*\.\s*([a-z_][a-z0-9_]*)",
     re.IGNORECASE,
 )
 ALIASED_CANONICAL_RELATION = re.compile(
-    r"\b(?:from|join)\s+(public|kb_stage)\.([a-z_][a-z0-9_]*)\s+as\s+([a-z_][a-z0-9_]*)",
+    r"\b(?:from|join)\s+(?:only\s+)?(public|kb_stage)\s*\.\s*([a-z_][a-z0-9_]*)"
+    r"\s+as\s+([a-z_][a-z0-9_]*)",
     re.IGNORECASE,
 )
 MUTATING_SQL_KEYWORD = re.compile(
     r"\b(?:insert|update|delete|truncate|merge)\b",
     re.IGNORECASE,
 )
+LARGE_OBJECT_MUTATOR_REFERENCE = re.compile(
+    r'(?<![a-z0-9_])(?:(?:"pg_catalog"|pg_catalog)\s*\.\s*)?'
+    r'(?:(?:"lo_creat"|lo_creat)|(?:"lo_create"|lo_create)|'
+    r'(?:"lo_export"|lo_export)|(?:"lo_from_bytea"|lo_from_bytea)|'
+    r'(?:"lo_import"|lo_import)|(?:"lo_open"|lo_open)|'
+    r'(?:"lo_put"|lo_put)|(?:"lo_truncate64"|lo_truncate64)|'
+    r'(?:"lo_truncate"|lo_truncate)|(?:"lo_unlink"|lo_unlink)|'
+    r'(?:"lowrite"|lowrite))\s*\(',
+    re.IGNORECASE,
+)
+PSQL_META_COMMAND = re.compile(r"\\")
 CANONICAL_NAMESPACE_REFERENCE = re.compile(r"\b(?:public|kb_stage)\s*\.", re.IGNORECASE)
+QUOTED_SQL_IDENTIFIER = re.compile(r'"(?:""|[^"])*"')
 DYNAMIC_SQL_VALUE = "__teleo_dynamic_sql_value__"
 UNREVIEWED_DYNAMIC_SQL = "__teleo_unreviewed_dynamic_sql__"
 REVIEWED_SQL_INTERPOLATION_CALLS = frozenset({"sql_array", "sql_integer", "sql_json_array", "sql_literal"})
+QUERY_RUNNERS = frozenset({"run_psql", "psql_json_lines"})
+REVIEWED_RUNTIME_SOURCE_SHA256 = "5f98218622803e0a1bbedc1e23e538d1f152ecb516e31a387a5eb39080aaf2cd"
+DYNAMIC_CALL_BUILTINS = frozenset({"eval", "exec", "globals", "locals"})
+READ_CAPABILITY = "read"
+STAGE_PROPOSAL_CAPABILITY = "stage-proposal"
 SQL_IDENTIFIER = re.compile(r"\b[a-z_][a-z0-9_]*\b", re.IGNORECASE)
+SQL_STRUCTURE_TOKEN = re.compile(r"[a-z_][a-z0-9_]*|[(),.;]", re.IGNORECASE)
+FROM_CLAUSE_END_KEYWORDS = frozenset(
+    {"except", "fetch", "for", "group", "having", "intersect", "limit", "offset", "order", "returning", "union", "where", "window"}
+)
 SQL_KEYWORDS = frozenset(
     """
     all and any array as asc between bigint bool boolean by case cast cross
     current_database current_user desc distinct else end exists false filter
     first following for from full group having ilike in inner interval is join
     json jsonb last lateral left like limit not null nulls offset on or order
-    outer over partition preceding range recursive right row rows select some
+    only outer over partition preceding range recursive right row rows select some
     symmetric table text then ties time timestamp timestamptz to true union
     unknown using uuid values varchar when where window with zone
     """.split()  # noqa: SIM905 - the compact vocabulary is easier to audit as SQL.
@@ -55,12 +78,22 @@ NONCANONICAL_QUERY_IDENTIFIERS = frozenset(
         "table_type",
     }
 )
+UNQUALIFIED_NONCANONICAL_RELATIONS = frozenset({"pg_extension", "pg_tables"})
 
 
 class QueryContractError(ValueError):
     """The runtime SQL cannot be safely reduced to an exact read contract."""
 
 
+def _assert_exact_runtime_source(source: str) -> None:
+    observed = hashlib.sha256(source.encode("utf-8")).hexdigest()
+    if observed != REVIEWED_RUNTIME_SOURCE_SHA256:
+        raise QueryContractError(
+            "runtime query surface differs from the exact reviewed runtime source: "
+            f"expected={REVIEWED_RUNTIME_SOURCE_SHA256}, observed={observed}"
+        )
+
+
 def _is_reviewed_sql_interpolation(node: ast.expr) -> bool:
     return (
         isinstance(node, ast.Call)
@@ -87,58 +120,362 @@ def _render_sql_expression(node: ast.expr) -> str | None:
     return None
 
 
-def extract_canonical_query_literals(source: str) -> tuple[tuple[int, str], ...]:
+def _statement_path(
-    """Return canonical SQL literals actually passed to the runtime query runners."""
+    node: ast.AST,
+    function: ast.FunctionDef | ast.AsyncFunctionDef,
+    parents: dict[ast.AST, ast.AST],
+) -> tuple[tuple[ast.AST, str, int, ast.stmt], ...]:
+    """Return the ordered statement blocks containing ``node``."""
+
+    path: list[tuple[ast.AST, str, int, ast.stmt]] = []
+    current = node
+    while current is not function:
+        parent = parents.get(current)
+        if parent is None:
+            break
+        if isinstance(current, ast.stmt):
+            for field, value in ast.iter_fields(parent):
+                if isinstance(value, list) and current in value:
+                    path.append((parent, field, value.index(current), current))
+                    break
+        current = parent
+    return tuple(reversed(path))
+
+
+def _ancestor_chain(node: ast.AST, parents: dict[ast.AST, ast.AST]) -> tuple[ast.AST, ...]:
+    ancestors: list[ast.AST] = []
+    current = node
+    while current in parents:
+        current = parents[current]
+        ancestors.append(current)
+    return tuple(ancestors)
+
+
+def _assignment_dominates_call(
+    assignment: ast.stmt,
+    call: ast.Call,
+    function: ast.FunctionDef | ast.AsyncFunctionDef,
+    parents: dict[ast.AST, ast.AST],
+) -> bool:
+    """Return whether one unconditional earlier assignment reaches ``call``."""
+
+    assignment_path = _statement_path(assignment, function, parents)
+    call_path = _statement_path(call, function, parents)
+    for index, assignment_entry in enumerate(assignment_path):
+        if index >= len(call_path):
+            return False
+        assignment_owner, assignment_field, assignment_index, assignment_statement = assignment_entry
+        call_owner, call_field, call_index, call_statement = call_path[index]
+        if assignment_owner is not call_owner or assignment_field != call_field:
+            return False
+        if assignment_index < call_index:
+            # An assignment hidden inside an earlier conditional does not
+            # dominate code after that conditional. The assignment itself must
+            # be the direct statement that precedes the call's branch.
+            return index == len(assignment_path) - 1
+        if assignment_index > call_index or assignment_statement is not call_statement:
+            return False
+    return False
+
+
+def _reject_query_runner_indirection(tree: ast.AST, parents: dict[ast.AST, ast.AST]) -> None:
+    """Require every runtime query runner reference to be a direct call."""
+
+    for node in ast.walk(tree):
+        if isinstance(node, ast.Call) and isinstance(node.func, (ast.Call, ast.Lambda, ast.Subscript)):
+            raise QueryContractError(f"line {node.lineno}: indirect call expressions are not auditable")
+        if isinstance(node, ast.Name) and node.id in DYNAMIC_CALL_BUILTINS and isinstance(node.ctx, ast.Load):
+            raise QueryContractError(f"line {node.lineno}: dynamic call indirection is not auditable")
+        if isinstance(node, ast.Attribute) and node.attr in DYNAMIC_CALL_BUILTINS:
+            raise QueryContractError(f"line {node.lineno}: dynamic call indirection is not auditable")
+        if isinstance(node, ast.Name) and node.id in QUERY_RUNNERS and isinstance(node.ctx, ast.Load):
+            parent = parents.get(node)
+            if not (isinstance(parent, ast.Call) and parent.func is node):
+                raise QueryContractError(
+                    f"line {node.lineno}: query runner {node.id!r} must be called directly"
+                )
+        elif isinstance(node, ast.Attribute) and node.attr in QUERY_RUNNERS:
+            raise QueryContractError(
+                f"line {node.lineno}: query runner {node.attr!r} must be called directly"
+            )
+        elif isinstance(node, ast.Constant) and node.value in QUERY_RUNNERS:
+            raise QueryContractError(
+                f"line {node.lineno}: query runner {node.value!r} cannot be resolved through indirection"
+            )
+
+
+def _is_reviewed_psql_json_lines_forwarder(
+    call: ast.Call,
+    function: ast.FunctionDef | ast.AsyncFunctionDef,
+) -> bool:
+    """Recognize the one query-transparent ``psql_json_lines`` forwarding call."""
+
+    if function.name != "psql_json_lines":
+        return False
+    if not (isinstance(call.func, ast.Name) and call.func.id == "run_psql"):
+        return False
+    if len(call.args) != 2 or not all(isinstance(argument, ast.Name) for argument in call.args):
+        return False
+    if [argument.id for argument in call.args] != ["args", "sql"]:
+        return False
+    keywords = {keyword.arg: keyword.value for keyword in call.keywords if keyword.arg is not None}
+    if len(keywords) != len(call.keywords) or set(keywords) != {"db", "capability"}:
+        return False
+    return all(
+        isinstance(keywords[name], ast.Name) and keywords[name].id == name
+        for name in ("db", "capability")
+    )
+
+
+def extract_runtime_query_literals(
+    source: str,
+    *,
+    enforce_source_pin: bool = True,
+) -> tuple[tuple[int, str, str], ...]:
+    """Return every statically rendered SQL statement and its capability."""
 
     tree = ast.parse(source)
-    queries: set[tuple[int, str]] = set()
+    parents = {child: parent for parent in ast.walk(tree) for child in ast.iter_child_nodes(parent)}
+    _reject_query_runner_indirection(tree, parents)
+    queries: set[tuple[int, str, str]] = set()
     for function in (node for node in ast.walk(tree) if isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef))):
-        assignments: dict[str, tuple[int, str | None]] = {}
+        function_calls = [node for node in ast.walk(function) if isinstance(node, ast.Call)]
+        reviewed_forwarders = [
+            call for call in function_calls if _is_reviewed_psql_json_lines_forwarder(call, function)
+        ]
+        if function.name == "psql_json_lines" and len(reviewed_forwarders) != 1:
+            raise QueryContractError(
+                f"line {function.lineno}: psql_json_lines must contain exactly one reviewed run_psql forwarder"
+            )
+        assignments: dict[str, list[tuple[ast.stmt, str | None]]] = defaultdict(list)
         for node in ast.walk(function):
             if isinstance(node, ast.Assign):
                 rendered = _render_sql_expression(node.value)
                 for target in node.targets:
                     if isinstance(target, ast.Name):
-                        assignments[target.id] = (node.lineno, rendered)
+                        assignments[target.id].append((node, rendered))
             elif isinstance(node, ast.AnnAssign) and isinstance(node.target, ast.Name):
-                assignments[node.target.id] = (node.lineno, _render_sql_expression(node.value))
+                assignments[node.target.id].append((node, _render_sql_expression(node.value)))
+            elif isinstance(node, ast.AugAssign) and isinstance(node.target, ast.Name):
+                assignments[node.target.id].append((node, None))
+            elif isinstance(node, ast.NamedExpr) and isinstance(node.target, ast.Name):
+                # Named expressions are deliberately not an accepted SQL
+                # construction surface; record them as an inexact write.
+                statement = next(
+                    (ancestor for ancestor in _ancestor_chain(node, parents) if isinstance(ancestor, ast.stmt)),
+                    function,
+                )
+                assignments[node.target.id].append((statement, None))
 
-        for call in (node for node in ast.walk(function) if isinstance(node, ast.Call)):
+        for call in sorted(
+            function_calls,
+            key=lambda node: (node.lineno, node.col_offset),
+        ):
             runner = call.func.id if isinstance(call.func, ast.Name) else None
-            if runner not in {"run_psql", "psql_json_lines"}:
+            if runner not in QUERY_RUNNERS:
                 continue
-            if function.name == "psql_json_lines" and runner == "run_psql":
+            if _is_reviewed_psql_json_lines_forwarder(call, function):
                 continue
             if len(call.args) < 2:
                 raise QueryContractError(f"line {call.lineno}: {runner} requires a statically inspectable SQL argument")
+            capability_keywords = [keyword for keyword in call.keywords if keyword.arg == "capability"]
+            if len(capability_keywords) > 1:
+                raise QueryContractError(f"line {call.lineno}: {runner} has ambiguous SQL capabilities")
+            capability = READ_CAPABILITY
+            if capability_keywords:
+                capability_expression = capability_keywords[0].value
+                if (
+                    isinstance(capability_expression, ast.Name)
+                    and capability_expression.id == "RUNTIME_SQL_STAGE_PROPOSAL_CAPABILITY"
+                ) or (
+                    isinstance(capability_expression, ast.Constant)
+                    and capability_expression.value == STAGE_PROPOSAL_CAPABILITY
+                ):
+                    capability = STAGE_PROPOSAL_CAPABILITY
+                else:
+                    raise QueryContractError(
+                        f"line {call.lineno}: {runner} SQL capability must remain statically reviewed"
+                    )
             expression = call.args[1]
             if isinstance(expression, ast.Name):
-                line, sql = assignments.get(expression.id, (call.lineno, None))
+                writes = assignments.get(expression.id, [])
+                if len(writes) != 1:
+                    raise QueryContractError(
+                        f"line {call.lineno}: {runner} SQL variable {expression.id!r} must have exactly one assignment"
+                    )
+                assignment, sql = writes[0]
+                if not _assignment_dominates_call(assignment, call, function, parents):
+                    raise QueryContractError(
+                        f"line {call.lineno}: {runner} SQL assignment must unambiguously precede the call"
+                    )
+                line = assignment.lineno
             else:
                 line, sql = call.lineno, _render_sql_expression(expression)
             if sql is None:
                 raise QueryContractError(f"line {call.lineno}: {runner} SQL must remain a local string literal")
-            scrubbed = _strip_sql_strings_and_comments(sql)
+            scrubbed = _normalize_sql_identifier_syntax(sql, line)
+            _reject_unqualified_relation_references(scrubbed, line)
+            _reject_comma_separated_base_relations(scrubbed, line)
             if UNREVIEWED_DYNAMIC_SQL in sql:
                 raise QueryContractError(
                     f"line {line}: dynamic SQL interpolation is not part of the reviewed runtime query surface"
                 )
+            if PSQL_META_COMMAND.search(scrubbed):
+                raise QueryContractError(
+                    f"line {line}: psql meta-commands are not part of the reviewed runtime query surface"
+                )
             if MUTATING_SQL_KEYWORD.search(scrubbed) and CANONICAL_NAMESPACE_REFERENCE.search(scrubbed):
                 raise QueryContractError(f"line {line}: runtime query performs direct canonical DML")
-            if CANONICAL_RELATION.search(sql):
+            if LARGE_OBJECT_MUTATOR_REFERENCE.search(scrubbed):
-                queries.add((line, sql))
+                raise QueryContractError(
+                    f"line {line}: runtime query invokes a provider-owned large-object mutator"
+                )
+            queries.add((line, sql, capability))
 
     if not queries:
-        raise QueryContractError("runtime source contains no canonical SQL query literals")
+        raise QueryContractError("runtime source contains no SQL query literals")
+    if enforce_source_pin:
+        _assert_exact_runtime_source(source)
     return tuple(sorted(queries, key=lambda item: item[0]))
 
 
+def extract_canonical_query_literals(
+    source: str,
+    *,
+    enforce_source_pin: bool = True,
+) -> tuple[tuple[int, str], ...]:
+    """Return canonical SQL literals actually passed to the runtime query runners."""
+
+    queries: list[tuple[int, str]] = []
+    for line, sql, _capability in extract_runtime_query_literals(source, enforce_source_pin=enforce_source_pin):
+        normalized = _normalize_sql_identifier_syntax(sql, line)
+        if CANONICAL_RELATION.search(normalized):
+            queries.append((line, normalized))
+    if not queries:
+        raise QueryContractError("runtime source contains no canonical SQL query literals")
+    return tuple(queries)
+
+
 def _strip_sql_strings_and_comments(sql: str) -> str:
     without_strings = re.sub(r"(?:[eE])?'(?:''|[^'])*'", " ", sql)
     without_block_comments = re.sub(r"/\*.*?\*/", " ", without_strings, flags=re.DOTALL)
     return re.sub(r"--[^\n]*", " ", without_block_comments)
 
 
+def _normalize_sql_identifier_syntax(sql: str, line: int) -> str:
+    """Return structural SQL with exact, lowercase quoted identifiers normalized."""
+
+    scrubbed = _strip_sql_strings_and_comments(sql)
+
+    def normalize_quoted_identifier(match: re.Match[str]) -> str:
+        identifier = match.group(0)[1:-1].replace('""', '"')
+        if (
+            not SQL_IDENTIFIER.fullmatch(identifier)
+            or identifier != identifier.lower()
+            or identifier in SQL_KEYWORDS
+        ):
+            raise QueryContractError(
+                f"line {line}: quoted SQL identifier {identifier!r} cannot be reduced exactly"
+            )
+        return identifier
+
+    normalized = QUOTED_SQL_IDENTIFIER.sub(normalize_quoted_identifier, scrubbed)
+    if '"' in normalized:
+        raise QueryContractError(f"line {line}: unterminated quoted SQL identifier")
+    return normalized
+
+
+def _reject_unqualified_relation_references(sql: str, line: int) -> None:
+    """Reject search-path-dependent base relations while allowing local CTEs/functions."""
+
+    cte_names = {
+        match.group(1).lower()
+        for match in re.finditer(
+            r"(?:\bwith\s+(?:recursive\s+)?|,)\s*([a-z_][a-z0-9_]*)"
+            r"\s*(?:\([^)]*\))?\s+as\s+(?:(?:not\s+)?materialized\s+)?\(",
+            sql,
+            re.IGNORECASE,
+        )
+    }
+    for match in re.finditer(
+        r"\b(?:from|join)\s+(?:only\s+)?([a-z_][a-z0-9_]*)",
+        sql,
+        re.IGNORECASE,
+    ):
+        relation = match.group(1).lower()
+        remainder = sql[match.end() :].lstrip()
+        if remainder.startswith((".", "(")):
+            continue
+        if relation in cte_names or relation in UNQUALIFIED_NONCANONICAL_RELATIONS:
+            continue
+        raise QueryContractError(
+            f"line {line}: unqualified relation {relation!r} is search-path dependent"
+        )
+
+
+def _reject_comma_separated_base_relations(sql: str, line: int) -> None:
+    """Require explicit JOIN syntax for base relations after a FROM-list comma."""
+
+    cte_names = {
+        match.group(1).lower()
+        for match in re.finditer(
+            r"(?:\bwith\s+(?:recursive\s+)?|,)\s*([a-z_][a-z0-9_]*)"
+            r"\s*(?:\([^)]*\))?\s+as\s+(?:(?:not\s+)?materialized\s+)?\(",
+            sql,
+            re.IGNORECASE,
+        )
+    }
+    tokens = [(match.group(0).lower(), match.start()) for match in SQL_STRUCTURE_TOKEN.finditer(sql)]
+    depth = 0
+    from_depths: set[int] = set()
+    for index, (token, _position) in enumerate(tokens):
+        if token == "(":
+            depth += 1
+            continue
+        if token == ")":
+            from_depths.discard(depth)
+            depth = max(0, depth - 1)
+            continue
+        if token == "from":
+            from_depths.add(depth)
+            continue
+        if token == ";" or token in FROM_CLAUSE_END_KEYWORDS:
+            from_depths.discard(depth)
+            continue
+        if token != "," or depth not in from_depths:
+            continue
+
+        item_index = index + 1
+        while item_index < len(tokens) and tokens[item_index][0] in {"lateral", "only"}:
+            item_index += 1
+        if item_index >= len(tokens):
+            raise QueryContractError(f"line {line}: incomplete comma-separated FROM item")
+        relation = tokens[item_index][0]
+        if relation == "(":
+            continue
+        if not SQL_IDENTIFIER.fullmatch(relation):
+            raise QueryContractError(f"line {line}: comma-separated FROM item cannot be reduced exactly")
+
+        next_token = tokens[item_index + 1][0] if item_index + 1 < len(tokens) else ""
+        if next_token == "(":
+            continue
+        if next_token == ".":
+            function_token = tokens[item_index + 3][0] if item_index + 3 < len(tokens) else ""
+            if function_token == "(":
+                continue
+            qualified = relation
+            if item_index + 2 < len(tokens):
+                qualified = f"{relation}.{tokens[item_index + 2][0]}"
+            raise QueryContractError(
+                f"line {line}: comma-separated base relation {qualified!r} must use explicit JOIN"
+            )
+        if relation in cte_names:
+            continue
+        raise QueryContractError(
+            f"line {line}: comma-separated base relation {relation!r} must use explicit JOIN"
+        )
+
+
 def _unqualified_identifier_candidates(sql: str) -> set[str]:
     """Find identifiers that are neither qualified columns nor query structure."""
 
@@ -248,7 +585,7 @@ def derive_runtime_read_columns(source: str) -> dict[tuple[str, str], frozenset[
     """Mechanically derive canonical relation columns from runtime SQL."""
 
     observed: dict[tuple[str, str], set[str]] = defaultdict(set)
-    for line, sql in extract_canonical_query_literals(source):
+    for line, sql in extract_canonical_query_literals(source, enforce_source_pin=False):
         aliases: dict[str, tuple[str, str]] = {}
         explicit_starts: set[int] = set()
         for match in ALIASED_CANONICAL_RELATION.finditer(sql):
@@ -273,7 +610,7 @@ def derive_runtime_read_columns(source: str) -> dict[tuple[str, str], frozenset[
             observed[relation].update(
                 column.lower()
                 for column in re.findall(
-                    rf"(?<![a-z0-9_]){re.escape(alias)}\.([a-z_][a-z0-9_]*)",
+                    rf"(?<![a-z0-9_]){re.escape(alias)}\s*\.\s*([a-z_][a-z0-9_]*)",
                     sql,
                     re.IGNORECASE,
                 )
@@ -284,6 +621,7 @@ def derive_runtime_read_columns(source: str) -> dict[tuple[str, str], frozenset[
             joined = ", ".join(sorted(unqualified))
             raise QueryContractError(f"line {line}: unqualified SQL identifiers prevent exact derivation: {joined}")
 
+    _assert_exact_runtime_source(source)
     return {relation: frozenset(columns) for relation, columns in observed.items()}
 
 

ops/gcp_leoclean_runtime_role.sql

@@ -261,53 +261,107 @@ select pg_catalog.format(
   from pg_catalog.pg_largeobject_metadata metadata
 \gexec
 
--- Default PostgreSQL grants EXECUTE on built-in large-object mutators to
+-- Cloud SQL owns pg_catalog and does not expose its true PostgreSQL
--- PUBLIC, which would let the runtime persist data despite all relation ACLs.
+-- superuser. Its customer administrator therefore cannot revoke the built-in
--- Snapshot and explicitly preserve every existing non-scoped role's effective
+-- PUBLIC EXECUTE grants on large-object mutators. Keep that provider-owned
--- access, then remove PUBLIC and both scoped roles from every mutator overload.
+-- residual immutable and explicit: reject inventory drift and any direct ACL
-create temporary table leoclean_preserved_lo_execute (
+-- granted to either scoped role, while separately proving that neither role
-  role_name name not null,
+-- owns or can reach an existing large object. The no-send staging runtime has
-  function_oid oid not null,
+-- no raw-SQL surface and its query-contract checker rejects every routine in
-  primary key (role_name, function_oid)
+-- this inventory. This is a staging-only compensating boundary, not a claim
+-- that arbitrary use of the raw database credential is write-free.
+create temporary table leoclean_expected_lo_routine_residual (
+  signature text primary key,
+  public_execute boolean not null
 ) on commit drop;
 
-insert into pg_temp.leoclean_preserved_lo_execute (role_name, function_oid)
+insert into pg_temp.leoclean_expected_lo_routine_residual (signature, public_execute) values
-select role_row.rolname, function_row.oid
+  ('lo_creat(integer)', true),
-  from pg_catalog.pg_roles role_row
+  ('lo_create(oid)', true),
- cross join pg_catalog.pg_proc function_row
+  ('lo_export(oid, text)', false),
-  join pg_catalog.pg_namespace namespace on namespace.oid = function_row.pronamespace
+  ('lo_from_bytea(oid, bytea)', true),
- where role_row.rolname not in ('leoclean_kb_runtime', 'leoclean_kb_stage_owner')
+  ('lo_import(text)', false),
-   and namespace.nspname = 'pg_catalog'
+  ('lo_import(text, oid)', false),
-   and function_row.proname in (
+  ('lo_open(oid, integer)', true),
-     'lo_creat', 'lo_create', 'lo_export', 'lo_from_bytea', 'lo_import',
+  ('lo_put(oid, bigint, bytea)', true),
-     'lo_open', 'lo_put', 'lo_truncate', 'lo_truncate64', 'lo_unlink', 'lowrite'
+  ('lo_truncate(integer, integer)', true),
-   )
+  ('lo_truncate64(integer, bigint)', true),
-   and pg_catalog.has_function_privilege(role_row.oid, function_row.oid, 'EXECUTE');
+  ('lo_unlink(oid)', true),
+  ('lowrite(integer, bytea)', true);
 
-select pg_catalog.format(
+do $large_object_residual$
-         'revoke execute on function pg_catalog.%I(%s) from public, leoclean_kb_runtime, leoclean_kb_stage_owner',
+declare
-         function_row.proname,
+  unexpected text;
-         pg_catalog.pg_get_function_identity_arguments(function_row.oid)
+begin
+  with actual as (
+    select function_row.oid,
+           pg_catalog.format(
+             '%I(%s)',
+             function_row.proname,
+             pg_catalog.pg_get_function_identity_arguments(function_row.oid)
+           ) as signature,
+           exists (
+             select 1
+               from pg_catalog.aclexplode(
+                 coalesce(function_row.proacl, pg_catalog.acldefault('f', function_row.proowner))
+               ) acl
+              where acl.grantee = 0
+                and acl.privilege_type = 'EXECUTE'
+           ) as public_execute,
+           pg_catalog.has_function_privilege('leoclean_kb_runtime', function_row.oid, 'EXECUTE') as runtime_execute,
+           pg_catalog.has_function_privilege('leoclean_kb_stage_owner', function_row.oid, 'EXECUTE') as owner_execute
+      from pg_catalog.pg_proc function_row
+      join pg_catalog.pg_namespace namespace on namespace.oid = function_row.pronamespace
+     where namespace.nspname = 'pg_catalog'
+       and function_row.proname in (
+         'lo_creat', 'lo_create', 'lo_export', 'lo_from_bytea', 'lo_import',
+         'lo_open', 'lo_put', 'lo_truncate', 'lo_truncate64', 'lo_unlink', 'lowrite'
        )
-  from pg_catalog.pg_proc function_row
+  )
-  join pg_catalog.pg_namespace namespace on namespace.oid = function_row.pronamespace
+  select pg_catalog.string_agg(
- where namespace.nspname = 'pg_catalog'
+           coalesce(expected.signature, actual.signature),
-   and function_row.proname in (
+           ', '
-     'lo_creat', 'lo_create', 'lo_export', 'lo_from_bytea', 'lo_import',
+           order by coalesce(expected.signature, actual.signature)
-     'lo_open', 'lo_put', 'lo_truncate', 'lo_truncate64', 'lo_unlink', 'lowrite'
+         )
-   )
+    into unexpected
-\gexec
+    from pg_temp.leoclean_expected_lo_routine_residual expected
+    full join actual using (signature)
+   where expected.signature is null
+      or actual.signature is null
+      or actual.public_execute is distinct from expected.public_execute
+      or actual.runtime_execute is distinct from expected.public_execute
+      or actual.owner_execute is distinct from expected.public_execute;
+  if unexpected is not null then
+    raise exception 'provider-owned large-object routine residual drifted: %', unexpected;
+  end if;
 
-select pg_catalog.format(
+  select pg_catalog.string_agg(
-         'grant execute on function pg_catalog.%I(%s) to %I',
+           pg_catalog.format(
-         function_row.proname,
+             '%I(%s):%s',
-         pg_catalog.pg_get_function_identity_arguments(function_row.oid),
+             function_row.proname,
-         preserved.role_name
+             pg_catalog.pg_get_function_identity_arguments(function_row.oid),
-       )
+             role_row.rolname
-  from pg_temp.leoclean_preserved_lo_execute preserved
+           ),
-  join pg_catalog.pg_proc function_row on function_row.oid = preserved.function_oid
+           ', '
- order by function_row.proname, preserved.role_name
+           order by function_row.proname, role_row.rolname
-\gexec
+         )
+    into unexpected
+    from pg_catalog.pg_proc function_row
+    join pg_catalog.pg_namespace namespace on namespace.oid = function_row.pronamespace
+   cross join lateral pg_catalog.aclexplode(function_row.proacl) acl
+    join pg_catalog.pg_roles role_row on role_row.oid = acl.grantee
+   where namespace.nspname = 'pg_catalog'
+     and function_row.proname in (
+       'lo_creat', 'lo_create', 'lo_export', 'lo_from_bytea', 'lo_import',
+       'lo_open', 'lo_put', 'lo_truncate', 'lo_truncate64', 'lo_unlink', 'lowrite'
+     )
+     and role_row.rolname in ('leoclean_kb_runtime', 'leoclean_kb_stage_owner')
+     and acl.privilege_type = 'EXECUTE';
+  if unexpected is not null then
+    raise exception 'scoped Leo roles have direct large-object routine ACLs: %', unexpected;
+  end if;
+end
+$large_object_residual$;
 
 select pg_catalog.format(
          'revoke all privileges on parameter %I from leoclean_kb_runtime, leoclean_kb_stage_owner',
@@ -1061,46 +1115,76 @@ begin
     raise exception 'scoped Leo roles retain large-object ownership or ACL reachability: %', unexpected;
   end if;
 
+  with actual as (
+    select function_row.oid,
+           pg_catalog.format(
+             '%I(%s)',
+             function_row.proname,
+             pg_catalog.pg_get_function_identity_arguments(function_row.oid)
+           ) as signature,
+           exists (
+             select 1
+               from pg_catalog.aclexplode(
+                 coalesce(function_row.proacl, pg_catalog.acldefault('f', function_row.proowner))
+               ) acl
+              where acl.grantee = 0
+                and acl.privilege_type = 'EXECUTE'
+           ) as public_execute,
+           pg_catalog.has_function_privilege('leoclean_kb_runtime', function_row.oid, 'EXECUTE') as runtime_execute,
+           pg_catalog.has_function_privilege('leoclean_kb_stage_owner', function_row.oid, 'EXECUTE') as owner_execute
+      from pg_catalog.pg_proc function_row
+      join pg_catalog.pg_namespace namespace on namespace.oid = function_row.pronamespace
+     where namespace.nspname = 'pg_catalog'
+       and function_row.proname in (
+         'lo_creat', 'lo_create', 'lo_export', 'lo_from_bytea', 'lo_import',
+         'lo_open', 'lo_put', 'lo_truncate', 'lo_truncate64', 'lo_unlink', 'lowrite'
+       )
+  )
   select pg_catalog.string_agg(
-           pg_catalog.format('%I(%s)', function_row.proname, pg_catalog.pg_get_function_identity_arguments(function_row.oid)),
+           coalesce(expected.signature, actual.signature),
            ', '
+           order by coalesce(expected.signature, actual.signature)
          )
     into unexpected
-    from pg_catalog.pg_proc function_row
+    from pg_temp.leoclean_expected_lo_routine_residual expected
-    join pg_catalog.pg_namespace namespace on namespace.oid = function_row.pronamespace
+    full join actual using (signature)
-   where namespace.nspname = 'pg_catalog'
+   where expected.signature is null
-     and function_row.proname in (
+      or actual.signature is null
-       'lo_creat', 'lo_create', 'lo_export', 'lo_from_bytea', 'lo_import',
+      or actual.public_execute is distinct from expected.public_execute
-       'lo_open', 'lo_put', 'lo_truncate', 'lo_truncate64', 'lo_unlink', 'lowrite'
+      or actual.runtime_execute is distinct from expected.public_execute
-     )
+      or actual.owner_execute is distinct from expected.public_execute;
-     and (
-       has_function_privilege('leoclean_kb_runtime', function_row.oid, 'EXECUTE')
-       or has_function_privilege('leoclean_kb_stage_owner', function_row.oid, 'EXECUTE')
-     );
   if unexpected is not null then
-    raise exception 'PUBLIC or inherited large-object mutators remain executable: %', unexpected;
+    raise exception 'provider-owned large-object routine residual drifted: %', unexpected;
   end if;
 
   select pg_catalog.string_agg(
-           pg_catalog.format('%I(%s)', function_row.proname, pg_catalog.pg_get_function_identity_arguments(function_row.oid)),
+           pg_catalog.format(
+             '%I(%s):%s',
+             function_row.proname,
+             pg_catalog.pg_get_function_identity_arguments(function_row.oid),
+             role_row.rolname
+           ),
            ', '
-           order by function_row.proname, pg_catalog.pg_get_function_identity_arguments(function_row.oid)
+           order by function_row.proname, role_row.rolname
          )
     into unexpected
     from pg_catalog.pg_proc function_row
     join pg_catalog.pg_namespace namespace on namespace.oid = function_row.pronamespace
-   cross join lateral pg_catalog.aclexplode(
+   cross join lateral pg_catalog.aclexplode(function_row.proacl) acl
-     coalesce(function_row.proacl, pg_catalog.acldefault('f', function_row.proowner))
+    join pg_catalog.pg_roles role_row on role_row.oid = acl.grantee
-   ) acl
    where namespace.nspname = 'pg_catalog'
      and function_row.proname in (
        'lo_creat', 'lo_create', 'lo_export', 'lo_from_bytea', 'lo_import',
        'lo_open', 'lo_put', 'lo_truncate', 'lo_truncate64', 'lo_unlink', 'lowrite'
      )
-     and acl.grantee = 0
+     and role_row.rolname in ('leoclean_kb_runtime', 'leoclean_kb_stage_owner')
      and acl.privilege_type = 'EXECUTE';
   if unexpected is not null then
-    raise exception 'PUBLIC retains large-object mutator EXECUTE: %', unexpected;
+    raise exception 'scoped Leo roles have direct large-object routine ACLs: %', unexpected;
+  end if;
+
+  if pg_catalog.current_setting('lo_compat_privileges') <> 'off' then
+    raise exception 'lo_compat_privileges must remain off for scoped large-object ACL enforcement';
   end if;
 
   select pg_catalog.string_agg(parameter_acl.parname, ', ' order by parameter_acl.parname)
@@ -1434,8 +1518,9 @@ end
 $verification$;
 
 -- LOGIN is the last state change in the same transaction as every database ACL,
--- canonical privilege, large-object, and topology assertion. Any failure rolls
+-- canonical privilege, large-object ownership/residual, and topology
--- back the complete privilege migration and leaves the role NOLOGIN.
+-- assertion. Any failure rolls back the complete privilege migration and
+-- leaves the role NOLOGIN.
 alter role leoclean_kb_runtime
   with login nocreatedb nocreaterole noinherit connection limit 8;
 

ops/verify_gcp_leoclean_runtime_permissions.py

@@ -233,13 +233,27 @@ READ_TABLE_ALLOWLIST: tuple[tuple[str, str], ...] = tuple(
     (schema, relation) for schema, relation, _columns in READ_COLUMN_ALLOWLIST
 )
 
+LARGE_OBJECT_MUTATION_ROUTINE_RESIDUAL: tuple[tuple[str, bool], ...] = (
+    ("lo_creat(integer)", True),
+    ("lo_create(oid)", True),
+    ("lo_export(oid, text)", False),
+    ("lo_from_bytea(oid, bytea)", True),
+    ("lo_import(text)", False),
+    ("lo_import(text, oid)", False),
+    ("lo_open(oid, integer)", True),
+    ("lo_put(oid, bigint, bytea)", True),
+    ("lo_truncate(integer, integer)", True),
+    ("lo_truncate64(integer, bigint)", True),
+    ("lo_unlink(oid)", True),
+    ("lowrite(integer, bytea)", True),
+)
+
 CATALOG_ZERO_COUNT_FIELDS: tuple[str, ...] = (
     "column_dml_grants",
     "database_create_grants",
+    "direct_large_object_mutation_routine_acl_entries",
     "missing_allowed_table_selects",
     "large_object_acl_privileges",
-    "large_object_mutation_routine_execute",
-    "public_large_object_mutation_routine_execute",
     "other_scoped_backends",
     "owned_database_objects",
     "owned_large_objects",
@@ -674,6 +688,7 @@ select pg_catalog.jsonb_build_object(
     (select version from pg_catalog.pg_stat_ssl where pid = pg_catalog.pg_backend_pid()),
     ''
   ),
+  'lo_compat_privileges', pg_catalog.current_setting('lo_compat_privileges'),
   'system_identifier', (select system_identifier::text from pg_catalog.pg_control_system())
 )::text;
 """.strip()
@@ -975,6 +990,7 @@ select pg_catalog.jsonb_build_object(
     and pg_catalog.current_setting('lock_timeout') = '2s'
     and pg_catalog.current_setting('session_preload_libraries') = ''
     and pg_catalog.current_setting('local_preload_libraries') = ''
+    and pg_catalog.current_setting('lo_compat_privileges') = 'off'
     and pg_catalog.current_setting('role') = 'none'
   ),
   'unexpected_connectable_databases', (
@@ -1278,33 +1294,53 @@ select pg_catalog.jsonb_build_object(
           and acl.privilege_type in ('SELECT', 'UPDATE')
      )
   ),
-  'large_object_mutation_routine_execute', (
+  'direct_large_object_mutation_routine_acl_entries', (
     select pg_catalog.count(*)::int
       from pg_catalog.pg_proc function_row
       join pg_catalog.pg_namespace namespace on namespace.oid = function_row.pronamespace
-     cross join (values ({_sql_literal(RUNTIME_DATABASE_ROLE)}), ({_sql_literal(STAGE_OWNER_DATABASE_ROLE)})) scoped_role(name)
+     cross join lateral pg_catalog.aclexplode(function_row.proacl) acl
+      join pg_catalog.pg_roles role_row on role_row.oid = acl.grantee
      where namespace.nspname = 'pg_catalog'
        and function_row.proname in (
          'lo_creat', 'lo_create', 'lo_export', 'lo_from_bytea', 'lo_import',
          'lo_open', 'lo_put', 'lo_truncate', 'lo_truncate64', 'lo_unlink', 'lowrite'
        )
-       and pg_catalog.has_function_privilege(scoped_role.name, function_row.oid, 'EXECUTE')
+       and role_row.rolname in ({_sql_literal(RUNTIME_DATABASE_ROLE)}, {_sql_literal(STAGE_OWNER_DATABASE_ROLE)})
-  ),
-  'public_large_object_mutation_routine_execute', (
-    select pg_catalog.count(*)::int
-      from pg_catalog.pg_proc function_row
-      join pg_catalog.pg_namespace namespace on namespace.oid = function_row.pronamespace
-      cross join lateral pg_catalog.aclexplode(
-        coalesce(function_row.proacl, pg_catalog.acldefault('f', function_row.proowner))
-      ) acl
-     where namespace.nspname = 'pg_catalog'
-       and function_row.proname in (
-         'lo_creat', 'lo_create', 'lo_export', 'lo_from_bytea', 'lo_import',
-         'lo_open', 'lo_put', 'lo_truncate', 'lo_truncate64', 'lo_unlink', 'lowrite'
-       )
-       and acl.grantee = 0
        and acl.privilege_type = 'EXECUTE'
   ),
+  'large_object_mutation_routine_residual', coalesce((
+    select pg_catalog.jsonb_agg(
+             pg_catalog.jsonb_build_object(
+               'signature', pg_catalog.format(
+                 '%I(%s)',
+                 function_row.proname,
+                 pg_catalog.pg_get_function_identity_arguments(function_row.oid)
+               ),
+               'public_execute', exists (
+                 select 1
+                   from pg_catalog.aclexplode(
+                     coalesce(function_row.proacl, pg_catalog.acldefault('f', function_row.proowner))
+                   ) acl
+                  where acl.grantee = 0
+                    and acl.privilege_type = 'EXECUTE'
+               ),
+               'runtime_execute', pg_catalog.has_function_privilege(
+                 {_sql_literal(RUNTIME_DATABASE_ROLE)}, function_row.oid, 'EXECUTE'
+               ),
+               'stage_owner_execute', pg_catalog.has_function_privilege(
+                 {_sql_literal(STAGE_OWNER_DATABASE_ROLE)}, function_row.oid, 'EXECUTE'
+               )
+             )
+             order by function_row.proname, pg_catalog.pg_get_function_identity_arguments(function_row.oid)
+           )
+      from pg_catalog.pg_proc function_row
+      join pg_catalog.pg_namespace namespace on namespace.oid = function_row.pronamespace
+     where namespace.nspname = 'pg_catalog'
+       and function_row.proname in (
+         'lo_creat', 'lo_create', 'lo_export', 'lo_from_bytea', 'lo_import',
+         'lo_open', 'lo_put', 'lo_truncate', 'lo_truncate64', 'lo_unlink', 'lowrite'
+       )
+  ), '[]'::pg_catalog.jsonb),
   'parameter_privileges', (
     select pg_catalog.count(*)::int
       from pg_catalog.pg_parameter_acl parameter_acl
@@ -1639,6 +1675,49 @@ def _transactional(sql: str) -> str:
     return f"begin;\n{sql.rstrip(';')};\nrollback;"
 
 
+def _large_object_owned_count_sql() -> str:
+    return """
+select pg_catalog.count(*)::text
+  from pg_catalog.pg_largeobject_metadata metadata
+ where metadata.lomowner = (
+   select role_row.oid from pg_catalog.pg_roles role_row where role_row.rolname = current_user
+ );
+""".strip()
+
+
+def _large_object_residual_probe_sql() -> str:
+    return """
+begin;
+with created as (
+  select pg_catalog.lo_from_bytea(0, ''::bytea) as oid
+)
+select pg_catalog.jsonb_build_object(
+  'created', created.oid <> 0,
+  'owned_inside', exists (
+    select 1
+      from pg_catalog.pg_largeobject_metadata metadata
+     where metadata.oid = created.oid
+       and metadata.lomowner = (
+         select role_row.oid from pg_catalog.pg_roles role_row where role_row.rolname = current_user
+       )
+  )
+)::text
+from created;
+rollback;
+""".strip()
+
+
+def _assert_large_object_residual_probe(value: dict[str, Any]) -> dict[str, bool | str]:
+    if value != {"created": True, "owned_inside": True}:
+        raise VerificationError("large_object_residual_probe_mismatch", "rolled_back_large_object_residual")
+    return {
+        "capability_present": True,
+        "owned_inside_transaction": True,
+        "persistence_after_rollback": False,
+        "transaction": "rolled_back",
+    }
+
+
 def _negative_checks(run_id: str, source_ref: str) -> tuple[NegativeCheck, ...]:
     nil_uuid = "00000000-0000-0000-0000-000000000000"
     return (
@@ -1678,24 +1757,6 @@ def _negative_checks(run_id: str, source_ref: str) -> tuple[NegativeCheck, ...]:
             _transactional("delete from public.claims where false"),
             "42501",
         ),
-        NegativeCheck(
-            "large_object_creat",
-            CANONICAL_DATABASE,
-            _transactional("select pg_catalog.lo_creat(0)"),
-            "42501",
-        ),
-        NegativeCheck(
-            "large_object_create",
-            CANONICAL_DATABASE,
-            _transactional("select pg_catalog.lo_create(0)"),
-            "42501",
-        ),
-        NegativeCheck(
-            "large_object_from_bytea",
-            CANONICAL_DATABASE,
-            _transactional("select pg_catalog.lo_from_bytea(0, ''::bytea)"),
-            "42501",
-        ),
         NegativeCheck(
             "forged_proposer_overload",
             CANONICAL_DATABASE,
@@ -1735,6 +1796,12 @@ def _negative_checks(run_id: str, source_ref: str) -> tuple[NegativeCheck, ...]:
             ),
             "42501",
         ),
+        NegativeCheck(
+            "set_lo_compat_privileges",
+            CANONICAL_DATABASE,
+            _transactional("set lo_compat_privileges = on"),
+            "42501",
+        ),
         NegativeCheck(
             "set_role_stage_owner",
             CANONICAL_DATABASE,
@@ -1839,9 +1906,12 @@ def _expect_sqlstate(
 
 
 def _assert_identity(identity: dict[str, Any]) -> dict[str, Any]:
+    if identity.get("lo_compat_privileges") != "off":
+        raise VerificationError("lo_compat_privileges_unsafe", "database_identity")
     expected = {
         "database": CANONICAL_DATABASE,
         "current_user": RUNTIME_DATABASE_ROLE,
+        "lo_compat_privileges": "off",
         "session_user": RUNTIME_DATABASE_ROLE,
         "server_addr": PRIVATE_CLOUDSQL_HOST,
         "server_port": PRIVATE_CLOUDSQL_PORT,
@@ -1973,8 +2043,8 @@ def _assert_stage_function_definition(posture: dict[str, Any]) -> dict[str, Any]
     }
 
 
-def _assert_catalog_privilege_posture(posture: dict[str, Any]) -> dict[str, int | bool]:
+def _assert_catalog_privilege_posture(posture: dict[str, Any]) -> dict[str, Any]:
-    sanitized: dict[str, int | bool] = {}
+    sanitized: dict[str, Any] = {}
     for field in CATALOG_ZERO_COUNT_FIELDS:
         value = posture.get(field)
         if isinstance(value, bool) or not isinstance(value, int) or value != 0:
@@ -1984,6 +2054,18 @@ def _assert_catalog_privilege_posture(posture: dict[str, Any]) -> dict[str, int
         if posture.get(field) is not True:
             raise VerificationError("catalog_privilege_posture_mismatch", "catalog_privilege_posture")
         sanitized[field] = True
+    expected_residual = [
+        {
+            "public_execute": public_execute,
+            "runtime_execute": public_execute,
+            "signature": signature,
+            "stage_owner_execute": public_execute,
+        }
+        for signature, public_execute in LARGE_OBJECT_MUTATION_ROUTINE_RESIDUAL
+    ]
+    if posture.get("large_object_mutation_routine_residual") != expected_residual:
+        raise VerificationError("large_object_residual_mismatch", "catalog_privilege_posture")
+    sanitized["large_object_mutation_routine_residual"] = expected_residual
     return sanitized
 
 
@@ -2162,6 +2244,38 @@ def verify_runtime_permissions(
             ),
             "canary_rows_after",
         )
+        large_objects_before = _parse_zero_count(
+            _run_psql_success(
+                check="large_objects_before",
+                database=CANONICAL_DATABASE,
+                sql=_large_object_owned_count_sql(),
+                env=psql_env,
+                runner=runner,
+            ),
+            "large_objects_before",
+        )
+        large_object_residual = _assert_large_object_residual_probe(
+            _parse_json_object(
+                _run_psql_success(
+                    check="rolled_back_large_object_residual",
+                    database=CANONICAL_DATABASE,
+                    sql=_large_object_residual_probe_sql(),
+                    env=psql_env,
+                    runner=runner,
+                ),
+                "rolled_back_large_object_residual",
+            )
+        )
+        large_objects_after = _parse_zero_count(
+            _run_psql_success(
+                check="large_objects_after",
+                database=CANONICAL_DATABASE,
+                sql=_large_object_owned_count_sql(),
+                env=psql_env,
+                runner=runner,
+            ),
+            "large_objects_after",
+        )
 
         negative_permissions = [
             _expect_sqlstate(check, env=psql_env, runner=runner) for check in _negative_checks(run_id, source_ref)
@@ -2176,6 +2290,11 @@ def verify_runtime_permissions(
             "canary_rows_after": rows_after,
             "canary_rows_before": rows_before,
             "function_privileges": function_privileges,
+            "large_object_residual": {
+                **large_object_residual,
+                "owned_objects_after": large_objects_after,
+                "owned_objects_before": large_objects_before,
+            },
             "negative_permissions": negative_permissions,
             "role_posture": role_posture,
             "stage_function_definition": stage_function_definition,
@@ -2188,7 +2307,9 @@ def verify_runtime_permissions(
         "current_tier": REQUIRED_TIER,
         "database_identity": identity,
         "execution": {
+            "arbitrary_database_writes_denied": False,
             "canonical_writes_committed": False,
+            "direct_relation_writes_denied": True,
             "service_independent": True,
             "unix_user": RUNTIME_UNIX_USER,
         },
@@ -2196,7 +2317,7 @@ def verify_runtime_permissions(
         "mode": "live_private_gcp_staging",
         "required_tier": REQUIRED_TIER,
         "run_id": run_id,
-        "schema_version": 1,
+        "schema_version": 2,
         "secret_access": {
             "administrator": {
                 **administrator_secret,
@@ -2210,10 +2331,17 @@ def verify_runtime_permissions(
         },
         "status": "pass",
         "safety": {
+            "arbitrary_database_writes_denied": False,
+            "canonical_relation_writes_denied": True,
             "canonical_write_committed": False,
+            "direct_stage_table_writes_denied": True,
+            "production_eligible": False,
             "production_promotion_attempted": False,
+            "proposal_staging": "function_only",
             "proposal_write_committed": False,
+            "public_large_object_mutation_residual": "present",
             "secret_value_retained": False,
+            "staging_only": True,
             "telegram_send_attempted": False,
         },
         "target": {
@@ -2235,7 +2363,7 @@ def failure_receipt(run_id: str, error: VerificationError) -> dict[str, Any]:
         "generated_at_utc": datetime.now(timezone.utc).isoformat(timespec="seconds").replace("+00:00", "Z"),
         "required_tier": REQUIRED_TIER,
         "run_id": run_id,
-        "schema_version": 1,
+        "schema_version": 2,
         "status": "fail",
     }
 

scripts/run_gcp_generated_db_direct_claim_suite.py

@@ -53,7 +53,7 @@ DEFAULT_SECRET = "gcp-teleo-pgvector-standby-postgres-password"
 DEFAULT_SSL_ROOT_CERT = Path("/usr/local/libexec/livingip/leoclean-kb/cloudsql-server-ca.pem")
 DEFAULT_SYSTEM_IDENTIFIER = "7659718422914359312"
 DEFAULT_MANIFEST_SQL = HERE.parent / "ops" / "postgres_parity_manifest.sql"
-REVIEWED_CLOUDSQL_TOOL_SHA256 = "cc435231adab4325489936ae18a156224ab0d4a9cb158dad46b4491511b64ea3"
+REVIEWED_CLOUDSQL_TOOL_SHA256 = "5f98218622803e0a1bbedc1e23e538d1f152ecb516e31a387a5eb39080aaf2cd"
 COUNT_READBACK_RE = re.compile(
     r"DB readback:\s*claims:\s*`?(?P<claims>\d+)`?;\s*"
     r"sources:\s*`?(?P<sources>\d+)`?;\s*"

tests/test_gcp_leoclean_runtime_permissions.py

@@ -89,6 +89,7 @@ class FakeRunner:
                 "current_user": verifier.RUNTIME_DATABASE_ROLE,
                 "database": verifier.CANONICAL_DATABASE,
                 "leak": SECRET_VALUE,
+                "lo_compat_privileges": "off",
                 "server_addr": verifier.PRIVATE_CLOUDSQL_HOST,
                 "server_port": verifier.PRIVATE_CLOUDSQL_PORT,
                 "session_user": verifier.RUNTIME_DATABASE_ROLE,
@@ -174,6 +175,15 @@ class FakeRunner:
             catalog_posture.update(
                 {
                     "leak": SECRET_VALUE,
+                    "large_object_mutation_routine_residual": [
+                        {
+                            "public_execute": public_execute,
+                            "runtime_execute": public_execute,
+                            "signature": signature,
+                            "stage_owner_execute": public_execute,
+                        }
+                        for signature, public_execute in verifier.LARGE_OBJECT_MUTATION_ROUTINE_RESIDUAL
+                    ],
                     **{field: True for field in verifier.CATALOG_TRUE_FIELDS},
                 }
             )
@@ -181,6 +191,14 @@ class FakeRunner:
             return verifier.CommandResult(0, json.dumps(catalog_posture), f"ignored {SECRET_VALUE}")
         if sql == verifier._canary_count_sql(SOURCE_REF):
             return verifier.CommandResult(0, "0\n", f"ignored {SECRET_VALUE}")
+        if sql == verifier._large_object_owned_count_sql():
+            return verifier.CommandResult(0, "0\n", f"ignored {SECRET_VALUE}")
+        if sql == verifier._large_object_residual_probe_sql():
+            return verifier.CommandResult(
+                0,
+                json.dumps({"created": True, "owned_inside": True}),
+                f"ignored {SECRET_VALUE}",
+            )
         if sql == verifier._stage_sql(RUN_ID, SOURCE_REF):
             return verifier.CommandResult(
                 0,
@@ -284,6 +302,7 @@ def test_live_receipt_contract_is_sanitized_and_uses_exact_child_environments()
     assert receipt["database_identity"] == {
         "current_user": verifier.RUNTIME_DATABASE_ROLE,
         "database": verifier.CANONICAL_DATABASE,
+        "lo_compat_privileges": "off",
         "server_addr": verifier.PRIVATE_CLOUDSQL_HOST,
         "server_port": verifier.PRIVATE_CLOUDSQL_PORT,
         "session_user": verifier.RUNTIME_DATABASE_ROLE,
@@ -301,6 +320,14 @@ def test_live_receipt_contract_is_sanitized_and_uses_exact_child_environments()
     }
     assert receipt["checks"]["canary_rows_before"] == 0
     assert receipt["checks"]["canary_rows_after"] == 0
+    assert receipt["checks"]["large_object_residual"] == {
+        "capability_present": True,
+        "owned_inside_transaction": True,
+        "owned_objects_after": 0,
+        "owned_objects_before": 0,
+        "persistence_after_rollback": False,
+        "transaction": "rolled_back",
+    }
     assert receipt["checks"]["role_posture"] == {
         "bypasses_rls": False,
         "can_create_db": False,
@@ -336,6 +363,15 @@ def test_live_receipt_contract_is_sanitized_and_uses_exact_child_environments()
     assert receipt["checks"]["catalog_privileges"] == {
         **{field: 0 for field in verifier.CATALOG_ZERO_COUNT_FIELDS},
         **{field: True for field in verifier.CATALOG_TRUE_FIELDS},
+        "large_object_mutation_routine_residual": [
+            {
+                "public_execute": public_execute,
+                "runtime_execute": public_execute,
+                "signature": signature,
+                "stage_owner_execute": public_execute,
+            }
+            for signature, public_execute in verifier.LARGE_OBJECT_MUTATION_ROUTINE_RESIDUAL
+        ],
     }
     assert receipt["checks"]["stage_function_definition"] == {
         "acl_exact": True,
@@ -349,6 +385,28 @@ def test_live_receipt_contract_is_sanitized_and_uses_exact_child_environments()
         "source_sha256": verifier.EXPECTED_STAGE_FUNCTION_SOURCE_SHA256,
     }
     assert len(receipt["checks"]["negative_permissions"]) == len(verifier._negative_checks(RUN_ID, SOURCE_REF))
+    assert receipt["schema_version"] == 2
+    assert receipt["execution"] == {
+        "arbitrary_database_writes_denied": False,
+        "canonical_writes_committed": False,
+        "direct_relation_writes_denied": True,
+        "service_independent": True,
+        "unix_user": verifier.RUNTIME_UNIX_USER,
+    }
+    assert receipt["safety"] == {
+        "arbitrary_database_writes_denied": False,
+        "canonical_relation_writes_denied": True,
+        "canonical_write_committed": False,
+        "direct_stage_table_writes_denied": True,
+        "production_eligible": False,
+        "production_promotion_attempted": False,
+        "proposal_staging": "function_only",
+        "proposal_write_committed": False,
+        "public_large_object_mutation_residual": "present",
+        "secret_value_retained": False,
+        "staging_only": True,
+        "telegram_send_attempted": False,
+    }
     assert receipt["secret_access"]["administrator"]["classification"] == "iam_permission_denied"
     assert receipt["secret_access"]["administrator"]["stdout_discarded"] is True
     assert SECRET_VALUE not in serialized
@@ -614,6 +672,143 @@ def test_runtime_query_contract_rejects_a_new_unallowlisted_column() -> None:
         query_contract.verify_runtime_read_columns(mutated, expected)
 
 
+def test_runtime_query_contract_pins_the_exact_reviewed_runtime_source() -> None:
+    runtime_path = Path(__file__).resolve().parents[1] / "hermes-agent" / "leoclean-bin" / "cloudsql_memory_tool.py"
+    runtime_source = runtime_path.read_text(encoding="utf-8")
+
+    assert hashlib.sha256(runtime_source.encode("utf-8")).hexdigest() == query_contract.REVIEWED_RUNTIME_SOURCE_SHA256
+
+
+def test_runtime_query_contract_rejects_recursive_proposal_execution_cardinality_mutation() -> None:
+    runtime_path = Path(__file__).resolve().parents[1] / "hermes-agent" / "leoclean-bin" / "cloudsql_memory_tool.py"
+    runtime_source = runtime_path.read_text(encoding="utf-8")
+    recursive_sql = '''    sql = """
+with recursive spam as (
+  select 1 as n, null::jsonb as proposal
+  union all
+  select n + 1, kb_stage.stage_leoclean_proposal(
+    'revise_claim', 'telegram', 'receipt:1', 'reason', '{}'::jsonb
+  )
+  from spam
+  where n < 10
+)
+select proposal from spam;
+"""
+'''
+    function_start = runtime_source.index("def propose_core_change")
+    sql_start = runtime_source.index('    sql = f"""\n', function_start)
+    sql_end = runtime_source.index('"""\n    rows = psql_json_lines(', sql_start) + len('"""\n')
+    mutated = runtime_source[:sql_start] + recursive_sql + runtime_source[sql_end:]
+    assert mutated != runtime_source
+
+    with pytest.raises(query_contract.QueryContractError, match="exact reviewed runtime source"):
+        query_contract.extract_runtime_query_literals(mutated)
+
+
+@pytest.mark.parametrize(
+    "query",
+    (
+        'select c.future_private_text from "public"."claims" as c',
+        "select c . future_private_text from public.claims as c",
+        "select c /* qualifier */ . /* column */ future_private_text "
+        "from public /* schema */ . /* relation */ claims as c",
+    ),
+    ids=("quoted-identifiers", "spaced-qualification", "commented-qualification"),
+)
+def test_runtime_query_contract_normalizes_valid_postgres_identifier_qualification(query: str) -> None:
+    runtime_path = Path(__file__).resolve().parents[1] / "hermes-agent" / "leoclean-bin" / "cloudsql_memory_tool.py"
+    runtime_source = runtime_path.read_text(encoding="utf-8")
+    expected = {
+        (schema, relation): frozenset(columns)
+        for schema, relation, columns in verifier.READ_COLUMN_ALLOWLIST
+    }
+    mutated = runtime_source.replace(
+        '    claims_sql = f"""\n',
+        f"    run_psql(args, {query!r})\n" '    claims_sql = f"""\n',
+        1,
+    )
+    assert mutated != runtime_source
+
+    with pytest.raises(query_contract.QueryContractError, match="runtime query surface differs"):
+        query_contract.verify_runtime_read_columns(mutated, expected)
+
+
+def test_runtime_query_contract_rejects_search_path_dependent_base_relation() -> None:
+    runtime_path = Path(__file__).resolve().parents[1] / "hermes-agent" / "leoclean-bin" / "cloudsql_memory_tool.py"
+    runtime_source = runtime_path.read_text(encoding="utf-8")
+    mutated = runtime_source.replace(
+        '    claims_sql = f"""\n',
+        "    run_psql(args, 'select c.future_private_text from claims as c')\n"
+        '    claims_sql = f"""\n',
+        1,
+    )
+    assert mutated != runtime_source
+
+    with pytest.raises(query_contract.QueryContractError, match="unqualified relation 'claims'"):
+        query_contract.derive_runtime_read_columns(mutated)
+
+
+def test_runtime_query_contract_tracks_postgres_only_relation_syntax() -> None:
+    runtime_path = Path(__file__).resolve().parents[1] / "hermes-agent" / "leoclean-bin" / "cloudsql_memory_tool.py"
+    runtime_source = runtime_path.read_text(encoding="utf-8")
+    expected = {
+        (schema, relation): frozenset(columns)
+        for schema, relation, columns in verifier.READ_COLUMN_ALLOWLIST
+    }
+    mutated = runtime_source.replace(
+        '    claims_sql = f"""\n',
+        "    run_psql(args, 'select c.future_private_text from ONLY public.claims as c')\n"
+        '    claims_sql = f"""\n',
+        1,
+    )
+    assert mutated != runtime_source
+
+    with pytest.raises(query_contract.QueryContractError, match="runtime query surface differs"):
+        query_contract.verify_runtime_read_columns(mutated, expected)
+
+
+@pytest.mark.parametrize(
+    "query",
+    (
+        "select c.future_private_text from pg_catalog.pg_tables as p, public.claims as c",
+        "select c.future_private_text from (values (1)) as v(x), claims as c",
+        "select c.id, s.future_private_text from public.claims as c, public.sources as s",
+    ),
+    ids=("qualified-after-table", "unqualified-after-subquery", "qualified-after-canonical"),
+)
+def test_runtime_query_contract_rejects_comma_separated_base_relations(query: str) -> None:
+    runtime_path = Path(__file__).resolve().parents[1] / "hermes-agent" / "leoclean-bin" / "cloudsql_memory_tool.py"
+    runtime_source = runtime_path.read_text(encoding="utf-8")
+    mutated = runtime_source.replace(
+        '    claims_sql = f"""\n',
+        f"    run_psql(args, {query!r})\n" '    claims_sql = f"""\n',
+        1,
+    )
+    assert mutated != runtime_source
+
+    with pytest.raises(query_contract.QueryContractError, match="comma-separated base relation"):
+        query_contract.derive_runtime_read_columns(mutated)
+
+
+@pytest.mark.parametrize("alias", ("where", "order"))
+def test_runtime_query_contract_rejects_quoted_structural_keyword_aliases(alias: str) -> None:
+    runtime_path = Path(__file__).resolve().parents[1] / "hermes-agent" / "leoclean-bin" / "cloudsql_memory_tool.py"
+    runtime_source = runtime_path.read_text(encoding="utf-8")
+    query = (
+        f'select "{alias}".id, s.future_private_text '
+        f'from public.claims as "{alias}", public.sources as s'
+    )
+    mutated = runtime_source.replace(
+        '    claims_sql = f"""\n',
+        f"    run_psql(args, {query!r})\n" '    claims_sql = f"""\n',
+        1,
+    )
+    assert mutated != runtime_source
+
+    with pytest.raises(query_contract.QueryContractError, match="quoted SQL identifier"):
+        query_contract.derive_runtime_read_columns(mutated)
+
+
 def test_runtime_query_contract_rejects_unqualified_canonical_columns() -> None:
     runtime_path = Path(__file__).resolve().parents[1] / "hermes-agent" / "leoclean-bin" / "cloudsql_memory_tool.py"
     runtime_source = runtime_path.read_text(encoding="utf-8")
@@ -688,6 +883,174 @@ def test_runtime_query_contract_fails_closed_for_adversarial_query_mutations(
         query_contract.verify_runtime_read_columns(mutated, expected)
 
 
+@pytest.mark.parametrize(
+    ("replacement", "error"),
+    (
+        (
+            '    claims_sql = "select " + "pg_catalog.lo_from_bytea(0, \'\'::bytea)"\n'
+            "    run_psql(args, claims_sql)\n"
+            '    claims_sql = f"""\n',
+            "must have exactly one assignment",
+        ),
+        (
+            '    claims_sql = f"""\n',
+            "placeholder",
+        ),
+        (
+            "    runner = run_psql\n"
+            "    runner(args, \"select pg_catalog.lo_from_bytea(0, ''::bytea)\")\n"
+            '    claims_sql = f"""\n',
+            "must be called directly",
+        ),
+        (
+            "    globals()['run_psql'](args, \"select pg_catalog.lo_from_bytea(0, ''::bytea)\")\n"
+            '    claims_sql = f"""\n',
+            "indirect call expressions",
+        ),
+        (
+            "    globals()['run_' + 'psql'](args, \"select c.future_private_text "
+            "from public.claims as c\")\n"
+            '    claims_sql = f"""\n',
+            "indirect call expressions",
+        ),
+        (
+            "    get_namespace = globals\n"
+            "    namespace = get_namespace()\n"
+            "    runner = namespace['run_' + 'psql']\n"
+            "    runner(args, \"select c.future_private_text from public.claims as c\")\n"
+            '    claims_sql = f"""\n',
+            "dynamic call indirection",
+        ),
+        (
+            "    {'query': run_psql}['query'](args, \"select pg_catalog.lo_from_bytea(0, ''::bytea)\")\n"
+            '    claims_sql = f"""\n',
+            "indirect call expressions",
+        ),
+    ),
+    ids=(
+        "late-reassignment",
+        "augmented-assignment",
+        "runner-alias",
+        "globals-indirection",
+        "computed-globals-indirection",
+        "aliased-globals-indirection",
+        "dict-indirection",
+    ),
+)
+def test_runtime_query_contract_rejects_query_runner_flow_bypasses(replacement: str, error: str) -> None:
+    runtime_path = Path(__file__).resolve().parents[1] / "hermes-agent" / "leoclean-bin" / "cloudsql_memory_tool.py"
+    runtime_source = runtime_path.read_text(encoding="utf-8")
+    if error == "placeholder":
+        mutated = runtime_source.replace(
+            "    claims = psql_json_lines(args, claims_sql, db=args.canonical_db)\n",
+            "    claims_sql += \"\\nselect pg_catalog.lo_from_bytea(0, ''::bytea)\"\n"
+            "    claims = psql_json_lines(args, claims_sql, db=args.canonical_db)\n",
+            1,
+        )
+        error = "must have exactly one assignment"
+    else:
+        mutated = runtime_source.replace('    claims_sql = f"""\n', replacement, 1)
+    assert mutated != runtime_source
+
+    with pytest.raises(query_contract.QueryContractError, match=error):
+        query_contract.derive_runtime_read_columns(mutated)
+
+
+def test_runtime_query_contract_inspects_extra_calls_inside_psql_json_lines_wrapper() -> None:
+    runtime_path = Path(__file__).resolve().parents[1] / "hermes-agent" / "leoclean-bin" / "cloudsql_memory_tool.py"
+    runtime_source = runtime_path.read_text(encoding="utf-8")
+    anchor = "    rows: list[dict[str, Any]] = []\n"
+    mutated = runtime_source.replace(
+        anchor,
+        anchor + '    run_psql(args, "select c.future_private_text from public.claims as c")\n',
+        1,
+    )
+    assert mutated != runtime_source
+
+    with pytest.raises(query_contract.QueryContractError, match="runtime query surface differs"):
+        query_contract.verify_runtime_read_columns(
+            mutated,
+            {
+                (schema, relation): frozenset(columns)
+                for schema, relation, columns in verifier.READ_COLUMN_ALLOWLIST
+            },
+        )
+
+
+@pytest.mark.parametrize(
+    "routine",
+    (
+        "lo_creat",
+        "lo_create",
+        "lo_export",
+        "lo_from_bytea",
+        "lo_import",
+        "lo_open",
+        "lo_put",
+        "lo_truncate",
+        "lo_truncate64",
+        "lo_unlink",
+        "lowrite",
+    ),
+)
+@pytest.mark.parametrize("qualification", ("", "pg_catalog."), ids=("unqualified", "qualified"))
+def test_runtime_query_contract_rejects_large_object_mutators(routine: str, qualification: str) -> None:
+    runtime_path = Path(__file__).resolve().parents[1] / "hermes-agent" / "leoclean-bin" / "cloudsql_memory_tool.py"
+    runtime_source = runtime_path.read_text(encoding="utf-8")
+    mutated = runtime_source.replace(
+        '    claims_sql = f"""\n',
+        f'    run_psql(args, "select {qualification}{routine}(0)")\n    claims_sql = f"""\n',
+        1,
+    )
+    assert mutated != runtime_source
+
+    with pytest.raises(query_contract.QueryContractError, match="provider-owned large-object mutator"):
+        query_contract.derive_runtime_read_columns(mutated)
+
+
+@pytest.mark.parametrize(
+    ("sql_literal", "error"),
+    (
+        ('select pg_catalog."lo_from_bytea"(0)', "provider-owned large-object mutator"),
+        ('select "pg_catalog".lo_from_bytea(0)', "provider-owned large-object mutator"),
+        ('select "pg_catalog"."lo_from_bytea"(0)', "provider-owned large-object mutator"),
+        ('select "lo_from_bytea"(0)', "provider-owned large-object mutator"),
+        (r"\lo_import /tmp/probe", "psql meta-commands"),
+        (r"\lo_unlink 42", "psql meta-commands"),
+        (r"\copy public.claims to '/tmp/probe'", "psql meta-commands"),
+        (r"select 1 \gexec", "psql meta-commands"),
+        (r"\! /usr/bin/psql --command='select pg_catalog.lo_create(0)'", "psql meta-commands"),
+    ),
+    ids=(
+        "quoted-routine",
+        "quoted-schema",
+        "quoted-schema-and-routine",
+        "unqualified-quoted-routine",
+        "lo-import-meta-command",
+        "lo-unlink-meta-command",
+        "copy-meta-command",
+        "gexec-meta-command",
+        "shell-meta-command",
+    ),
+)
+def test_runtime_query_contract_rejects_quoted_identifiers_and_psql_meta_commands(
+    sql_literal: str,
+    error: str,
+) -> None:
+    runtime_path = Path(__file__).resolve().parents[1] / "hermes-agent" / "leoclean-bin" / "cloudsql_memory_tool.py"
+    runtime_source = runtime_path.read_text(encoding="utf-8")
+    injected = repr(sql_literal)
+    mutated = runtime_source.replace(
+        '    claims_sql = f"""\n',
+        f"    run_psql(args, {injected})\n    claims_sql = f\"\"\"\n",
+        1,
+    )
+    assert mutated != runtime_source
+
+    with pytest.raises(query_contract.QueryContractError, match=error):
+        query_contract.derive_runtime_read_columns(mutated)
+
+
 def test_runtime_query_contract_rejects_unqualified_dynamic_column_interpolation() -> None:
     runtime_path = Path(__file__).resolve().parents[1] / "hermes-agent" / "leoclean-bin" / "cloudsql_memory_tool.py"
     runtime_source = runtime_path.read_text(encoding="utf-8")
@@ -749,12 +1112,14 @@ def test_runtime_query_contract_rejects_dynamic_canonical_namespace_interpolatio
 
 
 def test_catalog_queries_use_exact_regprocedure_signatures_and_both_membership_directions() -> None:
+    identity_sql = verifier._identity_sql()
     function_sql = verifier._function_privilege_posture_sql()
     role_sql = verifier._role_posture_sql()
     owner_role_sql = verifier._stage_owner_role_posture_sql()
     stage_definition_sql = verifier._stage_function_definition_sql()
     catalog_sql = verifier._catalog_privilege_posture_sql()
 
+    assert "lo_compat_privileges" in identity_sql
     assert "pg_catalog.to_regprocedure(signature)" in function_sql
     for _name, signature, _exists, _execute in verifier.FUNCTION_PRIVILEGE_EXPECTATIONS:
         assert signature in function_sql
@@ -781,7 +1146,9 @@ def test_catalog_queries_use_exact_regprocedure_signatures_and_both_membership_d
     assert "nspname <> 'information_schema'" in catalog_sql
     assert verifier.STAGE_OWNER_DATABASE_ROLE in catalog_sql
     assert "public_database_connect_grants" in catalog_sql
-    assert "public_large_object_mutation_routine_execute" in catalog_sql
+    assert "direct_large_object_mutation_routine_acl_entries" in catalog_sql
+    assert "large_object_mutation_routine_residual" in catalog_sql
+    assert "current_setting('lo_compat_privileges') = 'off'" in catalog_sql
     assert "'lo_creat'" in catalog_sql
     assert "'lo_open'" in catalog_sql
     for schema, relation in verifier.READ_TABLE_ALLOWLIST:
@@ -804,6 +1171,35 @@ def test_any_unexpected_catalog_privilege_fails_closed(field: str) -> None:
     assert SECRET_VALUE not in str(caught.value)
 
 
+def test_large_object_provider_residual_drift_fails_closed() -> None:
+    drifted = [
+        {
+            "public_execute": public_execute,
+            "runtime_execute": public_execute,
+            "signature": signature,
+            "stage_owner_execute": public_execute,
+        }
+        for signature, public_execute in verifier.LARGE_OBJECT_MUTATION_ROUTINE_RESIDUAL
+    ]
+    drifted[0] = {**drifted[0], "runtime_execute": False}
+    runner = FakeRunner(catalog_posture_override={"large_object_mutation_routine_residual": drifted})
+
+    with pytest.raises(verifier.VerificationError) as caught:
+        run_success(runner)
+
+    assert caught.value.code == "large_object_residual_mismatch"
+    assert caught.value.check == "catalog_privilege_posture"
+
+
+def test_large_object_residual_probe_must_be_rollback_only() -> None:
+    with pytest.raises(verifier.VerificationError) as caught:
+        verifier._assert_large_object_residual_probe({"created": True, "owned_inside": False})
+
+    assert caught.value.code == "large_object_residual_probe_mismatch"
+    assert verifier._large_object_residual_probe_sql().startswith("begin;\n")
+    assert verifier._large_object_residual_probe_sql().endswith("\nrollback;")
+
+
 @pytest.mark.parametrize("field", verifier.CATALOG_TRUE_FIELDS)
 def test_any_missing_required_catalog_privilege_fails_closed(field: str) -> None:
     runner = FakeRunner(catalog_posture_override={field: False})
@@ -943,6 +1339,17 @@ def test_wrong_identity_or_non_ssl_connection_fails_before_permission_probes() -
     assert SECRET_VALUE not in str(caught.value)
 
 
+def test_runtime_identity_rejects_legacy_large_object_compatibility_mode() -> None:
+    runner = FakeRunner(identity_override={"lo_compat_privileges": "on"})
+
+    with pytest.raises(verifier.VerificationError) as caught:
+        run_success(runner)
+
+    assert caught.value.code == "lo_compat_privileges_unsafe"
+    assert caught.value.check == "database_identity"
+    assert len(runner.calls) == 2
+
+
 def test_optional_receipt_is_canonical_mode_0600_and_refuses_symlink(tmp_path: Path) -> None:
     payload = {"z": 2, "a": {"safe": True}}
     output = tmp_path / "receipt.json"

tests/test_gcp_leoclean_runtime_role_postgres.py

@@ -1,5 +1,6 @@
 from __future__ import annotations
 
+import importlib.util
 import json
 import os
 import re
@@ -8,6 +9,7 @@ import subprocess
 import time
 import uuid
 from pathlib import Path
+from types import ModuleType
 
 import pytest
 import yaml
@@ -16,6 +18,7 @@ from ops import verify_gcp_leoclean_runtime_permissions as verifier
 
 REPO_ROOT = Path(__file__).resolve().parents[1]
 ROLE_SQL = REPO_ROOT / "ops" / "gcp_leoclean_runtime_role.sql"
+RUNTIME_TOOL = REPO_ROOT / "hermes-agent" / "leoclean-bin" / "cloudsql_memory_tool.py"
 RUN_ENV = "TELEO_RUN_LEOCLEAN_RUNTIME_POSTGRES_CANARY"
 POSTGRES_IMAGE = "postgres@sha256:eb4759788a2182f08257135e61a34f2cfc3c2914079f3465d64ee62350f4d081"
 POSTGRES_VERSION_NUM = "160014"
@@ -25,6 +28,14 @@ LOCAL_RUNTIME_PASSWORD = "disposable-r2-runtime-password"
 IDENTIFIER_RE = re.compile(r"[a-z_][a-z0-9_]*\Z")
 
 
+def _load_runtime_tool() -> ModuleType:
+    spec = importlib.util.spec_from_file_location("leoclean_runtime_canary_contract", RUNTIME_TOOL)
+    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 test_ci_runs_digest_pinned_permission_canary_and_always_cleans_up() -> None:
     workflow_path = REPO_ROOT / ".github" / "workflows" / "ci.yml"
     workflow = yaml.safe_load(workflow_path.read_text(encoding="utf-8"))
@@ -83,12 +94,13 @@ def _psql(
     *,
     role: str = "postgres",
     database: str = DATABASE,
+    pgoptions: str | None = None,
 ) -> subprocess.CompletedProcess[str]:
-    return _run(
+    command = ["docker", "exec", "-i"]
+    if pgoptions is not None:
+        command.extend(("--env", f"PGOPTIONS={pgoptions}"))
+    command.extend(
         [
-            "docker",
-            "exec",
-            "-i",
             container,
             "psql",
             "--no-psqlrc",
@@ -101,9 +113,9 @@ def _psql(
             role,
             "--dbname",
             database,
-        ],
+        ]
-        input_text=sql,
     )
+    return _run(command, input_text=sql)
 
 
 def _assert_denied(container: str, sql: str, sqlstate: str = "42501") -> None:
@@ -615,30 +627,47 @@ def _assert_exact_read_contract(container: str) -> None:
         _assert_denied(container, f"select {_quote_identifier(denied_column)} from {qualified} limit 0;")
 
 
-def _assert_proposal_function_and_rollback(container: str) -> None:
+def _assert_proposal_function_and_rollback(container: str, runtime_tool: ModuleType) -> None:
+    crafted_rationale = "permission canary\\'; select pg_catalog.lo_from_bytea(0, ''::bytea); --"
+    crafted_rationale_sql = runtime_tool.sql_literal(crafted_rationale)
     calls = []
     for proposal_type in ("revise_claim", "revise_strategy", "add_edge"):
+        rationale = crafted_rationale_sql if proposal_type == "revise_claim" else "'permission canary'"
         calls.append(
             "select concat_ws('|', staged->>'proposal_type', staged->>'status', "
             "staged->>'proposed_by_handle', staged->>'channel') "
             "from (select kb_stage.stage_leoclean_proposal("
-            f"'{proposal_type}', '', 'r2:{proposal_type}', 'permission canary', "
+            f"'{proposal_type}', '', 'r2:{proposal_type}', {rationale}, "
             "'{\"canary\": true}'::jsonb) as staged) probe;"
         )
     sql = "\n".join(
         [
             "begin;",
+            "select 'strings=' || current_setting('standard_conforming_strings');",
+            "select 'readonly=' || current_setting('transaction_read_only');",
             *calls,
+            f"select 'crafted=' || count(*) from kb_stage.kb_proposals where rationale = {crafted_rationale_sql};",
             "select 'inside=' || count(id) from kb_stage.kb_proposals;",
             "rollback;",
             "select 'after=' || count(id) from kb_stage.kb_proposals;",
         ]
     )
-    output = _require_success(_psql(container, sql, role=RUNTIME_ROLE), "proposal function rollback")
+    output = _require_success(
+        _psql(
+            container,
+            sql,
+            role=RUNTIME_ROLE,
+            pgoptions=runtime_tool.RUNTIME_STANDARD_STRINGS_PGOPTIONS,
+        ),
+        "proposal function rollback",
+    )
     lines = [line.strip() for line in output.splitlines() if line.strip()]
+    assert "strings=on" in lines
+    assert "readonly=off" in lines
     assert "revise_claim|pending_review|leo|telegram" in lines
     assert "revise_strategy|pending_review|leo|telegram" in lines
     assert "add_edge|pending_review|leo|telegram" in lines
+    assert "crafted=1" in lines
     assert "inside=3" in lines
     assert "after=0" in lines
 
@@ -658,6 +687,7 @@ def _assert_write_and_escalation_denials(container: str) -> None:
         _assert_denied(container, sql)
 
     escalation_attempts = (
+        "set lo_compat_privileges = on;",
         "set role leoclean_kb_stage_owner;",
         "set role kb_review;",
         "set role kb_apply;",
@@ -679,9 +709,117 @@ def _assert_write_and_escalation_denials(container: str) -> None:
     _assert_denied(container, "select body from public.private_notes;", "42501")
 
 
+def _assert_large_object_provider_residual_rolls_back(container: str) -> None:
+    inventory_sql = """
+select pg_catalog.format(
+         '%I(%s)|%s|%s',
+         function_row.proname,
+         pg_catalog.pg_get_function_identity_arguments(function_row.oid),
+         exists (
+           select 1
+             from pg_catalog.aclexplode(
+               coalesce(function_row.proacl, pg_catalog.acldefault('f', function_row.proowner))
+             ) acl
+            where acl.grantee = 0
+              and acl.privilege_type = 'EXECUTE'
+         ),
+         pg_catalog.has_function_privilege(current_user, function_row.oid, 'EXECUTE')
+       )
+  from pg_catalog.pg_proc function_row
+  join pg_catalog.pg_namespace namespace on namespace.oid = function_row.pronamespace
+ where namespace.nspname = 'pg_catalog'
+   and function_row.proname in (
+     'lo_creat', 'lo_create', 'lo_export', 'lo_from_bytea', 'lo_import',
+     'lo_open', 'lo_put', 'lo_truncate', 'lo_truncate64', 'lo_unlink', 'lowrite'
+   )
+ order by function_row.proname, pg_catalog.pg_get_function_identity_arguments(function_row.oid);
+"""
+    observed = _require_success(_psql(container, inventory_sql, role=RUNTIME_ROLE), "large-object residual inventory")
+    expected = "\n".join(
+        f"{signature}|{'t' if public_execute else 'f'}|{'t' if public_execute else 'f'}"
+        for signature, public_execute in verifier.LARGE_OBJECT_MUTATION_ROUTINE_RESIDUAL
+    )
+    assert observed == expected
+
+    probe_sql = """
+select 'before=' || count(*)
+  from pg_catalog.pg_largeobject_metadata metadata
+ where metadata.lomowner = (select oid from pg_catalog.pg_roles where rolname = current_user);
+begin;
+select 'created=' || (pg_catalog.lo_from_bytea(0, ''::bytea) <> 0);
+select 'inside=' || count(*)
+  from pg_catalog.pg_largeobject_metadata metadata
+ where metadata.lomowner = (select oid from pg_catalog.pg_roles where rolname = current_user);
+rollback;
+select 'after=' || count(*)
+  from pg_catalog.pg_largeobject_metadata metadata
+ where metadata.lomowner = (select oid from pg_catalog.pg_roles where rolname = current_user);
+"""
+    output = _require_success(_psql(container, probe_sql, role=RUNTIME_ROLE), "rolled-back large-object residual")
+    lines = [
+        line.strip()
+        for line in output.splitlines()
+        if line.strip() and line.strip() not in {"BEGIN", "ROLLBACK"}
+    ]
+    assert lines == ["before=0", "created=true", "inside=1", "after=0"]
+
+
+def _assert_runtime_read_only_guard(container: str, runtime_tool: ModuleType) -> None:
+    read_only_options = runtime_tool.RUNTIME_READ_ONLY_PGOPTIONS
+    owner_count_sql = f"""
+select count(*)
+  from pg_catalog.pg_largeobject_metadata metadata
+ where metadata.lomowner = (select oid from pg_catalog.pg_roles where rolname = '{RUNTIME_ROLE}');
+"""
+    assert _require_success(_psql(container, owner_count_sql), "large-object count before read guard") == "0"
+    assert (
+        _require_success(
+            _psql(
+                container,
+                "show standard_conforming_strings;",
+                role=RUNTIME_ROLE,
+                pgoptions=read_only_options,
+            ),
+            "runtime string parsing guard",
+        )
+        == "on"
+    )
+    assert (
+        _require_success(
+            _psql(
+                container,
+                "show transaction_read_only;",
+                role=RUNTIME_ROLE,
+                pgoptions=read_only_options,
+            ),
+            "runtime read-only transaction guard",
+        )
+        == "on"
+    )
+    _require_success(
+        _psql(
+            container,
+            "select id, text from public.claims limit 0;",
+            role=RUNTIME_ROLE,
+            pgoptions=read_only_options,
+        ),
+        "allowed read through runtime read-only guard",
+    )
+    denied = _psql(
+        container,
+        "select pg_catalog.lo_from_bytea(0, ''::bytea);",
+        role=RUNTIME_ROLE,
+        pgoptions=read_only_options,
+    )
+    assert denied.returncode != 0
+    assert re.search(r"ERROR:\s+25006:", denied.stderr), denied.stderr
+    assert _require_success(_psql(container, owner_count_sql), "large-object count after read guard") == "0"
+
+
 @pytest.mark.skipif(os.environ.get(RUN_ENV) != "1", reason=f"set {RUN_ENV}=1 for the local PG16 canary")
 def test_digest_pinned_network_disabled_postgres_16_14_permission_contract(tmp_path: Path) -> None:
     assert shutil.which("docker") is not None, "Docker is required for the disposable permission canary"
+    runtime_tool = _load_runtime_tool()
 
     image_output = _require_success(
         _run(["docker", "image", "inspect", "--format", "{{.Id}}|{{json .RepoDigests}}", POSTGRES_IMAGE]),
@@ -783,6 +921,38 @@ def test_digest_pinned_network_disabled_postgres_16_14_permission_contract(tmp_p
             "failure-injected role SQL copy",
         )
 
+        _require_success(
+            _psql(container, "alter database teleo_canonical set lo_compat_privileges = on;"),
+            "unsafe large-object compatibility fixture",
+        )
+        unsafe_compat = _provision(container, "/tmp/gcp_leoclean_runtime_role.sql")
+        assert unsafe_compat.returncode != 0
+        assert "lo_compat_privileges must remain off" in unsafe_compat.stderr
+        assert _require_success(
+            _psql(container, f"select rolcanlogin from pg_catalog.pg_roles where rolname = '{RUNTIME_ROLE}';"),
+            "unsafe compatibility refence",
+        ) == "f"
+        _require_success(
+            _psql(container, "alter database teleo_canonical reset lo_compat_privileges;"),
+            "large-object compatibility recovery",
+        )
+
+        _require_success(
+            _psql(container, f"grant execute on function pg_catalog.lo_create(oid) to {RUNTIME_ROLE};"),
+            "direct large-object ACL fixture",
+        )
+        direct_acl = _provision(container, "/tmp/gcp_leoclean_runtime_role.sql")
+        assert direct_acl.returncode != 0
+        assert "scoped Leo roles have direct large-object routine ACLs" in direct_acl.stderr
+        assert _require_success(
+            _psql(container, f"select rolcanlogin from pg_catalog.pg_roles where rolname = '{RUNTIME_ROLE}';"),
+            "direct ACL refence",
+        ) == "f"
+        _require_success(
+            _psql(container, f"revoke execute on function pg_catalog.lo_create(oid) from {RUNTIME_ROLE};"),
+            "direct large-object ACL recovery",
+        )
+
         successful_states: list[dict[str, object]] = []
         for provision_number in range(1, 4):
             _require_success(
@@ -801,6 +971,7 @@ def test_digest_pinned_network_disabled_postgres_16_14_permission_contract(tmp_p
             assert role_state[RUNTIME_ROLE]["password_is_null"] is False
             assert role_state["leoclean_kb_stage_owner"]["password_is_scram"] is False
             assert role_state["leoclean_kb_stage_owner"]["password_is_null"] is True
+            _assert_large_object_provider_residual_rolls_back(container)
             successful_states.append(state)
 
         assert successful_states[1:] == successful_states[:1] * 2
@@ -841,7 +1012,8 @@ def test_digest_pinned_network_disabled_postgres_16_14_permission_contract(tmp_p
         }
 
         _assert_exact_read_contract(container)
-        _assert_proposal_function_and_rollback(container)
+        _assert_runtime_read_only_guard(container, runtime_tool)
+        _assert_proposal_function_and_rollback(container, runtime_tool)
         _assert_write_and_escalation_denials(container)
 
         _require_success(

tests/test_hermes_leoclean_kb_bridge_source.py

@@ -17,6 +17,7 @@ from types import SimpleNamespace
 
 import pytest
 
+from ops import derive_leoclean_runtime_query_contract as query_contract
 from scripts.working_leo_open_ended_benchmark import (
     M3TAVERSAL_DIRECT_CLAIM_FOLLOWUP_SCENARIOS,
     score_reply,
@@ -657,9 +658,195 @@ def test_cloudsql_psql_uses_only_explicit_pg_environment_fields(monkeypatch: pyt
         "PGSSLROOTCERT": module.RUNTIME_SSL_ROOT_CERT,
         "PGCONNECT_TIMEOUT": "8",
         "PGPASSWORD": "scoped-runtime-password",
+        "PGOPTIONS": module.RUNTIME_READ_ONLY_PGOPTIONS,
     }
 
 
+def test_cloudsql_runtime_rejects_dynamic_large_object_sql_before_credentials_or_subprocess(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    module = _load_module(BRIDGE_DIR / "cloudsql_memory_tool.py")
+    args = _runtime_connection_args(module)
+    sql = "select " + "pg_catalog.lo_from_bytea(0, ''::bytea);"
+    monkeypatch.setattr(module, "password", lambda _args: pytest.fail("credentials must not be acquired"))
+    monkeypatch.setattr(
+        module.subprocess,
+        "run",
+        lambda *_args, **_kwargs: pytest.fail("psql must not be launched"),
+    )
+
+    with pytest.raises(SystemExit, match="large-object mutators"):
+        module.run_psql(args, sql)
+
+
+@pytest.mark.parametrize(
+    "sql",
+    (
+        "select 1; select 2;",
+        r"\copy public.claims to '/tmp/probe'",
+        "with changed as (update public.claims set status = 'closed' returning id) select id from changed;",
+        "copy public.claims to stdout;",
+        "select 1 into temporary runtime_probe;",
+        "call public.apply_proposal();",
+        "do $$ begin perform pg_catalog.lo_create(0); end $$;",
+        "select pg_catalog.set_config('default_transaction_read_only', 'off', false);",
+        "select pg_catalog.nextval('runtime_probe_sequence');",
+        "select pg_catalog.pg_notify('runtime_probe', 'payload');",
+        "select pg_catalog.pg_try_advisory_lock_shared(42);",
+        "select pg_catalog.pg_advisory_unlock_all();",
+        r'''select pg_catalog.U&"lo\005ffrom\005fbytea"(0, ''::bytea);''',
+        "select 1 /* unterminated",
+        "select 'unterminated",
+    ),
+)
+def test_cloudsql_runtime_sql_boundary_rejects_unreviewed_statement_shapes(sql: str) -> None:
+    module = _load_module(BRIDGE_DIR / "cloudsql_memory_tool.py")
+
+    with pytest.raises(SystemExit, match="Runtime SQL was rejected"):
+        module.validate_runtime_sql(sql)
+
+
+def test_cloudsql_runtime_sql_boundary_ignores_keywords_inside_literals_and_comments() -> None:
+    module = _load_module(BRIDGE_DIR / "cloudsql_memory_tool.py")
+
+    module.validate_runtime_sql(
+        "select 'update; \\lo_unlink pg_catalog.lo_from_bytea(0)'::text /* delete public.claims */;"
+    )
+
+
+def test_cloudsql_sql_literal_with_backslash_quote_remains_one_validated_value() -> None:
+    module = _load_module(BRIDGE_DIR / "cloudsql_memory_tool.py")
+    payload = "receipt\\'; select pg_catalog.lo_from_bytea(0, ''::bytea); --"
+    sql = f"select {module.sql_literal(payload)}::text;"
+
+    module.validate_runtime_sql(sql)
+    assert sql == "select 'receipt\\''; select pg_catalog.lo_from_bytea(0, ''''::bytea); --'::text;"
+
+
+def test_cloudsql_runtime_rejects_carriage_return_statement_smuggling_before_execution(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    module = _load_module(BRIDGE_DIR / "cloudsql_memory_tool.py")
+    args = _runtime_connection_args(module)
+    sql = (
+        "select kb_stage.stage_leoclean_proposal("
+        "'revise_claim', 'telegram', 'receipt:1', 'reason', '{}'::jsonb) "
+        "-- hidden from an LF-only lexer\r; select pg_catalog.lo_from_bytea(0, ''::bytea);"
+    )
+    monkeypatch.setattr(module, "password", lambda _args: pytest.fail("credentials must not be acquired"))
+    monkeypatch.setattr(
+        module.subprocess,
+        "run",
+        lambda *_args, **_kwargs: pytest.fail("psql must not be launched"),
+    )
+
+    with pytest.raises(SystemExit, match="carriage returns"):
+        module.run_psql(args, sql, capability=module.RUNTIME_SQL_STAGE_PROPOSAL_CAPABILITY)
+
+
+def test_cloudsql_runtime_sql_boundary_separates_reads_from_the_one_proposal_function() -> None:
+    module = _load_module(BRIDGE_DIR / "cloudsql_memory_tool.py")
+    proposal_sql = """
+with staged as (
+  select kb_stage.stage_leoclean_proposal(
+    'revise_claim', 'telegram', 'receipt:1', 'reason', '{}'::jsonb
+  ) as proposal
+)
+select proposal from staged;
+"""
+
+    module.validate_runtime_sql(sql=proposal_sql, capability=module.RUNTIME_SQL_STAGE_PROPOSAL_CAPABILITY)
+    with pytest.raises(SystemExit, match="application functions are not allowed for reads"):
+        module.validate_runtime_sql(proposal_sql)
+    with pytest.raises(SystemExit, match="large-object mutators"):
+        module.validate_runtime_sql(
+            proposal_sql.replace(
+                "select proposal from staged;",
+                "select proposal, pg_catalog.lo_from_bytea(0, ''::bytea) from staged;",
+            ),
+            capability=module.RUNTIME_SQL_STAGE_PROPOSAL_CAPABILITY,
+        )
+    with pytest.raises(SystemExit, match="exceeds the reviewed staging function surface"):
+        module.validate_runtime_sql(
+            proposal_sql.replace(
+                "select proposal from staged;",
+                "select proposal, kb_stage.stage_leoclean_proposal("
+                "'revise_claim', null, null, 'again', '{}'::jsonb) from staged;",
+            ),
+            capability=module.RUNTIME_SQL_STAGE_PROPOSAL_CAPABILITY,
+        )
+    with pytest.raises(SystemExit, match="side-effecting system functions"):
+        module.validate_runtime_sql(
+            proposal_sql.replace(
+                "select proposal from staged;",
+                "select proposal, pg_catalog.pg_notify('attacker', 'payload') from staged;",
+            ),
+            capability=module.RUNTIME_SQL_STAGE_PROPOSAL_CAPABILITY,
+        )
+
+
+def test_every_mechanically_extracted_runtime_query_passes_its_exact_capability() -> None:
+    module = _load_module(BRIDGE_DIR / "cloudsql_memory_tool.py")
+    runtime_source = (BRIDGE_DIR / "cloudsql_memory_tool.py").read_text(encoding="utf-8")
+    queries = query_contract.extract_runtime_query_literals(runtime_source)
+
+    assert len(queries) == 18
+    assert [capability for _line, _sql, capability in queries].count(
+        module.RUNTIME_SQL_STAGE_PROPOSAL_CAPABILITY
+    ) == 2
+    for _line, sql, capability in queries:
+        module.validate_runtime_sql(sql, capability=capability)
+
+
+def test_cloudsql_stage_proposal_capability_reaches_psql_without_read_only_mode(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    module = _load_module(BRIDGE_DIR / "cloudsql_memory_tool.py")
+    args = _runtime_connection_args(module)
+    _configure_runtime_environment(monkeypatch)
+    captured: dict[str, object] = {}
+    sql = (
+        "select kb_stage.stage_leoclean_proposal("
+        "'revise_claim', 'telegram', 'receipt:1', 'reason', '{}'::jsonb);"
+    )
+
+    def fake_run(command, **kwargs):
+        captured["command"] = command
+        captured["kwargs"] = kwargs
+        return SimpleNamespace(returncode=0, stdout='{"id":"proposal-id"}\n', stderr="")
+
+    monkeypatch.setattr(module, "runtime_password_from_metadata", lambda: "scoped-runtime-password")
+    monkeypatch.setattr(module.subprocess, "run", fake_run)
+
+    module.run_psql(args, sql, capability=module.RUNTIME_SQL_STAGE_PROPOSAL_CAPABILITY)
+
+    assert captured["command"] == ["/usr/bin/psql", "-X", "-At", "-q", "-v", "ON_ERROR_STOP=1"]
+    kwargs = captured["kwargs"]
+    assert isinstance(kwargs, dict)
+    assert kwargs["env"]["PGOPTIONS"] == module.RUNTIME_STANDARD_STRINGS_PGOPTIONS
+
+
+def test_cloudsql_clone_mode_rejects_proposal_capability_before_credentials_or_subprocess(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    module = _load_module(BRIDGE_DIR / "cloudsql_memory_tool.py")
+    args = _runtime_connection_args(module)
+    args.credential_mode = "clone-readonly"
+    sql = (
+        "select kb_stage.stage_leoclean_proposal("
+        "'revise_claim', 'telegram', 'receipt:1', 'reason', '{}'::jsonb);"
+    )
+    monkeypatch.setattr(module, "password", lambda _args: pytest.fail("credentials must not be acquired"))
+    monkeypatch.setattr(
+        module.subprocess,
+        "run",
+        lambda *_args, **_kwargs: pytest.fail("psql must not be launched"),
+    )
+
+    with pytest.raises(SystemExit, match="requires the scoped runtime credential mode"):
+        module.run_psql(args, sql, capability=module.RUNTIME_SQL_STAGE_PROPOSAL_CAPABILITY)
+
+
 def test_cloudsql_runtime_metadata_and_secret_requests_are_exact_and_silent(
     monkeypatch: pytest.MonkeyPatch,
     capsys: pytest.CaptureFixture[str],
@@ -927,10 +1114,11 @@ def test_cloudsql_runtime_read_marker_rejects_wrong_live_identity(
 
 def test_cloudsql_proposal_calls_leave_leo_identity_to_the_database(monkeypatch: pytest.MonkeyPatch) -> None:
     module = _load_module(BRIDGE_DIR / "cloudsql_memory_tool.py")
-    captured: list[str] = []
+    captured: list[tuple[str, str]] = []
 
-    def fake_psql_json_lines(_args, sql, db=None):
+    def fake_psql_json_lines(_args, sql, db=None, *, capability="read"):
-        captured.append(sql)
+        module.validate_runtime_sql(sql, capability=capability)
+        captured.append((sql, capability))
         return [{"id": "proposal-id", "database": db}]
 
     monkeypatch.setattr(module, "psql_json_lines", fake_psql_json_lines)
@@ -961,12 +1149,16 @@ def test_cloudsql_proposal_calls_leave_leo_identity_to_the_database(monkeypatch:
     module.propose_core_change(core_args)
     module.propose_edge(edge_args)
 
-    assert "p.proposed_by_handle" not in captured[0]
+    assert [capability for _sql, capability in captured] == [
-    assert "as proposed_by_handle" not in captured[0]
+        module.RUNTIME_SQL_STAGE_PROPOSAL_CAPABILITY,
-    assert "p.proposed_by_handle" not in captured[1]
+        module.RUNTIME_SQL_STAGE_PROPOSAL_CAPABILITY,
-    assert "as proposed_by_handle" not in captured[1]
+    ]
-    assert "stage_leoclean_proposal(\n    p.proposal_type,\n    p.channel," in captured[0]
+    assert "p.proposed_by_handle" not in captured[0][0]
-    assert "stage_leoclean_proposal(\n    'add_edge',\n    p.channel," in captured[1]
+    assert "as proposed_by_handle" not in captured[0][0]
+    assert "p.proposed_by_handle" not in captured[1][0]
+    assert "as proposed_by_handle" not in captured[1][0]
+    assert "stage_leoclean_proposal(\n    p.proposal_type,\n    p.channel," in captured[0][0]
+    assert "stage_leoclean_proposal(\n    'add_edge',\n    p.channel," in captured[1][0]
 
 
 def test_gcp_runtime_role_uses_one_narrow_staging_function() -> None:
@@ -981,11 +1173,13 @@ def test_gcp_runtime_role_uses_one_narrow_staging_function() -> None:
     assert "unexpectedly has schema CREATE" in sql
     assert "runtime CONNECT reaches a noncanonical database" in sql
     assert "PUBLIC retains database CONNECT" in sql
-    assert "PUBLIC retains large-object mutator EXECUTE" in sql
+    assert "provider-owned large-object routine residual drifted" in sql
+    assert "scoped Leo roles have direct large-object routine ACLs" in sql
+    assert "lo_compat_privileges must remain off" in sql
     assert "REVOKE TEMP FROM PUBLIC would be a" in sql
     assert "can execute unexpected SECURITY DEFINER functions" in sql
     assert "revoke connect on database %I from public" in sql
-    assert "revoke execute on function pg_catalog.%I(%s) from public" in sql
+    assert "revoke execute on function pg_catalog.%I(%s) from public" not in sql
     assert re.search(
         r"grant\s+insert\s*\([^)]*proposal_type[^)]*payload[^)]*\)\s+"
         r"on\s+kb_stage\.kb_proposals\s+to\s+leoclean_kb_stage_owner",

tests/test_teleo_agent_systemd.py

@@ -1,3 +1,4 @@
+import json
 import os
 import resource
 import signal
@@ -28,6 +29,13 @@ def load_service_context_runner() -> dict[str, object]:
     return namespace
 
 
+def embedded_permission_receipt_validator_source() -> str:
+    script = (REPO_ROOT / "deploy" / "sync-gcp-leoclean-runtime.sh").read_text()
+    marker = 'sudo /usr/bin/python3 -I - "$receipt_path" <<\'PY\'\n'
+    start = script.index(marker) + len(marker)
+    return script[start : script.index("\nPY\n", start)]
+
+
 def test_teleo_agent_template_supports_optional_per_agent_env_file():
     unit = (REPO_ROOT / "systemd" / "teleo-agent@.service").read_text()
 
@@ -396,6 +404,11 @@ def test_gcp_runtime_sync_is_source_bound_and_scoped():
     assert "for capability_field in CapInh CapPrm CapEff CapBnd CapAmb" in script
     assert "0000000000000000" in script
     assert '"current_tier": "T3_live_readonly"' in script
+    assert '"schema_version": 2' in script
+    assert '"arbitrary_database_writes_denied": False' in script
+    assert '"production_eligible": False' in script
+    assert '"public_large_object_mutation_residual": "present"' in script
+    assert '"staging_only": True' in script
     assert '"status": "pass"' in script
     assert "UnsetEnvironment=PGPASSWORD" in dropin
     assert "UnsetEnvironment=GOOGLE_APPLICATION_CREDENTIALS CLOUDSDK_CORE_PROJECT CLOUDSDK_AUTH_ACCESS_TOKEN" in dropin
@@ -430,6 +443,53 @@ def test_gcp_runtime_sync_is_source_bound_and_scoped():
     assert "gcp-teleo-pgvector-standby-postgres-password" not in dropin
 
 
+def test_gcp_permission_receipt_validator_binds_staging_only_safety_ceiling(tmp_path: Path) -> None:
+    source = embedded_permission_receipt_validator_source()
+    expected_safety: dict[str, object] = {
+        "arbitrary_database_writes_denied": False,
+        "canonical_relation_writes_denied": True,
+        "direct_stage_table_writes_denied": True,
+        "production_eligible": False,
+        "proposal_staging": "function_only",
+        "public_large_object_mutation_residual": "present",
+        "staging_only": True,
+    }
+    receipt: dict[str, object] = {
+        "artifact": "gcp_leoclean_runtime_permissions",
+        "current_tier": "T3_live_readonly",
+        "mode": "live_private_gcp_staging",
+        "required_tier": "T3_live_readonly",
+        "schema_version": 2,
+        "status": "pass",
+        "safety": expected_safety,
+    }
+    receipt_path = tmp_path / "receipt.json"
+
+    def validate(value: dict[str, object]) -> subprocess.CompletedProcess[str]:
+        receipt_path.write_text(json.dumps(value), encoding="utf-8")
+        return subprocess.run(
+            [sys.executable, "-I", "-", str(receipt_path)],
+            input=source,
+            text=True,
+            capture_output=True,
+            check=False,
+        )
+
+    accepted = validate(receipt)
+    assert accepted.returncode == 0, accepted.stderr
+    assert '"permission_receipt": "validated"' in accepted.stdout
+
+    for field, expected in expected_safety.items():
+        unsafe = not expected if isinstance(expected, bool) else "unexpected"
+        rejected = validate({**receipt, "safety": {**expected_safety, field: unsafe}})
+        assert rejected.returncode != 0
+        assert f"safety.{field}" in rejected.stderr
+
+    missing = validate({key: value for key, value in receipt.items() if key != "safety"})
+    assert missing.returncode != 0
+    assert "safety" in missing.stderr
+
+
 def test_gcp_runtime_embedded_remote_programs_are_bash_syntax_valid() -> None:
     script = (REPO_ROOT / "deploy" / "sync-gcp-leoclean-runtime.sh").read_text()
     blocks: dict[str, str] = {}