-- 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;

-- The independently provisioned GCP runtime keeps an exact column-level read
-- projection. Record that reviewed contract in this transaction so a later
-- review/apply prerequisite rerun can preserve the scoped grants without
-- accepting arbitrary runtime ACLs. The matching test binds these rows to the
-- runtime-role provisioning manifest and the executable query contract.
create temporary table kb_apply_prereqs_leoclean_read_column_allowlist (
  schema_name name not null,
  relation_name name not null,
  column_name name not null,
  column_ordinal smallint not null,
  primary key (schema_name, relation_name, column_name),
  unique (schema_name, relation_name, column_ordinal)
) on commit drop;

insert into pg_temp.kb_apply_prereqs_leoclean_read_column_allowlist (
  schema_name,
  relation_name,
  column_name,
  column_ordinal
) values
  ('public', 'agents', 'id', 1),
  ('public', 'agents', 'handle', 2),
  ('public', 'claims', 'id', 1),
  ('public', 'claims', 'type', 2),
  ('public', 'claims', 'text', 3),
  ('public', 'claims', 'status', 4),
  ('public', 'claims', 'confidence', 5),
  ('public', 'claims', 'tags', 6),
  ('public', 'claims', 'superseded_by', 7),
  ('public', 'claims', 'created_at', 8),
  ('public', 'claims', 'updated_at', 9),
  ('public', 'claim_evidence', 'claim_id', 1),
  ('public', 'claim_evidence', 'source_id', 2),
  ('public', 'claim_evidence', 'role', 3),
  ('public', 'claim_evidence', 'weight', 4),
  ('public', 'claim_edges', 'id', 1),
  ('public', 'claim_edges', 'from_claim', 2),
  ('public', 'claim_edges', 'to_claim', 3),
  ('public', 'claim_edges', 'edge_type', 4),
  ('public', 'sources', 'id', 1),
  ('public', 'sources', 'source_type', 2),
  ('public', 'sources', 'url', 3),
  ('public', 'sources', 'storage_path', 4),
  ('public', 'sources', 'excerpt', 5),
  ('public', 'sources', 'hash', 6),
  ('public', 'personas', 'agent_id', 1),
  ('public', 'personas', 'name', 2),
  ('public', 'personas', 'voice', 3),
  ('public', 'personas', 'role', 4),
  ('public', 'personas', 'source_ref', 5),
  ('public', 'personas', 'lens', 6),
  ('public', 'strategies', 'agent_id', 1),
  ('public', 'strategies', 'diagnosis', 2),
  ('public', 'strategies', 'guiding_policy', 3),
  ('public', 'strategies', 'proximate_objectives', 4),
  ('public', 'strategies', 'version', 5),
  ('public', 'strategies', 'active', 6),
  ('public', 'beliefs', 'agent_id', 1),
  ('public', 'beliefs', 'level', 2),
  ('public', 'beliefs', 'statement', 3),
  ('public', 'beliefs', 'falsifier', 4),
  ('public', 'beliefs', 'rank', 5),
  ('public', 'beliefs', 'status', 6),
  ('public', 'blindspots', 'agent_id', 1),
  ('public', 'blindspots', 'name', 2),
  ('public', 'blindspots', 'description', 3),
  ('public', 'blindspots', 'correction', 4),
  ('public', 'blindspots', 'kind', 5),
  ('public', 'blindspots', 'status', 6),
  ('public', 'agent_roles', 'agent_id', 1),
  ('public', 'agent_roles', 'title', 2),
  ('public', 'agent_roles', 'description', 3),
  ('public', 'peer_models', 'subject_id', 1),
  ('public', 'peer_models', 'peer_id', 2),
  ('public', 'peer_models', 'domain', 3),
  ('public', 'peer_models', 'outranks_on', 4),
  ('public', 'peer_models', 'deference_rule', 5),
  ('public', 'behavioral_rules', 'agent_id', 1),
  ('public', 'behavioral_rules', 'category', 2),
  ('public', 'behavioral_rules', 'rule', 3),
  ('public', 'behavioral_rules', 'rationale', 4),
  ('public', 'contributor_rules', 'agent_id', 1),
  ('public', 'contributor_rules', 'name', 2),
  ('public', 'contributor_rules', 'directive', 3),
  ('public', 'contributor_rules', 'ci_tier', 4),
  ('public', 'contributor_rules', 'weighting', 5),
  ('public', 'contributor_rules', 'rationale', 6),
  ('public', 'reasoning_tools', 'agent_id', 1),
  ('public', 'reasoning_tools', 'name', 2),
  ('public', 'reasoning_tools', 'description', 3),
  ('public', 'reasoning_tools', 'category', 4),
  ('public', 'governance_gates', 'agent_id', 1),
  ('public', 'governance_gates', 'name', 2),
  ('public', 'governance_gates', 'criteria', 3),
  ('public', 'governance_gates', 'evidence_bar', 4),
  ('public', 'governance_gates', 'pass_condition', 5),
  ('kb_stage', 'kb_proposals', 'id', 1),
  ('kb_stage', 'kb_proposals', 'proposal_type', 2),
  ('kb_stage', 'kb_proposals', 'status', 3),
  ('kb_stage', 'kb_proposals', 'proposed_by_handle', 4),
  ('kb_stage', 'kb_proposals', 'proposed_by_agent_id', 5),
  ('kb_stage', 'kb_proposals', 'channel', 6),
  ('kb_stage', 'kb_proposals', 'source_ref', 7),
  ('kb_stage', 'kb_proposals', 'rationale', 8),
  ('kb_stage', 'kb_proposals', 'payload', 9),
  ('kb_stage', 'kb_proposals', 'reviewed_by_handle', 10),
  ('kb_stage', 'kb_proposals', 'reviewed_at', 11),
  ('kb_stage', 'kb_proposals', 'review_note', 12),
  ('kb_stage', 'kb_proposals', 'applied_by_handle', 13),
  ('kb_stage', 'kb_proposals', 'applied_at', 14),
  ('kb_stage', 'kb_proposals', 'created_at', 15),
  ('kb_stage', 'kb_proposals', 'updated_at', 16);

-- The proposal function owner has only the exact underlying table rights its
-- SECURITY DEFINER implementation needs. These rows describe bounded ACLs that
-- may already exist when this prerequisite is rerun after runtime provisioning.
create temporary table kb_apply_prereqs_leoclean_stage_owner_column_allowlist (
  schema_name name not null,
  relation_name name not null,
  column_name name not null,
  privilege_type text not null,
  primary key (schema_name, relation_name, column_name, privilege_type)
) on commit drop;

insert into pg_temp.kb_apply_prereqs_leoclean_stage_owner_column_allowlist (
  schema_name,
  relation_name,
  column_name,
  privilege_type
) values
  ('public', 'agents', 'id', 'SELECT'),
  ('public', 'agents', 'handle', 'SELECT'),
  ('kb_stage', 'kb_proposals', 'proposal_type', 'INSERT'),
  ('kb_stage', 'kb_proposals', 'status', 'INSERT'),
  ('kb_stage', 'kb_proposals', 'proposed_by_handle', 'INSERT'),
  ('kb_stage', 'kb_proposals', 'proposed_by_agent_id', 'INSERT'),
  ('kb_stage', 'kb_proposals', 'channel', 'INSERT'),
  ('kb_stage', 'kb_proposals', 'source_ref', 'INSERT'),
  ('kb_stage', 'kb_proposals', 'rationale', 'INSERT'),
  ('kb_stage', 'kb_proposals', 'payload', 'INSERT');

-- Snapshot the exact bounded external ACLs present before normalization. Final
-- verification compares against these rows, so this migration must preserve
-- them exactly; it cannot silently add another otherwise allowlisted grant.
create temporary table kb_apply_prereqs_preserved_table_acls (
  schema_name name not null,
  relation_name name not null,
  grantee name not null,
  privilege_type text not null,
  is_grantable boolean not null,
  primary key (schema_name, relation_name, grantee, privilege_type)
) on commit drop;

insert into pg_temp.kb_apply_prereqs_preserved_table_acls (
  schema_name,
  relation_name,
  grantee,
  privilege_type,
  is_grantable
)
select namespace.nspname,
       relation.relname,
       grantee_role.rolname,
       pg_catalog.upper(acl.privilege_type),
       acl.is_grantable
  from pg_catalog.pg_class relation
  join pg_catalog.pg_namespace namespace on namespace.oid = relation.relnamespace
 cross join lateral pg_catalog.aclexplode(
   coalesce(relation.relacl, pg_catalog.acldefault('r', relation.relowner))
 ) acl
  join pg_catalog.pg_roles grantee_role on grantee_role.oid = acl.grantee
 where (
   grantee_role.rolname = 'kb_observatory_read'
   and (namespace.nspname, relation.relname) in (
     ('public', 'claims'),
     ('public', 'sources'),
     ('public', 'claim_evidence'),
     ('public', 'claim_edges'),
     ('kb_stage', 'kb_proposals')
   )
   and pg_catalog.upper(acl.privilege_type) = 'SELECT'
   and not acl.is_grantable
 ) or (
   grantee_role.rolname = 'leoclean_kb_stage_owner'
   and namespace.nspname = 'kb_stage'
   and relation.relname = 'kb_proposals'
   and pg_catalog.upper(acl.privilege_type) = 'SELECT'
   and not acl.is_grantable
 );

create temporary table kb_apply_prereqs_preserved_column_acls (
  schema_name name not null,
  relation_name name not null,
  column_name name not null,
  grantee name not null,
  privilege_type text not null,
  is_grantable boolean not null,
  primary key (schema_name, relation_name, column_name, grantee, privilege_type)
) on commit drop;

insert into pg_temp.kb_apply_prereqs_preserved_column_acls (
  schema_name,
  relation_name,
  column_name,
  grantee,
  privilege_type,
  is_grantable
)
select namespace.nspname,
       relation.relname,
       attribute.attname,
       grantee_role.rolname,
       pg_catalog.upper(acl.privilege_type),
       acl.is_grantable
  from pg_catalog.pg_attribute attribute
  join pg_catalog.pg_class relation on relation.oid = attribute.attrelid
  join pg_catalog.pg_namespace namespace on namespace.oid = relation.relnamespace
 cross join lateral pg_catalog.aclexplode(attribute.attacl) acl
  join pg_catalog.pg_roles grantee_role on grantee_role.oid = acl.grantee
 where (namespace.nspname, relation.relname) in (
         ('public', 'agents'),
         ('public', 'strategies'),
         ('public', 'strategy_nodes'),
         ('public', 'claim_evidence'),
         ('public', 'claim_evidence_assessments'),
         ('public', 'claim_edges'),
         ('public', 'claims'),
         ('public', 'sources'),
         ('public', 'reasoning_tools'),
         ('kb_stage', 'kb_proposals'),
         ('kb_stage', 'kb_review_principals'),
         ('kb_stage', 'kb_proposal_approvals'),
         ('kb_stage', 'teleo_v3_installation_ownership')
       )
   and attribute.attnum > 0
   and not attribute.attisdropped
   and not acl.is_grantable
   and (
     (
       grantee_role.rolname = 'leoclean_kb_runtime'
       and pg_catalog.upper(acl.privilege_type) = 'SELECT'
       and exists (
         select 1
           from pg_temp.kb_apply_prereqs_leoclean_read_column_allowlist allowlist
          where allowlist.schema_name = namespace.nspname
            and allowlist.relation_name = relation.relname
            and allowlist.column_name = attribute.attname
       )
     )
     or (
       grantee_role.rolname = 'leoclean_kb_stage_owner'
       and exists (
         select 1
           from pg_temp.kb_apply_prereqs_leoclean_stage_owner_column_allowlist allowlist
          where allowlist.schema_name = namespace.nspname
            and allowlist.relation_name = relation.relname
            and allowlist.column_name = attribute.attname
            and allowlist.privilege_type = pg_catalog.upper(acl.privilege_type)
       )
     )
   );

-- Protected review/apply roles are standalone principals. The Observatory
-- service principal is intentionally a member of kb_observatory_read, but the
-- read role itself must never be a member of another role: that would create an
-- 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',
           'leoclean_kb_runtime', 'leoclean_kb_stage_owner'
         )
      or granted_role.rolname in (
           'kb_gate_owner', 'kb_review', 'kb_apply',
           'leoclean_kb_runtime', 'leoclean_kb_stage_owner'
         );

  if v_memberships is not null then
    raise exception 'kb_apply_prereqs: unexpected protected role membership: %', v_memberships;
  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 exists (
         select 1
           from pg_temp.kb_apply_prereqs_preserved_table_acls preserved
          where preserved.schema_name = namespace.nspname
            and preserved.relation_name = relation.relname
            and preserved.grantee = pg_catalog.pg_get_userbyid(acl.grantee)
            and preserved.privilege_type = pg_catalog.upper(acl.privilege_type)
            and preserved.is_grantable = acl.is_grantable
       )
     );

  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 not (
       pg_catalog.pg_get_userbyid(acl.grantee) in (
         'kb_gate_owner', 'kb_review', 'kb_apply'
       )
       or exists (
         select 1
           from pg_temp.kb_apply_prereqs_preserved_column_acls preserved
          where preserved.schema_name = namespace.nspname
            and preserved.relation_name = relation.relname
            and preserved.column_name = attribute.attname
            and preserved.grantee = pg_catalog.pg_get_userbyid(acl.grantee)
            and preserved.privilege_type = pg_catalog.upper(acl.privilege_type)
            and preserved.is_grantable = acl.is_grantable
       )
     );

  if v_drift is not null then
    raise exception 'kb_apply_prereqs: unexpected protected table grantee: %', v_drift;
  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',
           'leoclean_kb_runtime', 'leoclean_kb_stage_owner'
         )
      or granted_role.rolname in (
           'kb_gate_owner', 'kb_review', 'kb_apply',
           'leoclean_kb_runtime', 'leoclean_kb_stage_owner'
         );

  if v_drift is not null then
    raise exception 'kb_apply_prereqs: protected role membership appeared during migration: %', v_drift;
  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')
  ), actual(schema_name, object_name, column_name, grantee, privilege_type, is_grantable) as (
    select namespace.nspname::text,
           relation.relname::text,
           attribute.attname::text,
           (case
              when acl.grantee = 0 then 'PUBLIC'
              else pg_catalog.pg_get_userbyid(acl.grantee)
            end)::text,
           pg_catalog.upper(acl.privilege_type),
           acl.is_grantable
      from tracked
      join pg_catalog.pg_namespace namespace on namespace.nspname = tracked.schema_name
      join pg_catalog.pg_class relation
        on relation.relnamespace = namespace.oid
       and relation.relname = tracked.object_name
      join pg_catalog.pg_attribute attribute on attribute.attrelid = relation.oid
     cross join lateral pg_catalog.aclexplode(attribute.attacl) acl
     where attribute.attnum > 0
       and not attribute.attisdropped
       and acl.grantee <> relation.relowner
  ), expected(schema_name, object_name, column_name, grantee, privilege_type, is_grantable) as (
    select preserved.schema_name::text,
           preserved.relation_name::text,
           preserved.column_name::text,
           preserved.grantee::text,
           preserved.privilege_type,
           preserved.is_grantable
      from pg_temp.kb_apply_prereqs_preserved_column_acls preserved
  ), delta(kind, schema_name, object_name, column_name, grantee, privilege_type, is_grantable) as (
    select 'unexpected', actual.*
      from actual
     where not exists (
       select 1 from expected where pg_catalog.to_jsonb(expected) = pg_catalog.to_jsonb(actual)
     )
    union all
    select 'missing', expected.*
      from expected
     where not exists (
       select 1 from actual where pg_catalog.to_jsonb(actual) = pg_catalog.to_jsonb(expected)
     )
  )
  select pg_catalog.string_agg(
           pg_catalog.format(
             '%s %I.%I.%I -> %s %s grantable=%s',
             kind,
             schema_name,
             object_name,
             column_name,
             grantee,
             privilege_type,
             is_grantable
           ),
           ', ' order by kind, schema_name, object_name, column_name, grantee, privilege_type
         )
    into v_drift
    from delta;

  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
    select preserved.schema_name::text,
           preserved.relation_name::text,
           preserved.grantee::text,
           preserved.privilege_type,
           preserved.is_grantable
      from pg_temp.kb_apply_prereqs_preserved_table_acls preserved
  ), tracked(schema_name, object_name) as (
    values
      ('public', 'agents'),
      ('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;