#!/usr/bin/env python3
"""Apply an approved kb_stage proposal into canonical public.* rows.

Stage 2 of the KB apply pipeline:  approve -> APPLY -> render -> surface.

Governance boundary
-------------------
This tool connects to Postgres as the narrow ``kb_apply`` login role, never as
superuser. ``kb_apply`` can insert reviewed claim/source/reasoning-tool rows and
write the strategy/evidence/edge tables required by supported contracts. It has
SELECT-only access to ``kb_stage.kb_proposals``; payload-bound SECURITY DEFINER
functions are the only way it can lock and finish an approved ledger row. It
holds no DELETE or DDL and no write access to unrelated tables (agents, budgets,
personas, ...). This enforces
"agents propose, but do not self-apply" at the database boundary rather than by
convention. The tool is an operator, invoked by a human/reviewer after approval;
it is deliberately not part of the agent runtime.

Flow
----
1. Read the proposal (must be ``status='approved'``).
2. Dispatch by ``proposal_type`` to build the canonical write from a strict
   ``apply_payload`` contract carried on the proposal payload.
3. Emit a single transaction: canonical writes, then a DO block that flips the
   ledger to ``applied`` asserting rowcount=1 (the concurrent double-apply
   guard), stamps ``applied_by_agent_id`` as a real FK, and verifies invariants
   (RAISE on violation -> rollback), COMMIT.

Prerequisites (one-time, superuser): scripts/kb_apply_prereqs.sql bootstraps the
``kb-apply`` service-agent row, grants ``kb_apply`` SELECT on public.agents, and
ensures the one-active-strategy unique index. Run it before the first apply.

``--dry-run`` prints the exact SQL and does not connect for writes.

Contract (v2)
-------------
Freeform Leo eval packets are NOT applied directly. Each proposal must carry a
strict ``apply_payload`` object; a separate normalizer produces it. Shapes:

  revise_strategy:
    {"apply_payload": {
        "agent_id": "<uuid>",
        "strategy": {"diagnosis": str, "guiding_policy": str,
                     "proximate_objectives": [ ... ]},
        "strategy_nodes": [{"node_type": "diagnosis|policy|objective",
                            "title": str, "body": str, "rank": int}, ...]}}

  add_edge:
    {"apply_payload": {"from_claim": "<uuid>", "to_claim": "<uuid>",
                       "edge_type": "<edge_type>", "weight": <0..1|null>}}

  attach_evidence:
    {"apply_payload": {"evidence": [{"claim_id": "<uuid>",
                                     "source_id": "<uuid>",
                                     "role": "<evidence_role>",
                                     "weight": <0..1|null>}, ...]}}

  approve_claim:
    {"apply_payload": {
        "agent_id": "<uuid|null>",
        "claims": [{"id": "<uuid>", "type": "<claim type>",
                    "text": str, "status": "open", "confidence": <0..1|null>,
                    "tags": [str, ...], "created_by": "<uuid|null>"}, ...],
        "sources": [{"id": "<uuid>", "source_type": "<source type>",
                     "url": str|null, "storage_path": str|null,
                     "excerpt": str|null, "hash": str,
                     "created_by": "<uuid|null>"}, ...],
        "evidence": [{"claim_id": "<uuid>", "source_id": "<uuid>",
                      "role": "<evidence role>", "weight": <0..1|null>,
                      "created_by": "<uuid|null>"}, ...],
        "edges": [{"from_claim": "<uuid>", "to_claim": "<uuid>",
                   "edge_type": "<edge type>", "weight": <0..1|null>,
                   "created_by": "<uuid|null>"}, ...],
        "reasoning_tools": [{"id": "<uuid>", "agent_id": "<uuid|null>",
                             "name": str, "description": str,
                             "category": str|null}, ...]}}

``approve_claim`` is the only source-creation contract. It is an atomic,
insert-only canonical bundle: existing IDs must match the reviewed payload
exactly, source hashes cannot alias another ID, and every row is verified before
the approved proposal ledger can flip to ``applied``.
"""

from __future__ import annotations

import argparse
import hashlib
import json
import shutil
import subprocess
import sys
import uuid
from pathlib import Path
from typing import Any

import kb_apply_replay_receipt as replay_receipt

DEFAULT_SECRETS_FILE = "/home/teleo/.hermes/profiles/leoclean/secrets/kb-apply.env"
DEFAULT_CONTAINER = "teleo-pg"
DEFAULT_DB = "teleo"
DEFAULT_HOST = "127.0.0.1"
DEFAULT_ROLE = "kb_apply"
V3_CUTOVER_FUNCTION = "kb_stage.teleo_v3_enforce_apply_contract_cutover()"
V3_CUTOVER_FUNCTION_NAME = "teleo_v3_enforce_apply_contract_cutover"
V3_CUTOVER_TRIGGER = "teleo_v3_00_enforce_apply_contract_cutover"
V3_GATE_OWNER = "kb_gate_owner"
V3_PROTECTED_ROLES = ("kb_gate_owner", "kb_review", "kb_apply")
V3_TRIGGER_PREFIX = "teleo_v3_"
V3_CATALOG_DDL_ADVISORY_LOCK_KEY = 6072343533146485505
V3_CATALOG_LOCK_FUNCTION = "teleo_v3_acquire_protected_relation_locks"
V3_CATALOG_LOCK_FUNCTION_SOURCE_SHA256 = "897345944ba0f1b2a5c01b098fb17de6f881a122e906277086fa7889e76219e8"
V3_TRIGGER_FUNCTION_SOURCE_SHA256 = {
    "teleo_v3_enforce_apply_contract_cutover": "3ab2d5e1d5d1d18b13579fe2c0cae487036c7b147cf7f1b6b9a3633274756c18",
    "teleo_v3_guard_immutable_fields": "d243741e552fe8c232440ffa5f853cf261f4b86b6f32d76f375734a2235c2d13",
    "teleo_v3_guard_linked_proposal_decision": "602ee8f81f4515b0237f09a503ecbfe4fce2dc4a799e13f833554103b0e83e32",
    "teleo_v3_reject_delete": "46190723ab4520954858dbd4bb6c9230e68e41c68b5aee3b8c2719e1416de43c",
    "teleo_v3_require_accepted_proposal": "ca398ef2f185488c0712bd5ebc22402ab9e41e5dfd2f5823b367e6a382e8a64f",
    "teleo_v3_require_linked_insert": "d8c190bd2fc90159b9a4824015c10a73a701a263bba90b67f8ccce75e454e6f8",
    "teleo_v3_validate_claim_edge": "6a7a3e9d1d5ac6e9490abc9b7b1f321bb2e0948aea2671415aad28101f4ee936",
    "teleo_v3_validate_claim_evidence": "4c697c497699760214aca99f255c215e393319fd4482d4b36efac839280c3733",
    "teleo_v3_validate_evidence_assessment": "860f939b4878abe8e2e604e03be9eeb6c23b4878af4b19f9ec42de4b821fac61",
}

# Handle of the service-agent row that performs canonical writes. Bootstrapped
# once by superuser (scripts/kb_apply_prereqs.sql); kb_apply holds SELECT-only on
# public.agents so it can resolve this into applied_by_agent_id, never INSERT.
SERVICE_AGENT_HANDLE = "kb-apply"

APPLYABLE_TYPES = ("revise_strategy", "add_edge", "attach_evidence", "approve_claim")

CLAIM_TYPES = {"empirical", "structural", "normative", "meta", "concept"}
SOURCE_TYPES = {"paper", "article", "transcript", "observation", "dataset", "post", "dm", "other"}
EVIDENCE_ROLES = {"grounds", "illustrates", "contradicts"}
EDGE_TYPES = {
    "supports",
    "challenges",
    "requires",
    "relates",
    "contradicts",
    "supersedes",
    "derives_from",
    "cites",
    "causes",
    "constrains",
    "accelerates",
}

MAX_BUNDLE_ROWS = {
    "claims": 100,
    "sources": 500,
    "evidence": 1000,
    "edges": 1000,
    "assessments": 100,
    "reasoning_tools": 50,
}


def _trigger_argument_hex(arguments: tuple[str, ...]) -> str:
    return b"".join(argument.encode("utf-8") + b"\0" for argument in arguments).hex()


def _v3_trigger_contract(
    schema_name: str,
    table_name: str,
    trigger_name: str,
    trigger_type: int,
    function_name: str,
    *,
    update_columns: tuple[str, ...] = (),
    arguments: tuple[str, ...] = (),
    security_definer: bool = False,
) -> dict[str, Any]:
    return {
        "schema_name": schema_name,
        "table_name": table_name,
        "trigger_name": trigger_name,
        "trigger_type": trigger_type,
        "update_columns": sorted(update_columns),
        "when_expression": None,
        "argument_count": len(arguments),
        "argument_hex": _trigger_argument_hex(arguments),
        "function_schema": "kb_stage",
        "function_name": function_name,
        "function_identity_arguments": "",
        "function_owner": V3_GATE_OWNER,
        "function_security_definer": security_definer,
        "function_config": ["search_path=pg_catalog, pg_temp"],
        "function_source_sha256": V3_TRIGGER_FUNCTION_SOURCE_SHA256[function_name],
        "function_owner_only_execute": True if security_definer else None,
    }


# PostgreSQL pg_trigger.tgtype is a bitmask: ROW=1, BEFORE=2, INSERT=4,
# DELETE=8, UPDATE=16. Every V3 trigger is an ordinary row-level BEFORE trigger.
V3_EXPECTED_TRIGGER_CONTRACT = (
    _v3_trigger_contract(
        "kb_stage",
        "kb_proposals",
        V3_CUTOVER_TRIGGER,
        23,
        V3_CUTOVER_FUNCTION_NAME,
        update_columns=("proposal_type", "payload", "status"),
        security_definer=True,
    ),
    _v3_trigger_contract(
        "public",
        "claims",
        "teleo_v3_00_claim_require_linked_insert",
        7,
        "teleo_v3_require_linked_insert",
        arguments=("accepted_by_proposal_id", "admission_state"),
    ),
    _v3_trigger_contract(
        "public",
        "claims",
        "teleo_v3_claim_require_accepted_proposal",
        23,
        "teleo_v3_require_accepted_proposal",
        update_columns=("accepted_by_proposal_id",),
        arguments=("accepted_by_proposal_id",),
        security_definer=True,
    ),
    _v3_trigger_contract(
        "public",
        "sources",
        "teleo_v3_00_source_require_linked_insert",
        7,
        "teleo_v3_require_linked_insert",
        arguments=("accepted_by_proposal_id",),
    ),
    _v3_trigger_contract(
        "public",
        "claims",
        "teleo_v3_claim_immutable",
        19,
        "teleo_v3_guard_immutable_fields",
        arguments=(
            "id",
            "owner_agent_id",
            "type",
            "text",
            "proposition",
            "body_markdown",
            "scope",
            "access_scope",
            "status",
            "admission_state",
            "revision_criteria",
            "revision_kind",
            "supersedes_id",
            "superseded_by",
            "confidence",
            "tags",
            "created_by",
            "created_at",
            "updated_at",
            "accepted_by_proposal_id",
        ),
    ),
    _v3_trigger_contract("public", "claims", "teleo_v3_claim_reject_delete", 11, "teleo_v3_reject_delete"),
    _v3_trigger_contract(
        "public",
        "claim_evidence",
        "teleo_v3_00_evidence_require_linked_insert",
        7,
        "teleo_v3_require_linked_insert",
        arguments=("accepted_from_proposal_id",),
    ),
    _v3_trigger_contract(
        "public",
        "sources",
        "teleo_v3_source_require_accepted_proposal",
        23,
        "teleo_v3_require_accepted_proposal",
        update_columns=("accepted_by_proposal_id",),
        arguments=("accepted_by_proposal_id",),
        security_definer=True,
    ),
    _v3_trigger_contract(
        "public",
        "claim_edges",
        "teleo_v3_00_edge_require_linked_insert",
        7,
        "teleo_v3_require_linked_insert",
        arguments=("accepted_by_proposal_id",),
    ),
    _v3_trigger_contract(
        "public",
        "sources",
        "teleo_v3_source_immutable",
        19,
        "teleo_v3_guard_immutable_fields",
        arguments=(
            "id",
            "source_type",
            "url",
            "storage_path",
            "excerpt",
            "hash",
            "canonical_url",
            "storage_uri",
            "content_hash",
            "access_scope",
            "ingestion_origin",
            "provenance_status",
            "source_class",
            "auto_promotion_policy",
            "created_by",
            "captured_at",
            "created_at",
            "accepted_by_proposal_id",
        ),
    ),
    _v3_trigger_contract("public", "sources", "teleo_v3_source_reject_delete", 11, "teleo_v3_reject_delete"),
    _v3_trigger_contract(
        "public",
        "claim_evidence",
        "teleo_v3_evidence_require_accepted_proposal",
        23,
        "teleo_v3_require_accepted_proposal",
        update_columns=("accepted_from_proposal_id",),
        arguments=("accepted_from_proposal_id",),
        security_definer=True,
    ),
    _v3_trigger_contract(
        "public",
        "claim_evidence",
        "teleo_v3_evidence_validate_provenance",
        23,
        "teleo_v3_validate_claim_evidence",
    ),
    _v3_trigger_contract(
        "public",
        "claim_evidence",
        "teleo_v3_evidence_immutable",
        19,
        "teleo_v3_guard_immutable_fields",
        arguments=(
            "id",
            "claim_id",
            "source_id",
            "role",
            "weight",
            "excerpt",
            "locator_json",
            "source_content_hash",
            "rationale",
            "polarity",
            "created_by",
            "created_at",
            "accepted_from_proposal_id",
        ),
    ),
    _v3_trigger_contract(
        "public",
        "claim_evidence",
        "teleo_v3_evidence_reject_delete",
        11,
        "teleo_v3_reject_delete",
    ),
    _v3_trigger_contract(
        "public",
        "claim_edges",
        "teleo_v3_edge_require_accepted_proposal",
        23,
        "teleo_v3_require_accepted_proposal",
        update_columns=("accepted_by_proposal_id",),
        arguments=("accepted_by_proposal_id",),
        security_definer=True,
    ),
    _v3_trigger_contract(
        "public", "claim_edges", "teleo_v3_edge_validate_endpoints", 23, "teleo_v3_validate_claim_edge"
    ),
    _v3_trigger_contract(
        "public",
        "claim_edges",
        "teleo_v3_edge_immutable",
        19,
        "teleo_v3_guard_immutable_fields",
        arguments=(
            "id",
            "from_claim",
            "to_claim",
            "edge_type",
            "weight",
            "relation_type",
            "rationale",
            "scope_or_conditions",
            "created_by",
            "created_at",
            "accepted_by_proposal_id",
        ),
    ),
    _v3_trigger_contract("public", "claim_edges", "teleo_v3_edge_reject_delete", 11, "teleo_v3_reject_delete"),
    _v3_trigger_contract(
        "public",
        "claim_evidence_assessments",
        "teleo_v3_assessment_require_accepted_proposal",
        23,
        "teleo_v3_require_accepted_proposal",
        update_columns=("accepted_by_proposal_id",),
        arguments=("accepted_by_proposal_id",),
        security_definer=True,
    ),
    _v3_trigger_contract(
        "public",
        "claim_evidence_assessments",
        "teleo_v3_assessment_validate",
        23,
        "teleo_v3_validate_evidence_assessment",
    ),
    _v3_trigger_contract(
        "public",
        "claim_evidence_assessments",
        "teleo_v3_assessment_immutable",
        19,
        "teleo_v3_guard_immutable_fields",
        arguments=(
            "id",
            "claim_id",
            "support_tier",
            "challenge_tier",
            "overall_state",
            "rationale",
            "evidence_set_hash",
            "accepted_by_proposal_id",
            "supersedes_assessment_id",
            "created_at",
        ),
    ),
    _v3_trigger_contract(
        "public",
        "claim_evidence_assessments",
        "teleo_v3_assessment_reject_delete",
        11,
        "teleo_v3_reject_delete",
    ),
    _v3_trigger_contract(
        "kb_stage",
        "kb_proposals",
        "teleo_v3_guard_linked_proposal_decision",
        19,
        "teleo_v3_guard_linked_proposal_decision",
        update_columns=(
            "proposal_type",
            "status",
            "rationale",
            "payload",
            "reviewed_by_handle",
            "reviewed_by_agent_id",
            "reviewed_at",
            "review_note",
            "applied_by_handle",
            "applied_by_agent_id",
            "applied_at",
        ),
    ),
)

V3_PROTECTED_RELATIONS = (
    ("kb_stage", "kb_proposals"),
    ("public", "claim_edges"),
    ("public", "claim_evidence"),
    ("public", "claim_evidence_assessments"),
    ("public", "claims"),
    ("public", "sources"),
)


def v3_expected_trigger_contract_json() -> str:
    return json.dumps(V3_EXPECTED_TRIGGER_CONTRACT, separators=(",", ":"), sort_keys=True)


def build_v3_contract_state_query(
    *,
    into: str | None = None,
    expected_contract_expression: str | None = None,
) -> str:
    """Return the exact catalog query shared by worker and transaction guards."""
    protected_roles = ", ".join(sql_literal(role) for role in V3_PROTECTED_ROLES)
    expected_contract = expected_contract_expression or sql_literal(v3_expected_trigger_contract_json())
    into_clause = f"\n  into {into}" if into is not None else ""
    return f"""with expected_triggers as (
  select expected.*
    from pg_catalog.jsonb_to_recordset({expected_contract}::jsonb) as expected(
      schema_name text,
      table_name text,
      trigger_name text,
      trigger_type integer,
      update_columns jsonb,
      when_expression text,
      argument_count integer,
      argument_hex text,
      function_schema text,
      function_name text,
      function_identity_arguments text,
      function_owner text,
      function_security_definer boolean,
      function_config jsonb,
      function_source_sha256 text,
      function_owner_only_execute boolean
    )
), expected_functions as (
  select distinct function_schema, function_name, function_identity_arguments,
         function_owner, function_security_definer, function_config,
         function_source_sha256, function_owner_only_execute
    from expected_triggers
), actual_v3_functions as (
  select procedure.oid, namespace.nspname as function_schema,
         procedure.proname as function_name,
         pg_catalog.pg_get_function_identity_arguments(procedure.oid) as function_identity_arguments,
         owner_role.rolname as function_owner,
         procedure.prosecdef as function_security_definer,
         pg_catalog.to_jsonb(procedure.proconfig) as function_config,
         pg_catalog.encode(
           pg_catalog.sha256(pg_catalog.convert_to(procedure.prosrc, 'UTF8')),
           'hex'
         ) as function_source_sha256,
         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 function_owner_only_execute,
         language.lanname as function_language,
         procedure.prokind as function_kind,
         procedure.pronargs as function_argument_count,
         procedure.prorettype
    from pg_catalog.pg_proc procedure
    join pg_catalog.pg_namespace namespace on namespace.oid = procedure.pronamespace
   join pg_catalog.pg_roles owner_role on owner_role.oid = procedure.proowner
    join pg_catalog.pg_language language on language.oid = procedure.prolang
   where namespace.nspname = 'kb_stage'
     and pg_catalog.left(procedure.proname, {len(V3_TRIGGER_PREFIX)}) = {sql_literal(V3_TRIGGER_PREFIX)}
     and procedure.proname <> {sql_literal(V3_CATALOG_LOCK_FUNCTION)}
), actual_v3_catalog_lock_functions as (
  select procedure.oid,
         owner_role.rolname as function_owner,
         procedure.prosecdef as function_security_definer,
         pg_catalog.to_jsonb(procedure.proconfig) as function_config,
         pg_catalog.encode(
           pg_catalog.sha256(pg_catalog.convert_to(procedure.prosrc, 'UTF8')),
           'hex'
         ) as function_source_sha256,
         language.lanname as function_language,
         procedure.prokind as function_kind,
         procedure.pronargs as function_argument_count,
         pg_catalog.pg_get_function_identity_arguments(procedure.oid) as function_identity_arguments,
         procedure.prorettype,
         exists (
           select 1
             from pg_catalog.aclexplode(
               coalesce(procedure.proacl, pg_catalog.acldefault('f', procedure.proowner))
             ) acl
             join pg_catalog.pg_roles grantee on grantee.oid = acl.grantee
            where acl.privilege_type = 'EXECUTE'
              and grantee.rolname = 'kb_apply'
         ) as apply_execute,
         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
              and acl.grantee <> (
                select role.oid from pg_catalog.pg_roles role where role.rolname = 'kb_apply'
              )
         ) as execute_acl_exact
    from pg_catalog.pg_proc procedure
    join pg_catalog.pg_namespace namespace on namespace.oid = procedure.pronamespace
    join pg_catalog.pg_roles owner_role on owner_role.oid = procedure.proowner
    join pg_catalog.pg_language language on language.oid = procedure.prolang
   where namespace.nspname = 'kb_stage'
     and procedure.proname = {sql_literal(V3_CATALOG_LOCK_FUNCTION)}
), actual_v3_triggers as (
  select trigger_row.*, table_namespace.nspname as schema_name,
         relation.relname as table_name,
         function_namespace.nspname as function_schema,
         procedure.proname as function_name,
         pg_catalog.pg_get_function_identity_arguments(procedure.oid) as function_identity_arguments,
         coalesce((
           select pg_catalog.jsonb_agg(attribute.attname order by attribute.attname)
             from pg_catalog.unnest(trigger_row.tgattr::smallint[]) update_column(attnum)
             join pg_catalog.pg_attribute attribute
               on attribute.attrelid = trigger_row.tgrelid
              and attribute.attnum = update_column.attnum
              and not attribute.attisdropped
         ), '[]'::jsonb) as update_columns,
         pg_catalog.pg_get_expr(trigger_row.tgqual, trigger_row.tgrelid, true) as when_expression,
         pg_catalog.encode(trigger_row.tgargs, 'hex') as argument_hex
    from pg_catalog.pg_trigger trigger_row
    join pg_catalog.pg_class relation on relation.oid = trigger_row.tgrelid
    join pg_catalog.pg_namespace table_namespace on table_namespace.oid = relation.relnamespace
    join pg_catalog.pg_proc procedure on procedure.oid = trigger_row.tgfoid
    join pg_catalog.pg_namespace function_namespace on function_namespace.oid = procedure.pronamespace
   where pg_catalog.left(trigger_row.tgname, {len(V3_TRIGGER_PREFIX)}) = {sql_literal(V3_TRIGGER_PREFIX)}
), function_contract as (
  select
    (select pg_catalog.count(*) from actual_v3_functions) =
      (select pg_catalog.count(*) from expected_functions)
    and not exists (
      select 1
        from expected_functions expected
       where not exists (
         select 1
           from actual_v3_functions actual
          where actual.function_schema = expected.function_schema
            and actual.function_name = expected.function_name
            and actual.function_identity_arguments = expected.function_identity_arguments
            and actual.function_owner = expected.function_owner
            and actual.function_security_definer = expected.function_security_definer
            and actual.function_config = expected.function_config
            and actual.function_source_sha256 = expected.function_source_sha256
            and (
              expected.function_owner_only_execute is null
              or actual.function_owner_only_execute = expected.function_owner_only_execute
            )
            and actual.function_language = 'plpgsql'
            and actual.function_kind = 'f'
            and actual.function_argument_count = 0
            and actual.prorettype = 'pg_catalog.trigger'::pg_catalog.regtype
       )
    ) as valid
), catalog_lock_function_contract as (
  select pg_catalog.count(*) = 1
     and pg_catalog.bool_and(
       function_owner = {sql_literal(V3_GATE_OWNER)}
       and function_security_definer
       and function_config = '["search_path=pg_catalog, pg_temp"]'::jsonb
       and function_source_sha256 = {sql_literal(V3_CATALOG_LOCK_FUNCTION_SOURCE_SHA256)}
       and function_language = 'plpgsql'
       and function_kind = 'f'
       and function_argument_count = 0
       and function_identity_arguments = ''
       and prorettype = 'pg_catalog.void'::pg_catalog.regtype
       and apply_execute
       and execute_acl_exact
     ) as valid
    from actual_v3_catalog_lock_functions
), trigger_contract as (
  select
    (select pg_catalog.count(*) from actual_v3_triggers) =
      (select pg_catalog.count(*) from expected_triggers)
    and not exists (
      select 1
        from expected_triggers expected
       where not exists (
         select 1
           from actual_v3_triggers actual
          where actual.schema_name = expected.schema_name
            and actual.table_name = expected.table_name
            and actual.tgname = expected.trigger_name
            and not actual.tgisinternal
            and actual.tgenabled = 'O'
            and actual.tgtype = expected.trigger_type
            and actual.update_columns = expected.update_columns
            and actual.when_expression is not distinct from expected.when_expression
            and actual.tgnargs = expected.argument_count
            and actual.argument_hex = expected.argument_hex
            and actual.tgconstraint = 0::pg_catalog.oid
            and not actual.tgdeferrable
            and not actual.tginitdeferred
            and actual.tgconstrrelid = 0::pg_catalog.oid
            and actual.tgparentid = 0::pg_catalog.oid
            and actual.tgoldtable is null
            and actual.tgnewtable is null
            and actual.function_schema = expected.function_schema
            and actual.function_name = expected.function_name
            and actual.function_identity_arguments = expected.function_identity_arguments
            and exists (
              select 1
                from actual_v3_functions function_row
               where function_row.oid = actual.tgfoid
                 and function_row.function_schema = expected.function_schema
                 and function_row.function_name = expected.function_name
                 and function_row.function_identity_arguments = expected.function_identity_arguments
            )
       )
    ) as valid
), catalog_state as (
  select
    exists (select 1 from actual_v3_functions)
      or exists (select 1 from actual_v3_catalog_lock_functions)
      or exists (select 1 from actual_v3_triggers) as artifacts_present,
    (select valid from function_contract) as functions_valid,
    (select valid from catalog_lock_function_contract) as catalog_lock_function_valid,
    (select valid from trigger_contract) as triggers_valid,
    exists (
      select 1
        from pg_catalog.pg_roles owner_role
       where owner_role.rolname = {sql_literal(V3_GATE_OWNER)}
         and not owner_role.rolcanlogin
         and not owner_role.rolinherit
         and not owner_role.rolsuper
         and not owner_role.rolcreatedb
         and not owner_role.rolcreaterole
         and not owner_role.rolreplication
         and not owner_role.rolbypassrls
    ) as owner_role_valid,
    not exists (
      select 1
        from pg_catalog.pg_auth_members membership
        join pg_catalog.pg_roles member_role on member_role.oid = membership.member
        join pg_catalog.pg_roles granted_role on granted_role.oid = membership.roleid
       where member_role.rolname in ({protected_roles})
          or granted_role.rolname in ({protected_roles})
    ) as memberships_valid
)
select case
  when not artifacts_present then 'V2'
  when functions_valid and catalog_lock_function_valid and triggers_valid
    and owner_role_valid and memberships_valid then 'V3'
  else 'INVALID'
end{into_clause}
from catalog_state"""


def _v3_contract_lock_sql() -> str:
    statements: list[str] = []
    for schema_name, table_name in V3_PROTECTED_RELATIONS:
        if (schema_name, table_name) == ("kb_stage", "kb_proposals"):
            statements.append(f"select kb_stage.{V3_CATALOG_LOCK_FUNCTION}();")
        else:
            statements.append(f"lock table {schema_name}.{table_name} in row exclusive mode;")
    return "\n".join(statements)


def _v3_contract_transaction_guard_sql() -> str:
    state_query = build_v3_contract_state_query(into="v_contract_state")
    return f"""{_v3_contract_lock_sql()}
do $v3_contract_guard$
declare
  v_contract_state text;
begin
{state_query};
  if v_contract_state <> 'V3' then
    raise exception using
      errcode = '55000',
      message = pg_catalog.format(
        'apply_proposal: V3 catalog contract is %s; refusing mutating transaction',
        v_contract_state
      );
  end if;
end
$v3_contract_guard$;"""


V3_CLAIM_TYPES = {"empirical", "predictive", "causal", "normative", "conceptual"}
V3_ACCESS_SCOPES = {"agent_private", "collective_internal", "public"}
V3_SOURCE_ORIGINS = {"operator_upload", "agent_fetch", "contributor", "system_import", "unknown"}
V3_PROVENANCE_STATES = {"verified", "unverified", "disputed"}
V3_SOURCE_CLASSES = {
    "primary_record",
    "peer_reviewed",
    "institutional",
    "journalistic",
    "self_published",
    "unknown",
}
V3_AUTO_PROMOTION_POLICIES = {"eligible", "human_required", "blocked"}
V3_EVIDENCE_POLARITIES = {"supports", "challenges", "illustrates"}
V3_RELATION_TYPES = {"supports", "challenges", "depends_on", "contradicts"}
V3_SUPPORT_TIERS = {"none", "weak", "moderate", "strong"}
V3_ASSESSMENT_STATES = {"insufficient", "support_dominant", "challenge_dominant", "mixed"}


# --------------------------------------------------------------------------- #
# SQL helpers                                                                  #
# --------------------------------------------------------------------------- #
def sql_literal(value: Any) -> str:
    """Render a Python value as an explicit PostgreSQL escape-string literal."""
    if value is None:
        return "null"
    if isinstance(value, bool):
        return "true" if value else "false"
    if isinstance(value, (int, float)):
        return repr(value)
    escaped = str(value).replace("\\", "\\\\").replace("'", "''")
    return "E'" + escaped + "'"


def _jsonb(value: Any) -> str:
    return sql_literal(json.dumps(value, sort_keys=True)) + "::jsonb"


def deterministic_row_uuid(proposal_id: str, row_kind: str, *semantic_key: Any) -> str:
    """Derive a stable persisted row id from the proposal and semantic key.

    Evidence and edge ids are not part of the reviewed v2 payload.  Deriving
    them here makes clone/production receipts and a pre-authorized rollback
    packet bind the same exact ids instead of relying on database randomness.
    """
    material = json.dumps(
        ["livingip", "kb-apply", str(proposal_id), row_kind, *semantic_key],
        ensure_ascii=True,
        separators=(",", ":"),
    )
    return str(uuid.uuid5(uuid.NAMESPACE_URL, material))


def _text_array(values: list[str]) -> str:
    if not values:
        return "'{}'::text[]"
    return "array[" + ", ".join(sql_literal(value) for value in values) + "]::text[]"


def _uuid(value: Any, path: str, *, nullable: bool = False) -> str | None:
    if value is None and nullable:
        return None
    try:
        return str(uuid.UUID(str(value)))
    except (TypeError, ValueError, AttributeError) as exc:
        raise ValueError(f"{path} must be a UUID{' or null' if nullable else ''}") from exc


def _v3_uuid(value: Any, path: str, *, nullable: bool = False) -> str | None:
    """Require the reviewed V3 spelling to equal PostgreSQL's UUID text form."""
    canonical = _uuid(value, path, nullable=nullable)
    if canonical is None:
        return None
    if not isinstance(value, str) or value != canonical:
        raise ValueError(f"{path} must be a canonical lowercase hyphenated UUID")
    return canonical


def _weight(value: Any, path: str) -> Any:
    if value is None:
        return None
    if isinstance(value, bool) or not isinstance(value, (int, float)):
        raise ValueError(f"{path} must be a number from 0 to 1 or null")
    if not 0 <= value <= 1:
        raise ValueError(f"{path} must be between 0 and 1")
    return value


def _required_text(value: Any, path: str) -> str:
    if not isinstance(value, str) or not value.strip():
        raise ValueError(f"{path} must be a non-empty string")
    return value


def _optional_text(value: Any, path: str) -> str | None:
    if value is None:
        return None
    if not isinstance(value, str):
        raise ValueError(f"{path} must be a string or null")
    return value


def _sha256(value: Any, path: str) -> str:
    text = _required_text(value, path)
    if len(text) != 64 or any(character not in "0123456789abcdef" for character in text):
        raise ValueError(f"{path} must be a lowercase SHA-256 hex digest")
    return text


def _utc_timestamp(value: Any, path: str) -> str:
    try:
        return replay_receipt.canonical_utc_timestamp(value, path=path)
    except ValueError as exc:
        raise ValueError(f"{path} must be a timezone-aware ISO-8601 timestamp") from exc


def _json_object(value: Any, path: str, *, nonempty: bool = False) -> dict[str, Any]:
    if not isinstance(value, dict) or (nonempty and not value):
        qualifier = "non-empty " if nonempty else ""
        raise ValueError(f"{path} must be a {qualifier}object")
    return value


def _rows(payload: dict[str, Any], key: str, *, required: bool = False) -> list[dict[str, Any]]:
    value = payload.get(key)
    if value is None:
        value = []
    if not isinstance(value, list) or any(not isinstance(row, dict) for row in value):
        raise ValueError(f"approve_claim apply_payload.{key} must be a list of objects")
    if required and not value:
        raise ValueError(f"approve_claim apply_payload requires non-empty '{key}'")
    if len(value) > MAX_BUNDLE_ROWS[key]:
        raise ValueError(f"approve_claim apply_payload.{key} exceeds {MAX_BUNDLE_ROWS[key]} rows")
    return value


def _dedupe(values: list[Any], path: str) -> None:
    seen = set()
    for value in values:
        marker = json.dumps(value, sort_keys=True)
        if marker in seen:
            raise ValueError(f"{path} contains a duplicate: {value!r}")
        seen.add(marker)


def _reject_unknown(mapping: dict[str, Any], allowed: set[str], path: str) -> None:
    unknown = sorted(set(mapping) - allowed)
    if unknown:
        raise ValueError(f"{path} contains unknown fields: {unknown}")


def _v3_evidence_set_hash(claim_id: str, rows: list[dict[str, Any]]) -> str:
    evidence = [
        {
            "source_id": row["source_id"],
            "source_content_hash": row["source_content_hash"],
            "polarity": row["polarity"],
            "locator": row["locator_json"],
        }
        for row in rows
        if row["claim_id"] == claim_id
    ]
    evidence.sort(key=lambda row: json.dumps(row, separators=(",", ":"), sort_keys=True))
    payload = json.dumps(
        {"claim_id": claim_id, "evidence": evidence},
        separators=(",", ":"),
        sort_keys=True,
    )
    return hashlib.sha256(payload.encode("utf-8")).hexdigest()


def _normalize_v3_approve_claim(apply_payload: dict[str, Any]) -> dict[str, Any]:
    _reject_unknown(
        apply_payload,
        {
            "contract_version",
            "source_proposal_id",
            "agent_id",
            "claims",
            "sources",
            "evidence",
            "edges",
            "assessments",
            "reasoning_tools",
            "normalization_manifest",
        },
        "approve_claim apply_payload",
    )
    if apply_payload.get("contract_version") != 3:
        raise ValueError("V3 approve_claim apply_payload.contract_version must equal 3")
    for key in ("claims", "sources", "evidence", "edges", "assessments", "reasoning_tools"):
        if key not in apply_payload:
            raise ValueError(f"V3 approve_claim apply_payload requires explicit '{key}' collection")
    if apply_payload.get("reasoning_tools") != []:
        raise ValueError("V3 approve_claim does not admit reasoning_tools; use a separately reviewed capability")
    if apply_payload.get("source_proposal_id") is not None:
        _v3_uuid(apply_payload["source_proposal_id"], "approve_claim.source_proposal_id")
    manifest = apply_payload.get("normalization_manifest")
    if manifest is not None:
        _json_object(manifest, "approve_claim.normalization_manifest")

    agent_id = _v3_uuid(apply_payload.get("agent_id"), "approve_claim.agent_id", nullable=True)
    raw_claims = _rows(apply_payload, "claims", required=True)
    raw_sources = _rows(apply_payload, "sources", required=True)
    raw_evidence = _rows(apply_payload, "evidence", required=True)
    raw_edges = _rows(apply_payload, "edges")
    raw_assessments = _rows(apply_payload, "assessments", required=True)

    claims: list[dict[str, Any]] = []
    for index, row in enumerate(raw_claims):
        path = f"approve_claim.claims[{index}]"
        _reject_unknown(
            row,
            {
                "id",
                "owner_agent_id",
                "type",
                "proposition",
                "body_markdown",
                "scope",
                "access_scope",
                "status",
                "revision_criteria",
                "revision_kind",
                "supersedes_id",
                "tags",
                "created_by",
            },
            path,
        )
        claim_type = row.get("type")
        if claim_type not in V3_CLAIM_TYPES:
            raise ValueError(f"{path}.type must be one of {sorted(V3_CLAIM_TYPES)}")
        status = row.get("status", "active")
        if status not in {"active", "contested"}:
            raise ValueError(f"{path}.status must be active or contested for an original admission")
        revision_kind = row.get("revision_kind", "original")
        if revision_kind != "original" or row.get("supersedes_id") is not None:
            raise ValueError("V3 minimum writer admits original claims only; revision transition is not yet proven")
        access_scope = row.get("access_scope")
        if access_scope not in V3_ACCESS_SCOPES:
            raise ValueError(f"{path}.access_scope must be one of {sorted(V3_ACCESS_SCOPES)}")
        owner_agent_id = _v3_uuid(row.get("owner_agent_id"), f"{path}.owner_agent_id", nullable=True)
        if access_scope == "agent_private" and owner_agent_id is None:
            raise ValueError(f"{path}.owner_agent_id is required for agent_private claims")
        created_by = _v3_uuid(row.get("created_by", agent_id), f"{path}.created_by", nullable=True)
        if created_by is None:
            raise ValueError(f"{path}.created_by or apply_payload.agent_id is required")
        tags = row.get("tags") or []
        if not isinstance(tags, list) or any(not isinstance(tag, str) or not tag.strip() for tag in tags):
            raise ValueError(f"{path}.tags must be a list of non-empty strings")
        claims.append(
            {
                "id": _v3_uuid(row.get("id"), f"{path}.id"),
                "owner_agent_id": owner_agent_id,
                "type": claim_type,
                "proposition": _required_text(row.get("proposition"), f"{path}.proposition"),
                "body_markdown": _required_text(row.get("body_markdown"), f"{path}.body_markdown"),
                "scope": _required_text(row.get("scope"), f"{path}.scope"),
                "access_scope": access_scope,
                "status": status,
                "revision_criteria": _required_text(row.get("revision_criteria"), f"{path}.revision_criteria"),
                "revision_kind": revision_kind,
                "supersedes_id": None,
                "tags": tags,
                "created_by": created_by,
            }
        )

    sources: list[dict[str, Any]] = []
    for index, row in enumerate(raw_sources):
        path = f"approve_claim.sources[{index}]"
        _reject_unknown(
            row,
            {
                "id",
                "source_type",
                "canonical_url",
                "storage_uri",
                "excerpt",
                "content_hash",
                "access_scope",
                "ingestion_origin",
                "provenance_status",
                "source_class",
                "auto_promotion_policy",
                "captured_at",
                "created_by",
            },
            path,
        )
        source_type = row.get("source_type")
        if source_type not in SOURCE_TYPES:
            raise ValueError(f"{path}.source_type must be one of {sorted(SOURCE_TYPES)}")
        canonical_url = _optional_text(row.get("canonical_url"), f"{path}.canonical_url")
        storage_uri = _optional_text(row.get("storage_uri"), f"{path}.storage_uri")
        if not (canonical_url and canonical_url.strip()) and not (storage_uri and storage_uri.strip()):
            raise ValueError(f"{path} requires canonical_url or storage_uri")
        access_scope = row.get("access_scope")
        if access_scope not in V3_ACCESS_SCOPES:
            raise ValueError(f"{path}.access_scope must be one of {sorted(V3_ACCESS_SCOPES)}")
        ingestion_origin = row.get("ingestion_origin")
        if ingestion_origin not in V3_SOURCE_ORIGINS:
            raise ValueError(f"{path}.ingestion_origin must be one of {sorted(V3_SOURCE_ORIGINS)}")
        provenance_status = row.get("provenance_status")
        if provenance_status not in V3_PROVENANCE_STATES:
            raise ValueError(f"{path}.provenance_status must be one of {sorted(V3_PROVENANCE_STATES)}")
        source_class = row.get("source_class")
        if source_class not in V3_SOURCE_CLASSES:
            raise ValueError(f"{path}.source_class must be one of {sorted(V3_SOURCE_CLASSES)}")
        auto_promotion_policy = row.get("auto_promotion_policy")
        if auto_promotion_policy not in V3_AUTO_PROMOTION_POLICIES:
            raise ValueError(f"{path}.auto_promotion_policy must be one of {sorted(V3_AUTO_PROMOTION_POLICIES)}")
        created_by = _v3_uuid(row.get("created_by", agent_id), f"{path}.created_by", nullable=True)
        if created_by is None:
            raise ValueError(f"{path}.created_by or apply_payload.agent_id is required")
        sources.append(
            {
                "id": _v3_uuid(row.get("id"), f"{path}.id"),
                "source_type": source_type,
                "canonical_url": canonical_url,
                "storage_uri": storage_uri,
                "excerpt": _optional_text(row.get("excerpt"), f"{path}.excerpt"),
                "content_hash": _sha256(row.get("content_hash"), f"{path}.content_hash"),
                "access_scope": access_scope,
                "ingestion_origin": ingestion_origin,
                "provenance_status": provenance_status,
                "source_class": source_class,
                "auto_promotion_policy": auto_promotion_policy,
                "captured_at": _utc_timestamp(row.get("captured_at"), f"{path}.captured_at"),
                "created_by": created_by,
            }
        )

    claim_ids = {row["id"] for row in claims}
    source_by_id = {row["id"]: row for row in sources}
    evidence: list[dict[str, Any]] = []
    polarity_to_role = {"supports": "grounds", "challenges": "contradicts", "illustrates": "illustrates"}
    for index, row in enumerate(raw_evidence):
        path = f"approve_claim.evidence[{index}]"
        _reject_unknown(
            row,
            {
                "id",
                "claim_id",
                "source_id",
                "polarity",
                "excerpt",
                "locator_json",
                "source_content_hash",
                "rationale",
                "created_by",
            },
            path,
        )
        claim_id = _v3_uuid(row.get("claim_id"), f"{path}.claim_id")
        source_id = _v3_uuid(row.get("source_id"), f"{path}.source_id")
        if claim_id not in claim_ids or source_id not in source_by_id:
            raise ValueError(f"{path} must link payload-owned claim and source rows")
        polarity = row.get("polarity")
        if polarity not in V3_EVIDENCE_POLARITIES:
            raise ValueError(f"{path}.polarity must be one of {sorted(V3_EVIDENCE_POLARITIES)}")
        source_content_hash = _sha256(row.get("source_content_hash"), f"{path}.source_content_hash")
        if source_content_hash != source_by_id[source_id]["content_hash"]:
            raise ValueError(f"{path}.source_content_hash does not match the payload source")
        created_by = _v3_uuid(row.get("created_by", agent_id), f"{path}.created_by", nullable=True)
        if created_by is None:
            raise ValueError(f"{path}.created_by or apply_payload.agent_id is required")
        evidence.append(
            {
                "id": _v3_uuid(row.get("id"), f"{path}.id"),
                "claim_id": claim_id,
                "source_id": source_id,
                "role": polarity_to_role[polarity],
                "polarity": polarity,
                "excerpt": _optional_text(row.get("excerpt"), f"{path}.excerpt"),
                "locator_json": _json_object(row.get("locator_json"), f"{path}.locator_json", nonempty=True),
                "source_content_hash": source_content_hash,
                "rationale": _required_text(row.get("rationale"), f"{path}.rationale"),
                "created_by": created_by,
            }
        )

    edges: list[dict[str, Any]] = []
    for index, row in enumerate(raw_edges):
        path = f"approve_claim.edges[{index}]"
        _reject_unknown(
            row,
            {
                "id",
                "from_claim",
                "to_claim",
                "relation_type",
                "rationale",
                "scope_or_conditions",
                "created_by",
            },
            path,
        )
        from_claim = _v3_uuid(row.get("from_claim"), f"{path}.from_claim")
        to_claim = _v3_uuid(row.get("to_claim"), f"{path}.to_claim")
        relation_type = row.get("relation_type")
        if relation_type not in V3_RELATION_TYPES:
            raise ValueError(f"{path}.relation_type must be one of {sorted(V3_RELATION_TYPES)}")
        if from_claim == to_claim:
            raise ValueError(f"{path} cannot be a self-edge")
        if not ({from_claim, to_claim} & claim_ids):
            raise ValueError(f"{path} must involve at least one payload-owned claim")
        if relation_type == "contradicts" and from_claim > to_claim:
            raise ValueError(f"{path} contradicts endpoints must use canonical UUID order")
        created_by = _v3_uuid(row.get("created_by", agent_id), f"{path}.created_by", nullable=True)
        if created_by is None:
            raise ValueError(f"{path}.created_by or apply_payload.agent_id is required")
        edges.append(
            {
                "id": _v3_uuid(row.get("id"), f"{path}.id"),
                "from_claim": from_claim,
                "to_claim": to_claim,
                "relation_type": relation_type,
                "edge_type": "requires" if relation_type == "depends_on" else relation_type,
                "rationale": _required_text(row.get("rationale"), f"{path}.rationale"),
                "scope_or_conditions": _optional_text(row.get("scope_or_conditions"), f"{path}.scope_or_conditions"),
                "created_by": created_by,
            }
        )

    assessments: list[dict[str, Any]] = []
    claim_type_by_id = {row["id"]: row["type"] for row in claims}
    for index, row in enumerate(raw_assessments):
        path = f"approve_claim.assessments[{index}]"
        _reject_unknown(
            row,
            {
                "id",
                "claim_id",
                "support_tier",
                "challenge_tier",
                "overall_state",
                "rationale",
                "evidence_set_hash",
                "supersedes_assessment_id",
            },
            path,
        )
        claim_id = _v3_uuid(row.get("claim_id"), f"{path}.claim_id")
        if claim_id not in claim_ids:
            raise ValueError(f"{path}.claim_id must identify a payload-owned claim")
        support_tier = row.get("support_tier")
        challenge_tier = row.get("challenge_tier")
        if support_tier is not None and support_tier not in V3_SUPPORT_TIERS:
            raise ValueError(f"{path}.support_tier must be null or one of {sorted(V3_SUPPORT_TIERS)}")
        if challenge_tier is not None and challenge_tier not in V3_SUPPORT_TIERS:
            raise ValueError(f"{path}.challenge_tier must be null or one of {sorted(V3_SUPPORT_TIERS)}")
        if claim_type_by_id[claim_id] in {"empirical", "predictive", "causal"} and (
            support_tier is None or challenge_tier is None
        ):
            raise ValueError(f"{path} requires support_tier and challenge_tier for a truth-apt claim")
        overall_state = row.get("overall_state")
        if overall_state not in V3_ASSESSMENT_STATES:
            raise ValueError(f"{path}.overall_state must be one of {sorted(V3_ASSESSMENT_STATES)}")
        evidence_set_hash = _sha256(row.get("evidence_set_hash"), f"{path}.evidence_set_hash")
        if evidence_set_hash != _v3_evidence_set_hash(claim_id, evidence):
            raise ValueError(f"{path}.evidence_set_hash does not match the accepted evidence set")
        supersedes_assessment_id = _v3_uuid(
            row.get("supersedes_assessment_id"), f"{path}.supersedes_assessment_id", nullable=True
        )
        if supersedes_assessment_id is not None:
            raise ValueError("V3 minimum writer admits first assessments only; assessment supersession is not proven")
        assessments.append(
            {
                "id": _v3_uuid(row.get("id"), f"{path}.id"),
                "claim_id": claim_id,
                "support_tier": support_tier,
                "challenge_tier": challenge_tier,
                "overall_state": overall_state,
                "rationale": _required_text(row.get("rationale"), f"{path}.rationale"),
                "evidence_set_hash": evidence_set_hash,
                "supersedes_assessment_id": None,
            }
        )

    _dedupe([row["id"] for row in claims], "V3 approve_claim claims ids")
    _dedupe([row["id"] for row in sources], "V3 approve_claim sources ids")
    _dedupe([row["content_hash"] for row in sources], "V3 approve_claim source hashes")
    _dedupe([row["id"] for row in evidence], "V3 approve_claim evidence ids")
    _dedupe(
        [[row["claim_id"], row["source_id"], row["polarity"]] for row in evidence],
        "V3 approve_claim evidence keys",
    )
    _dedupe([row["id"] for row in edges], "V3 approve_claim edge ids")
    _dedupe(
        [[row["from_claim"], row["to_claim"], row["relation_type"]] for row in edges],
        "V3 approve_claim edge keys",
    )
    _dedupe([row["id"] for row in assessments], "V3 approve_claim assessment ids")
    _dedupe([row["claim_id"] for row in assessments], "V3 approve_claim assessment claim ids")
    assessed_claims = {row["claim_id"] for row in assessments}
    if assessed_claims != claim_ids:
        raise ValueError("V3 approve_claim requires exactly one evidence assessment for every admitted claim")

    return {
        "agent_id": agent_id,
        "claims": claims,
        "sources": sources,
        "evidence": evidence,
        "edges": edges,
        "assessments": assessments,
    }


def _validate_approval_meta(approval: dict[str, Any], proposal_type: str, apply_payload: dict[str, Any]) -> None:
    if not isinstance(approval, dict):
        raise ValueError("approved proposal metadata is required")
    if approval.get("proposal_type") != proposal_type:
        raise ValueError("approved proposal_type does not match the apply builder")
    payload = approval.get("payload")
    if not isinstance(payload, dict) or payload.get("apply_payload") != apply_payload:
        raise ValueError("approved proposal payload does not match the strict apply payload")
    if not str(approval.get("reviewed_by_handle") or "").strip():
        raise ValueError("approved proposal requires reviewed_by_handle")
    if not str(approval.get("reviewed_at") or "").strip():
        raise ValueError("approved proposal requires reviewed_at")
    _utc_timestamp(approval["reviewed_at"], "approved proposal reviewed_at")
    if not str(approval.get("review_note") or "").strip():
        raise ValueError("approved proposal requires review_note")
    _uuid(
        approval.get("reviewed_by_agent_id"),
        "approved proposal reviewed_by_agent_id",
        nullable=True,
    )


def _approval_guard_sql(proposal_id: str, approval: dict[str, Any]) -> str:
    reviewed_at = _utc_timestamp(approval["reviewed_at"], "approved proposal reviewed_at")
    return f"""select kb_stage.assert_approved_proposal(
  {sql_literal(proposal_id)}::uuid,
  {sql_literal(approval["proposal_type"])},
  {_jsonb(approval["payload"])},
  {sql_literal(approval["reviewed_by_handle"])},
  {sql_literal(approval.get("reviewed_by_agent_id"))}::uuid,
  {sql_literal(reviewed_at)}::timestamptz,
  {sql_literal(approval["review_note"])}
);"""


def _ledger_and_verify(
    proposal_id: str,
    applied_by: str,
    extra_checks: str,
    approval: dict[str, Any],
) -> str:
    """Verify canonical rows, then finish the locked, payload-bound apply."""
    reviewed_at = _utc_timestamp(approval["reviewed_at"], "approved proposal reviewed_at")
    checks_sql = f"""do $verify$
begin
{extra_checks}
end
$verify$;"""
    finish_sql = f"""select kb_stage.finish_approved_proposal(
  {sql_literal(proposal_id)}::uuid,
  {sql_literal(approval["proposal_type"])},
  {_jsonb(approval["payload"])},
  {sql_literal(approval["reviewed_by_handle"])},
  {sql_literal(approval.get("reviewed_by_agent_id"))}::uuid,
  {sql_literal(reviewed_at)}::timestamptz,
  {sql_literal(approval["review_note"])},
  {sql_literal(applied_by)}
);"""
    return checks_sql + "\n" + finish_sql


# --------------------------------------------------------------------------- #
# Per-type SQL builders (pure; unit-tested without a DB)                       #
# --------------------------------------------------------------------------- #
def build_revise_strategy_sql(
    apply_payload: dict[str, Any],
    proposal_id: str,
    applied_by: str | None,
    approval: dict[str, Any],
) -> str:
    _validate_approval_meta(approval, "revise_strategy", apply_payload)
    agent_id = apply_payload.get("agent_id")
    strategy = apply_payload.get("strategy") or {}
    nodes: list[dict[str, Any]] = apply_payload.get("strategy_nodes") or []
    if not agent_id:
        raise ValueError("revise_strategy apply_payload requires 'agent_id'")
    for key in ("diagnosis", "guiding_policy", "proximate_objectives"):
        if strategy.get(key) is None:
            raise ValueError(f"revise_strategy apply_payload.strategy requires '{key}'")
    if not nodes:
        raise ValueError("revise_strategy apply_payload requires non-empty 'strategy_nodes'")

    node_values = []
    for n in nodes:
        nt = n.get("node_type")
        if nt not in ("diagnosis", "policy", "objective"):
            raise ValueError(f"invalid strategy node_type: {nt!r}")
        if not n.get("title") or n.get("body") is None:
            raise ValueError("each strategy_node requires 'title' and 'body'")
        node_values.append(
            f"({sql_literal(agent_id)}::uuid, {sql_literal(nt)}, "
            f"{sql_literal(n['title'])}, {sql_literal(n['body'])}, "
            f"{int(n.get('rank', 1))})"
        )

    node_values_sql = ",\n  ".join(node_values)
    canonical = f"""
-- deactivate the current active strategy for this agent (one_active invariant)
update public.strategies
   set active = false
 where agent_id = {sql_literal(agent_id)}::uuid and active;

-- insert the new strategy as the next version, active
insert into public.strategies
       (agent_id, diagnosis, guiding_policy, proximate_objectives, version, active)
select {sql_literal(agent_id)}::uuid,
       {sql_literal(strategy["diagnosis"])},
       {sql_literal(strategy["guiding_policy"])},
       {_jsonb(strategy["proximate_objectives"])},
       coalesce(max(version), 0) + 1,
       true
  from public.strategies
 where agent_id = {sql_literal(agent_id)}::uuid;

-- retire the agent's current strategy nodes (shared NULL-agent nodes untouched)
update public.strategy_nodes
   set status = 'retired', updated_at = now()
 where agent_id = {sql_literal(agent_id)}::uuid and status <> 'retired';

-- insert the new node set
insert into public.strategy_nodes (agent_id, node_type, title, body, rank)
values
  {node_values_sql};
""".rstrip()

    checks = f"""  if (select count(*) from public.strategies
       where agent_id = {sql_literal(agent_id)}::uuid and active) <> 1 then
    raise exception 'apply_proposal: expected exactly one active strategy for agent %', {sql_literal(agent_id)};
  end if;"""

    ledger = _ledger_and_verify(proposal_id, applied_by or SERVICE_AGENT_HANDLE, checks, approval)
    return _wrap_txn(canonical, ledger, _approval_guard_sql(proposal_id, approval))


def build_add_edge_sql(
    apply_payload: dict[str, Any],
    proposal_id: str,
    applied_by: str | None,
    approval: dict[str, Any],
) -> str:
    _validate_approval_meta(approval, "add_edge", apply_payload)
    _reject_unknown(apply_payload, {"from_claim", "to_claim", "edge_type", "weight"}, "add_edge apply_payload")
    f = _uuid(apply_payload.get("from_claim"), "add_edge.from_claim")
    t = _uuid(apply_payload.get("to_claim"), "add_edge.to_claim")
    et = apply_payload.get("edge_type")
    if et not in EDGE_TYPES:
        raise ValueError(f"add_edge.edge_type must be one of {sorted(EDGE_TYPES)}")
    w = _weight(apply_payload.get("weight"), "add_edge.weight")
    if f == t:
        raise ValueError("add_edge from_claim and to_claim must differ")
    row_id = deterministic_row_uuid(proposal_id, "claim_edge", f, t, et)

    edge_lock_key = f"claim_edge:{f}:{t}:{et}"
    canonical = f"""
-- serialize semantic edge creation because the historical table has no
-- (from_claim,to_claim,edge_type) uniqueness constraint.
select pg_advisory_xact_lock(hashtextextended({sql_literal(edge_lock_key)}, 0));

-- insert the edge unless an identical (from,to,type) edge already exists
insert into public.claim_edges (id, from_claim, to_claim, edge_type, weight)
select {sql_literal(row_id)}::uuid, {sql_literal(f)}::uuid, {sql_literal(t)}::uuid,
       {sql_literal(et)}::edge_type, {sql_literal(w)}
 where not exists (
   select 1 from public.claim_edges
    where from_claim = {sql_literal(f)}::uuid
      and to_claim = {sql_literal(t)}::uuid
      and edge_type = {sql_literal(et)}::edge_type);
""".rstrip()

    checks = f"""  -- Historical replay receipts can carry a pre-deterministic row id.
  -- Require one exact semantic row; fresh applies still insert row_id above.
  if (select count(*) from public.claim_edges
       where from_claim = {sql_literal(f)}::uuid
         and to_claim = {sql_literal(t)}::uuid
         and edge_type = {sql_literal(et)}::edge_type) <> 1 then
    raise exception 'apply_proposal: claim_edge row does not match strict apply payload';
  end if;
  if exists (select 1 from public.claim_edges
       where from_claim = {sql_literal(f)}::uuid
         and to_claim = {sql_literal(t)}::uuid
         and edge_type = {sql_literal(et)}::edge_type
         and weight is distinct from {sql_literal(w)}) then
    raise exception 'apply_proposal: claim_edge row does not match strict apply payload';
  end if;
  if not exists (select 1 from public.claim_edges
       where from_claim = {sql_literal(f)}::uuid
         and to_claim = {sql_literal(t)}::uuid
         and edge_type = {sql_literal(et)}::edge_type
         and weight is not distinct from {sql_literal(w)}) then
    raise exception 'apply_proposal: claim_edge row does not match strict apply payload';
  end if;"""

    ledger = _ledger_and_verify(proposal_id, applied_by or SERVICE_AGENT_HANDLE, checks, approval)
    return _wrap_txn(canonical, ledger, _approval_guard_sql(proposal_id, approval))


def build_attach_evidence_sql(
    apply_payload: dict[str, Any],
    proposal_id: str,
    applied_by: str | None,
    approval: dict[str, Any],
) -> str:
    _validate_approval_meta(approval, "attach_evidence", apply_payload)
    _reject_unknown(apply_payload, {"evidence"}, "attach_evidence apply_payload")
    evidence: list[dict[str, Any]] = apply_payload.get("evidence") or []
    if not evidence:
        raise ValueError("attach_evidence apply_payload requires non-empty 'evidence'")

    statements = []
    checks = []
    for i, ev in enumerate(evidence):
        _reject_unknown(
            ev,
            {"claim_id", "source_id", "role", "weight"},
            f"attach_evidence.evidence[{i}]",
        )
        claim_id = _uuid(ev.get("claim_id"), f"attach_evidence.evidence[{i}].claim_id")
        source_id = _uuid(ev.get("source_id"), f"attach_evidence.evidence[{i}].source_id")
        role = ev.get("role") or "grounds"
        if role not in EVIDENCE_ROLES:
            raise ValueError(f"attach_evidence.evidence[{i}].role must be one of {sorted(EVIDENCE_ROLES)}")
        weight = _weight(ev.get("weight"), f"attach_evidence.evidence[{i}].weight")
        statements.append(
            f"""insert into public.claim_evidence (claim_id, source_id, role, weight)
select {sql_literal(claim_id)}::uuid, {sql_literal(source_id)}::uuid,
       {sql_literal(role)}::evidence_role, {sql_literal(weight)}
on conflict (claim_id, source_id, role) do nothing;"""
        )
        checks.append(
            f"""  -- Accept a legacy receipt id only when exactly one semantic row matches.
  if (select count(*) from public.claim_evidence
       where claim_id = {sql_literal(claim_id)}::uuid
         and source_id = {sql_literal(source_id)}::uuid
         and role = {sql_literal(role)}::evidence_role) <> 1 then
    raise exception 'apply_proposal: evidence row % does not match strict apply payload', {i};
  end if;
  if exists (select 1 from public.claim_evidence
       where claim_id = {sql_literal(claim_id)}::uuid
         and source_id = {sql_literal(source_id)}::uuid
         and role = {sql_literal(role)}::evidence_role
         and weight is distinct from {sql_literal(weight)}) then
    raise exception 'apply_proposal: evidence row % does not match strict apply payload', {i};
  end if;
  if not exists (select 1 from public.claim_evidence
       where claim_id = {sql_literal(claim_id)}::uuid
         and source_id = {sql_literal(source_id)}::uuid
         and role = {sql_literal(role)}::evidence_role
         and weight is not distinct from {sql_literal(weight)}) then
    raise exception 'apply_proposal: evidence row % does not match strict apply payload', {i};
  end if;"""
        )

    canonical = "\n".join(statements)
    ledger = _ledger_and_verify(proposal_id, applied_by or SERVICE_AGENT_HANDLE, "\n".join(checks), approval)
    return _wrap_txn(canonical, ledger, _approval_guard_sql(proposal_id, approval))


def build_approve_claim_v3_sql(
    apply_payload: dict[str, Any],
    proposal_id: str,
    applied_by: str | None,
    approval: dict[str, Any],
    *,
    fresh_owned: bool = False,
    fresh_preflight_sha256: str | None = None,
) -> str:
    """Build an exact-row idempotent V3 claim/source/evidence transaction."""
    _validate_approval_meta(approval, "approve_claim", apply_payload)
    bundle = _normalize_v3_approve_claim(apply_payload)
    if fresh_owned:
        _sha256(fresh_preflight_sha256, "fresh-owned approve_claim preflight")
    elif fresh_preflight_sha256 is not None:
        raise ValueError("fresh preflight SHA-256 is only valid for fresh-owned approve_claim")

    statements: list[str] = []
    checks: list[str] = []
    if fresh_owned:
        statements.extend(
            [
                f"-- fresh-owned preflight SHA-256: {fresh_preflight_sha256}",
                (
                    "select pg_advisory_xact_lock(hashtextextended("
                    f"'proposal-fresh-apply:' || {sql_literal(proposal_id)}, 0));"
                ),
            ]
        )

    for row in bundle["sources"]:
        statements.append(
            f"""do $source_conflict$
begin
  if exists (select 1 from public.sources
       where hash = {sql_literal(row["content_hash"])}
         and id <> {sql_literal(row["id"])}::uuid) then
    raise exception 'V3 approve_claim: source hash collision for source %', {sql_literal(row["id"])};
  end if;
end
$source_conflict$;
insert into public.sources (
  id, source_type, url, storage_path, excerpt, hash, created_by, captured_at,
  canonical_url, storage_uri, content_hash, access_scope, ingestion_origin,
  provenance_status, source_class, auto_promotion_policy, accepted_by_proposal_id
) values (
  {sql_literal(row["id"])}::uuid, {sql_literal(row["source_type"])},
  {sql_literal(row["canonical_url"])}, {sql_literal(row["storage_uri"])},
  {sql_literal(row["excerpt"])}, {sql_literal(row["content_hash"])},
  {sql_literal(row["created_by"])}::uuid, {sql_literal(row["captured_at"])}::timestamptz,
  {sql_literal(row["canonical_url"])}, {sql_literal(row["storage_uri"])},
  {sql_literal(row["content_hash"])}, {sql_literal(row["access_scope"])},
  {sql_literal(row["ingestion_origin"])}, {sql_literal(row["provenance_status"])},
  {sql_literal(row["source_class"])}, {sql_literal(row["auto_promotion_policy"])},
  {sql_literal(proposal_id)}::uuid
)
on conflict do nothing;"""
        )
        checks.append(
            f"""  if not exists (select 1 from public.sources
       where id = {sql_literal(row["id"])}::uuid
         and source_type = {sql_literal(row["source_type"])}
         and url is not distinct from {sql_literal(row["canonical_url"])}
         and storage_path is not distinct from {sql_literal(row["storage_uri"])}
         and excerpt is not distinct from {sql_literal(row["excerpt"])}
         and hash = {sql_literal(row["content_hash"])}
         and created_by is not distinct from {sql_literal(row["created_by"])}::uuid
         and captured_at = {sql_literal(row["captured_at"])}::timestamptz
         and canonical_url is not distinct from {sql_literal(row["canonical_url"])}
         and storage_uri is not distinct from {sql_literal(row["storage_uri"])}
         and content_hash = {sql_literal(row["content_hash"])}
         and access_scope = {sql_literal(row["access_scope"])}
         and ingestion_origin = {sql_literal(row["ingestion_origin"])}
         and provenance_status = {sql_literal(row["provenance_status"])}
         and source_class = {sql_literal(row["source_class"])}
         and auto_promotion_policy = {sql_literal(row["auto_promotion_policy"])}
         and accepted_by_proposal_id = {sql_literal(proposal_id)}::uuid) then
    raise exception 'V3 approve_claim: source row does not match strict payload: %', {sql_literal(row["id"])};
  end if;"""
        )

    for row in bundle["claims"]:
        tags = _text_array(row["tags"])
        statements.append(
            f"""insert into public.claims (
  id, owner_agent_id, type, text, proposition, body_markdown, scope, access_scope,
  status, admission_state, revision_criteria, revision_kind, supersedes_id,
  confidence, tags, created_by, superseded_by, accepted_by_proposal_id
) values (
  {sql_literal(row["id"])}::uuid, {sql_literal(row["owner_agent_id"])}::uuid,
  {sql_literal(row["type"])}, {sql_literal(row["proposition"])}, {sql_literal(row["proposition"])},
  {sql_literal(row["body_markdown"])}, {sql_literal(row["scope"])}, {sql_literal(row["access_scope"])},
  {sql_literal(row["status"])}, 'admitted', {sql_literal(row["revision_criteria"])},
  {sql_literal(row["revision_kind"])}, null, null, {tags},
  {sql_literal(row["created_by"])}::uuid, null, {sql_literal(proposal_id)}::uuid
)
on conflict do nothing;"""
        )
        checks.append(
            f"""  if not exists (select 1 from public.claims
       where id = {sql_literal(row["id"])}::uuid
         and owner_agent_id is not distinct from {sql_literal(row["owner_agent_id"])}::uuid
         and type = {sql_literal(row["type"])}
         and text = {sql_literal(row["proposition"])}
         and proposition = {sql_literal(row["proposition"])}
         and body_markdown = {sql_literal(row["body_markdown"])}
         and scope = {sql_literal(row["scope"])}
         and access_scope = {sql_literal(row["access_scope"])}
         and status = {sql_literal(row["status"])}
         and admission_state = 'admitted'
         and revision_criteria = {sql_literal(row["revision_criteria"])}
         and revision_kind = 'original'
         and supersedes_id is null
         and confidence is null
         and tags = {tags}
         and created_by is not distinct from {sql_literal(row["created_by"])}::uuid
         and superseded_by is null
         and accepted_by_proposal_id = {sql_literal(proposal_id)}::uuid) then
    raise exception 'V3 approve_claim: claim row does not match strict payload: %', {sql_literal(row["id"])};
  end if;"""
        )

    for row in bundle["evidence"]:
        statements.append(
            f"""insert into public.claim_evidence (
  id, claim_id, source_id, role, weight, created_by, excerpt, locator_json,
  source_content_hash, rationale, polarity, accepted_from_proposal_id
) values (
  {sql_literal(row["id"])}::uuid, {sql_literal(row["claim_id"])}::uuid,
  {sql_literal(row["source_id"])}::uuid, {sql_literal(row["role"])}::evidence_role,
  null, {sql_literal(row["created_by"])}::uuid, {sql_literal(row["excerpt"])},
  {_jsonb(row["locator_json"])}, {sql_literal(row["source_content_hash"])},
  {sql_literal(row["rationale"])}, {sql_literal(row["polarity"])}, {sql_literal(proposal_id)}::uuid
)
on conflict do nothing;"""
        )
        checks.append(
            f"""  if not exists (select 1 from public.claim_evidence
       where id = {sql_literal(row["id"])}::uuid
         and claim_id = {sql_literal(row["claim_id"])}::uuid
         and source_id = {sql_literal(row["source_id"])}::uuid
         and role = {sql_literal(row["role"])}::evidence_role
         and weight is null
         and created_by is not distinct from {sql_literal(row["created_by"])}::uuid
         and excerpt is not distinct from {sql_literal(row["excerpt"])}
         and locator_json = {_jsonb(row["locator_json"])}
         and source_content_hash = {sql_literal(row["source_content_hash"])}
         and rationale = {sql_literal(row["rationale"])}
         and polarity = {sql_literal(row["polarity"])}
         and accepted_from_proposal_id = {sql_literal(proposal_id)}::uuid) then
    raise exception 'V3 approve_claim: evidence row does not match strict payload: %', {sql_literal(row["id"])};
  end if;"""
        )

    for row in bundle["edges"]:
        statements.append(
            f"""insert into public.claim_edges (
  id, from_claim, to_claim, edge_type, weight, created_by, relation_type,
  rationale, scope_or_conditions, accepted_by_proposal_id
) values (
  {sql_literal(row["id"])}::uuid, {sql_literal(row["from_claim"])}::uuid,
  {sql_literal(row["to_claim"])}::uuid, {sql_literal(row["edge_type"])}::edge_type,
  null, {sql_literal(row["created_by"])}::uuid, {sql_literal(row["relation_type"])},
  {sql_literal(row["rationale"])}, {sql_literal(row["scope_or_conditions"])},
  {sql_literal(proposal_id)}::uuid
)
on conflict do nothing;"""
        )
        checks.append(
            f"""  if not exists (select 1 from public.claim_edges
       where id = {sql_literal(row["id"])}::uuid
         and from_claim = {sql_literal(row["from_claim"])}::uuid
         and to_claim = {sql_literal(row["to_claim"])}::uuid
         and edge_type = {sql_literal(row["edge_type"])}::edge_type
         and weight is null
         and created_by is not distinct from {sql_literal(row["created_by"])}::uuid
         and relation_type = {sql_literal(row["relation_type"])}
         and rationale = {sql_literal(row["rationale"])}
         and scope_or_conditions is not distinct from {sql_literal(row["scope_or_conditions"])}
         and accepted_by_proposal_id = {sql_literal(proposal_id)}::uuid) then
    raise exception 'V3 approve_claim: edge row does not match strict payload: %', {sql_literal(row["id"])};
  end if;"""
        )

    for row in bundle["assessments"]:
        statements.append(
            f"""insert into public.claim_evidence_assessments (
  id, claim_id, support_tier, challenge_tier, overall_state, rationale,
  evidence_set_hash, accepted_by_proposal_id, supersedes_assessment_id
) values (
  {sql_literal(row["id"])}::uuid, {sql_literal(row["claim_id"])}::uuid,
  {sql_literal(row["support_tier"])}, {sql_literal(row["challenge_tier"])},
  {sql_literal(row["overall_state"])}, {sql_literal(row["rationale"])},
  {sql_literal(row["evidence_set_hash"])}, {sql_literal(proposal_id)}::uuid,
  {sql_literal(row["supersedes_assessment_id"])}::uuid
)
on conflict do nothing;"""
        )
        checks.append(
            f"""  if not exists (select 1 from public.claim_evidence_assessments
       where id = {sql_literal(row["id"])}::uuid
         and claim_id = {sql_literal(row["claim_id"])}::uuid
         and support_tier is not distinct from {sql_literal(row["support_tier"])}
         and challenge_tier is not distinct from {sql_literal(row["challenge_tier"])}
         and overall_state = {sql_literal(row["overall_state"])}
         and rationale = {sql_literal(row["rationale"])}
         and evidence_set_hash = {sql_literal(row["evidence_set_hash"])}
         and accepted_by_proposal_id = {sql_literal(proposal_id)}::uuid
         and supersedes_assessment_id is not distinct from {sql_literal(row["supersedes_assessment_id"])}::uuid) then
    raise exception 'V3 approve_claim: assessment row does not match strict payload: %', {sql_literal(row["id"])};
  end if;"""
        )

    canonical = "\n\n".join(statements)
    ledger = _ledger_and_verify(proposal_id, applied_by or SERVICE_AGENT_HANDLE, "\n".join(checks), approval)
    return _wrap_txn(
        canonical,
        ledger,
        _approval_guard_sql(proposal_id, approval),
        serializable=True,
        contract_guard_sql=_v3_contract_transaction_guard_sql(),
        session_advisory_lock_key=V3_CATALOG_DDL_ADVISORY_LOCK_KEY,
    )


def build_approve_claim_sql(
    apply_payload: dict[str, Any],
    proposal_id: str,
    applied_by: str | None,
    approval: dict[str, Any],
    *,
    fresh_owned: bool = False,
    fresh_preflight_sha256: str | None = None,
) -> str:
    """Build one conflict-safe transaction for a reviewed canonical graph bundle."""
    _validate_approval_meta(approval, "approve_claim", apply_payload)
    if apply_payload.get("contract_version") == 3:
        return build_approve_claim_v3_sql(
            apply_payload,
            proposal_id,
            applied_by,
            approval,
            fresh_owned=fresh_owned,
            fresh_preflight_sha256=fresh_preflight_sha256,
        )
    _reject_unknown(
        apply_payload,
        {
            "contract_version",
            "source_proposal_id",
            "agent_id",
            "claims",
            "sources",
            "evidence",
            "edges",
            "reasoning_tools",
            "normalization_manifest",
        },
        "approve_claim apply_payload",
    )
    if apply_payload.get("contract_version") != 2:
        raise ValueError("approve_claim apply_payload.contract_version must equal 2")
    for key in ("claims", "sources", "evidence", "edges", "reasoning_tools"):
        if key not in apply_payload:
            raise ValueError(f"approve_claim apply_payload requires explicit '{key}' collection")
    if apply_payload.get("source_proposal_id") is not None:
        _uuid(apply_payload["source_proposal_id"], "approve_claim.source_proposal_id")
    if apply_payload.get("normalization_manifest") is not None and not isinstance(
        apply_payload["normalization_manifest"], dict
    ):
        raise ValueError("approve_claim.normalization_manifest must be an object")
    agent_id = _uuid(apply_payload.get("agent_id"), "approve_claim.agent_id", nullable=True)
    raw_claims = _rows(apply_payload, "claims", required=True)
    raw_sources = _rows(apply_payload, "sources")
    raw_evidence = _rows(apply_payload, "evidence")
    raw_edges = _rows(apply_payload, "edges")
    raw_tools = _rows(apply_payload, "reasoning_tools")

    claims: list[dict[str, Any]] = []
    for index, row in enumerate(raw_claims):
        path = f"approve_claim.claims[{index}]"
        _reject_unknown(
            row,
            {"id", "type", "text", "status", "confidence", "tags", "created_by", "superseded_by"},
            path,
        )
        claim_type = row.get("type")
        if claim_type not in CLAIM_TYPES:
            raise ValueError(f"{path}.type must be one of {sorted(CLAIM_TYPES)}")
        text = row.get("text")
        if not isinstance(text, str) or not text.strip():
            raise ValueError(f"{path}.text must be a non-empty string")
        status = row.get("status", "open")
        if status != "open":
            raise ValueError(f"{path}.status must be 'open' for an insert-only approval")
        tags = row.get("tags") or []
        if not isinstance(tags, list) or any(not isinstance(tag, str) for tag in tags):
            raise ValueError(f"{path}.tags must be a list of strings")
        claims.append(
            {
                "id": _uuid(row.get("id"), f"{path}.id"),
                "type": claim_type,
                "text": text,
                "status": status,
                "confidence": _weight(row.get("confidence"), f"{path}.confidence"),
                "tags": tags,
                "created_by": _uuid(row.get("created_by", agent_id), f"{path}.created_by", nullable=True),
                "superseded_by": _uuid(row.get("superseded_by"), f"{path}.superseded_by", nullable=True),
            }
        )

    sources: list[dict[str, Any]] = []
    for index, row in enumerate(raw_sources):
        path = f"approve_claim.sources[{index}]"
        _reject_unknown(
            row,
            {"id", "source_type", "url", "storage_path", "excerpt", "hash", "created_by"},
            path,
        )
        source_type = row.get("source_type")
        if source_type not in SOURCE_TYPES:
            raise ValueError(f"{path}.source_type must be one of {sorted(SOURCE_TYPES)}")
        source_hash = row.get("hash")
        if not isinstance(source_hash, str) or not source_hash.strip():
            raise ValueError(f"{path}.hash must be a non-empty string")
        for key in ("url", "storage_path", "excerpt"):
            if row.get(key) is not None and not isinstance(row.get(key), str):
                raise ValueError(f"{path}.{key} must be a string or null")
        sources.append(
            {
                "id": _uuid(row.get("id"), f"{path}.id"),
                "source_type": source_type,
                "url": row.get("url"),
                "storage_path": row.get("storage_path"),
                "excerpt": row.get("excerpt"),
                "hash": source_hash,
                "created_by": _uuid(row.get("created_by", agent_id), f"{path}.created_by", nullable=True),
            }
        )

    evidence: list[dict[str, Any]] = []
    for index, row in enumerate(raw_evidence):
        path = f"approve_claim.evidence[{index}]"
        _reject_unknown(
            row,
            {"claim_id", "source_id", "role", "weight", "created_by"},
            path,
        )
        role = row.get("role", "grounds")
        if role not in EVIDENCE_ROLES:
            raise ValueError(f"{path}.role must be one of {sorted(EVIDENCE_ROLES)}")
        evidence.append(
            {
                "claim_id": _uuid(row.get("claim_id"), f"{path}.claim_id"),
                "source_id": _uuid(row.get("source_id"), f"{path}.source_id"),
                "role": role,
                "weight": _weight(row.get("weight"), f"{path}.weight"),
                "created_by": _uuid(row.get("created_by", agent_id), f"{path}.created_by", nullable=True),
            }
        )

    edges: list[dict[str, Any]] = []
    for index, row in enumerate(raw_edges):
        path = f"approve_claim.edges[{index}]"
        _reject_unknown(
            row,
            {"from_claim", "to_claim", "edge_type", "weight", "created_by"},
            path,
        )
        edge_type = row.get("edge_type")
        if edge_type not in EDGE_TYPES:
            raise ValueError(f"{path}.edge_type must be one of {sorted(EDGE_TYPES)}")
        from_claim = _uuid(row.get("from_claim"), f"{path}.from_claim")
        to_claim = _uuid(row.get("to_claim"), f"{path}.to_claim")
        if from_claim == to_claim:
            raise ValueError(f"{path} cannot be a self-edge")
        edges.append(
            {
                "id": deterministic_row_uuid(proposal_id, "claim_edge", from_claim, to_claim, edge_type),
                "from_claim": from_claim,
                "to_claim": to_claim,
                "edge_type": edge_type,
                "weight": _weight(row.get("weight"), f"{path}.weight"),
                "created_by": _uuid(row.get("created_by", agent_id), f"{path}.created_by", nullable=True),
            }
        )

    reasoning_tools: list[dict[str, Any]] = []
    for index, row in enumerate(raw_tools):
        path = f"approve_claim.reasoning_tools[{index}]"
        _reject_unknown(
            row,
            {"id", "agent_id", "name", "description", "category"},
            path,
        )
        name = row.get("name")
        description = row.get("description")
        category = row.get("category")
        if not isinstance(name, str) or not name.strip():
            raise ValueError(f"{path}.name must be a non-empty string")
        if not isinstance(description, str) or not description.strip():
            raise ValueError(f"{path}.description must be a non-empty string")
        if category is not None and not isinstance(category, str):
            raise ValueError(f"{path}.category must be a string or null")
        reasoning_tools.append(
            {
                "id": _uuid(row.get("id"), f"{path}.id"),
                "agent_id": _uuid(row.get("agent_id", agent_id), f"{path}.agent_id", nullable=True),
                "name": name,
                "description": description,
                "category": category,
            }
        )

    _dedupe([row["id"] for row in claims], "approve_claim.claims ids")
    _dedupe([row["id"] for row in sources], "approve_claim.sources ids")
    _dedupe([row["hash"] for row in sources], "approve_claim.sources hashes")
    _dedupe(
        [[row["claim_id"], row["source_id"], row["role"]] for row in evidence],
        "approve_claim.evidence keys",
    )
    _dedupe(
        [[row["from_claim"], row["to_claim"], row["edge_type"]] for row in edges],
        "approve_claim.edges keys",
    )
    _dedupe([row["id"] for row in reasoning_tools], "approve_claim.reasoning_tools ids")

    if fresh_owned:
        if (
            not isinstance(fresh_preflight_sha256, str)
            or len(fresh_preflight_sha256) != 64
            or any(character not in "0123456789abcdef" for character in fresh_preflight_sha256)
        ):
            raise ValueError("fresh-owned approve_claim requires an exact preflight SHA-256")
    elif fresh_preflight_sha256 is not None:
        raise ValueError("fresh preflight SHA-256 is only valid for fresh-owned approve_claim")

    statements: list[str] = []
    checks: list[str] = []

    if fresh_owned:
        statements.extend(
            [
                f"-- fresh-owned preflight SHA-256: {fresh_preflight_sha256}",
                (
                    "select pg_advisory_xact_lock(hashtextextended("
                    f"'proposal-fresh-apply:' || {sql_literal(proposal_id)}, 0));"
                ),
            ]
        )

    if sources:
        collision_checks = []
        for row in sources:
            collision_checks.append(
                f"""  if exists (select 1 from public.sources
       where hash = {sql_literal(row["hash"])}
         and id <> {sql_literal(row["id"])}::uuid) then
    raise exception 'approve_claim: source hash collision for source %', {sql_literal(row["id"])};
  end if;"""
            )
        statements.append("do $source_conflicts$\nbegin\n" + "\n".join(collision_checks) + "\nend\n$source_conflicts$;")

    for row in claims:
        tags = _text_array(row["tags"])
        conflict_clause = ";" if fresh_owned else "\non conflict (id) do nothing;"
        statements.append(
            f"""insert into public.claims (id, type, text, status, confidence, tags, created_by, superseded_by)
select {sql_literal(row["id"])}::uuid, {sql_literal(row["type"])}, {sql_literal(row["text"])},
       {sql_literal(row["status"])}, {sql_literal(row["confidence"])}, {tags},
       {sql_literal(row["created_by"])}::uuid, {sql_literal(row["superseded_by"])}::uuid{conflict_clause}"""
        )
        checks.append(
            f"""  if not exists (select 1 from public.claims
       where id = {sql_literal(row["id"])}::uuid
         and type = {sql_literal(row["type"])}
         and text = {sql_literal(row["text"])}
         and status = {sql_literal(row["status"])}
         and confidence is not distinct from {sql_literal(row["confidence"])}
         and tags is not distinct from {tags}
         and created_by is not distinct from {sql_literal(row["created_by"])}::uuid
         and superseded_by is not distinct from {sql_literal(row["superseded_by"])}::uuid) then
    raise exception 'approve_claim: claim row does not match strict apply payload: %', {sql_literal(row["id"])};
  end if;"""
        )

    for row in sources:
        conflict_clause = ";" if fresh_owned else "\non conflict do nothing;"
        statements.append(
            f"""insert into public.sources (id, source_type, url, storage_path, excerpt, hash, created_by)
select {sql_literal(row["id"])}::uuid, {sql_literal(row["source_type"])}, {sql_literal(row["url"])},
       {sql_literal(row["storage_path"])}, {sql_literal(row["excerpt"])}, {sql_literal(row["hash"])},
       {sql_literal(row["created_by"])}::uuid{conflict_clause}"""
        )
        checks.append(
            f"""  if exists (select 1 from public.sources
       where hash = {sql_literal(row["hash"])}
         and id <> {sql_literal(row["id"])}::uuid) then
    raise exception 'approve_claim: source hash collision for source %', {sql_literal(row["id"])};
  end if;
  if not exists (select 1 from public.sources
       where id = {sql_literal(row["id"])}::uuid
         and source_type = {sql_literal(row["source_type"])}
         and url is not distinct from {sql_literal(row["url"])}
         and storage_path is not distinct from {sql_literal(row["storage_path"])}
         and excerpt is not distinct from {sql_literal(row["excerpt"])}
         and hash = {sql_literal(row["hash"])}
         and created_by is not distinct from {sql_literal(row["created_by"])}::uuid) then
    raise exception 'approve_claim: source row does not match strict apply payload: %', {sql_literal(row["id"])};
  end if;"""
        )

    for row in reasoning_tools:
        conflict_clause = ";" if fresh_owned else "\non conflict (id) do nothing;"
        statements.append(
            f"""insert into public.reasoning_tools (id, agent_id, name, description, category)
select {sql_literal(row["id"])}::uuid, {sql_literal(row["agent_id"])}::uuid,
       {sql_literal(row["name"])}, {sql_literal(row["description"])}, {sql_literal(row["category"])}{conflict_clause}"""
        )
        checks.append(
            f"""  if not exists (select 1 from public.reasoning_tools
       where id = {sql_literal(row["id"])}::uuid
         and agent_id is not distinct from {sql_literal(row["agent_id"])}::uuid
         and name = {sql_literal(row["name"])}
         and description = {sql_literal(row["description"])}
         and category is not distinct from {sql_literal(row["category"])}) then
    raise exception 'approve_claim: reasoning tool row does not match strict apply payload: %', {sql_literal(row["id"])};
  end if;"""
        )

    for row in evidence:
        conflict_clause = ";" if fresh_owned else "\non conflict (claim_id, source_id, role) do nothing;"
        statements.append(
            f"""insert into public.claim_evidence (claim_id, source_id, role, weight, created_by)
select {sql_literal(row["claim_id"])}::uuid, {sql_literal(row["source_id"])}::uuid,
       {sql_literal(row["role"])}::evidence_role, {sql_literal(row["weight"])},
       {sql_literal(row["created_by"])}::uuid{conflict_clause}"""
        )
        checks.append(
            f"""  -- Accept a legacy receipt id only when exactly one semantic row matches.
  if (select count(*) from public.claim_evidence
       where claim_id = {sql_literal(row["claim_id"])}::uuid
         and source_id = {sql_literal(row["source_id"])}::uuid
         and role = {sql_literal(row["role"])}::evidence_role) <> 1 then
    raise exception 'approve_claim: evidence row does not match strict apply payload';
  end if;
  if exists (select 1 from public.claim_evidence
       where claim_id = {sql_literal(row["claim_id"])}::uuid
         and source_id = {sql_literal(row["source_id"])}::uuid
         and role = {sql_literal(row["role"])}::evidence_role
         and (weight is distinct from {sql_literal(row["weight"])}
           or created_by is distinct from {sql_literal(row["created_by"])}::uuid)) then
    raise exception 'approve_claim: evidence row does not match strict apply payload';
  end if;
  if not exists (select 1 from public.claim_evidence
       where claim_id = {sql_literal(row["claim_id"])}::uuid
         and source_id = {sql_literal(row["source_id"])}::uuid
         and role = {sql_literal(row["role"])}::evidence_role
         and weight is not distinct from {sql_literal(row["weight"])}
         and created_by is not distinct from {sql_literal(row["created_by"])}::uuid) then
    raise exception 'approve_claim: evidence row does not match strict apply payload';
  end if;"""
        )

    for row in edges:
        edge_lock_key = f"claim_edge:{row['from_claim']}:{row['to_claim']}:{row['edge_type']}"
        edge_insert_guard = (
            ";"
            if fresh_owned
            else f"""\n where not exists (
   select 1 from public.claim_edges
    where from_claim = {sql_literal(row["from_claim"])}::uuid
      and to_claim = {sql_literal(row["to_claim"])}::uuid
      and edge_type = {sql_literal(row["edge_type"])}::edge_type);"""
        )
        statements.append(
            f"""select pg_advisory_xact_lock(hashtextextended({sql_literal(edge_lock_key)}, 0));
insert into public.claim_edges (id, from_claim, to_claim, edge_type, weight, created_by)
select {sql_literal(row["id"])}::uuid, {sql_literal(row["from_claim"])}::uuid, {sql_literal(row["to_claim"])}::uuid,
       {sql_literal(row["edge_type"])}::edge_type, {sql_literal(row["weight"])},
       {sql_literal(row["created_by"])}::uuid{edge_insert_guard}"""
        )
        checks.append(
            f"""  -- Accept a legacy receipt id only when exactly one semantic row matches.
  if (select count(*) from public.claim_edges
       where from_claim = {sql_literal(row["from_claim"])}::uuid
         and to_claim = {sql_literal(row["to_claim"])}::uuid
         and edge_type = {sql_literal(row["edge_type"])}::edge_type) <> 1 then
    raise exception 'approve_claim: edge row does not match strict apply payload';
  end if;
  if exists (select 1 from public.claim_edges
       where from_claim = {sql_literal(row["from_claim"])}::uuid
         and to_claim = {sql_literal(row["to_claim"])}::uuid
         and edge_type = {sql_literal(row["edge_type"])}::edge_type
         and (weight is distinct from {sql_literal(row["weight"])}
           or created_by is distinct from {sql_literal(row["created_by"])}::uuid)) then
    raise exception 'approve_claim: edge row does not match strict apply payload';
  end if;
  if not exists (select 1 from public.claim_edges
       where from_claim = {sql_literal(row["from_claim"])}::uuid
         and to_claim = {sql_literal(row["to_claim"])}::uuid
         and edge_type = {sql_literal(row["edge_type"])}::edge_type
         and weight is not distinct from {sql_literal(row["weight"])}
         and created_by is not distinct from {sql_literal(row["created_by"])}::uuid) then
    raise exception 'approve_claim: edge row does not match strict apply payload';
  end if;"""
        )

    canonical = "\n\n".join(statements)
    ledger = _ledger_and_verify(proposal_id, applied_by or SERVICE_AGENT_HANDLE, "\n".join(checks), approval)
    return _wrap_txn(
        canonical,
        ledger,
        _approval_guard_sql(proposal_id, approval),
        serializable=fresh_owned,
    )


def _wrap_txn(
    canonical_sql: str,
    ledger_sql: str,
    approval_guard_sql: str,
    *,
    serializable: bool = False,
    contract_guard_sql: str = "",
    session_advisory_lock_key: int | None = None,
) -> str:
    session_lock_sql = ""
    session_unlock_sql = ""
    if session_advisory_lock_key is not None:
        session_lock_sql = f"select pg_catalog.pg_advisory_lock_shared({session_advisory_lock_key});\n"
        session_unlock_sql = f"select pg_catalog.pg_advisory_unlock_shared({session_advisory_lock_key});\n"
    return (
        session_lock_sql
        + "begin;\n"
        + ("set transaction isolation level serializable;\n" if serializable else "")
        + "set local timezone = 'UTC';\n"
        + "set local standard_conforming_strings = on;\n"
        + (f"{contract_guard_sql}\n" if contract_guard_sql else "")
        + f"{approval_guard_sql}\n"
        + f"{canonical_sql}\n"
        + f"{ledger_sql}\n"
        + "commit;\n"
        + session_unlock_sql
    )


BUILDERS = {
    "revise_strategy": build_revise_strategy_sql,
    "add_edge": build_add_edge_sql,
    "attach_evidence": build_attach_evidence_sql,
    "approve_claim": build_approve_claim_sql,
}


def proposal_contract_version(proposal: dict[str, Any]) -> int:
    """Return the exact typed apply contract or reject malformed/hybrid shapes."""
    proposal_type = proposal.get("proposal_type")
    payload = proposal.get("payload")
    if not isinstance(payload, dict):
        raise ValueError("proposal payload must be a JSON object")
    apply_payload = payload.get("apply_payload")
    if not isinstance(apply_payload, dict):
        raise ValueError("proposal payload.apply_payload must be a JSON object")

    raw_version = apply_payload.get("contract_version")
    if proposal_type == "approve_claim":
        if type(raw_version) is not int or raw_version not in {2, 3}:
            raise ValueError("approve_claim apply_payload.contract_version must be integer 2 or 3")
        return raw_version
    if "contract_version" in apply_payload:
        raise ValueError(f"proposal_type {proposal_type!r} is a V2 legacy contract and must omit contract_version")
    return 2


def assert_expected_contract_version(proposal: dict[str, Any], expected: int | None) -> int:
    actual = proposal_contract_version(proposal)
    if expected is not None and actual != expected:
        raise ValueError(f"proposal contract version {actual} does not match expected version {expected}")
    if actual == 3 and proposal.get("proposal_type") != "approve_claim":
        raise ValueError("V3 apply contract supports only proposal_type 'approve_claim'")
    return actual


def build_apply_sql(
    proposal: dict[str, Any],
    applied_by: str | None,
    *,
    fresh_owned: bool = False,
    fresh_preflight_sha256: str | None = None,
) -> str:
    assert_expected_contract_version(proposal, None)
    ptype = proposal["proposal_type"]
    if ptype not in BUILDERS:
        raise ValueError(f"apply_proposal cannot apply proposal_type {ptype!r}")
    payload = proposal["payload"]
    apply_payload = payload.get("apply_payload")
    if apply_payload is None:
        raise ValueError(
            "proposal payload has no 'apply_payload' strict contract. "
            "Run the normalizer to produce apply_payload before applying."
        )
    if fresh_owned and ptype != "approve_claim":
        raise ValueError("fresh-owned apply is supported only for approve_claim")
    if ptype == "approve_claim":
        return build_approve_claim_sql(
            apply_payload,
            proposal["id"],
            applied_by or SERVICE_AGENT_HANDLE,
            proposal,
            fresh_owned=fresh_owned,
            fresh_preflight_sha256=fresh_preflight_sha256,
        )
    return BUILDERS[ptype](
        apply_payload,
        proposal["id"],
        applied_by or SERVICE_AGENT_HANDLE,
        proposal,
    )


def assert_applyable(proposal: dict[str, Any]) -> None:
    status = proposal.get("status")
    if status == "applied":
        raise SystemExit(f"proposal {proposal['id']} is already applied (idempotent no-op)")
    if status != "approved":
        raise SystemExit(f"proposal {proposal['id']} has status {status!r}; only 'approved' proposals apply")


def assert_receiptable(proposal: dict[str, Any]) -> None:
    if proposal.get("status") != "applied":
        raise SystemExit(
            f"proposal {proposal['id']} has status {proposal.get('status')!r}; "
            "--receipt-only requires an applied proposal"
        )
    replay_receipt.expected_row_counts(proposal)


# --------------------------------------------------------------------------- #
# Execution (docker exec as the kb_apply role)                                 #
# --------------------------------------------------------------------------- #
def load_password(secrets_file: str) -> str:
    path = Path(secrets_file)
    if not path.is_file():
        raise SystemExit(f"secrets file not found: {secrets_file}")
    for line in path.read_text(encoding="utf-8").splitlines():
        line = line.strip()
        if not line or line.startswith("#") or "=" not in line:
            continue
        key, _, val = line.partition("=")
        if key.strip() in ("KB_APPLY_PASSWORD", "KB_APPLY_DB_PASSWORD", "PGPASSWORD"):
            return val.strip().strip('"').strip("'")
    raise SystemExit(f"no KB_APPLY_PASSWORD/KB_APPLY_DB_PASSWORD/PGPASSWORD entry in {secrets_file}")


def run_psql(
    args: argparse.Namespace,
    sql: str,
    password: str,
    *,
    redact_output_on_error: bool = False,
) -> str:
    docker_binary = shutil.which("docker") or "docker"
    command = [
        docker_binary,
        "exec",
        "-e",
        "PGPASSWORD",
        "-i",
        args.container,
        "psql",
        "-U",
        args.role,
        "-h",
        args.host,
        "-d",
        args.db,
        "-v",
        "ON_ERROR_STOP=1",
        "-At",
        "-q",
    ]
    session_sql = "set timezone = 'UTC';\n" + sql
    result = subprocess.run(
        command,
        input=session_sql,
        text=True,
        capture_output=True,
        env={"PGPASSWORD": password, "PATH": "/usr/bin:/bin:/usr/local/bin"},
        check=False,
    )
    if result.returncode != 0:
        if redact_output_on_error:
            raise SystemExit(f"psql failed ({result.returncode}); private postflight output redacted")
        raise SystemExit(f"psql failed ({result.returncode})\nSTDOUT:\n{result.stdout}\nSTDERR:\n{result.stderr}")
    return result.stdout


def load_proposal(args: argparse.Namespace, password: str) -> dict[str, Any]:
    sql = f"""select jsonb_build_object(
        'id', id::text,
        'proposal_type', proposal_type,
        'status', status,
        'payload', payload,
        'reviewed_by_handle', reviewed_by_handle,
        'reviewed_by_agent_id', reviewed_by_agent_id::text,
        'reviewed_at', case when reviewed_at is null then null else
          to_char(reviewed_at at time zone 'UTC', 'YYYY-MM-DD"T"HH24:MI:SS.US"Z"') end,
        'review_note', review_note,
        'applied_by_handle', applied_by_handle,
        'applied_by_agent_id', applied_by_agent_id::text,
        'applied_at', case when applied_at is null then null else
          to_char(applied_at at time zone 'UTC', 'YYYY-MM-DD"T"HH24:MI:SS.US"Z"') end)::text
      from kb_stage.kb_proposals
     where id = {sql_literal(args.proposal_id)}::uuid;"""
    out = run_psql(args, sql, password).strip()
    if not out:
        raise SystemExit(f"proposal not found: {args.proposal_id}")
    return json.loads(out)


def capture_replay_receipt(
    args: argparse.Namespace,
    proposal: dict[str, Any],
    password: str,
    *,
    apply_sql: str | None = None,
) -> tuple[Path, dict[str, Any]]:
    """Read exact postflight rows and atomically write a private replay receipt."""
    postflight_sql = replay_receipt.build_postflight_sql(proposal)
    raw = run_psql(
        args,
        postflight_sql,
        password,
        redact_output_on_error=True,
    ).strip()
    if not raw:
        raise RuntimeError("postflight query returned no row")
    canonical_rows = json.loads(raw)
    if not isinstance(canonical_rows, dict):
        raise RuntimeError("postflight query did not return a JSON object")
    apply_sql_source = "exact_executed_sql"
    if apply_sql is None:
        apply_sql = build_apply_sql_for_args(proposal, args, proposal.get("applied_by_handle"))
        apply_sql_source = "reconstructed_current_engine"
    receipt = replay_receipt.build_replay_receipt(
        proposal,
        canonical_rows,
        apply_sql_sha256=replay_receipt.sha256_text(apply_sql),
        apply_sql_source=apply_sql_source,
    )
    path = replay_receipt.resolve_receipt_path(
        str(proposal["id"]),
        receipt_out=args.receipt_out,
        receipt_dir=args.receipt_dir,
    )
    replay_receipt.write_private_receipt(path, receipt)
    return path, receipt


def parse_args(argv: list[str]) -> argparse.Namespace:
    p = argparse.ArgumentParser(description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter)
    p.add_argument("proposal_id", help="UUID of the approved proposal to apply")
    p.add_argument(
        "--applied-by",
        default=SERVICE_AGENT_HANDLE,
        help="agent handle recorded as applied_by_handle and resolved to the "
        "applied_by_agent_id FK (default: the kb-apply service agent)",
    )
    p.add_argument("--dry-run", action="store_true", help="print the SQL, do not execute")
    p.add_argument("--secrets-file", default=DEFAULT_SECRETS_FILE)
    p.add_argument("--container", default=DEFAULT_CONTAINER)
    p.add_argument("--db", default=DEFAULT_DB)
    p.add_argument("--host", default=DEFAULT_HOST)
    p.add_argument("--role", default=DEFAULT_ROLE)
    p.add_argument(
        "--expected-contract-version",
        type=int,
        choices=(2, 3),
        default=None,
        help="reassert one exact proposal contract before any mutating SQL or receipt recovery",
    )
    p.add_argument(
        "--fresh-preflight",
        type=Path,
        help=(
            "exact proposal_apply_fresh_preflight JSON; makes approve_claim use strict fresh-owned inserts "
            "bound to the timestamped preflight"
        ),
    )
    p.add_argument(
        "--receipt-only",
        action="store_true",
        help="recover the private row-level replay receipt for an already applied proposal; no writes",
    )
    p.add_argument(
        "--receipt-dir",
        default=None,
        help=f"private receipt directory (default: KB_APPLY_RECEIPT_DIR or {replay_receipt.DEFAULT_RECEIPT_DIR})",
    )
    p.add_argument(
        "--receipt-out",
        default=None,
        help="exact private receipt path; overrides --receipt-dir",
    )
    args = p.parse_args(argv)
    if args.receipt_only and args.dry_run:
        p.error("--receipt-only and --dry-run are mutually exclusive")
    return args


def build_apply_sql_for_args(
    proposal: dict[str, Any],
    args: argparse.Namespace,
    applied_by: str | None,
) -> str:
    fresh_preflight_path = getattr(args, "fresh_preflight", None)
    if fresh_preflight_path is None:
        return build_apply_sql(proposal, applied_by)
    if not fresh_preflight_path.is_file():
        raise SystemExit(f"fresh preflight file not found: {fresh_preflight_path}")
    try:
        preflight = json.loads(fresh_preflight_path.read_text(encoding="utf-8"))
    except (OSError, json.JSONDecodeError) as exc:
        raise SystemExit(f"fresh preflight is unreadable: {fresh_preflight_path}") from exc
    import proposal_apply_lifecycle as lifecycle

    try:
        return lifecycle.build_fresh_owned_apply_sql(proposal, preflight, applied_by)
    except ValueError as exc:
        raise SystemExit(f"fresh preflight refused: {exc}") from exc


def main(argv: list[str] | None = None) -> int:
    args = parse_args(sys.argv[1:] if argv is None else argv)

    if args.receipt_only:
        password = load_password(args.secrets_file)
        proposal = load_proposal(args, password)
        assert_expected_contract_version(proposal, args.expected_contract_version)
        assert_receiptable(proposal)
        path, receipt = capture_replay_receipt(args, proposal, password)
        print(
            json.dumps(
                {
                    "proposal_id": proposal["id"],
                    "receipt_path": str(path),
                    "replay_material_sha256": receipt["hashes"]["replay_material_sha256"],
                    "replay_ready": receipt["replay_ready"],
                },
                sort_keys=True,
            )
        )
        return 0

    if args.dry_run:
        # Dry-run still needs the proposal to build SQL; read it as kb_apply.
        password = load_password(args.secrets_file)
        proposal = load_proposal(args, password)
        assert_expected_contract_version(proposal, args.expected_contract_version)
        assert_applyable(proposal)
        print(build_apply_sql_for_args(proposal, args, args.applied_by))
        return 0

    password = load_password(args.secrets_file)
    proposal = load_proposal(args, password)
    assert_expected_contract_version(proposal, args.expected_contract_version)
    assert_applyable(proposal)
    sql = build_apply_sql_for_args(proposal, args, args.applied_by)
    run_psql(args, sql, password)
    applied_proposal = load_proposal(args, password)
    try:
        path, receipt = capture_replay_receipt(
            args,
            applied_proposal,
            password,
            apply_sql=sql,
        )
    except Exception as exc:
        raise SystemExit(
            f"proposal {proposal['id']} committed, but private replay receipt capture failed: {exc}. "
            f"Recover without reapplying via: {Path(__file__).name} {proposal['id']} --receipt-only"
        ) from exc
    print(
        json.dumps(
            {
                "applied": True,
                "proposal_id": proposal["id"],
                "proposal_type": proposal["proposal_type"],
                "receipt_path": str(path),
                "replay_material_sha256": receipt["hashes"]["replay_material_sha256"],
                "replay_ready": receipt["replay_ready"],
            },
            sort_keys=True,
        )
    )
    return 0


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