from __future__ import annotations

import json
import os
import re
import shutil
import subprocess
import time
import uuid
from pathlib import Path

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"
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"
LOCAL_RUNTIME_PASSWORD = "disposable-r2-runtime-password"
IDENTIFIER_RE = re.compile(r"[a-z_][a-z0-9_]*\Z")


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,
) -> subprocess.CompletedProcess[str]:
    return _run(
        [
            "docker",
            "exec",
            "-i",
            container,
            "psql",
            "--no-psqlrc",
            "--no-password",
            "--set=ON_ERROR_STOP=1",
            "--set=VERBOSITY=verbose",
            "--tuples-only",
            "--no-align",
            "--username",
            role,
            "--dbname",
            database,
        ],
        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 _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 _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) -> None:
    calls = []
    for proposal_type in ("revise_claim", "revise_strategy", "add_edge"):
        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}', 'permission canary', "
            "'{\"canary\": true}'::jsonb) as staged) probe;"
        )
    sql = "\n".join(
        [
            "begin;",
            *calls,
            "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), "proposal function rollback")
    lines = [line.strip() for line in output.splitlines() if line.strip()]
    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 "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 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")


@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"

    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}"
            )

        _require_success(_psql(container, "create database teleo_kb;"), "legacy database fixture")
        _require_success(_psql(container, _bootstrap_sql()), "permission fixture bootstrap")
        _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",
        )

        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
            successful_states.append(state)

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

        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(stable_state, can_login=False)
        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}"

        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

        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_proposal_function_and_rollback(container)
        _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}")