\set ON_ERROR_STOP on

-- One-time GCP staging provisioning for Leo's Cloud SQL identity. Run this
-- only through ops/provision_gcp_leoclean_runtime_role.sh; psql's \password
-- command encrypts the supplied value client-side, so cleartext is never SQL.
--
-- The login can read the canonical allowlist and can stage a pending proposal
-- through one narrowly-scoped SECURITY DEFINER function. It cannot inherit or
-- SET ROLE into another principal, forge the proposer identity, write directly
-- to canonical/staging tables, or inspect the legacy teleo_restore audit schema.
-- Keep the runtime unable to authenticate until both database phases have
-- completed and verified. Any failure before the final transaction therefore
-- leaves a disabled role instead of a partially provisioned login.
select 'create role leoclean_kb_runtime nologin nosuperuser nocreatedb nocreaterole noinherit noreplication nobypassrls connection limit 8'
where not exists (select 1 from pg_catalog.pg_roles where rolname = 'leoclean_kb_runtime')
\gexec

select 'create role leoclean_kb_stage_owner nologin nosuperuser nocreatedb nocreaterole noinherit noreplication nobypassrls connection limit -1'
where not exists (select 1 from pg_catalog.pg_roles where rolname = 'leoclean_kb_stage_owner')
\gexec

alter role leoclean_kb_runtime
  with nologin nosuperuser nocreatedb nocreaterole noinherit noreplication nobypassrls connection limit 8;
alter role leoclean_kb_stage_owner
  with nologin nosuperuser nocreatedb nocreaterole noinherit noreplication nobypassrls connection limit -1
  password null;

-- Snapshot every non-superuser principal connected to either scoped role before
-- removing the direct membership edges. Existing sessions can retain an
-- already-selected SET ROLE identity after NOLOGIN/revocation, so all other
-- connected sessions must be terminated before password or privilege work.
create temporary table leoclean_fenced_roles (
  role_oid oid primary key,
  role_name name unique not null
) on commit preserve rows;

with recursive connected(role_oid) as (
  values
    ('leoclean_kb_runtime'::pg_catalog.regrole::oid),
    ('leoclean_kb_stage_owner'::pg_catalog.regrole::oid)
  union
  select membership.member
    from connected
    join pg_catalog.pg_auth_members membership
      on membership.roleid = connected.role_oid
)
insert into pg_temp.leoclean_fenced_roles (role_oid, role_name)
select role_row.oid, role_row.rolname
  from connected
  join pg_catalog.pg_roles role_row on role_row.oid = connected.role_oid
 where not role_row.rolsuper;

-- NOINHERIT does not prevent SET ROLE. Remove every direct membership edge in
-- either direction before terminating the snapshotted sessions, closing the
-- race for newly connected membership principals.
select format('revoke %I from %I', granted_role.rolname, member_role.rolname)
  from pg_catalog.pg_auth_members membership
  join pg_catalog.pg_roles granted_role on granted_role.oid = membership.roleid
  join pg_catalog.pg_roles member_role on member_role.oid = membership.member
 where granted_role.rolname in ('leoclean_kb_runtime', 'leoclean_kb_stage_owner')
    or member_role.rolname in ('leoclean_kb_runtime', 'leoclean_kb_stage_owner')
\gexec

do $session_fence$
declare
  target record;
  remaining text;
begin
  for target in
    select activity.pid
      from pg_catalog.pg_stat_activity activity
      join pg_temp.leoclean_fenced_roles fenced on fenced.role_name = activity.usename
     where activity.pid <> pg_catalog.pg_backend_pid()
       and activity.backend_type = 'client backend'
  loop
    if not pg_catalog.pg_terminate_backend(target.pid, 5000) then
      raise exception 'could not terminate a pre-existing scoped-role session';
    end if;
  end loop;

  select pg_catalog.string_agg(activity.pid::text, ',' order by activity.pid)
    into remaining
    from pg_catalog.pg_stat_activity activity
    join pg_temp.leoclean_fenced_roles fenced on fenced.role_name = activity.usename
   where activity.pid <> pg_catalog.pg_backend_pid()
     and activity.backend_type = 'client backend';
  if remaining is not null then
    raise exception 'pre-existing scoped-role sessions remain after termination';
  end if;

  select pg_catalog.string_agg(prepared.gid, ',' order by prepared.gid)
    into remaining
    from pg_catalog.pg_prepared_xacts prepared
    join pg_temp.leoclean_fenced_roles fenced on fenced.role_name = prepared.owner;
  if remaining is not null then
    raise exception 'scoped-role prepared transactions must be resolved before provisioning';
  end if;
end
$session_fence$;

-- This is deliberately after the autocommitted NOLOGIN/session/membership
-- fence. Interruption can leave a disabled role, never a still-enabled old
-- login or an old connected session. The wrapper supplies the two prompts.
\password leoclean_kb_runtime

alter role leoclean_kb_runtime reset all;
alter role leoclean_kb_stage_owner reset all;
drop table pg_temp.leoclean_fenced_roles;

select pg_catalog.format(
         'alter role %I in database %I reset all',
         role_row.rolname,
         database_row.datname
       )
  from pg_catalog.pg_db_role_setting setting_row
  join pg_catalog.pg_roles role_row on role_row.oid = setting_row.setrole
  join pg_catalog.pg_database database_row on database_row.oid = setting_row.setdatabase
 where role_row.rolname in ('leoclean_kb_runtime', 'leoclean_kb_stage_owner')
\gexec

-- PostgreSQL normally grants TEMP to PUBLIC. A role-specific REVOKE cannot
-- override that inherited grant, while REVOKE TEMP FROM PUBLIC would be a
-- database-wide change that could break kb_apply/review or operator workflows.
-- Leave PUBLIC TEMP policy unchanged here and surface it in the final receipt.
-- TEMP cannot shadow this SECURITY DEFINER boundary: pg_catalog is first,
-- pg_temp is explicitly last, and every relation reference is schema-qualified.
-- Database CREATE, direct TEMP grants, and all unexpected direct database ACLs
-- are rejected below. A database-wide TEMP removal requires a separate role-use
-- inventory and explicit grants to every workload that still needs it.

alter role leoclean_kb_runtime in database teleo_canonical reset all;
alter role leoclean_kb_stage_owner in database teleo_canonical reset all;
alter role leoclean_kb_runtime in database teleo_kb reset all;
alter role leoclean_kb_stage_owner in database teleo_kb reset all;

alter role leoclean_kb_runtime in database teleo_canonical
  set search_path = pg_catalog, public, kb_stage;
alter role leoclean_kb_runtime in database teleo_canonical
  set statement_timeout = '15s';
alter role leoclean_kb_runtime in database teleo_canonical
  set lock_timeout = '2s';

\connect teleo_canonical

begin;

-- CONNECT granted to PUBLIC cannot be denied to one role with a role-specific
-- REVOKE. Preserve every existing non-scoped principal's effective CONNECT
-- set as explicit ACLs, remove the cluster-wide PUBLIC grant from every
-- database (including templates), and then grant the runtime its sole target.
-- This runs in the same transaction as every remaining privilege/topology
-- assertion, so a later failure restores the original ACLs atomically.
create temporary table leoclean_preserved_database_connect (
  role_name name not null,
  database_name name not null,
  primary key (role_name, database_name)
) on commit drop;

insert into pg_temp.leoclean_preserved_database_connect (role_name, database_name)
select role_row.rolname, database_row.datname
  from pg_catalog.pg_roles role_row
 cross join pg_catalog.pg_database database_row
 where role_row.rolname not in ('leoclean_kb_runtime', 'leoclean_kb_stage_owner')
   and pg_catalog.has_database_privilege(role_row.oid, database_row.oid, 'CONNECT');

select pg_catalog.format('revoke connect on database %I from public', database_row.datname)
  from pg_catalog.pg_database database_row
\gexec

select pg_catalog.format(
         'revoke all privileges on database %I from leoclean_kb_runtime, leoclean_kb_stage_owner',
         database_row.datname
       )
  from pg_catalog.pg_database database_row
\gexec

select pg_catalog.format(
         'grant connect on database %I to %I',
         preserved.database_name,
         preserved.role_name
       )
  from pg_temp.leoclean_preserved_database_connect preserved
 order by preserved.database_name, preserved.role_name
\gexec

grant connect on database teleo_canonical to leoclean_kb_runtime;

-- The deploy principal needs temporary ownership authority to replace an
-- existing function on idempotent reruns. This membership is transaction-local
-- and is revoked before verification/commit.
select format('grant leoclean_kb_stage_owner to %I', current_user)
 where current_user <> 'leoclean_kb_stage_owner'
\gexec

revoke all on schema public, kb_stage
  from leoclean_kb_runtime, leoclean_kb_stage_owner;
revoke all privileges on all tables in schema public, kb_stage
  from leoclean_kb_runtime, leoclean_kb_stage_owner;
revoke all privileges on all sequences in schema public, kb_stage
  from leoclean_kb_runtime, leoclean_kb_stage_owner;
revoke all privileges on all functions in schema public, kb_stage
  from leoclean_kb_runtime, leoclean_kb_stage_owner;
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,
-- 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.
create temporary table leoclean_preserved_app_routine_execute (
  role_name name not null,
  function_oid oid not null,
  primary key (role_name, function_oid)
) on commit drop;

insert into pg_temp.leoclean_preserved_app_routine_execute (role_name, function_oid)
select role_row.rolname, function_row.oid
  from pg_catalog.pg_roles role_row
 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 namespace.nspname in ('public', 'kb_stage')
   and pg_catalog.has_function_privilege(role_row.oid, function_row.oid, 'EXECUTE');

select pg_catalog.format(
         'revoke execute on %s %I.%I(%s) from public, leoclean_kb_runtime, leoclean_kb_stage_owner',
         case when function_row.prokind = 'p' then 'procedure' else 'function' end,
         namespace.nspname,
         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 in ('public', 'kb_stage')
\gexec

select pg_catalog.format(
         'grant execute on %s %I.%I(%s) to %I',
         case when function_row.prokind = 'p' then 'procedure' else 'function' end,
         namespace.nspname,
         function_row.proname,
         pg_catalog.pg_get_function_identity_arguments(function_row.oid),
         preserved.role_name
       )
  from pg_temp.leoclean_preserved_app_routine_execute preserved
  join pg_catalog.pg_proc function_row on function_row.oid = preserved.function_oid
  join pg_catalog.pg_namespace namespace on namespace.oid = function_row.pronamespace
 order by namespace.nspname, function_row.proname, preserved.role_name
\gexec

select pg_catalog.format(
         'revoke all privileges on large object %s from leoclean_kb_runtime, leoclean_kb_stage_owner',
         metadata.oid
       )
  from pg_catalog.pg_largeobject_metadata metadata
\gexec

-- Default PostgreSQL grants EXECUTE on built-in large-object mutators to
-- PUBLIC, which would let the runtime persist data despite all relation ACLs.
-- Snapshot and explicitly preserve every existing non-scoped role's effective
-- access, then remove PUBLIC and both scoped roles from every mutator overload.
create temporary table leoclean_preserved_lo_execute (
  role_name name not null,
  function_oid oid not null,
  primary key (role_name, function_oid)
) on commit drop;

insert into pg_temp.leoclean_preserved_lo_execute (role_name, function_oid)
select role_row.rolname, function_row.oid
  from pg_catalog.pg_roles role_row
 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 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(role_row.oid, function_row.oid, 'EXECUTE');

select pg_catalog.format(
         'revoke execute on function pg_catalog.%I(%s) from public, leoclean_kb_runtime, leoclean_kb_stage_owner',
         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'
   )
\gexec

select pg_catalog.format(
         'grant execute on function pg_catalog.%I(%s) to %I',
         function_row.proname,
         pg_catalog.pg_get_function_identity_arguments(function_row.oid),
         preserved.role_name
       )
  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
\gexec

select pg_catalog.format(
         'revoke all privileges on parameter %I from leoclean_kb_runtime, leoclean_kb_stage_owner',
         parameter_acl.parname
       )
  from pg_catalog.pg_parameter_acl parameter_acl
\gexec

-- Table-level REVOKE does not clear column ACLs. Clear every column ACL for both
-- principals before rebuilding the exact allowlist below.
select format(
         'revoke all privileges (%s) on table %I.%I from %I',
         pg_catalog.string_agg(pg_catalog.quote_ident(attribute.attname), ', ' order by attribute.attnum),
         namespace.nspname,
         relation.relname,
         target.rolname
       )
  from pg_catalog.pg_class relation
  join pg_catalog.pg_namespace namespace on namespace.oid = relation.relnamespace
  join pg_catalog.pg_attribute attribute on attribute.attrelid = relation.oid
 cross join (
   values ('leoclean_kb_runtime'), ('leoclean_kb_stage_owner')
 ) as target(rolname)
 where namespace.nspname in ('public', 'kb_stage')
   and relation.relkind in ('r', 'p', 'v', 'm', 'f')
   and attribute.attnum > 0
   and not attribute.attisdropped
 group by namespace.nspname, relation.relname, target.rolname
\gexec

grant usage on schema public, kb_stage to leoclean_kb_runtime;

grant select on
  public.agents,
  public.claims,
  public.claim_evidence,
  public.claim_edges,
  public.sources,
  public.personas,
  public.strategies,
  public.beliefs,
  public.blindspots,
  public.agent_roles,
  public.peer_models,
  public.behavioral_rules,
  public.contributor_rules,
  public.reasoning_tools,
  public.governance_gates,
  kb_stage.kb_proposals
to leoclean_kb_runtime;

-- The function owner is a dedicated NOLOGIN role. Grant only the columns the
-- function reads/inserts. CREATE is temporary and is revoked immediately after
-- ownership is assigned.
grant usage, create on schema kb_stage to leoclean_kb_stage_owner;
grant usage on schema public to leoclean_kb_stage_owner;
grant select (id, handle) on public.agents to leoclean_kb_stage_owner;
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;

-- Remove the prior caller-supplied-identity overload before installing the
-- session-bound form. A runtime caller can only stage as canonical agent Leo.
drop function if exists kb_stage.stage_leoclean_proposal(text, text, text, text, text, jsonb);

create or replace function kb_stage.stage_leoclean_proposal(
  p_proposal_type text,
  p_channel text,
  p_source_ref text,
  p_rationale text,
  p_payload jsonb
) returns jsonb
language plpgsql
security definer
set search_path = pg_catalog, pg_temp
as $function$
declare
  staged kb_stage.kb_proposals%rowtype;
  proposer_id uuid;
begin
  if session_user <> 'leoclean_kb_runtime' then
    raise exception 'stage_leoclean_proposal is restricted to leoclean_kb_runtime'
      using errcode = '42501';
  end if;
  if p_proposal_type not in ('revise_claim', 'revise_strategy', 'add_edge') then
    raise exception 'unsupported Leo proposal type: %', p_proposal_type using errcode = '22023';
  end if;
  if nullif(btrim(p_source_ref), '') is null then
    raise exception 'source_ref is required' using errcode = '22023';
  end if;
  if nullif(btrim(p_rationale), '') is null then
    raise exception 'rationale is required' using errcode = '22023';
  end if;
  if p_payload is null or jsonb_typeof(p_payload) <> 'object' then
    raise exception 'proposal payload must be a JSON object' using errcode = '22023';
  end if;

  select agent.id
    into proposer_id
    from public.agents as agent
   where agent.handle = 'leo';

  if proposer_id is null then
    raise exception 'canonical Leo agent row is required' using errcode = '23503';
  end if;

  insert into kb_stage.kb_proposals (
    proposal_type,
    status,
    proposed_by_handle,
    proposed_by_agent_id,
    channel,
    source_ref,
    rationale,
    payload
  ) values (
    p_proposal_type,
    'pending_review',
    'leo',
    proposer_id,
    coalesce(nullif(p_channel, ''), 'telegram'),
    p_source_ref,
    p_rationale,
    p_payload
  )
  returning * into staged;

  return to_jsonb(staged);
end
$function$;

alter function kb_stage.stage_leoclean_proposal(text, text, text, text, jsonb)
  owner to leoclean_kb_stage_owner;
revoke create on schema kb_stage from leoclean_kb_stage_owner;

-- Normalize explicit ACL entries left by CREATE OR REPLACE or an earlier grant.
revoke all on function kb_stage.stage_leoclean_proposal(text, text, text, text, jsonb)
  from public;
select format(
         'revoke all on function kb_stage.stage_leoclean_proposal(text, text, text, text, jsonb) from %I',
         grantee.rolname
       )
  from pg_catalog.pg_proc function_row
 cross join lateral pg_catalog.aclexplode(
   coalesce(
     function_row.proacl,
     pg_catalog.acldefault('f', function_row.proowner)
   )
 ) as acl
  join pg_catalog.pg_roles grantee on grantee.oid = acl.grantee
 where function_row.oid = 'kb_stage.stage_leoclean_proposal(text,text,text,text,jsonb)'::pg_catalog.regprocedure
   and grantee.rolname <> 'leoclean_kb_stage_owner'
\gexec

grant execute on function kb_stage.stage_leoclean_proposal(text, text, text, text, jsonb)
  to leoclean_kb_runtime;

select format('revoke leoclean_kb_stage_owner from %I', current_user)
 where current_user <> 'leoclean_kb_stage_owner'
\gexec

do $verification$
declare
  unexpected text;
  runtime_oid oid := 'leoclean_kb_runtime'::pg_catalog.regrole::oid;
  owner_oid oid := 'leoclean_kb_stage_owner'::pg_catalog.regrole::oid;
  stage_function oid := 'kb_stage.stage_leoclean_proposal(text,text,text,text,jsonb)'::pg_catalog.regprocedure::oid;
begin
  if exists (
    select 1
      from pg_catalog.pg_roles role_row
     where role_row.rolname = 'leoclean_kb_runtime'
       and (role_row.rolcanlogin
            or role_row.rolsuper
            or role_row.rolcreatedb
            or role_row.rolcreaterole
            or role_row.rolinherit
            or role_row.rolreplication
            or role_row.rolbypassrls
            or role_row.rolconnlimit <> 8)
  ) then
    raise exception 'leoclean_kb_runtime has unsafe provisioning role attributes';
  end if;

  if exists (
    select 1
      from pg_catalog.pg_roles role_row
     where role_row.rolname = 'leoclean_kb_stage_owner'
       and (role_row.rolcanlogin
            or role_row.rolsuper
            or role_row.rolcreatedb
            or role_row.rolcreaterole
            or role_row.rolinherit
            or role_row.rolreplication
            or role_row.rolbypassrls
            or role_row.rolconnlimit <> -1)
  ) then
    raise exception 'leoclean_kb_stage_owner has unsafe role attributes';
  end if;

  select pg_catalog.string_agg(
           pg_catalog.format('%I->%I', granted_role.rolname, member_role.rolname),
           ', '
           order by granted_role.rolname, member_role.rolname
         )
    into unexpected
    from pg_catalog.pg_auth_members membership
    join pg_catalog.pg_roles granted_role on granted_role.oid = membership.roleid
    join pg_catalog.pg_roles member_role on member_role.oid = membership.member
   where membership.roleid in (runtime_oid, owner_oid)
      or membership.member in (runtime_oid, owner_oid);
  if unexpected is not null then
    raise exception 'scoped Leo roles retain role memberships: %', unexpected;
  end if;

  if not exists (
    select 1
      from pg_catalog.pg_class relation
      join pg_catalog.pg_namespace namespace on namespace.oid = relation.relnamespace
     where namespace.nspname = 'kb_stage'
       and relation.relname = 'kb_proposals'
       and relation.relkind = 'r'
       and relation.relpersistence = 'p'
       and not relation.relrowsecurity
       and not relation.relforcerowsecurity
  ) then
    raise exception 'kb_stage.kb_proposals is not the reviewed permanent base table';
  end if;

  select pg_catalog.string_agg(
           pg_catalog.format('%I.%I:%s', namespace.nspname, relation.relname, relation.relkind),
           ', '
           order by namespace.nspname, relation.relname
         )
    into unexpected
    from pg_catalog.pg_class relation
    join pg_catalog.pg_namespace namespace on namespace.oid = relation.relnamespace
   where (namespace.nspname, relation.relname) in (
     ('public', 'agents'), ('public', 'claims'), ('public', 'claim_evidence'),
     ('public', 'claim_edges'), ('public', 'sources'), ('public', 'personas'),
     ('public', 'strategies'), ('public', 'beliefs'), ('public', 'blindspots'),
     ('public', 'agent_roles'), ('public', 'peer_models'), ('public', 'behavioral_rules'),
     ('public', 'contributor_rules'), ('public', 'reasoning_tools'),
     ('public', 'governance_gates'), ('kb_stage', 'kb_proposals')
   )
     and relation.relkind <> 'r';
  if unexpected is not null then
    raise exception 'Leo read allowlist contains a non-base relation: %', unexpected;
  end if;

  select pg_catalog.string_agg(
           pg_catalog.format('%s->%s', inheritance.inhparent::pg_catalog.regclass, inheritance.inhrelid::pg_catalog.regclass),
           ', '
         )
    into unexpected
    from pg_catalog.pg_inherits inheritance
   where inheritance.inhparent in (
           'public.agents'::pg_catalog.regclass, 'public.claims'::pg_catalog.regclass,
           'public.claim_evidence'::pg_catalog.regclass, 'public.claim_edges'::pg_catalog.regclass,
           'public.sources'::pg_catalog.regclass, 'public.personas'::pg_catalog.regclass,
           'public.strategies'::pg_catalog.regclass, 'public.beliefs'::pg_catalog.regclass,
           'public.blindspots'::pg_catalog.regclass, 'public.agent_roles'::pg_catalog.regclass,
           'public.peer_models'::pg_catalog.regclass, 'public.behavioral_rules'::pg_catalog.regclass,
           'public.contributor_rules'::pg_catalog.regclass, 'public.reasoning_tools'::pg_catalog.regclass,
           'public.governance_gates'::pg_catalog.regclass, 'kb_stage.kb_proposals'::pg_catalog.regclass
         )
      or inheritance.inhrelid in (
           'public.agents'::pg_catalog.regclass, 'public.claims'::pg_catalog.regclass,
           'public.claim_evidence'::pg_catalog.regclass, 'public.claim_edges'::pg_catalog.regclass,
           'public.sources'::pg_catalog.regclass, 'public.personas'::pg_catalog.regclass,
           'public.strategies'::pg_catalog.regclass, 'public.beliefs'::pg_catalog.regclass,
           'public.blindspots'::pg_catalog.regclass, 'public.agent_roles'::pg_catalog.regclass,
           'public.peer_models'::pg_catalog.regclass, 'public.behavioral_rules'::pg_catalog.regclass,
           'public.contributor_rules'::pg_catalog.regclass, 'public.reasoning_tools'::pg_catalog.regclass,
           'public.governance_gates'::pg_catalog.regclass, 'kb_stage.kb_proposals'::pg_catalog.regclass
         );
  if unexpected is not null then
    raise exception 'Leo read allowlist has inheritance edges: %', unexpected;
  end if;

  with expected(ordinal, attname, typname, not_null, has_default, collated) as (
    values
      (1, 'id', 'uuid', true, true, false),
      (2, 'proposal_type', 'text', true, false, true),
      (3, 'status', 'text', true, true, true),
      (4, 'proposed_by_handle', 'text', false, false, true),
      (5, 'proposed_by_agent_id', 'uuid', false, false, false),
      (6, 'channel', 'text', true, true, true),
      (7, 'source_ref', 'text', false, false, true),
      (8, 'rationale', 'text', true, false, true),
      (9, 'payload', 'jsonb', true, false, false),
      (10, 'reviewed_by_handle', 'text', false, false, true),
      (11, 'reviewed_by_agent_id', 'uuid', false, false, false),
      (12, 'reviewed_at', 'timestamptz', false, false, false),
      (13, 'review_note', 'text', false, false, true),
      (14, 'applied_by_handle', 'text', false, false, true),
      (15, 'applied_by_agent_id', 'uuid', false, false, false),
      (16, 'applied_at', 'timestamptz', false, false, false),
      (17, 'created_at', 'timestamptz', true, true, false),
      (18, 'updated_at', 'timestamptz', true, true, false)
  ), actual as (
    select attribute.attnum::int as ordinal,
           attribute.attname,
           type_row.typname,
           attribute.attnotnull as not_null,
           attribute.atthasdef as has_default,
           attribute.attidentity,
           attribute.attgenerated,
           attribute.atttypmod,
           type_namespace.nspname as type_namespace,
           attribute.attcollation = 'pg_catalog.default'::pg_catalog.regcollation as collated
      from pg_catalog.pg_attribute attribute
      join pg_catalog.pg_type type_row on type_row.oid = attribute.atttypid
      join pg_catalog.pg_namespace type_namespace on type_namespace.oid = type_row.typnamespace
     where attribute.attrelid = 'kb_stage.kb_proposals'::pg_catalog.regclass
       and attribute.attnum > 0
       and not attribute.attisdropped
  )
  select pg_catalog.string_agg(coalesce(expected.attname, actual.attname), ', ' order by coalesce(expected.ordinal, actual.ordinal))
    into unexpected
    from expected
    full join actual using (ordinal, attname)
   where expected.ordinal is null
      or actual.ordinal is null
      or expected.typname is distinct from actual.typname
      or expected.not_null is distinct from actual.not_null
      or expected.has_default is distinct from actual.has_default
      or expected.collated is distinct from actual.collated
      or actual.attidentity <> ''
      or actual.attgenerated <> ''
      or actual.atttypmod <> -1
      or actual.type_namespace <> 'pg_catalog';
  if unexpected is not null then
    raise exception 'kb_stage.kb_proposals column contract drifted: %', unexpected;
  end if;

  if exists (
    select 1
      from pg_catalog.pg_trigger trigger_row
     where trigger_row.tgrelid = 'kb_stage.kb_proposals'::pg_catalog.regclass
       and not trigger_row.tgisinternal
  ) or exists (
    select 1
      from pg_catalog.pg_rewrite rewrite_row
     where rewrite_row.ev_class = 'kb_stage.kb_proposals'::pg_catalog.regclass
  ) or exists (
    select 1
      from pg_catalog.pg_policy policy_row
     where policy_row.polrelid = 'kb_stage.kb_proposals'::pg_catalog.regclass
  ) or exists (
    select 1
      from pg_catalog.pg_attribute attribute
     where attribute.attrelid = 'kb_stage.kb_proposals'::pg_catalog.regclass
       and attribute.attnum > 0
       and not attribute.attisdropped
       and attribute.attgenerated <> ''
  ) then
    raise exception 'kb_stage.kb_proposals has unsafe trigger/rule/RLS/generated topology';
  end if;

  with expected(attname, expression) as (
    values
      ('id', 'gen_random_uuid()'),
      ('status', '''pending_review''::text'),
      ('channel', '''cli''::text'),
      ('created_at', 'now()'),
      ('updated_at', 'now()')
  ), actual as (
    select attribute.attname,
           pg_catalog.pg_get_expr(default_row.adbin, default_row.adrelid) as expression
      from pg_catalog.pg_attribute attribute
      join pg_catalog.pg_attrdef default_row
        on default_row.adrelid = attribute.attrelid
       and default_row.adnum = attribute.attnum
     where attribute.attrelid = 'kb_stage.kb_proposals'::pg_catalog.regclass
       and attribute.attnum > 0
       and not attribute.attisdropped
  )
  select pg_catalog.string_agg(coalesce(expected.attname, actual.attname), ', ')
    into unexpected
    from expected
    full join actual using (attname)
   where expected.expression is distinct from actual.expression;
  if unexpected is not null then
    raise exception 'kb_stage.kb_proposals default topology drifted: %', unexpected;
  end if;

  select pg_catalog.string_agg(
           pg_catalog.format('%s:%s', dependency.refclassid::pg_catalog.regclass, dependency.refobjid),
           ', '
         )
    into unexpected
    from pg_catalog.pg_attrdef default_row
    join pg_catalog.pg_depend dependency
      on dependency.classid = 'pg_catalog.pg_attrdef'::pg_catalog.regclass
     and dependency.objid = default_row.oid
   where default_row.adrelid = 'kb_stage.kb_proposals'::pg_catalog.regclass
     and (
       dependency.refclassid = 'pg_catalog.pg_proc'::pg_catalog.regclass
       and dependency.refobjid not in (
         'pg_catalog.gen_random_uuid()'::pg_catalog.regprocedure,
         'pg_catalog.now()'::pg_catalog.regprocedure
       )
       or dependency.refclassid = 'pg_catalog.pg_type'::pg_catalog.regclass
          and not exists (
            select 1
              from pg_catalog.pg_type type_row
              join pg_catalog.pg_namespace namespace on namespace.oid = type_row.typnamespace
             where type_row.oid = dependency.refobjid
               and namespace.nspname = 'pg_catalog'
          )
       or dependency.refclassid = 'pg_catalog.pg_operator'::pg_catalog.regclass
          and not exists (
            select 1
              from pg_catalog.pg_operator operator_row
              join pg_catalog.pg_namespace namespace on namespace.oid = operator_row.oprnamespace
             where operator_row.oid = dependency.refobjid
               and namespace.nspname = 'pg_catalog'
          )
       or dependency.refclassid = 'pg_catalog.pg_collation'::pg_catalog.regclass
          and not exists (
            select 1
              from pg_catalog.pg_collation collation_row
              join pg_catalog.pg_namespace namespace on namespace.oid = collation_row.collnamespace
             where collation_row.oid = dependency.refobjid
               and namespace.nspname = 'pg_catalog'
          )
     );
  if unexpected is not null then
    raise exception 'kb_stage.kb_proposals defaults depend on unreviewed objects: %', unexpected;
  end if;

  if (
    select pg_catalog.count(*) <> 6
        or pg_catalog.count(*) filter (
          where (constraint_row.conname, constraint_row.contype) in (
            ('kb_proposals_pkey', 'p'),
            ('kb_proposals_proposal_type_check', 'c'),
            ('kb_proposals_status_check', 'c'),
            ('kb_proposals_applied_by_agent_id_fkey', 'f'),
            ('kb_proposals_proposed_by_agent_id_fkey', 'f'),
            ('kb_proposals_reviewed_by_agent_id_fkey', 'f')
          )
        ) <> 6
      from pg_catalog.pg_constraint constraint_row
     where constraint_row.conrelid = 'kb_stage.kb_proposals'::pg_catalog.regclass
  ) or exists (
    select 1
      from pg_catalog.pg_constraint constraint_row
      join pg_catalog.pg_depend dependency
        on dependency.classid = 'pg_catalog.pg_constraint'::pg_catalog.regclass
       and dependency.objid = constraint_row.oid
       and dependency.refclassid = 'pg_catalog.pg_proc'::pg_catalog.regclass
      join pg_catalog.pg_proc function_row on function_row.oid = dependency.refobjid
      join pg_catalog.pg_namespace namespace on namespace.oid = function_row.pronamespace
     where constraint_row.conrelid = 'kb_stage.kb_proposals'::pg_catalog.regclass
       and namespace.nspname <> 'pg_catalog'
  ) or exists (
    select 1
      from pg_catalog.pg_index index_row
     where index_row.indrelid = 'kb_stage.kb_proposals'::pg_catalog.regclass
       and (
         index_row.indexprs is not null
         or (
           index_row.indpred is not null
           and pg_catalog.pg_get_expr(index_row.indpred, index_row.indrelid)
               <> '(proposed_by_handle IS NOT NULL)'
         )
       )
  ) then
    raise exception 'kb_stage.kb_proposals constraint/index topology drifted';
  end if;

  select pg_catalog.string_agg(
           pg_catalog.format(
             '%I:%s:%s:grantable=%s',
             database_row.datname,
             grantee.rolname,
             acl.privilege_type,
             acl.is_grantable
           ),
           ', '
           order by database_row.datname, grantee.rolname, acl.privilege_type
         )
    into unexpected
    from pg_catalog.pg_database database_row
   cross join lateral pg_catalog.aclexplode(
     coalesce(
       database_row.datacl,
       pg_catalog.acldefault('d', database_row.datdba)
     )
   ) as acl
    join pg_catalog.pg_roles grantee on grantee.oid = acl.grantee
   where grantee.rolname in ('leoclean_kb_runtime', 'leoclean_kb_stage_owner')
     and not (
       database_row.datname = 'teleo_canonical'
       and grantee.rolname = 'leoclean_kb_runtime'
       and acl.privilege_type = 'CONNECT'
       and not acl.is_grantable
     );
  if unexpected is not null then
    raise exception 'scoped Leo roles retain unexpected direct database ACLs: %', unexpected;
  end if;

  select pg_catalog.string_agg(database_row.datname, ', ' order by database_row.datname)
    into unexpected
    from pg_catalog.pg_database database_row
   where database_row.datname <> 'teleo_canonical'
     and has_database_privilege('leoclean_kb_runtime', database_row.oid, 'CONNECT');
  if unexpected is not null then
    raise exception 'runtime CONNECT reaches a noncanonical database: %', unexpected;
  end if;

  select pg_catalog.string_agg(database_row.datname, ', ' order by database_row.datname)
    into unexpected
    from pg_catalog.pg_database database_row
   where has_database_privilege('leoclean_kb_stage_owner', database_row.oid, 'CONNECT');
  if unexpected is not null then
    raise exception 'stage owner CONNECT reaches a database: %', unexpected;
  end if;

  select pg_catalog.string_agg(database_row.datname, ', ' order by database_row.datname)
    into unexpected
    from pg_catalog.pg_database database_row
   cross join lateral pg_catalog.aclexplode(
     coalesce(database_row.datacl, pg_catalog.acldefault('d', database_row.datdba))
   ) acl
   where acl.grantee = 0
     and acl.privilege_type = 'CONNECT';
  if unexpected is not null then
    raise exception 'PUBLIC retains database CONNECT: %', unexpected;
  end if;

  select pg_catalog.string_agg(
           pg_catalog.format('%s:%s', setting_row.setdatabase, setting_row.setconfig),
           ', '
         )
    into unexpected
    from pg_catalog.pg_db_role_setting setting_row
   where setting_row.setrole in (runtime_oid, owner_oid)
     and not (
       setting_row.setrole = runtime_oid
       and setting_row.setdatabase = (
         select database_row.oid
           from pg_catalog.pg_database database_row
          where database_row.datname = 'teleo_canonical'
       )
       and pg_catalog.cardinality(setting_row.setconfig) = 3
       and setting_row.setconfig @> array[
         'search_path=pg_catalog, public, kb_stage',
         'statement_timeout=15s',
         'lock_timeout=2s'
       ]::text[]
     );
  if unexpected is not null or not exists (
    select 1
      from pg_catalog.pg_db_role_setting setting_row
     where setting_row.setrole = runtime_oid
       and setting_row.setdatabase = (
         select database_row.oid
           from pg_catalog.pg_database database_row
          where database_row.datname = 'teleo_canonical'
       )
       and pg_catalog.cardinality(setting_row.setconfig) = 3
       and setting_row.setconfig @> array[
         'search_path=pg_catalog, public, kb_stage',
         'statement_timeout=15s',
         'lock_timeout=2s'
       ]::text[]
  ) then
    raise exception 'scoped Leo role settings differ from the reviewed contract';
  end if;

  select pg_catalog.string_agg(
           pg_catalog.format('%I:%s', database_row.datname, scoped_role.name),
           ', '
           order by database_row.datname, scoped_role.name
         )
    into unexpected
    from pg_catalog.pg_database database_row
   cross join (values ('leoclean_kb_runtime'), ('leoclean_kb_stage_owner')) scoped_role(name)
   where has_database_privilege(scoped_role.name, database_row.oid, 'CREATE');
  if unexpected is not null then
    raise exception 'a scoped Leo role unexpectedly has database CREATE: %', unexpected;
  end if;

  select pg_catalog.string_agg(
           pg_catalog.format('%s:%s:%s', dependency.classid::pg_catalog.regclass, dependency.objid, dependency.objsubid),
           ', '
         )
    into unexpected
    from pg_catalog.pg_shdepend dependency
   where dependency.refclassid = 'pg_catalog.pg_authid'::pg_catalog.regclass
     and dependency.refobjid = runtime_oid
     and dependency.deptype = 'o';
  if unexpected is not null then
    raise exception 'leoclean_kb_runtime unexpectedly owns database objects: %', unexpected;
  end if;

  select pg_catalog.string_agg(
           pg_catalog.format('%s:%s:%s', dependency.classid::pg_catalog.regclass, dependency.objid, dependency.objsubid),
           ', '
         )
    into unexpected
    from pg_catalog.pg_shdepend dependency
   where dependency.refclassid = 'pg_catalog.pg_authid'::pg_catalog.regclass
     and dependency.refobjid = owner_oid
     and dependency.deptype = 'o'
     and not (
       dependency.dbid = (select database_row.oid from pg_catalog.pg_database database_row where database_row.datname = current_database())
       and
       dependency.classid = 'pg_catalog.pg_proc'::pg_catalog.regclass
       and dependency.objid = stage_function
       and dependency.objsubid = 0
     );
  if unexpected is not null then
    raise exception 'leoclean_kb_stage_owner owns unexpected database objects: %', unexpected;
  end if;

  select pg_catalog.string_agg(metadata.oid::text, ', ' order by metadata.oid)
    into unexpected
    from pg_catalog.pg_largeobject_metadata metadata
   where metadata.lomowner in (runtime_oid, owner_oid)
      or exists (
        select 1
          from pg_catalog.aclexplode(
            coalesce(metadata.lomacl, pg_catalog.acldefault('L', metadata.lomowner))
          ) acl
         where acl.grantee in (0, runtime_oid, owner_oid)
           and acl.privilege_type in ('SELECT', 'UPDATE')
      );
  if unexpected is not null then
    raise exception 'scoped Leo roles retain large-object ownership or ACL reachability: %', 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)),
           ', '
         )
    into unexpected
    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'
     )
     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;
  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)),
           ', '
           order by function_row.proname, pg_catalog.pg_get_function_identity_arguments(function_row.oid)
         )
    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(
     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';
  if unexpected is not null then
    raise exception 'PUBLIC retains large-object mutator EXECUTE: %', unexpected;
  end if;

  select pg_catalog.string_agg(parameter_acl.parname, ', ' order by parameter_acl.parname)
    into unexpected
    from pg_catalog.pg_parameter_acl parameter_acl
   where has_parameter_privilege('leoclean_kb_runtime', parameter_acl.parname, 'SET')
      or has_parameter_privilege('leoclean_kb_runtime', parameter_acl.parname, 'ALTER SYSTEM')
      or has_parameter_privilege('leoclean_kb_stage_owner', parameter_acl.parname, 'SET')
      or has_parameter_privilege('leoclean_kb_stage_owner', parameter_acl.parname, 'ALTER SYSTEM');
  if unexpected is not null then
    raise exception 'scoped Leo roles retain parameter privileges: %', unexpected;
  end if;

  if has_schema_privilege('leoclean_kb_runtime', 'public', 'CREATE')
     or has_schema_privilege('leoclean_kb_runtime', 'kb_stage', 'CREATE')
     or has_schema_privilege('leoclean_kb_stage_owner', 'public', 'CREATE')
     or has_schema_privilege('leoclean_kb_stage_owner', 'kb_stage', 'CREATE') then
    raise exception 'a scoped Leo role unexpectedly has schema CREATE';
  end if;

  select pg_catalog.string_agg(
           pg_catalog.format('%I.%I:%s', namespace.nspname, relation.relname, privilege.name),
           ', '
           order by namespace.nspname, relation.relname, privilege.name
         )
    into unexpected
    from pg_catalog.pg_class relation
    join pg_catalog.pg_namespace namespace on namespace.oid = relation.relnamespace
   cross join (values ('INSERT'), ('UPDATE'), ('DELETE'), ('TRUNCATE'), ('REFERENCES'), ('TRIGGER')) privilege(name)
   where namespace.nspname in ('public', 'kb_stage')
     and relation.relkind in ('r', 'p', 'v', 'm', 'f')
     and has_table_privilege('leoclean_kb_runtime', relation.oid, privilege.name);
  if unexpected is not null then
    raise exception 'leoclean_kb_runtime unexpectedly has table-level DML: %', unexpected;
  end if;

  select pg_catalog.string_agg(
           pg_catalog.format('%I.%I.%I:%s', namespace.nspname, relation.relname, attribute.attname, privilege.name),
           ', '
           order by namespace.nspname, relation.relname, attribute.attnum, privilege.name
         )
    into unexpected
    from pg_catalog.pg_class relation
    join pg_catalog.pg_namespace namespace on namespace.oid = relation.relnamespace
    join pg_catalog.pg_attribute attribute on attribute.attrelid = relation.oid
   cross join (values ('INSERT'), ('UPDATE'), ('REFERENCES')) privilege(name)
   where namespace.nspname in ('public', 'kb_stage')
     and relation.relkind in ('r', 'p', 'v', 'm', 'f')
     and attribute.attnum > 0
     and not attribute.attisdropped
     and has_column_privilege('leoclean_kb_runtime', relation.oid, attribute.attnum, privilege.name);
  if unexpected is not null then
    raise exception 'leoclean_kb_runtime unexpectedly has column-level DML: %', unexpected;
  end if;

  select pg_catalog.string_agg(
           pg_catalog.format('%I.%I', namespace.nspname, relation.relname),
           ', '
           order by namespace.nspname, relation.relname
         )
    into unexpected
    from pg_catalog.pg_class relation
    join pg_catalog.pg_namespace namespace on namespace.oid = relation.relnamespace
   where namespace.nspname in ('public', 'kb_stage')
     and relation.relkind in ('r', 'p', 'v', 'm', 'f')
     and has_table_privilege('leoclean_kb_runtime', relation.oid, 'SELECT')
     and (namespace.nspname, relation.relname) not in (
       ('public', 'agents'),
       ('public', 'claims'),
       ('public', 'claim_evidence'),
       ('public', 'claim_edges'),
       ('public', 'sources'),
       ('public', 'personas'),
       ('public', 'strategies'),
       ('public', 'beliefs'),
       ('public', 'blindspots'),
       ('public', 'agent_roles'),
       ('public', 'peer_models'),
       ('public', 'behavioral_rules'),
       ('public', 'contributor_rules'),
       ('public', 'reasoning_tools'),
       ('public', 'governance_gates'),
       ('kb_stage', 'kb_proposals')
     );
  if unexpected is not null then
    raise exception 'leoclean_kb_runtime can SELECT outside its table allowlist: %', unexpected;
  end if;

  select pg_catalog.string_agg(
           pg_catalog.format('%I.%I.%I', namespace.nspname, relation.relname, attribute.attname),
           ', '
           order by namespace.nspname, relation.relname, attribute.attnum
         )
    into unexpected
    from pg_catalog.pg_class relation
    join pg_catalog.pg_namespace namespace on namespace.oid = relation.relnamespace
    join pg_catalog.pg_attribute attribute on attribute.attrelid = relation.oid
   where namespace.nspname in ('public', 'kb_stage')
     and relation.relkind in ('r', 'p', 'v', 'm', 'f')
     and attribute.attnum > 0
     and not attribute.attisdropped
     and has_column_privilege('leoclean_kb_runtime', relation.oid, attribute.attnum, 'SELECT')
     and (namespace.nspname, relation.relname) not in (
       ('public', 'agents'),
       ('public', 'claims'),
       ('public', 'claim_evidence'),
       ('public', 'claim_edges'),
       ('public', 'sources'),
       ('public', 'personas'),
       ('public', 'strategies'),
       ('public', 'beliefs'),
       ('public', 'blindspots'),
       ('public', 'agent_roles'),
       ('public', 'peer_models'),
       ('public', 'behavioral_rules'),
       ('public', 'contributor_rules'),
       ('public', 'reasoning_tools'),
       ('public', 'governance_gates'),
       ('kb_stage', 'kb_proposals')
     );
  if unexpected is not null then
    raise exception 'leoclean_kb_runtime has column SELECT outside its allowlist: %', unexpected;
  end if;

  if has_table_privilege('leoclean_kb_stage_owner', 'public.agents', 'SELECT')
     or not has_column_privilege('leoclean_kb_stage_owner', 'public.agents', 'id', 'SELECT')
     or not has_column_privilege('leoclean_kb_stage_owner', 'public.agents', 'handle', 'SELECT')
     or not has_table_privilege('leoclean_kb_stage_owner', 'kb_stage.kb_proposals', 'SELECT')
     or has_table_privilege('leoclean_kb_stage_owner', 'kb_stage.kb_proposals', 'INSERT') then
    raise exception 'leoclean_kb_stage_owner does not have the exact read/insert grant shape';
  end if;

  select pg_catalog.string_agg(
           pg_catalog.format('%I.%I.%I', namespace.nspname, relation.relname, attribute.attname),
           ', '
           order by namespace.nspname, relation.relname, attribute.attnum
         )
    into unexpected
    from pg_catalog.pg_class relation
    join pg_catalog.pg_namespace namespace on namespace.oid = relation.relnamespace
    join pg_catalog.pg_attribute attribute on attribute.attrelid = relation.oid
   where namespace.nspname in ('public', 'kb_stage')
     and relation.relkind in ('r', 'p', 'v', 'm', 'f')
     and attribute.attnum > 0
     and not attribute.attisdropped
     and has_column_privilege('leoclean_kb_stage_owner', relation.oid, attribute.attnum, 'INSERT')
     and not (
       namespace.nspname = 'kb_stage'
       and relation.relname = 'kb_proposals'
       and attribute.attname in (
         'proposal_type',
         'status',
         'proposed_by_handle',
         'proposed_by_agent_id',
         'channel',
         'source_ref',
         'rationale',
         'payload'
       )
     );
  if unexpected is not null then
    raise exception 'leoclean_kb_stage_owner can INSERT unexpected columns: %', unexpected;
  end if;

  select pg_catalog.string_agg(
           pg_catalog.format('%I.%I:%s', namespace.nspname, relation.relname, privilege.name),
           ', '
           order by namespace.nspname, relation.relname, privilege.name
         )
    into unexpected
    from pg_catalog.pg_class relation
    join pg_catalog.pg_namespace namespace on namespace.oid = relation.relnamespace
   cross join (values ('INSERT'), ('UPDATE'), ('DELETE'), ('TRUNCATE'), ('REFERENCES'), ('TRIGGER')) privilege(name)
   where namespace.nspname in ('public', 'kb_stage')
     and relation.relkind in ('r', 'p', 'v', 'm', 'f')
     and has_table_privilege('leoclean_kb_stage_owner', relation.oid, privilege.name);
  if unexpected is not null then
    raise exception 'leoclean_kb_stage_owner unexpectedly has table-level DML: %', unexpected;
  end if;

  select pg_catalog.string_agg(
           pg_catalog.format('%I.%I.%I:%s', namespace.nspname, relation.relname, attribute.attname, privilege.name),
           ', '
           order by namespace.nspname, relation.relname, attribute.attnum, privilege.name
         )
    into unexpected
    from pg_catalog.pg_class relation
    join pg_catalog.pg_namespace namespace on namespace.oid = relation.relnamespace
    join pg_catalog.pg_attribute attribute on attribute.attrelid = relation.oid
   cross join (values ('UPDATE'), ('REFERENCES')) privilege(name)
   where namespace.nspname in ('public', 'kb_stage')
     and relation.relkind in ('r', 'p', 'v', 'm', 'f')
     and attribute.attnum > 0
     and not attribute.attisdropped
     and has_column_privilege('leoclean_kb_stage_owner', relation.oid, attribute.attnum, privilege.name);
  if unexpected is not null then
    raise exception 'leoclean_kb_stage_owner unexpectedly has column-level DML: %', unexpected;
  end if;

  select pg_catalog.string_agg(
           pg_catalog.format('%I.%I.%I', namespace.nspname, relation.relname, attribute.attname),
           ', '
           order by namespace.nspname, relation.relname, attribute.attnum
         )
    into unexpected
    from pg_catalog.pg_class relation
    join pg_catalog.pg_namespace namespace on namespace.oid = relation.relnamespace
    join pg_catalog.pg_attribute attribute on attribute.attrelid = relation.oid
   where namespace.nspname in ('public', 'kb_stage')
     and relation.relkind in ('r', 'p', 'v', 'm', 'f')
     and attribute.attnum > 0
     and not attribute.attisdropped
     and has_column_privilege('leoclean_kb_stage_owner', relation.oid, attribute.attnum, 'SELECT')
     and not (
       (namespace.nspname = 'kb_stage' and relation.relname = 'kb_proposals')
       or (
         namespace.nspname = 'public'
         and relation.relname = 'agents'
         and attribute.attname in ('id', 'handle')
       )
     );
  if unexpected is not null then
    raise exception 'leoclean_kb_stage_owner can SELECT unexpected columns: %', unexpected;
  end if;

  select pg_catalog.string_agg(
           pg_catalog.format('%I.%I:%s:%s', namespace.nspname, relation.relname, scoped_role.name, privilege.name),
           ', '
           order by namespace.nspname, relation.relname, scoped_role.name, privilege.name
         )
    into unexpected
    from pg_catalog.pg_class relation
    join pg_catalog.pg_namespace namespace on namespace.oid = relation.relnamespace
   cross join (values ('leoclean_kb_runtime'), ('leoclean_kb_stage_owner')) scoped_role(name)
   cross join (values ('USAGE'), ('SELECT'), ('UPDATE')) privilege(name)
   where namespace.nspname in ('public', 'kb_stage')
     and relation.relkind = 'S'
     and has_sequence_privilege(scoped_role.name, relation.oid, privilege.name);
  if unexpected is not null then
    raise exception 'a scoped Leo role unexpectedly has sequence privileges: %', unexpected;
  end if;

  if exists (
    select 1
      from pg_catalog.pg_attribute attribute
     where attribute.attrelid = 'kb_stage.kb_proposals'::pg_catalog.regclass
       and attribute.attname in (
         'proposal_type',
         'status',
         'proposed_by_handle',
         'proposed_by_agent_id',
         'channel',
         'source_ref',
         'rationale',
         'payload'
       )
       and not has_column_privilege(
         'leoclean_kb_stage_owner',
         attribute.attrelid,
         attribute.attnum,
         'INSERT'
       )
  ) then
    raise exception 'leoclean_kb_stage_owner is missing a required INSERT column grant';
  end if;

  select pg_catalog.string_agg(
           pg_catalog.format('%I.%I(%s)', namespace.nspname, function_row.proname, pg_catalog.pg_get_function_identity_arguments(function_row.oid)),
           ', '
         )
    into unexpected
    from pg_catalog.pg_proc function_row
    join pg_catalog.pg_namespace namespace on namespace.oid = function_row.pronamespace
   where namespace.nspname in ('public', 'kb_stage')
     and function_row.prosecdef
     and function_row.oid <> stage_function
     and has_function_privilege('leoclean_kb_runtime', function_row.oid, 'EXECUTE');
  if unexpected is not null then
    raise exception 'leoclean_kb_runtime can execute unexpected SECURITY DEFINER functions: %', unexpected;
  end if;

  if not exists (
    select 1
      from pg_catalog.pg_proc function_row
     where function_row.oid = stage_function
       and function_row.prosecdef
       and function_row.proowner = owner_oid
       and function_row.proconfig = array['search_path=pg_catalog, pg_temp']::text[]
  ) then
    raise exception 'stage_leoclean_proposal owner/security/search_path contract is invalid';
  end if;

  select pg_catalog.string_agg(
           pg_catalog.format(
             'grantee=%s privilege=%s grantable=%s',
             acl.grantee,
             acl.privilege_type,
             acl.is_grantable
           ),
           ', '
         )
    into unexpected
    from pg_catalog.pg_proc function_row
   cross join lateral pg_catalog.aclexplode(
     coalesce(
       function_row.proacl,
       pg_catalog.acldefault('f', function_row.proowner)
     )
   ) as acl
   where function_row.oid = stage_function
     and not (
       acl.grantee in (runtime_oid, owner_oid)
       and acl.privilege_type = 'EXECUTE'
       and (acl.grantee = owner_oid or not acl.is_grantable)
     );
  if unexpected is not null then
    raise exception 'stage_leoclean_proposal has unexpected ACL entries: %', unexpected;
  end if;

  if not has_function_privilege(
    'leoclean_kb_runtime',
    'kb_stage.stage_leoclean_proposal(text,text,text,text,jsonb)',
    'EXECUTE'
  ) then
    raise exception 'leoclean_kb_runtime cannot execute stage_leoclean_proposal';
  end if;
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
-- back the complete privilege migration and leaves the role NOLOGIN.
alter role leoclean_kb_runtime
  with login nosuperuser nocreatedb nocreaterole noinherit noreplication nobypassrls connection limit 8;

do $login_verification$
begin
  if exists (
    select 1
      from pg_catalog.pg_roles role_row
     where role_row.rolname = 'leoclean_kb_runtime'
       and (
         not role_row.rolcanlogin
         or role_row.rolsuper
         or role_row.rolcreatedb
         or role_row.rolcreaterole
         or role_row.rolinherit
         or role_row.rolreplication
         or role_row.rolbypassrls
         or role_row.rolconnlimit <> 8
       )
  ) then
    raise exception 'leoclean_kb_runtime final login attributes are unsafe';
  end if;
end
$login_verification$;

commit;