Plan: chain-v2 verifiers — family-leased tail + scrub engine, token-confidential fencing, the unified chain namespace, citation attestation (#1205 audit, #1206 FTI, epic &73)

/// Which projections a family reads — ALWAYS view-mediated (C8 + rename
/// stability): audit → audit_hashed_preimage_v / _archive_v; fti →
/// fti_hashed_preimage_v / _archive_v (which gain `id` in MR-1).
/// EleGrant → UnsupportedFamily.
pub fn preimage_views(family: ChainFamily) -> Result<(&'static str, &'static str), ChainVerifyError>;

/// One row as read from a preimage view — the FULL projection. The payload is
/// fetched as TEXT (`canonical_event_payload::text` in the SELECT list — the
/// views themselves are unchanged) so the D14 number fence sees the stored
/// rendering BEFORE any serde normalization; it is parsed into `Value` only
/// after the fence passes.
pub struct PreimageRow {
    pub chain_seq: i64,
    pub previous_hash: EventHash,
    pub event_hash: EventHash,
    pub chain_source: ChainSource,
    pub formula: u16,
    pub side: RowSide,                // Live | Archive — the union tag (D1a)
    pub payload_text: String,         // fenced (D14), then parsed
    pub payload: serde_json::Value,   // STRUCTURAL-VALUE: canonical JSON, re-hashed after validation
    pub derived: DerivedColumns,      // per-family enum: every typed hashed column the view exposes
}
pub enum RowSide { Live, Archive }
pub enum DerivedColumns { Audit(AuditDerived), Fti(FtiDerived) }
// AuditDerived: event_id, event_type, event_timestamp, user_id, user_role, action,
//   resource_type, resource_id, source_service, household_id, metadata (fetched as
//   TEXT + fenced like the payload, then Value — STRUCTURAL-VALUE).
// FtiDerived: id, accessed_by, accessed_at, purpose_code, data_elements_accessed,
//   originating_system, action, resource_type, resource_id.

#[derive(Debug, thiserror::Error)]
pub enum ChainVerifyError {
    Database(#[from] sqlx::Error),        // availability — retry/backoff
    Topology(canopy_common::chain_append::ChainAppendError), // fetch_topology seam (explicit wrap, not From)
    Reject(#[from] VerifyReject),         // integrity — latch material, NEVER retried
    UnsupportedFamily { family: ChainFamily },
    Fenced,                               // lost the token/CAS — stop silently
}
/// Decode/canonicalization/number-fence failures on PERSISTED data construct
/// VerifyReject::MalformedRow — an integrity finding. canopy_chain::ChainError
/// from the environment stays an error path; there is NO blanket Chain→retry arm.

/// The C6 rejection taxonomy = the pinned chain_incidents.kind vocabulary,
/// now ALSO a CHECK constraint (D2) — Rust enum, SQL CHECK, and the kind→loop
/// mapping are test-pinned against each other:
/// hash_mismatch, linkage_break, noncontiguous_seq, duplicate_seq,
/// formula_version, payload_set_violation, derived_column_mismatch,
/// routing_mismatch, source_mismatch, id_mismatch, genesis_mismatch,
/// terminal_mismatch, head_regression, rows_beyond_head, missing_head,
/// unexpected_head, missing_shard_rows, target_hash_mismatch,
/// manifest_divergence, manifest_metadata_mismatch, boundary_unavailable,
/// malformed_row.
pub enum VerifyReject { /* one variant per kind, positional evidence fields */ }

pub struct VerifyTarget { pub seq: i64, pub hash: EventHash }
pub struct VerifyCursor { pub seq: i64, pub hash: EventHash }

/// Verify ONE bounded batch (single-statement archive ∪ live — D1a).
/// Per row, in order:
///   0. the D14 raw number fence over payload_text (and audit metadata text)
///      — violation → MalformedRow;
///   1. contiguity (seq == prev+1; <= prev → DuplicateSeq — including the
///      cross-side duplicate case, which arrives adjacent under D1a);
///   2. linkage (row.previous_hash == running hash);
///   3. formula == 2;
///   4. closed key/type set for the family (payload_set_violation — the same
///      key sets the append fns enforce, mirrored in Rust, test-pinned);
///   5. derived-column consistency: every DerivedColumns field == its payload
///      field (derived_column_mismatch);
///   6. routing placement: shard_for(routing_id, shard_count) == shard
///      (routing_mismatch; routing id = payload event_id for audit, payload id
///      for fti — the Amendment 7 binding);
///   7. source: row.chain_source == topo.source (source_mismatch);
///   8. side sanity: side == Archive requires seq <= archived_through(at
///      fetch); side == Live requires seq > archived_through — a wrong-side
///      row latches (the archived-orphan / wrong-side class, X6);
///   9. (fti) row.id == payload id (id_mismatch);
///  10. envelope rebuild + event_hash recompute == stored (hash_mismatch).
/// If the batch reaches target.seq the running hash MUST equal target.hash
/// (TargetHashMismatch). Returns the advanced cursor + rows/bytes consumed.
pub async fn verify_batch(
    chain: &PgPool, topo: &ActiveTopology, family: ChainFamily, shard: ShardId,
    cursor: VerifyCursor, target: &VerifyTarget, batch: BatchBudget,
) -> Result<BatchOutcome, ChainVerifyError>;
pub struct BatchBudget { pub rows: u32, pub bytes: u64 }
pub struct BatchOutcome { pub cursor: VerifyCursor, pub reached_target: bool,
    pub rows: u32, pub bytes: u64 }

/// Boundary hash for a cursor at `seq` (reads the CHAIN pool): seq == 0 →
/// empty_head_hash(instance, family, epoch, shard, None); else the event_hash
/// of the row at `seq` from whichever side holds it — absent entirely
/// (post-purge, #1208-era) → Reject::BoundaryUnavailable.
pub async fn cursor_hash_at(chain: &PgPool, topo: &ActiveTopology,
    family: ChainFamily, shard: ShardId, seq: i64) -> Result<EventHash, ChainVerifyError>;

/// One-statement heads snapshot (single MVCC snapshot).
pub struct ShardHead { pub shard: ShardId, pub last_seq: i64, pub last_hash: EventHash,
    pub archived_through: i64 }
pub async fn capture_heads(chain: &PgPool, topo: &ActiveTopology, family: ChainFamily)
    -> Result<Vec<ShardHead>, ChainVerifyError>;

/// Structural census (family-lease-serialized, its own cadence — one replica
/// per family per cadence, X2; WHOLE retained history, X6):
/// - head set complete over 0..shard_count, none beyond (MissingHead /
///   UnexpectedHead);
/// - genesis arm: an empty head (last_seq == 0) must carry last_hash ==
///   empty_head_hash (genesis_mismatch);
/// - terminal row exists with event_hash == last_hash, or last_seq <=
///   archived_through with the archive row matching (TerminalMismatch);
/// - rows beyond the CURRENT head — ONE statement per side unioned with the
///   head subselect (single snapshot);
/// - identity-filtered whole-range count == span, computed over archive ∪ live
///   in ONE union statement (MissingShardRows); rows carrying a foreign
///   instance/epoch/shard on EITHER side are detected (the archived-orphan
///   MUST now covers archived rows); on any mismatch the census RE-READS the
///   head/boundary and re-counts ONCE before latching (an archive move between
///   cadences is legal interleaving, never a torn-read breach).
pub async fn structural_census(chain: &PgPool, topo: &ActiveTopology,
    family: ChainFamily, heads: &[ShardHead]) -> Result<(), ChainVerifyError>;

/// Manifest binding — order matters: (1) fetch the latest CONFIRMED anchor row
/// (security pool); (2) decode manifest_bytes (from_canonical_bytes) and
/// compare the DECODED fields against the anchor ROW's caller-supplied
/// metadata — the SEVEN row columns: anchor_seq, kind, epoch,
/// previous_anchor_hash, manifest_hash, instance, family
/// (manifest_metadata_mismatch; chain_anchors has NO source column — the
/// decoded source is hash-bound inside manifest_bytes and checked in step 3);
/// (3) compare identity vs topology (instance/family/source/epoch/shard_count)
/// and vs the COMPILED formula/routing constants; (4) THEN capture heads and
/// check per-tip prefix consistency (tip.last_seq <= head.last_seq AND the
/// chain hash AT tip.last_seq equals tip.last_hash). A tip BEYOND the head
/// triggers ONE re-fetch + re-capture before latching manifest_divergence.
/// On success returns the checked anchor id — stamped on the FAMILY checkpoint
/// row as trusted_manifest_ref via the guarded advance (X5: the family row is
/// the ONE trusted-manifest source; per-shard stamping is REMOVED).
pub async fn manifest_check(security: &PgPool, chain: &PgPool, topo: &ActiveTopology,
    family: ChainFamily) -> Result<Option<CheckedManifest>, ChainVerifyError>;
pub struct CheckedManifest { pub anchor_id: Uuid, pub anchor_seq: i64,
    pub tips: Vec<ShardTipSeq> }

/// Owns the security pool + the verifier's own timeout knobs. Each call opens
/// a transaction, applies SET LOCAL lock_timeout/statement_timeout (the
/// router.rs:27-53 idiom, D11 knobs), runs the statement(s), commits. The
/// job-finalize path exposes the transaction so the run write and the job
/// finalize commit ATOMICALLY (D8).
pub struct VerifierDb { /* security: PgPool, timeouts: VerifyTimeouts */ }

pub struct CheckpointKey { pub instance: ChainInstanceId, pub family: ChainFamily,
    pub epoch: Epoch, pub shard: ShardId, pub loop_kind: LoopKind }
pub enum LoopKind { Tail, Scrub, Family }   // "tail" | "scrub" | "family"
/// The FAMILY key is (instance, family, epoch, shard 0, Family) — PK-distinct
/// from shard 0's tail/scrub rows by loop_kind (D2/D3).

pub struct Lease { pub token: Uuid, pub fence: i64 }

/// Reads go through the token-free view chain_checkpoints_verify_v (X1: the
/// verify role's raw SELECT on the table is REVOKED — the ONLY way to hold a
/// token is to have minted it via acquire).
pub struct CheckpointState { pub fence: i64, pub cursor: Option<VerifyCursor>,
    pub target: Option<VerifyTarget>, pub trusted_manifest_ref: Option<Uuid>,
    pub lease_expires_at: Option<DateTime<Utc>>, pub updated_at: DateTime<Utc>,
    pub cycle_started_at: Option<DateTime<Utc>>,
    pub cycle_completed_at: Option<DateTime<Utc>> }
impl VerifierDb {
  pub async fn read_checkpoint(&self, key: &CheckpointKey)
      -> Result<Option<CheckpointState>, ChainVerifyError>;

  /// chain_checkpoint_acquire: Some(Lease), or None = not acquired (an
  /// unexpired lease — ANY owner's, including our own crashed predecessor
  /// (X1: expiry-only takeover, owner display-only) — or a lost cycle-start
  /// CAS). Duration is a BOUNDED number of seconds; expiry is computed inside
  /// the locked fn (no caller clocks). `init` is REQUIRED for a first-ever
  /// tail/scrub acquire; family acquires carry neither init nor cycle.
  /// Scrub cycle-starts pass `cycle` — a cursor-CAS (X5): the target+reset
  /// write applies ONLY if the row's verified_through_seq still equals
  /// expected_seq, so a delayed worker's stale cycle-start LOSES under the
  /// row lock.
  pub async fn acquire(&self, key: &CheckpointKey, owner: &str, lease_secs: u32,
      init: Option<&VerifyCursor>, cycle: Option<&ScrubCycleStart>)
      -> Result<Option<Lease>, ChainVerifyError>;

  /// chain_checkpoint_advance: exact-token-bound, existing-row-only,
  /// cursor-monotonic. The token stays valid PAST expiry until a takeover
  /// mints a successor (X1: correctness never reads the clock; a finding from
  /// a long batch is never lost to a clock). Health stamps are PROVEN in-fn
  /// (X5): on SCRUB keys cycle_complete only when the presented cursor EQUALS
  /// the stored target (seq AND hash); on the FAMILY key cycle_complete is
  /// the census-cadence stamp — token-gated observability, deliberately NOT
  /// target-proven (census completion is a read-side check with no relational
  /// witness, and no D6 status rule reads the stamp); manifest_ref only on
  /// the FAMILY key and only when it references a CONFIRMED chain_anchors row
  /// of the same (instance, family). cursor is None exactly for family-key
  /// advances (lease refresh / census stamp — the family row has no cursor).
  pub async fn advance(&self, key: &CheckpointKey, lease: &Lease,
      cursor: Option<&VerifyCursor>, lease_secs: u32, cycle_complete: bool,
      manifest_ref: Option<Uuid>) -> Result<(), ChainVerifyError>;   // Err(Fenced) on token mismatch
}
pub struct ScrubCycleStart { pub expected_seq: i64, pub target: VerifyTarget,
    pub boundary: VerifyCursor }

/// Guarded writes (X2): run recording and incident latching validate a token
/// inside the DB. Runs validate the FAMILY token (run rows are family-scoped
/// with an optional shard for error attribution — the any-shard-token hole is
/// gone). Latches validate the token of the lease named by the finding's
/// scope: shard-scoped findings present the shard lease of the DETECTED loop;
/// family-scoped findings (missing/unexpected heads, genesis mismatch,
/// manifest divergence, boundary-unavailable at init) present the FAMILY
/// token with shard NULL.
pub struct VerifyRun { pub loop_kind: LoopKind, pub shard: Option<ShardId>,
    pub started_at: DateTime<Utc>, pub finished_at: DateTime<Utc>,
    pub outcome: RunOutcome, pub error: Option<String>, pub rows_verified: i64,
    pub mode: RunMode, pub job_id: Option<Uuid> }
pub enum RunOutcome { Ok, Rejected, Error }     // 'ok' | 'rejected' | 'error'
pub enum RunMode { Scheduled, Manual }          // manual runs NEVER feed status (X4)
impl VerifierDb {
  pub async fn record_run(&self, family_lease: &Lease, key_family: &CheckpointKey,
      run: &VerifyRun) -> Result<Uuid, ChainVerifyError>;
  /// Same statement inside a caller-owned transaction — the job-finalize
  /// atomicity seam (D8).
  pub async fn record_run_in(&self, tx: &mut PgConnection, family_lease: &Lease,
      key_family: &CheckpointKey, run: &VerifyRun) -> Result<Uuid, ChainVerifyError>;
  pub async fn latch_incident(&self, lease: &Lease, key: &CheckpointKey,
      detected: LoopKind, reject: &VerifyReject) -> Result<Option<Uuid>, ChainVerifyError>;
  /// Halt-gate read — via the evidence-free chain_incidents_verify_v (X3).
  pub async fn family_has_unresolved_incident(&self, instance: ChainInstanceId,
      family: ChainFamily) -> Result<Option<Uuid>, ChainVerifyError>;
}

-- Checkpoints: the family loop kind, the token, the cycle stamps, the CHECK matrix.
ALTER TABLE chain_verification_checkpoints
    DROP CONSTRAINT chain_verification_checkpoints_loop_kind_check;  -- the :296 inline CHECK
ALTER TABLE chain_verification_checkpoints
    ADD COLUMN lease_token UUID,
    ADD COLUMN cycle_started_at TIMESTAMPTZ,
    ADD COLUMN cycle_completed_at TIMESTAMPTZ,
    ADD CONSTRAINT ..._loop_kind CHECK (loop_kind IN ('tail','scrub','family')),
    -- The FAMILY row is (instance, family, epoch, shard 0, 'family') — PK-distinct
    -- from shard 0's tail/scrub rows. It carries NO cursor and NO target; it
    -- carries the family lease, trusted_manifest_ref (X5: the ONE source), and
    -- cycle_completed_at as the census-cadence stamp.
    ADD CONSTRAINT ..._family_shape CHECK (loop_kind <> 'family'
        OR (shard_id = 0 AND target_seq IS NULL AND verified_through_seq = 0
            AND verified_through_hash IS NULL)),
    ADD CONSTRAINT ..._cursor_pair CHECK (loop_kind = 'family'
        OR verified_through_hash IS NOT NULL),
    ADD CONSTRAINT ..._hash_len CHECK ((verified_through_hash IS NULL
            OR octet_length(verified_through_hash) = 32)
        AND (target_hash IS NULL OR octet_length(target_hash) = 32)),
    ADD CONSTRAINT ..._nonneg CHECK (verified_through_seq >= 0 AND fence >= 0
        AND (target_seq IS NULL OR target_seq >= 0)),
    ADD CONSTRAINT ..._target_pair CHECK ((target_seq IS NULL) = (target_hash IS NULL)),
    ADD CONSTRAINT ..._target_ge_cursor CHECK (target_seq IS NULL
        OR target_seq >= verified_through_seq),
    ADD CONSTRAINT ..._lease_trio CHECK ((lease_owner IS NULL) = (lease_expires_at IS NULL)
        AND (lease_owner IS NULL) = (lease_token IS NULL)),
    ADD CONSTRAINT ..._token_v7 CHECK (lease_token IS NULL
        OR uuid_extract_version(lease_token) = 7);

-- Runs: family loop kind, per-shard attribution, the manual/scheduled split.
ALTER TABLE chain_verification_runs
    DROP CONSTRAINT chain_verification_runs_loop_kind_check;         -- the :314 inline CHECK
ALTER TABLE chain_verification_runs
    ADD COLUMN shard_id SMALLINT,          -- NULL = family-scoped / whole-family coverage
    ADD COLUMN mode TEXT NOT NULL CHECK (mode IN ('scheduled','manual')),
    ADD COLUMN job_id UUID,                -- FK added in MR-2 (the jobs table lands there)
    ADD CONSTRAINT ..._loop_kind CHECK (loop_kind IN ('tail','scrub','family')),
    ADD CONSTRAINT ..._outcome CHECK (outcome IN ('ok','rejected','error')),
    ADD CONSTRAINT ..._finish_pair CHECK ((finished_at IS NULL) = (outcome IS NULL)),
    ADD CONSTRAINT ..._finish_order CHECK (finished_at IS NULL OR finished_at >= started_at),
    ADD CONSTRAINT ..._rows_nonneg CHECK (rows_verified >= 0);
CREATE INDEX chain_verification_runs_latest_idx ON chain_verification_runs
    (chain_instance_id, chain_family, chain_epoch, loop_kind, shard_id,
     finished_at DESC NULLS LAST)
    WHERE mode = 'scheduled';   -- per-SCOPE latest-run reads (X5); status reads
                                -- scheduled runs ONLY, and manual runs are
                                -- reached by id via their job — the partial
                                -- predicate matches the D6 view exactly

-- Incidents: the detecting loop is STORED, dedup is NULL-safe, the vocabulary
-- is closed in-schema, evidence is bounded.
ALTER TABLE chain_incidents
    ADD COLUMN detected_loop_kind TEXT NOT NULL
        CHECK (detected_loop_kind IN ('tail','scrub','family')),
    ADD CONSTRAINT ..._family_scope CHECK ((detected_loop_kind = 'family')
        = (shard_id IS NULL)),
    ADD CONSTRAINT ..._kind_vocab CHECK (kind IN ('hash_mismatch','linkage_break',
        'noncontiguous_seq','duplicate_seq','formula_version','payload_set_violation',
        'derived_column_mismatch','routing_mismatch','source_mismatch','id_mismatch',
        'genesis_mismatch','terminal_mismatch','head_regression','rows_beyond_head',
        'missing_head','unexpected_head','missing_shard_rows','target_hash_mismatch',
        'manifest_divergence','manifest_metadata_mismatch','boundary_unavailable',
        'malformed_row')),
    ADD CONSTRAINT ..._evidence_shape CHECK (jsonb_typeof(evidence) = 'object'
        AND pg_column_size(evidence) <= 16384);
CREATE UNIQUE INDEX chain_incidents_latched_uq ON chain_incidents
    (chain_instance_id, chain_family, chain_epoch, shard_id, kind, detected_loop_kind)
    NULLS NOT DISTINCT WHERE state = 'latched';   -- family-scoped (NULL epoch/shard) dedups (X3)

CREATE VIEW chain_checkpoints_verify_v AS SELECT
    chain_instance_id, chain_family, chain_epoch, shard_id, loop_kind,
    target_seq, target_hash, verified_through_seq, verified_through_hash,
    trusted_manifest_ref, lease_owner, lease_expires_at, fence, updated_at,
    cycle_started_at, cycle_completed_at
  FROM chain_verification_checkpoints;            -- EVERYTHING except lease_token
CREATE VIEW chain_incidents_verify_v AS SELECT
    id, chain_instance_id, chain_family, chain_epoch, shard_id,
    detected_at, kind, detected_loop_kind, state
  FROM chain_incidents;                           -- no evidence, no resolution text
-- (owner-transfer both to canopy_chain_owner_security, the substrate $own$ pattern)
REVOKE SELECT ON chain_verification_checkpoints, chain_verification_runs,
    chain_incidents FROM canopy_chain_verify;     -- closes the :863 token/evidence leak
REVOKE INSERT ON chain_verification_runs FROM canopy_chain_verify;  -- the :864 open write
GRANT SELECT ON chain_checkpoints_verify_v, chain_incidents_verify_v
    TO canopy_chain_verify;
GRANT SELECT ON chain_incidents TO canopy_chain_incident_admin;  -- the D7 evidence reader

CREATE FUNCTION chain_checkpoint_acquire(
    p_instance uuid, p_family text, p_epoch int, p_shard smallint, p_loop text,
    p_owner text, p_lease_secs int,
    p_init_seq bigint DEFAULT NULL, p_init_hash bytea DEFAULT NULL,
    p_cycle_expected_seq bigint DEFAULT NULL,
    p_cycle_target_seq bigint DEFAULT NULL, p_cycle_target_hash bytea DEFAULT NULL,
    p_cycle_reset_seq bigint DEFAULT NULL, p_cycle_reset_hash bytea DEFAULT NULL
) RETURNS TABLE (lease_token uuid, fence bigint)
-- Domain: p_lease_secs BETWEEN 1 AND 600 (RAISE — X1: bounded DURATION; expiry
--   := now() + make_interval(secs => p_lease_secs) computed HERE, no caller
--   clocks, no lock-wait erosion).
-- Family arm: p_loop = 'family' requires p_shard = 0 and every init/cycle
--   param NULL (RAISE otherwise); the INSERT arm creates the family row with a
--   NULL cursor hash (the _family_shape/_cursor_pair CHECKs).
-- Tail/scrub arm: INSERT ... ON CONFLICT DO NOTHING with the REQUIRED init
--   cursor, then re-SELECT FOR UPDATE (concurrent first-acquires serialize;
--   absent row + NULL init → RAISE).
-- Takeover is EXPIRY-ONLY for everyone (X1): an unexpired lease — ANY owner,
--   including the caller's own text — returns EMPTY. Owner is display-only; a
--   restarted process waits out its own lease (bounded by p_lease_secs).
-- Cycle-start (scrub keys; all five p_cycle_* or none — RAISE on a partial
--   set) is a cursor-CAS (X5): it applies ONLY if verified_through_seq =
--   p_cycle_expected_seq; a stale worker's delayed cycle-start returns EMPTY
--   under the row lock and resets NOTHING. On success it writes target +
--   cursor reset + cycle_started_at := now() in the same UPDATE
--   (cycle_completed_at is NOT cleared — it stamps the LAST completed cycle).
-- Grant: fence := fence + 1, lease_token := uuidv7(); returns the minted
--   token + new fence — the ONLY fence-raising, row-creating path.

CREATE FUNCTION chain_checkpoint_advance(
    p_instance uuid, p_family text, p_epoch int, p_shard smallint, p_loop text,
    p_lease_token uuid, p_lease_secs int,
    p_verified_seq bigint DEFAULT NULL, p_verified_hash bytea DEFAULT NULL,
    p_cycle_complete boolean DEFAULT false, p_manifest_ref uuid DEFAULT NULL
) RETURNS boolean
-- Existing-row-only (absent → false). Token must EQUAL the stored lease_token
--   (else false — the fence); expiry is NEVER read (X1: a token stays valid
--   past expiry until a takeover mints a successor — a long batch's finding
--   is never lost to a clock; p_lease_secs, domain-checked 1..=600, renews
--   expiry := now() + interval on success).
-- Family arm: p_loop = 'family' requires p_verified_seq/hash NULL and permits
--   p_manifest_ref / p_cycle_complete (the census-cadence stamp). Tail/scrub
--   arms REQUIRE the cursor pair and REJECT p_manifest_ref (RAISE — X5: the
--   family row is the ONE trusted-manifest home).
-- Cursor monotonic: p_verified_seq < verified_through_seq → false; equal seq
--   with a DIFFERENT hash → RAISE 'chain-v2: checkpoint cursor hash
--   divergence' (corruption, never silent).
-- PROOFS (X5 — the scrub-coverage and manifest stamps are never caller
--   assertions; the family census stamp is the stated exception: token-gated
--   observability with no relational witness, and no D6 status rule reads it):
--   p_cycle_complete on a scrub key RAISEs unless p_verified_seq = target_seq
--   AND p_verified_hash = target_hash (checked against the STORED target
--   in-fn); on success stamps cycle_completed_at := now().
--   p_manifest_ref RAISEs unless it references a chain_anchors row with
--   notarization_state = 'confirmed' AND matching (chain_instance_id,
--   chain_family) — a relational check, not trust.
-- Never touches fence/target otherwise; stamps updated_at.

-- Guarded writers:
CREATE FUNCTION chain_run_record(
    p_family_token uuid,
    p_instance uuid, p_family text, p_epoch int, p_loop text, p_shard smallint,
    p_started timestamptz, p_finished timestamptz, p_outcome text, p_error text,
    p_rows bigint, p_mode text, p_job_id uuid
) RETURNS uuid
-- Validates p_family_token = the FAMILY row's lease_token for (p_instance,
-- p_family, p_epoch) — X2: runs are recorded ONLY by the family-lease holder
-- (returns NULL when fenced; the engine maps NULL → Fenced). INSERTs the
-- immutable run row (id := uuidv7()). p_job_id is stored from MR-1 (the FK
-- and the mode⇔job CHECK arrive with the MR-2 jobs table — same-signature
-- CREATE OR REPLACE there, explicitly NOT an overload).

CREATE FUNCTION chain_incident_latch(
    p_token uuid,
    p_instance uuid, p_family text, p_epoch int, p_shard smallint,
    p_detected_loop text, p_kind text, p_evidence jsonb
) RETURNS uuid
-- Scope rule (X3): p_detected_loop = 'family' ⟺ p_shard IS NULL (RAISE
-- otherwise). Token validation matches the finding's scope: family-scoped →
-- the FAMILY row's token; shard-scoped → the (p_shard, p_detected_loop) row's
-- token. p_kind is CHECK-constrained by the table; p_evidence bounded by the
-- _evidence_shape CHECK. INSERT ... ON CONFLICT (chain_instance_id,
-- chain_family, chain_epoch, shard_id, kind, detected_loop_kind)
-- WHERE state = 'latched' DO NOTHING; returns the existing latched id on
-- conflict (idempotent, race-free). Fenced token → NULL.

CREATE FUNCTION chain_incident_resolve(p_id uuid, p_reason text,
    p_evidence_ref text, p_revalidation_run uuid) RETURNS void
-- actor := session_user (recorded by the fn, never caller-supplied);
-- p_evidence_ref REQUIRED. The revalidation run must be: outcome = 'ok',
-- mode = 'manual' (X4: scheduled runs NEVER resolve), matching
-- instance + family, finished_at > incident.detected_at, loop_kind =
-- incident.detected_loop_kind (STORED, never inferred — X3), and
-- scope-covering: run.shard_id IS NULL (whole-family coverage) OR
-- run.shard_id = incident.shard_id.

CREATE ROLE canopy_chain_anchor_emitter NOLOGIN;   -- reconciled fail-closed like the rest
CREATE FUNCTION chain_anchor_transition_emit(
    p_id uuid, p_to text, p_jws text, p_kid text,
    p_external_ref text, p_external_version text) RETURNS void
-- The :701-707 edge matrix MINUS confirmation: pending→{submitted,failed},
-- submitted→failed, failed→submitted; p_to = 'confirmed' → RAISE. COALESCE
-- updates of jws/kid/external_ref/external_version as today.
CREATE FUNCTION chain_anchor_transition_confirm(p_id uuid) RETURNS void
-- submitted→confirmed ONLY; stamps verifier_confirmed_at := now(); touches
-- NO caller-supplied columns (the emitter arms own those).
REVOKE EXECUTE ON FUNCTION chain_anchor_append(uuid,text,int,bigint,text,bytea,bytea,bytea)
    FROM canopy_chain_verify;
GRANT EXECUTE ON FUNCTION chain_anchor_append(uuid,text,int,bigint,text,bytea,bytea,bytea)
    TO canopy_chain_anchor_emitter;
GRANT EXECUTE ON FUNCTION chain_anchor_transition_emit(uuid,text,text,text,text,text)
    TO canopy_chain_anchor_emitter;
GRANT EXECUTE ON FUNCTION chain_anchor_transition_confirm(uuid)
    TO canopy_chain_verify;

CREATE UNIQUE INDEX audit_events_archive_v2_event_id_uq
    ON audit_events_archive_v2 (((canonical_event_payload->>'event_id')));
-- and in the tanf/medicaid MR-1 migrations (alongside the view change):
--   CREATE UNIQUE INDEX fti_audit_log_archive_v2_id_uq
--       ON fti_audit_log_archive_v2 (id);

DROP FUNCTION chain_checkpoint_advance(uuid, text, int, smallint, text,
    bigint, text, timestamptz, bigint, bytea);
DROP FUNCTION chain_incident_latch(uuid, text, int, smallint, text, jsonb);
DROP FUNCTION chain_incident_resolve(uuid, text, text, text, uuid);
DROP FUNCTION chain_anchor_transition(uuid, text, text, text, text, text);
-- (chain_checkpoint_acquire and chain_run_record are new — nothing to drop.
-- The grant-matrix test asserts to_regprocedure(<each old signature>) IS NULL
-- post-migration.)

#[derive(Serialize, Deserialize)] #[serde(rename_all = "snake_case")]
pub enum ChainFamilyWire { Audit, Fti }
#[derive(Serialize, Deserialize)]
pub enum ChainServiceWire {
    #[serde(rename = "canopy-tanf")] CanopyTanf,
    #[serde(rename = "canopy-medicaid")] CanopyMedicaid }
#[derive(Serialize, Deserialize)] #[serde(rename_all = "snake_case")]
pub enum ChainVerifyState { Unknown, Verifying, Healthy, Stale, Error, Breached }
#[derive(Serialize, Deserialize)] #[serde(rename_all = "snake_case")]
pub enum ChainLoopWire { Tail, Scrub, Family }
#[derive(Serialize, Deserialize)] #[serde(rename_all = "kebab-case")]
pub enum RequestedLoopWire { Tail, Scrub, FamilyFull }      // matches the DB vocabulary
#[derive(Serialize, Deserialize)] #[serde(rename_all = "snake_case")]
pub enum JobStateWire { Queued, Running, Done, Error }
#[derive(Serialize, Deserialize)] #[serde(rename_all = "snake_case")]
pub enum JobErrorCode { CoverageIncomplete, VerifierError, IntegrityRejected, Crashed }
#[derive(Serialize, Deserialize)] #[serde(rename_all = "snake_case")]
pub enum AttestRefusal { UnknownEvent, NewerThanCheckpoint, BeyondTrustedManifest,
    StateNotAttestable, VerifierUnavailable, ForeignTopology }
// StatusReason: the closed reasons vocabulary (one variant per firing D6 input),
// #[serde(rename_all = "snake_case")], enumerated + test-pinned in the source.

pub struct ChainPosition { pub instance: Uuid, pub family: ChainFamilyWire,
    pub epoch: i32, pub shard: u16, pub seq: i64 }   // instance+family: X9 —
    // attestation and the row banner filter to the EXACT topology
pub struct ChainBacklog { pub staged: i64, pub parked: i64,
    pub inbox_parked: i64, pub dlq_depth: i64 }

pub struct ShardCoverage {
    pub shard_id: u16,
    pub tail_verified_through: i64,
    pub head_seq: i64,
    pub tail_lag: i64,
    pub tail_updated_at: Option<DateTime<Utc>>,
    pub scrub_verified_through: i64,
    pub scrub_target: Option<i64>,
    pub scrub_updated_at: Option<DateTime<Utc>>,
    pub scrub_cycle_started_at: Option<DateTime<Utc>>,    // first-cycle staleness input
    pub scrub_cycle_completed_at: Option<DateTime<Utc>>,  // coverage stamp
}

pub struct ChainStatusResponse {
    pub family: ChainFamilyWire,
    pub service: Option<ChainServiceWire>,
    pub state: ChainVerifyState,
    pub reasons: Vec<StatusReason>,
    pub epoch: Option<i32>,
    pub shards: Vec<ShardCoverage>,
    pub backlog: Option<ChainBacklog>,   // None for fti — typed applicability,
                                         // never zeroed i64s (X5)
    pub trusted_manifest: Option<TrustedManifestSummary>,
    pub incident_id: Option<Uuid>,
    pub breached_position: Option<ChainPosition>,  // the typed row-banner source (X9)
}
pub struct TrustedManifestSummary { pub anchor_id: Uuid, pub anchor_seq: i64,
    pub age_secs: i64 }

pub struct ChainVerifyJobAccepted { pub job_id: Uuid, pub poll_url: String }
pub struct ChainVerifyJobStatus { pub job_id: Uuid, pub state: JobStateWire,
    pub requested_loop: RequestedLoopWire, pub attempts: i32,
    pub run: Option<ChainRunSummary>, pub error_code: Option<JobErrorCode> }
#[derive(Serialize, Deserialize)] #[serde(rename_all = "snake_case")]
pub enum RunOutcomeWire { Ok, Rejected, Error }     // = the D2 CHECK vocabulary
pub struct ChainRunSummary { pub run_id: Uuid, pub loop_kind: ChainLoopWire,
    pub outcome: Option<RunOutcomeWire>, pub rows_verified: i64,
    pub started_at: DateTime<Utc>, pub finished_at: Option<DateTime<Utc>> }
pub struct ChainAttestation { pub attested: bool, pub event_id: Uuid,
    pub position: Option<ChainPosition>, pub verified_through: Option<i64>,
    pub trusted_anchor_seq: Option<i64>, pub state: ChainVerifyState,
    pub reason: Option<AttestRefusal> }

CREATE TABLE chain_verify_jobs (
  id UUID PRIMARY KEY CHECK (uuid_extract_version(id) = 7),
  chain_family TEXT NOT NULL CHECK (chain_family IN ('audit','fti')),
  fti_source TEXT CHECK (fti_source IN ('canopy-tanf','canopy-medicaid')),
  CHECK ((chain_family = 'fti') = (fti_source IS NOT NULL)),
  requested_loop TEXT NOT NULL CHECK (requested_loop IN ('tail','scrub','family-full')),
  incident_id UUID REFERENCES chain_incidents(id),   -- revalidation jobs
  requested_by TEXT NOT NULL CHECK (requested_by <> ''),
  requested_at TIMESTAMPTZ NOT NULL DEFAULT now(),
  state TEXT NOT NULL DEFAULT 'queued'
      CHECK (state IN ('queued','running','done','error')),
  -- The durable work definition, written ONCE at first claim (X4): reclaim
  -- resumes the SAME instance/epoch/target vector — a crash never re-captures
  -- weaker targets. Shape: {"shards":[{"shard":N,"seq":S,"hash":"<hex>"},…]}.
  chain_instance_id UUID, chain_epoch INT, captured_targets JSONB,
  CHECK ((captured_targets IS NULL) = (chain_instance_id IS NULL)
     AND (captured_targets IS NULL) = (chain_epoch IS NULL)),
  -- The claim is DB-minted-token-based (the checkpoint ABA fix, symmetric):
  claim_owner TEXT, claim_token UUID, claim_expires_at TIMESTAMPTZ,
  heartbeat_at TIMESTAMPTZ,
  attempts INT NOT NULL DEFAULT 0 CHECK (attempts >= 0),
  run_id UUID REFERENCES chain_verification_runs(id),
  error_code TEXT CHECK (error_code IN
      ('coverage_incomplete','verifier_error','integrity_rejected','crashed')),
  finished_at TIMESTAMPTZ,
  -- The full state matrix (X4):
  CHECK ((state = 'running') = (claim_owner IS NOT NULL)),
  CHECK ((claim_owner IS NULL) = (claim_token IS NULL)
     AND (claim_owner IS NULL) = (claim_expires_at IS NULL)
     AND (claim_owner IS NULL) = (heartbeat_at IS NULL)),
  CHECK ((state IN ('done','error')) = (finished_at IS NOT NULL)),
  CHECK (state <> 'done' OR run_id IS NOT NULL),
  CHECK ((state = 'error') = (error_code IS NOT NULL)),
  CHECK (state <> 'queued' OR (run_id IS NULL AND captured_targets IS NULL
         AND attempts = 0))
);
CREATE INDEX chain_verify_jobs_queued_idx ON chain_verify_jobs (requested_at)
    WHERE state = 'queued';
CREATE INDEX chain_verify_jobs_reclaim_idx ON chain_verify_jobs (claim_expires_at)
    WHERE state = 'running';                                    -- reclaim scan (X4)
CREATE INDEX chain_verify_jobs_reap_idx ON chain_verify_jobs (finished_at)
    WHERE state IN ('done','error');                            -- reap scan (X4)
CREATE UNIQUE INDEX chain_verify_jobs_active_target_uq ON chain_verify_jobs
    (chain_family, fti_source) NULLS NOT DISTINCT
    WHERE state IN ('queued','running');       -- ONE active job per target (X4)

chain_job_enqueue(p_family text, p_source text, p_loop text, p_requested_by text,
    p_incident uuid, p_max_queued int) RETURNS TABLE (job_id uuid, created boolean)
--   An existing active job for the target → (its id, false) — the handler maps
--   to 409 verification_in_progress (idempotent: no duplicate work, the caller
--   learns the in-flight id). Global queued count >= p_max_queued → RAISE
--   (handler → 503). p_incident must reference an existing incident.  [_app]
chain_job_claim(p_family text, p_source text, p_worker text, p_claim_secs int)
    RETURNS SETOF chain_verify_jobs
--   TARGET-SCOPED (X4): each family task claims ONLY its own (family, source).
--   FOR UPDATE SKIP LOCKED, oldest queued OR running-with-expired-claim
--   (reclaim). Mints claim_token := uuidv7(), attempts := attempts + 1,
--   claim_expires_at := now() + p_claim_secs (domain 1..=600 in-fn).
--   captured_targets is PRESERVED on reclaim.                        [verify]
chain_job_capture(p_id uuid, p_claim_token uuid, p_instance uuid, p_epoch int,
    p_targets jsonb) RETURNS void
--   Token-validated; writes the work definition ONLY when captured_targets IS
--   NULL (first capture wins; a reclaimer sees non-NULL and RESUMES).  [verify]
chain_job_heartbeat(p_id uuid, p_claim_token uuid, p_claim_secs int) RETURNS void
--   Token-validated; extends claim_expires_at, stamps heartbeat_at.   [verify]
chain_job_finalize(p_id uuid, p_claim_token uuid, p_state text, p_run uuid,
    p_error_code text) RETURNS void
--   Token-validated; state ∈ {done, error}; nulls the claim fields; called in
--   the SAME transaction as chain_run_record (VerifierDb::record_run_in).
--                                                                     [verify]
chain_job_reap(p_older_than_days int) RETURNS bigint
--   Floor enforced IN the fn: p_older_than_days >= 7 (RAISE below).
--   Terminal-only; rows whose incident_id references a still-latched incident
--   are EXEMPT until resolved.                                        [verify]