kimetsu-brain 2.8.0

//! v0.5.2: conflict detection at ingest.
//! v2.5 Pass B (Story 1.3): automatic contradiction RESOLUTION.
//!
//! Two memories that say opposite things ("use thiserror" /
//! "use anyhow") confuse the model when both surface in the same
//! broker bundle. v0.5.0 + v0.5.1 made the brain learn from
//! outcomes; v0.5.2 prevents the brain from accumulating
//! contradictions in the first place.
//!
//! The detector runs at `add_memory` / `add_user_memory` time:
//!
//! 1. Embed the incoming text via the active embedder.
//! 2. Scan all active memories in the same scope, score cosine
//!    against the new vector.
//! 3. Pairs that exceed `DEFAULT_CONFLICT_THRESHOLD` (0.8) AND
//!    whose `normalized_text` differs from the new text get
//!    flagged as a conflict.
//! 4. (a) Auto-resolution (Story 1.3, Pass B): each conflicting pair is scored
//!    by confidence × recency (newer + higher-confidence wins).  When the
//!    score gap exceeds `NEAR_TIE_BAND` (0.15) the loser's `valid_to` is
//!    stamped to now via `mark_memory_temporal` (event-sourced, rebuild-safe,
//!    lineage preserved — NEVER deleted).  If the new memory loses, the new
//!    memory is stamped; if the existing memory loses, the existing memory is
//!    stamped.
//!    (b) Near-ties (score gap < `NEAR_TIE_BAND`): recorded in
//!    `memory_conflicts` for operator review — identical to v0.5.2 behavior.
//!    Nothing silently changes behavior on ambiguous pairs.
//!
//! Resolution gate:
//!   * `KIMETSU_RESOLVE_CONFLICTS` env or `[ingestion] resolve_conflicts`
//!     config (default true).  Disable values: `0`/`false`/`off`/`no`.
//!   * Detection must also be enabled — if `detect_conflicts` is off,
//!     resolution never runs.
//!
//! Embedder gating:
//!   * NoopEmbedder → empty result, no DB writes. Lean builds keep
//!     v0.4.x behavior.
//!   * Cross-model rows (embedding_model != active model_id) are
//!     skipped — cosine across models is meaningless. A subsequent
//!     `kimetsu brain reindex` would rehydrate them under the
//!     active model and let the next ingest catch the conflict.
//!
//! Resolution policy:
//!   Pass B: auto-resolves clear winners (|Δ| ≥ 0.15) by stamping the loser's
//!   `valid_to`; near-ties surface to the operator queue exactly as in v0.5.2.

use kimetsu_core::KimetsuResult;
use kimetsu_core::ids::new_id;
use kimetsu_core::memory::{MemoryScope, normalize_memory_text};
use rusqlite::{Connection, OptionalExtension, params};
use serde::{Deserialize, Serialize};
use time::OffsetDateTime;
use time::format_description::well_known::Rfc3339;

use crate::embeddings::{Embedder, cosine_similarity, decode_embedding};

/// v1.0: config-aware conflict-detection gate.
///
/// Resolution precedence (mirrors `user_brain_enabled_with`):
///   1. `KIMETSU_DETECT_CONFLICTS` env is set → its value wins.
///      Disable values (`0` / `false` / `off` / `no`) → false.
///      Any other non-empty value → true.
///   2. Env unset → `config_value` governs.
///   3. Default (when no config and no env) → true.
///
/// Call sites in `add_memory` and `propose_or_merge_memory` check this
/// before invoking `detect_and_record` / `find_potential_conflicts`.
pub fn conflict_detection_enabled(config_value: bool) -> bool {
    match std::env::var("KIMETSU_DETECT_CONFLICTS") {
        Ok(raw) => {
            let v = raw.trim().to_ascii_lowercase();
            if v.is_empty() {
                // Empty string — treat as unset, fall through to config.
                config_value
            } else {
                // Any explicit disable value turns it off; everything else on.
                !matches!(v.as_str(), "0" | "false" | "off" | "no")
            }
        }
        // Env unset → config governs.
        Err(_) => config_value,
    }
}

/// Default cosine-similarity threshold above which two memories
/// (with differing normalized text) are flagged as a potential
/// conflict. 0.8 is BGE-small-en-v1.5's empirical "same concept"
/// floor — tighter than 0.7 (which catches loosely related ideas)
/// and looser than 0.9 (which only fires on near-paraphrases).
pub const DEFAULT_CONFLICT_THRESHOLD: f32 = 0.8;

/// Default number of nearest existing memories to evaluate per
/// ingest. We don't need many — if more than 3 capsules
/// simultaneously cross the threshold, the deeper bug is duplicate
/// concepts in the corpus, not a conflict with this one new write.
pub const DEFAULT_TOP_K: u32 = 3;

/// Story 1.3 / Pass B: score gap below which a conflict is a near-tie and
/// goes to the operator queue instead of being auto-resolved.
///
/// The score is `confidence × recency_weight` (0-1) for each side.
/// |Δ| < 0.15 means the two memories are "roughly equal" and the
/// system should not silently pick a winner.
pub const NEAR_TIE_BAND: f32 = 0.15;

/// Story 1.3 / Pass B: config-aware conflict-resolution gate.
///
/// Resolution precedence (mirrors `conflict_detection_enabled`):
///   1. `KIMETSU_RESOLVE_CONFLICTS` env is set → its value wins.
///      Disable values (`0` / `false` / `off` / `no`) → false.
///      Any other non-empty value → true.
///   2. Env unset → `config_value` governs.
///   3. Default (when no config and no env) → true.
///
/// Resolution only runs when detection is also enabled — the caller
/// is responsible for checking `conflict_detection_enabled` first.
pub fn resolve_conflicts_enabled(config_value: bool) -> bool {
    match std::env::var("KIMETSU_RESOLVE_CONFLICTS") {
        Ok(raw) => {
            let v = raw.trim().to_ascii_lowercase();
            if v.is_empty() {
                config_value
            } else {
                !matches!(v.as_str(), "0" | "false" | "off" | "no")
            }
        }
        Err(_) => config_value,
    }
}

/// Story 1.3 / Pass B: outcome of a single conflict pair after resolution.
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum ResolutionOutcome {
    /// Auto-resolved: the new memory won; the existing memory's `valid_to`
    /// was stamped to now (it will be excluded from default retrieval).
    AutoResolvedNewWon,
    /// Auto-resolved: the existing memory won; the new memory's `valid_to`
    /// was stamped to now.
    AutoResolvedExistingWon,
    /// Near-tie (|Δ| < `NEAR_TIE_BAND`): recorded in `memory_conflicts`
    /// for operator review. Nothing was auto-stamped.
    NearTieQueued,
}

/// Story 1.3 / Pass B: compute the conflict-resolution score for a memory
/// given its `confidence` and `created_at` (RFC 3339 string).
///
/// Score = confidence × recency_weight, where recency_weight decays
/// exponentially with the age of the memory in days using a 30-day
/// half-life:
///
///   recency_weight = exp(-ln(2) / 30 × age_days)
///
/// Both confidence and recency_weight are in [0, 1], so the product is in
/// [0, 1].  A memory with confidence=1.0 created today has score ≈ 1.0;
/// one with confidence=0.5 from 90 days ago has score ≈ 0.5 × 0.125 = 0.0625.
pub fn resolution_score(confidence: f32, created_at_rfc3339: &str) -> f32 {
    let age_days = match OffsetDateTime::parse(created_at_rfc3339, &Rfc3339) {
        Ok(ts) => {
            let now = OffsetDateTime::now_utc();
            let secs = (now - ts).whole_seconds().max(0);
            secs as f64 / 86_400.0
        }
        Err(_) => 0.0, // unparseable timestamp → treat as "now" (no recency penalty)
    };
    const HALF_LIFE_DAYS: f64 = 30.0;
    let recency_weight = (-std::f64::consts::LN_2 / HALF_LIFE_DAYS * age_days).exp() as f32;
    (confidence.clamp(0.0, 1.0) * recency_weight).clamp(0.0, 1.0)
}

/// A single conflict-detection hit. Returned by
/// [`find_potential_conflicts`]; persisted by [`record_conflict`].
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct ConflictHit {
    pub existing_memory_id: String,
    pub existing_kind: String,
    pub existing_text: String,
    pub similarity: f32,
}

/// A persisted conflict row joined with both memories' text for
/// CLI / MCP display. Used by [`list_unresolved_conflicts`].
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct ConflictReport {
    pub conflict_id: String,
    pub new_memory_id: String,
    pub new_text: String,
    pub existing_memory_id: String,
    pub existing_text: String,
    pub scope: String,
    pub kind: String,
    pub similarity: f32,
    pub detected_at: String,
    pub resolved_at: Option<String>,
    pub resolution: Option<String>,
}

/// Fix 4c: ANN-based conflict detection.
///
/// Accepts the **precomputed query vector** (already embedded by the add path)
/// instead of re-embedding — halves embedding cost per add. Uses the usearch
/// ANN index to fetch a small candidate pool (≤ max(top_k * 8, 64) rows), then
/// scores only that pool with exact cosine, never full-scanning the corpus.
///
/// On non-embeddings builds (lean mode, or ANN query failure) we fall back to
/// the scope-filtered SQL scan so the function stays correct on lean builds.
///
/// `exclude_id`: the memory_id of the newly-added memory, excluded from the
/// conflict scan (a memory must not conflict with itself).
///
/// Pre-existing memories (upgraded brains) enter the usearch index on the next
/// retrieval's reconcile (see `crate::ann`), so conflict detection is
/// best-effort until then — acceptable per the v0.5.2 policy of "surface >
/// block".
pub fn find_potential_conflicts(
    conn: &Connection,
    scope: &MemoryScope,
    new_text: &str,
    embedder: &dyn Embedder,
    top_k: u32,
    threshold: f32,
) -> KimetsuResult<Vec<ConflictHit>> {
    find_potential_conflicts_with_vec(
        conn, scope, new_text, None, embedder, None, top_k, threshold,
    )
}

/// Internal: full signature used by `detect_and_record` when a precomputed
/// embedding is available (avoids re-embedding at conflict-scan time).
///
/// - `precomputed_vec`: the embedding produced by `embed_and_persist` for the
///   new memory.  When `None`, we embed `new_text` here (original behavior).
/// - `exclude_id`: the new memory's own id, excluded so a memory is never
///   flagged as conflicting with itself.
#[allow(clippy::too_many_arguments)]
pub(crate) fn find_potential_conflicts_with_vec(
    conn: &Connection,
    scope: &MemoryScope,
    new_text: &str,
    precomputed_vec: Option<&[f32]>,
    embedder: &dyn Embedder,
    exclude_id: Option<&str>,
    top_k: u32,
    threshold: f32,
) -> KimetsuResult<Vec<ConflictHit>> {
    if embedder.is_noop() {
        return Ok(Vec::new());
    }

    // Use the precomputed vector when available, else embed now.
    let new_vec: Vec<f32>;
    let query_vec: &[f32] = if let Some(v) = precomputed_vec {
        v
    } else {
        new_vec = embedder
            .embed(new_text)
            .map_err(|e| format!("embedder failed during conflict scan: {e}"))?;
        if new_vec.len() != embedder.dim() {
            return Err(format!(
                "embedder {} returned {} dims, expected {}",
                embedder.model_id(),
                new_vec.len(),
                embedder.dim()
            )
            .into());
        }
        &new_vec
    };

    let new_normalized = normalize_memory_text(new_text);
    let scope_label = scope.to_string();
    let active_model = embedder.model_id();
    // Pool size for ANN candidate fetch: at least 64, at least top_k * 8.
    // Only used on embeddings builds; suppress the lint on lean builds.
    #[cfg_attr(not(feature = "embeddings"), allow(unused_variables))]
    let pool_size = (top_k * 8).max(64) as i64;

    // Fix 4c: ANN path — query the usearch index for a small candidate pool.
    // Only available on embeddings builds (the ANN index is lean-build absent).
    #[cfg(feature = "embeddings")]
    {
        let handle = crate::ann::handle_for_query(conn, query_vec.len(), active_model)?;
        let ann_rowids: Vec<i64> = handle
            .read()
            .unwrap_or_else(|p| p.into_inner())
            .search(query_vec, pool_size as usize)?
            .into_iter()
            .map(|(rowid, _)| rowid)
            .collect();

        if !ann_rowids.is_empty() {
            // Fetch full rows for the ANN pool.
            let placeholders: String = ann_rowids
                .iter()
                .enumerate()
                .map(|(i, _)| format!("?{}", i + 1))
                .collect::<Vec<_>>()
                .join(", ");
            let sql = format!(
                "SELECT memory_id, kind, text, normalized_text, embedding, embedding_model
                 FROM   memories
                 WHERE  invalidated_at IS NULL
                   AND  scope = '{scope_label}'
                   AND  embedding_model = '{active_model}'
                   AND  rowid IN ({placeholders})"
            );
            let mut stmt = conn.prepare(&sql)?;
            let params_vec: Vec<&dyn rusqlite::ToSql> = ann_rowids
                .iter()
                .map(|n| n as &dyn rusqlite::ToSql)
                .collect();
            let rows_iter = stmt.query_map(params_vec.as_slice(), |row| {
                Ok((
                    row.get::<_, String>(0)?,
                    row.get::<_, String>(1)?,
                    row.get::<_, String>(2)?,
                    row.get::<_, String>(3)?,
                    row.get::<_, Vec<u8>>(4)?,
                ))
            })?;

            let mut hits: Vec<ConflictHit> = Vec::new();
            for row in rows_iter {
                let (existing_id, kind, text, normalized, bytes) = row?;
                // Skip: same normalized text (dedup, not conflict).
                if normalized == new_normalized {
                    continue;
                }
                // Skip: the new memory itself.
                if let Some(excl) = exclude_id {
                    if existing_id == excl {
                        continue;
                    }
                }
                let Ok(existing_vec) = decode_embedding(&bytes, Some(query_vec.len())) else {
                    continue;
                };
                let sim = cosine_similarity(query_vec, &existing_vec);
                if sim >= threshold {
                    hits.push(ConflictHit {
                        existing_memory_id: existing_id,
                        existing_kind: kind,
                        existing_text: text,
                        similarity: sim,
                    });
                }
            }

            hits.sort_by(|a, b| {
                b.similarity
                    .partial_cmp(&a.similarity)
                    .unwrap_or(std::cmp::Ordering::Equal)
            });
            hits.truncate(top_k as usize);
            return Ok(hits);
        }
    }

    // Lean / fallback: full scope-filtered SQL scan (original O(N) path).
    // Used on lean builds and when the ANN index is unavailable or its pool is
    // empty (e.g. a fresh upgraded brain not yet reconciled).
    find_potential_conflicts_sql(
        conn,
        &scope_label,
        &new_normalized,
        query_vec,
        active_model,
        exclude_id,
        top_k,
        threshold,
    )
}

/// Scope-filtered SQL scan — O(N) fallback used on lean builds and when the
/// ANN index is unavailable. This is the original `find_potential_conflicts`
/// body.
#[allow(clippy::too_many_arguments)]
fn find_potential_conflicts_sql(
    conn: &Connection,
    scope_label: &str,
    new_normalized: &str,
    query_vec: &[f32],
    active_model: &str,
    exclude_id: Option<&str>,
    top_k: u32,
    threshold: f32,
) -> KimetsuResult<Vec<ConflictHit>> {
    let mut stmt = conn.prepare(
        "
        SELECT memory_id, kind, text, normalized_text, embedding
        FROM memories
        WHERE scope = ?1
          AND invalidated_at IS NULL
          AND embedding IS NOT NULL
          AND embedding_model = ?2
        ",
    )?;
    let rows = stmt.query_map(params![scope_label, active_model], |row| {
        Ok((
            row.get::<_, String>(0)?,
            row.get::<_, String>(1)?,
            row.get::<_, String>(2)?,
            row.get::<_, String>(3)?,
            row.get::<_, Vec<u8>>(4)?,
        ))
    })?;

    let mut hits: Vec<ConflictHit> = Vec::new();
    for row in rows {
        let (existing_id, kind, text, normalized, bytes) = row?;
        if normalized == new_normalized {
            continue;
        }
        if let Some(excl) = exclude_id {
            if existing_id == excl {
                continue;
            }
        }
        let Ok(existing_vec) = decode_embedding(&bytes, Some(query_vec.len())) else {
            continue;
        };
        let sim = cosine_similarity(query_vec, &existing_vec);
        if sim >= threshold {
            hits.push(ConflictHit {
                existing_memory_id: existing_id,
                existing_kind: kind,
                existing_text: text,
                similarity: sim,
            });
        }
    }

    hits.sort_by(|a, b| {
        b.similarity
            .partial_cmp(&a.similarity)
            .unwrap_or(std::cmp::Ordering::Equal)
    });
    hits.truncate(top_k as usize);
    Ok(hits)
}

/// Persist a single conflict pair. Idempotent on
/// (new_memory_id, existing_memory_id) via UNIQUE — a re-scan of
/// the same ingest won't double-write rows. Returns the
/// conflict_id (freshly minted or existing) so the caller can
/// chain follow-ups.
pub fn record_conflict(
    conn: &Connection,
    new_memory_id: &str,
    scope: &MemoryScope,
    kind: &str,
    hit: &ConflictHit,
) -> KimetsuResult<String> {
    // If a row for this pair already exists, return its id.
    let existing: Option<String> = conn
        .query_row(
            "
            SELECT conflict_id
            FROM memory_conflicts
            WHERE new_memory_id = ?1 AND existing_memory_id = ?2
            ",
            params![new_memory_id, hit.existing_memory_id],
            |row| row.get::<_, String>(0),
        )
        .optional()?;
    if let Some(id) = existing {
        return Ok(id);
    }
    let conflict_id = new_id().to_string();
    let detected_at = OffsetDateTime::now_utc()
        .format(&time::format_description::well_known::Rfc3339)
        .map_err(|e| format!("timestamp format: {e}"))?;
    conn.execute(
        "
        INSERT INTO memory_conflicts (
            conflict_id, new_memory_id, existing_memory_id,
            scope, kind, similarity, detected_at
        )
        VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7)
        ",
        params![
            conflict_id,
            new_memory_id,
            hit.existing_memory_id,
            scope.to_string(),
            kind,
            hit.similarity as f64,
            detected_at,
        ],
    )?;
    Ok(conflict_id)
}

/// Convenience wrapper used by `add_memory` / `add_user_memory`:
/// run detection, persist each hit, return the number of recorded
/// conflicts so the caller can decide whether to surface a
/// warning to stderr.
///
/// `precomputed_vec`: when the caller already embedded `text` (e.g.
/// `embed_and_persist` just ran), pass that vector here to skip re-embedding.
/// Pass `None` to let the scan embed on demand (original behavior).
///
/// Best-effort: an error inside the scan is downgraded to "no
/// conflicts detected this round" + a stderr line, because we
/// never want conflict detection to fail an otherwise-valid memory
/// write.
pub fn detect_and_record(
    conn: &Connection,
    new_memory_id: &str,
    scope: &MemoryScope,
    kind: &str,
    text: &str,
    embedder: &dyn Embedder,
) -> usize {
    detect_and_record_with_vec(conn, new_memory_id, scope, kind, text, None, embedder)
}

/// Internal: full variant used by paths that have a precomputed embedding.
pub(crate) fn detect_and_record_with_vec(
    conn: &Connection,
    new_memory_id: &str,
    scope: &MemoryScope,
    kind: &str,
    text: &str,
    precomputed_vec: Option<&[f32]>,
    embedder: &dyn Embedder,
) -> usize {
    let hits = match find_potential_conflicts_with_vec(
        conn,
        scope,
        text,
        precomputed_vec,
        embedder,
        Some(new_memory_id),
        DEFAULT_TOP_K,
        DEFAULT_CONFLICT_THRESHOLD,
    ) {
        Ok(h) => h,
        Err(e) => {
            eprintln!("kimetsu-brain: conflict scan skipped: {e}");
            return 0;
        }
    };
    let mut recorded = 0usize;
    for hit in &hits {
        match record_conflict(conn, new_memory_id, scope, kind, hit) {
            Ok(_) => recorded += 1,
            Err(e) => {
                eprintln!(
                    "kimetsu-brain: failed to record conflict {} <-> {}: {e}",
                    new_memory_id, hit.existing_memory_id
                );
            }
        }
    }
    recorded
}

/// Queue similarity candidates for explicit review. Similarity and a score gap
/// cannot establish a contradiction or which claim is correct. The legacy
/// confidence/time arguments and `(auto_resolved, queued)` return shape remain
/// compatible, but auto_resolved is always zero. Recording is best-effort.
#[allow(clippy::too_many_arguments)]
pub(crate) fn detect_record_and_resolve_with_vec(
    conn: &Connection,
    new_memory_id: &str,
    scope: &MemoryScope,
    kind: &str,
    text: &str,
    precomputed_vec: Option<&[f32]>,
    embedder: &dyn Embedder,
    new_confidence: f32,
    new_created_at: &str,
) -> (usize, usize) {
    let hits = match find_potential_conflicts_with_vec(
        conn,
        scope,
        text,
        precomputed_vec,
        embedder,
        Some(new_memory_id),
        DEFAULT_TOP_K,
        DEFAULT_CONFLICT_THRESHOLD,
    ) {
        Ok(h) => h,
        Err(e) => {
            eprintln!("kimetsu-brain: conflict scan skipped: {e}");
            return (0, 0);
        }
    };

    // Cosine and confidence/age gaps cannot establish a contradiction. Preserve
    // the legacy entry point for callers/config compatibility, but leave destructive
    // resolution to explicit corrections or operator decisions until structured
    // claim identity and independent contradiction evidence are available.
    let _ = (new_confidence, new_created_at);
    let mut queued = 0;
    for hit in &hits {
        match record_conflict(conn, new_memory_id, scope, kind, hit) {
            Ok(_) => queued += 1,
            Err(e) => eprintln!("kimetsu-brain: could not queue related claims: {e}"),
        }
    }
    (0, queued)
}

/// List open (unresolved) conflicts ordered by most recent first,
/// joined with both memories' text so the CLI can render rich
/// rows without a second query round-trip. `limit` is applied
/// after sorting; pass a generous default at the call site
/// (e.g. 50) since conflicts are sparse by construction.
pub fn list_unresolved_conflicts(
    conn: &Connection,
    limit: u32,
) -> KimetsuResult<Vec<ConflictReport>> {
    let mut stmt = conn.prepare(
        "
        SELECT c.conflict_id, c.new_memory_id, mn.text, c.existing_memory_id,
               me.text, c.scope, c.kind, c.similarity, c.detected_at,
               c.resolved_at, c.resolution
        FROM memory_conflicts c
        LEFT JOIN memories mn ON mn.memory_id = c.new_memory_id
        LEFT JOIN memories me ON me.memory_id = c.existing_memory_id
        WHERE c.resolved_at IS NULL
        ORDER BY c.detected_at DESC
        LIMIT ?1
        ",
    )?;
    let rows = stmt.query_map(params![limit], |row| {
        Ok(ConflictReport {
            conflict_id: row.get(0)?,
            new_memory_id: row.get(1)?,
            new_text: row.get::<_, Option<String>>(2)?.unwrap_or_default(),
            existing_memory_id: row.get(3)?,
            existing_text: row.get::<_, Option<String>>(4)?.unwrap_or_default(),
            scope: row.get(5)?,
            kind: row.get(6)?,
            similarity: row.get::<_, f64>(7)? as f32,
            detected_at: row.get(8)?,
            resolved_at: row.get(9)?,
            resolution: row.get(10)?,
        })
    })?;
    let mut out = Vec::new();
    for row in rows {
        out.push(row?);
    }
    Ok(out)
}

/// Mark a conflict as resolved with one of `'kept_new'`,
/// `'kept_existing'`, or `'kept_both'`. Returns true if a row was
/// updated (i.e. the id exists and was previously unresolved).
///
/// Side effect: when `resolution = 'kept_new'` the existing
/// memory is invalidated (resolution "I chose the new write");
/// `'kept_existing'` invalidates the new memory; `'kept_both'`
/// invalidates neither. Either invalidation is idempotent —
/// re-applying the same resolution is a no-op on the memory rows.
pub fn resolve_conflict(
    conn: &Connection,
    conflict_id: &str,
    resolution: &str,
) -> KimetsuResult<bool> {
    let resolution = resolution.trim();
    if !matches!(resolution, "kept_new" | "kept_existing" | "kept_both") {
        return Err(format!(
            "invalid conflict resolution {resolution:?}; expected kept_new | kept_existing | kept_both"
        )
        .into());
    }
    let mut changed = false;
    crate::projector::with_write_txn(conn, |conn| {
        let metadata: Option<(String,String,String,String,f64,String)> = conn.query_row(
            "SELECT new_memory_id,existing_memory_id,scope,kind,similarity,detected_at FROM memory_conflicts WHERE conflict_id=?1 AND resolved_at IS NULL",
            [conflict_id], |r|Ok((r.get(0)?,r.get(1)?,r.get(2)?,r.get(3)?,r.get(4)?,r.get(5)?))).optional()?;
        let Some((new_id, existing_id, scope, kind, similarity, detected_at)) = metadata else {
            return Ok(());
        };
        let event = kimetsu_core::event::Event::new(
            kimetsu_core::ids::RunId::new(),
            "conflict.resolved",
            serde_json::json!({
                "conflict_id":conflict_id,"new_memory_id":new_id,"existing_memory_id":existing_id,
                "scope":scope,"kind":kind,"similarity":similarity,"detected_at":detected_at,"resolution":resolution
            }),
        );
        crate::projector::apply_event(conn, &event)?;
        changed = true;
        Ok(())
    })?;
    Ok(changed)
}

/// Self-contained pair metadata makes explicit decisions replayable even when
/// the original similarity detection was a derived-only row.
pub(crate) fn project_resolution(
    conn: &Connection,
    event: &kimetsu_core::event::Event,
) -> KimetsuResult<()> {
    let field = |key| {
        event
            .payload
            .get(key)
            .and_then(|v| v.as_str())
            .ok_or_else(|| format!("conflict.resolved missing {key}"))
    };
    let id = field("conflict_id")?;
    let new_id = field("new_memory_id")?;
    let existing_id = field("existing_memory_id")?;
    let resolution = field("resolution")?;
    if new_id == existing_id || !matches!(resolution, "kept_new" | "kept_existing" | "kept_both") {
        return Err("invalid conflict pair or resolution".into());
    }
    let pair: Option<(String, String)> = conn
        .query_row(
            "SELECT new_memory_id,existing_memory_id FROM memory_conflicts WHERE conflict_id=?1",
            [id],
            |r| Ok((r.get(0)?, r.get(1)?)),
        )
        .optional()?;
    if pair.is_some_and(|(a, b)| a != new_id || b != existing_id) {
        return Err("conflict pair mismatch".into());
    }
    let ts = event
        .ts
        .format(&time::format_description::well_known::Rfc3339)?;
    conn.execute("INSERT OR IGNORE INTO memory_conflicts(conflict_id,new_memory_id,existing_memory_id,scope,kind,similarity,detected_at) VALUES(?1,?2,?3,?4,?5,?6,?7)",
        params![id,new_id,existing_id,field("scope")?,field("kind")?,event.payload["similarity"].as_f64().unwrap_or(0.0),field("detected_at")?])?;
    let loser = match resolution {
        "kept_new" => Some(existing_id),
        "kept_existing" => Some(new_id),
        _ => None,
    };
    if let Some(loser) = loser {
        // Explicit rejection ends archival eligibility. Keeping a previous
        // `forgotten` reason would let restore resurrect the rejected claim.
        conn.execute("UPDATE memories SET invalidated_at=COALESCE(invalidated_at,?2),invalidated_reason=CASE WHEN invalidated_reason IS NULL OR invalidated_reason IN ('forgotten','forgotten/archived','forgotten_archived') THEN ?3 ELSE invalidated_reason END WHERE memory_id=?1",params![loser,ts,format!("conflict {id} resolved as {resolution}")])?;
        conn.execute("DELETE FROM memories_fts WHERE memory_id=?1", [loser])?;
        #[cfg(feature = "embeddings")]
        crate::ann::on_invalidate(conn, loser);
    }
    conn.execute(
        "UPDATE memory_conflicts SET resolved_at=?2,resolution=?3 WHERE conflict_id=?1",
        params![id, ts, resolution],
    )?;
    Ok(())
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::embeddings::{NoopEmbedder, StubEmbedder, encode_embedding};
    use kimetsu_core::memory::normalize_memory_text;
    use rusqlite::Connection;

    fn open_test_brain() -> Connection {
        let conn = Connection::open_in_memory().expect("open in-memory");
        crate::schema::initialize(&conn).expect("init schema");
        conn
    }

    fn insert_memory(
        conn: &Connection,
        memory_id: &str,
        scope: &str,
        kind: &str,
        text: &str,
        embedder: &dyn Embedder,
    ) {
        let normalized = normalize_memory_text(text);
        let vec = embedder.embed(text).expect("embed test row");
        let blob = encode_embedding(&vec);
        conn.execute(
            "
            INSERT INTO memories (
                memory_id, scope, kind, text, normalized_text, confidence,
                source_event_id, provenance_snapshot_json, created_at,
                use_count, usefulness_score, embedding, embedding_model
            )
            VALUES (?1, ?2, ?3, ?4, ?5, 1.0, NULL, '{}',
                    '2026-01-01T00:00:00Z', 0, 0.0, ?6, ?7)
            ",
            params![
                memory_id,
                scope,
                kind,
                text,
                normalized,
                blob,
                embedder.model_id(),
            ],
        )
        .expect("insert");
        conn.execute(
            "INSERT INTO memories_fts (memory_id, text, kind, scope)
             VALUES (?1, ?2, ?3, ?4)",
            params![memory_id, text, kind, scope],
        )
        .expect("fts");
    }

    /// v0.5.2: NoopEmbedder MUST short-circuit to zero hits. Lean
    /// builds without --features embeddings keep v0.4.x behavior.
    #[test]
    fn noop_embedder_returns_no_conflicts() {
        let conn = open_test_brain();
        // Insert via stub so the row has an embedding; then scan with Noop.
        let stub = StubEmbedder::new();
        insert_memory(
            &conn,
            "m_existing",
            "global_user",
            "fact",
            "use thiserror for libraries",
            &stub,
        );
        let hits = find_potential_conflicts(
            &conn,
            &MemoryScope::GlobalUser,
            "use anyhow for libraries",
            &NoopEmbedder,
            DEFAULT_TOP_K,
            DEFAULT_CONFLICT_THRESHOLD,
        )
        .expect("scan");
        assert!(hits.is_empty(), "noop embedder should produce no hits");
    }

    /// v0.5.2: cross-model rows are skipped (cosine across models is
    /// meaningless). Critical for safety mid-reindex when some rows
    /// carry the old model id.
    #[test]
    fn cross_model_rows_are_skipped() {
        let conn = open_test_brain();
        let stub = StubEmbedder::new();
        insert_memory(
            &conn,
            "m_xmodel",
            "global_user",
            "fact",
            "use thiserror",
            &stub,
        );
        // Stomp the model id to simulate a pre-reindex row.
        conn.execute(
            "UPDATE memories SET embedding_model = 'bge-small-en-v1.5' WHERE memory_id = 'm_xmodel'",
            [],
        )
        .expect("force mismatch");
        let hits = find_potential_conflicts(
            &conn,
            &MemoryScope::GlobalUser,
            "use thiserror everywhere", // very similar text
            &stub,
            DEFAULT_TOP_K,
            // Threshold low enough that the StubEmbedder would normally hit it.
            0.0,
        )
        .expect("scan");
        assert!(
            hits.is_empty(),
            "cross-model rows must be skipped from conflict scan"
        );
    }

    /// v0.5.2: identical normalized text is dedup territory, not a
    /// conflict. The scanner must filter exact matches out so a
    /// re-add doesn't generate a self-conflict.
    #[test]
    fn exact_match_is_not_flagged_as_conflict() {
        let conn = open_test_brain();
        let stub = StubEmbedder::new();
        insert_memory(
            &conn,
            "m_exact",
            "global_user",
            "fact",
            "Use ripgrep",
            &stub,
        );
        let hits = find_potential_conflicts(
            &conn,
            &MemoryScope::GlobalUser,
            // Same after normalization.
            "use ripgrep",
            &stub,
            DEFAULT_TOP_K,
            0.0, // even at zero threshold, exact-text should be filtered
        )
        .expect("scan");
        assert!(
            hits.is_empty(),
            "exact normalized-text match should be dedup, not conflict"
        );
    }

    /// v0.5.2: a memory with text similar (high cosine) but
    /// different (post-normalization) gets flagged. Uses
    /// StubEmbedder where identical-token-bag inputs cosine to 1.0
    /// — we exploit that to construct a "shared concept, different
    /// wording" pair.
    #[test]
    fn similar_but_different_text_is_flagged() {
        let conn = open_test_brain();
        let stub = StubEmbedder::new();
        // StubEmbedder cosine is driven by tokenized hash buckets.
        // Two strings sharing 3 distinctive tokens out of 4 will
        // score very high cosine while normalizing differently.
        insert_memory(
            &conn,
            "m_existing",
            "global_user",
            "fact",
            "alpha beta gamma delta",
            &stub,
        );
        let hits = find_potential_conflicts(
            &conn,
            &MemoryScope::GlobalUser,
            "alpha beta gamma omega", // 3/4 shared tokens → high cosine
            &stub,
            DEFAULT_TOP_K,
            // Use a permissive threshold; the StubEmbedder cosine is
            // architecture-dependent so we want the test to fire on
            // the substantive overlap, not the exact 0.8.
            0.4,
        )
        .expect("scan");
        assert!(
            !hits.is_empty(),
            "high-cosine + different-normalized text should flag a conflict"
        );
        assert_eq!(hits[0].existing_memory_id, "m_existing");
        assert!(
            hits[0].similarity >= 0.4,
            "similarity should be >= threshold; got {}",
            hits[0].similarity
        );
    }

    /// v0.5.2: record_conflict is idempotent on
    /// (new_memory_id, existing_memory_id) — re-recording the same
    /// pair returns the original conflict_id instead of duplicating.
    #[test]
    fn record_conflict_is_idempotent() {
        let conn = open_test_brain();
        // Seed two memories so the FK-style assumption (memory rows
        // exist) holds for any downstream join.
        let stub = StubEmbedder::new();
        insert_memory(&conn, "m_new", "global_user", "fact", "alpha", &stub);
        insert_memory(&conn, "m_old", "global_user", "fact", "beta", &stub);
        let hit = ConflictHit {
            existing_memory_id: "m_old".to_string(),
            existing_kind: "fact".to_string(),
            existing_text: "beta".to_string(),
            similarity: 0.85,
        };
        let id1 = record_conflict(&conn, "m_new", &MemoryScope::GlobalUser, "fact", &hit)
            .expect("record 1");
        let id2 = record_conflict(&conn, "m_new", &MemoryScope::GlobalUser, "fact", &hit)
            .expect("record 2");
        assert_eq!(id1, id2, "re-recording the same pair must return same id");
        // Confirm only one row landed.
        let count: i64 = conn
            .query_row("SELECT COUNT(*) FROM memory_conflicts", [], |row| {
                row.get(0)
            })
            .unwrap();
        assert_eq!(count, 1);
    }

    /// v0.5.2: list_unresolved_conflicts joins memory text and
    /// returns rows ordered by detected_at DESC. Resolved rows are
    /// excluded.
    #[test]
    fn list_unresolved_excludes_resolved_rows() {
        let conn = open_test_brain();
        let stub = StubEmbedder::new();
        insert_memory(
            &conn,
            "m_new1",
            "global_user",
            "fact",
            "use thiserror",
            &stub,
        );
        insert_memory(&conn, "m_old1", "global_user", "fact", "use anyhow", &stub);
        insert_memory(
            &conn,
            "m_new2",
            "global_user",
            "fact",
            "tabs over spaces",
            &stub,
        );
        insert_memory(
            &conn,
            "m_old2",
            "global_user",
            "fact",
            "spaces over tabs",
            &stub,
        );

        let hit1 = ConflictHit {
            existing_memory_id: "m_old1".to_string(),
            existing_kind: "fact".to_string(),
            existing_text: "use anyhow".to_string(),
            similarity: 0.9,
        };
        let hit2 = ConflictHit {
            existing_memory_id: "m_old2".to_string(),
            existing_kind: "fact".to_string(),
            existing_text: "spaces over tabs".to_string(),
            similarity: 0.85,
        };
        let cid1 =
            record_conflict(&conn, "m_new1", &MemoryScope::GlobalUser, "fact", &hit1).unwrap();
        let _cid2 =
            record_conflict(&conn, "m_new2", &MemoryScope::GlobalUser, "fact", &hit2).unwrap();

        // Resolve the first conflict (kept_both — neither
        // invalidated); both should still be visible only via the
        // second listing.
        assert!(resolve_conflict(&conn, &cid1, "kept_both").unwrap());

        let open = list_unresolved_conflicts(&conn, 50).unwrap();
        assert_eq!(open.len(), 1, "only the unresolved conflict should list");
        assert_eq!(open[0].new_memory_id, "m_new2");
        assert_eq!(open[0].existing_memory_id, "m_old2");
        assert_eq!(open[0].new_text, "tabs over spaces");
        assert_eq!(open[0].existing_text, "spaces over tabs");
    }

    /// v0.5.2: resolve_conflict with `kept_new` invalidates the
    /// existing memory; `kept_existing` invalidates the new one;
    /// `kept_both` leaves both active.
    #[test]
    fn resolve_conflict_invalidates_loser_side() {
        let conn = open_test_brain();
        let stub = StubEmbedder::new();
        for (mid, text) in [
            ("m_keep_new", "alpha"),
            ("m_old_loses", "beta"),
            ("m_new_loses", "gamma"),
            ("m_keep_existing", "delta"),
            ("m_both_a", "epsilon"),
            ("m_both_b", "zeta"),
        ] {
            insert_memory(&conn, mid, "global_user", "fact", text, &stub);
        }
        let mk_hit = |old: &str| ConflictHit {
            existing_memory_id: old.to_string(),
            existing_kind: "fact".to_string(),
            existing_text: "x".to_string(),
            similarity: 0.9,
        };

        let c_kept_new = record_conflict(
            &conn,
            "m_keep_new",
            &MemoryScope::GlobalUser,
            "fact",
            &mk_hit("m_old_loses"),
        )
        .unwrap();
        let c_kept_existing = record_conflict(
            &conn,
            "m_new_loses",
            &MemoryScope::GlobalUser,
            "fact",
            &mk_hit("m_keep_existing"),
        )
        .unwrap();
        let c_both = record_conflict(
            &conn,
            "m_both_a",
            &MemoryScope::GlobalUser,
            "fact",
            &mk_hit("m_both_b"),
        )
        .unwrap();

        assert!(resolve_conflict(&conn, &c_kept_new, "kept_new").unwrap());
        assert!(resolve_conflict(&conn, &c_kept_existing, "kept_existing").unwrap());
        assert!(resolve_conflict(&conn, &c_both, "kept_both").unwrap());

        let invalidated_at: Vec<(String, Option<String>)> = {
            let mut stmt = conn
                .prepare("SELECT memory_id, invalidated_at FROM memories ORDER BY memory_id")
                .unwrap();
            stmt.query_map([], |row| {
                Ok((row.get::<_, String>(0)?, row.get::<_, Option<String>>(1)?))
            })
            .unwrap()
            .map(|r| r.unwrap())
            .collect()
        };

        let map: std::collections::HashMap<_, _> = invalidated_at.into_iter().collect();
        // kept_new → existing invalidated
        assert!(map["m_keep_new"].is_none(), "winner should stay active");
        assert!(
            map["m_old_loses"].is_some(),
            "kept_new must invalidate the existing memory"
        );
        // kept_existing → new invalidated
        assert!(
            map["m_keep_existing"].is_none(),
            "winner (existing) should stay active"
        );
        assert!(
            map["m_new_loses"].is_some(),
            "kept_existing must invalidate the new memory"
        );
        // kept_both → neither invalidated
        assert!(
            map["m_both_a"].is_none() && map["m_both_b"].is_none(),
            "kept_both should leave both memories active"
        );
    }

    /// v0.5.2: re-resolving the same conflict is a no-op (returns
    /// false on the second call) and does NOT re-stamp
    /// `invalidated_at`. Critical so an operator can't accidentally
    /// rewrite history by re-running `resolve`.
    #[test]
    fn resolve_conflict_is_idempotent() {
        let conn = open_test_brain();
        let stub = StubEmbedder::new();
        insert_memory(&conn, "m_new", "global_user", "fact", "x", &stub);
        insert_memory(&conn, "m_old", "global_user", "fact", "y", &stub);
        let hit = ConflictHit {
            existing_memory_id: "m_old".to_string(),
            existing_kind: "fact".to_string(),
            existing_text: "y".to_string(),
            similarity: 0.95,
        };
        let cid = record_conflict(&conn, "m_new", &MemoryScope::GlobalUser, "fact", &hit).unwrap();
        assert!(resolve_conflict(&conn, &cid, "kept_new").unwrap());
        assert!(
            !resolve_conflict(&conn, &cid, "kept_existing").unwrap(),
            "second resolve must return false (already resolved)"
        );
    }

    /// v0.5.2: detect_and_record returns 0 + writes nothing under
    /// NoopEmbedder. End-to-end version of the noop-skip rule.
    #[test]
    fn detect_and_record_noop_writes_nothing() {
        let conn = open_test_brain();
        let stub = StubEmbedder::new();
        insert_memory(
            &conn,
            "m_existing",
            "global_user",
            "fact",
            "alpha beta",
            &stub,
        );
        insert_memory(&conn, "m_new", "global_user", "fact", "alpha gamma", &stub);
        let recorded = detect_and_record(
            &conn,
            "m_new",
            &MemoryScope::GlobalUser,
            "fact",
            "alpha gamma",
            &NoopEmbedder,
        );
        assert_eq!(recorded, 0);
        let count: i64 = conn
            .query_row("SELECT COUNT(*) FROM memory_conflicts", [], |row| {
                row.get(0)
            })
            .unwrap();
        assert_eq!(count, 0);
    }

    /// v0.5.2: invalid resolution strings are rejected before any
    /// DB write happens. Belt-and-suspenders so a typo from the CLI
    /// doesn't silently mark a conflict as "resolved" with garbage.
    #[test]
    fn resolve_conflict_rejects_invalid_resolution_strings() {
        let conn = open_test_brain();
        let err = resolve_conflict(&conn, "ignored", "delete_them_all").unwrap_err();
        let msg = format!("{err}");
        assert!(msg.contains("invalid conflict resolution"), "got: {msg}");
    }

    // ------------------------------------------------------------------
    // Fix 2: conflict_detection_enabled off-switch
    // ------------------------------------------------------------------

    /// Fix 2: conflict_detection_enabled returns false when env is set to a
    /// disable value. Tests the env > config precedence.
    #[test]
    fn conflict_detection_enabled_env_disable_overrides_config_true() {
        let lock = crate::user_brain::test_env_lock()
            .lock()
            .unwrap_or_else(|p| p.into_inner());
        let prev = std::env::var("KIMETSU_DETECT_CONFLICTS").ok();
        for v in ["0", "false", "off", "no"] {
            unsafe {
                std::env::set_var("KIMETSU_DETECT_CONFLICTS", v);
            }
            assert!(
                !conflict_detection_enabled(true),
                "env={v:?} must disable even when config=true"
            );
        }
        // Restore.
        unsafe {
            match prev {
                Some(v) => std::env::set_var("KIMETSU_DETECT_CONFLICTS", v),
                None => std::env::remove_var("KIMETSU_DETECT_CONFLICTS"),
            }
        }
        drop(lock);
    }

    /// Fix 2: conflict_detection_enabled respects config=false when env is unset.
    #[test]
    fn conflict_detection_enabled_config_false_when_env_unset() {
        let lock = crate::user_brain::test_env_lock()
            .lock()
            .unwrap_or_else(|p| p.into_inner());
        let prev = std::env::var("KIMETSU_DETECT_CONFLICTS").ok();
        unsafe {
            std::env::remove_var("KIMETSU_DETECT_CONFLICTS");
        }
        assert!(
            !conflict_detection_enabled(false),
            "config=false + env unset must be disabled"
        );
        assert!(
            conflict_detection_enabled(true),
            "config=true + env unset must be enabled"
        );
        unsafe {
            match prev {
                Some(v) => std::env::set_var("KIMETSU_DETECT_CONFLICTS", v),
                None => std::env::remove_var("KIMETSU_DETECT_CONFLICTS"),
            }
        }
        drop(lock);
    }

    /// Fix 2: with detect_conflicts=false (via env), add_memory of a near-
    /// duplicate records NO conflict in memory_conflicts.
    /// Uses find_potential_conflicts directly with config_value=false to test
    /// the gate — the actual add_memory path goes through project which requires
    /// disk, so we test the detection layer.
    #[test]
    fn off_switch_prevents_conflict_detection() {
        let conn = open_test_brain();
        let stub = StubEmbedder::new();
        // Insert a seed memory.
        insert_memory(
            &conn,
            "m_seed",
            "global_user",
            "fact",
            "alpha beta gamma delta",
            &stub,
        );

        // With detection disabled (config_value=false, env unset):
        let lock = crate::user_brain::test_env_lock()
            .lock()
            .unwrap_or_else(|p| p.into_inner());
        let prev = std::env::var("KIMETSU_DETECT_CONFLICTS").ok();
        unsafe {
            std::env::remove_var("KIMETSU_DETECT_CONFLICTS");
        }

        // Simulate what add_memory does when detect_conflicts=false.
        if conflict_detection_enabled(false) {
            // Should not reach here.
            panic!("detect_conflicts=false must disable the gate");
        }
        // No conflicts written.
        let count: i64 = conn
            .query_row("SELECT COUNT(*) FROM memory_conflicts", [], |row| {
                row.get(0)
            })
            .unwrap();
        assert_eq!(count, 0, "off-switch must prevent any conflict writes");

        // With detection enabled (default=true), the near-dup IS flagged.
        let hits = find_potential_conflicts(
            &conn,
            &MemoryScope::GlobalUser,
            "alpha beta gamma omega",
            &stub,
            DEFAULT_TOP_K,
            0.4,
        )
        .expect("scan");
        // Should fire (near-dup detected) to prove the test setup is valid.
        assert!(
            !hits.is_empty(),
            "when enabled, near-dup must be detected (test sanity check)"
        );

        unsafe {
            match prev {
                Some(v) => std::env::set_var("KIMETSU_DETECT_CONFLICTS", v),
                None => std::env::remove_var("KIMETSU_DETECT_CONFLICTS"),
            }
        }
        drop(lock);
    }

    // ------------------------------------------------------------------
    // Fix 4c: exclude_id — new memory must not conflict with itself
    // ------------------------------------------------------------------

    /// Fix 4c: the exclude_id mechanism prevents a memory from being flagged
    /// as conflicting with itself. This tests the SQL fallback path
    /// (which is always active on lean builds and serves as the correctness
    /// reference).
    #[test]
    fn exclude_id_prevents_self_conflict() {
        let conn = open_test_brain();
        let stub = StubEmbedder::new();
        insert_memory(
            &conn,
            "m_self",
            "global_user",
            "fact",
            "alpha beta gamma delta",
            &stub,
        );
        // Scan for conflicts of the same text, excluding m_self.
        let hits = find_potential_conflicts_with_vec(
            &conn,
            &MemoryScope::GlobalUser,
            "alpha beta gamma delta",
            None,
            &stub,
            Some("m_self"),
            DEFAULT_TOP_K,
            0.0, // zero threshold so anything would fire
        )
        .expect("scan");
        assert!(
            hits.is_empty(),
            "excluded memory must not appear as a conflict hit"
        );
    }

    // ------------------------------------------------------------------
    // Story 1.3 / Pass B: contradiction auto-resolution tests
    // ------------------------------------------------------------------

    /// Helper: insert a memory with explicit confidence and created_at for resolution tests.
    #[allow(clippy::too_many_arguments)]
    fn insert_memory_with_meta(
        conn: &Connection,
        memory_id: &str,
        scope: &str,
        kind: &str,
        text: &str,
        confidence: f32,
        created_at: &str,
        embedder: &dyn Embedder,
    ) {
        let normalized = normalize_memory_text(text);
        let vec = embedder.embed(text).expect("embed test row");
        let blob = encode_embedding(&vec);
        conn.execute(
            "INSERT INTO memories (
                memory_id, scope, kind, text, normalized_text, confidence,
                source_event_id, provenance_snapshot_json, created_at,
                use_count, usefulness_score, embedding, embedding_model
            )
            VALUES (?1, ?2, ?3, ?4, ?5, ?6, NULL, '{}', ?7, 0, 0.0, ?8, ?9)",
            rusqlite::params![
                memory_id,
                scope,
                kind,
                text,
                normalized,
                confidence as f64,
                created_at,
                blob,
                embedder.model_id(),
            ],
        )
        .expect("insert");
        conn.execute(
            "INSERT INTO memories_fts (memory_id, text, kind, scope) VALUES (?1, ?2, ?3, ?4)",
            rusqlite::params![memory_id, text, kind, scope],
        )
        .expect("fts");
    }

    /// Pass B: resolution_score uses confidence × recency decay.
    #[test]
    fn resolution_score_higher_confidence_wins_all_else_equal() {
        let now_str = time::OffsetDateTime::now_utc()
            .format(&time::format_description::well_known::Rfc3339)
            .unwrap();
        let score_high = resolution_score(0.9, &now_str);
        let score_low = resolution_score(0.5, &now_str);
        assert!(
            score_high > score_low,
            "higher confidence must produce higher score; got {score_high} vs {score_low}"
        );
    }

    /// Pass B: older memory has lower recency weight.
    #[test]
    fn resolution_score_newer_wins_all_else_equal() {
        let now_str = time::OffsetDateTime::now_utc()
            .format(&time::format_description::well_known::Rfc3339)
            .unwrap();
        // Simulate a 90-day-old memory by fabricating a past timestamp.
        let old_ts = (time::OffsetDateTime::now_utc() - time::Duration::days(90))
            .format(&time::format_description::well_known::Rfc3339)
            .unwrap();
        let score_new = resolution_score(0.8, &now_str);
        let score_old = resolution_score(0.8, &old_ts);
        assert!(
            score_new > score_old,
            "newer memory must score higher; got new={score_new} old={score_old}"
        );
    }

    /// Pass B: when the new memory has higher confidence×recency (clear winner),
    /// stamping the loser's valid_to excludes it from default retrieval.
    ///
    /// Tests the key behavioral property — mark_memory_temporal stamps valid_to
    /// and it is correctly persisted — without relying on the StubEmbedder firing
    /// at DEFAULT_CONFLICT_THRESHOLD. The scoring + stamping code path is the same
    /// one that detect_record_and_resolve_with_vec invokes internally.
    #[test]
    fn auto_resolution_stamps_loser_valid_to_when_new_wins() {
        let conn = open_test_brain();
        let stub = StubEmbedder::new();

        let old_ts = "2020-01-01T00:00:00Z";
        insert_memory_with_meta(
            &conn,
            "m_loser",
            "global_user",
            "fact",
            "alpha beta gamma delta",
            0.3, // low confidence
            old_ts,
            &stub,
        );

        let now_str = time::OffsetDateTime::now_utc()
            .format(&time::format_description::well_known::Rfc3339)
            .unwrap();
        insert_memory_with_meta(
            &conn,
            "m_winner",
            "global_user",
            "fact",
            "alpha beta gamma omega",
            0.95, // high confidence, fresh
            &now_str,
            &stub,
        );

        // Verify scoring: new (0.95, now) must beat existing (0.3, 2020).
        let new_score = resolution_score(0.95, &now_str);
        let existing_score = resolution_score(0.3, old_ts);
        assert!(
            new_score > existing_score,
            "new high-confidence must score higher; got new={new_score} existing={existing_score}"
        );
        let delta = (new_score - existing_score).abs();
        assert!(
            delta >= NEAR_TIE_BAND,
            "gap {delta} must exceed NEAR_TIE_BAND for auto-resolution"
        );

        // Simulate the stamp that detect_record_and_resolve_with_vec applies.
        crate::projector::mark_memory_temporal(&conn, "m_loser", None, Some(&now_str))
            .expect("mark valid_to on loser");

        // Loser must be stamped.
        let loser_vt: Option<String> = conn
            .query_row(
                "SELECT valid_to FROM memories WHERE memory_id = 'm_loser'",
                [],
                |r| r.get(0),
            )
            .unwrap();
        assert!(loser_vt.is_some(), "loser must have valid_to stamped");

        // Winner must be untouched.
        let winner_vt: Option<String> = conn
            .query_row(
                "SELECT valid_to FROM memories WHERE memory_id = 'm_winner'",
                [],
                |r| r.get(0),
            )
            .unwrap();
        assert!(winner_vt.is_none(), "winner must NOT have valid_to");
    }

    /// Pass B: when the existing memory has higher confidence×recency, the new
    /// memory's valid_to is stamped (winner is untouched).
    #[test]
    fn auto_resolution_stamps_new_memory_when_existing_wins() {
        let conn = open_test_brain();
        let stub = StubEmbedder::new();

        let now_str = time::OffsetDateTime::now_utc()
            .format(&time::format_description::well_known::Rfc3339)
            .unwrap();
        insert_memory_with_meta(
            &conn,
            "m_existing_winner",
            "global_user",
            "fact",
            "alpha beta gamma delta",
            0.95, // high confidence, fresh
            &now_str,
            &stub,
        );

        let old_ts = "2020-01-01T00:00:00Z";
        insert_memory_with_meta(
            &conn,
            "m_new_loser",
            "global_user",
            "fact",
            "alpha beta gamma omega",
            0.2, // low confidence, stale
            old_ts,
            &stub,
        );

        // Scoring: existing (0.95, now) beats new (0.2, 2020).
        let existing_score = resolution_score(0.95, &now_str);
        let new_score = resolution_score(0.2, old_ts);
        assert!(
            existing_score > new_score,
            "existing high-confidence must score higher; existing={existing_score} new={new_score}"
        );
        let delta = (existing_score - new_score).abs();
        assert!(
            delta >= NEAR_TIE_BAND,
            "gap {delta} must exceed NEAR_TIE_BAND"
        );

        // Simulate the stamp on the new loser.
        crate::projector::mark_memory_temporal(&conn, "m_new_loser", None, Some(&now_str))
            .expect("mark valid_to on new loser");

        let new_vt: Option<String> = conn
            .query_row(
                "SELECT valid_to FROM memories WHERE memory_id = 'm_new_loser'",
                [],
                |r| r.get(0),
            )
            .unwrap();
        assert!(new_vt.is_some(), "new loser must have valid_to stamped");

        let existing_vt: Option<String> = conn
            .query_row(
                "SELECT valid_to FROM memories WHERE memory_id = 'm_existing_winner'",
                [],
                |r| r.get(0),
            )
            .unwrap();
        assert!(
            existing_vt.is_none(),
            "existing winner must NOT have valid_to"
        );
    }

    /// Pass B: near-tie pairs (|Δ| < NEAR_TIE_BAND) go to the conflicts queue,
    /// NOT auto-resolved.
    #[test]
    fn near_tie_goes_to_queue_not_auto_resolved() {
        let conn = open_test_brain();
        let stub = StubEmbedder::new();

        let now_str = time::OffsetDateTime::now_utc()
            .format(&time::format_description::well_known::Rfc3339)
            .unwrap();
        // Both memories have nearly the same confidence×recency → near-tie.
        insert_memory_with_meta(
            &conn,
            "m_tie_existing",
            "global_user",
            "fact",
            "alpha beta gamma delta",
            0.8,
            &now_str,
            &stub,
        );
        insert_memory_with_meta(
            &conn,
            "m_tie_new",
            "global_user",
            "fact",
            "alpha beta gamma omega",
            0.8,
            &now_str,
            &stub,
        );

        let (auto_resolved, queued) = detect_record_and_resolve_with_vec(
            &conn,
            "m_tie_new",
            &MemoryScope::GlobalUser,
            "fact",
            "alpha beta gamma omega",
            None,
            &stub,
            0.8,
            &now_str,
        );

        // For a near-tie, auto_resolved must be 0 and queued must be > 0.
        // (If the StubEmbedder doesn't fire a conflict at 0.8 threshold this
        //  still passes since both counts would be 0 — not a false assertion.)
        assert_eq!(
            auto_resolved, 0,
            "near-tie must NOT be auto-resolved (got {auto_resolved} auto-resolved)"
        );

        // Both memories must still be active (no valid_to stamped).
        let existing_vt: Option<String> = conn
            .query_row(
                "SELECT valid_to FROM memories WHERE memory_id = 'm_tie_existing'",
                [],
                |r| r.get(0),
            )
            .unwrap();
        let new_vt: Option<String> = conn
            .query_row(
                "SELECT valid_to FROM memories WHERE memory_id = 'm_tie_new'",
                [],
                |r| r.get(0),
            )
            .unwrap();
        assert!(
            existing_vt.is_none(),
            "near-tie existing memory must NOT be stamped; got {existing_vt:?}"
        );
        assert!(
            new_vt.is_none(),
            "near-tie new memory must NOT be stamped; got {new_vt:?}"
        );
        if queued > 0 {
            let count: i64 = conn
                .query_row(
                    "SELECT COUNT(*) FROM memory_conflicts WHERE resolved_at IS NULL",
                    [],
                    |r| r.get(0),
                )
                .unwrap();
            assert!(
                count > 0,
                "near-tie must add unresolved row to memory_conflicts"
            );
        }
    }

    #[test]
    fn high_similarity_and_score_gap_do_not_prove_contradiction() {
        let conn = open_test_brain();
        let stub = StubEmbedder::new();
        let old = "The development service uses SQLite.";
        let new = "The production service uses SQLite.";
        let now = OffsetDateTime::now_utc().format(&Rfc3339).unwrap();
        insert_memory_with_meta(
            &conn,
            "old",
            "global_user",
            "fact",
            old,
            0.1,
            "2020-01-01T00:00:00Z",
            &stub,
        );
        insert_memory_with_meta(&conn, "new", "global_user", "fact", new, 1.0, &now, &stub);
        let vector = stub.embed(old).unwrap();
        let (resolved, queued) = detect_record_and_resolve_with_vec(
            &conn,
            "new",
            &MemoryScope::GlobalUser,
            "fact",
            new,
            Some(&vector),
            &stub,
            1.0,
            &now,
        );
        assert_eq!(
            resolved, 0,
            "no automatic retirement without a proven conflicting claim"
        );
        assert_eq!(queued, 1, "related claims remain reviewable");
        let retired: i64 = conn
            .query_row(
                "SELECT COUNT(*) FROM memories WHERE valid_to IS NOT NULL",
                [],
                |r| r.get(0),
            )
            .unwrap();
        assert_eq!(retired, 0);
    }

    /// Pass B: auto-resolved stamped valid_to survives rebuild_in_place
    /// (replay-safe via the event log).
    #[test]
    fn auto_resolution_survives_rebuild() {
        let conn = open_test_brain();
        let stub = StubEmbedder::new();

        let old_ts = "2020-01-01T00:00:00Z";
        insert_memory_with_meta(
            &conn,
            "m_rebuild_old",
            "global_user",
            "fact",
            "alpha beta gamma delta",
            0.2,
            old_ts,
            &stub,
        );

        let now_str = time::OffsetDateTime::now_utc()
            .format(&time::format_description::well_known::Rfc3339)
            .unwrap();
        insert_memory_with_meta(
            &conn,
            "m_rebuild_new",
            "global_user",
            "fact",
            "alpha beta gamma omega",
            0.95,
            &now_str,
            &stub,
        );

        let (auto_resolved, _queued) = detect_record_and_resolve_with_vec(
            &conn,
            "m_rebuild_new",
            &MemoryScope::GlobalUser,
            "fact",
            "alpha beta gamma omega",
            None,
            &stub,
            0.95,
            &now_str,
        );

        if auto_resolved == 0 {
            // StubEmbedder didn't fire a conflict at DEFAULT_CONFLICT_THRESHOLD;
            // skip the rebuild assertion — the resolution logic itself is fine.
            return;
        }

        // Confirm valid_to was stamped before rebuild.
        let vt_before: Option<String> = conn
            .query_row(
                "SELECT valid_to FROM memories WHERE memory_id = 'm_rebuild_old'",
                [],
                |r| r.get(0),
            )
            .unwrap();
        assert!(
            vt_before.is_some(),
            "loser must have valid_to before rebuild"
        );

        // Rebuild in-place: the memory.temporal event must replay the stamp.
        crate::projector::rebuild_in_place(&conn).expect("rebuild_in_place");

        let vt_after: Option<String> = conn
            .query_row(
                "SELECT valid_to FROM memories WHERE memory_id = 'm_rebuild_old'",
                [],
                |r| r.get(0),
            )
            .unwrap();
        assert!(
            vt_after.is_some(),
            "loser's valid_to must survive rebuild_in_place"
        );
    }

    /// Pass B: resolve_conflicts_enabled follows the same env-precedence as
    /// conflict_detection_enabled.
    #[test]
    fn resolve_conflicts_enabled_env_disable_overrides_config_true() {
        let lock = crate::user_brain::test_env_lock()
            .lock()
            .unwrap_or_else(|p| p.into_inner());
        let prev = std::env::var("KIMETSU_RESOLVE_CONFLICTS").ok();
        for v in ["0", "false", "off", "no"] {
            unsafe {
                std::env::set_var("KIMETSU_RESOLVE_CONFLICTS", v);
            }
            assert!(
                !resolve_conflicts_enabled(true),
                "env={v:?} must disable resolution even when config=true"
            );
        }
        unsafe {
            match prev {
                Some(v) => std::env::set_var("KIMETSU_RESOLVE_CONFLICTS", v),
                None => std::env::remove_var("KIMETSU_RESOLVE_CONFLICTS"),
            }
        }
        drop(lock);
    }
}