yantrikdb 0.7.1

//! RFC 008 Phase 1: Warrant Flow — the control stack that replaces scalar
//! confidence. This module implements the three operators (⊕, ⋈, ↝_m) and
//! the mobility state M(c|ρ) that they operate on.
//!
//! Start here when reading: the mobility state is a 13-dim vector per
//! (proposition_id, regime). It is NOT a confidence score. It represents
//! how the claim's warrant is moving through its epistemic neighborhood:
//!
//!   σ — support mass (weighted, dependence-discounted)
//!   α — attack mass
//!   δ — source diversity
//!   ι — effective independence (ratio of support to raw sum)
//!   τ — temporal coherence
//!   γ — transportability across regimes
//!   μ — mutability
//!   λ — load-bearingness in downstream dependencies
//!   χ — modality consilience (cross-modal independent corroboration)
//!   ψ_l — self-generation ratio (immediate)
//!   ψ_a — self-generation ratio (ancestral)
//!   κ — contamination risk (shared pipelines)
//!   ν — novelty isolation
//!
//! Components are materialized at three tiers:
//!   write (<10ms per claim insert): σ, α, χ, ψ_l
//!   read (50-200ms, query-cached): Γ, R_c(u)
//!   background (async snapshot-tagged): ψ_a, κ, ν, τ, γ, μ, λ, δ, ι refinements
//!
//! M3 locked spec (Saga note 14, GPT-5.4 red-team session bab6d0b7):
//!   ⊕ is a *deterministic functional over the live claim set* — not a
//!   stream-fold, not order-sensitive. For each live claim k, overlap is
//!   measured against the union of *all other* live claims' lineages
//!   (leave-one-out symmetric). Adding or removing a claim is not a local
//!   delta; it is a full recompute. The recompute is idempotent via
//!   content_hash — if the hash of the current live claim set matches the
//!   stored hash, we skip the UPSERT.

use rusqlite::{params, Connection};
use serde::{Deserialize, Serialize};
use std::collections::{HashMap, HashSet};

use crate::error::Result;

/// Current formula version. Bump when any part of the ⊕ definition, the
/// per-dimension overlap semantics, or the content_hash input changes —
/// doing so causes existing `mobility_state` rows to be recomputed the
/// next time they are accessed (state_status='stale_formula'). Background
/// reconciler upgrades stale rows proactively.
pub const FORMULA_VERSION: i32 = 1;

/// Dependence-discount penalty weights for the accumulation operator ⊕.
/// Fixed in v1; RFC 008 Phase 2 will make them learned.
///
///   ω_k = 1 / (1 + 0.5·D_k + 0.3·P_k + 0.7·S_k)
///
/// where D_k, P_k, S_k are the leave-one-out overlaps described on
/// `accumulate_mass`. Rationale for the ratios:
///   0.5 — source overlap is the primary anti-echo-chamber lever
///   0.3 — pipeline overlap is secondary (same extractor ≠ full duplication)
///   0.7 — self-generation is the strongest discount (prevents self-loops)
pub const DEPENDENCE_WEIGHT_SOURCE: f64 = 0.5;
pub const DEPENDENCE_WEIGHT_PIPELINE: f64 = 0.3;
pub const DEPENDENCE_WEIGHT_SELF_GEN: f64 = 0.7;

/// Max modality count used to normalize χ into [0, 1]: text, image,
/// numeric, audio, code, telemetry.
const MAX_MODALITIES: f64 = 6.0;

/// State lifecycle values for `mobility_state.state_status`.
pub mod state_status {
    pub const FRESH: &str = "fresh";
    pub const RECOMPUTING: &str = "recomputing";
    pub const FAILED: &str = "failed";
    pub const STALE_FORMULA: &str = "stale_formula";
}

/// A single row of the `mobility_state` table. All 13 mobility components
/// are Option because they are populated at different tiers — a freshly
/// recomputed write-tier row only fills in σ, α, χ, ψ_l; the rest are
/// NULL until the background job computes them.
#[derive(Debug, Clone, Serialize, Deserialize, Default)]
pub struct MobilityState {
    pub proposition_id: String,
    pub regime: String,
    pub snapshot_ts: f64,
    pub support_mass: Option<f64>,            // σ
    pub attack_mass: Option<f64>,             // α
    pub source_diversity: Option<f64>,        // δ
    pub effective_independence: Option<f64>,  // ι
    pub temporal_coherence: Option<f64>,      // τ
    pub transportability: Option<f64>,        // γ
    pub mutability: Option<f64>,              // μ
    pub load_bearingness: Option<f64>,        // λ
    pub modality_consilience: Option<f64>,    // χ
    pub self_gen_local: Option<f64>,          // ψ_l
    pub self_gen_ancestral: Option<f64>,      // ψ_a
    pub contamination_risk: Option<f64>,      // κ
    pub novelty_isolation: Option<f64>,       // ν
    /// JSON array of component names populated at each tier.
    pub tier_write_components: Vec<String>,
    pub tier_read_components: Vec<String>,
    pub tier_bg_components: Vec<String>,
    /// M3 reproducibility fields.
    pub formula_version: i32,
    pub content_hash: String,
    pub live_claim_count: i64,
    pub state_status: String,
    pub computed_at: i64,
}

/// A narrow projection of a claim row containing the fields needed by
/// both the mobility (⊕) and contest (⋈) write-tier computations. Fetched
/// once per recompute pass and shared between the two.
#[derive(Debug, Clone)]
struct ClaimRow {
    claim_id: String,
    polarity: i32,
    weight: f64,
    extractor: String,
    source_lineage: Vec<String>, // normalized to deduped, sorted
    self_generated: bool,
    modality_signal: String,
    // RFC 008 M4 additions — used only by contest_state computation.
    // source_memory_rid is the artifact-level identity (same document/event)
    // used to gate the same-artifact-extractor-polarity-conflict counter.
    // namespace drives referent_schema_heterogeneity_count. valid_from/to
    // drive the temporal split between overlap-conflict and separable-
    // opposition.
    source_memory_rid: Option<String>,
    namespace: String,
    valid_from: Option<f64>,
    valid_to: Option<f64>,
}

impl crate::engine::YantrikDB {
    /// Compute the write-tier mobility state for (proposition_id, regime).
    ///
    /// Reads all live (non-tombstoned) claims, computes a content_hash
    /// over the normalized input. If a mobility_state row with matching
    /// hash already exists, returns it unchanged (idempotent — repeated
    /// calls on the same live set are free). Otherwise, runs the full
    /// ⊕ accumulation and upserts a fresh row with state_status='fresh'.
    ///
    /// This is the hot-path entry point from the ingestion hook and the
    /// tombstone path. Transaction scope is controlled by the caller —
    /// typically `ingest_claim` wraps both the INSERT and this call in
    /// one BEGIN IMMEDIATE → COMMIT.
    pub fn compute_write_tier_mobility(
        &self,
        proposition_id: &str,
        regime: &str,
    ) -> Result<MobilityState> {
        let conn = self.conn.lock();
        compute_write_tier_mobility_conn(&conn, proposition_id, regime)
    }

    /// Read mobility state for (proposition_id, regime). Returns the most
    /// recent snapshot by snapshot_ts. None if none recorded.
    pub fn get_mobility_state(
        &self,
        proposition_id: &str,
        regime: &str,
    ) -> Result<Option<MobilityState>> {
        let conn = self.conn.lock();
        read_latest_state(&conn, proposition_id, regime)
    }

    /// Upsert mobility state, replacing any row with the same
    /// (proposition_id, regime, snapshot_ts) key. Usually called by
    /// `compute_write_tier_mobility`; background tiers will also use it
    /// to append snapshot-tagged rows with later snapshot_ts values.
    pub fn upsert_mobility_state(&self, state: &MobilityState) -> Result<()> {
        let conn = self.conn.lock();
        upsert_mobility_state_inner(&conn, state)
    }

    /// RFC 008 M6 — background-tier mobility recompute. Fills the
    /// currently-NULL components τ (temporal_coherence), λ (load_bearingness),
    /// and ψ_a (self_gen_ancestral) on the existing mobility_state row for
    /// (proposition_id, regime). Does NOT recompute write-tier components
    /// (σ, α, χ, ψ_l) — those are authoritative from M3's leave-one-out
    /// ingestion hook.
    ///
    /// Idempotent under an unchanged claim + move graph: repeat calls
    /// produce the same values. If no write-tier row exists yet, this is
    /// a no-op (returns None).
    ///
    /// Designed to be called by a scan job or explicit app-code. No lock-
    /// scope coupling with claim ingestion; consumers can run it whenever.
    pub fn compute_background_mobility(
        &self,
        proposition_id: &str,
        regime: &str,
    ) -> Result<Option<MobilityState>> {
        let conn = self.conn.lock();
        let Some(mut state) = read_latest_state(&conn, proposition_id, regime)? else {
            return Ok(None);
        };
        let tau = compute_temporal_coherence(&conn, proposition_id, regime)?;
        let lambda = compute_load_bearingness(&conn, proposition_id)?;
        let psi_a = compute_self_gen_ancestral(&conn, proposition_id, 2)?;
        state.temporal_coherence = Some(tau);
        state.load_bearingness = Some(lambda);
        state.self_gen_ancestral = Some(psi_a);
        // Mark background tier components as present. Extend the tier_bg
        // component list idempotently so reads know what's been populated.
        for name in ["temporal_coherence", "load_bearingness", "self_gen_ancestral"] {
            if !state.tier_bg_components.iter().any(|c| c == name) {
                state.tier_bg_components.push(name.to_string());
            }
        }
        upsert_mobility_state_inner(&conn, &state)?;
        Ok(Some(state))
    }

    /// RFC 008 M6 — batch scan. Walks mobility_state rows whose background
    /// components are NULL (temporal_coherence IS NULL serves as the
    /// sentinel — the three background components are populated together),
    /// up to `limit` rows, and fills them in. Returns the number of rows
    /// updated.
    ///
    /// Safe to call from `think()` or a periodic job. Each row is
    /// processed in its own lock acquisition so long batches don't
    /// monopolize the writer.
    pub fn recompute_background_mobility_batch(&self, limit: usize) -> Result<usize> {
        let pending = self.list_background_pending(limit)?;
        let mut count = 0;
        for (prop_id, regime) in pending {
            if self.compute_background_mobility(&prop_id, &regime)?.is_some() {
                count += 1;
            }
        }
        Ok(count)
    }

    fn list_background_pending(&self, limit: usize) -> Result<Vec<(String, String)>> {
        let conn = self.conn.lock();
        let mut stmt = conn.prepare(
            "SELECT proposition_id, regime FROM mobility_state \
             WHERE temporal_coherence IS NULL \
             ORDER BY computed_at ASC LIMIT ?1",
        )?;
        let rows = stmt
            .query_map(params![limit as i64], |row| {
                Ok((row.get::<_, String>(0)?, row.get::<_, String>(1)?))
            })?
            .collect::<std::result::Result<Vec<_>, _>>()?;
        Ok(rows)
    }
}

// ──────────────────────────────────────────────────────────────────
// Connection-level helpers. These accept a `&Connection` so callers
// can use them inside their own transaction scope.
// ──────────────────────────────────────────────────────────────────

fn read_latest_state(
    conn: &Connection,
    proposition_id: &str,
    regime: &str,
) -> Result<Option<MobilityState>> {
    let mut stmt = conn.prepare(
        "SELECT proposition_id, regime, snapshot_ts, \
         support_mass, attack_mass, source_diversity, effective_independence, \
         temporal_coherence, transportability, mutability, load_bearingness, \
         modality_consilience, self_gen_local, self_gen_ancestral, \
         contamination_risk, novelty_isolation, \
         tier_write_components, tier_read_components, tier_bg_components, \
         formula_version, content_hash, live_claim_count, state_status, computed_at \
         FROM mobility_state \
         WHERE proposition_id = ?1 AND regime = ?2 \
         ORDER BY snapshot_ts DESC LIMIT 1",
    )?;
    let result = stmt.query_row(params![proposition_id, regime], row_to_mobility_state);
    match result {
        Ok(state) => Ok(Some(state)),
        Err(rusqlite::Error::QueryReturnedNoRows) => Ok(None),
        Err(e) => Err(e.into()),
    }
}

fn upsert_mobility_state_inner(conn: &Connection, state: &MobilityState) -> Result<()> {
    let tier_write = serde_json::to_string(&state.tier_write_components)?;
    let tier_read = serde_json::to_string(&state.tier_read_components)?;
    let tier_bg = serde_json::to_string(&state.tier_bg_components)?;
    conn.execute(
        "INSERT INTO mobility_state (\
         proposition_id, regime, snapshot_ts, \
         support_mass, attack_mass, source_diversity, effective_independence, \
         temporal_coherence, transportability, mutability, load_bearingness, \
         modality_consilience, self_gen_local, self_gen_ancestral, \
         contamination_risk, novelty_isolation, \
         tier_write_components, tier_read_components, tier_bg_components, \
         formula_version, content_hash, live_claim_count, state_status, computed_at) \
         VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9, ?10, ?11, ?12, ?13, ?14, ?15, ?16, \
                 ?17, ?18, ?19, ?20, ?21, ?22, ?23, ?24) \
         ON CONFLICT(proposition_id, regime, snapshot_ts) DO UPDATE SET \
         support_mass = excluded.support_mass, \
         attack_mass = excluded.attack_mass, \
         source_diversity = excluded.source_diversity, \
         effective_independence = excluded.effective_independence, \
         temporal_coherence = excluded.temporal_coherence, \
         transportability = excluded.transportability, \
         mutability = excluded.mutability, \
         load_bearingness = excluded.load_bearingness, \
         modality_consilience = excluded.modality_consilience, \
         self_gen_local = excluded.self_gen_local, \
         self_gen_ancestral = excluded.self_gen_ancestral, \
         contamination_risk = excluded.contamination_risk, \
         novelty_isolation = excluded.novelty_isolation, \
         tier_write_components = excluded.tier_write_components, \
         tier_read_components = excluded.tier_read_components, \
         tier_bg_components = excluded.tier_bg_components, \
         formula_version = excluded.formula_version, \
         content_hash = excluded.content_hash, \
         live_claim_count = excluded.live_claim_count, \
         state_status = excluded.state_status, \
         computed_at = excluded.computed_at",
        params![
            state.proposition_id, state.regime, state.snapshot_ts,
            state.support_mass, state.attack_mass, state.source_diversity,
            state.effective_independence, state.temporal_coherence,
            state.transportability, state.mutability, state.load_bearingness,
            state.modality_consilience, state.self_gen_local,
            state.self_gen_ancestral, state.contamination_risk,
            state.novelty_isolation,
            tier_write, tier_read, tier_bg,
            state.formula_version, state.content_hash, state.live_claim_count,
            state.state_status, state.computed_at,
        ],
    )?;
    Ok(())
}

/// Connection-level variant of `compute_write_tier_mobility` so the
/// ingestion hook can call it from within an already-locked connection
/// scope (same transaction as the claim INSERT).
pub(super) fn compute_write_tier_mobility_conn(
    conn: &Connection,
    proposition_id: &str,
    regime: &str,
) -> Result<MobilityState> {
    let claims = fetch_claims_for_mobility(conn, proposition_id, regime)?;
    let hash = content_hash(&claims);

    // Idempotence: if the latest stored row was computed under the current
    // FORMULA_VERSION with an identical content_hash and is still fresh,
    // no recompute is needed. Repeated inserts of the same (src, dst, rel,
    // extractor, polarity, namespace) resolve to the same set of live
    // claims, so this path is the common case on UPSERT.
    if let Some(existing) = read_latest_state(conn, proposition_id, regime)? {
        if existing.formula_version == FORMULA_VERSION
            && existing.content_hash == hash
            && existing.state_status == state_status::FRESH
        {
            return Ok(existing);
        }
    }

    let support: Vec<&ClaimRow> = claims.iter().filter(|c| c.polarity == 1).collect();
    let attack: Vec<&ClaimRow> = claims.iter().filter(|c| c.polarity == -1).collect();

    let support_mass = accumulate_mass(&support);
    let attack_mass = accumulate_mass(&attack);
    let chi = compute_modality_consilience(&support);
    let psi_l = compute_self_gen_local(&support);

    let state = MobilityState {
        proposition_id: proposition_id.to_string(),
        regime: regime.to_string(),
        snapshot_ts: crate::engine::now(),
        support_mass: Some(support_mass),
        attack_mass: Some(attack_mass),
        modality_consilience: Some(chi),
        self_gen_local: Some(psi_l),
        tier_write_components: vec![
            "support_mass".into(),
            "attack_mass".into(),
            "modality_consilience".into(),
            "self_gen_local".into(),
        ],
        formula_version: FORMULA_VERSION,
        content_hash: hash,
        live_claim_count: claims.len() as i64,
        state_status: state_status::FRESH.to_string(),
        computed_at: unix_seconds(),
        ..Default::default()
    };

    upsert_mobility_state_inner(conn, &state)?;
    Ok(state)
}

// ──────────────────────────────────────────────────────────────────
// Free functions — the actual math, testable without the engine.
// ──────────────────────────────────────────────────────────────────

/// Fetch a narrow projection of claim rows for (proposition_id, regime),
/// deduping and sorting each lineage inline so the downstream hash and
/// Jaccard math is deterministic. ORDER BY claim_id for stable hashes.
fn fetch_claims_for_mobility(
    conn: &Connection,
    proposition_id: &str,
    regime: &str,
) -> Result<Vec<ClaimRow>> {
    let mut stmt = conn.prepare(
        "SELECT claim_id, polarity, weight, extractor, source_lineage, self_generated, \
         modality_signal, source_memory_rid, namespace, valid_from, valid_to \
         FROM claims \
         WHERE proposition_id = ?1 AND regime_tag = ?2 AND tombstoned = 0 \
         ORDER BY claim_id ASC",
    )?;
    let rows = stmt
        .query_map(params![proposition_id, regime], |row| {
            let lineage_json: String = row.get(4)?;
            let lineage = normalize_lineage(&lineage_json);
            let self_gen_int: i64 = row.get(5)?;
            Ok(ClaimRow {
                claim_id: row.get(0)?,
                polarity: row.get(1)?,
                weight: row.get(2)?,
                extractor: row.get(3)?,
                source_lineage: lineage,
                self_generated: self_gen_int != 0,
                modality_signal: row.get(6)?,
                source_memory_rid: row.get(7)?,
                namespace: row.get(8)?,
                valid_from: row.get(9)?,
                valid_to: row.get(10)?,
            })
        })?
        .collect::<std::result::Result<Vec<_>, _>>()?;
    Ok(rows)
}

/// Normalize a source_lineage JSON string into a deduped, sorted vector.
/// NULL and unparseable input become empty vectors. This normalization is
/// also what makes content_hash reproducible across equivalent inputs.
fn normalize_lineage(lineage_json: &str) -> Vec<String> {
    let parsed: Vec<String> = serde_json::from_str(lineage_json).unwrap_or_default();
    let mut set: Vec<String> = parsed.into_iter().collect::<HashSet<_>>().into_iter().collect();
    set.sort();
    set
}

/// Compute the M3 content_hash. Inputs are normalized and sorted so
/// equivalent claim sets produce identical hashes regardless of row
/// order, extractor casing insignificance, or JSON serialization noise.
///
/// The hash covers:
///   1. FORMULA_VERSION (so a version bump invalidates every row)
///   2. sorted claim_ids
///   3. per-claim: polarity, self_generated flag, extractor, modality_signal
///   4. per-claim: sorted source_lineage
///   5. claim weight (3 decimal places to avoid float drift)
///
/// NOT in the hash: created_at, any mutable metadata. We are hashing the
/// *semantic input to the accumulator*, not the row metadata.
fn content_hash(claims: &[ClaimRow]) -> String {
    let mut hasher = blake3::Hasher::new();
    hasher.update(b"yantrikdb.warrant.v");
    hasher.update(&FORMULA_VERSION.to_le_bytes());
    hasher.update(b"\x00claims\x00");
    // claims are already ORDER BY claim_id from fetch_claims_for_mobility,
    // but we do not rely on that — hash is stable under any input order
    // because we re-sort defensively. This also lets the pure-function
    // accumulator tests call content_hash directly.
    let mut indexed: Vec<(&String, &ClaimRow)> = claims.iter().map(|c| (&c.claim_id, c)).collect();
    indexed.sort_by(|a, b| a.0.cmp(b.0));
    for (cid, c) in indexed {
        hasher.update(cid.as_bytes());
        hasher.update(b"|");
        hasher.update(&c.polarity.to_le_bytes());
        hasher.update(b"|");
        hasher.update(&[c.self_generated as u8]);
        hasher.update(b"|");
        hasher.update(c.extractor.as_bytes());
        hasher.update(b"|");
        hasher.update(c.modality_signal.as_bytes());
        hasher.update(b"|");
        // weight rounded to millesimals for hash stability across f64
        // representation variations; the arithmetic still uses full precision.
        let w_scaled = (c.weight * 1000.0).round() as i64;
        hasher.update(&w_scaled.to_le_bytes());
        hasher.update(b"|");
        for src in &c.source_lineage {
            hasher.update(src.as_bytes());
            hasher.update(b",");
        }
        hasher.update(b"\x00");
    }
    hex::encode(hasher.finalize().as_bytes())
}

fn unix_seconds() -> i64 {
    use std::time::{SystemTime, UNIX_EPOCH};
    SystemTime::now()
        .duration_since(UNIX_EPOCH)
        .map(|d| d.as_secs() as i64)
        .unwrap_or(0)
}

/// ⊕ accumulation: sum of weighted supports with leave-one-out dependence
/// discount. For each claim k:
///
///   ω_k = 1 / (1 + λ·D_k + μ·P_k + τ·S_k)
///
/// where the three overlaps are measured against the rest of the live set:
///
///   D_k — Jaccard of source_lineage(k) vs. union of all others' lineages
///   P_k — fraction of other claims sharing the same extractor
///   S_k — 1 if claim k is self-generated AND any other claim is, else 0
///
/// The result is a **deterministic functional over the live claim set** —
/// order-invariant, tombstone-correct. P_k and S_k are degenerate overlap
/// proxies in v1: P_k could become Jaccard over pipeline_lineage once that
/// column exists on claims; S_k could become Jaccard over a self-gen
/// ancestry lineage. Both preserve the set-symmetry property regardless.
pub(super) fn accumulate_mass(claims: &[&ClaimRow]) -> f64 {
    compute_omegas(claims).iter().enumerate().map(|(k, w)| w * claims[k].weight).sum()
}

/// Compute the per-claim discount ω_k vector for a polarity-homogeneous
/// live set. Returns `vec![]` when the input is empty. The result is
/// position-aligned with `claims`. Shared by both ⊕ (mass) and ⋈
/// (effective_independence = Σ ω_k) so both operators use the same
/// leave-one-out semantics from the same live-set snapshot.
fn compute_omegas(claims: &[&ClaimRow]) -> Vec<f64> {
    if claims.is_empty() {
        return Vec::new();
    }
    let mut freq: HashMap<&str, usize> = HashMap::new();
    for c in claims {
        for e in &c.source_lineage {
            *freq.entry(e.as_str()).or_insert(0) += 1;
        }
    }
    let total_distinct = freq.len();

    let mut omegas = Vec::with_capacity(claims.len());
    for (k, claim) in claims.iter().enumerate() {
        let d_k = leave_one_out_jaccard(&claim.source_lineage, &freq, total_distinct);
        let p_k = pipeline_overlap_ratio(claim, claims, k);
        let s_k = self_gen_overlap_binary(claim, claims, k);
        let discount = 1.0
            + DEPENDENCE_WEIGHT_SOURCE * d_k
            + DEPENDENCE_WEIGHT_PIPELINE * p_k
            + DEPENDENCE_WEIGHT_SELF_GEN * s_k;
        omegas.push(1.0 / discount);
    }
    omegas
}

/// Leave-one-out Jaccard of a claim's source_lineage set X against the
/// union U of every *other* claim's source_lineage.
///
///   intersection(X, U) = elements of X that appear in >= 1 other claim
///   union(X, U) = |total_distinct| − |elements only in this claim|
///                 + 0 (since X is a subset of {e : freq[e] ≥ 1})
///                 ... equivalently |total_distinct|.
///
/// Convention for empty inputs (M3 locked spec, revised):
///   Jaccard(∅, _) = Jaccard(_, ∅) = 0  — empty lineage carries no signal.
///
/// Rationale: an empty/missing lineage is informationally inert — absent
/// positive evidence of shared provenance, we don't penalize. This is also
/// what preserves the "single claim has no rest" intuition: with one claim,
/// the rest union is empty, overlap is 0, discount is neutral, ω = 1.
fn leave_one_out_jaccard(
    claim_set: &[String],
    freq: &HashMap<&str, usize>,
    total_distinct: usize,
) -> f64 {
    if claim_set.is_empty() || total_distinct == 0 {
        return 0.0;
    }
    let rest_union_is_empty = {
        // "rest" is empty iff every element in freq has count == 1 AND is
        // in claim_set — i.e., claim_set is the only contributor.
        let claim_set_lookup: HashSet<&str> = claim_set.iter().map(|s| s.as_str()).collect();
        freq.iter().all(|(e, c)| *c == 1 && claim_set_lookup.contains(e))
    };
    if rest_union_is_empty {
        return 0.0;
    }
    // |X ∩ rest_union| = # of claim_set elements e with freq[e] > 1
    let intersection = claim_set
        .iter()
        .filter(|e| freq.get(e.as_str()).copied().unwrap_or(0) > 1)
        .count();
    // |X ∪ rest_union| = |total_distinct| — every element with freq ≥ 1
    // lives somewhere, and claim_set ⊆ freq-keys.
    intersection as f64 / total_distinct as f64
}

/// P_k — fraction of other claims sharing the same extractor. Symmetric
/// under permutation of `claims`, so order-invariant.
fn pipeline_overlap_ratio(claim: &ClaimRow, claims: &[&ClaimRow], own_index: usize) -> f64 {
    if claims.len() <= 1 {
        return 0.0;
    }
    let matching = claims
        .iter()
        .enumerate()
        .filter(|(i, c)| *i != own_index && c.extractor == claim.extractor)
        .count();
    matching as f64 / (claims.len() - 1) as f64
}

/// S_k — 1.0 if this claim is self-generated AND at least one other claim
/// in the live set is also self-generated. 0.0 otherwise. Binary v1;
/// background tier will refine to consider ancestry depth.
fn self_gen_overlap_binary(claim: &ClaimRow, claims: &[&ClaimRow], own_index: usize) -> f64 {
    if !claim.self_generated {
        return 0.0;
    }
    let has_other = claims
        .iter()
        .enumerate()
        .any(|(i, c)| i != own_index && c.self_generated);
    if has_other {
        1.0
    } else {
        0.0
    }
}

/// χ — modality consilience: distinct modalities in the live set, normalized
/// by MAX_MODALITIES. Higher when independent modalities corroborate. [0, 1].
fn compute_modality_consilience(claims: &[&ClaimRow]) -> f64 {
    if claims.is_empty() {
        return 0.0;
    }
    let distinct: HashSet<&String> = claims.iter().map(|c| &c.modality_signal).collect();
    (distinct.len() as f64 / MAX_MODALITIES).min(1.0)
}

/// ψ_l — local self-generation ratio: fraction of supporting claims whose
/// `self_generated` flag is set.
fn compute_self_gen_local(claims: &[&ClaimRow]) -> f64 {
    if claims.is_empty() {
        return 0.0;
    }
    let self_gen_count = claims.iter().filter(|c| c.self_generated).count();
    self_gen_count as f64 / claims.len() as f64
}

// ──────────────────────────────────────────────────────────────────
// RFC 008 Phase 1 M6: Background-tier mobility components.
//
// τ — temporal_coherence: polarity persistence across a proposition's
//     claim history. 1.0 means polarity never flipped; lower means it
//     oscillated. Uses all claims (including tombstoned) because the
//     historical polarity record is the input — tombstoning doesn't
//     erase a past flip.
//
// λ — load_bearingness: raw count of cognitive moves whose inputs
//     reference any claim of this proposition. High λ = heavily load-
//     bearing = revising it has cascading effects. Stored as REAL for
//     schema consistency; consumers can normalize or thresholds as
//     needed.
//
// ψ_a — self_gen_ancestral: fraction of ancestry DAG nodes whose
//     self_generated flag is set. Bounded BFS at depth 2 by default.
//     Returns 0.0 when there is no ancestry to trace (e.g. all claims
//     are leaf evidence with no producing moves).
// ──────────────────────────────────────────────────────────────────

fn compute_temporal_coherence(
    conn: &Connection,
    proposition_id: &str,
    regime: &str,
) -> Result<f64> {
    // Use all claims in the proposition — tombstoned included — because
    // a past polarity flip is real historical information even if later
    // reconciled. Order by created_at for chronological sequence.
    let mut stmt = conn.prepare(
        "SELECT polarity FROM claims \
         WHERE proposition_id = ?1 AND regime_tag = ?2 \
         ORDER BY created_at ASC, claim_id ASC",
    )?;
    let polarities: Vec<i32> = stmt
        .query_map(params![proposition_id, regime], |row| row.get::<_, i32>(0))?
        .collect::<std::result::Result<Vec<_>, _>>()?;
    if polarities.len() < 2 {
        // Single claim (or none) — no basis to judge coherence.
        // Convention: fully coherent by default, since there's no evidence
        // of inconsistency. Callers that want "undefined" can look at
        // live_claim_count on the mobility row instead.
        return Ok(1.0);
    }
    let mut flips = 0;
    for w in polarities.windows(2) {
        if w[0] != w[1] {
            flips += 1;
        }
    }
    let max_flips = (polarities.len() - 1) as f64;
    Ok(1.0 - (flips as f64 / max_flips))
}

fn compute_load_bearingness(conn: &Connection, proposition_id: &str) -> Result<f64> {
    // Count distinct moves that consume any claim of this proposition as
    // input. This measures downstream-reasoning-weight: how many cognitive
    // operations depend on these claims.
    let count: i64 = conn.query_row(
        "SELECT COUNT(DISTINCT mie.move_id) \
         FROM move_input_edge mie \
         INNER JOIN claims c ON mie.claim_id = c.claim_id \
         WHERE c.proposition_id = ?1",
        params![proposition_id],
        |row| row.get(0),
    )?;
    Ok(count as f64)
}

fn compute_self_gen_ancestral(
    conn: &Connection,
    proposition_id: &str,
    max_depth: u32,
) -> Result<f64> {
    use std::collections::HashSet;

    let seed_claims = fetch_proposition_live_claim_ids(conn, proposition_id)?;
    if seed_claims.is_empty() {
        return Ok(0.0);
    }

    // BFS backward through move_output_edge → move_events → move_input_edge.
    // The seed layer itself is NOT counted as ancestry (it's the claims
    // we're evaluating the ancestry of).
    let mut ancestry: HashSet<String> = HashSet::new();
    let mut frontier: Vec<String> = seed_claims;
    for _depth in 0..max_depth {
        if frontier.is_empty() {
            break;
        }
        let mut next: Vec<String> = Vec::new();
        for claim_id in &frontier {
            let moves = fetch_producing_moves(conn, claim_id)?;
            for mv in &moves {
                let inputs = fetch_move_input_claim_ids(conn, mv)?;
                for inp in inputs {
                    if ancestry.insert(inp.clone()) {
                        next.push(inp);
                    }
                }
            }
        }
        frontier = next;
    }
    if ancestry.is_empty() {
        return Ok(0.0);
    }

    // Count self-generated claims in the ancestry set.
    let placeholders: String = (0..ancestry.len())
        .map(|i| format!("?{}", i + 1))
        .collect::<Vec<_>>()
        .join(",");
    let sql = format!(
        "SELECT COUNT(*) FROM claims \
         WHERE claim_id IN ({}) AND self_generated = 1",
        placeholders
    );
    let params_vec: Vec<Box<dyn rusqlite::types::ToSql>> = ancestry
        .iter()
        .map(|c| Box::new(c.clone()) as Box<dyn rusqlite::types::ToSql>)
        .collect();
    let params_ref: Vec<&dyn rusqlite::types::ToSql> =
        params_vec.iter().map(|b| b.as_ref()).collect();
    let self_gen: i64 = conn.query_row(&sql, params_ref.as_slice(), |row| row.get(0))?;
    Ok(self_gen as f64 / ancestry.len() as f64)
}

fn fetch_proposition_live_claim_ids(conn: &Connection, proposition_id: &str) -> Result<Vec<String>> {
    let mut stmt = conn.prepare(
        "SELECT claim_id FROM claims \
         WHERE proposition_id = ?1 AND tombstoned = 0",
    )?;
    let rows = stmt
        .query_map(params![proposition_id], |row| row.get::<_, String>(0))?
        .collect::<std::result::Result<Vec<_>, _>>()?;
    Ok(rows)
}

fn fetch_producing_moves(conn: &Connection, claim_id: &str) -> Result<Vec<String>> {
    let mut stmt = conn.prepare("SELECT move_id FROM move_output_edge WHERE claim_id = ?1")?;
    let rows = stmt
        .query_map(params![claim_id], |row| row.get::<_, String>(0))?
        .collect::<std::result::Result<Vec<_>, _>>()?;
    Ok(rows)
}

fn fetch_move_input_claim_ids(conn: &Connection, move_id: &str) -> Result<Vec<String>> {
    let mut stmt = conn.prepare("SELECT claim_id FROM move_input_edge WHERE move_id = ?1")?;
    let rows = stmt
        .query_map(params![move_id], |row| row.get::<_, String>(0))?
        .collect::<std::result::Result<Vec<_>, _>>()?;
    Ok(rows)
}

// ──────────────────────────────────────────────────────────────────
// RFC 008 Phase 1 M4: Contest operator ⋈ — Γ(c) grounded diagnostics.
//
// ⋈ produces a compact reproducible summary of the SHAPE of contest in
// the live claim set, using only features defensible from lineage and
// claim metadata. No speculative contradiction typing, no pair storage,
// no action recommendations. See Saga note 16 for the locked spec.
// ──────────────────────────────────────────────────────────────────

/// Contest derivation version. Bumped when counter definitions, gating
/// rules, or content_hash input changes. Independent of mobility's
/// FORMULA_VERSION — contest logic evolves on its own schedule.
pub const CONTEST_DERIVATION_VERSION: i32 = 1;

/// Heuristic flag bit positions (mirrors SQL schema comments).
pub mod contest_flags {
    /// σ > 2.0 AND support_effective_independence < 2.0 — apparent consensus
    /// that collapses under dependence discount (rumor amplification risk).
    pub const DUPLICATION_RISK: u64 = 1 << 0;
    /// Same-source opposite-polarity claims detected (source self-conflict,
    /// routes to source audit / chronology review).
    pub const SAME_SOURCE_CONFLICT: u64 = 1 << 1;
    /// Claims span multiple namespaces — referent alignment may be
    /// heterogeneous (NOT asserted as incompatible — just observed).
    pub const REFERENT_HETEROGENEITY_PRESENT: u64 = 1 << 2;
    /// Same underlying source_memory_rid yielded opposite-polarity claims
    /// through different extractors (extraction pathology signal).
    pub const SAME_ARTIFACT_EXTRACTOR_CONFLICT: u64 = 1 << 3;
    /// Opposite-polarity claims with overlapping validity intervals —
    /// present-tense contradiction (not mere state-change-over-time).
    pub const PRESENT_TENSE_CONFLICT: u64 = 1 << 4;
}

/// Materialized contest state for a (proposition_id, regime). This is
/// Γ(c) — a structured contradiction signature, not a net-confidence
/// scalar. Every field is directly computable from claim metadata with
/// no speculative inference.
#[derive(Debug, Clone, Serialize, Deserialize, Default)]
pub struct ContestState {
    pub proposition_id: String,
    pub regime: String,
    pub support_mass: f64,
    pub attack_mass: f64,
    pub support_effective_independence: f64,
    pub attack_effective_independence: f64,
    pub support_distinct_source_count: i64,
    pub attack_distinct_source_count: i64,
    pub same_source_opposite_polarity_count: i64,
    pub same_artifact_extractor_polarity_conflict_count: i64,
    pub temporal_overlap_conflict_count: i64,
    pub temporal_separable_opposition_count: i64,
    pub referent_schema_heterogeneity_count: i64,
    pub heuristic_flags: u64,
    pub derivation_version: i32,
    pub content_hash: String,
    pub live_claim_count: i64,
    pub state_status: String,
    pub computed_at: i64,
}

/// On-demand inspection payload — exemplar claim pairs that drove the
/// contest flags. NOT persisted; recomputed on read. Per GPT-5.4's
/// "top-k exemplars via query, not persisted state" pattern: we keep
/// the substrate table compact (counters only) and expose pair detail
/// only when callers ask for it.
#[derive(Debug, Clone, Serialize, Deserialize, Default)]
pub struct ContestConflictReport {
    pub proposition_id: String,
    pub regime: String,
    pub heuristic_flags: u64,
    /// Pairs of claim_ids where both claims share identical normalized
    /// source_lineage AND have opposite polarity.
    pub same_source_opposite_polarity_pairs: Vec<(String, String)>,
    /// Pairs of claim_ids deriving from the same source_memory_rid via
    /// different extractors, with opposite polarity.
    pub same_artifact_extractor_conflict_pairs: Vec<(String, String)>,
    /// Pairs with opposite polarity AND overlapping validity intervals
    /// (real present-tense contradictions, not state changes over time).
    pub temporal_overlap_conflict_pairs: Vec<(String, String)>,
}

impl crate::engine::YantrikDB {
    /// Compute contest state ⋈ for (proposition_id, regime). Reads all
    /// live claims, computes content_hash, short-circuits on idempotent
    /// match, otherwise derives counters and heuristic flags, then
    /// upserts. Same lock/transaction discipline as M3 mobility.
    pub fn compute_contest_state(
        &self,
        proposition_id: &str,
        regime: &str,
    ) -> Result<ContestState> {
        let conn = self.conn.lock();
        compute_contest_state_conn(&conn, proposition_id, regime)
    }

    /// Read contest state. Returns None if never computed.
    pub fn get_contest_state(
        &self,
        proposition_id: &str,
        regime: &str,
    ) -> Result<Option<ContestState>> {
        let conn = self.conn.lock();
        read_contest_state(&conn, proposition_id, regime)
    }

    /// RFC 008 M4.5 — propositions whose contest_state has ANY of the
    /// bits in `flag_mask` set. Indexed query via idx_contest_flags.
    /// This is the primary post-M4 consumer of contest_state: audit
    /// tools, UIs, or periodic review jobs can list all flagged
    /// propositions by category (e.g. all SAME_SOURCE_CONFLICT for
    /// source-audit routing).
    ///
    /// Passing `flag_mask = 0` returns the empty vec (no flags = no
    /// matches). Use `contest_flags::*` constants to build masks.
    /// Example: `flag_mask = SAME_SOURCE_CONFLICT | SAME_ARTIFACT_EXTRACTOR_CONFLICT`.
    pub fn list_flagged_propositions(
        &self,
        flag_mask: u64,
        limit: usize,
    ) -> Result<Vec<ContestState>> {
        if flag_mask == 0 {
            return Ok(Vec::new());
        }
        let conn = self.conn.lock();
        let mut stmt = conn.prepare(
            "SELECT proposition_id, regime, support_mass, attack_mass, \
             support_effective_independence, attack_effective_independence, \
             support_distinct_source_count, attack_distinct_source_count, \
             same_source_opposite_polarity_count, \
             same_artifact_extractor_polarity_conflict_count, \
             temporal_overlap_conflict_count, temporal_separable_opposition_count, \
             referent_schema_heterogeneity_count, heuristic_flags, \
             derivation_version, content_hash, live_claim_count, state_status, computed_at \
             FROM contest_state \
             WHERE (heuristic_flags & ?1) != 0 \
             ORDER BY computed_at DESC \
             LIMIT ?2",
        )?;
        let rows = stmt
            .query_map(params![flag_mask as i64, limit as i64], |row| {
                let flags_int: i64 = row.get(13)?;
                Ok(ContestState {
                    proposition_id: row.get(0)?,
                    regime: row.get(1)?,
                    support_mass: row.get(2)?,
                    attack_mass: row.get(3)?,
                    support_effective_independence: row.get(4)?,
                    attack_effective_independence: row.get(5)?,
                    support_distinct_source_count: row.get(6)?,
                    attack_distinct_source_count: row.get(7)?,
                    same_source_opposite_polarity_count: row.get(8)?,
                    same_artifact_extractor_polarity_conflict_count: row.get(9)?,
                    temporal_overlap_conflict_count: row.get(10)?,
                    temporal_separable_opposition_count: row.get(11)?,
                    referent_schema_heterogeneity_count: row.get(12)?,
                    heuristic_flags: flags_int as u64,
                    derivation_version: row.get(14)?,
                    content_hash: row.get(15)?,
                    live_claim_count: row.get(16)?,
                    state_status: row.get(17)?,
                    computed_at: row.get(18)?,
                })
            })?
            .collect::<std::result::Result<Vec<_>, _>>()?;
        Ok(rows)
    }

    /// RFC 008 M4.5 — produce a structured report of exemplar claim pairs
    /// that drove the contest flags for a given proposition. Recomputed
    /// on demand from the current live claim set; not persisted. Use
    /// this when an audit tool or review UI needs the actual claim IDs
    /// behind a flag like SAME_SOURCE_CONFLICT. Returns None if the
    /// proposition has no contest_state row yet.
    pub fn inspect_contest_conflicts(
        &self,
        proposition_id: &str,
        regime: &str,
    ) -> Result<Option<ContestConflictReport>> {
        let conn = self.conn.lock();
        let Some(state) = read_contest_state(&conn, proposition_id, regime)? else {
            return Ok(None);
        };
        let claims = fetch_claims_for_mobility(&conn, proposition_id, regime)?;
        drop(conn);
        let supports: Vec<&ClaimRow> = claims.iter().filter(|c| c.polarity == 1).collect();
        let attacks: Vec<&ClaimRow> = claims.iter().filter(|c| c.polarity == -1).collect();
        Ok(Some(ContestConflictReport {
            proposition_id: proposition_id.to_string(),
            regime: regime.to_string(),
            heuristic_flags: state.heuristic_flags,
            same_source_opposite_polarity_pairs: same_source_pairs(&supports, &attacks),
            same_artifact_extractor_conflict_pairs: same_artifact_pairs(&supports, &attacks),
            temporal_overlap_conflict_pairs: temporal_overlap_pairs(&supports, &attacks),
        }))
    }
}

// Pair-listing helpers for ContestConflictReport. These duplicate the
// pair-gating logic used by the counter functions; kept separate so the
// hot-path counters stay minimal and the exemplar listing is opt-in via
// `inspect_contest_conflicts`.

fn same_source_pairs(supports: &[&ClaimRow], attacks: &[&ClaimRow]) -> Vec<(String, String)> {
    let mut out = Vec::new();
    for s in supports {
        if s.source_lineage.is_empty() {
            continue;
        }
        let s_key = lineage_key(&s.source_lineage);
        for a in attacks {
            if a.source_lineage.is_empty() {
                continue;
            }
            if lineage_key(&a.source_lineage) == s_key {
                out.push((s.claim_id.clone(), a.claim_id.clone()));
            }
        }
    }
    out
}

fn same_artifact_pairs(supports: &[&ClaimRow], attacks: &[&ClaimRow]) -> Vec<(String, String)> {
    let mut out = Vec::new();
    for s in supports {
        let Some(s_rid) = &s.source_memory_rid else { continue };
        for a in attacks {
            if let Some(a_rid) = &a.source_memory_rid {
                if a_rid == s_rid && a.extractor != s.extractor {
                    out.push((s.claim_id.clone(), a.claim_id.clone()));
                }
            }
        }
    }
    out
}

fn temporal_overlap_pairs(supports: &[&ClaimRow], attacks: &[&ClaimRow]) -> Vec<(String, String)> {
    let mut out = Vec::new();
    for s in supports {
        for a in attacks {
            if !intervals_disjoint(s, a) {
                out.push((s.claim_id.clone(), a.claim_id.clone()));
            }
        }
    }
    out
}

pub(super) fn compute_contest_state_conn(
    conn: &Connection,
    proposition_id: &str,
    regime: &str,
) -> Result<ContestState> {
    let claims = fetch_claims_for_mobility(conn, proposition_id, regime)?;
    let hash = contest_content_hash(&claims);

    // Idempotence: unchanged live set + current derivation_version + fresh
    // status → no recompute needed.
    if let Some(existing) = read_contest_state(conn, proposition_id, regime)? {
        if existing.derivation_version == CONTEST_DERIVATION_VERSION
            && existing.content_hash == hash
            && existing.state_status == state_status::FRESH
        {
            return Ok(existing);
        }
    }

    // Capture pre-state flags for M10: if the conflict bits flip from off
    // to on during this recompute, we'll auto-file adversarial candidates
    // for any moves whose outputs are claims of this proposition.
    let prior_flags = read_contest_state(conn, proposition_id, regime)?
        .map(|s| s.heuristic_flags)
        .unwrap_or(0);

    let state = derive_contest_state(proposition_id, regime, &claims, hash);
    upsert_contest_state(conn, &state)?;

    // RFC 008 M10: auto-adversarial candidates on newly-set conflict flags.
    // Scan the bits that transitioned off → on; for each affected flag,
    // find moves whose outputs are among this proposition's claims and
    // file a candidate. De-duplicated: an existing candidate on the same
    // (move_id, discovered_via) combination is skipped.
    let newly_set = state.heuristic_flags & !prior_flags;
    if newly_set & 0b0010 != 0 {
        // Bit 1 = SAME_SOURCE_CONFLICT
        auto_file_adversarial_for_proposition(conn, proposition_id, "contradiction")?;
    }
    if newly_set & 0b1000 != 0 {
        // Bit 3 = SAME_ARTIFACT_EXTRACTOR_CONFLICT
        auto_file_adversarial_for_proposition(conn, proposition_id, "contradiction")?;
    }

    Ok(state)
}

/// Auto-file adversarial candidates for moves whose outputs are among a
/// flagged proposition's claims. De-duplicates on (move_id, discovered_via)
/// so repeat recomputes don't spam the candidate queue.
fn auto_file_adversarial_for_proposition(
    conn: &Connection,
    proposition_id: &str,
    discovered_via: &str,
) -> Result<()> {
    let now_ts = crate::engine::now();
    // Find moves whose output_edge points to any claim of this proposition.
    let mut stmt = conn.prepare(
        "SELECT DISTINCT moe.move_id FROM move_output_edge moe \
         INNER JOIN claims c ON moe.claim_id = c.claim_id \
         WHERE c.proposition_id = ?1",
    )?;
    let moves: Vec<String> = stmt
        .query_map(params![proposition_id], |row| row.get::<_, String>(0))?
        .collect::<std::result::Result<Vec<_>, _>>()?;
    drop(stmt);
    for move_id in moves {
        // Skip if an existing candidate for this (move_id, discovered_via)
        // already exists — we don't want repeat flags to spam the queue.
        let already_exists: bool = conn
            .query_row(
                "SELECT 1 FROM move_adversarial_instance \
                 WHERE move_id = ?1 AND discovered_via = ?2 LIMIT 1",
                params![move_id, discovered_via],
                |_| Ok(true),
            )
            .unwrap_or(false);
        if already_exists {
            continue;
        }
        let instance_id = crate::id::new_id();
        let root_cause = format!(
            "auto-generated: proposition {} contest flag transitioned for move {}",
            proposition_id, move_id
        );
        conn.execute(
            "INSERT INTO move_adversarial_instance (\
             instance_id, move_id, status, discovered_via, traced_root_cause, \
             discovered_at, created_at) \
             VALUES (?1, ?2, 'candidate', ?3, ?4, ?5, ?5)",
            params![instance_id, move_id, discovered_via, root_cause, now_ts],
        )?;
    }
    Ok(())
}

/// Pure-function derivation of ContestState from a live claim set. All
/// counters computed with leave-one-out symmetry and strict gating rules.
fn derive_contest_state(
    proposition_id: &str,
    regime: &str,
    claims: &[ClaimRow],
    content_hash: String,
) -> ContestState {
    let supports: Vec<&ClaimRow> = claims.iter().filter(|c| c.polarity == 1).collect();
    let attacks: Vec<&ClaimRow> = claims.iter().filter(|c| c.polarity == -1).collect();

    // Polarity aggregates using shared ⊕ ω_k math.
    let support_omegas = compute_omegas(&supports);
    let attack_omegas = compute_omegas(&attacks);
    let support_mass: f64 = support_omegas
        .iter()
        .enumerate()
        .map(|(k, w)| w * supports[k].weight)
        .sum();
    let attack_mass: f64 = attack_omegas
        .iter()
        .enumerate()
        .map(|(k, w)| w * attacks[k].weight)
        .sum();
    let support_effective_independence: f64 = support_omegas.iter().sum();
    let attack_effective_independence: f64 = attack_omegas.iter().sum();

    // Distinct source counts — cheap, grounded, directly interpretable.
    let support_distinct_source_count = distinct_source_count(&supports);
    let attack_distinct_source_count = distinct_source_count(&attacks);

    // Grounded contest diagnostics. Each uses strict gating.
    let same_source_opposite_polarity_count =
        count_same_source_opposite_polarity(&supports, &attacks);
    let same_artifact_extractor_polarity_conflict_count =
        count_same_artifact_extractor_conflict(&supports, &attacks);
    let (temporal_overlap_conflict_count, temporal_separable_opposition_count) =
        count_temporal_conflicts(&supports, &attacks);
    let referent_schema_heterogeneity_count = count_referent_heterogeneity(claims);

    // Heuristic flags derived from the above counters.
    let mut flags: u64 = 0;
    if support_mass > 2.0 && support_effective_independence < 2.0 {
        flags |= contest_flags::DUPLICATION_RISK;
    }
    if same_source_opposite_polarity_count > 0 {
        flags |= contest_flags::SAME_SOURCE_CONFLICT;
    }
    if referent_schema_heterogeneity_count > 0 {
        flags |= contest_flags::REFERENT_HETEROGENEITY_PRESENT;
    }
    if same_artifact_extractor_polarity_conflict_count > 0 {
        flags |= contest_flags::SAME_ARTIFACT_EXTRACTOR_CONFLICT;
    }
    if temporal_overlap_conflict_count > 0 {
        flags |= contest_flags::PRESENT_TENSE_CONFLICT;
    }

    ContestState {
        proposition_id: proposition_id.to_string(),
        regime: regime.to_string(),
        support_mass,
        attack_mass,
        support_effective_independence,
        attack_effective_independence,
        support_distinct_source_count,
        attack_distinct_source_count,
        same_source_opposite_polarity_count,
        same_artifact_extractor_polarity_conflict_count,
        temporal_overlap_conflict_count,
        temporal_separable_opposition_count,
        referent_schema_heterogeneity_count,
        heuristic_flags: flags,
        derivation_version: CONTEST_DERIVATION_VERSION,
        content_hash,
        live_claim_count: claims.len() as i64,
        state_status: state_status::FRESH.to_string(),
        computed_at: unix_seconds(),
    }
}

/// Distinct source elements across a set of claims (after normalization
/// and dedup per-claim — source_lineage is already normalized at fetch).
fn distinct_source_count(claims: &[&ClaimRow]) -> i64 {
    let mut set: HashSet<&str> = HashSet::new();
    for c in claims {
        for e in &c.source_lineage {
            set.insert(e.as_str());
        }
    }
    set.len() as i64
}

/// Count opposite-polarity pairs where both claims share IDENTICAL
/// normalized source_lineage sets. Groups supports by their lineage set
/// serialization, then joins against attacks' groups — O(n + matches).
fn count_same_source_opposite_polarity(supports: &[&ClaimRow], attacks: &[&ClaimRow]) -> i64 {
    if supports.is_empty() || attacks.is_empty() {
        return 0;
    }
    let mut support_groups: HashMap<String, usize> = HashMap::new();
    for s in supports {
        if s.source_lineage.is_empty() {
            continue; // empty lineage provides no identity for this gate
        }
        *support_groups.entry(lineage_key(&s.source_lineage)).or_insert(0) += 1;
    }
    let mut count: i64 = 0;
    for a in attacks {
        if a.source_lineage.is_empty() {
            continue;
        }
        if let Some(n_support) = support_groups.get(&lineage_key(&a.source_lineage)) {
            count += *n_support as i64;
        }
    }
    count
}

/// Count opposite-polarity pairs that derive from the same
/// source_memory_rid (same underlying document/event) via DIFFERENT
/// extractors. The same-artifact gate distinguishes extraction pathology
/// from ordinary cross-source disagreement.
fn count_same_artifact_extractor_conflict(
    supports: &[&ClaimRow],
    attacks: &[&ClaimRow],
) -> i64 {
    if supports.is_empty() || attacks.is_empty() {
        return 0;
    }
    // Index supports by source_memory_rid → vec of extractors.
    let mut by_artifact: HashMap<&str, Vec<&str>> = HashMap::new();
    for s in supports {
        if let Some(rid) = &s.source_memory_rid {
            by_artifact.entry(rid.as_str()).or_default().push(s.extractor.as_str());
        }
    }
    let mut count: i64 = 0;
    for a in attacks {
        if let Some(rid) = &a.source_memory_rid {
            if let Some(support_extractors) = by_artifact.get(rid.as_str()) {
                // Count supports on the same artifact whose extractor differs
                // from this attack's extractor.
                for se in support_extractors {
                    if *se != a.extractor {
                        count += 1;
                    }
                }
            }
        }
    }
    count
}

/// Split opposite-polarity pairs into temporal-overlap conflicts (real
/// contradictions) and temporal-separable oppositions (state changes
/// over time). Gated on both sides having validity intervals; if either
/// is fully unknown (both bounds NULL), treat as overlap-possible since
/// we cannot rule out temporal overlap.
fn count_temporal_conflicts(supports: &[&ClaimRow], attacks: &[&ClaimRow]) -> (i64, i64) {
    if supports.is_empty() || attacks.is_empty() {
        return (0, 0);
    }
    let mut overlap = 0i64;
    let mut separable = 0i64;
    for s in supports {
        for a in attacks {
            if intervals_disjoint(s, a) {
                separable += 1;
            } else {
                overlap += 1;
            }
        }
    }
    (overlap, separable)
}

/// Two claims' validity intervals are disjoint if both have at least one
/// bound AND one interval ends before the other begins. Missing bounds
/// are treated as open-ended (-∞ for valid_from, +∞ for valid_to); if
/// both claims have no bounds at all, they are NOT disjoint.
fn intervals_disjoint(a: &ClaimRow, b: &ClaimRow) -> bool {
    let a_fully_open = a.valid_from.is_none() && a.valid_to.is_none();
    let b_fully_open = b.valid_from.is_none() && b.valid_to.is_none();
    if a_fully_open && b_fully_open {
        return false;
    }
    let a_from = a.valid_from.unwrap_or(f64::NEG_INFINITY);
    let a_to = a.valid_to.unwrap_or(f64::INFINITY);
    let b_from = b.valid_from.unwrap_or(f64::NEG_INFINITY);
    let b_to = b.valid_to.unwrap_or(f64::INFINITY);
    // Disjoint iff one ends strictly before the other begins.
    a_to < b_from || b_to < a_from
}

/// Count distinct namespaces present across all live claims. Values > 1
/// indicate referent-schema heterogeneity (claims may be talking about
/// subtly different referents under different ontologies — NOT an
/// assertion of incompatibility, only observation of heterogeneity).
fn count_referent_heterogeneity(claims: &[ClaimRow]) -> i64 {
    let distinct: HashSet<&str> = claims.iter().map(|c| c.namespace.as_str()).collect();
    if distinct.len() > 1 {
        distinct.len() as i64
    } else {
        0
    }
}

fn lineage_key(lineage: &[String]) -> String {
    // lineage is already normalized (sorted, deduped) at fetch time.
    lineage.join("\x01")
}

/// Contest content_hash. Input set differs from mobility's: adds
/// source_memory_rid, namespace, valid_from/to (which contest counters
/// depend on) and excludes modality_signal/self_generated (which mobility
/// uses but contest doesn't). Separate blake3 domain tag prevents any
/// confusion between the two hashes.
fn contest_content_hash(claims: &[ClaimRow]) -> String {
    let mut hasher = blake3::Hasher::new();
    hasher.update(b"yantrikdb.contest.v");
    hasher.update(&CONTEST_DERIVATION_VERSION.to_le_bytes());
    hasher.update(b"\x00claims\x00");
    let mut indexed: Vec<(&String, &ClaimRow)> = claims.iter().map(|c| (&c.claim_id, c)).collect();
    indexed.sort_by(|a, b| a.0.cmp(b.0));
    for (cid, c) in indexed {
        hasher.update(cid.as_bytes());
        hasher.update(b"|");
        hasher.update(&c.polarity.to_le_bytes());
        hasher.update(b"|");
        hasher.update(c.extractor.as_bytes());
        hasher.update(b"|");
        hasher.update(c.namespace.as_bytes());
        hasher.update(b"|");
        hasher.update(c.source_memory_rid.as_deref().unwrap_or("").as_bytes());
        hasher.update(b"|");
        let w_scaled = (c.weight * 1000.0).round() as i64;
        hasher.update(&w_scaled.to_le_bytes());
        hasher.update(b"|");
        for src in &c.source_lineage {
            hasher.update(src.as_bytes());
            hasher.update(b",");
        }
        hasher.update(b"|");
        // valid_from/to rounded to milliseconds for stable hashing.
        let vf = c.valid_from.map(|t| (t * 1000.0).round() as i64).unwrap_or(i64::MIN);
        let vt = c.valid_to.map(|t| (t * 1000.0).round() as i64).unwrap_or(i64::MAX);
        hasher.update(&vf.to_le_bytes());
        hasher.update(&vt.to_le_bytes());
        hasher.update(b"\x00");
    }
    hex::encode(hasher.finalize().as_bytes())
}

fn read_contest_state(
    conn: &Connection,
    proposition_id: &str,
    regime: &str,
) -> Result<Option<ContestState>> {
    let mut stmt = conn.prepare(
        "SELECT proposition_id, regime, support_mass, attack_mass, \
         support_effective_independence, attack_effective_independence, \
         support_distinct_source_count, attack_distinct_source_count, \
         same_source_opposite_polarity_count, \
         same_artifact_extractor_polarity_conflict_count, \
         temporal_overlap_conflict_count, temporal_separable_opposition_count, \
         referent_schema_heterogeneity_count, heuristic_flags, \
         derivation_version, content_hash, live_claim_count, state_status, computed_at \
         FROM contest_state \
         WHERE proposition_id = ?1 AND regime = ?2",
    )?;
    let result = stmt.query_row(params![proposition_id, regime], |row| {
        let flags_int: i64 = row.get(13)?;
        Ok(ContestState {
            proposition_id: row.get(0)?,
            regime: row.get(1)?,
            support_mass: row.get(2)?,
            attack_mass: row.get(3)?,
            support_effective_independence: row.get(4)?,
            attack_effective_independence: row.get(5)?,
            support_distinct_source_count: row.get(6)?,
            attack_distinct_source_count: row.get(7)?,
            same_source_opposite_polarity_count: row.get(8)?,
            same_artifact_extractor_polarity_conflict_count: row.get(9)?,
            temporal_overlap_conflict_count: row.get(10)?,
            temporal_separable_opposition_count: row.get(11)?,
            referent_schema_heterogeneity_count: row.get(12)?,
            heuristic_flags: flags_int as u64,
            derivation_version: row.get(14)?,
            content_hash: row.get(15)?,
            live_claim_count: row.get(16)?,
            state_status: row.get(17)?,
            computed_at: row.get(18)?,
        })
    });
    match result {
        Ok(state) => Ok(Some(state)),
        Err(rusqlite::Error::QueryReturnedNoRows) => Ok(None),
        Err(e) => Err(e.into()),
    }
}

fn upsert_contest_state(conn: &Connection, s: &ContestState) -> Result<()> {
    conn.execute(
        "INSERT INTO contest_state (\
         proposition_id, regime, support_mass, attack_mass, \
         support_effective_independence, attack_effective_independence, \
         support_distinct_source_count, attack_distinct_source_count, \
         same_source_opposite_polarity_count, \
         same_artifact_extractor_polarity_conflict_count, \
         temporal_overlap_conflict_count, temporal_separable_opposition_count, \
         referent_schema_heterogeneity_count, heuristic_flags, \
         derivation_version, content_hash, live_claim_count, state_status, computed_at) \
         VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9, ?10, ?11, ?12, ?13, ?14, ?15, ?16, ?17, ?18, ?19) \
         ON CONFLICT(proposition_id, regime) DO UPDATE SET \
         support_mass = excluded.support_mass, \
         attack_mass = excluded.attack_mass, \
         support_effective_independence = excluded.support_effective_independence, \
         attack_effective_independence = excluded.attack_effective_independence, \
         support_distinct_source_count = excluded.support_distinct_source_count, \
         attack_distinct_source_count = excluded.attack_distinct_source_count, \
         same_source_opposite_polarity_count = excluded.same_source_opposite_polarity_count, \
         same_artifact_extractor_polarity_conflict_count = excluded.same_artifact_extractor_polarity_conflict_count, \
         temporal_overlap_conflict_count = excluded.temporal_overlap_conflict_count, \
         temporal_separable_opposition_count = excluded.temporal_separable_opposition_count, \
         referent_schema_heterogeneity_count = excluded.referent_schema_heterogeneity_count, \
         heuristic_flags = excluded.heuristic_flags, \
         derivation_version = excluded.derivation_version, \
         content_hash = excluded.content_hash, \
         live_claim_count = excluded.live_claim_count, \
         state_status = excluded.state_status, \
         computed_at = excluded.computed_at",
        params![
            s.proposition_id, s.regime, s.support_mass, s.attack_mass,
            s.support_effective_independence, s.attack_effective_independence,
            s.support_distinct_source_count, s.attack_distinct_source_count,
            s.same_source_opposite_polarity_count,
            s.same_artifact_extractor_polarity_conflict_count,
            s.temporal_overlap_conflict_count, s.temporal_separable_opposition_count,
            s.referent_schema_heterogeneity_count,
            s.heuristic_flags as i64,
            s.derivation_version, s.content_hash, s.live_claim_count,
            s.state_status, s.computed_at,
        ],
    )?;
    Ok(())
}

/// Row-to-struct decoder for mobility_state SELECTs.
fn row_to_mobility_state(row: &rusqlite::Row) -> rusqlite::Result<MobilityState> {
    let tier_write_json: String = row.get(16)?;
    let tier_read_json: String = row.get(17)?;
    let tier_bg_json: String = row.get(18)?;
    Ok(MobilityState {
        proposition_id: row.get(0)?,
        regime: row.get(1)?,
        snapshot_ts: row.get(2)?,
        support_mass: row.get(3)?,
        attack_mass: row.get(4)?,
        source_diversity: row.get(5)?,
        effective_independence: row.get(6)?,
        temporal_coherence: row.get(7)?,
        transportability: row.get(8)?,
        mutability: row.get(9)?,
        load_bearingness: row.get(10)?,
        modality_consilience: row.get(11)?,
        self_gen_local: row.get(12)?,
        self_gen_ancestral: row.get(13)?,
        contamination_risk: row.get(14)?,
        novelty_isolation: row.get(15)?,
        tier_write_components: serde_json::from_str(&tier_write_json).unwrap_or_default(),
        tier_read_components: serde_json::from_str(&tier_read_json).unwrap_or_default(),
        tier_bg_components: serde_json::from_str(&tier_bg_json).unwrap_or_default(),
        formula_version: row.get(19)?,
        content_hash: row.get(20)?,
        live_claim_count: row.get(21)?,
        state_status: row.get(22)?,
        computed_at: row.get(23)?,
    })
}

// ──────────────────────────────────────────────────────────────────
// Unit tests for the pure functions. Integration tests (round-trip
// through YantrikDB) live in engine/tests.rs.
// ──────────────────────────────────────────────────────────────────

#[cfg(test)]
mod tests {
    use super::*;

    fn mk_claim(
        claim_id: &str,
        extractor: &str,
        lineage: &[&str],
        self_gen: bool,
        modality: &str,
        weight: f64,
    ) -> ClaimRow {
        let mut normalized: Vec<String> = lineage.iter().map(|s| s.to_string()).collect();
        normalized.sort();
        normalized.dedup();
        ClaimRow {
            claim_id: claim_id.to_string(),
            polarity: 1,
            weight,
            extractor: extractor.to_string(),
            source_lineage: normalized,
            self_generated: self_gen,
            modality_signal: modality.to_string(),
            source_memory_rid: None,
            namespace: "default".to_string(),
            valid_from: None,
            valid_to: None,
        }
    }

    #[test]
    fn accumulate_single_claim_returns_weight() {
        let c = mk_claim("c1", "ext_a", &["src_1"], false, "text", 1.0);
        let refs = vec![&c];
        let total = accumulate_mass(&refs);
        assert!((total - 1.0).abs() < 1e-9);
    }

    #[test]
    fn accumulate_independent_claims_scales_linearly() {
        let c1 = mk_claim("c1", "ext_a", &["src_a1", "src_a2"], false, "text", 1.0);
        let c2 = mk_claim("c2", "ext_b", &["src_b1"], false, "image", 1.0);
        let c3 = mk_claim("c3", "ext_c", &["src_c1"], false, "numeric", 1.0);
        let total = accumulate_mass(&[&c1, &c2, &c3]);
        // Disjoint lineages, different extractors, no self-gen → ω = 1 each.
        assert!(
            (total - 3.0).abs() < 1e-9,
            "independent claims should sum to raw weights, got {}",
            total
        );
    }

    #[test]
    fn accumulate_duplicate_lineage_discounts() {
        let c1 = mk_claim("c1", "ext_a", &["src_shared"], false, "text", 1.0);
        let c2 = mk_claim("c2", "ext_a", &["src_shared"], false, "text", 1.0);
        let c3 = mk_claim("c3", "ext_a", &["src_shared"], false, "text", 1.0);
        let total = accumulate_mass(&[&c1, &c2, &c3]);
        // D_k = 1, P_k = 1, S_k = 0 → discount = 1.8; ω = 1/1.8; total ≈ 1.67
        assert!(total < 2.0, "duplicate lineage should discount, got {}", total);
        assert!(total > 1.5, "discount shouldn't be excessive, got {}", total);
    }

    #[test]
    fn accumulate_self_generated_is_strongly_discounted() {
        let c1 = mk_claim("c1", "self_reasoning", &["self_1"], true, "text", 1.0);
        let c2 = mk_claim("c2", "self_reasoning", &["self_1"], true, "text", 1.0);
        let total = accumulate_mass(&[&c1, &c2]);
        // discount = 1 + 0.5·1 + 0.3·1 + 0.7·1 = 2.5; ω = 0.4; total ≈ 0.8
        assert!(total < 1.0, "self-generated duplicates should collapse; got {}", total);
    }

    #[test]
    fn accumulate_is_order_invariant() {
        let c1 = mk_claim("c1", "ext_a", &["src_a"], false, "text", 1.0);
        let c2 = mk_claim("c2", "ext_b", &["src_b"], false, "image", 1.0);
        let c3 = mk_claim("c3", "ext_c", &["src_c"], false, "numeric", 1.0);
        let forward = accumulate_mass(&[&c1, &c2, &c3]);
        let reverse = accumulate_mass(&[&c3, &c2, &c1]);
        let shuffled = accumulate_mass(&[&c2, &c3, &c1]);
        assert!((forward - reverse).abs() < 1e-9);
        assert!((forward - shuffled).abs() < 1e-9);
    }

    #[test]
    fn modality_consilience_tracks_distinct_modalities() {
        let c1 = mk_claim("c1", "ext_a", &["src_1"], false, "text", 1.0);
        let c2 = mk_claim("c2", "ext_a", &["src_2"], false, "text", 1.0);
        let mono = compute_modality_consilience(&[&c1, &c2]);
        let c3 = mk_claim("c3", "ext_a", &["src_3"], false, "image", 1.0);
        let c4 = mk_claim("c4", "ext_a", &["src_4"], false, "numeric", 1.0);
        let multi = compute_modality_consilience(&[&c1, &c3, &c4]);
        assert!(multi > mono);
    }

    #[test]
    fn self_gen_local_is_correct_ratio() {
        let c1 = mk_claim("c1", "ext_a", &["src_1"], true, "text", 1.0);
        let c2 = mk_claim("c2", "ext_a", &["src_2"], false, "text", 1.0);
        let c3 = mk_claim("c3", "ext_a", &["src_3"], true, "text", 1.0);
        let r = compute_self_gen_local(&[&c1, &c2, &c3]);
        assert!((r - 2.0 / 3.0).abs() < 1e-9);
    }

    #[test]
    fn content_hash_is_order_invariant() {
        let c1 = mk_claim("c1", "ext_a", &["src_a"], false, "text", 1.0);
        let c2 = mk_claim("c2", "ext_b", &["src_b"], false, "image", 1.0);
        let c3 = mk_claim("c3", "ext_c", &["src_c"], false, "numeric", 1.0);
        let h_forward = content_hash(&[c1.clone(), c2.clone(), c3.clone()]);
        let h_reverse = content_hash(&[c3.clone(), c2.clone(), c1.clone()]);
        assert_eq!(h_forward, h_reverse);
    }

    #[test]
    fn content_hash_discriminates_on_lineage_change() {
        let c1 = mk_claim("c1", "ext_a", &["src_a"], false, "text", 1.0);
        let c2 = mk_claim("c1", "ext_a", &["src_b"], false, "text", 1.0);
        assert_ne!(content_hash(&[c1]), content_hash(&[c2]));
    }

    #[test]
    fn jaccard_empty_inputs_are_zero() {
        // Empty claim set or empty rest union → 0 (no signal, no discount).
        // This is the revised M3 convention; it means a single claim with
        // empty lineage gets ω = 1, which matches the "no one to be
        // dependent on" intuition.
        let freq: HashMap<&str, usize> = HashMap::new();
        assert_eq!(leave_one_out_jaccard(&[], &freq, 0), 0.0);
        let claim = vec!["a".to_string()];
        assert_eq!(leave_one_out_jaccard(&claim, &freq, 0), 0.0);
    }

    #[test]
    fn jaccard_disjoint_is_zero() {
        // claim = {a}, others contribute {b}. freq = {a:1, b:1}, total=2.
        // rest_union for claim = {b} (only b has freq outside claim).
        // intersection = {} (a has freq 1, not > 1).
        // union = 2. Jaccard = 0.
        let mut freq: HashMap<&str, usize> = HashMap::new();
        freq.insert("a", 1);
        freq.insert("b", 1);
        let j = leave_one_out_jaccard(&["a".to_string()], &freq, 2);
        assert_eq!(j, 0.0);
    }

    #[test]
    fn jaccard_fully_shared_is_one() {
        // claim = {a}, another claim also has {a}. freq = {a:2}, total=1.
        // intersection = 1 (a is in both); union = 1. Jaccard = 1.
        let mut freq: HashMap<&str, usize> = HashMap::new();
        freq.insert("a", 2);
        let j = leave_one_out_jaccard(&["a".to_string()], &freq, 1);
        assert_eq!(j, 1.0);
    }
}