from __future__ import annotations

import importlib.util
import json
import os
import re
import shutil
import subprocess
import time
import uuid
from pathlib import Path
from types import ModuleType

import pytest
import yaml

from ops import verify_gcp_leoclean_runtime_permissions as verifier

REPO_ROOT = Path(__file__).resolve().parents[1]
ROLE_SQL = REPO_ROOT / "ops" / "gcp_leoclean_runtime_role.sql"
RUNTIME_TOOL = REPO_ROOT / "hermes-agent" / "leoclean-bin" / "cloudsql_memory_tool.py"
RUN_ENV = "TELEO_RUN_LEOCLEAN_RUNTIME_POSTGRES_CANARY"
POSTGRES_IMAGE = "postgres@sha256:eb4759788a2182f08257135e61a34f2cfc3c2914079f3465d64ee62350f4d081"
POSTGRES_VERSION_NUM = "160014"
DATABASE = "teleo_canonical"
RUNTIME_ROLE = "leoclean_kb_runtime"
CONNECT_SENTINEL_DATABASE = "leoclean_connect_sentinel"
LOCAL_RUNTIME_PASSWORD = "disposable-r2-runtime-password"
IDENTIFIER_RE = re.compile(r"[a-z_][a-z0-9_]*\Z")


def _load_runtime_tool() -> ModuleType:
    spec = importlib.util.spec_from_file_location("leoclean_runtime_canary_contract", RUNTIME_TOOL)
    assert spec is not None and spec.loader is not None
    module = importlib.util.module_from_spec(spec)
    spec.loader.exec_module(module)
    return module


def test_ci_runs_digest_pinned_permission_canary_and_always_cleans_up() -> None:
    workflow_path = REPO_ROOT / ".github" / "workflows" / "ci.yml"
    workflow = yaml.safe_load(workflow_path.read_text(encoding="utf-8"))
    assert workflow["env"]["LEOCLEAN_RUNTIME_POSTGRES_IMAGE"] == POSTGRES_IMAGE

    steps = workflow["jobs"]["test"]["steps"]
    pull_step = next(
        step for step in steps if step.get("name") == "Pull digest-pinned PostgreSQL permission canary image"
    )
    pytest_step = next(step for step in steps if step.get("name") == "Pytest")
    cleanup_step = next(
        step for step in steps if step.get("name") == "Remove interrupted PostgreSQL permission canaries"
    )

    assert pull_step["run"] == 'docker pull "${LEOCLEAN_RUNTIME_POSTGRES_IMAGE}"'
    assert pytest_step["env"] == {RUN_ENV: "1"}
    assert "python -m pytest" in pytest_step["run"]
    assert cleanup_step["if"] == "always()"
    assert "label=livingip.r2.permission-canary" in cleanup_step["run"]
    assert "docker rm --force" in cleanup_step["run"]


def _quote_identifier(value: str) -> str:
    assert IDENTIFIER_RE.fullmatch(value), value
    return f'"{value}"'


def _run(
    command: list[str],
    *,
    input_text: str | None = None,
    timeout: int = 120,
) -> subprocess.CompletedProcess[str]:
    return subprocess.run(
        command,
        input=input_text,
        text=True,
        capture_output=True,
        check=False,
        timeout=timeout,
    )


def _require_success(completed: subprocess.CompletedProcess[str], action: str) -> str:
    if completed.returncode != 0:
        raise AssertionError(
            f"{action} failed with exit {completed.returncode}\n"
            f"stdout:\n{completed.stdout}\nstderr:\n{completed.stderr}"
        )
    return completed.stdout.strip()


def _psql(
    container: str,
    sql: str,
    *,
    role: str = "postgres",
    database: str = DATABASE,
    pgoptions: str | None = None,
) -> subprocess.CompletedProcess[str]:
    command = ["docker", "exec", "-i"]
    if pgoptions is not None:
        command.extend(("--env", f"PGOPTIONS={pgoptions}"))
    command.extend(
        [
            container,
            "psql",
            "--no-psqlrc",
            "--no-password",
            "--set=ON_ERROR_STOP=1",
            "--set=VERBOSITY=verbose",
            "--tuples-only",
            "--no-align",
            "--username",
            role,
            "--dbname",
            database,
        ]
    )
    return _run(command, input_text=sql)


def _assert_denied(container: str, sql: str, sqlstate: str = "42501") -> None:
    completed = _psql(container, sql, role=RUNTIME_ROLE)
    assert completed.returncode != 0, f"unexpected success for: {sql}"
    assert re.search(rf"(?:ERROR|FATAL):\s+{re.escape(sqlstate)}:", completed.stderr), (
        f"expected SQLSTATE {sqlstate} for {sql!r}\nstderr:\n{completed.stderr}"
    )


def _assert_database_connect_denied(container: str, database: str) -> None:
    completed = _psql(container, "select 1;", role=RUNTIME_ROLE, database=database)
    assert completed.returncode != 0, f"unexpected database connection success: {database}"
    assert f'permission denied for database "{database}"' in completed.stderr
    assert "does not have CONNECT privilege" in completed.stderr


def _bootstrap_provider_database_connect_residual(container: str) -> list[dict[str, object]]:
    _require_success(_psql(container, "create role cloudsqladmin nologin;"), "provider owner fixture")
    _require_success(
        _psql(container, "alter database template0 owner to cloudsqladmin;"),
        "provider template owner fixture",
    )
    _require_success(
        _psql(container, "create database cloudsqladmin owner cloudsqladmin template template0;"),
        "provider system database fixture",
    )
    _require_success(
        _psql(container, f"create database {CONNECT_SENTINEL_DATABASE};"),
        "operator-owned connect sentinel fixture",
    )
    return _read_provider_database_connect_residual(container)


def _read_provider_database_connect_residual(container: str) -> list[dict[str, object]]:
    output = _require_success(
        _psql(
            container,
            """
select coalesce(
         pg_catalog.jsonb_agg(
           pg_catalog.jsonb_build_object(
             'allow_connections', observed.allow_connections,
             'database', observed.database_name,
             'database_acl', observed.database_acl,
             'is_template', observed.is_template,
             'owner', observed.owner_name,
             'public_connect', observed.public_connect,
             'public_connect_acl_entries', observed.public_connect_acl_entries,
             'public_connect_grantable', observed.public_connect_grantable
           )
           order by observed.database_name
         ),
         '[]'::pg_catalog.jsonb
       )::text
  from (
    select database_row.datname as database_name,
           owner_role.rolname as owner_name,
           database_row.datallowconn as allow_connections,
           database_row.datistemplate as is_template,
           database_row.datacl::text as database_acl,
           pg_catalog.count(*) filter (
             where acl.grantee = 0
               and acl.privilege_type = 'CONNECT'
           )::integer as public_connect_acl_entries,
           coalesce(
             pg_catalog.bool_or(true) filter (
               where acl.grantee = 0
                 and acl.privilege_type = 'CONNECT'
             ),
             false
           ) as public_connect,
           coalesce(
             pg_catalog.bool_or(acl.is_grantable) filter (
               where acl.grantee = 0
                 and acl.privilege_type = 'CONNECT'
             ),
             false
           ) as public_connect_grantable
      from pg_catalog.pg_database database_row
      join pg_catalog.pg_roles owner_role on owner_role.oid = database_row.datdba
     cross join lateral pg_catalog.aclexplode(
       coalesce(
         database_row.datacl,
         pg_catalog.acldefault('d', database_row.datdba)
       )
     ) acl
     where database_row.datname in ('cloudsqladmin', 'template0')
     group by database_row.datname,
              owner_role.rolname,
              database_row.datallowconn,
              database_row.datistemplate,
              database_row.datacl
  ) observed;
""",
        ),
        "provider database-connect residual inventory",
    )
    return json.loads(output)


def _generated_read_tables_sql() -> str:
    statements: list[str] = []
    for schema, relation, columns in verifier.READ_COLUMN_ALLOWLIST:
        if (schema, relation) in {("public", "agents"), ("kb_stage", "kb_proposals")}:
            continue
        definitions = [f"{_quote_identifier(column)} text" for column in columns]
        definitions.append('"runtime_hidden_marker" text')
        statements.append(
            f"create table {_quote_identifier(schema)}.{_quote_identifier(relation)} ({', '.join(definitions)});"
        )
    return "\n".join(statements)


def _bootstrap_sql() -> str:
    # This fixture models the reviewed privilege topology, not production-schema parity.
    return f"""
create role kb_review nologin noinherit;
create role kb_apply nologin noinherit;
create schema kb_stage;

create table public.agents (
  id uuid primary key,
  handle text not null unique,
  runtime_hidden_marker text
);

{_generated_read_tables_sql()}

create table kb_stage.kb_proposals (
  id uuid primary key default gen_random_uuid(),
  proposal_type text not null,
  status text not null default 'pending_review',
  proposed_by_handle text,
  proposed_by_agent_id uuid,
  channel text not null default 'cli',
  source_ref text,
  rationale text not null,
  payload jsonb not null,
  reviewed_by_handle text,
  reviewed_by_agent_id uuid,
  reviewed_at timestamptz,
  review_note text,
  applied_by_handle text,
  applied_by_agent_id uuid,
  applied_at timestamptz,
  created_at timestamptz not null default now(),
  updated_at timestamptz not null default now(),
  constraint kb_proposals_proposal_type_check check (
    proposal_type in ('revise_claim', 'revise_strategy', 'add_edge')
  ),
  constraint kb_proposals_status_check check (
    status in ('pending_review', 'approved', 'rejected', 'applied', 'canceled')
  ),
  constraint kb_proposals_proposed_by_agent_id_fkey
    foreign key (proposed_by_agent_id) references public.agents(id),
  constraint kb_proposals_reviewed_by_agent_id_fkey
    foreign key (reviewed_by_agent_id) references public.agents(id),
  constraint kb_proposals_applied_by_agent_id_fkey
    foreign key (applied_by_agent_id) references public.agents(id)
);

create table public.private_notes (id uuid primary key, body text not null);

insert into public.agents (id, handle, runtime_hidden_marker)
values ('11111111-1111-4111-8111-111111111111', 'leo', 'not-runtime-readable');

create function kb_stage.approve_strict_proposal(uuid, text, jsonb, text, text)
returns jsonb language sql security definer set search_path = pg_catalog, pg_temp
as $function$ select '{{}}'::jsonb $function$;

create function kb_stage.assert_approved_proposal(
  uuid, text, jsonb, text, uuid, timestamptz, text
) returns void language plpgsql security definer set search_path = pg_catalog, pg_temp
as $function$ begin return; end; $function$;

create function kb_stage.finish_approved_proposal(
  uuid, text, jsonb, text, uuid, timestamptz, text, text
) returns jsonb language sql security definer set search_path = pg_catalog, pg_temp
as $function$ select '{{}}'::jsonb $function$;
"""


def _read_fingerprint(container: str) -> dict[str, int]:
    pairs: list[str] = []
    for schema, relation, columns in verifier.READ_COLUMN_ALLOWLIST:
        key = f"{schema}.{relation}"
        first_column = _quote_identifier(columns[0])
        qualified = f"{_quote_identifier(schema)}.{_quote_identifier(relation)}"
        pairs.extend((f"'{key}'", f"(select count({first_column}) from {qualified})"))
    sql = f"select jsonb_build_object({', '.join(pairs)})::text;"
    output = _require_success(_psql(container, sql, role=RUNTIME_ROLE), "read fingerprint")
    return json.loads(output)


def _provision(container: str, remote_sql_path: str) -> subprocess.CompletedProcess[str]:
    return _run(
        [
            "docker",
            "exec",
            "--interactive",
            container,
            "psql",
            "--no-psqlrc",
            "--no-password",
            "--set=ON_ERROR_STOP=1",
            "--set=VERBOSITY=verbose",
            "--username",
            "postgres",
            "--dbname",
            DATABASE,
            f"--file={remote_sql_path}",
        ],
        input_text=f"{LOCAL_RUNTIME_PASSWORD}\n{LOCAL_RUNTIME_PASSWORD}\n",
    )


def _assert_cloudsql_admin_attribute_compatibility(container: str) -> None:
    probe_role = "cloudsql_admin_probe"
    target_role = "cloudsql_attribute_target"
    _require_success(
        _psql(
            container,
            f"""
create role {probe_role}
  login createrole createdb nosuperuser noreplication nobypassrls;
create role {target_role}
  nologin nosuperuser nocreatedb nocreaterole noinherit noreplication nobypassrls;
grant {target_role} to {probe_role} with admin option;
""",
        ),
        "Cloud SQL administrator attribute fixture",
    )

    _require_success(
        _psql(
            container,
            f"""
alter role {target_role}
  with nologin nocreatedb nocreaterole noinherit connection limit 8;
alter role {target_role}
  with login nocreatedb nocreaterole noinherit connection limit 8;
""",
            role=probe_role,
        ),
        "Cloud SQL-compatible scoped attribute changes",
    )

    for forbidden_attribute in ("nosuperuser", "noreplication", "nobypassrls"):
        denied = _psql(
            container,
            f"alter role {target_role} with {forbidden_attribute};",
            role=probe_role,
        )
        assert denied.returncode != 0
        assert re.search(r"(?:ERROR|FATAL):\s+42501:", denied.stderr), denied.stderr


def _read_table_rows_state(container: str) -> dict[str, dict[str, object]]:
    table_output = _require_success(
        _psql(
            container,
            """
select namespace.nspname || '|' || relation.relname
  from pg_catalog.pg_class relation
  join pg_catalog.pg_namespace namespace on namespace.oid = relation.relnamespace
 where namespace.nspname in ('public', 'kb_stage')
   and relation.relkind in ('r', 'p')
 order by namespace.nspname, relation.relname;
""",
        ),
        "provisioning table inventory",
    )
    state: dict[str, dict[str, object]] = {}
    for line in table_output.splitlines():
        schema, relation = line.strip().split("|", 1)
        assert IDENTIFIER_RE.fullmatch(schema), schema
        assert IDENTIFIER_RE.fullmatch(relation), relation
        qualified = f"{_quote_identifier(schema)}.{_quote_identifier(relation)}"
        output = _require_success(
            _psql(
                container,
                f"""
select pg_catalog.jsonb_build_object(
         'row_count', pg_catalog.count(*),
         'rowset_md5', pg_catalog.md5(
           coalesce(
             pg_catalog.string_agg(
               pg_catalog.to_jsonb(row_data)::text,
               E'\\n'
               order by pg_catalog.to_jsonb(row_data)::text
             ),
             ''
           )
         )
       )::text
  from {qualified} as row_data;
""",
            ),
            f"row-state snapshot for {schema}.{relation}",
        )
        state[f"{schema}.{relation}"] = json.loads(output)
    return state


def _read_catalog_state(container: str) -> dict[str, object]:
    output = _require_success(
        _psql(
            container,
            """
with scoped_role_names(role_name) as (
  values ('leoclean_kb_runtime'), ('leoclean_kb_stage_owner')
), acl_entries(kind, object_name, grantor_name, grantee_name, privilege_type, is_grantable) as (
  select 'database',
         database_row.datname,
         grantor_role.rolname,
         case when acl.grantee = 0 then 'PUBLIC' else grantee_role.rolname end,
         acl.privilege_type,
         acl.is_grantable
    from pg_catalog.pg_database database_row
   cross join lateral pg_catalog.aclexplode(
     coalesce(
       database_row.datacl,
       pg_catalog.acldefault('d', database_row.datdba)
     )
   ) acl
    left join pg_catalog.pg_roles grantor_role on grantor_role.oid = acl.grantor
    left join pg_catalog.pg_roles grantee_role on grantee_role.oid = acl.grantee
  union all
  select 'schema',
         namespace.nspname,
         grantor_role.rolname,
         case when acl.grantee = 0 then 'PUBLIC' else grantee_role.rolname end,
         acl.privilege_type,
         acl.is_grantable
    from pg_catalog.pg_namespace namespace
   cross join lateral pg_catalog.aclexplode(
     coalesce(namespace.nspacl, pg_catalog.acldefault('n', namespace.nspowner))
   ) acl
    left join pg_catalog.pg_roles grantor_role on grantor_role.oid = acl.grantor
    left join pg_catalog.pg_roles grantee_role on grantee_role.oid = acl.grantee
   where namespace.nspname in ('public', 'kb_stage')
  union all
  select case when relation.relkind = 'S' then 'sequence' else 'relation' end,
         pg_catalog.format('%I.%I', namespace.nspname, relation.relname),
         grantor_role.rolname,
         case when acl.grantee = 0 then 'PUBLIC' else grantee_role.rolname end,
         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(case when relation.relkind = 'S' then 'S'::"char" else 'r'::"char" end, relation.relowner)
     )
   ) acl
    left join pg_catalog.pg_roles grantor_role on grantor_role.oid = acl.grantor
    left join pg_catalog.pg_roles grantee_role on grantee_role.oid = acl.grantee
   where namespace.nspname in ('public', 'kb_stage')
     and relation.relkind in ('r', 'p', 'v', 'm', 'f', 'S')
  union all
  select 'column',
         pg_catalog.format('%I.%I.%I', namespace.nspname, relation.relname, attribute.attname),
         grantor_role.rolname,
         case when acl.grantee = 0 then 'PUBLIC' else grantee_role.rolname end,
         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
    left join pg_catalog.pg_roles grantor_role on grantor_role.oid = acl.grantor
    left join pg_catalog.pg_roles grantee_role on grantee_role.oid = acl.grantee
   where namespace.nspname in ('public', 'kb_stage')
     and attribute.attnum > 0
     and not attribute.attisdropped
  union all
  select 'routine',
         pg_catalog.format(
           '%I.%I(%s)',
           namespace.nspname,
           function_row.proname,
           pg_catalog.pg_get_function_identity_arguments(function_row.oid)
         ),
         grantor_role.rolname,
         case when acl.grantee = 0 then 'PUBLIC' else grantee_role.rolname end,
         acl.privilege_type,
         acl.is_grantable
    from pg_catalog.pg_proc function_row
    join pg_catalog.pg_namespace namespace on namespace.oid = function_row.pronamespace
   cross join lateral pg_catalog.aclexplode(
     coalesce(function_row.proacl, pg_catalog.acldefault('f', function_row.proowner))
   ) acl
    left join pg_catalog.pg_roles grantor_role on grantor_role.oid = acl.grantor
    left join pg_catalog.pg_roles grantee_role on grantee_role.oid = acl.grantee
   where namespace.nspname in ('public', 'kb_stage')
  union all
  select 'large_object',
         metadata.oid::text,
         grantor_role.rolname,
         case when acl.grantee = 0 then 'PUBLIC' else grantee_role.rolname end,
         acl.privilege_type,
         acl.is_grantable
    from pg_catalog.pg_largeobject_metadata metadata
   cross join lateral pg_catalog.aclexplode(
     coalesce(metadata.lomacl, pg_catalog.acldefault('L', metadata.lomowner))
   ) acl
    left join pg_catalog.pg_roles grantor_role on grantor_role.oid = acl.grantor
    left join pg_catalog.pg_roles grantee_role on grantee_role.oid = acl.grantee
  union all
  select 'parameter',
         parameter_acl.parname,
         grantor_role.rolname,
         case when acl.grantee = 0 then 'PUBLIC' else grantee_role.rolname end,
         acl.privilege_type,
         acl.is_grantable
    from pg_catalog.pg_parameter_acl parameter_acl
   cross join lateral pg_catalog.aclexplode(parameter_acl.paracl) acl
    left join pg_catalog.pg_roles grantor_role on grantor_role.oid = acl.grantor
    left join pg_catalog.pg_roles grantee_role on grantee_role.oid = acl.grantee
), role_state as (
  select coalesce(
           pg_catalog.jsonb_agg(
             pg_catalog.jsonb_build_object(
               'role', role_row.rolname,
               'can_login', role_row.rolcanlogin,
               'is_superuser', role_row.rolsuper,
               'can_create_db', role_row.rolcreatedb,
               'can_create_role', role_row.rolcreaterole,
               'inherits_privileges', role_row.rolinherit,
               'can_replicate', role_row.rolreplication,
               'bypasses_rls', role_row.rolbypassrls,
               'connection_limit', role_row.rolconnlimit,
               'role_config', role_row.rolconfig,
               'password_is_null', auth_row.rolpassword is null,
               'password_is_scram', coalesce(auth_row.rolpassword like 'SCRAM-SHA-256$%', false)
             )
             order by role_row.rolname
           ),
           '[]'::pg_catalog.jsonb
         ) as value
    from pg_catalog.pg_roles role_row
    join pg_catalog.pg_authid auth_row on auth_row.oid = role_row.oid
    join scoped_role_names scoped on scoped.role_name = role_row.rolname
), database_role_settings as (
  select coalesce(
           pg_catalog.jsonb_agg(
             pg_catalog.jsonb_build_object(
               'role', role_row.rolname,
               'database', database_row.datname,
               'setting', setting_value.setting
             )
             order by role_row.rolname, database_row.datname, setting_value.setting
           ),
           '[]'::pg_catalog.jsonb
         ) as value
    from pg_catalog.pg_db_role_setting setting_row
    join pg_catalog.pg_roles role_row on role_row.oid = setting_row.setrole
    join scoped_role_names scoped on scoped.role_name = role_row.rolname
    left join pg_catalog.pg_database database_row on database_row.oid = setting_row.setdatabase
   cross join lateral pg_catalog.unnest(setting_row.setconfig) setting_value(setting)
), membership_state as (
  select coalesce(
           pg_catalog.jsonb_agg(
             pg_catalog.jsonb_build_object(
               'granted_role', granted_role.rolname,
               'member_role', member_role.rolname,
               'grantor_role', grantor_role.rolname,
               'admin_option', membership.admin_option,
               'inherit_option', membership.inherit_option,
               'set_option', membership.set_option
             )
             order by granted_role.rolname, member_role.rolname, grantor_role.rolname
           ),
           '[]'::pg_catalog.jsonb
         ) as value
    from pg_catalog.pg_auth_members membership
    join pg_catalog.pg_roles granted_role on granted_role.oid = membership.roleid
    join pg_catalog.pg_roles member_role on member_role.oid = membership.member
    join pg_catalog.pg_roles grantor_role on grantor_role.oid = membership.grantor
   where granted_role.rolname in (select role_name from scoped_role_names)
      or member_role.rolname in (select role_name from scoped_role_names)
), ownership_state as (
  select coalesce(
           pg_catalog.jsonb_agg(
             pg_catalog.jsonb_build_object(
               'kind', owned.kind,
               'object', owned.object_name,
               'owner', owned.owner_name
             )
             order by owned.kind, owned.object_name, owned.owner_name
           ),
           '[]'::pg_catalog.jsonb
         ) as value
    from (
      select 'relation' as kind,
             pg_catalog.format('%I.%I', namespace.nspname, relation.relname) as object_name,
             owner_role.rolname as owner_name
        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 in ('public', 'kb_stage')
      union all
      select 'routine',
             pg_catalog.format(
               '%I.%I(%s)',
               namespace.nspname,
               function_row.proname,
               pg_catalog.pg_get_function_identity_arguments(function_row.oid)
             ),
             owner_role.rolname
        from pg_catalog.pg_proc function_row
        join pg_catalog.pg_namespace namespace on namespace.oid = function_row.pronamespace
        join pg_catalog.pg_roles owner_role on owner_role.oid = function_row.proowner
       where namespace.nspname in ('public', 'kb_stage')
    ) owned
), normalized_acls as (
  select coalesce(
           pg_catalog.jsonb_agg(
             pg_catalog.jsonb_build_object(
               'kind', entries.kind,
               'object', entries.object_name,
               'grantor', entries.grantor_name,
               'grantee', entries.grantee_name,
               'privilege', entries.privilege_type,
               'grantable', entries.is_grantable
             )
             order by entries.kind,
                      entries.object_name,
                      entries.grantor_name,
                      entries.grantee_name,
                      entries.privilege_type,
                      entries.is_grantable
           ),
           '[]'::pg_catalog.jsonb
         ) as value
    from acl_entries entries
)
select pg_catalog.jsonb_build_object(
         'roles', (select value from role_state),
         'database_role_settings', (select value from database_role_settings),
         'memberships', (select value from membership_state),
         'ownership', (select value from ownership_state),
         'acls', (select value from normalized_acls),
         'stage_function', pg_catalog.jsonb_build_object(
           'owner', (
             select owner_role.rolname
               from pg_catalog.pg_proc function_row
               join pg_catalog.pg_roles owner_role on owner_role.oid = function_row.proowner
              where function_row.oid = 'kb_stage.stage_leoclean_proposal(text,text,text,text,jsonb)'::pg_catalog.regprocedure
           ),
           'runtime_execute', pg_catalog.has_function_privilege(
             'leoclean_kb_runtime',
             'kb_stage.stage_leoclean_proposal(text,text,text,text,jsonb)',
             'EXECUTE'
           ),
           'owner_execute', pg_catalog.has_function_privilege(
             'leoclean_kb_stage_owner',
             'kb_stage.stage_leoclean_proposal(text,text,text,text,jsonb)',
             'EXECUTE'
           )
         )
       )::text;
""",
        ),
        "provisioning catalog-state snapshot",
    )
    return json.loads(output)


def _read_provisioning_state(container: str) -> dict[str, object]:
    return {
        "rows": _read_table_rows_state(container),
        "catalog": _read_catalog_state(container),
    }


def _state_with_runtime_login(state: dict[str, object], *, can_login: bool) -> dict[str, object]:
    adjusted = json.loads(json.dumps(state))
    roles = adjusted["catalog"]["roles"]
    runtime = next(role for role in roles if role["role"] == RUNTIME_ROLE)
    runtime["can_login"] = can_login
    return adjusted


def _assert_exact_read_contract(container: str) -> None:
    for schema, relation, columns in verifier.READ_COLUMN_ALLOWLIST:
        qualified = f"{_quote_identifier(schema)}.{_quote_identifier(relation)}"
        projection = ", ".join(_quote_identifier(column) for column in columns)
        _require_success(
            _psql(container, f"select {projection} from {qualified} limit 0;", role=RUNTIME_ROLE),
            f"allowed projection {schema}.{relation}",
        )

        _assert_denied(container, f"select * from {qualified} limit 0;")
        denied_column = (
            "reviewed_by_agent_id" if (schema, relation) == ("kb_stage", "kb_proposals") else "runtime_hidden_marker"
        )
        _assert_denied(container, f"select {_quote_identifier(denied_column)} from {qualified} limit 0;")


def _assert_proposal_function_and_rollback(container: str, runtime_tool: ModuleType) -> None:
    crafted_rationale = "permission canary\\'; select pg_catalog.lo_from_bytea(0, ''::bytea); --"
    crafted_rationale_sql = runtime_tool.sql_literal(crafted_rationale)
    calls = []
    for proposal_type in ("revise_claim", "revise_strategy", "add_edge"):
        rationale = crafted_rationale_sql if proposal_type == "revise_claim" else "'permission canary'"
        calls.append(
            "select concat_ws('|', staged->>'proposal_type', staged->>'status', "
            "staged->>'proposed_by_handle', staged->>'channel') "
            "from (select kb_stage.stage_leoclean_proposal("
            f"'{proposal_type}', '', 'r2:{proposal_type}', {rationale}, "
            "'{\"canary\": true}'::jsonb) as staged) probe;"
        )
    sql = "\n".join(
        [
            "begin;",
            "select 'strings=' || current_setting('standard_conforming_strings');",
            "select 'readonly=' || current_setting('transaction_read_only');",
            *calls,
            f"select 'crafted=' || count(*) from kb_stage.kb_proposals where rationale = {crafted_rationale_sql};",
            "select 'inside=' || count(id) from kb_stage.kb_proposals;",
            "rollback;",
            "select 'after=' || count(id) from kb_stage.kb_proposals;",
        ]
    )
    output = _require_success(
        _psql(
            container,
            sql,
            role=RUNTIME_ROLE,
            pgoptions=runtime_tool.RUNTIME_STANDARD_STRINGS_PGOPTIONS,
        ),
        "proposal function rollback",
    )
    lines = [line.strip() for line in output.splitlines() if line.strip()]
    assert "strings=on" in lines
    assert "readonly=off" in lines
    assert "revise_claim|pending_review|leo|telegram" in lines
    assert "revise_strategy|pending_review|leo|telegram" in lines
    assert "add_edge|pending_review|leo|telegram" in lines
    assert "crafted=1" in lines
    assert "inside=3" in lines
    assert "after=0" in lines


def _assert_write_and_escalation_denials(container: str) -> None:
    direct_writes = (
        "insert into public.claims (id) values ('direct-canonical-write');",
        "update public.claims set runtime_hidden_marker = 'x' where false;",
        "delete from public.claims where false;",
        "truncate public.claims;",
        "insert into kb_stage.kb_proposals (proposal_type, rationale, payload) values ('revise_claim', 'direct-stage-write', '{}'::jsonb);",
        "update kb_stage.kb_proposals set rationale = 'x' where false;",
        "delete from kb_stage.kb_proposals where false;",
        "truncate kb_stage.kb_proposals;",
    )
    for sql in direct_writes:
        _assert_denied(container, sql)

    escalation_attempts = (
        "set lo_compat_privileges = on;",
        "set role leoclean_kb_stage_owner;",
        "set role kb_review;",
        "set role kb_apply;",
        "set role postgres;",
        "create role r2_escape;",
        "set session authorization postgres;",
        "select kb_stage.approve_strict_proposal('00000000-0000-0000-0000-000000000000', 'leo', '{}'::jsonb, 'approve', 'note');",
        "select kb_stage.assert_approved_proposal('00000000-0000-0000-0000-000000000000', 'revise_claim', '{}'::jsonb, 'leo', '11111111-1111-4111-8111-111111111111', now(), 'note');",
        "select kb_stage.finish_approved_proposal('00000000-0000-0000-0000-000000000000', 'revise_claim', '{}'::jsonb, 'leo', '11111111-1111-4111-8111-111111111111', now(), 'note', 'applied');",
    )
    for sql in escalation_attempts:
        _assert_denied(container, sql)

    _assert_denied(
        container,
        "select kb_stage.stage_leoclean_proposal('leo', 'revise_claim', 'telegram', 'r2', 'forged', '{}'::jsonb);",
        "42883",
    )
    _assert_denied(container, "select body from public.private_notes;", "42501")


def _assert_large_object_provider_residual_rolls_back(container: str) -> None:
    inventory_sql = """
select pg_catalog.format(
         '%I(%s)|%s|%s',
         function_row.proname,
         pg_catalog.pg_get_function_identity_arguments(function_row.oid),
         exists (
           select 1
             from pg_catalog.aclexplode(
               coalesce(function_row.proacl, pg_catalog.acldefault('f', function_row.proowner))
             ) acl
            where acl.grantee = 0
              and acl.privilege_type = 'EXECUTE'
         ),
         pg_catalog.has_function_privilege(current_user, function_row.oid, 'EXECUTE')
       )
  from pg_catalog.pg_proc function_row
  join pg_catalog.pg_namespace namespace on namespace.oid = function_row.pronamespace
 where namespace.nspname = 'pg_catalog'
   and function_row.proname in (
     'lo_creat', 'lo_create', 'lo_export', 'lo_from_bytea', 'lo_import',
     'lo_open', 'lo_put', 'lo_truncate', 'lo_truncate64', 'lo_unlink', 'lowrite'
   )
 order by function_row.proname, pg_catalog.pg_get_function_identity_arguments(function_row.oid);
"""
    observed = _require_success(_psql(container, inventory_sql, role=RUNTIME_ROLE), "large-object residual inventory")
    expected = "\n".join(
        f"{signature}|{'t' if public_execute else 'f'}|{'t' if public_execute else 'f'}"
        for signature, public_execute in verifier.LARGE_OBJECT_MUTATION_ROUTINE_RESIDUAL
    )
    assert observed == expected

    probe_sql = """
select 'before=' || count(*)
  from pg_catalog.pg_largeobject_metadata metadata
 where metadata.lomowner = (select oid from pg_catalog.pg_roles where rolname = current_user);
begin;
select 'created=' || (pg_catalog.lo_from_bytea(0, ''::bytea) <> 0);
select 'inside=' || count(*)
  from pg_catalog.pg_largeobject_metadata metadata
 where metadata.lomowner = (select oid from pg_catalog.pg_roles where rolname = current_user);
rollback;
select 'after=' || count(*)
  from pg_catalog.pg_largeobject_metadata metadata
 where metadata.lomowner = (select oid from pg_catalog.pg_roles where rolname = current_user);
"""
    output = _require_success(_psql(container, probe_sql, role=RUNTIME_ROLE), "rolled-back large-object residual")
    lines = [
        line.strip()
        for line in output.splitlines()
        if line.strip() and line.strip() not in {"BEGIN", "ROLLBACK"}
    ]
    assert lines == ["before=0", "created=true", "inside=1", "after=0"]


def _assert_runtime_read_only_guard(container: str, runtime_tool: ModuleType) -> None:
    read_only_options = runtime_tool.RUNTIME_READ_ONLY_PGOPTIONS
    owner_count_sql = f"""
select count(*)
  from pg_catalog.pg_largeobject_metadata metadata
 where metadata.lomowner = (select oid from pg_catalog.pg_roles where rolname = '{RUNTIME_ROLE}');
"""
    assert _require_success(_psql(container, owner_count_sql), "large-object count before read guard") == "0"
    assert (
        _require_success(
            _psql(
                container,
                "show standard_conforming_strings;",
                role=RUNTIME_ROLE,
                pgoptions=read_only_options,
            ),
            "runtime string parsing guard",
        )
        == "on"
    )
    assert (
        _require_success(
            _psql(
                container,
                "show transaction_read_only;",
                role=RUNTIME_ROLE,
                pgoptions=read_only_options,
            ),
            "runtime read-only transaction guard",
        )
        == "on"
    )
    _require_success(
        _psql(
            container,
            "select id, text from public.claims limit 0;",
            role=RUNTIME_ROLE,
            pgoptions=read_only_options,
        ),
        "allowed read through runtime read-only guard",
    )
    denied = _psql(
        container,
        "select pg_catalog.lo_from_bytea(0, ''::bytea);",
        role=RUNTIME_ROLE,
        pgoptions=read_only_options,
    )
    assert denied.returncode != 0
    assert re.search(r"ERROR:\s+25006:", denied.stderr), denied.stderr
    assert _require_success(_psql(container, owner_count_sql), "large-object count after read guard") == "0"


@pytest.mark.skipif(os.environ.get(RUN_ENV) != "1", reason=f"set {RUN_ENV}=1 for the local PG16 canary")
def test_digest_pinned_network_disabled_postgres_16_14_permission_contract(tmp_path: Path) -> None:
    assert shutil.which("docker") is not None, "Docker is required for the disposable permission canary"
    runtime_tool = _load_runtime_tool()

    image_output = _require_success(
        _run(["docker", "image", "inspect", "--format", "{{.Id}}|{{json .RepoDigests}}", POSTGRES_IMAGE]),
        "pinned image inspection",
    )
    image_id, repo_digests_json = image_output.split("|", 1)
    assert re.fullmatch(r"sha256:[0-9a-f]{64}", image_id)
    assert POSTGRES_IMAGE in json.loads(repo_digests_json)

    run_token = uuid.uuid4().hex[:12]
    container = f"leoclean-r2-{run_token}"
    label = f"livingip.r2.permission-canary={run_token}"
    cleanup_error: str | None = None
    try:
        started = _run(
            [
                "docker",
                "run",
                "--detach",
                "--name",
                container,
                "--label",
                label,
                "--pull=never",
                "--network=none",
                "--tmpfs",
                "/var/lib/postgresql/data:rw,noexec,nosuid,nodev,size=256m",
                "--env",
                "POSTGRES_DB=teleo_canonical",
                "--env",
                "POSTGRES_HOST_AUTH_METHOD=trust",
                POSTGRES_IMAGE,
            ]
        )
        _require_success(started, "disposable PostgreSQL start")

        inspect = json.loads(_require_success(_run(["docker", "inspect", container]), "container inspection"))[0]
        assert inspect["Image"] == image_id
        assert inspect["HostConfig"]["NetworkMode"] == "none"
        assert not inspect["HostConfig"].get("PortBindings")
        assert not inspect["HostConfig"].get("Binds")
        tmpfs_spec = inspect["HostConfig"]["Tmpfs"]["/var/lib/postgresql/data"]
        assert {"rw", "noexec", "nosuid", "nodev"}.issubset(tmpfs_spec.split(","))
        assert any(size in tmpfs_spec.split(",") for size in ("size=256m", "size=268435456"))
        assert all(mount["Type"] not in {"bind", "volume"} for mount in inspect["Mounts"])

        last_startup_probe: subprocess.CompletedProcess[str] | None = None
        for _ in range(120):
            final_server = _run(
                [
                    "docker",
                    "exec",
                    container,
                    "sh",
                    "-c",
                    'test "$(cat /proc/1/comm)" = postgres',
                ],
                timeout=5,
            )
            if final_server.returncode == 0:
                version = _psql(container, "show server_version_num;")
                last_startup_probe = version
                if version.returncode == 0 and version.stdout.strip() == POSTGRES_VERSION_NUM:
                    break
            else:
                last_startup_probe = final_server
            time.sleep(0.25)
        else:
            logs = _run(["docker", "logs", container], timeout=10)
            probe_output = ""
            if last_startup_probe is not None:
                probe_output = f"\nlast probe stdout:\n{last_startup_probe.stdout}\nlast probe stderr:\n{last_startup_probe.stderr}"
            raise AssertionError(
                f"final PostgreSQL server did not report version {POSTGRES_VERSION_NUM}"
                f"{probe_output}\ncontainer logs:\n{logs.stdout}\n{logs.stderr}"
            )

        provider_residual_before = _bootstrap_provider_database_connect_residual(container)
        expected_provider_core = [
            {key: value for key, value in row.items() if key not in {"runtime_connect", "stage_owner_connect"}}
            for row in verifier.PROVIDER_DATABASE_CONNECT_RESIDUAL
        ]
        assert [
            {key: value for key, value in row.items() if key != "database_acl"}
            for row in provider_residual_before
        ] == expected_provider_core
        assert provider_residual_before[0]["database_acl"] is None
        _require_success(_psql(container, "create database teleo_kb;"), "legacy database fixture")
        _require_success(_psql(container, _bootstrap_sql()), "permission fixture bootstrap")
        _assert_cloudsql_admin_attribute_compatibility(container)
        _require_success(
            _run(["docker", "cp", str(ROLE_SQL), f"{container}:/tmp/gcp_leoclean_runtime_role.sql"]),
            "role SQL copy",
        )

        failure_anchor = "\n-- LOGIN is the last state change"
        role_sql = ROLE_SQL.read_text(encoding="utf-8")
        assert role_sql.count(failure_anchor) == 1
        failure_sql = role_sql.replace(
            failure_anchor,
            "\ngrant update on public.private_notes to leoclean_kb_runtime;"
            "\nselect 1 / 0 as injected_mid_provision_failure;"
            f"{failure_anchor}",
        )
        failure_path = tmp_path / "gcp_leoclean_runtime_role_injected_failure.sql"
        failure_path.write_text(failure_sql, encoding="utf-8")
        _require_success(
            _run(["docker", "cp", str(failure_path), f"{container}:/tmp/gcp_leoclean_runtime_role_failure.sql"]),
            "failure-injected role SQL copy",
        )

        _require_success(
            _psql(container, "alter database teleo_canonical set lo_compat_privileges = on;"),
            "unsafe large-object compatibility fixture",
        )
        unsafe_compat = _provision(container, "/tmp/gcp_leoclean_runtime_role.sql")
        assert unsafe_compat.returncode != 0
        assert "lo_compat_privileges must remain off" in unsafe_compat.stderr
        assert _require_success(
            _psql(container, f"select rolcanlogin from pg_catalog.pg_roles where rolname = '{RUNTIME_ROLE}';"),
            "unsafe compatibility refence",
        ) == "f"
        _require_success(
            _psql(container, "alter database teleo_canonical reset lo_compat_privileges;"),
            "large-object compatibility recovery",
        )

        _require_success(
            _psql(container, f"grant execute on function pg_catalog.lo_create(oid) to {RUNTIME_ROLE};"),
            "direct large-object ACL fixture",
        )
        direct_acl = _provision(container, "/tmp/gcp_leoclean_runtime_role.sql")
        assert direct_acl.returncode != 0
        assert "scoped Leo roles have direct large-object routine ACLs" in direct_acl.stderr
        assert _require_success(
            _psql(container, f"select rolcanlogin from pg_catalog.pg_roles where rolname = '{RUNTIME_ROLE}';"),
            "direct ACL refence",
        ) == "f"
        _require_success(
            _psql(container, f"revoke execute on function pg_catalog.lo_create(oid) from {RUNTIME_ROLE};"),
            "direct large-object ACL recovery",
        )

        _require_success(
            _psql(
                container,
                "create role provider_acl_probe nologin; "
                "grant connect on database template0 to provider_acl_probe;",
            ),
            "provider extra-principal ACL fixture",
        )
        provider_acl_drift = _provision(container, "/tmp/gcp_leoclean_runtime_role.sql")
        assert provider_acl_drift.returncode != 0
        assert "provider databases retain unexpected ACL entries" in provider_acl_drift.stderr
        assert _require_success(
            _psql(container, f"select rolcanlogin from pg_catalog.pg_roles where rolname = '{RUNTIME_ROLE}';"),
            "provider ACL drift role fence",
        ) == "f"
        _require_success(
            _psql(
                container,
                "revoke connect on database template0 from provider_acl_probe; "
                "drop role provider_acl_probe;",
            ),
            "provider extra-principal ACL recovery",
        )

        _require_success(
            _psql(container, "alter database teleo_canonical set session_preload_libraries = 'auto_explain';"),
            "preload-library drift fixture",
        )
        preload_drift = _provision(container, "/tmp/gcp_leoclean_runtime_role.sql")
        assert preload_drift.returncode != 0
        assert "session and local preload libraries must remain empty" in preload_drift.stderr
        assert _require_success(
            _psql(container, f"select rolcanlogin from pg_catalog.pg_roles where rolname = '{RUNTIME_ROLE}';"),
            "preload-library drift role fence",
        ) == "f"
        _require_success(
            _psql(container, "alter database teleo_canonical reset session_preload_libraries;"),
            "preload-library drift recovery",
        )

        _require_success(
            _psql(container, "alter database cloudsqladmin allow_connections false;"),
            "provider residual drift fixture",
        )
        provider_drift = _provision(container, "/tmp/gcp_leoclean_runtime_role.sql")
        assert provider_drift.returncode != 0
        assert "Cloud SQL provider database CONNECT residual drifted" in provider_drift.stderr
        assert _require_success(
            _psql(container, f"select rolcanlogin from pg_catalog.pg_roles where rolname = '{RUNTIME_ROLE}';"),
            "provider residual drift role fence",
        ) == "f"
        _require_success(
            _psql(container, "alter database cloudsqladmin allow_connections true;"),
            "provider residual drift recovery",
        )

        successful_states: list[dict[str, object]] = []
        for provision_number in range(1, 4):
            _require_success(
                _provision(container, "/tmp/gcp_leoclean_runtime_role.sql"),
                f"runtime role provisioning run {provision_number}",
            )
            state = _read_provisioning_state(container)
            assert state["catalog"]["stage_function"] == {
                "owner": "leoclean_kb_stage_owner",
                "owner_execute": False,
                "runtime_execute": True,
            }
            assert state["catalog"]["memberships"] == []
            role_state = {role["role"]: role for role in state["catalog"]["roles"]}
            assert role_state[RUNTIME_ROLE]["password_is_scram"] is True
            assert role_state[RUNTIME_ROLE]["password_is_null"] is False
            assert role_state["leoclean_kb_stage_owner"]["password_is_scram"] is False
            assert role_state["leoclean_kb_stage_owner"]["password_is_null"] is True
            _assert_large_object_provider_residual_rolls_back(container)
            assert _read_provider_database_connect_residual(container) == provider_residual_before
            successful_states.append(state)

        assert successful_states[1:] == successful_states[:1] * 2
        stable_state = successful_states[0]

        _require_success(
            _psql(container, f"grant connect on database {CONNECT_SENTINEL_DATABASE} to public;"),
            "rollback database ACL fixture",
        )
        pre_failure_state = _read_provisioning_state(container)
        failed = _provision(container, "/tmp/gcp_leoclean_runtime_role_failure.sql")
        assert failed.returncode != 0
        assert re.search(r"ERROR:\s+22012:", failed.stderr)
        assert _read_provisioning_state(container) == _state_with_runtime_login(pre_failure_state, can_login=False)
        assert _read_provider_database_connect_residual(container) == provider_residual_before
        disabled_login = _psql(container, "select 1;", role=RUNTIME_ROLE)
        assert disabled_login.returncode != 0
        assert f'role "{RUNTIME_ROLE}" is not permitted to log in' in disabled_login.stderr

        _require_success(
            _provision(container, "/tmp/gcp_leoclean_runtime_role.sql"),
            "runtime role recovery provisioning",
        )
        assert _read_provisioning_state(container) == stable_state

        identity = _require_success(
            _psql(container, "select current_user || '|' || current_database();", role=RUNTIME_ROLE),
            "runtime identity",
        )
        assert identity == f"{RUNTIME_ROLE}|{DATABASE}"

        for denied_database in ("postgres", "teleo_kb", "template1", CONNECT_SENTINEL_DATABASE):
            _assert_database_connect_denied(container, denied_database)
        provider_identity = _require_success(
            _psql(container, "select current_user || '|' || current_database();", role=RUNTIME_ROLE, database="cloudsqladmin"),
            "provider-owned residual connection",
        )
        assert provider_identity == f"{RUNTIME_ROLE}|cloudsqladmin"

        stage_definition = json.loads(
            _require_success(
                _psql(container, verifier._stage_function_definition_sql(), role=RUNTIME_ROLE),
                "external verifier stage-function query",
            )
        )
        assert verifier._assert_stage_function_definition(stage_definition)["owner_execute"] is False

        catalog_output = _require_success(
            _psql(container, verifier._catalog_privilege_posture_sql(), role=RUNTIME_ROLE),
            "external verifier catalog query",
        )
        assert len(catalog_output.splitlines()) == 1, repr(catalog_output)
        _parsed, parsed_end = json.JSONDecoder().raw_decode(catalog_output)
        assert not catalog_output[parsed_end:].strip(), repr(catalog_output[parsed_end : parsed_end + 200])
        catalog_posture = json.loads(catalog_output)
        verified_catalog = verifier._assert_catalog_privilege_posture(catalog_posture)
        assert verified_catalog["provider_database_connect_residual"] == [
            dict(row) for row in verifier.PROVIDER_DATABASE_CONNECT_RESIDUAL
        ]

        before = _read_fingerprint(container)
        assert before == {
            f"{schema}.{relation}": (1 if (schema, relation) == ("public", "agents") else 0)
            for schema, relation, _columns in verifier.READ_COLUMN_ALLOWLIST
        }

        _assert_exact_read_contract(container)
        _assert_runtime_read_only_guard(container, runtime_tool)
        _assert_proposal_function_and_rollback(container, runtime_tool)
        _assert_write_and_escalation_denials(container)

        _require_success(
            _psql(container, "alter table public.claims add column future_private_text text;"),
            "future-column sentinel",
        )
        _assert_denied(container, "select future_private_text from public.claims limit 0;")
        assert _read_fingerprint(container) == before
    finally:
        removed = _run(["docker", "rm", "--force", container], timeout=30)
        if removed.returncode != 0 and "No such container" not in f"{removed.stdout}\n{removed.stderr}":
            cleanup_error = removed.stderr or removed.stdout
        leftovers = _run(
            ["docker", "ps", "--all", "--filter", f"label={label}", "--format", "{{.Names}}"],
            timeout=10,
        )
        if leftovers.returncode != 0 or leftovers.stdout.strip():
            cleanup_error = cleanup_error or leftovers.stderr or leftovers.stdout
        if cleanup_error is not None:
            raise AssertionError(f"disposable PostgreSQL cleanup failed: {cleanup_error}")