#!/usr/bin/env python3
"""Verify the minimum V3 epistemic contract in disposable PostgreSQL 16.

The verifier accepts no database address or credentials. It creates its own
networkless, tmpfs-backed Docker container, installs the additive migration,
round-trips proposal-linked epistemic rows, exercises adversarial failures,
rolls the contract back, reinstalls it, rolls it back again, and removes the
container. Production and persistent databases are unreachable by design.
"""

from __future__ import annotations

import argparse
import hashlib
import json
import os
import re
import secrets
import shutil
import subprocess
import time
import uuid
from pathlib import Path
from typing import Any

REPO_ROOT = Path(__file__).resolve().parents[1]
MIGRATION_SQL = REPO_ROOT / "ops" / "teleo_v3_epistemic_contract.sql"
ROLLBACK_SQL = REPO_ROOT / "ops" / "rollback_teleo_v3_epistemic_contract.sql"

RECEIPT_SCHEMA = "livingip.teleoV3EpistemicContractReceipt.v1"
POSTGRES_IMAGE = os.environ.get("LEOCLEAN_RUNTIME_POSTGRES_IMAGE", "postgres:16-alpine")
DATABASE = "teleo_v3_contract"
CONTAINER_LABEL = "livingip.canary=teleo-v3-epistemic-contract"
TRIGGER_CONTRACT_PATTERN = re.compile(
    r"\$trigger_contract\$\s*(\[.*?\])\s*\$trigger_contract\$",
    re.DOTALL,
)
CATALOG_LOCK_FUNCTION_NAME = "teleo_v3_acquire_protected_relation_locks"
CATALOG_LOCK_FUNCTION_SOURCE_SHA256 = "897345944ba0f1b2a5c01b098fb17de6f881a122e906277086fa7889e76219e8"

AGENT_ID = "11111111-1111-4111-8111-111111111111"
LEGACY_CLAIM_ID = "20000000-0000-4000-8000-000000000001"
LEGACY_RELATED_CLAIM_ID = "20000000-0000-4000-8000-000000000002"
LEGACY_SOURCE_ID = "20000000-0000-4000-8000-000000000003"
LEGACY_EVIDENCE_ID = "20000000-0000-4000-8000-000000000004"
LEGACY_EDGE_ID = "20000000-0000-4000-8000-000000000005"

PROPOSAL_ID = "30000000-0000-4000-8000-000000000001"
PENDING_PROPOSAL_ID = "30000000-0000-4000-8000-000000000002"
APPLIED_WITHOUT_RECEIPT_PROPOSAL_ID = "30000000-0000-4000-8000-000000000003"
PRIMARY_CLAIM_ID = "40000000-0000-4000-8000-000000000001"
SUPPORT_CLAIM_ID = "40000000-0000-4000-8000-000000000002"
SOURCE_ID = "40000000-0000-4000-8000-000000000003"
EVIDENCE_ID = "40000000-0000-4000-8000-000000000004"
ASSESSMENT_ID = "40000000-0000-4000-8000-000000000005"
EDGE_ID = "40000000-0000-4000-8000-000000000006"

SOURCE_PAYLOAD = b"LivingIP V3 source payload: immutable epistemic provenance.\n"
SOURCE_CONTENT_HASH = hashlib.sha256(SOURCE_PAYLOAD).hexdigest()
EVIDENCE_SET_HASH = hashlib.sha256(
    json.dumps(
        {
            "claim_id": PRIMARY_CLAIM_ID,
            "evidence": [
                {
                    "source_id": SOURCE_ID,
                    "source_content_hash": SOURCE_CONTENT_HASH,
                    "polarity": "supports",
                    "locator": {"paragraph": 1, "section": "epistemic-contract"},
                }
            ],
        },
        separators=(",", ":"),
        sort_keys=True,
    ).encode("utf-8")
).hexdigest()

PRIMARY_PROPOSITION = "An admitted V3 claim preserves exact evidence provenance."
PRIMARY_BODY = (
    "The claim records a proposition and explanatory body together, pins a content-hashed source locator, "
    "and remains attributable to the reviewed proposal that admitted it."
)
PRIMARY_SCOPE = "Minimum disposable PostgreSQL V3 contract verification."
PRIMARY_REVISION_CRITERIA = (
    "Revise if the stored source hash, locator, proposal decision, or evidence assessment cannot be reproduced."
)


def load_expected_trigger_contract() -> list[dict[str, Any]]:
    match = TRIGGER_CONTRACT_PATTERN.search(MIGRATION_SQL.read_text(encoding="utf-8"))
    if match is None:
        raise VerificationError("migration does not declare the V3 trigger catalog contract")
    contract = json.loads(match.group(1))
    if not isinstance(contract, list):
        raise VerificationError("migration V3 trigger catalog contract is not a JSON array")
    return contract


def trigger_catalog_is_exact(catalog: dict[str, Any]) -> bool:
    expected_triggers = load_expected_trigger_contract()
    actual_triggers = catalog["triggers"]
    if len(actual_triggers) != len(expected_triggers):
        return False

    expected_by_key = {(row["schema_name"], row["table_name"], row["trigger_name"]): row for row in expected_triggers}
    actual_by_key = {(row["schema_name"], row["table_name"], row["trigger_name"]): row for row in actual_triggers}
    if len(expected_by_key) != len(expected_triggers) or len(actual_by_key) != len(actual_triggers):
        return False
    if actual_by_key.keys() != expected_by_key.keys():
        return False

    for key, expected in expected_by_key.items():
        actual = actual_by_key[key]
        if actual["enabled_state"] != "O" or actual["is_internal"]:
            return False
        if actual["trigger_type"] != expected["trigger_type"]:
            return False
        if actual["update_columns"] != expected["update_columns"]:
            return False
        if actual["when_expression"] != expected["when_expression"]:
            return False
        if actual["argument_count"] != expected["argument_count"]:
            return False
        if actual["argument_hex"] != expected["argument_hex"]:
            return False
        if any(
            (
                actual["constraint_oid"] != "0",
                actual["is_deferrable"],
                actual["is_initially_deferred"],
                actual["constraint_relation_oid"] != "0",
                actual["parent_trigger_oid"] != "0",
                actual["old_transition_table"] is not None,
                actual["new_transition_table"] is not None,
            )
        ):
            return False
        if actual["function_schema"] != expected["function_schema"]:
            return False
        if actual["function_name"] != expected["function_name"]:
            return False
        if actual["function_identity_arguments"] != expected["function_identity_arguments"]:
            return False

    expected_functions = {
        (
            row["function_schema"],
            row["function_name"],
            row["function_identity_arguments"],
        ): row
        for row in expected_triggers
    }
    actual_functions = {
        (row["schema_name"], row["function_name"], row["arguments"]): row for row in catalog["functions"]
    }
    if len(actual_functions) != len(catalog["functions"]):
        return False
    catalog_lock_function = actual_functions.pop(
        ("kb_stage", CATALOG_LOCK_FUNCTION_NAME, ""),
        None,
    )
    if catalog_lock_function is None or any(
        (
            catalog_lock_function["owner_name"] != "kb_gate_owner",
            not catalog_lock_function["security_definer"],
            catalog_lock_function["config"] != ["search_path=pg_catalog, pg_temp"],
            catalog_lock_function["source_sha256"] != CATALOG_LOCK_FUNCTION_SOURCE_SHA256,
            catalog_lock_function["language_name"] != "plpgsql",
            catalog_lock_function["function_kind"] != "f",
            catalog_lock_function["argument_count"] != 0,
            catalog_lock_function["return_type"] != "void",
            catalog_lock_function["non_owner_execute_grantees"] != ["kb_apply"],
        )
    ):
        return False
    if actual_functions.keys() != expected_functions.keys():
        return False
    for key, expected in expected_functions.items():
        actual = actual_functions[key]
        if actual["owner_name"] != expected["function_owner"]:
            return False
        if actual["security_definer"] != expected["function_security_definer"]:
            return False
        if actual["config"] != expected["function_config"]:
            return False
        if actual["source_sha256"] != expected["function_source_sha256"]:
            return False
        if expected["function_owner_only_execute"] is not None and (
            actual["owner_only_execute"] != expected["function_owner_only_execute"]
        ):
            return False
        if actual["language_name"] != "plpgsql":
            return False
        if actual["function_kind"] != "f" or actual["argument_count"] != 0:
            return False
        if actual["return_type"] != "trigger":
            return False
    return True


class VerificationError(RuntimeError):
    """Raised when the disposable runtime cannot prove the contract."""


def canonical_sha256(value: Any) -> str:
    payload = json.dumps(value, ensure_ascii=False, separators=(",", ":"), sort_keys=True)
    return hashlib.sha256(payload.encode("utf-8")).hexdigest()


def parse_last_json_line(output: str) -> Any:
    for line in reversed(output.splitlines()):
        candidate = line.strip()
        if not candidate:
            continue
        try:
            return json.loads(candidate)
        except json.JSONDecodeError:
            continue
    return None


def safe_error_text(value: str, *, limit: int = 1600) -> str:
    return " ".join(value.split())[-limit:]


def sql_literal(value: str) -> str:
    escaped = value.replace("\\", "\\\\").replace("'", "''")
    return f"E'{escaped}'"


def docker_available() -> bool:
    docker = shutil.which("docker")
    if not docker:
        return False
    result = subprocess.run(
        [docker, "info", "--format", "{{.ServerVersion}}"],
        text=True,
        capture_output=True,
        check=False,
        timeout=15,
    )
    return result.returncode == 0


BOOTSTRAP_SQL = f"""
\\set ON_ERROR_STOP on
create schema kb_stage;
create role kb_gate_owner nologin noinherit;
create role kb_apply nologin noinherit;
create role kb_review login noinherit;
create type evidence_role as enum ('grounds', 'illustrates', 'contradicts');
create type edge_type as enum (
  'supports', 'challenges', 'requires', 'relates', 'contradicts', 'supersedes',
  'derives_from', 'cites', 'causes', 'constrains', 'accelerates'
);

create table public.agents (
  id uuid primary key,
  handle text not null unique,
  kind text not null,
  created_at timestamptz not null default now()
);

create table public.claims (
  id uuid primary key,
  type text not null,
  text text not null,
  status text not null,
  confidence numeric,
  tags text[] not null default '{{}}',
  created_by uuid references public.agents(id),
  superseded_by uuid references public.claims(id),
  created_at timestamptz not null default now(),
  updated_at timestamptz not null default now()
);

create table public.sources (
  id uuid primary key,
  source_type text not null,
  url text,
  storage_path text,
  excerpt text,
  hash text not null,
  created_by uuid references public.agents(id),
  captured_at timestamptz,
  created_at timestamptz not null default now()
);

create table public.claim_evidence (
  id uuid primary key default gen_random_uuid(),
  claim_id uuid not null references public.claims(id),
  source_id uuid not null references public.sources(id),
  role evidence_role not null,
  weight numeric,
  created_by uuid references public.agents(id),
  created_at timestamptz not null default now()
);

create table public.claim_edges (
  id uuid primary key,
  from_claim uuid not null references public.claims(id),
  to_claim uuid not null references public.claims(id),
  edge_type edge_type not null,
  weight numeric,
  created_by uuid references public.agents(id),
  created_at timestamptz not null default now()
);

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

grant usage on schema kb_stage to kb_gate_owner;
grant select, update on kb_stage.kb_proposals to kb_gate_owner;

insert into public.agents (id, handle, kind, created_at)
values ('{AGENT_ID}', 'v3-contract-reviewer', 'human', '2026-07-18T08:00:00Z');

insert into public.claims
  (id, type, text, status, confidence, tags, created_by, created_at, updated_at)
values
  ('{LEGACY_CLAIM_ID}', 'structural', 'Legacy claim without a body.', 'open', 0.4,
   array['legacy'], '{AGENT_ID}', '2026-07-18T08:01:00Z', '2026-07-18T08:01:00Z'),
  ('{LEGACY_RELATED_CLAIM_ID}', 'structural', 'Second legacy claim.', 'open', null,
   array['legacy'], '{AGENT_ID}', '2026-07-18T08:02:00Z', '2026-07-18T08:02:00Z');

insert into public.sources
  (id, source_type, url, storage_path, excerpt, hash, created_by, captured_at, created_at)
values
  ('{LEGACY_SOURCE_ID}', 'legacy_document', null, '/legacy/source.txt', 'Legacy excerpt.', 'legacy-hash',
   '{AGENT_ID}', '2026-07-18T08:03:00Z', '2026-07-18T08:03:00Z');

insert into public.claim_evidence
  (id, claim_id, source_id, role, weight, created_by, created_at)
values
  ('{LEGACY_EVIDENCE_ID}', '{LEGACY_CLAIM_ID}', '{LEGACY_SOURCE_ID}', 'grounds', 0.25,
   '{AGENT_ID}', '2026-07-18T08:04:00Z');

insert into public.claim_edges
  (id, from_claim, to_claim, edge_type, weight, created_by, created_at)
values
  ('{LEGACY_EDGE_ID}', '{LEGACY_CLAIM_ID}', '{LEGACY_RELATED_CLAIM_ID}', 'relates', 0.5,
   '{AGENT_ID}', '2026-07-18T08:05:00Z');

-- This malformed historical row proves the NOT VALID proposal constraint is
-- additive while the canonical-row trigger still refuses it as authority.
insert into kb_stage.kb_proposals (
  id, proposal_type, status, proposed_by_handle, proposed_by_agent_id, channel,
  rationale, payload, reviewed_by_handle, reviewed_by_agent_id, reviewed_at, review_note,
  created_at, updated_at
) values (
  '{APPLIED_WITHOUT_RECEIPT_PROPOSAL_ID}', 'approve_claim', 'applied', 'leo', '{AGENT_ID}',
  'legacy_fixture', 'Historical applied status without a receipt.',
  '{{"contract_version": 2}}'::jsonb, 'v3-contract-reviewer', '{AGENT_ID}',
  '2026-07-18T08:06:00Z', 'Historical malformed receipt retained for compatibility testing.',
  '2026-07-18T08:06:00Z', '2026-07-18T08:06:00Z'
);
"""


LEGACY_STATE_SQL = """
select jsonb_build_object(
  'claims', coalesce((
    select jsonb_agg(to_jsonb(row_data) order by row_data.id)
      from (
        select id, type, text, status, confidence, tags, created_by, superseded_by, created_at, updated_at
          from public.claims
         where id in (
           '20000000-0000-4000-8000-000000000001'::uuid,
           '20000000-0000-4000-8000-000000000002'::uuid
         )
      ) row_data
  ), '[]'::jsonb),
  'sources', coalesce((
    select jsonb_agg(to_jsonb(row_data) order by row_data.id)
      from (
        select id, source_type, url, storage_path, excerpt, hash, created_by, captured_at, created_at
          from public.sources
         where id = '20000000-0000-4000-8000-000000000003'::uuid
      ) row_data
  ), '[]'::jsonb),
  'claim_evidence', coalesce((
    select jsonb_agg(to_jsonb(row_data) order by row_data.id)
      from (
        select id, claim_id, source_id, role, weight, created_by, created_at
          from public.claim_evidence
         where id = '20000000-0000-4000-8000-000000000004'::uuid
      ) row_data
  ), '[]'::jsonb),
  'claim_edges', coalesce((
    select jsonb_agg(to_jsonb(row_data) order by row_data.id)
      from (
        select id, from_claim, to_claim, edge_type, weight, created_by, created_at
          from public.claim_edges
         where id = '20000000-0000-4000-8000-000000000005'::uuid
      ) row_data
  ), '[]'::jsonb)
)::text;
"""


CONTRACT_CATALOG_SQL = """
with contract_columns as (
  select table_schema, table_name, column_name, data_type, udt_name, is_nullable, column_default
    from information_schema.columns
   where table_schema = 'public'
     and (
       (table_name = 'claims' and column_name in (
         'owner_agent_id', 'proposition', 'body_markdown', 'scope', 'access_scope', 'admission_state',
         'revision_criteria', 'revision_kind', 'supersedes_id', 'accepted_by_proposal_id'
       ))
       or (table_name = 'sources' and column_name in (
         'canonical_url', 'storage_uri', 'content_hash', 'access_scope', 'ingestion_origin',
         'provenance_status', 'source_class', 'auto_promotion_policy', 'accepted_by_proposal_id'
       ))
       or (table_name = 'claim_evidence' and column_name in (
         'excerpt', 'locator_json', 'source_content_hash', 'rationale', 'polarity', 'accepted_from_proposal_id'
       ))
       or (table_name = 'claim_edges' and column_name in (
         'relation_type', 'rationale', 'scope_or_conditions', 'accepted_by_proposal_id'
       ))
     )
), contract_constraints as (
  select namespace.nspname as schema_name, relation.relname as table_name,
         constraint_row.conname as constraint_name,
         pg_get_constraintdef(constraint_row.oid, true) as definition
    from pg_constraint constraint_row
    join pg_class relation on relation.oid = constraint_row.conrelid
    join pg_namespace namespace on namespace.oid = relation.relnamespace
   where namespace.nspname in ('public', 'kb_stage')
     and constraint_row.conname like 'teleo_v3_%'
), contract_triggers as (
  select namespace.nspname as schema_name, relation.relname as table_name,
         trigger_row.tgname as trigger_name,
         trigger_row.tgenabled as enabled_state,
         trigger_row.tgisinternal as is_internal,
         trigger_row.tgtype as trigger_type,
         coalesce((
           select jsonb_agg(attribute.attname order by attribute.attname)
             from unnest(trigger_row.tgattr::smallint[]) update_column(attnum)
             join pg_attribute attribute
               on attribute.attrelid = trigger_row.tgrelid
              and attribute.attnum = update_column.attnum
              and not attribute.attisdropped
         ), '[]'::jsonb) as update_columns,
         pg_get_expr(trigger_row.tgqual, trigger_row.tgrelid, true) as when_expression,
         trigger_row.tgnargs as argument_count,
         encode(trigger_row.tgargs, 'hex') as argument_hex,
         trigger_row.tgconstraint::text as constraint_oid,
         trigger_row.tgdeferrable as is_deferrable,
         trigger_row.tginitdeferred as is_initially_deferred,
         trigger_row.tgconstrrelid::text as constraint_relation_oid,
         trigger_row.tgparentid::text as parent_trigger_oid,
         trigger_row.tgoldtable as old_transition_table,
         trigger_row.tgnewtable as new_transition_table,
         function_namespace.nspname as function_schema,
         trigger_function.proname as function_name,
         pg_get_function_identity_arguments(trigger_function.oid) as function_identity_arguments,
         pg_get_triggerdef(trigger_row.oid, true) as definition
    from pg_trigger trigger_row
    join pg_class relation on relation.oid = trigger_row.tgrelid
    join pg_namespace namespace on namespace.oid = relation.relnamespace
    join pg_proc trigger_function on trigger_function.oid = trigger_row.tgfoid
    join pg_namespace function_namespace on function_namespace.oid = trigger_function.pronamespace
   where left(trigger_row.tgname, 9) = 'teleo_v3_'
), contract_functions as (
  select namespace.nspname as schema_name, procedure.proname as function_name,
         pg_get_function_identity_arguments(procedure.oid) as arguments,
         owner_role.rolname as owner_name,
         procedure.prosecdef as security_definer,
         procedure.proconfig as config,
         encode(sha256(convert_to(procedure.prosrc, 'UTF8')), 'hex') as source_sha256,
         language.lanname as language_name,
         procedure.prokind as function_kind,
         procedure.pronargs as argument_count,
         procedure.prorettype::regtype::text as return_type,
         not exists (
           select 1
             from pg_catalog.aclexplode(
               coalesce(
                 procedure.proacl,
                 pg_catalog.acldefault('f', procedure.proowner)
               )
             ) acl
            where acl.privilege_type = 'EXECUTE'
              and acl.grantee <> procedure.proowner
         ) as owner_only_execute,
         coalesce((
           select jsonb_agg(coalesce(grantee.rolname, 'PUBLIC') order by coalesce(grantee.rolname, 'PUBLIC'))
             from pg_catalog.aclexplode(
               coalesce(
                 procedure.proacl,
                 pg_catalog.acldefault('f', procedure.proowner)
               )
             ) acl
             left join pg_catalog.pg_roles grantee on grantee.oid = acl.grantee
            where acl.privilege_type = 'EXECUTE'
              and acl.grantee <> procedure.proowner
         ), '[]'::jsonb) as non_owner_execute_grantees
    from pg_proc procedure
    join pg_namespace namespace on namespace.oid = procedure.pronamespace
    join pg_roles owner_role on owner_role.oid = procedure.proowner
    join pg_language language on language.oid = procedure.prolang
   where namespace.nspname = 'kb_stage'
     and left(procedure.proname, 9) = 'teleo_v3_'
), contract_indexes as (
  select schemaname, tablename, indexname, indexdef
    from pg_indexes
   where schemaname = 'public' and indexname like 'teleo_v3_%'
)
select jsonb_build_object(
  'assessment_table', to_regclass('public.claim_evidence_assessments')::text,
  'columns', coalesce((
    select jsonb_agg(to_jsonb(contract_columns) order by table_name, column_name) from contract_columns
  ), '[]'::jsonb),
  'constraints', coalesce((
    select jsonb_agg(to_jsonb(contract_constraints) order by schema_name, table_name, constraint_name)
      from contract_constraints
  ), '[]'::jsonb),
  'triggers', coalesce((
    select jsonb_agg(to_jsonb(contract_triggers) order by schema_name, table_name, trigger_name)
      from contract_triggers
  ), '[]'::jsonb),
  'functions', coalesce((
    select jsonb_agg(to_jsonb(contract_functions) order by schema_name, function_name, arguments)
      from contract_functions
  ), '[]'::jsonb),
  'indexes', coalesce((
    select jsonb_agg(to_jsonb(contract_indexes) order by schemaname, tablename, indexname)
      from contract_indexes
  ), '[]'::jsonb)
)::text;
"""


POSTGRES_READY_LOG_LINE = "database system is ready to accept connections"
POSTGRES_STARTUP_TIMEOUT_SECONDS = 45.0
POSTGRES_STARTUP_PROBE_TIMEOUT_SECONDS = 10.0
POSTGRES_STARTUP_POLL_INTERVAL_SECONDS = 0.25
POSTGRES_CLEANUP_COMMAND_TIMEOUT_SECONDS = 10.0
POSTGRES_CLEANUP_MAX_ATTEMPTS = 5
POSTGRES_CLEANUP_POLL_INTERVAL_SECONDS = 0.25
DOCKER_CONTAINER_ID_PATTERN = re.compile(r"[0-9a-f]{12,64}")


def postgres_final_startup_ready(
    logs: str,
    database_probe: subprocess.CompletedProcess[str],
) -> bool:
    return (
        logs.count(POSTGRES_READY_LOG_LINE) >= 2
        and database_probe.returncode == 0
        and database_probe.stdout.strip() == "1"
    )


class DisposablePostgres:
    def __init__(self) -> None:
        self.docker = shutil.which("docker")
        if not self.docker:
            raise VerificationError("docker executable is unavailable")
        self.name = f"teleo-v3-contract-{uuid.uuid4().hex[:12]}"
        self.password = secrets.token_urlsafe(32)
        self.started = False
        self.cleanup_required = False
        self.environment: dict[str, Any] = {}

    def run(
        self,
        arguments: list[str],
        *,
        input_text: str | None = None,
        check: bool = True,
        timeout: float = 120,
        environment: dict[str, str] | None = None,
    ) -> subprocess.CompletedProcess[str]:
        result = subprocess.run(
            [self.docker, *arguments],
            input=input_text,
            text=True,
            capture_output=True,
            check=False,
            timeout=timeout,
            env=environment,
        )
        if check and result.returncode != 0:
            detail = safe_error_text(result.stderr or result.stdout)
            raise VerificationError(f"docker {arguments[0]} failed ({result.returncode}): {detail}")
        return result

    def _redacted_startup_detail(self, value: str | bytes | None) -> str:
        if isinstance(value, bytes):
            value = value.decode("utf-8", errors="replace")
        raw_detail = value or ""
        if self.password:
            raw_detail = raw_detail.replace(self.password, "<redacted>")
        detail = safe_error_text(raw_detail, limit=500)
        return detail or "<empty>"

    def _startup_result_diagnostic(
        self,
        stage: str,
        result: subprocess.CompletedProcess[str],
    ) -> str:
        detail = self._redacted_startup_detail(result.stderr or result.stdout)
        return f"stage={stage} kind=nonzero returncode={result.returncode} detail={detail}"

    def _startup_exception_diagnostic(
        self,
        stage: str,
        exc: OSError | subprocess.TimeoutExpired,
    ) -> str:
        if isinstance(exc, subprocess.TimeoutExpired):
            detail = self._redacted_startup_detail(exc.stderr or exc.output)
            return f"stage={stage} kind=timeout timeout_seconds={exc.timeout:g} detail={detail}"
        return f"stage={stage} kind=launch_error detail={self._redacted_startup_detail(str(exc))}"

    def _wait_for_final_startup(
        self,
        *,
        startup_timeout: float = POSTGRES_STARTUP_TIMEOUT_SECONDS,
        probe_timeout: float = POSTGRES_STARTUP_PROBE_TIMEOUT_SECONDS,
        poll_interval: float = POSTGRES_STARTUP_POLL_INTERVAL_SECONDS,
    ) -> None:
        if startup_timeout <= 0 or probe_timeout <= 0 or poll_interval < 0:
            raise ValueError("PostgreSQL startup timing values must be positive")

        deadline = time.monotonic() + startup_timeout
        ready_streak = 0
        attempts = 0
        last_diagnostic = "stage=startup kind=no_probe_completed"
        while True:
            remaining = deadline - time.monotonic()
            if remaining <= 0:
                break

            attempts += 1
            try:
                logs = self.run(
                    ["logs", self.name],
                    check=False,
                    timeout=min(probe_timeout, remaining),
                )
            except (OSError, subprocess.TimeoutExpired) as exc:
                ready_streak = 0
                last_diagnostic = self._startup_exception_diagnostic("docker_logs", exc)
            else:
                if logs.returncode != 0:
                    ready_streak = 0
                    last_diagnostic = self._startup_result_diagnostic("docker_logs", logs)
                else:
                    remaining = deadline - time.monotonic()
                    if remaining <= 0:
                        last_diagnostic = "stage=target_database_probe kind=deadline_exhausted_after_logs"
                        break
                    try:
                        database_probe = self.psql(
                            "select 1;",
                            check=False,
                            timeout=min(probe_timeout, remaining),
                        )
                    except (OSError, subprocess.TimeoutExpired) as exc:
                        ready_streak = 0
                        last_diagnostic = self._startup_exception_diagnostic("target_database_probe", exc)
                    else:
                        if postgres_final_startup_ready(logs.stdout + logs.stderr, database_probe):
                            ready_streak += 1
                            if ready_streak == 2:
                                return
                        else:
                            ready_streak = 0
                            if database_probe.returncode != 0:
                                last_diagnostic = self._startup_result_diagnostic(
                                    "target_database_probe",
                                    database_probe,
                                )
                            else:
                                ready_transitions = (logs.stdout + logs.stderr).count(POSTGRES_READY_LOG_LINE)
                                probe_output = self._redacted_startup_detail(database_probe.stdout)
                                last_diagnostic = (
                                    "stage=readiness_gate kind=not_ready "
                                    f"ready_transitions={ready_transitions} "
                                    f"database_probe_returncode={database_probe.returncode} "
                                    f"database_probe_output={probe_output}"
                                )

            remaining = deadline - time.monotonic()
            if remaining <= 0:
                break
            time.sleep(min(poll_interval, remaining))

        raise VerificationError(
            f"disposable PostgreSQL did not become stably ready within {startup_timeout:g}s "
            f"after {attempts} attempt(s); last_probe={last_diagnostic}"
        )

    def _launch_container(self) -> subprocess.CompletedProcess[str]:
        arguments = [
            "run",
            "--detach",
            "--rm",
            "--name",
            self.name,
            "--network",
            "none",
            "--label",
            CONTAINER_LABEL,
            "--label",
            f"livingip.canary.instance={self.name}",
            "--tmpfs",
            "/var/lib/postgresql/data:rw,nosuid,nodev,size=384m",
            "--env",
            "POSTGRES_PASSWORD",
            "--env",
            f"POSTGRES_DB={DATABASE}",
            POSTGRES_IMAGE,
        ]
        launch_environment = os.environ.copy()
        launch_environment["POSTGRES_PASSWORD"] = self.password
        self.cleanup_required = True
        try:
            launched = self.run(
                arguments,
                check=False,
                environment=launch_environment,
            )
        except (OSError, subprocess.TimeoutExpired) as exc:
            diagnostic = self._startup_exception_diagnostic("docker_run", exc)
            raise VerificationError(
                "disposable PostgreSQL launch state is ambiguous; cleanup_required=true; "
                f"{diagnostic}"
            ) from None
        if launched.returncode != 0:
            diagnostic = self._startup_result_diagnostic("docker_run", launched)
            raise VerificationError(
                "disposable PostgreSQL launch failed; cleanup_required=true; "
                f"{diagnostic}"
            )
        self.started = True
        return launched

    def start(self) -> dict[str, Any]:
        launched = self._launch_container()

        self._wait_for_final_startup()

        inspected = json.loads(self.run(["inspect", self.name]).stdout)[0]
        labels = inspected.get("Config", {}).get("Labels", {})
        mounts = inspected.get("Mounts", [])
        tmpfs = inspected.get("HostConfig", {}).get("Tmpfs", {})
        self.environment = {
            "container_id_sha256": hashlib.sha256(launched.stdout.strip().encode("utf-8")).hexdigest(),
            "container_name": self.name,
            "image": inspected.get("Config", {}).get("Image"),
            "image_id": inspected.get("Image"),
            "network_mode": inspected.get("HostConfig", {}).get("NetworkMode"),
            "tmpfs_data_dir": "/var/lib/postgresql/data" in tmpfs,
            "docker_volume_mounts": [
                {"type": mount.get("Type"), "destination": mount.get("Destination")} for mount in mounts
            ],
            "canary_label": labels.get("livingip.canary"),
            "instance_label": labels.get("livingip.canary.instance"),
        }
        if self.environment["network_mode"] != "none":
            raise VerificationError("disposable PostgreSQL is not networkless")
        if not self.environment["tmpfs_data_dir"] or mounts:
            raise VerificationError("disposable PostgreSQL must use tmpfs without Docker volumes")
        if self.environment["canary_label"] != "teleo-v3-epistemic-contract":
            raise VerificationError("disposable PostgreSQL canary label is missing")
        if self.environment["instance_label"] != self.name:
            raise VerificationError("disposable PostgreSQL instance label is missing")
        return self.environment

    def psql(
        self,
        sql: str,
        *,
        check: bool = True,
        timeout: float = 120,
    ) -> subprocess.CompletedProcess[str]:
        return self.run(
            [
                "exec",
                "--interactive",
                self.name,
                "psql",
                "--no-psqlrc",
                "--no-password",
                "--set=ON_ERROR_STOP=1",
                "--set=VERBOSITY=verbose",
                "--tuples-only",
                "--no-align",
                "--username",
                "postgres",
                "--dbname",
                DATABASE,
            ],
            input_text=sql,
            check=check,
            timeout=timeout,
        )

    def psql_json(self, sql: str) -> Any:
        result = self.psql(sql)
        parsed = parse_last_json_line(result.stdout)
        if parsed is None:
            raise VerificationError("PostgreSQL readback did not contain JSON")
        return parsed

    def apply_file(self, path: Path, *, check: bool = True) -> subprocess.CompletedProcess[str]:
        return self.psql(path.read_text(encoding="utf-8"), check=check)

    def _cleanup_run(
        self,
        arguments: list[str],
        *,
        stage: str,
    ) -> tuple[subprocess.CompletedProcess[str] | None, str | None]:
        try:
            return (
                self.run(
                    arguments,
                    check=False,
                    timeout=POSTGRES_CLEANUP_COMMAND_TIMEOUT_SECONDS,
                ),
                None,
            )
        except (OSError, subprocess.TimeoutExpired) as exc:
            return None, self._startup_exception_diagnostic(stage, exc)

    def cleanup(self) -> dict[str, Any]:
        ownership_filters = [
            "container",
            "ls",
            "--all",
            "--no-trunc",
            "--filter",
            f"name=^/{self.name}$",
            "--filter",
            "label=livingip.canary=teleo-v3-epistemic-contract",
            "--filter",
            f"label=livingip.canary.instance={self.name}",
            "--format",
            "{{.ID}}",
        ]
        name_readback = [
            "container",
            "ls",
            "--all",
            "--filter",
            f"name=^/{self.name}$",
            "--format",
            "{{.Names}}",
        ]
        label_readback = [
            "container",
            "ls",
            "--all",
            "--filter",
            "label=livingip.canary=teleo-v3-epistemic-contract",
            "--filter",
            f"label=livingip.canary.instance={self.name}",
            "--format",
            "{{.Names}}",
        ]
        errors: list[str] = []
        observed_owned_ids: set[str] = set()
        remove_returncodes: list[int] = []
        remove_attempted = False
        stable_clear_observations = 0
        inspect: subprocess.CompletedProcess[str] | None = None
        names: subprocess.CompletedProcess[str] | None = None
        labels: subprocess.CompletedProcess[str] | None = None
        attempts = 0

        for attempts in range(1, POSTGRES_CLEANUP_MAX_ATTEMPTS + 1):
            ownership, ownership_error = self._cleanup_run(
                ownership_filters,
                stage="cleanup_ownership_readback",
            )
            if ownership_error:
                errors.append(ownership_error)
            owned_ids: list[str] = []
            if ownership is not None and ownership.returncode == 0:
                candidates = [line.strip() for line in ownership.stdout.splitlines() if line.strip()]
                if all(DOCKER_CONTAINER_ID_PATTERN.fullmatch(candidate) for candidate in candidates):
                    owned_ids = candidates
                    observed_owned_ids.update(candidates)
                else:
                    errors.append("stage=cleanup_ownership_readback kind=invalid_container_id")
            elif ownership is not None:
                errors.append(self._startup_result_diagnostic("cleanup_ownership_readback", ownership))

            if self.cleanup_required:
                for container_id in owned_ids:
                    remove_attempted = True
                    removed, remove_error = self._cleanup_run(
                        ["rm", "--force", container_id],
                        stage="cleanup_remove",
                    )
                    if remove_error:
                        errors.append(remove_error)
                    elif removed is not None:
                        remove_returncodes.append(removed.returncode)
                        if removed.returncode != 0:
                            errors.append(self._startup_result_diagnostic("cleanup_remove", removed))

            inspect, inspect_error = self._cleanup_run(
                ["inspect", self.name],
                stage="cleanup_inspect_readback",
            )
            names, names_error = self._cleanup_run(name_readback, stage="cleanup_name_readback")
            labels, labels_error = self._cleanup_run(label_readback, stage="cleanup_label_readback")
            errors.extend(error for error in (inspect_error, names_error, labels_error) if error)
            if names is not None and names.returncode != 0:
                errors.append(self._startup_result_diagnostic("cleanup_name_readback", names))
            if labels is not None and labels.returncode != 0:
                errors.append(self._startup_result_diagnostic("cleanup_label_readback", labels))

            name_clear = names is not None and names.returncode == 0 and not names.stdout.strip()
            label_clear = labels is not None and labels.returncode == 0 and not labels.stdout.strip()
            inspect_clear = inspect is not None and inspect.returncode != 0
            if inspect_clear and name_clear and label_clear:
                stable_clear_observations += 1
                if stable_clear_observations == 2:
                    break
            else:
                stable_clear_observations = 0
            if attempts < POSTGRES_CLEANUP_MAX_ATTEMPTS:
                time.sleep(POSTGRES_CLEANUP_POLL_INTERVAL_SECONDS)

        name_readback_clear = names is not None and names.returncode == 0 and not names.stdout.strip()
        instance_label_readback_clear = labels is not None and labels.returncode == 0 and not labels.stdout.strip()
        container_absent = (
            inspect is not None
            and inspect.returncode != 0
            and name_readback_clear
            and instance_label_readback_clear
        )
        return {
            "cleanup_required": self.cleanup_required,
            "cleanup_reconciliation_attempted": self.cleanup_required,
            "remove_attempted": remove_attempted,
            "remove_returncode": remove_returncodes[-1] if remove_returncodes else None,
            "remove_returncodes": remove_returncodes,
            "owned_container_ids_observed": sorted(observed_owned_ids),
            "container_absent": container_absent,
            "name_readback_clear": name_readback_clear,
            "instance_label_readback_clear": instance_label_readback_clear,
            "stable_absence_readback": stable_clear_observations >= 2,
            "readback_attempts": attempts,
            "foreign_name_collision_detected": not name_readback_clear and instance_label_readback_clear,
            "errors": errors,
            "network_mode_was_none": self.environment.get("network_mode") == "none",
            "tmpfs_data_dir_was_present": self.environment.get("tmpfs_data_dir") is True,
            "docker_volume_mounts_observed": self.environment.get("docker_volume_mounts", []),
        }


def catalog_readback(postgres: DisposablePostgres) -> dict[str, Any]:
    value = postgres.psql_json(CONTRACT_CATALOG_SQL)
    if not isinstance(value, dict):
        raise VerificationError("contract catalog readback was not an object")
    return value


def catalog_is_absent(catalog: dict[str, Any]) -> bool:
    return catalog.get("assessment_table") is None and all(
        catalog.get(key) == [] for key in ("columns", "constraints", "triggers", "functions", "indexes")
    )


def failure_receipt(result: subprocess.CompletedProcess[str], expected_sqlstate: str) -> dict[str, Any]:
    match = re.search(r"(?:ERROR|FATAL):\s+([0-9A-Z]{5}):", result.stderr)
    observed_sqlstate = match.group(1) if match else None
    return {
        "refused": result.returncode != 0,
        "returncode": result.returncode,
        "expected_sqlstate": expected_sqlstate,
        "observed_sqlstate": observed_sqlstate,
        "expected_sqlstate_observed": observed_sqlstate == expected_sqlstate,
        "stderr_tail": safe_error_text(result.stderr, limit=500),
    }


def expect_failure(
    postgres: DisposablePostgres,
    sql: str,
    *,
    expected_sqlstate: str,
) -> dict[str, Any]:
    result = postgres.psql(sql, check=False)
    receipt = failure_receipt(result, expected_sqlstate)
    if not receipt["refused"] or not receipt["expected_sqlstate_observed"]:
        raise VerificationError(
            f"adversarial statement did not fail with SQLSTATE {expected_sqlstate}: {receipt['stderr_tail']}"
        )
    return receipt


def proposal_decision_lock_race_check(postgres: DisposablePostgres) -> dict[str, Any]:
    """Prove a canonical writer locks the accepted decision against concurrent downgrade."""

    assert postgres.docker is not None
    command = [
        postgres.docker,
        "exec",
        "--interactive",
        postgres.name,
        "psql",
        "--no-psqlrc",
        "--no-password",
        "--set=ON_ERROR_STOP=1",
        "--set=VERBOSITY=verbose",
        "--tuples-only",
        "--no-align",
        "--username",
        "postgres",
        "--dbname",
        DATABASE,
    ]
    holder = subprocess.Popen(
        command,
        stdin=subprocess.PIPE,
        stdout=subprocess.PIPE,
        stderr=subprocess.PIPE,
        text=True,
    )
    assert holder.stdin is not None
    holder.stdin.write(
        f"""
set application_name = 'teleo_v3_proposal_lock_holder';
begin;
select status from kb_stage.kb_proposals where id = '{PROPOSAL_ID}'::uuid for share;
select pg_sleep(4);
rollback;
"""
    )
    holder.stdin.close()
    holder.stdin = None
    try:
        deadline = time.monotonic() + 8
        lock_held = False
        while time.monotonic() < deadline:
            lock_held = postgres.psql_json(
                """
select jsonb_build_object(
  'held', exists (
    select 1 from pg_stat_activity
     where application_name = 'teleo_v3_proposal_lock_holder'
       and state = 'active'
       and wait_event = 'PgSleep'
  )
)::text;
"""
            )["held"]
            if lock_held:
                break
            time.sleep(0.1)
        if not lock_held:
            raise VerificationError("proposal lock holder did not reach the protected transaction")
        downgrade = postgres.psql(
            f"""
set statement_timeout = '750ms';
update kb_stage.kb_proposals set status = 'rejected' where id = '{PROPOSAL_ID}'::uuid;
""",
            check=False,
        )
        downgrade_receipt = failure_receipt(downgrade, "57014")
        holder_stdout, holder_stderr = holder.communicate(timeout=8)
        status = postgres.psql_json(
            f"select jsonb_build_object('status', status)::text from kb_stage.kb_proposals "
            f"where id = '{PROPOSAL_ID}'::uuid;"
        )["status"]
        return {
            "lock_observed": lock_held,
            "concurrent_downgrade": downgrade_receipt,
            "holder_returncode": holder.returncode,
            "holder_stderr_tail": safe_error_text(holder_stderr, limit=300),
            "holder_output_observed": bool(holder_stdout.strip()),
            "proposal_status_after": status,
        }
    finally:
        if holder.poll() is None:
            holder.terminate()
            try:
                holder.wait(timeout=3)
            except subprocess.TimeoutExpired:
                holder.kill()
                holder.wait(timeout=3)


def accepted_rows_sql() -> str:
    return f"""
insert into kb_stage.kb_proposals (
  id, proposal_type, status, proposed_by_handle, proposed_by_agent_id, channel,
  source_ref, rationale, payload, reviewed_by_handle, reviewed_by_agent_id,
  reviewed_at, review_note, created_at, updated_at
) values (
  '{PROPOSAL_ID}', 'approve_claim', 'approved', 'leo', '{AGENT_ID}', 'disposable_verifier',
  'fixture://livingip-v3-contract', 'Admit exact hash-bound V3 epistemic rows.',
  '{{"apply_payload": {{"contract_version": 3}}}}'::jsonb, 'v3-contract-reviewer', '{AGENT_ID}',
  '2026-07-18T09:00:00Z', 'Accepted only for the disposable PostgreSQL contract verifier.',
  '2026-07-18T08:59:00Z', '2026-07-18T09:00:00Z'
);

insert into kb_stage.kb_proposals (
  id, proposal_type, status, proposed_by_handle, proposed_by_agent_id, channel,
  rationale, payload, created_at, updated_at
) values (
  '{PENDING_PROPOSAL_ID}', 'approve_claim', 'pending_review', 'leo', '{AGENT_ID}',
  'disposable_verifier', 'Pending proposal must not authorize canonical rows.',
  '{{"apply_payload": {{"contract_version": 3}}}}'::jsonb,
  '2026-07-18T09:01:00Z', '2026-07-18T09:01:00Z'
);

insert into public.sources (
  id, source_type, url, storage_path, excerpt, hash, created_by, captured_at, created_at,
  canonical_url, storage_uri, content_hash, access_scope, ingestion_origin,
  provenance_status, source_class, auto_promotion_policy, accepted_by_proposal_id
) values (
  '{SOURCE_ID}', 'document', 'https://example.invalid/livingip-v3-source', null,
  'Immutable epistemic provenance.', '{SOURCE_CONTENT_HASH}', '{AGENT_ID}',
  '2026-07-18T09:02:00Z', '2026-07-18T09:02:00Z',
  'https://example.invalid/livingip-v3-source', null, '{SOURCE_CONTENT_HASH}',
  'collective_internal', 'operator_upload', 'verified', 'primary_record', 'human_required',
  '{PROPOSAL_ID}'
);

insert into public.claims (
  id, type, text, status, confidence, tags, created_by, created_at, updated_at,
  owner_agent_id, proposition, body_markdown, scope, access_scope, admission_state,
  revision_criteria, revision_kind, supersedes_id, accepted_by_proposal_id
) values (
  '{PRIMARY_CLAIM_ID}', 'empirical', {sql_literal(PRIMARY_PROPOSITION)}, 'active', null,
  array['v3-contract'], '{AGENT_ID}', '2026-07-18T09:03:00Z', '2026-07-18T09:03:00Z',
  null, {sql_literal(PRIMARY_PROPOSITION)}, {sql_literal(PRIMARY_BODY)}, {sql_literal(PRIMARY_SCOPE)},
  'collective_internal', 'admitted', {sql_literal(PRIMARY_REVISION_CRITERIA)}, 'semantic',
  '{LEGACY_CLAIM_ID}', '{PROPOSAL_ID}'
), (
  '{SUPPORT_CLAIM_ID}', 'conceptual',
  'Content-hashed locators make source versions reproducible.', 'active', null,
  array['v3-contract'], '{AGENT_ID}', '2026-07-18T09:04:00Z', '2026-07-18T09:04:00Z',
  null, 'Content-hashed locators make source versions reproducible.',
  'A locator and immutable source hash identify the exact source payload used by an evidence link.',
  'Minimum disposable PostgreSQL V3 contract verification.', 'collective_internal', 'admitted',
  'Revise if the hash and locator no longer identify one exact source payload.', 'original', null,
  '{PROPOSAL_ID}'
);

insert into public.claim_evidence (
  id, claim_id, source_id, role, weight, created_by, created_at,
  excerpt, locator_json, source_content_hash, rationale, polarity, accepted_from_proposal_id
) values (
  '{EVIDENCE_ID}', '{PRIMARY_CLAIM_ID}', '{SOURCE_ID}', 'grounds', null, '{AGENT_ID}',
  '2026-07-18T09:05:00Z', 'Immutable epistemic provenance.',
  '{{"paragraph": 1, "section": "epistemic-contract"}}'::jsonb, '{SOURCE_CONTENT_HASH}',
  'The cited passage directly supports reproducible source provenance.', 'supports', '{PROPOSAL_ID}'
);

insert into public.claim_evidence_assessments (
  id, claim_id, support_tier, challenge_tier, overall_state, rationale,
  evidence_set_hash, accepted_by_proposal_id, created_at
) values (
  '{ASSESSMENT_ID}', '{PRIMARY_CLAIM_ID}', 'strong', 'none', 'support_dominant',
  'One exact primary-record source supports the bounded contract claim and no challenge is linked.',
  '{EVIDENCE_SET_HASH}', '{PROPOSAL_ID}', '2026-07-18T09:06:00Z'
);

insert into public.claim_edges (
  id, from_claim, to_claim, edge_type, weight, created_by, created_at,
  relation_type, rationale, scope_or_conditions, accepted_by_proposal_id
) values (
  '{EDGE_ID}', '{SUPPORT_CLAIM_ID}', '{PRIMARY_CLAIM_ID}', 'supports', null, '{AGENT_ID}',
  '2026-07-18T09:07:00Z', 'supports',
  'Reproducible source versions are a prerequisite for exact evidence provenance.',
  'Within the minimum V3 epistemic storage contract.', '{PROPOSAL_ID}'
);
"""


def round_trip_sql() -> str:
    return f"""
select jsonb_build_object(
  'proposal', (select to_jsonb(row_data) from (
    select id, proposal_type, status, reviewed_by_handle, reviewed_by_agent_id,
           reviewed_at, review_note
      from kb_stage.kb_proposals where id = '{PROPOSAL_ID}'::uuid
  ) row_data),
  'claim', (select to_jsonb(row_data) from (
    select id, type, text, proposition, body_markdown, scope, access_scope,
           admission_state, revision_criteria, revision_kind, supersedes_id,
           accepted_by_proposal_id
      from public.claims where id = '{PRIMARY_CLAIM_ID}'::uuid
  ) row_data),
  'support_claim', (select to_jsonb(row_data) from (
    select id, admission_state, revision_kind, accepted_by_proposal_id
      from public.claims where id = '{SUPPORT_CLAIM_ID}'::uuid
  ) row_data),
  'source', (select to_jsonb(row_data) from (
    select id, canonical_url, storage_uri, hash, content_hash, access_scope,
           ingestion_origin, provenance_status, source_class, auto_promotion_policy,
           accepted_by_proposal_id
      from public.sources where id = '{SOURCE_ID}'::uuid
  ) row_data),
  'evidence', (select to_jsonb(row_data) from (
    select id, claim_id, source_id, role, excerpt, locator_json, source_content_hash,
           rationale, polarity, accepted_from_proposal_id
      from public.claim_evidence where id = '{EVIDENCE_ID}'::uuid
  ) row_data),
  'assessment', (select to_jsonb(row_data) from (
    select id, claim_id, support_tier, challenge_tier, overall_state, rationale,
           evidence_set_hash, accepted_by_proposal_id
      from public.claim_evidence_assessments where id = '{ASSESSMENT_ID}'::uuid
  ) row_data),
  'edge', (select to_jsonb(row_data) from (
    select id, from_claim, to_claim, relation_type, rationale, scope_or_conditions,
           accepted_by_proposal_id
      from public.claim_edges where id = '{EDGE_ID}'::uuid
  ) row_data)
)::text;
"""


def adversarial_checks(postgres: DisposablePostgres) -> dict[str, Any]:
    valid_source_columns = """
      source_type, url, hash, canonical_url, content_hash, access_scope, ingestion_origin,
      provenance_status, source_class, auto_promotion_policy, accepted_by_proposal_id
    """
    valid_source_values = f"""
      'document', 'https://example.invalid/pending', '{SOURCE_CONTENT_HASH}',
      'https://example.invalid/pending', '{SOURCE_CONTENT_HASH}', 'collective_internal',
      'operator_upload', 'verified', 'primary_record', 'human_required', '{PENDING_PROPOSAL_ID}'
    """
    checks = {
        "new_legacy_claim_cannot_bypass_v3_contract": expect_failure(
            postgres,
            """
insert into public.claims (id, type, text, status)
values ('50000000-0000-4000-8000-000000000011', 'structural', 'Legacy bypass.', 'open');
""",
            expected_sqlstate="23514",
        ),
        "new_unlinked_source_cannot_bypass_v3_contract": expect_failure(
            postgres,
            f"""
insert into public.sources (id, source_type, hash)
values ('50000000-0000-4000-8000-000000000012', 'document', '{SOURCE_CONTENT_HASH}');
""",
            expected_sqlstate="23514",
        ),
        "new_unlinked_evidence_cannot_bypass_v3_contract": expect_failure(
            postgres,
            f"""
insert into public.claim_evidence (id, claim_id, source_id, role)
values (
  '50000000-0000-4000-8000-000000000013', '{LEGACY_CLAIM_ID}', '{LEGACY_SOURCE_ID}', 'grounds'
);
""",
            expected_sqlstate="23514",
        ),
        "new_unlinked_edge_cannot_bypass_v3_contract": expect_failure(
            postgres,
            f"""
insert into public.claim_edges (id, from_claim, to_claim, edge_type)
values (
  '50000000-0000-4000-8000-000000000014',
  '{LEGACY_CLAIM_ID}', '{LEGACY_RELATED_CLAIM_ID}', 'relates'
);
""",
            expected_sqlstate="23514",
        ),
        "pending_proposal_cannot_authorize_source": expect_failure(
            postgres,
            f"insert into public.sources (id, {valid_source_columns}) "
            f"values ('50000000-0000-4000-8000-000000000001', {valid_source_values});",
            expected_sqlstate="23514",
        ),
        "applied_status_without_apply_receipt_cannot_authorize_source": expect_failure(
            postgres,
            f"""
insert into public.sources (id, {valid_source_columns}) values (
  '50000000-0000-4000-8000-000000000015',
  'document', 'https://example.invalid/missing-apply-receipt', '{SOURCE_CONTENT_HASH}',
  'https://example.invalid/missing-apply-receipt', '{SOURCE_CONTENT_HASH}', 'collective_internal',
  'operator_upload', 'verified', 'primary_record', 'human_required',
  '{APPLIED_WITHOUT_RECEIPT_PROPOSAL_ID}'
);
""",
            expected_sqlstate="23514",
        ),
        "new_applied_proposal_without_receipt_is_rejected": expect_failure(
            postgres,
            """
insert into kb_stage.kb_proposals (
  id, proposal_type, status, rationale, payload, reviewed_by_handle, reviewed_at, review_note
) values (
  '30000000-0000-4000-8000-000000000004', 'approve_claim', 'applied',
  'Invalid new apply receipt.', '{"apply_payload": {"contract_version": 3}}'::jsonb,
  'v3-contract-reviewer', '2026-07-18T09:00:00Z', 'Missing applicator and applied_at.'
);
""",
            expected_sqlstate="23514",
        ),
        "non_applied_proposal_cannot_carry_apply_metadata": expect_failure(
            postgres,
            f"""
update kb_stage.kb_proposals
   set applied_by_handle = 'premature-applicator', applied_at = '2026-07-18T09:00:00Z'
 where id = '{PENDING_PROPOSAL_ID}'::uuid;
""",
            expected_sqlstate="55000",
        ),
        "admitted_claim_requires_body_scope_and_revision_criteria": expect_failure(
            postgres,
            f"""
insert into public.claims (
  id, type, text, status, proposition, body_markdown, scope, access_scope,
  admission_state, revision_criteria, revision_kind, accepted_by_proposal_id
) values (
  '50000000-0000-4000-8000-000000000002', 'empirical', 'Incomplete admitted claim.', 'active',
  'Incomplete admitted claim.', null, 'Verifier scope.', 'collective_internal',
  'admitted', 'Revise when incomplete.', 'original', '{PROPOSAL_ID}'
);
""",
            expected_sqlstate="23514",
        ),
        "null_revision_kind_cannot_bypass_claim_contract": expect_failure(
            postgres,
            f"""
insert into public.claims (
  id, type, text, status, proposition, body_markdown, scope, access_scope,
  admission_state, revision_criteria, revision_kind, accepted_by_proposal_id
) values (
  '50000000-0000-4000-8000-000000000007', 'empirical', 'Null revision kind claim.', 'active',
  'Null revision kind claim.', 'A complete body cannot compensate for a null required revision kind.',
  'Verifier scope.', 'collective_internal', 'admitted', 'Revise when null checks fail.', null, '{PROPOSAL_ID}'
);
""",
            expected_sqlstate="23514",
        ),
        "null_source_classification_cannot_bypass_source_contract": expect_failure(
            postgres,
            f"""
insert into public.sources (
  id, source_type, url, hash, canonical_url, content_hash, access_scope, ingestion_origin,
  provenance_status, source_class, auto_promotion_policy, accepted_by_proposal_id
) values (
  '50000000-0000-4000-8000-000000000008', 'document', 'https://example.invalid/null-class',
  '{SOURCE_CONTENT_HASH}', 'https://example.invalid/null-class', '{SOURCE_CONTENT_HASH}',
  'collective_internal', 'operator_upload', 'verified', null, 'human_required', '{PROPOSAL_ID}'
);
""",
            expected_sqlstate="23514",
        ),
        "evidence_hash_must_match_source_version": expect_failure(
            postgres,
            f"""
insert into public.claim_evidence (
  id, claim_id, source_id, role, weight, locator_json, source_content_hash,
  rationale, polarity, accepted_from_proposal_id
) values (
  '50000000-0000-4000-8000-000000000003', '{PRIMARY_CLAIM_ID}', '{SOURCE_ID}', 'grounds', null,
  '{{"paragraph": 1}}'::jsonb, '{"0" * 64}', 'Mismatched source hash.', 'supports', '{PROPOSAL_ID}'
);
""",
            expected_sqlstate="23514",
        ),
        "evidence_requires_locator_and_rationale": expect_failure(
            postgres,
            f"""
insert into public.claim_evidence (
  id, claim_id, source_id, role, weight, locator_json, source_content_hash,
  rationale, polarity, accepted_from_proposal_id
) values (
  '50000000-0000-4000-8000-000000000004', '{PRIMARY_CLAIM_ID}', '{SOURCE_ID}', 'grounds', null,
  '{{}}'::jsonb, '{SOURCE_CONTENT_HASH}', '', 'supports', '{PROPOSAL_ID}'
);
""",
            expected_sqlstate="23514",
        ),
        "null_polarity_cannot_bypass_evidence_contract": expect_failure(
            postgres,
            f"""
insert into public.claim_evidence (
  id, claim_id, source_id, role, weight, locator_json, source_content_hash,
  rationale, polarity, accepted_from_proposal_id
) values (
  '50000000-0000-4000-8000-000000000009', '{PRIMARY_CLAIM_ID}', '{SOURCE_ID}', 'grounds', null,
  '{{"paragraph": 1}}'::jsonb, '{SOURCE_CONTENT_HASH}', 'Null polarity must fail.', null, '{PROPOSAL_ID}'
);
""",
            expected_sqlstate="23514",
        ),
        "generic_relates_edge_is_not_admitted": expect_failure(
            postgres,
            f"""
insert into public.claim_edges (
  id, from_claim, to_claim, edge_type, weight, relation_type, rationale, accepted_by_proposal_id
) values (
  '50000000-0000-4000-8000-000000000005', '{SUPPORT_CLAIM_ID}', '{PRIMARY_CLAIM_ID}',
  'relates', null, 'relates', 'Generic relation must fail.', '{PROPOSAL_ID}'
);
""",
            expected_sqlstate="23514",
        ),
        "null_relation_type_cannot_bypass_edge_contract": expect_failure(
            postgres,
            f"""
insert into public.claim_edges (
  id, from_claim, to_claim, edge_type, weight, relation_type, rationale, accepted_by_proposal_id
) values (
  '50000000-0000-4000-8000-000000000010', '{SUPPORT_CLAIM_ID}', '{PRIMARY_CLAIM_ID}',
  'supports', null, null, 'Null relation type must fail.', '{PROPOSAL_ID}'
);
""",
            expected_sqlstate="23514",
        ),
        "truth_apt_assessment_requires_both_tiers": expect_failure(
            postgres,
            f"""
insert into public.claim_evidence_assessments (
  id, claim_id, support_tier, challenge_tier, overall_state, rationale,
  evidence_set_hash, accepted_by_proposal_id
) values (
  '50000000-0000-4000-8000-000000000006', '{PRIMARY_CLAIM_ID}', 'strong', null,
  'support_dominant', 'Missing challenge tier must fail.', '{EVIDENCE_SET_HASH}', '{PROPOSAL_ID}'
);
""",
            expected_sqlstate="23514",
        ),
        "source_hash_is_immutable": expect_failure(
            postgres,
            f"update public.sources set content_hash = '{'f' * 64}' where id = '{SOURCE_ID}'::uuid;",
            expected_sqlstate="55000",
        ),
        "source_locator_is_immutable": expect_failure(
            postgres,
            f"update public.sources set canonical_url = 'https://example.invalid/mutated' "
            f"where id = '{SOURCE_ID}'::uuid;",
            expected_sqlstate="55000",
        ),
        "source_classification_is_immutable": expect_failure(
            postgres,
            f"update public.sources set source_class = 'journalistic' where id = '{SOURCE_ID}'::uuid;",
            expected_sqlstate="55000",
        ),
        "admitted_claim_semantics_are_immutable": expect_failure(
            postgres,
            f"update public.claims set body_markdown = 'Mutated body.' where id = '{PRIMARY_CLAIM_ID}'::uuid;",
            expected_sqlstate="55000",
        ),
        "admitted_claim_status_is_immutable": expect_failure(
            postgres,
            f"update public.claims set status = 'retracted' where id = '{PRIMARY_CLAIM_ID}'::uuid;",
            expected_sqlstate="55000",
        ),
        "admitted_claim_update_timestamp_is_immutable": expect_failure(
            postgres,
            f"update public.claims set updated_at = updated_at + interval '1 second' "
            f"where id = '{PRIMARY_CLAIM_ID}'::uuid;",
            expected_sqlstate="55000",
        ),
        "admitted_claim_cannot_be_deleted": expect_failure(
            postgres,
            f"delete from public.claims where id = '{PRIMARY_CLAIM_ID}'::uuid;",
            expected_sqlstate="55000",
        ),
        "proposal_linked_source_cannot_be_deleted": expect_failure(
            postgres,
            f"delete from public.sources where id = '{SOURCE_ID}'::uuid;",
            expected_sqlstate="55000",
        ),
        "proposal_linked_evidence_cannot_be_deleted": expect_failure(
            postgres,
            f"delete from public.claim_evidence where id = '{EVIDENCE_ID}'::uuid;",
            expected_sqlstate="55000",
        ),
        "proposal_linked_edge_cannot_be_deleted": expect_failure(
            postgres,
            f"delete from public.claim_edges where id = '{EDGE_ID}'::uuid;",
            expected_sqlstate="55000",
        ),
        "evidence_assessment_cannot_be_deleted": expect_failure(
            postgres,
            f"delete from public.claim_evidence_assessments where id = '{ASSESSMENT_ID}'::uuid;",
            expected_sqlstate="55000",
        ),
        "linked_proposal_cannot_lose_accepted_decision": expect_failure(
            postgres,
            f"update kb_stage.kb_proposals set status = 'rejected' where id = '{PROPOSAL_ID}'::uuid;",
            expected_sqlstate="55000",
        ),
        "linked_proposal_decision_metadata_is_immutable": expect_failure(
            postgres,
            f"update kb_stage.kb_proposals set review_note = 'Rewritten decision.' where id = '{PROPOSAL_ID}'::uuid;",
            expected_sqlstate="55000",
        ),
    }
    return checks


def applied_receipt_adversarial_checks(postgres: DisposablePostgres) -> dict[str, Any]:
    return {
        "applied_proposal_status_is_immutable": expect_failure(
            postgres,
            f"update kb_stage.kb_proposals set status = 'approved' where id = '{PROPOSAL_ID}'::uuid;",
            expected_sqlstate="55000",
        ),
        "applied_proposal_timestamp_is_immutable": expect_failure(
            postgres,
            f"update kb_stage.kb_proposals set applied_at = applied_at + interval '1 second' "
            f"where id = '{PROPOSAL_ID}'::uuid;",
            expected_sqlstate="55000",
        ),
        "applied_proposal_applicator_is_immutable": expect_failure(
            postgres,
            f"update kb_stage.kb_proposals set applied_by_handle = 'rewritten-applicator' "
            f"where id = '{PROPOSAL_ID}'::uuid;",
            expected_sqlstate="55000",
        ),
    }


def cleanup_canary_rows_sql() -> str:
    return f"""
begin;
-- This verifier owns an isolated disposable database and runs as its superuser.
-- Production writers must never receive a role capable of disabling triggers.
set local session_replication_role = replica;
delete from public.claim_evidence_assessments where id = '{ASSESSMENT_ID}'::uuid;
delete from public.claim_edges where id = '{EDGE_ID}'::uuid;
delete from public.claim_evidence where id = '{EVIDENCE_ID}'::uuid;
delete from public.claims where id in ('{PRIMARY_CLAIM_ID}'::uuid, '{SUPPORT_CLAIM_ID}'::uuid);
delete from public.sources where id = '{SOURCE_ID}'::uuid;
delete from kb_stage.kb_proposals where id in (
  '{PROPOSAL_ID}'::uuid,
  '{PENDING_PROPOSAL_ID}'::uuid,
  '{APPLIED_WITHOUT_RECEIPT_PROPOSAL_ID}'::uuid
);
commit;
"""


def build_checks(
    receipt: dict[str, Any],
    *,
    first_catalog: dict[str, Any],
    second_catalog: dict[str, Any],
    reinstall_catalog: dict[str, Any],
    rollback_catalog: dict[str, Any],
    final_catalog: dict[str, Any],
    baseline_legacy: dict[str, Any],
    after_install_legacy: dict[str, Any],
    after_cleanup_legacy: dict[str, Any],
    after_rollback_legacy: dict[str, Any],
) -> dict[str, bool]:
    round_trip = receipt["round_trip"]
    claim = round_trip["claim"]
    source = round_trip["source"]
    evidence = round_trip["evidence"]
    assessment = round_trip["assessment"]
    edge = round_trip["edge"]
    proposal = round_trip["proposal"]
    linked_ids = {
        claim["accepted_by_proposal_id"],
        source["accepted_by_proposal_id"],
        evidence["accepted_from_proposal_id"],
        assessment["accepted_by_proposal_id"],
        edge["accepted_by_proposal_id"],
    }
    adversarial = receipt["adversarial"]
    accepted_proposal_gate = next(
        (
            function
            for function in first_catalog["functions"]
            if function["function_name"] == "teleo_v3_require_accepted_proposal" and function["arguments"] == ""
        ),
        {},
    )
    cutover_gate = next(
        (
            function
            for function in first_catalog["functions"]
            if function["function_name"] == "teleo_v3_enforce_apply_contract_cutover" and function["arguments"] == ""
        ),
        {},
    )
    cutover_trigger = next(
        (
            trigger
            for trigger in first_catalog["triggers"]
            if trigger["trigger_name"] == "teleo_v3_00_enforce_apply_contract_cutover"
        ),
        {},
    )
    return {
        "container_network_is_none": receipt["environment"]["network_mode"] == "none",
        "container_data_is_tmpfs": receipt["environment"]["tmpfs_data_dir"] is True,
        "container_has_no_volume_mounts": receipt["environment"]["docker_volume_mounts"] == [],
        "canary_cleanup_bypass_is_disposable_superuser_only": receipt["canary_cleanup"]
        == {
            "trigger_bypass": "set local session_replication_role=replica",
            "scope": "isolated-disposable-superuser-transaction",
        },
        "legacy_rows_unchanged_by_install": baseline_legacy == after_install_legacy,
        "legacy_rows_unchanged_by_canary_cleanup": baseline_legacy == after_cleanup_legacy,
        "legacy_rows_unchanged_by_rollback": baseline_legacy == after_rollback_legacy,
        "migration_is_idempotent": first_catalog == second_catalog,
        "security_definer_trigger_owner_and_acl_are_narrow": accepted_proposal_gate
        == {
            "schema_name": "kb_stage",
            "function_name": "teleo_v3_require_accepted_proposal",
            "arguments": "",
            "owner_name": "kb_gate_owner",
            "security_definer": True,
            "config": ["search_path=pg_catalog, pg_temp"],
            "source_sha256": "ca398ef2f185488c0712bd5ebc22402ab9e41e5dfd2f5823b367e6a382e8a64f",
            "language_name": "plpgsql",
            "function_kind": "f",
            "argument_count": 0,
            "return_type": "trigger",
            "owner_only_execute": True,
            "non_owner_execute_grantees": [],
        },
        "cutover_trigger_function_owner_and_acl_are_narrow": cutover_gate
        == {
            "schema_name": "kb_stage",
            "function_name": "teleo_v3_enforce_apply_contract_cutover",
            "arguments": "",
            "owner_name": "kb_gate_owner",
            "security_definer": True,
            "config": ["search_path=pg_catalog, pg_temp"],
            "source_sha256": "3ab2d5e1d5d1d18b13579fe2c0cae487036c7b147cf7f1b6b9a3633274756c18",
            "language_name": "plpgsql",
            "function_kind": "f",
            "argument_count": 0,
            "return_type": "trigger",
            "owner_only_execute": True,
            "non_owner_execute_grantees": [],
        },
        "full_trigger_catalog_is_exact": all(
            trigger_catalog_is_exact(catalog) for catalog in (first_catalog, second_catalog, reinstall_catalog)
        ),
        "cutover_trigger_is_enabled_and_exactly_bound": (
            cutover_trigger.get("schema_name") == "kb_stage"
            and cutover_trigger.get("table_name") == "kb_proposals"
            and cutover_trigger.get("enabled_state") == "O"
            and cutover_trigger.get("function_schema") == "kb_stage"
            and cutover_trigger.get("function_name") == "teleo_v3_enforce_apply_contract_cutover"
            and cutover_trigger.get("function_identity_arguments") == ""
        ),
        "claim_contract_round_trip": claim
        == {
            "id": PRIMARY_CLAIM_ID,
            "type": "empirical",
            "text": PRIMARY_PROPOSITION,
            "proposition": PRIMARY_PROPOSITION,
            "body_markdown": PRIMARY_BODY,
            "scope": PRIMARY_SCOPE,
            "access_scope": "collective_internal",
            "admission_state": "admitted",
            "revision_criteria": PRIMARY_REVISION_CRITERIA,
            "revision_kind": "semantic",
            "supersedes_id": LEGACY_CLAIM_ID,
            "accepted_by_proposal_id": PROPOSAL_ID,
        },
        "source_contract_round_trip": source["content_hash"] == SOURCE_CONTENT_HASH
        and source["hash"] == SOURCE_CONTENT_HASH
        and source["canonical_url"] == "https://example.invalid/livingip-v3-source"
        and source["provenance_status"] == "verified"
        and source["source_class"] == "primary_record",
        "evidence_contract_round_trip": evidence["claim_id"] == PRIMARY_CLAIM_ID
        and evidence["source_id"] == SOURCE_ID
        and evidence["source_content_hash"] == SOURCE_CONTENT_HASH
        and evidence["locator_json"] == {"paragraph": 1, "section": "epistemic-contract"}
        and evidence["polarity"] == "supports"
        and bool(evidence["rationale"]),
        "assessment_contract_round_trip": assessment["claim_id"] == PRIMARY_CLAIM_ID
        and assessment["support_tier"] == "strong"
        and assessment["challenge_tier"] == "none"
        and assessment["overall_state"] == "support_dominant"
        and assessment["evidence_set_hash"] == EVIDENCE_SET_HASH,
        "typed_edge_contract_round_trip": edge["from_claim"] == SUPPORT_CLAIM_ID
        and edge["to_claim"] == PRIMARY_CLAIM_ID
        and edge["relation_type"] == "supports"
        and bool(edge["rationale"]),
        "accepted_proposal_linkage_round_trip": linked_ids == {PROPOSAL_ID}
        and proposal["status"] == "approved"
        and proposal["reviewed_at"] is not None
        and bool(proposal["review_note"]),
        "accepted_proposal_lock_blocks_concurrent_downgrade": (
            receipt["proposal_decision_lock_race"]["lock_observed"] is True
            and receipt["proposal_decision_lock_race"]["concurrent_downgrade"]["refused"] is True
            and receipt["proposal_decision_lock_race"]["concurrent_downgrade"]["expected_sqlstate_observed"] is True
            and receipt["proposal_decision_lock_race"]["holder_returncode"] == 0
            and receipt["proposal_decision_lock_race"]["proposal_status_after"] == "approved"
        ),
        "all_adversarial_writes_refused": all(
            check["refused"] and check["expected_sqlstate_observed"] for check in adversarial.values()
        ),
        "approved_to_applied_transition_remains_compatible": (
            receipt["proposal_applied_transition"]["status"] == "applied"
            and receipt["proposal_applied_transition"]["review_note"] == proposal["review_note"]
            and receipt["proposal_applied_transition"]["applied_at"] is not None
        ),
        "rollback_refuses_v3_data_loss": receipt["rollback_with_data"]["refused"]
        and receipt["rollback_with_data"]["expected_sqlstate_observed"],
        "rollback_removes_contract": catalog_is_absent(rollback_catalog),
        "reinstall_matches_original_contract": reinstall_catalog == first_catalog,
        "final_rollback_removes_contract": catalog_is_absent(final_catalog),
        "rollback_is_idempotent": receipt["idempotent_rollback_returncode"] == 0,
    }


def write_receipt(output: Path, receipt: dict[str, Any]) -> None:
    output.parent.mkdir(parents=True, exist_ok=True)
    temporary = output.with_name(f".{output.name}.{uuid.uuid4().hex}.tmp")
    temporary.write_text(json.dumps(receipt, indent=2, sort_keys=True) + "\n", encoding="utf-8")
    os.replace(temporary, output)


def run_verification(output: Path) -> dict[str, Any]:
    receipt: dict[str, Any] = {
        "schema": RECEIPT_SCHEMA,
        "status": "fail",
        "required_tier": "T2_runtime",
        "current_tier": "not_verified",
        "production_access_attempted": False,
        "production_migration_shipped": False,
        "database_target_mode": "internally-created-disposable-container-only",
        "migration_files": {
            "install": str(MIGRATION_SQL.relative_to(REPO_ROOT)),
            "rollback": str(ROLLBACK_SQL.relative_to(REPO_ROOT)),
            "install_sha256": hashlib.sha256(MIGRATION_SQL.read_bytes()).hexdigest(),
            "rollback_sha256": hashlib.sha256(ROLLBACK_SQL.read_bytes()).hexdigest(),
        },
        "not_shipped": [
            "production migration execution",
            "live apply-worker wiring",
            "full V3 proposal/evaluation/identity architecture",
        ],
    }
    postgres = DisposablePostgres()
    error: str | None = None
    try:
        receipt["environment"] = postgres.start()
        postgres.psql(BOOTSTRAP_SQL)
        baseline_legacy = postgres.psql_json(LEGACY_STATE_SQL)
        receipt["legacy_baseline"] = baseline_legacy

        first_install = postgres.apply_file(MIGRATION_SQL)
        first_catalog = catalog_readback(postgres)
        after_install_legacy = postgres.psql_json(LEGACY_STATE_SQL)
        receipt["install_first"] = {
            "returncode": first_install.returncode,
            "catalog_sha256": canonical_sha256(first_catalog),
            "catalog": first_catalog,
        }

        second_install = postgres.apply_file(MIGRATION_SQL)
        second_catalog = catalog_readback(postgres)
        receipt["install_second"] = {
            "returncode": second_install.returncode,
            "catalog_sha256": canonical_sha256(second_catalog),
        }

        postgres.psql(accepted_rows_sql())
        receipt["round_trip"] = postgres.psql_json(round_trip_sql())
        receipt["proposal_decision_lock_race"] = proposal_decision_lock_race_check(postgres)
        receipt["adversarial"] = adversarial_checks(postgres)
        postgres.psql(
            f"""
update kb_stage.kb_proposals
   set status = 'applied',
       applied_by_handle = 'v3-contract-apply',
       applied_at = '2026-07-18T09:08:00Z',
       updated_at = '2026-07-18T09:08:00Z'
 where id = '{PROPOSAL_ID}'::uuid;
"""
        )
        receipt["proposal_applied_transition"] = postgres.psql_json(
            f"""
select jsonb_build_object(
  'id', id,
  'status', status,
  'review_note', review_note,
  'applied_by_handle', applied_by_handle,
  'applied_at', applied_at
)::text
from kb_stage.kb_proposals
where id = '{PROPOSAL_ID}'::uuid;
"""
        )
        receipt["adversarial"].update(applied_receipt_adversarial_checks(postgres))

        rollback_with_data = postgres.apply_file(ROLLBACK_SQL, check=False)
        receipt["rollback_with_data"] = failure_receipt(rollback_with_data, "55000")
        if not (
            receipt["rollback_with_data"]["refused"] and receipt["rollback_with_data"]["expected_sqlstate_observed"]
        ):
            raise VerificationError("rollback did not refuse proposal-linked V3 data")

        receipt["canary_cleanup"] = {
            "trigger_bypass": "set local session_replication_role=replica",
            "scope": "isolated-disposable-superuser-transaction",
        }
        postgres.psql(cleanup_canary_rows_sql())
        after_cleanup_legacy = postgres.psql_json(LEGACY_STATE_SQL)

        rollback = postgres.apply_file(ROLLBACK_SQL)
        rollback_catalog = catalog_readback(postgres)
        after_rollback_legacy = postgres.psql_json(LEGACY_STATE_SQL)
        receipt["rollback"] = {
            "returncode": rollback.returncode,
            "catalog": rollback_catalog,
            "legacy_sha256": canonical_sha256(after_rollback_legacy),
        }

        idempotent_rollback = postgres.apply_file(ROLLBACK_SQL)
        receipt["idempotent_rollback_returncode"] = idempotent_rollback.returncode

        reinstall = postgres.apply_file(MIGRATION_SQL)
        reinstall_catalog = catalog_readback(postgres)
        receipt["reinstall"] = {
            "returncode": reinstall.returncode,
            "catalog_sha256": canonical_sha256(reinstall_catalog),
        }

        final_rollback = postgres.apply_file(ROLLBACK_SQL)
        final_catalog = catalog_readback(postgres)
        receipt["final_rollback"] = {
            "returncode": final_rollback.returncode,
            "catalog": final_catalog,
        }

        receipt["checks"] = build_checks(
            receipt,
            first_catalog=first_catalog,
            second_catalog=second_catalog,
            reinstall_catalog=reinstall_catalog,
            rollback_catalog=rollback_catalog,
            final_catalog=final_catalog,
            baseline_legacy=baseline_legacy,
            after_install_legacy=after_install_legacy,
            after_cleanup_legacy=after_cleanup_legacy,
            after_rollback_legacy=after_rollback_legacy,
        )
    except Exception as exc:  # retain a bounded failure receipt before returning non-zero
        error = safe_error_text(str(exc))
        receipt["error"] = error
    finally:
        cleanup = postgres.cleanup()
        receipt["cleanup"] = cleanup
        cleanup_ok = (
            cleanup["container_absent"]
            and cleanup["name_readback_clear"]
            and cleanup["instance_label_readback_clear"]
            and cleanup["stable_absence_readback"]
            and cleanup["network_mode_was_none"]
            and cleanup["tmpfs_data_dir_was_present"]
            and cleanup["docker_volume_mounts_observed"] == []
        )
        receipt.setdefault("checks", {})["disposable_container_cleanup_verified"] = cleanup_ok
        all_checks_pass = bool(receipt["checks"]) and all(receipt["checks"].values())
        if error is None and all_checks_pass:
            receipt["status"] = "pass"
            receipt["current_tier"] = "T2_runtime_disposable_postgresql"
        receipt["strongest_claim"] = (
            "The additive minimum V3 epistemic contract installed, round-tripped accepted provenance, "
            "rejected adversarial writes, rolled back, reinstalled, and cleaned up in networkless disposable PostgreSQL."
            if receipt["status"] == "pass"
            else "The disposable V3 epistemic contract verification did not pass every runtime check."
        )
        write_receipt(output, receipt)
    return receipt


def parse_args(argv: list[str] | None = None) -> argparse.Namespace:
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument("--output", type=Path, required=True, help="retained JSON receipt path")
    return parser.parse_args(argv)


def main(argv: list[str] | None = None) -> int:
    args = parse_args(argv)
    receipt = run_verification(args.output)
    print(
        json.dumps(
            {
                "status": receipt["status"],
                "current_tier": receipt["current_tier"],
                "receipt": str(args.output.resolve()),
                "checks_passed": sum(1 for value in receipt.get("checks", {}).values() if value),
                "checks_total": len(receipt.get("checks", {})),
                "cleanup": receipt.get("checks", {}).get("disposable_container_cleanup_verified"),
                "production_access_attempted": receipt["production_access_attempted"],
            },
            sort_keys=True,
        )
    )
    return 0 if receipt["status"] == "pass" else 1


if __name__ == "__main__":
    raise SystemExit(main())