Harden Cloud SQL staging runtime boundary

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.

ops/derive_leoclean_runtime_query_contract.py

@@ -29,6 +29,17 @@ 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)
 DYNAMIC_SQL_VALUE = "__teleo_dynamic_sql_value__"
 UNREVIEWED_DYNAMIC_SQL = "__teleo_unreviewed_dynamic_sql__"
@@ -123,8 +134,16 @@ def extract_canonical_query_literals(source: str) -> tuple[tuple[int, str], ...]
                 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 LARGE_OBJECT_MUTATOR_REFERENCE.search(scrubbed):
+                raise QueryContractError(
+                    f"line {line}: runtime query invokes a provider-owned large-object mutator"
+                )
             if CANONICAL_RELATION.search(sql):
                 queries.add((line, sql))
 

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",
     }
 

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
@@ -688,6 +746,80 @@ def test_runtime_query_contract_fails_closed_for_adversarial_query_mutations(
         query_contract.verify_runtime_read_columns(mutated, expected)
 
 
+@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 +881,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 +915,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 +940,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 +1108,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

@@ -658,6 +658,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,6 +680,61 @@ 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"]
+
+
 @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"
@@ -783,6 +839,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 +889,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

tests/test_hermes_leoclean_kb_bridge_source.py

@@ -981,11 +981,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] = {}