Merge pull request #229 from living-ip/fix/cloudsql-prereq-composability

Preserve scoped runtime ACLs across prerequisite reruns

docs/workstreams/gcp-leoclean/goals.md

@@ -24,37 +24,56 @@ until the GCP service has passed soak and restore testing.
 
 ## Current state
 
-- Current main/base: `832f0c3f47abe3eded61f4be585fde3c4f00b91a`.
-- The GCP VM, private Cloud SQL instance, immutable image, and scoped Secret
-  Manager secret exist.
-- The Cloud SQL-compatible least-privilege correction is implemented on
-  `fix/cloudsql-large-object-boundary` and is awaiting publication and human
-  review.
-- Exact local evidence: 2,601 repository tests passed with 2 expected skips;
-  the digest-pinned, network-disabled PostgreSQL 16.14 lifecycle canary passed
-  2 tests; Ruff, Bash syntax, and `git diff --check` passed.
-- The `leoclean_kb_runtime` role has not yet been successfully provisioned in
-  Cloud SQL.
-- The VM service account has not yet been granted access to the scoped secret.
-- The no-send GCP service is not installed or running.
-- V3 knowledge reconstruction and promotion have not yet been reconciled from
-  PR #220 into canonical Cloud SQL.
-- Telegram still does not route to GCP.
+- Current main/base: `3569f4f2fb016e36f5ebeb27f01f3af91e1c360f`.
+- PRs #220, #221, #226, #227, and #228 are merged. Main therefore contains the
+  source-first V3 rebuild foundation, the PostgreSQL large-object boundary,
+  bounded Observatory prerequisite preservation, Cloud SQL provider-database
+  residual handling, and the manual exact-revision no-send publication
+  workflow.
+- A narrow current-main correction is in progress so rerunning
+  `scripts/kb_apply_prereqs.sql` preserves only the exact scoped runtime and
+  proposal-owner ACLs created by `ops/gcp_leoclean_runtime_role.sql` on the
+  prerequisite's protected table/column surface. Broader grants on that
+  surface and direct role membership continue to fail closed; the runtime
+  provisioner and verifier remain authoritative for role attributes, schema
+  privileges, routines, and the complete least-privilege posture.
+- Local evidence for that correction: 230 focused tests passed; the
+  digest-pinned, network-disabled PostgreSQL 16.14 lifecycle canary passed;
+  two prerequisite reruns were stable; injected ACL and membership drift
+  failed closed; Ruff and `git diff --check` passed.
+- The GCP VM, private Cloud SQL instance, and scoped Secret Manager secret are
+  known to exist, but no new current-main no-send image has yet been published
+  by the merged #228 workflow.
+- Live Cloud SQL provisioning of `leoclean_kb_runtime`, VM access to the scoped
+  secret, installation of the no-send service, effective running-process
+  identity, restart behavior, and rollback remain unproven.
+- The source-first V3 reconstruction code is on main, but no canonical GCP
+  promotion is implied by that merge.
+- Telegram still does not route to GCP. The VPS, production traffic, and
+  canonical proposal approvals remain untouched.
 
 ## Execution order
 
-1. Publish and review the Cloud SQL authority correction.
-2. Provision `leoclean_kb_runtime`, grant scoped-secret access, install the
-   immutable no-send service, and prove identity, permissions, restart, and
-   rollback.
-3. Extract the required V3 reconstruction changes from PR #220, rebuild or
-   migrate into a separately identified canonical candidate database, and
-   prove row/hash/source receipts before promotion.
-4. Bind the GCP runtime to the verified V3 canonical database and run database
+1. Review and merge the narrow PostgreSQL migration-order/composability
+   correction from current main.
+2. Run the merged GCP preflight against that exact main revision, then manually
+   dispatch #228's publication workflow for the same revision. Verify the
+   resulting immutable image/config digests and publication receipt.
+3. Provision `leoclean_kb_runtime`, grant only the staging VM access to the
+   scoped secret, and prove the Cloud SQL role's allowed reads and
+   function-only proposal staging plus denied writes and escalation.
+4. Install the exact immutable image as a separate no-send GCP service. Prove
+   the running process uses the scoped identity without administrator fallback,
+   then prove restart and rollback.
+5. Reconstruct or reconcile the V3 knowledge database into a separately
+   identified canonical candidate and prove row/hash/source receipts before
+   any promotion.
+6. Bind the GCP runtime to the verified V3 canonical database and run database
    and behavioral parity checks without Telegram sends.
-5. Switch Telegram delivery to the GCP service, prove real response receipts,
-   soak, restart, and restore behavior, and retain an immediate rollback path.
-6. Revoke obsolete runtime credentials after soak. VPS destruction remains a
+7. Switch Telegram delivery to the GCP service only after a separate human
+   authorization, prove real response receipts, soak, restart, and restore
+   behavior, and retain an immediate rollback path.
+8. Revoke obsolete runtime credentials after soak. VPS destruction remains a
    separate explicit decision.
 
 ## Authority and rollback boundaries
@@ -72,7 +91,8 @@ until the GCP service has passed soak and restore testing.
 
 ## 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.
+The composability correction requires exact-revision database/security review
+before merge. GCP role/IAM mutation, V3 canonical promotion, and Telegram
+cutover remain 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/gcp_leoclean_runtime_role.sql

@@ -397,10 +397,12 @@ revoke all privileges on all procedures in schema public, kb_stage
   from leoclean_kb_runtime, leoclean_kb_stage_owner;
 
 -- Built-in defaults also grant PUBLIC EXECUTE on newly created application
--- routines. Preserve every existing non-scoped role's effective access,
+-- routines. Preserve every existing non-scoped, non-superuser role's effective access,
 -- remove PUBLIC/scoped execution from the complete application routine set,
 -- and restore the inventory explicitly. The reviewed staging function is
--- normalized again after CREATE OR REPLACE below.
+-- normalized again after CREATE OR REPLACE below. Superusers retain their
+-- inherent access and must not receive redundant explicit ACLs that would make
+-- the guarded review/apply prerequisite non-composable on a local PG16 canary.
 create temporary table leoclean_preserved_app_routine_execute (
   role_name name not null,
   function_oid oid not null,
@@ -413,6 +415,7 @@ select role_row.rolname, function_row.oid
  cross join pg_catalog.pg_proc function_row
   join pg_catalog.pg_namespace namespace on namespace.oid = function_row.pronamespace
  where role_row.rolname not in ('leoclean_kb_runtime', 'leoclean_kb_stage_owner')
+   and not role_row.rolsuper
    and namespace.nspname in ('public', 'kb_stage')
    and pg_catalog.has_function_privilege(role_row.oid, function_row.oid, 'EXECUTE');
 

scripts/kb_apply_prereqs.sql

@@ -50,6 +50,268 @@ alter role kb_gate_owner nologin noinherit nosuperuser nocreatedb nocreaterole n
 alter role kb_review login noinherit nosuperuser nocreatedb nocreaterole noreplication nobypassrls;
 alter role kb_apply login noinherit nosuperuser nocreatedb nocreaterole noreplication nobypassrls;
 
+-- The independently provisioned GCP runtime keeps an exact column-level read
+-- projection. Record that reviewed contract in this transaction so a later
+-- review/apply prerequisite rerun can preserve the scoped grants without
+-- accepting arbitrary runtime ACLs. The matching test binds these rows to the
+-- runtime-role provisioning manifest and the executable query contract.
+create temporary table kb_apply_prereqs_leoclean_read_column_allowlist (
+  schema_name name not null,
+  relation_name name not null,
+  column_name name not null,
+  column_ordinal smallint not null,
+  primary key (schema_name, relation_name, column_name),
+  unique (schema_name, relation_name, column_ordinal)
+) on commit drop;
+
+insert into pg_temp.kb_apply_prereqs_leoclean_read_column_allowlist (
+  schema_name,
+  relation_name,
+  column_name,
+  column_ordinal
+) values
+  ('public', 'agents', 'id', 1),
+  ('public', 'agents', 'handle', 2),
+  ('public', 'claims', 'id', 1),
+  ('public', 'claims', 'type', 2),
+  ('public', 'claims', 'text', 3),
+  ('public', 'claims', 'status', 4),
+  ('public', 'claims', 'confidence', 5),
+  ('public', 'claims', 'tags', 6),
+  ('public', 'claims', 'superseded_by', 7),
+  ('public', 'claims', 'created_at', 8),
+  ('public', 'claims', 'updated_at', 9),
+  ('public', 'claim_evidence', 'claim_id', 1),
+  ('public', 'claim_evidence', 'source_id', 2),
+  ('public', 'claim_evidence', 'role', 3),
+  ('public', 'claim_evidence', 'weight', 4),
+  ('public', 'claim_edges', 'id', 1),
+  ('public', 'claim_edges', 'from_claim', 2),
+  ('public', 'claim_edges', 'to_claim', 3),
+  ('public', 'claim_edges', 'edge_type', 4),
+  ('public', 'sources', 'id', 1),
+  ('public', 'sources', 'source_type', 2),
+  ('public', 'sources', 'url', 3),
+  ('public', 'sources', 'storage_path', 4),
+  ('public', 'sources', 'excerpt', 5),
+  ('public', 'sources', 'hash', 6),
+  ('public', 'personas', 'agent_id', 1),
+  ('public', 'personas', 'name', 2),
+  ('public', 'personas', 'voice', 3),
+  ('public', 'personas', 'role', 4),
+  ('public', 'personas', 'source_ref', 5),
+  ('public', 'personas', 'lens', 6),
+  ('public', 'strategies', 'agent_id', 1),
+  ('public', 'strategies', 'diagnosis', 2),
+  ('public', 'strategies', 'guiding_policy', 3),
+  ('public', 'strategies', 'proximate_objectives', 4),
+  ('public', 'strategies', 'version', 5),
+  ('public', 'strategies', 'active', 6),
+  ('public', 'beliefs', 'agent_id', 1),
+  ('public', 'beliefs', 'level', 2),
+  ('public', 'beliefs', 'statement', 3),
+  ('public', 'beliefs', 'falsifier', 4),
+  ('public', 'beliefs', 'rank', 5),
+  ('public', 'beliefs', 'status', 6),
+  ('public', 'blindspots', 'agent_id', 1),
+  ('public', 'blindspots', 'name', 2),
+  ('public', 'blindspots', 'description', 3),
+  ('public', 'blindspots', 'correction', 4),
+  ('public', 'blindspots', 'kind', 5),
+  ('public', 'blindspots', 'status', 6),
+  ('public', 'agent_roles', 'agent_id', 1),
+  ('public', 'agent_roles', 'title', 2),
+  ('public', 'agent_roles', 'description', 3),
+  ('public', 'peer_models', 'subject_id', 1),
+  ('public', 'peer_models', 'peer_id', 2),
+  ('public', 'peer_models', 'domain', 3),
+  ('public', 'peer_models', 'outranks_on', 4),
+  ('public', 'peer_models', 'deference_rule', 5),
+  ('public', 'behavioral_rules', 'agent_id', 1),
+  ('public', 'behavioral_rules', 'category', 2),
+  ('public', 'behavioral_rules', 'rule', 3),
+  ('public', 'behavioral_rules', 'rationale', 4),
+  ('public', 'contributor_rules', 'agent_id', 1),
+  ('public', 'contributor_rules', 'name', 2),
+  ('public', 'contributor_rules', 'directive', 3),
+  ('public', 'contributor_rules', 'ci_tier', 4),
+  ('public', 'contributor_rules', 'weighting', 5),
+  ('public', 'contributor_rules', 'rationale', 6),
+  ('public', 'reasoning_tools', 'agent_id', 1),
+  ('public', 'reasoning_tools', 'name', 2),
+  ('public', 'reasoning_tools', 'description', 3),
+  ('public', 'reasoning_tools', 'category', 4),
+  ('public', 'governance_gates', 'agent_id', 1),
+  ('public', 'governance_gates', 'name', 2),
+  ('public', 'governance_gates', 'criteria', 3),
+  ('public', 'governance_gates', 'evidence_bar', 4),
+  ('public', 'governance_gates', 'pass_condition', 5),
+  ('kb_stage', 'kb_proposals', 'id', 1),
+  ('kb_stage', 'kb_proposals', 'proposal_type', 2),
+  ('kb_stage', 'kb_proposals', 'status', 3),
+  ('kb_stage', 'kb_proposals', 'proposed_by_handle', 4),
+  ('kb_stage', 'kb_proposals', 'proposed_by_agent_id', 5),
+  ('kb_stage', 'kb_proposals', 'channel', 6),
+  ('kb_stage', 'kb_proposals', 'source_ref', 7),
+  ('kb_stage', 'kb_proposals', 'rationale', 8),
+  ('kb_stage', 'kb_proposals', 'payload', 9),
+  ('kb_stage', 'kb_proposals', 'reviewed_by_handle', 10),
+  ('kb_stage', 'kb_proposals', 'reviewed_at', 11),
+  ('kb_stage', 'kb_proposals', 'review_note', 12),
+  ('kb_stage', 'kb_proposals', 'applied_by_handle', 13),
+  ('kb_stage', 'kb_proposals', 'applied_at', 14),
+  ('kb_stage', 'kb_proposals', 'created_at', 15),
+  ('kb_stage', 'kb_proposals', 'updated_at', 16);
+
+-- The proposal function owner has only the exact underlying table rights its
+-- SECURITY DEFINER implementation needs. These rows describe bounded ACLs that
+-- may already exist when this prerequisite is rerun after runtime provisioning.
+create temporary table kb_apply_prereqs_leoclean_stage_owner_column_allowlist (
+  schema_name name not null,
+  relation_name name not null,
+  column_name name not null,
+  privilege_type text not null,
+  primary key (schema_name, relation_name, column_name, privilege_type)
+) on commit drop;
+
+insert into pg_temp.kb_apply_prereqs_leoclean_stage_owner_column_allowlist (
+  schema_name,
+  relation_name,
+  column_name,
+  privilege_type
+) values
+  ('public', 'agents', 'id', 'SELECT'),
+  ('public', 'agents', 'handle', 'SELECT'),
+  ('kb_stage', 'kb_proposals', 'proposal_type', 'INSERT'),
+  ('kb_stage', 'kb_proposals', 'status', 'INSERT'),
+  ('kb_stage', 'kb_proposals', 'proposed_by_handle', 'INSERT'),
+  ('kb_stage', 'kb_proposals', 'proposed_by_agent_id', 'INSERT'),
+  ('kb_stage', 'kb_proposals', 'channel', 'INSERT'),
+  ('kb_stage', 'kb_proposals', 'source_ref', 'INSERT'),
+  ('kb_stage', 'kb_proposals', 'rationale', 'INSERT'),
+  ('kb_stage', 'kb_proposals', 'payload', 'INSERT');
+
+-- Snapshot the exact bounded external ACLs present before normalization. Final
+-- verification compares against these rows, so this migration must preserve
+-- them exactly; it cannot silently add another otherwise allowlisted grant.
+create temporary table kb_apply_prereqs_preserved_table_acls (
+  schema_name name not null,
+  relation_name name not null,
+  grantee name not null,
+  privilege_type text not null,
+  is_grantable boolean not null,
+  primary key (schema_name, relation_name, grantee, privilege_type)
+) on commit drop;
+
+insert into pg_temp.kb_apply_prereqs_preserved_table_acls (
+  schema_name,
+  relation_name,
+  grantee,
+  privilege_type,
+  is_grantable
+)
+select namespace.nspname,
+       relation.relname,
+       grantee_role.rolname,
+       pg_catalog.upper(acl.privilege_type),
+       acl.is_grantable
+  from pg_catalog.pg_class relation
+  join pg_catalog.pg_namespace namespace on namespace.oid = relation.relnamespace
+ cross join lateral pg_catalog.aclexplode(
+   coalesce(relation.relacl, pg_catalog.acldefault('r', relation.relowner))
+ ) acl
+  join pg_catalog.pg_roles grantee_role on grantee_role.oid = acl.grantee
+ where (
+   grantee_role.rolname = 'kb_observatory_read'
+   and (namespace.nspname, relation.relname) in (
+     ('public', 'claims'),
+     ('public', 'sources'),
+     ('public', 'claim_evidence'),
+     ('public', 'claim_edges'),
+     ('kb_stage', 'kb_proposals')
+   )
+   and pg_catalog.upper(acl.privilege_type) = 'SELECT'
+   and not acl.is_grantable
+ ) or (
+   grantee_role.rolname = 'leoclean_kb_stage_owner'
+   and namespace.nspname = 'kb_stage'
+   and relation.relname = 'kb_proposals'
+   and pg_catalog.upper(acl.privilege_type) = 'SELECT'
+   and not acl.is_grantable
+ );
+
+create temporary table kb_apply_prereqs_preserved_column_acls (
+  schema_name name not null,
+  relation_name name not null,
+  column_name name not null,
+  grantee name not null,
+  privilege_type text not null,
+  is_grantable boolean not null,
+  primary key (schema_name, relation_name, column_name, grantee, privilege_type)
+) on commit drop;
+
+insert into pg_temp.kb_apply_prereqs_preserved_column_acls (
+  schema_name,
+  relation_name,
+  column_name,
+  grantee,
+  privilege_type,
+  is_grantable
+)
+select namespace.nspname,
+       relation.relname,
+       attribute.attname,
+       grantee_role.rolname,
+       pg_catalog.upper(acl.privilege_type),
+       acl.is_grantable
+  from pg_catalog.pg_attribute attribute
+  join pg_catalog.pg_class relation on relation.oid = attribute.attrelid
+  join pg_catalog.pg_namespace namespace on namespace.oid = relation.relnamespace
+ cross join lateral pg_catalog.aclexplode(attribute.attacl) acl
+  join pg_catalog.pg_roles grantee_role on grantee_role.oid = acl.grantee
+ where (namespace.nspname, relation.relname) in (
+         ('public', 'agents'),
+         ('public', 'strategies'),
+         ('public', 'strategy_nodes'),
+         ('public', 'claim_evidence'),
+         ('public', 'claim_evidence_assessments'),
+         ('public', 'claim_edges'),
+         ('public', 'claims'),
+         ('public', 'sources'),
+         ('public', 'reasoning_tools'),
+         ('kb_stage', 'kb_proposals'),
+         ('kb_stage', 'kb_review_principals'),
+         ('kb_stage', 'kb_proposal_approvals'),
+         ('kb_stage', 'teleo_v3_installation_ownership')
+       )
+   and attribute.attnum > 0
+   and not attribute.attisdropped
+   and not acl.is_grantable
+   and (
+     (
+       grantee_role.rolname = 'leoclean_kb_runtime'
+       and pg_catalog.upper(acl.privilege_type) = 'SELECT'
+       and exists (
+         select 1
+           from pg_temp.kb_apply_prereqs_leoclean_read_column_allowlist allowlist
+          where allowlist.schema_name = namespace.nspname
+            and allowlist.relation_name = relation.relname
+            and allowlist.column_name = attribute.attname
+       )
+     )
+     or (
+       grantee_role.rolname = 'leoclean_kb_stage_owner'
+       and exists (
+         select 1
+           from pg_temp.kb_apply_prereqs_leoclean_stage_owner_column_allowlist allowlist
+          where allowlist.schema_name = namespace.nspname
+            and allowlist.relation_name = relation.relname
+            and allowlist.column_name = attribute.attname
+            and allowlist.privilege_type = pg_catalog.upper(acl.privilege_type)
+       )
+     )
+   );
+
 -- Protected review/apply roles are standalone principals. The Observatory
 -- service principal is intentionally a member of kb_observatory_read, but the
 -- read role itself must never be a member of another role: that would create an
@@ -67,9 +329,13 @@ begin
     join pg_catalog.pg_roles member_role on member_role.oid = membership.member
     join pg_catalog.pg_roles granted_role on granted_role.oid = membership.roleid
    where member_role.rolname in (
-           'kb_gate_owner', 'kb_review', 'kb_apply', 'kb_observatory_read'
+           'kb_gate_owner', 'kb_review', 'kb_apply', 'kb_observatory_read',
+           'leoclean_kb_runtime', 'leoclean_kb_stage_owner'
          )
-      or granted_role.rolname in ('kb_gate_owner', 'kb_review', 'kb_apply');
+      or granted_role.rolname in (
+           'kb_gate_owner', 'kb_review', 'kb_apply',
+           'leoclean_kb_runtime', 'leoclean_kb_stage_owner'
+         );
 
   if v_memberships is not null then
     raise exception 'kb_apply_prereqs: unexpected protected role membership: %', v_memberships;
@@ -308,17 +574,14 @@ begin
        pg_catalog.pg_get_userbyid(acl.grantee) in (
          'kb_gate_owner', 'kb_review', 'kb_apply'
        )
-       or (
-         pg_catalog.pg_get_userbyid(acl.grantee) = 'kb_observatory_read'
-         and (namespace.nspname, relation.relname) in (
-           ('public', 'claims'),
-           ('public', 'sources'),
-           ('public', 'claim_evidence'),
-           ('public', 'claim_edges'),
-           ('kb_stage', 'kb_proposals')
-         )
-         and pg_catalog.upper(acl.privilege_type) = 'SELECT'
-         and not acl.is_grantable
+       or exists (
+         select 1
+           from pg_temp.kb_apply_prereqs_preserved_table_acls preserved
+          where preserved.schema_name = namespace.nspname
+            and preserved.relation_name = relation.relname
+            and preserved.grantee = pg_catalog.pg_get_userbyid(acl.grantee)
+            and preserved.privilege_type = pg_catalog.upper(acl.privilege_type)
+            and preserved.is_grantable = acl.is_grantable
        )
      );
 
@@ -362,9 +625,21 @@ begin
      and not attribute.attisdropped
      and acl.grantee <> 0
      and acl.grantee <> relation.relowner
-     and pg_catalog.pg_get_userbyid(acl.grantee) not in (
-           'kb_gate_owner', 'kb_review', 'kb_apply'
-         );
+     and not (
+       pg_catalog.pg_get_userbyid(acl.grantee) in (
+         'kb_gate_owner', 'kb_review', 'kb_apply'
+       )
+       or exists (
+         select 1
+           from pg_temp.kb_apply_prereqs_preserved_column_acls preserved
+          where preserved.schema_name = namespace.nspname
+            and preserved.relation_name = relation.relname
+            and preserved.column_name = attribute.attname
+            and preserved.grantee = pg_catalog.pg_get_userbyid(acl.grantee)
+            and preserved.privilege_type = pg_catalog.upper(acl.privilege_type)
+            and preserved.is_grantable = acl.is_grantable
+       )
+     );
 
   if v_drift is not null then
     raise exception 'kb_apply_prereqs: unexpected protected table grantee: %', v_drift;
@@ -1007,9 +1282,13 @@ begin
     join pg_catalog.pg_roles member_role on member_role.oid = membership.member
     join pg_catalog.pg_roles granted_role on granted_role.oid = membership.roleid
    where member_role.rolname in (
-           'kb_gate_owner', 'kb_review', 'kb_apply', 'kb_observatory_read'
+           'kb_gate_owner', 'kb_review', 'kb_apply', 'kb_observatory_read',
+           'leoclean_kb_runtime', 'leoclean_kb_stage_owner'
          )
-      or granted_role.rolname in ('kb_gate_owner', 'kb_review', 'kb_apply');
+      or granted_role.rolname in (
+           'kb_gate_owner', 'kb_review', 'kb_apply',
+           'leoclean_kb_runtime', 'leoclean_kb_stage_owner'
+         );
 
   if v_drift is not null then
     raise exception 'kb_apply_prereqs: protected role membership appeared during migration: %', v_drift;
@@ -1201,34 +1480,62 @@ begin
       ('kb_stage', 'kb_review_principals'),
       ('kb_stage', 'kb_proposal_approvals'),
       ('kb_stage', 'teleo_v3_installation_ownership')
+  ), actual(schema_name, object_name, column_name, grantee, privilege_type, is_grantable) as (
+    select namespace.nspname::text,
+           relation.relname::text,
+           attribute.attname::text,
+           (case
+              when acl.grantee = 0 then 'PUBLIC'
+              else pg_catalog.pg_get_userbyid(acl.grantee)
+            end)::text,
+           pg_catalog.upper(acl.privilege_type),
+           acl.is_grantable
+      from tracked
+      join pg_catalog.pg_namespace namespace on namespace.nspname = tracked.schema_name
+      join pg_catalog.pg_class relation
+        on relation.relnamespace = namespace.oid
+       and relation.relname = tracked.object_name
+      join pg_catalog.pg_attribute attribute on attribute.attrelid = relation.oid
+     cross join lateral pg_catalog.aclexplode(attribute.attacl) acl
+     where attribute.attnum > 0
+       and not attribute.attisdropped
+       and acl.grantee <> relation.relowner
+  ), expected(schema_name, object_name, column_name, grantee, privilege_type, is_grantable) as (
+    select preserved.schema_name::text,
+           preserved.relation_name::text,
+           preserved.column_name::text,
+           preserved.grantee::text,
+           preserved.privilege_type,
+           preserved.is_grantable
+      from pg_temp.kb_apply_prereqs_preserved_column_acls preserved
+  ), delta(kind, schema_name, object_name, column_name, grantee, privilege_type, is_grantable) as (
+    select 'unexpected', actual.*
+      from actual
+     where not exists (
+       select 1 from expected where pg_catalog.to_jsonb(expected) = pg_catalog.to_jsonb(actual)
+     )
+    union all
+    select 'missing', expected.*
+      from expected
+     where not exists (
+       select 1 from actual where pg_catalog.to_jsonb(actual) = pg_catalog.to_jsonb(expected)
+     )
   )
   select pg_catalog.string_agg(
            pg_catalog.format(
-             '%I.%I.%I -> %s %s grantable=%s',
-             namespace.nspname,
-             relation.relname,
-             attribute.attname,
-             case
-               when acl.grantee = 0 then 'PUBLIC'
-               else pg_catalog.pg_get_userbyid(acl.grantee)
-             end,
-             pg_catalog.upper(acl.privilege_type),
-             acl.is_grantable
+             '%s %I.%I.%I -> %s %s grantable=%s',
+             kind,
+             schema_name,
+             object_name,
+             column_name,
+             grantee,
+             privilege_type,
+             is_grantable
            ),
-           ', ' order by namespace.nspname, relation.relname, attribute.attnum, acl.grantee
+           ', ' order by kind, schema_name, object_name, column_name, grantee, privilege_type
          )
     into v_drift
-    from tracked
-    join pg_catalog.pg_namespace namespace on namespace.nspname = tracked.schema_name
-    join pg_catalog.pg_class relation
-      on relation.relnamespace = namespace.oid
-     and relation.relname = tracked.object_name
-    join pg_catalog.pg_attribute attribute on attribute.attrelid = relation.oid
-    cross join lateral pg_catalog.aclexplode(attribute.attacl) acl
-    left join pg_catalog.pg_roles grantee_role on grantee_role.oid = acl.grantee
-   where attribute.attnum > 0
-     and not attribute.attisdropped
-     and acl.grantee <> relation.relowner;
+    from delta;
 
   if v_drift is not null then
     raise exception 'kb_apply_prereqs: protected column ACL mismatch: %', v_drift;
@@ -1283,30 +1590,12 @@ begin
   ), expected(schema_name, object_name, grantee, privilege_type, is_grantable) as (
     select * from core_expected
     union all
-    -- Preserve only the independently installed Observatory read grants. The
-    -- preflight above rejects writes, grant options, column ACLs, and any table
-    -- outside this exact surface before the migration can change state.
-    select namespace.nspname::text,
-           relation.relname::text,
-           grantee_role.rolname::text,
-           pg_catalog.upper(acl.privilege_type),
-           acl.is_grantable
-      from pg_catalog.pg_class relation
-      join pg_catalog.pg_namespace namespace on namespace.oid = relation.relnamespace
-      cross join lateral pg_catalog.aclexplode(
-        coalesce(relation.relacl, pg_catalog.acldefault('r', relation.relowner))
-      ) acl
-      join pg_catalog.pg_roles grantee_role on grantee_role.oid = acl.grantee
-     where grantee_role.rolname = 'kb_observatory_read'
-       and (namespace.nspname, relation.relname) in (
-         ('public', 'claims'),
-         ('public', 'sources'),
-         ('public', 'claim_evidence'),
-         ('public', 'claim_edges'),
-         ('kb_stage', 'kb_proposals')
-       )
-       and pg_catalog.upper(acl.privilege_type) = 'SELECT'
-       and not acl.is_grantable
+    select preserved.schema_name::text,
+           preserved.relation_name::text,
+           preserved.grantee::text,
+           preserved.privilege_type,
+           preserved.is_grantable
+      from pg_temp.kb_apply_prereqs_preserved_table_acls preserved
   ), tracked(schema_name, object_name) as (
     values
       ('public', 'agents'),

tests/test_gcp_leoclean_runtime_role_postgres.py

@@ -18,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"
+PREREQUISITE_SQL = REPO_ROOT / "scripts" / "kb_apply_prereqs.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"
@@ -223,7 +224,11 @@ def _generated_read_tables_sql() -> str:
     for schema, relation, columns in verifier.READ_COLUMN_ALLOWLIST:
         if (schema, relation) in {("public", "agents"), ("kb_stage", "kb_proposals")}:
             continue
-        definitions = [f"{_quote_identifier(column)} text" for column in columns]
+        definitions = [
+            f"{_quote_identifier(column)} "
+            f"{'boolean' if (schema, relation, column) == ('public', 'strategies', 'active') else 'text'}"
+            for column in columns
+        ]
         definitions.append('"runtime_hidden_marker" text')
         statements.append(
             f"create table {_quote_identifier(schema)}.{_quote_identifier(relation)} ({', '.join(definitions)});"
@@ -241,11 +246,14 @@ create schema kb_stage;
 create table public.agents (
   id uuid primary key,
   handle text not null unique,
+  kind text not null default 'service',
   runtime_hidden_marker text
 );
 
 {_generated_read_tables_sql()}
 
+create table public.strategy_nodes (id uuid primary key);
+
 create table kb_stage.kb_proposals (
   id uuid primary key default gen_random_uuid(),
   proposal_type text not null,
@@ -283,6 +291,8 @@ create table public.private_notes (id uuid primary key, body text not null);
 
 insert into public.agents (id, handle, runtime_hidden_marker)
 values ('11111111-1111-4111-8111-111111111111', 'leo', 'not-runtime-readable');
+insert into public.agents (id, handle, kind, runtime_hidden_marker)
+values ('22222222-2222-4222-8222-222222222222', 'm3ta', 'human', null);
 
 create function kb_stage.approve_strict_proposal(uuid, text, jsonb, text, text)
 returns jsonb language sql security definer set search_path = pg_catalog, pg_temp
@@ -334,6 +344,10 @@ def _provision(container: str, remote_sql_path: str) -> subprocess.CompletedProc
     )
 
 
+def _apply_prerequisites(container: str) -> subprocess.CompletedProcess[str]:
+    return _psql(container, PREREQUISITE_SQL.read_text(encoding="utf-8"))
+
+
 def _assert_cloudsql_admin_attribute_compatibility(container: str) -> None:
     probe_role = "cloudsql_admin_probe"
     target_role = "cloudsql_attribute_target"
@@ -1001,6 +1015,7 @@ def test_digest_pinned_network_disabled_postgres_16_14_permission_contract(tmp_p
         assert provider_residual_before[0]["database_acl"] is None
         _require_success(_psql(container, "create database teleo_kb;"), "legacy database fixture")
         _require_success(_psql(container, _bootstrap_sql()), "permission fixture bootstrap")
+        _require_success(_apply_prerequisites(container), "initial KB apply prerequisite installation")
         _assert_cloudsql_admin_attribute_compatibility(container)
         _require_success(
             _run(["docker", "cp", str(ROLE_SQL), f"{container}:/tmp/gcp_leoclean_runtime_role.sql"]),
@@ -1156,6 +1171,56 @@ def test_digest_pinned_network_disabled_postgres_16_14_permission_contract(tmp_p
         )
         assert _read_provisioning_state(container) == stable_state
 
+        for rerun_number in range(1, 3):
+            _require_success(
+                _apply_prerequisites(container),
+                f"post-runtime KB apply prerequisite rerun {rerun_number}",
+            )
+            assert _read_provisioning_state(container) == stable_state
+
+        for grant_sql, revoke_sql, expected_error in (
+            (
+                "grant select (reviewed_by_agent_id) on kb_stage.kb_proposals "
+                "to leoclean_kb_runtime;",
+                "revoke select (reviewed_by_agent_id) on kb_stage.kb_proposals "
+                "from leoclean_kb_runtime;",
+                "unexpected protected table grantee",
+            ),
+            (
+                "grant update on kb_stage.kb_proposals to leoclean_kb_stage_owner;",
+                "revoke update on kb_stage.kb_proposals from leoclean_kb_stage_owner;",
+                "unexpected protected table grantee",
+            ),
+        ):
+            _require_success(_psql(container, grant_sql), "prerequisite drift fixture")
+            refused = _apply_prerequisites(container)
+            assert refused.returncode != 0
+            assert expected_error in refused.stderr
+            _require_success(_psql(container, revoke_sql), "prerequisite drift recovery")
+            assert _read_provisioning_state(container) == stable_state
+
+        _require_success(
+            _psql(container, "create role prereq_membership_probe nologin; "),
+            "prerequisite membership probe role",
+        )
+        _require_success(
+            _psql(container, "grant prereq_membership_probe to leoclean_kb_runtime;"),
+            "prerequisite membership drift fixture",
+        )
+        refused_membership = _apply_prerequisites(container)
+        assert refused_membership.returncode != 0
+        assert "unexpected protected role membership" in refused_membership.stderr
+        _require_success(
+            _psql(
+                container,
+                "revoke prereq_membership_probe from leoclean_kb_runtime; "
+                "drop role prereq_membership_probe;",
+            ),
+            "prerequisite membership drift recovery",
+        )
+        _require_success(_apply_prerequisites(container), "final KB apply prerequisite recovery")
+        assert _read_provisioning_state(container) == stable_state
+
         identity = _require_success(
             _psql(container, "select current_user || '|' || current_database();", role=RUNTIME_ROLE),
             "runtime identity",
@@ -1193,7 +1258,7 @@ def test_digest_pinned_network_disabled_postgres_16_14_permission_contract(tmp_p
 
         before = _read_fingerprint(container)
         assert before == {
-            f"{schema}.{relation}": (1 if (schema, relation) == ("public", "agents") else 0)
+            f"{schema}.{relation}": (3 if (schema, relation) == ("public", "agents") else 0)
             for schema, relation, _columns in verifier.READ_COLUMN_ALLOWLIST
         }
 

tests/test_kb_apply_prereqs.py

@@ -2,6 +2,7 @@ from __future__ import annotations
 
 import json
 import os
+import re
 import shutil
 import subprocess
 import time
@@ -12,6 +13,7 @@ from pathlib import Path
 
 import pytest
 
+from ops import verify_gcp_leoclean_runtime_permissions as runtime_verifier
 from tests.docker_volume_lifecycle import (
     POSTGRES_TMPFS_SPEC,
     acquire_docker_volume_set_guard,
@@ -22,6 +24,7 @@ from tests.docker_volume_lifecycle import (
 
 REPO_ROOT = Path(__file__).resolve().parents[1]
 MIGRATION = REPO_ROOT / "scripts" / "kb_apply_prereqs.sql"
+RUNTIME_ROLE_SQL = REPO_ROOT / "ops" / "gcp_leoclean_runtime_role.sql"
 POSTGRES_IMAGE = os.environ.get("LEOCLEAN_RUNTIME_POSTGRES_IMAGE", "postgres:16-alpine")
 DATABASE = "teleo_clone"
 LEGACY_APPROVED_ID = "aaaaaaaa-0000-4000-8000-000000000001"
@@ -75,6 +78,11 @@ create table kb_stage.kb_proposals (
   id uuid primary key,
   proposal_type text not null,
   status text not null,
+  proposed_by_handle text,
+  proposed_by_agent_id uuid,
+  channel text not null default 'cli',
+  source_ref text,
+  rationale text not null default '',
   payload jsonb not null,
   reviewed_by_handle text,
   reviewed_by_agent_id uuid,
@@ -83,6 +91,7 @@ create table kb_stage.kb_proposals (
   applied_by_handle text,
   applied_by_agent_id uuid,
   applied_at timestamptz,
+  created_at timestamptz not null default now(),
   updated_at timestamptz not null default now()
 );
 create table kb_stage.kb_review_principals (
@@ -269,6 +278,58 @@ def _lines(completed: subprocess.CompletedProcess[str]) -> set[str]:
     return {line for line in completed.stdout.splitlines() if line}
 
 
+def _read_allowlist_rows(sql: str, anchor: str) -> set[tuple[str, str, str, int]]:
+    manifest = sql.split(anchor, 1)[1].split(";", 1)[0]
+    return {
+        (schema, relation, column, int(ordinal))
+        for schema, relation, column, ordinal in re.findall(
+            r"\('([^']+)', '([^']+)', '([^']+)', ([0-9]+)\)", manifest
+        )
+    }
+
+
+def test_scoped_runtime_acl_allowlists_match_the_executable_runtime_contract() -> None:
+    migration_sql = MIGRATION.read_text(encoding="utf-8")
+    runtime_sql = RUNTIME_ROLE_SQL.read_text(encoding="utf-8")
+    runtime_rows = _read_allowlist_rows(
+        runtime_sql,
+        "insert into pg_temp.leoclean_runtime_read_column_allowlist (",
+    )
+    prerequisite_rows = _read_allowlist_rows(
+        migration_sql,
+        "insert into pg_temp.kb_apply_prereqs_leoclean_read_column_allowlist (",
+    )
+    executable_rows = {
+        (schema, relation, column, ordinal)
+        for schema, relation, columns in runtime_verifier.READ_COLUMN_ALLOWLIST
+        for ordinal, column in enumerate(columns, start=1)
+    }
+
+    assert len(runtime_rows) == 92
+    assert prerequisite_rows == runtime_rows == executable_rows
+
+    stage_manifest = migration_sql.split(
+        "insert into pg_temp.kb_apply_prereqs_leoclean_stage_owner_column_allowlist (",
+        1,
+    )[1].split(";", 1)[0]
+    stage_rows = set(
+        re.findall(
+            r"\('([^']+)', '([^']+)', '([^']+)', '([^']+)'\)",
+            stage_manifest,
+        )
+    )
+    expected_stage_rows = {
+        ("public", "agents", "id", "SELECT"),
+        ("public", "agents", "handle", "SELECT"),
+        *{
+            ("kb_stage", "kb_proposals", column, "INSERT")
+            for column in runtime_verifier.STAGE_OWNER_INSERT_COLUMNS
+        },
+    }
+    assert stage_rows == expected_stage_rows
+    assert "stage_leoclean_proposal" not in migration_sql
+
+
 def test_legacy_only_reviewer_is_canonicalized_without_rewriting_history(
     migrated_postgres: str,
 ) -> None:
@@ -1051,6 +1112,140 @@ grant select on public.sources, kb_stage.kb_proposals to kb_observatory_read;
         _apply_migration(container)
 
 
+def _scoped_runtime_acl_snapshot(container: str) -> str:
+    return _psql(
+        container,
+        r"""
+select 'role|' || rolname || '|' || rolcanlogin || '|' || rolinherit || '|' || rolsuper
+  from pg_catalog.pg_roles
+ where rolname in ('leoclean_kb_runtime', 'leoclean_kb_stage_owner')
+ order by rolname;
+select 'membership|' || member_role.rolname || '|' || granted_role.rolname
+  from pg_catalog.pg_auth_members membership
+  join pg_catalog.pg_roles member_role on member_role.oid = membership.member
+  join pg_catalog.pg_roles granted_role on granted_role.oid = membership.roleid
+ where member_role.rolname in ('leoclean_kb_runtime', 'leoclean_kb_stage_owner')
+    or granted_role.rolname in ('leoclean_kb_runtime', 'leoclean_kb_stage_owner')
+ order by member_role.rolname, granted_role.rolname;
+select 'table|' || namespace.nspname || '|' || relation.relname || '|' ||
+       grantee_role.rolname || '|' || pg_catalog.upper(acl.privilege_type) || '|' ||
+       acl.is_grantable
+  from pg_catalog.pg_class relation
+  join pg_catalog.pg_namespace namespace on namespace.oid = relation.relnamespace
+ cross join lateral pg_catalog.aclexplode(
+   coalesce(relation.relacl, pg_catalog.acldefault('r', relation.relowner))
+ ) acl
+  join pg_catalog.pg_roles grantee_role on grantee_role.oid = acl.grantee
+ where grantee_role.rolname in ('leoclean_kb_runtime', 'leoclean_kb_stage_owner')
+ order by namespace.nspname, relation.relname, grantee_role.rolname, acl.privilege_type;
+select 'column|' || namespace.nspname || '|' || relation.relname || '|' ||
+       attribute.attname || '|' || grantee_role.rolname || '|' ||
+       pg_catalog.upper(acl.privilege_type) || '|' || acl.is_grantable
+  from pg_catalog.pg_attribute attribute
+  join pg_catalog.pg_class relation on relation.oid = attribute.attrelid
+  join pg_catalog.pg_namespace namespace on namespace.oid = relation.relnamespace
+ cross join lateral pg_catalog.aclexplode(attribute.attacl) acl
+  join pg_catalog.pg_roles grantee_role on grantee_role.oid = acl.grantee
+ where grantee_role.rolname in ('leoclean_kb_runtime', 'leoclean_kb_stage_owner')
+ order by namespace.nspname, relation.relname, attribute.attnum,
+          grantee_role.rolname, acl.privilege_type;
+""",
+    ).stdout
+
+
+def test_migration_composes_with_scoped_leoclean_acls_and_rejects_drift() -> None:
+    with _postgres_clone() as container:
+        _apply_migration(container)
+        _psql(
+            container,
+            """
+create role leoclean_kb_runtime
+  nologin noinherit nosuperuser nocreatedb nocreaterole noreplication nobypassrls;
+create role leoclean_kb_stage_owner
+  nologin noinherit nosuperuser nocreatedb nocreaterole noreplication nobypassrls;
+
+grant usage on schema public, kb_stage to leoclean_kb_runtime, leoclean_kb_stage_owner;
+grant select (id, handle) on public.agents to leoclean_kb_runtime, leoclean_kb_stage_owner;
+grant select (agent_id, active) on public.strategies to leoclean_kb_runtime;
+grant select (claim_id, source_id, role) on public.claim_evidence to leoclean_kb_runtime;
+grant select (id) on public.claim_edges, public.claims to leoclean_kb_runtime;
+grant select (id, hash) on public.sources to leoclean_kb_runtime;
+grant select (
+  id, proposal_type, status, proposed_by_handle, proposed_by_agent_id, channel,
+  source_ref, rationale, payload, reviewed_by_handle, reviewed_at, review_note,
+  applied_by_handle, applied_at, created_at, updated_at
+) on kb_stage.kb_proposals to leoclean_kb_runtime;
+
+grant select on kb_stage.kb_proposals to leoclean_kb_stage_owner;
+grant insert (
+  proposal_type, status, proposed_by_handle, proposed_by_agent_id, channel,
+  source_ref, rationale, payload
+) on kb_stage.kb_proposals to leoclean_kb_stage_owner;
+""",
+        )
+        before = _scoped_runtime_acl_snapshot(container)
+
+        _apply_migration(container)
+        _apply_migration(container)
+        assert _scoped_runtime_acl_snapshot(container) == before
+
+        allowed_reads = _psql(
+            container,
+            """
+set session authorization leoclean_kb_runtime;
+select id, handle from public.agents limit 0;
+select id, proposal_type, status, payload from kb_stage.kb_proposals limit 0;
+""",
+        )
+        assert allowed_reads.returncode == 0
+
+        for denied_sql in (
+            "set session authorization leoclean_kb_runtime; "
+            "insert into public.claims (id) values ('99999999-9999-4999-8999-999999999998');",
+            "set session authorization leoclean_kb_runtime; "
+            "insert into kb_stage.kb_proposals (id, proposal_type, status, rationale, payload) "
+            "values ('99999999-9999-4999-8999-999999999997', 'add_edge', "
+            "'pending_review', 'direct write', '{}'::jsonb);",
+            "set session authorization leoclean_kb_runtime; set role kb_apply;",
+        ):
+            denied = _psql(container, denied_sql, check=False)
+            assert denied.returncode != 0
+            assert "permission denied" in denied.stderr
+
+        for grant_sql, revoke_sql, expected_error in (
+            (
+                "grant select (reviewed_by_agent_id) on kb_stage.kb_proposals "
+                "to leoclean_kb_runtime;",
+                "revoke select (reviewed_by_agent_id) on kb_stage.kb_proposals "
+                "from leoclean_kb_runtime;",
+                "unexpected protected table grantee",
+            ),
+            (
+                "grant select (id) on public.claims to leoclean_kb_runtime with grant option;",
+                "revoke grant option for select (id) on public.claims from leoclean_kb_runtime;",
+                "unexpected protected table grantee",
+            ),
+            (
+                "grant update on kb_stage.kb_proposals to leoclean_kb_stage_owner;",
+                "revoke update on kb_stage.kb_proposals from leoclean_kb_stage_owner;",
+                "unexpected protected table grantee",
+            ),
+            (
+                "grant broad_writer to leoclean_kb_runtime;",
+                "revoke broad_writer from leoclean_kb_runtime;",
+                "unexpected protected role membership",
+            ),
+        ):
+            _psql(container, grant_sql)
+            try:
+                _assert_migration_fails(container, expected_error)
+            finally:
+                _psql(container, revoke_sql)
+
+        _apply_migration(container)
+        assert _scoped_runtime_acl_snapshot(container) == before
+
+
 def test_migration_refuses_rogue_canonical_insert_before_install_and_normalizes_after_removal() -> None:
     with _postgres_clone() as container:
         _psql(container, "grant usage on schema public to rogue_grantee;")