-- Guarded KB review/apply prerequisites. Run as postgres at deploy time.
--
-- Authority is deliberately split:
--   * kb_review can read proposals and execute one constrained approval function;
--   * kb_apply can read approved proposals, perform narrow canonical writes, and
--     execute payload-bound assert/finish functions;
--   * kb_gate_owner is a NOLOGIN owner for gate tables/functions;
--   * neither login role can directly UPDATE kb_stage.kb_proposals or mutate the
--     immutable approval snapshot.
--
-- The deployer provisions passwords separately in kb-review.env/kb-apply.env.
-- This file contains no credentials and is safe to reuse in disposable clones.
-- Before the first V3 contract migration, run this file, stop the V2 worker,
-- then ALTER ROLE kb_apply NOLOGIN and end its sessions. After V3 is installed,
-- this file may re-enable the login: a V2-configured worker detects the cutover
-- and leaves historical V2 approvals review-visible until migrate/reject.

set timezone = 'UTC';

begin;

do $deployer$
begin
  if not exists (
    select 1
      from pg_catalog.pg_roles
     where rolname = current_user
       and rolsuper
  ) then
    raise exception 'kb_apply_prereqs: current role % must be a superuser', current_user;
  end if;
  if not exists (select 1 from pg_catalog.pg_roles where rolname = 'kb_apply') then
    raise exception 'kb_apply_prereqs: required runtime role kb_apply does not exist';
  end if;
end
$deployer$;

do $roles$
begin
  if not exists (select 1 from pg_catalog.pg_roles where rolname = 'kb_gate_owner') then
    create role kb_gate_owner nologin;
  end if;
  if not exists (select 1 from pg_catalog.pg_roles where rolname = 'kb_review') then
    create role kb_review login;
  end if;
end
$roles$;

alter role kb_gate_owner nologin noinherit nosuperuser nocreatedb nocreaterole noreplication nobypassrls;
alter role kb_review login noinherit nosuperuser nocreatedb nocreaterole noreplication nobypassrls;
alter role kb_apply login noinherit nosuperuser nocreatedb nocreaterole noreplication nobypassrls;

-- 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
-- inherited or SET ROLE path beyond its bounded SELECT surface.
do $memberships$
declare
  v_memberships text;
begin
  select pg_catalog.string_agg(
           pg_catalog.format('%I -> %I', member_role.rolname, granted_role.rolname),
           ', ' order by member_role.rolname, granted_role.rolname
         )
    into v_memberships
    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 (
           'kb_gate_owner', 'kb_review', 'kb_apply', 'kb_observatory_read'
         )
      or granted_role.rolname in ('kb_gate_owner', 'kb_review', 'kb_apply');

  if v_memberships is not null then
    raise exception 'kb_apply_prereqs: unexpected protected role membership: %', v_memberships;
  end if;
end
$memberships$;

-- Service identity used for the applied_by_agent_id foreign key. kb_apply can
-- read this row but can never create or modify agent rows.
insert into public.agents (id, handle, kind)
values ('44444444-4444-4444-4444-444444444444', 'kb-apply', 'service')
on conflict (handle) do nothing;

-- Canonicalize the existing human reviewer identity without changing its UUID.
-- Historical proposal/approval reviewer text is deliberately left untouched.
do $reviewer_identity$
declare
  v_legacy_id uuid;
  v_canonical_id uuid;
begin
  select id into v_legacy_id
    from public.agents
   where handle = 'm3ta'
   for update;

  select id into v_canonical_id
    from public.agents
   where handle = 'm3taversal'
   for update;

  if v_legacy_id is not null
     and v_canonical_id is not null
     and v_legacy_id <> v_canonical_id then
    raise exception
      'kb_apply_prereqs: both reviewer handles m3ta (%) and m3taversal (%) exist with different agent UUIDs; resolve the identity conflict before rerunning',
      v_legacy_id, v_canonical_id;
  end if;

  if v_canonical_id is null and v_legacy_id is not null then
    update public.agents
       set handle = 'm3taversal'
     where id = v_legacy_id
       and handle = 'm3ta';
    if not found then
      raise exception
        'kb_apply_prereqs: legacy reviewer m3ta changed during canonical identity migration';
    end if;
    v_canonical_id := v_legacy_id;
  end if;

  if v_canonical_id is null then
    raise exception
      'kb_apply_prereqs: required human reviewer agent is missing; restore exactly one public.agents row with handle m3ta or m3taversal before rerunning';
  end if;
end
$reviewer_identity$;

-- DB-login to human-reviewer mapping. Only the gate owner/admin can change it;
-- the approval function verifies --reviewed-by against this mapping rather than
-- trusting caller-supplied identity. The dedicated reviewer credential always
-- maps to the canonical m3taversal human agent.
create table if not exists kb_stage.kb_review_principals (
  db_role name primary key,
  reviewed_by_handle text not null,
  reviewed_by_agent_id uuid not null references public.agents(id),
  active boolean not null default true,
  created_at timestamptz not null default now()
);

-- One immutable snapshot per reviewed proposal. Apply is bound to this row as
-- well as the current proposal ledger, so an approved payload cannot drift.
create table if not exists kb_stage.kb_proposal_approvals (
  proposal_id uuid primary key references kb_stage.kb_proposals(id),
  proposal_type text not null,
  payload jsonb not null,
  reviewed_by_handle text not null,
  reviewed_by_agent_id uuid not null references public.agents(id),
  reviewed_by_db_role name not null,
  reviewed_at timestamptz not null,
  review_note text not null,
  approved_proposal_snapshot jsonb
);

-- Existing canonical/proposal tables must remain owned by the deployer. Gate
-- tables may be owned by the deployer during a legacy upgrade or kb_gate_owner
-- after an earlier guarded run. Anything else is ownership drift.
do $gate_preflight$
declare
  v_drift text;
begin
  select pg_catalog.string_agg(
           pg_catalog.format('%I.%I owned by %I', namespace.nspname, relation.relname, owner_role.rolname),
           ', ' order by namespace.nspname, relation.relname
         )
    into v_drift
    from pg_catalog.pg_class relation
    join pg_catalog.pg_namespace namespace on namespace.oid = relation.relnamespace
    join pg_catalog.pg_roles owner_role on owner_role.oid = relation.relowner
   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 relation.relkind in ('r', 'p')
     and not (
       (namespace.nspname = 'kb_stage'
        and relation.relname in (
          'kb_review_principals', 'kb_proposal_approvals', 'teleo_v3_installation_ownership'
        )
        and owner_role.rolname in (current_user, 'kb_gate_owner'))
       or (not (
             namespace.nspname = 'kb_stage'
             and relation.relname in (
               'kb_review_principals', 'kb_proposal_approvals', 'teleo_v3_installation_ownership'
             )
           )
           and owner_role.rolname = current_user)
     );

  if v_drift is not null then
    raise exception 'kb_apply_prereqs: unexpected protected table owner: %', v_drift;
  end if;

  select pg_catalog.string_agg(
           pg_catalog.format(
             '%I.%I(%s) owned by %I',
             namespace.nspname,
             procedure.proname,
             pg_catalog.oidvectortypes(procedure.proargtypes),
             owner_role.rolname
           ),
           ', ' order by procedure.proname, pg_catalog.oidvectortypes(procedure.proargtypes)
         )
    into v_drift
    from pg_catalog.pg_proc procedure
    join pg_catalog.pg_namespace namespace on namespace.oid = procedure.pronamespace
    join pg_catalog.pg_roles owner_role on owner_role.oid = procedure.proowner
   where namespace.nspname = 'kb_stage'
     and procedure.proname in (
           'approve_strict_proposal',
           'assert_approved_proposal',
           'finish_approved_proposal',
           'export_applied_proposal_replay_rows'
         )
     and owner_role.rolname not in (current_user, 'kb_gate_owner');

  if v_drift is not null then
    raise exception 'kb_apply_prereqs: unexpected protected function owner: %', v_drift;
  end if;

  select pg_catalog.string_agg(
           pg_catalog.format(
             '%I.%I(%s)',
             namespace.nspname,
             procedure.proname,
             pg_catalog.oidvectortypes(procedure.proargtypes)
           ),
           ', ' order by procedure.proname, pg_catalog.oidvectortypes(procedure.proargtypes)
         )
    into v_drift
    from pg_catalog.pg_proc procedure
    join pg_catalog.pg_namespace namespace on namespace.oid = procedure.pronamespace
   where namespace.nspname = 'kb_stage'
     and procedure.proname in (
           'approve_strict_proposal',
           'assert_approved_proposal',
           'finish_approved_proposal',
           'export_applied_proposal_replay_rows'
         )
     and not (
       (procedure.proname = 'approve_strict_proposal'
        and pg_catalog.oidvectortypes(procedure.proargtypes) in (
          'uuid, text, text',
          'uuid, text, jsonb, text, text'
        ))
       or (procedure.proname = 'assert_approved_proposal'
           and pg_catalog.oidvectortypes(procedure.proargtypes) =
             'uuid, text, jsonb, text, uuid, timestamp with time zone, text')
       or (procedure.proname = 'finish_approved_proposal'
           and pg_catalog.oidvectortypes(procedure.proargtypes) =
             'uuid, text, jsonb, text, uuid, timestamp with time zone, text, text')
       or (procedure.proname = 'export_applied_proposal_replay_rows'
           and pg_catalog.oidvectortypes(procedure.proargtypes) = 'uuid')
     );

  if v_drift is not null then
    raise exception 'kb_apply_prereqs: unexpected protected function overload: %', v_drift;
  end if;

  select pg_catalog.string_agg(
           pg_catalog.format(
             '%I.%I -> %s',
             namespace.nspname,
             relation.relname,
             case
               when acl.grantee = 0 then 'PUBLIC'
               else pg_catalog.pg_get_userbyid(acl.grantee)
             end
           ),
           ', ' order by namespace.nspname, relation.relname, acl.grantee
         )
    into v_drift
    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
   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 acl.grantee <> 0
     and acl.grantee <> relation.relowner
     and not (
       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
       )
     );

  if v_drift is not null then
    raise exception 'kb_apply_prereqs: unexpected protected table grantee: %', v_drift;
  end if;

  select pg_catalog.string_agg(
           pg_catalog.format(
             '%I.%I.%I -> %s',
             namespace.nspname,
             relation.relname,
             attribute.attname,
             case
               when acl.grantee = 0 then 'PUBLIC'
               else pg_catalog.pg_get_userbyid(acl.grantee)
             end
           ),
           ', ' order by namespace.nspname, relation.relname, attribute.attnum, acl.grantee
         )
    into v_drift
    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
   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 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'
         );

  if v_drift is not null then
    raise exception 'kb_apply_prereqs: unexpected protected table grantee: %', v_drift;
  end if;

  select pg_catalog.string_agg(
           pg_catalog.format(
             '%I.%I(%s) -> %s',
             namespace.nspname,
             procedure.proname,
             pg_catalog.oidvectortypes(procedure.proargtypes),
             case
               when acl.grantee = 0 then 'PUBLIC'
               else pg_catalog.pg_get_userbyid(acl.grantee)
             end
           ),
           ', ' order by procedure.proname, pg_catalog.oidvectortypes(procedure.proargtypes), acl.grantee
         )
    into v_drift
    from pg_catalog.pg_proc procedure
    join pg_catalog.pg_namespace namespace on namespace.oid = procedure.pronamespace
    cross join lateral pg_catalog.aclexplode(
      coalesce(procedure.proacl, pg_catalog.acldefault('f', procedure.proowner))
    ) acl
   where namespace.nspname = 'kb_stage'
     and procedure.proname in (
           'approve_strict_proposal',
           'assert_approved_proposal',
           'finish_approved_proposal',
           'export_applied_proposal_replay_rows'
         )
     and acl.grantee <> 0
     and acl.grantee <> procedure.proowner
     and pg_catalog.pg_get_userbyid(acl.grantee) not in (
           'kb_gate_owner', 'kb_review', 'kb_apply'
         );

  if v_drift is not null then
    raise exception 'kb_apply_prereqs: unexpected protected function grantee: %', v_drift;
  end if;
end
$gate_preflight$;

-- This legacy three-argument SECURITY DEFINER path trusted caller-supplied
-- reviewer text. Its exact signature is the only obsolete overload auto-removed.
drop function if exists kb_stage.approve_strict_proposal(uuid, text, text);

alter table kb_stage.kb_review_principals owner to kb_gate_owner;
alter table kb_stage.kb_proposal_approvals owner to kb_gate_owner;

alter table kb_stage.kb_proposal_approvals
  add column if not exists approved_proposal_snapshot jsonb;

comment on column kb_stage.kb_proposal_approvals.approved_proposal_snapshot is
  'Exact post-approval kb_proposals row. NULL means exact replay material is unavailable.';

-- A still-approved legacy row is the only state whose complete approved shape
-- remains recoverable without guessing. Already-applied rows stay NULL because
-- their pre-apply fields, especially updated_at, cannot be reconstructed.
update kb_stage.kb_proposal_approvals approval
   set approved_proposal_snapshot = pg_catalog.to_jsonb(proposal)
  from kb_stage.kb_proposals proposal
 where approval.proposal_id = proposal.id
   and approval.approved_proposal_snapshot is null
   and proposal.status = 'approved'
   and proposal.applied_by_handle is null
   and proposal.applied_by_agent_id is null
   and proposal.applied_at is null
   and approval.proposal_type = proposal.proposal_type
   and approval.payload = proposal.payload
   and approval.reviewed_by_handle = proposal.reviewed_by_handle
   and approval.reviewed_by_agent_id is not distinct from proposal.reviewed_by_agent_id
   and approval.reviewed_at = proposal.reviewed_at
   and approval.review_note = proposal.review_note;

insert into kb_stage.kb_review_principals
  (db_role, reviewed_by_handle, reviewed_by_agent_id, active, created_at)
select 'kb_review'::name, 'm3taversal', a.id, true,
       '1970-01-01 00:00:00+00'::timestamptz
  from public.agents a
 where a.handle = 'm3taversal'
on conflict (db_role) do update
  set reviewed_by_handle = excluded.reviewed_by_handle,
      reviewed_by_agent_id = excluded.reviewed_by_agent_id,
      active = excluded.active;

-- Reset every explicit privilege held by the protected roles before granting
-- the documented matrix. This removes stale DELETE/TRUNCATE/REFERENCES/TRIGGER
-- rights from upgrades instead of layering narrow grants on top of them.
revoke all privileges on schema public, kb_stage
  from kb_gate_owner, kb_apply, kb_review;
grant usage on schema public, kb_stage to kb_gate_owner;
grant usage on schema public, kb_stage to kb_apply;
grant usage on schema public, kb_stage to kb_review;

revoke all privileges on table
  public.agents,
  public.strategies,
  public.strategy_nodes,
  public.claim_evidence,
  public.claim_edges,
  public.claims,
  public.sources,
  public.reasoning_tools,
  kb_stage.kb_proposals,
  kb_stage.kb_review_principals,
  kb_stage.kb_proposal_approvals
  from kb_gate_owner, kb_apply, kb_review;
revoke all privileges on table
  kb_stage.kb_proposals,
  kb_stage.kb_review_principals,
  kb_stage.kb_proposal_approvals
  from public;

-- Column grants are stored separately from table ACLs and survive a table-level
-- REVOKE. Clear them explicitly so stale per-column writes cannot bypass the
-- normalized table matrix.
do $column_acl_reset$
declare
  protected_table record;
begin
  for protected_table in
    select namespace.nspname as schema_name,
           relation.relname as table_name,
           pg_catalog.string_agg(
             pg_catalog.format('%I', attribute.attname),
             ', ' order by attribute.attnum
           ) as column_list
      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, 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
     group by namespace.nspname, relation.relname
  loop
    execute pg_catalog.format(
      'revoke all privileges (%s) on table %I.%I from kb_gate_owner, kb_apply, kb_review',
      protected_table.column_list,
      protected_table.schema_name,
      protected_table.table_name
    );
    if protected_table.schema_name = 'kb_stage'
       and protected_table.table_name in (
         'kb_proposals', 'kb_review_principals', 'kb_proposal_approvals',
         'teleo_v3_installation_ownership'
       ) then
      execute pg_catalog.format(
        'revoke all privileges (%s) on table %I.%I from public',
        protected_table.column_list,
        protected_table.schema_name,
        protected_table.table_name
      );
    end if;
  end loop;
end
$column_acl_reset$;

grant select on public.agents to kb_gate_owner;
grant select on public.agents to kb_apply;
grant select on public.agents to kb_review;

grant select, insert, update on public.strategies to kb_apply;
grant select, insert, update on public.strategy_nodes to kb_apply;
grant select, insert on public.claim_evidence to kb_apply;
grant select, insert on public.claim_edges to kb_apply;
grant select, insert on public.claims to kb_apply;
grant select, insert on public.sources to kb_apply;
grant select, insert on public.reasoning_tools to kb_apply;

-- The V3 assessment table is installed by a separate additive migration. Keep
-- this prerequisite runnable before that migration, then normalize its ACLs on
-- the first post-migration rerun.
do $v3_assessment_acl$
declare
  v_columns text;
begin
  if pg_catalog.to_regclass('public.claim_evidence_assessments') is not null then
    execute 'revoke all privileges on table public.claim_evidence_assessments '
            'from public, kb_gate_owner, kb_apply, kb_review';
    select pg_catalog.string_agg(pg_catalog.format('%I', attribute.attname), ', ' order by attribute.attnum)
      into v_columns
      from pg_catalog.pg_attribute attribute
     where attribute.attrelid = 'public.claim_evidence_assessments'::regclass
       and attribute.attnum > 0
       and not attribute.attisdropped;
    if v_columns is not null then
      execute pg_catalog.format(
        'revoke all privileges (%s) on table public.claim_evidence_assessments '
        'from public, kb_gate_owner, kb_apply, kb_review',
        v_columns
      );
    end if;
    execute 'grant select, insert on public.claim_evidence_assessments to kb_apply';
    execute 'grant select on public.claims, public.sources, public.claim_evidence, '
            'public.claim_edges, public.claim_evidence_assessments to kb_gate_owner';
  end if;
end
$v3_assessment_acl$;

do $v3_installation_ownership_acl$
begin
  if pg_catalog.to_regclass('kb_stage.teleo_v3_installation_ownership') is not null then
    execute 'revoke all privileges on table kb_stage.teleo_v3_installation_ownership '
            'from public, kb_gate_owner, kb_apply, kb_review';
  end if;
end
$v3_installation_ownership_acl$;

-- Keep the primary ledger denial explicit in addition to the full reset above.
revoke update on kb_stage.kb_proposals from kb_apply;
grant select on kb_stage.kb_proposals to kb_apply;
grant select on kb_stage.kb_proposals to kb_review;

grant select, update on kb_stage.kb_proposals to kb_gate_owner;
grant select on kb_stage.kb_review_principals to kb_gate_owner;
grant select, insert on kb_stage.kb_proposal_approvals to kb_gate_owner;

-- Required conflict arbiters. Existing production constraints satisfy these
-- checks; fresh clones receive named indexes only when no equivalent unique
-- index exists. Migration failure on pre-existing duplicates is intentional.
do $indexes$
begin
  if not exists (
    select 1
      from pg_catalog.pg_index i
      join pg_catalog.pg_class t on t.oid = i.indrelid
      join pg_catalog.pg_namespace n on n.oid = t.relnamespace
     where n.nspname = 'public'
       and t.relname = 'sources'
       and i.indisunique
       and i.indpred is null
       and replace(pg_catalog.pg_get_indexdef(i.indexrelid), ' ', '') like '%(hash)%'
  ) then
    execute 'create unique index kb_apply_sources_hash_unique on public.sources (hash)';
  end if;

  if not exists (
    select 1
      from pg_catalog.pg_index i
      join pg_catalog.pg_class t on t.oid = i.indrelid
      join pg_catalog.pg_namespace n on n.oid = t.relnamespace
     where n.nspname = 'public'
       and t.relname = 'claim_evidence'
       and i.indisunique
       and i.indpred is null
       and replace(pg_catalog.pg_get_indexdef(i.indexrelid), ' ', '')
           like '%(claim_id,source_id,role)%'
  ) then
    execute 'create unique index kb_apply_claim_evidence_semantic_unique '
            'on public.claim_evidence (claim_id, source_id, role)';
  end if;
end
$indexes$;

create unique index if not exists one_active_strategy_per_agent
    on public.strategies (agent_id) where active;

-- Reviewer-only state transition. The expected type/payload are the exact
-- Python-validated snapshot; a concurrent pending-row mutation makes the
-- locked WHERE clause fail. Reviewer identity comes from session_user mapping.
create or replace function kb_stage.approve_strict_proposal(
  p_proposal_id uuid,
  p_expected_proposal_type text,
  p_expected_payload jsonb,
  p_expected_reviewed_by_handle text,
  p_review_note text
) returns jsonb
language plpgsql
security definer
set search_path = pg_catalog, pg_temp
set timezone = 'UTC'
as $function$
declare
  v_row kb_stage.kb_proposals%rowtype;
  v_principal kb_stage.kb_review_principals%rowtype;
  v_reviewed_at timestamptz;
begin
  if nullif(pg_catalog.btrim(p_expected_reviewed_by_handle), '') is null then
    raise exception 'approve_strict_proposal: expected reviewer handle is required';
  end if;
  if nullif(pg_catalog.btrim(p_review_note), '') is null then
    raise exception 'approve_strict_proposal: review note is required';
  end if;

  select * into v_principal
    from kb_stage.kb_review_principals
   where db_role = session_user::name
     and active;
  if not found then
    raise exception 'approve_strict_proposal: session role % has no active reviewer principal',
      session_user;
  end if;
  if v_principal.reviewed_by_handle <> pg_catalog.btrim(p_expected_reviewed_by_handle) then
    raise exception 'approve_strict_proposal: session role % is mapped to reviewer %, not %',
      session_user, v_principal.reviewed_by_handle, p_expected_reviewed_by_handle;
  end if;

  select * into v_row
    from kb_stage.kb_proposals
   where id = p_proposal_id
     and status = 'pending_review'
     and proposal_type = p_expected_proposal_type
     and payload = p_expected_payload
   for update;
  if not found then
    raise exception 'approve_strict_proposal: proposal % is not the exact pending snapshot',
      p_proposal_id;
  end if;
  if v_row.proposal_type not in ('revise_strategy', 'add_edge', 'attach_evidence', 'approve_claim') then
    raise exception 'approve_strict_proposal: unsupported proposal_type %', v_row.proposal_type;
  end if;
  if pg_catalog.jsonb_typeof(v_row.payload) <> 'object'
     or not (v_row.payload ? 'apply_payload')
     or pg_catalog.jsonb_typeof(v_row.payload->'apply_payload') <> 'object' then
    raise exception 'approve_strict_proposal: proposal % lacks object payload.apply_payload',
      p_proposal_id;
  end if;

  v_reviewed_at := pg_catalog.clock_timestamp();
  update kb_stage.kb_proposals
     set status = 'approved',
         reviewed_by_handle = v_principal.reviewed_by_handle,
         reviewed_by_agent_id = v_principal.reviewed_by_agent_id,
         reviewed_at = v_reviewed_at,
         review_note = pg_catalog.btrim(p_review_note),
         updated_at = pg_catalog.clock_timestamp()
   where id = p_proposal_id
     and status = 'pending_review'
     and proposal_type = p_expected_proposal_type
     and payload = p_expected_payload
   returning * into v_row;
  if not found then
    raise exception 'approve_strict_proposal: proposal % changed while approval was recorded',
      p_proposal_id;
  end if;

  insert into kb_stage.kb_proposal_approvals
    (proposal_id, proposal_type, payload, reviewed_by_handle,
     reviewed_by_agent_id, reviewed_by_db_role, reviewed_at, review_note,
     approved_proposal_snapshot)
  values
    (v_row.id, v_row.proposal_type, v_row.payload, v_principal.reviewed_by_handle,
     v_principal.reviewed_by_agent_id, session_user::name, v_reviewed_at,
     pg_catalog.btrim(p_review_note), pg_catalog.to_jsonb(v_row));

  return pg_catalog.jsonb_build_object(
    'id', v_row.id::text,
    'proposal_type', v_row.proposal_type,
    'status', v_row.status,
    'reviewed_by_handle', v_row.reviewed_by_handle,
    'reviewed_by_agent_id', v_row.reviewed_by_agent_id::text,
    'reviewed_by_db_role', session_user::text,
    'reviewed_at', v_row.reviewed_at::text,
    'review_note', v_row.review_note
  );
end
$function$;

-- Apply-side precondition. It locks the proposal for the surrounding transaction
-- and compares both the mutable ledger and immutable approval snapshot.
create or replace function kb_stage.assert_approved_proposal(
  p_proposal_id uuid,
  p_proposal_type text,
  p_payload jsonb,
  p_reviewed_by_handle text,
  p_reviewed_by_agent_id uuid,
  p_reviewed_at timestamptz,
  p_review_note text
) returns void
language plpgsql
security definer
set search_path = pg_catalog, pg_temp
set timezone = 'UTC'
as $function$
begin
  perform 1
    from kb_stage.kb_proposals p
    join kb_stage.kb_proposal_approvals a on a.proposal_id = p.id
   where p.id = p_proposal_id
     and p.status = 'approved'
     and p.proposal_type = p_proposal_type
     and p.payload = p_payload
     and p.reviewed_by_handle = p_reviewed_by_handle
     and p.reviewed_by_agent_id is not distinct from p_reviewed_by_agent_id
     and p.reviewed_at = p_reviewed_at
     and p.review_note = p_review_note
     and a.proposal_type = p_proposal_type
     and a.payload = p_payload
     and a.reviewed_by_handle = p_reviewed_by_handle
     and a.reviewed_by_agent_id is not distinct from p_reviewed_by_agent_id
     and a.reviewed_at = p_reviewed_at
     and a.review_note = p_review_note
     and a.approved_proposal_snapshot = pg_catalog.to_jsonb(p)
   for update of p;
  if not found then
    raise exception 'assert_approved_proposal: proposal % is not the exact immutable approved snapshot',
      p_proposal_id;
  end if;
end
$function$;

-- Apply-side terminal transition. It repeats every snapshot comparison while
-- the proposal row remains locked, resolves a real service-agent FK, and changes
-- ledger fields only. kb_apply remains a trusted narrow canonical writer; the
-- Python transaction performs and independently verifies typed writes first.
create or replace function kb_stage.finish_approved_proposal(
  p_proposal_id uuid,
  p_proposal_type text,
  p_payload jsonb,
  p_reviewed_by_handle text,
  p_reviewed_by_agent_id uuid,
  p_reviewed_at timestamptz,
  p_review_note text,
  p_applied_by_handle text
) returns jsonb
language plpgsql
security definer
set search_path = pg_catalog, pg_temp
set timezone = 'UTC'
as $function$
declare
  v_row kb_stage.kb_proposals%rowtype;
  v_applied_by_agent_id uuid;
begin
  if nullif(pg_catalog.btrim(p_applied_by_handle), '') is null then
    raise exception 'finish_approved_proposal: applied-by handle is required';
  end if;

  perform 1
    from kb_stage.kb_proposals p
    join kb_stage.kb_proposal_approvals a on a.proposal_id = p.id
   where p.id = p_proposal_id
     and p.status = 'approved'
     and p.proposal_type = p_proposal_type
     and p.payload = p_payload
     and p.reviewed_by_handle = p_reviewed_by_handle
     and p.reviewed_by_agent_id is not distinct from p_reviewed_by_agent_id
     and p.reviewed_at = p_reviewed_at
     and p.review_note = p_review_note
     and a.proposal_type = p_proposal_type
     and a.payload = p_payload
     and a.reviewed_by_handle = p_reviewed_by_handle
     and a.reviewed_by_agent_id is not distinct from p_reviewed_by_agent_id
     and a.reviewed_at = p_reviewed_at
     and a.review_note = p_review_note
     and a.approved_proposal_snapshot = pg_catalog.to_jsonb(p)
   for update of p;
  if not found then
    raise exception 'finish_approved_proposal: proposal % is not the locked immutable approved snapshot',
      p_proposal_id;
  end if;

  select id into v_applied_by_agent_id
    from public.agents
   where handle = pg_catalog.btrim(p_applied_by_handle)
   limit 1;
  if v_applied_by_agent_id is null then
    raise exception 'finish_approved_proposal: applied-by handle % has no public.agents row',
      p_applied_by_handle;
  end if;

  update kb_stage.kb_proposals
     set status = 'applied',
         applied_by_handle = pg_catalog.btrim(p_applied_by_handle),
         applied_by_agent_id = v_applied_by_agent_id,
         applied_at = pg_catalog.clock_timestamp(),
         updated_at = pg_catalog.clock_timestamp()
   where id = p_proposal_id
     and status = 'approved'
     and proposal_type = p_proposal_type
     and payload = p_payload
     and reviewed_by_handle = p_reviewed_by_handle
     and reviewed_by_agent_id is not distinct from p_reviewed_by_agent_id
     and reviewed_at = p_reviewed_at
     and review_note = p_review_note
   returning * into v_row;
  if not found then
    raise exception 'finish_approved_proposal: proposal % changed during apply', p_proposal_id;
  end if;

  return pg_catalog.jsonb_build_object(
    'id', v_row.id::text,
    'proposal_type', v_row.proposal_type,
    'status', v_row.status,
    'applied_by_handle', v_row.applied_by_handle,
    'applied_by_agent_id', v_row.applied_by_agent_id::text,
    'applied_at', v_row.applied_at::text
  );
end
$function$;

-- Read-only transition export. kb_apply receives only EXECUTE on this function,
-- never SELECT on the approval table. The stored approved row must be a complete
-- kb_proposals record and differ from the applied row only by the guarded apply
-- transition before any private material leaves PostgreSQL.
create or replace function kb_stage.export_applied_proposal_replay_rows(
  p_proposal_id uuid
) returns jsonb
language plpgsql
stable
security definer
set search_path = pg_catalog, pg_temp
set timezone = 'UTC'
as $function$
declare
  v_rows jsonb;
begin
  select pg_catalog.jsonb_build_object(
           'approved_proposal', approval.approved_proposal_snapshot,
           'approval_snapshot',
             pg_catalog.to_jsonb(approval) - 'approved_proposal_snapshot',
           'applied_proposal', pg_catalog.to_jsonb(proposal)
         )
    into v_rows
    from kb_stage.kb_proposals proposal
    join kb_stage.kb_proposal_approvals approval
      on approval.proposal_id = proposal.id
   where proposal.id = p_proposal_id
     and proposal.status = 'applied'
     and proposal.applied_by_handle is not null
     and proposal.applied_by_agent_id is not null
     and proposal.applied_at is not null
     and proposal.reviewed_by_handle is not null
     and proposal.reviewed_by_agent_id is not null
     and proposal.reviewed_at is not null
     and proposal.review_note is not null
     and nullif(pg_catalog.btrim(approval.reviewed_by_db_role::text), '') is not null
     and pg_catalog.jsonb_typeof(approval.approved_proposal_snapshot) = 'object'
     and approval.approved_proposal_snapshot = pg_catalog.to_jsonb(
           pg_catalog.jsonb_populate_record(
             null::kb_stage.kb_proposals,
             approval.approved_proposal_snapshot
           )
         )
     and approval.approved_proposal_snapshot->>'status' = 'approved'
     and approval.approved_proposal_snapshot->'applied_by_handle' = 'null'::jsonb
     and approval.approved_proposal_snapshot->'applied_by_agent_id' = 'null'::jsonb
     and approval.approved_proposal_snapshot->'applied_at' = 'null'::jsonb
     and approval.proposal_id = (approval.approved_proposal_snapshot->>'id')::uuid
     and approval.proposal_type = approval.approved_proposal_snapshot->>'proposal_type'
     and approval.payload = approval.approved_proposal_snapshot->'payload'
     and approval.reviewed_by_handle =
           approval.approved_proposal_snapshot->>'reviewed_by_handle'
     and approval.reviewed_by_agent_id =
           (approval.approved_proposal_snapshot->>'reviewed_by_agent_id')::uuid
     and approval.reviewed_at =
           (approval.approved_proposal_snapshot->>'reviewed_at')::timestamptz
     and approval.review_note = approval.approved_proposal_snapshot->>'review_note'
     and (
       approval.approved_proposal_snapshot || pg_catalog.jsonb_build_object(
         'status', proposal.status,
         'applied_by_handle', proposal.applied_by_handle,
         'applied_by_agent_id', proposal.applied_by_agent_id,
         'applied_at', proposal.applied_at,
         'updated_at', proposal.updated_at
       )
     ) = pg_catalog.to_jsonb(proposal);

  if v_rows is null then
    raise exception 'export_applied_proposal_replay_rows: exact applied transition unavailable for proposal %',
      p_proposal_id;
  end if;
  return v_rows;
end
$function$;

alter function kb_stage.approve_strict_proposal(uuid, text, jsonb, text, text)
  owner to kb_gate_owner;
alter function kb_stage.assert_approved_proposal(uuid, text, jsonb, text, uuid, timestamptz, text)
  owner to kb_gate_owner;
alter function kb_stage.finish_approved_proposal(uuid, text, jsonb, text, uuid, timestamptz, text, text)
  owner to kb_gate_owner;
alter function kb_stage.export_applied_proposal_replay_rows(uuid)
  owner to kb_gate_owner;

revoke all on function kb_stage.approve_strict_proposal(uuid, text, jsonb, text, text)
  from public, kb_gate_owner, kb_apply, kb_review;
revoke all on function kb_stage.assert_approved_proposal(uuid, text, jsonb, text, uuid, timestamptz, text)
  from public, kb_gate_owner, kb_apply, kb_review;
revoke all on function kb_stage.finish_approved_proposal(uuid, text, jsonb, text, uuid, timestamptz, text, text)
  from public, kb_gate_owner, kb_apply, kb_review;
revoke all on function kb_stage.export_applied_proposal_replay_rows(uuid)
  from public, kb_gate_owner, kb_apply, kb_review;

grant execute on function kb_stage.approve_strict_proposal(uuid, text, jsonb, text, text)
  to kb_review;
grant execute on function kb_stage.assert_approved_proposal(uuid, text, jsonb, text, uuid, timestamptz, text)
  to kb_apply;
grant execute on function kb_stage.finish_approved_proposal(uuid, text, jsonb, text, uuid, timestamptz, text, text)
  to kb_apply;
grant execute on function kb_stage.export_applied_proposal_replay_rows(uuid)
  to kb_apply;

-- Assert the complete protected catalog contract before commit. These checks
-- make reruns an upgrade proof rather than a best-effort sequence of grants.
do $final_invariants$
declare
  v_drift text;
begin
  select pg_catalog.string_agg(role.rolname, ', ' order by role.rolname)
    into v_drift
    from pg_catalog.pg_roles role
   where role.rolname in ('kb_gate_owner', 'kb_review', 'kb_apply')
     and (
       role.rolsuper
       or role.rolinherit
       or role.rolcreatedb
       or role.rolcreaterole
       or role.rolreplication
       or role.rolbypassrls
       or (role.rolname = 'kb_gate_owner' and role.rolcanlogin)
       or (role.rolname in ('kb_review', 'kb_apply') and not role.rolcanlogin)
     );

  if v_drift is not null then
    raise exception 'kb_apply_prereqs: protected role attributes did not normalize: %', v_drift;
  end if;

  select pg_catalog.string_agg(
           pg_catalog.format('%I -> %I', member_role.rolname, granted_role.rolname),
           ', ' order by member_role.rolname, granted_role.rolname
         )
    into v_drift
    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 (
           'kb_gate_owner', 'kb_review', 'kb_apply', 'kb_observatory_read'
         )
      or granted_role.rolname in ('kb_gate_owner', 'kb_review', 'kb_apply');

  if v_drift is not null then
    raise exception 'kb_apply_prereqs: protected role membership appeared during migration: %', v_drift;
  end if;

  if exists (select 1 from public.agents where handle = 'm3ta')
     or not exists (
       select 1
         from kb_stage.kb_review_principals principal
         join public.agents reviewer
           on reviewer.id = principal.reviewed_by_agent_id
        where principal.db_role = 'kb_review'::name
          and principal.reviewed_by_handle = 'm3taversal'
          and principal.active
          and reviewer.handle = 'm3taversal'
     ) then
    raise exception 'kb_apply_prereqs: kb_review principal did not normalize to active m3taversal';
  end if;

  select pg_catalog.string_agg(
           pg_catalog.format(
             '%I on %I usage=%s create=%s',
             protected_role.rolname,
             protected_schema.nspname,
             pg_catalog.has_schema_privilege(protected_role.rolname, protected_schema.nspname, 'USAGE'),
             pg_catalog.has_schema_privilege(protected_role.rolname, protected_schema.nspname, 'CREATE')
           ),
           ', ' order by protected_role.rolname, protected_schema.nspname
         )
    into v_drift
    from (values ('kb_gate_owner'), ('kb_review'), ('kb_apply')) protected_role(rolname)
    cross join (values ('public'), ('kb_stage')) protected_schema(nspname)
   where not pg_catalog.has_schema_privilege(
               protected_role.rolname, protected_schema.nspname, 'USAGE'
             )
      or pg_catalog.has_schema_privilege(
           protected_role.rolname, protected_schema.nspname, 'CREATE'
         );

  if v_drift is not null then
    raise exception 'kb_apply_prereqs: protected schema ACL mismatch: %', v_drift;
  end if;

  select pg_catalog.string_agg(
           pg_catalog.format('%I.%I owned by %I', namespace.nspname, relation.relname, owner_role.rolname),
           ', ' order by namespace.nspname, relation.relname
         )
    into v_drift
    from pg_catalog.pg_class relation
    join pg_catalog.pg_namespace namespace on namespace.oid = relation.relnamespace
    join pg_catalog.pg_roles owner_role on owner_role.oid = relation.relowner
   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 not (
       (namespace.nspname = 'kb_stage'
        and relation.relname in (
          'kb_review_principals', 'kb_proposal_approvals', 'teleo_v3_installation_ownership'
        )
        and owner_role.rolname = 'kb_gate_owner')
       or (not (
             namespace.nspname = 'kb_stage'
             and relation.relname in (
               'kb_review_principals', 'kb_proposal_approvals', 'teleo_v3_installation_ownership'
             )
           )
           and owner_role.rolname = current_user)
     );

  if v_drift is not null then
    raise exception 'kb_apply_prereqs: protected table ownership mismatch: %', v_drift;
  end if;

  if not exists (
    select 1
      from pg_catalog.pg_attribute attribute
     where attribute.attrelid = 'kb_stage.kb_proposal_approvals'::regclass
       and attribute.attname = 'approved_proposal_snapshot'
       and attribute.atttypid = 'pg_catalog.jsonb'::regtype
       and not attribute.attnotnull
       and not attribute.attisdropped
  ) then
    raise exception 'kb_apply_prereqs: approved proposal snapshot column is not nullable JSONB';
  end if;

  with expected(signature) as (
    values
      ('approve_strict_proposal(uuid, text, jsonb, text, text)'),
      ('assert_approved_proposal(uuid, text, jsonb, text, uuid, timestamp with time zone, text)'),
      ('finish_approved_proposal(uuid, text, jsonb, text, uuid, timestamp with time zone, text, text)'),
      ('export_applied_proposal_replay_rows(uuid)')
  ), actual(signature) as (
    select procedure.proname || '(' || pg_catalog.oidvectortypes(procedure.proargtypes) || ')'
      from pg_catalog.pg_proc procedure
      join pg_catalog.pg_namespace namespace on namespace.oid = procedure.pronamespace
     where namespace.nspname = 'kb_stage'
       and procedure.proname in (
             'approve_strict_proposal',
             'assert_approved_proposal',
             'finish_approved_proposal',
             'export_applied_proposal_replay_rows'
           )
  ), delta(kind, signature) as (
    select 'unexpected', actual.signature
      from actual
     where not exists (select 1 from expected where expected.signature = actual.signature)
    union all
    select 'missing', expected.signature
      from expected
     where not exists (select 1 from actual where actual.signature = expected.signature)
  )
  select pg_catalog.string_agg(kind || ' ' || signature, ', ' order by kind, signature)
    into v_drift
    from delta;

  if v_drift is not null then
    raise exception 'kb_apply_prereqs: protected function signature mismatch: %', v_drift;
  end if;

  select pg_catalog.string_agg(
           pg_catalog.format(
             '%I.%I(%s) owner=%I security_definer=%s config=%s',
             namespace.nspname,
             procedure.proname,
             pg_catalog.oidvectortypes(procedure.proargtypes),
             owner_role.rolname,
             procedure.prosecdef,
             procedure.proconfig
           ),
           ', ' order by procedure.proname
         )
    into v_drift
    from pg_catalog.pg_proc procedure
    join pg_catalog.pg_namespace namespace on namespace.oid = procedure.pronamespace
    join pg_catalog.pg_roles owner_role on owner_role.oid = procedure.proowner
   where namespace.nspname = 'kb_stage'
     and procedure.proname in (
           'approve_strict_proposal',
           'assert_approved_proposal',
           'finish_approved_proposal',
           'export_applied_proposal_replay_rows'
         )
     and (
       owner_role.rolname <> 'kb_gate_owner'
       or not procedure.prosecdef
       or pg_catalog.cardinality(procedure.proconfig) <> 2
       or not procedure.proconfig @> array[
         'search_path=pg_catalog, pg_temp'::text,
         'TimeZone=UTC'::text
       ]
       or (
         procedure.proname = 'export_applied_proposal_replay_rows'
         and (
           procedure.provolatile <> 's'
           or procedure.prokind <> 'f'
           or procedure.prorettype <> 'pg_catalog.jsonb'::regtype
         )
       )
     );

  if v_drift is not null then
    raise exception 'kb_apply_prereqs: protected function attribute mismatch: %', v_drift;
  end if;

  with tracked(schema_name, object_name) as (
    values
      ('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')
  )
  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
           ),
           ', ' order by namespace.nspname, relation.relname, attribute.attnum, acl.grantee
         )
    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;

  if v_drift is not null then
    raise exception 'kb_apply_prereqs: protected column ACL mismatch: %', v_drift;
  end if;

  with core_expected(schema_name, object_name, grantee, privilege_type, is_grantable) as (
    values
      ('public', 'agents', 'kb_gate_owner', 'SELECT', false),
      ('public', 'agents', 'kb_apply', 'SELECT', false),
      ('public', 'agents', 'kb_review', 'SELECT', false),
      ('public', 'strategies', 'kb_apply', 'SELECT', false),
      ('public', 'strategies', 'kb_apply', 'INSERT', false),
      ('public', 'strategies', 'kb_apply', 'UPDATE', false),
      ('public', 'strategy_nodes', 'kb_apply', 'SELECT', false),
      ('public', 'strategy_nodes', 'kb_apply', 'INSERT', false),
      ('public', 'strategy_nodes', 'kb_apply', 'UPDATE', false),
      ('public', 'claim_evidence', 'kb_apply', 'SELECT', false),
      ('public', 'claim_evidence', 'kb_apply', 'INSERT', false),
      ('public', 'claim_edges', 'kb_apply', 'SELECT', false),
      ('public', 'claim_edges', 'kb_apply', 'INSERT', false),
      ('public', 'claims', 'kb_apply', 'SELECT', false),
      ('public', 'claims', 'kb_apply', 'INSERT', false),
      ('public', 'sources', 'kb_apply', 'SELECT', false),
      ('public', 'sources', 'kb_apply', 'INSERT', false),
      ('public', 'reasoning_tools', 'kb_apply', 'SELECT', false),
      ('public', 'reasoning_tools', 'kb_apply', 'INSERT', false),
      ('kb_stage', 'kb_proposals', 'kb_gate_owner', 'SELECT', false),
      ('kb_stage', 'kb_proposals', 'kb_gate_owner', 'UPDATE', false),
      ('kb_stage', 'kb_proposals', 'kb_apply', 'SELECT', false),
      ('kb_stage', 'kb_proposals', 'kb_review', 'SELECT', false)
    union all
    select 'public', 'claim_evidence_assessments', 'kb_apply', 'SELECT', false
     where pg_catalog.to_regclass('public.claim_evidence_assessments') is not null
    union all
    select 'public', 'claim_evidence_assessments', 'kb_apply', 'INSERT', false
     where pg_catalog.to_regclass('public.claim_evidence_assessments') is not null
    union all
    select 'public', 'claims', 'kb_gate_owner', 'SELECT', false
     where pg_catalog.to_regclass('public.claim_evidence_assessments') is not null
    union all
    select 'public', 'sources', 'kb_gate_owner', 'SELECT', false
     where pg_catalog.to_regclass('public.claim_evidence_assessments') is not null
    union all
    select 'public', 'claim_evidence', 'kb_gate_owner', 'SELECT', false
     where pg_catalog.to_regclass('public.claim_evidence_assessments') is not null
    union all
    select 'public', 'claim_edges', 'kb_gate_owner', 'SELECT', false
     where pg_catalog.to_regclass('public.claim_evidence_assessments') is not null
    union all
    select 'public', 'claim_evidence_assessments', 'kb_gate_owner', 'SELECT', false
     where pg_catalog.to_regclass('public.claim_evidence_assessments') is not null
  ), 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
  ), tracked(schema_name, object_name) as (
    values
      ('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')
  ), actual(schema_name, object_name, grantee, privilege_type, is_grantable) as (
    select distinct
           namespace.nspname::text,
           relation.relname::text,
           (case
              when acl.grantee = 0 then 'PUBLIC'
              else grantee_role.rolname
            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
      cross join lateral pg_catalog.aclexplode(
        coalesce(relation.relacl, pg_catalog.acldefault('r', relation.relowner))
      ) acl
      left join pg_catalog.pg_roles grantee_role on grantee_role.oid = acl.grantee
     where acl.grantee <> relation.relowner
  ), delta(kind, schema_name, object_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(
             '%s %I.%I -> %I %s grantable=%s',
             kind, schema_name, object_name, grantee, privilege_type, is_grantable
           ),
           ', ' order by kind, schema_name, object_name, grantee, privilege_type
         )
    into v_drift
    from delta;

  if v_drift is not null then
    raise exception 'kb_apply_prereqs: protected table ACL mismatch: %', v_drift;
  end if;

  with expected(signature, grantee, privilege_type, is_grantable) as (
    values
      ('approve_strict_proposal(uuid, text, jsonb, text, text)', 'kb_review', 'EXECUTE', false),
      ('assert_approved_proposal(uuid, text, jsonb, text, uuid, timestamp with time zone, text)',
       'kb_apply', 'EXECUTE', false),
      ('finish_approved_proposal(uuid, text, jsonb, text, uuid, timestamp with time zone, text, text)',
       'kb_apply', 'EXECUTE', false),
      ('export_applied_proposal_replay_rows(uuid)',
       'kb_apply', 'EXECUTE', false)
  ), actual(signature, grantee, privilege_type, is_grantable) as (
    select distinct
           procedure.proname || '(' || pg_catalog.oidvectortypes(procedure.proargtypes) || ')',
           (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 pg_catalog.pg_proc procedure
      join pg_catalog.pg_namespace namespace on namespace.oid = procedure.pronamespace
      cross join lateral pg_catalog.aclexplode(
        coalesce(procedure.proacl, pg_catalog.acldefault('f', procedure.proowner))
      ) acl
     where namespace.nspname = 'kb_stage'
       and procedure.proname in (
             'approve_strict_proposal',
             'assert_approved_proposal',
             'finish_approved_proposal',
             'export_applied_proposal_replay_rows'
           )
       and acl.grantee <> procedure.proowner
  ), delta(kind, signature, 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(
             '%s %s -> %I %s grantable=%s',
             kind, signature, grantee, privilege_type, is_grantable
           ),
           ', ' order by kind, signature, grantee, privilege_type
         )
    into v_drift
    from delta;

  if v_drift is not null then
    raise exception 'kb_apply_prereqs: protected function ACL mismatch: %', v_drift;
  end if;
end
$final_invariants$;

commit;
