yantrikdb 0.7.16

//! Conflict detection and resolution.
//!
//! Rule-based detection engine for semantic contradictions across synced memories.
//! Conflicts are first-class data: stored in their own table, queryable, auditable,
//! and replicated via the oplog.

use rusqlite::params;

use crate::engine::YantrikDB;
use crate::error::Result;
use crate::types::{Conflict, ConflictType};

/// Rel types that indicate unique-value identity facts (should not have multiple values).
const IDENTITY_REL_TYPES: &[&str] = &[
    "birthday",
    "age",
    "lives_in",
    "works_at",
    "email",
    "phone",
    "full_name",
    "spouse",
    "hometown",
];

/// Rel types that indicate preferences (concurrent differences are suspicious).
const PREFERENCE_REL_TYPES: &[&str] = &["prefers", "favorite", "likes", "dislikes"];

/// Classify a conflict type from the rel_type.
fn classify_conflict(rel_type: &str) -> ConflictType {
    if IDENTITY_REL_TYPES.contains(&rel_type) {
        ConflictType::IdentityFact
    } else if PREFERENCE_REL_TYPES.contains(&rel_type) {
        ConflictType::Preference
    } else {
        ConflictType::Minor
    }
}

/// Entity types that, when substituted in otherwise-identical sentences, indicate
/// a factual contradiction (identity-level conflict).
const IDENTITY_ENTITY_TYPES: &[&str] = &["organization", "place", "person"];

/// Entity types where substitution indicates a preference contradiction.
const PREFERENCE_ENTITY_TYPES: &[&str] = &["tech"];

/// Temporal keywords whose presence (when differing) suggests a temporal conflict.
const TEMPORAL_KEYWORDS: &[&str] = &[
    "january",
    "february",
    "march",
    "april",
    "may",
    "june",
    "july",
    "august",
    "september",
    "october",
    "november",
    "december",
    "monday",
    "tuesday",
    "wednesday",
    "thursday",
    "friday",
    "saturday",
    "sunday",
    "morning",
    "afternoon",
    "evening",
    "night",
    "today",
    "tomorrow",
    "yesterday",
    "2024",
    "2025",
    "2026",
    "2027",
    "q1",
    "q2",
    "q3",
    "q4",
];

/// Date-like regex patterns for temporal substitution detection.
const DATE_PATTERNS: &[&str] = &[
    // Already handled by TEMPORAL_KEYWORDS: month names, day names, etc.
    // These catch numeric dates that TEMPORAL_KEYWORDS miss.
];

/// Check if a token looks like a date component (numeric date parts).
fn is_date_like(token: &str) -> bool {
    // ISO date: 2024-01-15 (split into parts: 2024, 01, 15)
    // Already covered by year keywords for 4-digit years.
    // Catch day/month numbers: 1-31
    if let Ok(n) = token.parse::<u32>() {
        return (1..=31).contains(&n);
    }
    // Ordinals: 1st, 2nd, 3rd, 15th, etc.
    if token.len() >= 3 && token.ends_with("st")
        || token.ends_with("nd")
        || token.ends_with("rd")
        || token.ends_with("th")
    {
        let num_part = &token[..token.len() - 2];
        if let Ok(n) = num_part.parse::<u32>() {
            return (1..=31).contains(&n);
        }
    }
    false
}

/// Map substitution category names to ConflictType.
/// Identity-like categories produce IdentityFact; everything else produces Preference.
const IDENTITY_CATEGORIES: &[&str] = &["cloud_providers"];

/// Check substitution_members table for category-based conflict.
fn check_category_substitution(
    conn: &rusqlite::Connection,
    diff_a: &[&String],
    diff_b: &[&String],
) -> Option<(ConflictType, String)> {
    // Try each pair of diff tokens
    let mut stmt = match conn.prepare_cached(
        "SELECT c.name, c.conflict_mode
         FROM substitution_members m1
         JOIN substitution_members m2 ON m1.category_id = m2.category_id
         JOIN substitution_categories c ON c.id = m1.category_id
         WHERE m1.token_normalized = ?1 AND m2.token_normalized = ?2
           AND m1.status = 'active' AND m2.status = 'active'
           AND m1.confidence >= 0.6 AND m2.confidence >= 0.6
           AND c.status = 'active'
         LIMIT 1",
    ) {
        Ok(s) => s,
        Err(_) => return None,
    };

    for token_a in diff_a {
        for token_b in diff_b {
            if let Ok((cat_name, _conflict_mode)) = stmt
                .query_row(params![token_a.as_str(), token_b.as_str()], |row| {
                    Ok((row.get::<_, String>(0)?, row.get::<_, String>(1)?))
                })
            {
                let conflict_type = if IDENTITY_CATEGORIES.contains(&cat_name.as_str()) {
                    ConflictType::IdentityFact
                } else {
                    ConflictType::Preference
                };
                let desc = format!(
                    "{} category substitution: {{{}}} vs {{{}}}",
                    cat_name, token_a, token_b,
                );
                return Some((conflict_type, desc));
            }
        }
    }

    // Try multi-word: join all diff tokens and check
    if diff_a.len() >= 2 || diff_b.len() >= 2 {
        let joined_a: String = diff_a
            .iter()
            .map(|s| s.as_str())
            .collect::<Vec<_>>()
            .join(" ");
        let joined_b: String = diff_b
            .iter()
            .map(|s| s.as_str())
            .collect::<Vec<_>>()
            .join(" ");
        if let Ok((cat_name, _)) = stmt.query_row(params![joined_a, joined_b], |row| {
            Ok((row.get::<_, String>(0)?, row.get::<_, String>(1)?))
        }) {
            let conflict_type = if IDENTITY_CATEGORIES.contains(&cat_name.as_str()) {
                ConflictType::IdentityFact
            } else {
                ConflictType::Preference
            };
            let desc = format!(
                "{} category substitution: {{{}}} vs {{{}}}",
                cat_name, joined_a, joined_b,
            );
            return Some((conflict_type, desc));
        }
    }

    None
}

/// Detect entity substitution in two memory texts.
///
/// Detection flow (in priority order):
/// 1. Temporal keywords (month names, days, years, etc.)
/// 2. Date-like numeric patterns (ordinals, day numbers)
/// 3. Substitution category lookup (learned + seed categories)
/// 4. Entity table lookup (legacy fallback)
fn classify_entity_substitution(
    conn: &rusqlite::Connection,
    text_a: &str,
    text_b: &str,
) -> (ConflictType, Option<String>) {
    let words_a: std::collections::HashSet<String> = text_a
        .split_whitespace()
        .map(|w| {
            w.trim_matches(|c: char| !c.is_alphanumeric())
                .to_lowercase()
        })
        .filter(|w| !w.is_empty())
        .collect();
    let words_b: std::collections::HashSet<String> = text_b
        .split_whitespace()
        .map(|w| {
            w.trim_matches(|c: char| !c.is_alphanumeric())
                .to_lowercase()
        })
        .filter(|w| !w.is_empty())
        .collect();

    let diff_a: Vec<&String> = words_a.difference(&words_b).collect();
    let diff_b: Vec<&String> = words_b.difference(&words_a).collect();

    // ── Step 1: Temporal keyword substitution ──
    let temporal_a = diff_a
        .iter()
        .any(|w| TEMPORAL_KEYWORDS.contains(&w.as_str()));
    let temporal_b = diff_b
        .iter()
        .any(|w| TEMPORAL_KEYWORDS.contains(&w.as_str()));
    if temporal_a && temporal_b {
        let diff_desc = format!(
            "temporal substitution: {{{}}} vs {{{}}}",
            diff_a
                .iter()
                .map(|s| s.as_str())
                .collect::<Vec<_>>()
                .join(", "),
            diff_b
                .iter()
                .map(|s| s.as_str())
                .collect::<Vec<_>>()
                .join(", "),
        );
        return (ConflictType::Temporal, Some(diff_desc));
    }

    // ── Step 2: Date-like numeric patterns ──
    let date_a = diff_a.iter().any(|w| is_date_like(w));
    let date_b = diff_b.iter().any(|w| is_date_like(w));
    if (temporal_a || date_a) && (temporal_b || date_b) {
        let diff_desc = format!(
            "date substitution: {{{}}} vs {{{}}}",
            diff_a
                .iter()
                .map(|s| s.as_str())
                .collect::<Vec<_>>()
                .join(", "),
            diff_b
                .iter()
                .map(|s| s.as_str())
                .collect::<Vec<_>>()
                .join(", "),
        );
        return (ConflictType::Temporal, Some(diff_desc));
    }

    // ── Step 3: Substitution category lookup (seed + learned) ──
    if let Some((conflict_type, desc)) = check_category_substitution(conn, &diff_a, &diff_b) {
        return (conflict_type, Some(desc));
    }

    // ── Step 4: Entity table lookup (legacy fallback) ──
    let mut entity_types_a: Vec<String> = Vec::new();
    let mut entity_types_b: Vec<String> = Vec::new();
    let mut entity_names_a: Vec<String> = Vec::new();
    let mut entity_names_b: Vec<String> = Vec::new();

    if let Ok(mut stmt) =
        conn.prepare_cached("SELECT name, entity_type FROM entities WHERE LOWER(name) = ?1")
    {
        for word in &diff_a {
            if let Ok(etype) = stmt.query_row(params![word.as_str()], |row| row.get::<_, String>(1))
            {
                entity_types_a.push(etype);
                entity_names_a.push(word.to_string());
            }
        }
        for word in &diff_b {
            if let Ok(etype) = stmt.query_row(params![word.as_str()], |row| row.get::<_, String>(1))
            {
                entity_types_b.push(etype);
                entity_names_b.push(word.to_string());
            }
        }
    }

    // Multi-word entity matching
    if entity_types_a.is_empty() && diff_a.len() >= 2 {
        let joined: String = diff_a
            .iter()
            .map(|s| s.as_str())
            .collect::<Vec<_>>()
            .join(" ");
        if let Ok(mut stmt) =
            conn.prepare_cached("SELECT name, entity_type FROM entities WHERE LOWER(name) = ?1")
        {
            if let Ok((name, etype)) = stmt.query_row(params![joined], |row| {
                Ok((row.get::<_, String>(0)?, row.get::<_, String>(1)?))
            }) {
                entity_types_a.push(etype);
                entity_names_a.push(name);
            }
        }
    }
    if entity_types_b.is_empty() && diff_b.len() >= 2 {
        let joined: String = diff_b
            .iter()
            .map(|s| s.as_str())
            .collect::<Vec<_>>()
            .join(" ");
        if let Ok(mut stmt) =
            conn.prepare_cached("SELECT name, entity_type FROM entities WHERE LOWER(name) = ?1")
        {
            if let Ok((name, etype)) = stmt.query_row(params![joined], |row| {
                Ok((row.get::<_, String>(0)?, row.get::<_, String>(1)?))
            }) {
                entity_types_b.push(etype);
                entity_names_b.push(name);
            }
        }
    }

    // Check if both sides have entities of the same type
    for type_a in &entity_types_a {
        for type_b in &entity_types_b {
            if type_a == type_b {
                let diff_desc = format!(
                    "{} substitution: {{{}}} vs {{{}}}",
                    type_a,
                    entity_names_a.join(", "),
                    entity_names_b.join(", "),
                );

                if IDENTITY_ENTITY_TYPES.contains(&type_a.as_str()) {
                    return (ConflictType::IdentityFact, Some(diff_desc));
                }
                if PREFERENCE_ENTITY_TYPES.contains(&type_a.as_str()) {
                    return (ConflictType::Preference, Some(diff_desc));
                }
                return (ConflictType::Minor, Some(diff_desc));
            }
        }
    }

    // No substitution detected
    (ConflictType::Minor, None)
}

/// Map a substitution category name to the appropriate ConflictType.
pub(crate) fn category_to_conflict_type(cat_name: &str) -> ConflictType {
    if IDENTITY_CATEGORIES.contains(&cat_name) {
        ConflictType::IdentityFact
    } else {
        ConflictType::Preference
    }
}

/// Check if a conflict already exists for this (memory_a, memory_b) pair.
/// Checks both orderings.
pub(crate) fn conflict_exists(db: &YantrikDB, rid_a: &str, rid_b: &str) -> Result<bool> {
    let conn = db.conn();
    let exists: bool = conn.query_row(
        "SELECT COUNT(*) > 0 FROM conflicts
         WHERE (memory_a = ?1 AND memory_b = ?2)
            OR (memory_a = ?2 AND memory_b = ?1)",
        params![rid_a, rid_b],
        |row| row.get(0),
    )?;
    Ok(exists)
}

/// Find the oplog target_rid for a relate op with given (src, dst, rel_type).
fn find_memory_for_edge(
    conn: &rusqlite::Connection,
    src: &str,
    dst: &str,
    rel_type: &str,
) -> Result<Option<String>> {
    let result = conn.query_row(
        "SELECT target_rid FROM oplog
         WHERE op_type = 'relate'
           AND json_extract(payload, '$.src') = ?1
           AND json_extract(payload, '$.dst') = ?2
           AND json_extract(payload, '$.rel_type') = ?3
         ORDER BY hlc DESC LIMIT 1",
        params![src, dst, rel_type],
        |row| row.get::<_, Option<String>>(0),
    );

    match result {
        Ok(rid) => Ok(rid),
        Err(rusqlite::Error::QueryReturnedNoRows) => Ok(None),
        Err(e) => Err(e.into()),
    }
}

/// Create a conflict record and log it to the oplog for replication.
pub fn create_conflict(
    db: &YantrikDB,
    conflict_type: &ConflictType,
    memory_a: &str,
    memory_b: &str,
    entity: Option<&str>,
    rel_type: Option<&str>,
    detection_reason: &str,
) -> Result<Conflict> {
    let conflict_id = crate::id::new_id();
    let ts = crate::time::now_secs();
    let priority = conflict_type.default_priority();
    let hlc_ts = db.tick_hlc();
    let hlc_bytes = hlc_ts.to_bytes().to_vec();
    let actor_id = db.actor_id().to_string();

    db.conn().execute(
        "INSERT OR IGNORE INTO conflicts
         (conflict_id, conflict_type, priority, status, memory_a, memory_b,
          entity, rel_type, detected_at, detected_by, detection_reason,
          hlc, origin_actor)
         VALUES (?1, ?2, ?3, 'open', ?4, ?5, ?6, ?7, ?8, ?9, ?10, ?11, ?12)",
        params![
            conflict_id,
            conflict_type.as_str(),
            priority,
            memory_a,
            memory_b,
            entity,
            rel_type,
            ts,
            actor_id,
            detection_reason,
            hlc_bytes,
            actor_id,
        ],
    )?;

    // Log to oplog for replication
    db.log_op(
        "conflict_detect",
        Some(&conflict_id),
        &serde_json::json!({
            "conflict_id": conflict_id,
            "conflict_type": conflict_type.as_str(),
            "priority": priority,
            "memory_a": memory_a,
            "memory_b": memory_b,
            "entity": entity,
            "rel_type": rel_type,
            "detected_at": ts,
            "detected_by": actor_id,
            "detection_reason": detection_reason,
        }),
        None,
    )?;

    Ok(Conflict {
        conflict_id,
        conflict_type: conflict_type.as_str().to_string(),
        priority: priority.to_string(),
        status: "open".to_string(),
        memory_a: memory_a.to_string(),
        memory_b: memory_b.to_string(),
        entity: entity.map(String::from),
        rel_type: rel_type.map(String::from),
        detected_at: ts,
        detected_by: actor_id,
        detection_reason: detection_reason.to_string(),
        resolved_at: None,
        resolved_by: None,
        strategy: None,
        winner_rid: None,
        resolution_note: None,
    })
}

/// Detect edge-based contradictions for a newly materialized edge.
/// Called from materialize_relate in replication.rs during sync.
pub fn detect_edge_conflicts(
    db: &YantrikDB,
    src: &str,
    dst: &str,
    rel_type: &str,
    incoming_target_rid: Option<&str>,
) -> Result<Vec<Conflict>> {
    let mut conflicts = Vec::new();

    // Only check identity and preference rel_types
    let is_identity = IDENTITY_REL_TYPES.contains(&rel_type);
    let is_preference = PREFERENCE_REL_TYPES.contains(&rel_type);
    if !is_identity && !is_preference {
        return Ok(conflicts);
    }

    // Collect data while holding the conn lock, then release before calling
    // conflict_exists/create_conflict (which also acquire the lock).
    let edge_data: Vec<(String, Option<String>, Option<String>)> = {
        let conn = db.conn();
        let mut stmt = conn.prepare(
            "SELECT edge_id, dst FROM edges
             WHERE src = ?1 AND rel_type = ?2 AND dst != ?3 AND tombstoned = 0",
        )?;

        let existing: Vec<(String, String)> = stmt
            .query_map(params![src, rel_type, dst], |row| {
                Ok((row.get::<_, String>(0)?, row.get::<_, String>(1)?))
            })?
            .collect::<std::result::Result<Vec<_>, _>>()?;

        existing
            .into_iter()
            .map(|(_edge_id, existing_dst)| {
                let memory_a = find_memory_for_edge(&conn, src, &existing_dst, rel_type)
                    .ok()
                    .flatten();
                let memory_b = incoming_target_rid.map(String::from).or_else(|| {
                    find_memory_for_edge(&conn, src, dst, rel_type)
                        .ok()
                        .flatten()
                });
                (existing_dst, memory_a, memory_b)
            })
            .collect()
    }; // conn lock released here

    for (existing_dst, memory_a, memory_b) in edge_data {
        let conflict_type = classify_conflict(rel_type);

        if let (Some(ref mem_a), Some(ref mem_b)) = (&memory_a, &memory_b) {
            if !conflict_exists(db, mem_a, mem_b)? {
                let conflict = create_conflict(
                    db,
                    &conflict_type,
                    mem_a,
                    mem_b,
                    Some(src),
                    Some(rel_type),
                    &format!(
                        "Entity '{}' has conflicting {} values: '{}' vs '{}'",
                        src, rel_type, existing_dst, dst
                    ),
                )?;
                conflicts.push(conflict);
            }
        }
    }

    Ok(conflicts)
}

/// Full-database conflict scan. Finds all edge-based contradictions
/// and concurrent consolidation conflicts.
pub fn scan_conflicts(db: &YantrikDB) -> Result<Vec<Conflict>> {
    scan_conflicts_limited(db, 50)
}

/// Scan for conflicts with a limit on max conflicts to detect per scan.
pub fn scan_conflicts_limited(db: &YantrikDB, max_conflicts: usize) -> Result<Vec<Conflict>> {
    let mut conflicts = Vec::new();

    // Phase 1: Collect edge-based conflict candidates while holding conn lock.
    // Each candidate: (src, rel_type, dst_i, dst_j, mem_a, mem_b)
    let edge_candidates: Vec<(
        String,
        String,
        String,
        String,
        Option<String>,
        Option<String>,
    )>;
    let entity_groups: std::collections::HashMap<String, Vec<(String, String, Vec<u8>)>>;
    let cm_rows: Vec<(String, String, String)>;

    {
        let conn = db.conn();

        // Scan for contradicting edges: same (src, rel_type) with different dst values
        let mut stmt = conn.prepare(
            "SELECT src, rel_type, GROUP_CONCAT(DISTINCT dst) as dsts, COUNT(DISTINCT dst) as cnt
             FROM edges
             WHERE tombstoned = 0
             GROUP BY src, rel_type
             HAVING cnt > 1",
        )?;

        let rows: Vec<(String, String, String)> = stmt
            .query_map([], |row| {
                Ok((
                    row.get::<_, String>(0)?,
                    row.get::<_, String>(1)?,
                    row.get::<_, String>(2)?,
                ))
            })?
            .collect::<std::result::Result<Vec<_>, _>>()?;

        let mut candidates = Vec::new();
        for (src, rel_type, dsts_csv) in rows {
            let is_identity = IDENTITY_REL_TYPES.contains(&rel_type.as_str());
            let is_preference = PREFERENCE_REL_TYPES.contains(&rel_type.as_str());
            if !is_identity && !is_preference {
                continue;
            }

            let dsts: Vec<String> = dsts_csv.split(',').map(|s| s.trim().to_string()).collect();
            if dsts.len() < 2 {
                continue;
            }

            for i in 0..dsts.len() {
                for j in (i + 1)..dsts.len() {
                    let mem_a = find_memory_for_edge(&conn, &src, &dsts[i], &rel_type)
                        .ok()
                        .flatten();
                    let mem_b = find_memory_for_edge(&conn, &src, &dsts[j], &rel_type)
                        .ok()
                        .flatten();
                    candidates.push((
                        src.clone(),
                        rel_type.clone(),
                        dsts[i].clone(),
                        dsts[j].clone(),
                        mem_a,
                        mem_b,
                    ));
                }
            }
        }
        edge_candidates = candidates;

        // Scan for entity-based semantic conflicts (cap to recent memories).
        // Limit scales with max_conflicts to keep incremental calls fast.
        let entity_scan_limit = (max_conflicts * 10).max(50).min(200);
        let entity_query = format!(
            "SELECT me.entity_name, m.rid, m.text, m.embedding
             FROM memory_entities me
             JOIN memories m ON m.rid = me.memory_rid
             WHERE m.consolidation_status = 'active'
             AND m.embedding IS NOT NULL
             ORDER BY m.created_at DESC
             LIMIT {}",
            entity_scan_limit
        );
        let mut entity_mem_stmt = conn.prepare(&entity_query)?;

        let em_rows: Vec<(String, String, String, Vec<u8>)> = entity_mem_stmt
            .query_map([], |row| {
                Ok((
                    row.get::<_, String>(0)?,
                    row.get::<_, String>(1)?,
                    row.get::<_, String>(2)?,
                    row.get::<_, Vec<u8>>(3)?,
                ))
            })?
            .collect::<std::result::Result<Vec<_>, _>>()?;

        let mut groups: std::collections::HashMap<String, Vec<(String, String, Vec<u8>)>> =
            std::collections::HashMap::new();
        for (entity, rid, text, emb) in em_rows {
            let text = db.decrypt_text(&text).unwrap_or(text);
            let emb = db.decrypt_embedding(&emb).unwrap_or(emb);
            groups.entry(entity).or_default().push((rid, text, emb));
        }
        entity_groups = groups;

        // Scan for concurrent consolidation conflicts
        let mut cm_stmt = conn.prepare(
            "SELECT cm1.consolidation_rid, cm2.consolidation_rid, cm1.source_rid
             FROM consolidation_members cm1
             JOIN consolidation_members cm2
               ON cm1.source_rid = cm2.source_rid
              AND cm1.consolidation_rid < cm2.consolidation_rid",
        )?;

        cm_rows = cm_stmt
            .query_map([], |row| {
                Ok((
                    row.get::<_, String>(0)?,
                    row.get::<_, String>(1)?,
                    row.get::<_, String>(2)?,
                ))
            })?
            .collect::<std::result::Result<Vec<_>, _>>()?;
    } // conn lock released here

    // Phase 2: Create conflicts (these functions acquire conn lock internally).

    // Edge-based conflicts
    for (src, rel_type, dst_i, dst_j, mem_a, mem_b) in &edge_candidates {
        if let (Some(a), Some(b)) = (mem_a, mem_b) {
            if !conflict_exists(db, a, b)? {
                let conflict_type = classify_conflict(rel_type);
                let conflict = create_conflict(
                    db,
                    &conflict_type,
                    a,
                    b,
                    Some(src),
                    Some(rel_type),
                    &format!(
                        "Entity '{}' has conflicting {} values: '{}' vs '{}'",
                        src, rel_type, dst_i, dst_j
                    ),
                )?;
                conflicts.push(conflict);
            }
        }
    }

    // Entity-based semantic conflicts
    {
        let mut seen_pairs: std::collections::HashSet<(String, String)> =
            std::collections::HashSet::new();

        for (entity, memories) in &entity_groups {
            if conflicts.len() >= max_conflicts {
                break;
            }
            if memories.len() < 2 {
                continue;
            }
            // Cap pairwise comparisons per entity to avoid O(n²) blowup
            let mem_limit = memories.len().min(20);
            for i in 0..mem_limit {
                for j in (i + 1)..mem_limit {
                    let (ref rid_a, ref text_a, ref emb_a) = memories[i];
                    let (ref rid_b, ref text_b, ref emb_b) = memories[j];

                    let pair = if rid_a < rid_b {
                        (rid_a.clone(), rid_b.clone())
                    } else {
                        (rid_b.clone(), rid_a.clone())
                    };
                    if seen_pairs.contains(&pair) {
                        continue;
                    }

                    let emb_a_f32 = crate::serde_helpers::deserialize_f32(emb_a);
                    let emb_b_f32 = crate::serde_helpers::deserialize_f32(emb_b);
                    let sim = crate::consolidate::cosine_similarity(&emb_a_f32, &emb_b_f32);

                    // Similar topic (>0.5) but not exact duplicate (<0.98)
                    if sim > 0.5 && sim < 0.98 {
                        // Compute word-level Jaccard to detect different content
                        let words_a: std::collections::HashSet<&str> = text_a
                            .split_whitespace()
                            .map(|w| w.trim_matches(|c: char| !c.is_alphanumeric()))
                            .filter(|w| !w.is_empty())
                            .collect();
                        let words_b: std::collections::HashSet<&str> = text_b
                            .split_whitespace()
                            .map(|w| w.trim_matches(|c: char| !c.is_alphanumeric()))
                            .filter(|w| !w.is_empty())
                            .collect();

                        let intersection = words_a.intersection(&words_b).count();
                        let union = words_a.union(&words_b).count();
                        let jaccard = if union > 0 {
                            intersection as f64 / union as f64
                        } else {
                            1.0
                        };

                        // High semantic similarity + low word overlap = likely contradiction
                        // (they're about the same topic but say different things)
                        if jaccard < 0.7 {
                            seen_pairs.insert(pair);
                            if !conflict_exists(db, rid_a, rid_b)? {
                                // Run entity substitution classifier to determine
                                // conflict type and generate a specific reason
                                let (conflict_type, substitution_desc) =
                                    classify_entity_substitution(&*db.conn(), text_a, text_b);

                                let reason = match substitution_desc {
                                    Some(ref desc) => format!(
                                        "Memories sharing entity '{}' contradict via {}: \
                                         similarity={:.0}%, word_overlap={:.0}%",
                                        entity,
                                        desc,
                                        sim * 100.0,
                                        jaccard * 100.0
                                    ),
                                    None => format!(
                                        "Memories sharing entity '{}' may contradict: \
                                         similarity={:.0}%, word_overlap={:.0}%",
                                        entity,
                                        sim * 100.0,
                                        jaccard * 100.0
                                    ),
                                };

                                let conflict = create_conflict(
                                    db,
                                    &conflict_type,
                                    rid_a,
                                    rid_b,
                                    Some(entity),
                                    None,
                                    &reason,
                                )?;
                                conflicts.push(conflict);
                            }
                        }
                    }
                }
            }
        }
    }

    // Consolidation conflicts
    let mut seen_pairs = std::collections::HashSet::new();
    for (rid_a, rid_b, shared_source) in cm_rows {
        let pair = if rid_a < rid_b {
            (rid_a.clone(), rid_b.clone())
        } else {
            (rid_b.clone(), rid_a.clone())
        };
        if seen_pairs.contains(&pair) {
            continue;
        }
        seen_pairs.insert(pair);

        if !conflict_exists(db, &rid_a, &rid_b)? {
            let conflict = create_conflict(
                db,
                &ConflictType::Consolidation,
                &rid_a,
                &rid_b,
                None,
                None,
                &format!(
                    "Concurrent consolidation: both '{}' and '{}' consumed source '{}'",
                    rid_a, rid_b, shared_source
                ),
            )?;
            conflicts.push(conflict);
        }
    }

    Ok(conflicts)
}

// ── RFC 006 Phase 1: Claim-Aware Conflict Scanner ──

/// Reason codes for claim-based conflicts (RFC 006 Phase 1).
pub mod reason_codes {
    pub const SAME_SUBJECT_SAME_REL_DISTINCT_OBJECT: &str =
        "same_subject_same_relation_distinct_object";
    pub const OVERLAPPING_VALIDITY: &str = "overlapping_validity_windows";
    pub const MISSING_TEMPORAL: &str = "missing_temporal_qualifier";
    pub const POSSIBLE_SUCCESSION: &str = "possible_temporal_succession";
    /// A positive claim and a negative claim about the same (src, rel_type, dst)
    /// exist — someone asserted X and someone else denied X.
    pub const POLARITY_CONTRADICTION: &str = "polarity_contradiction";
    /// A claim has modality other than 'asserted' — reported, hypothetical, denied.
    /// Lower confidence but still relevant for conflict tracking.
    pub const MODALITY_MISMATCH: &str = "modality_mismatch";
}

/// Check if two time intervals overlap.
fn intervals_overlap(
    from_a: Option<f64>,
    to_a: Option<f64>,
    from_b: Option<f64>,
    to_b: Option<f64>,
) -> Option<bool> {
    // If both have at least from or to, we can check
    let start_a = from_a?;
    let start_b = from_b?;
    let end_a = to_a.unwrap_or(f64::MAX);
    let end_b = to_b.unwrap_or(f64::MAX);
    Some(start_a < end_b && start_b < end_a)
}

/// Scan claims (extended edges) for scoped conflicts using RFC 006 logic.
///
/// Unlike the existing entity-based scanner, this operates on structured
/// claims with polarity, modality, and temporal qualifiers. It produces
/// conflicts with severity bands and reason codes.
///
/// Filtering:
/// - Only positive polarity (polarity = 1)
/// - Only asserted or reported modality
/// - Groups by (resolved_src, rel_type) to find distinct dst values
/// - Uses entity aliases for normalization
pub fn scan_claim_conflicts(db: &YantrikDB, max_conflicts: usize) -> Result<Vec<Conflict>> {
    let mut conflicts = Vec::new();

    // Phase 1: Query claim groups (same src + rel_type, different dst)
    // Only positive, asserted/reported claims participate
    let candidates: Vec<(
        String,
        String,
        String,
        String,
        Option<f64>,
        Option<f64>,
        Option<f64>,
        Option<f64>,
        String,
    )>;
    {
        let conn = db.conn();

        // Find all (src, rel_type) pairs with multiple distinct dst values
        // among positive, asserted claims
        let mut stmt = conn.prepare(
            "SELECT e1.src, e1.rel_type, e1.dst, e2.dst,
                    e1.valid_from, e1.valid_to, e2.valid_from, e2.valid_to,
                    e1.namespace
             FROM edges e1
             JOIN edges e2
               ON e1.src = e2.src
              AND e1.rel_type = e2.rel_type
              AND e1.dst < e2.dst
              AND e1.namespace = e2.namespace
             WHERE e1.tombstoned = 0 AND e2.tombstoned = 0
               AND e1.polarity = 1 AND e2.polarity = 1
               AND e1.modality IN ('asserted', 'reported')
               AND e2.modality IN ('asserted', 'reported')
             ORDER BY e1.created_at DESC
             LIMIT ?1",
        )?;

        candidates = stmt
            .query_map(params![max_conflicts * 5], |row| {
                Ok((
                    row.get::<_, String>(0)?,      // src
                    row.get::<_, String>(1)?,      // rel_type
                    row.get::<_, String>(2)?,      // dst1
                    row.get::<_, String>(3)?,      // dst2
                    row.get::<_, Option<f64>>(4)?, // valid_from_1
                    row.get::<_, Option<f64>>(5)?, // valid_to_1
                    row.get::<_, Option<f64>>(6)?, // valid_from_2
                    row.get::<_, Option<f64>>(7)?, // valid_to_2
                    row.get::<_, String>(8)?,      // namespace
                ))
            })?
            .collect::<std::result::Result<Vec<_>, _>>()?;
    } // conn released

    // Phase 2: Evaluate each candidate pair with policy awareness
    for (src, rel_type, dst1, dst2, vf1, vt1, vf2, vt2, namespace) in &candidates {
        if conflicts.len() >= max_conflicts {
            break;
        }

        // RFC 006 Phase 3: check relation policy before flagging.
        // Look up namespace-specific policy first, then global '*'.
        let policy: Option<(bool, bool, String)> = {
            let conn = db.conn();
            conn.query_row(
                "SELECT overlap_allowed, temporal_required, missing_time_severity \
                 FROM relation_policies \
                 WHERE relation_type = ?1 AND (namespace = ?2 OR namespace = '*') \
                 ORDER BY CASE WHEN namespace = ?2 THEN 0 ELSE 1 END \
                 LIMIT 1",
                params![rel_type, namespace],
                |row| {
                    Ok((
                        row.get::<_, bool>(0)?,
                        row.get::<_, bool>(1)?,
                        row.get::<_, String>(2)?,
                    ))
                },
            )
            .ok()
        };

        // If policy says overlap is allowed (e.g., works_at, speaks), skip
        if let Some((overlap_allowed, _, _)) = &policy {
            if *overlap_allowed {
                continue; // Multiple values are normal for this relation
            }
        }

        // Resolve aliases
        let src_canonical = db.resolve_alias(src, namespace);
        let dst1_canonical = db.resolve_alias(dst1, namespace);
        let dst2_canonical = db.resolve_alias(dst2, namespace);

        if dst1_canonical == dst2_canonical {
            continue;
        }

        // Determine severity — use policy's missing_time_severity if available
        let policy_missing_severity = policy
            .as_ref()
            .map(|(_, _, s)| s.as_str())
            .unwrap_or("medium");
        let policy_temporal_required = policy.as_ref().map(|(_, t, _)| *t).unwrap_or(false);

        let mut reason_codes =
            vec![reason_codes::SAME_SUBJECT_SAME_REL_DISTINCT_OBJECT.to_string()];
        let priority;

        match intervals_overlap(*vf1, *vt1, *vf2, *vt2) {
            Some(true) => {
                // Both have time, and they overlap → real conflict
                reason_codes.push(reason_codes::OVERLAPPING_VALIDITY.to_string());
                priority = "high";
            }
            Some(false) => {
                // Both have time, but they DON'T overlap → succession, not conflict.
                // Mark the OLDER claim as superseded (non-destructive: just a metadata update).
                let (older_src, older_dst, newer_vf) = if vf1.unwrap_or(0.0) < vf2.unwrap_or(0.0) {
                    (dst1.as_str(), dst2.as_str(), vf2)
                } else {
                    (dst2.as_str(), dst1.as_str(), vf1)
                };
                // Set valid_to on the older claim to mark it as historical
                {
                    let conn = db.conn();
                    let _ = conn.execute(
                        "UPDATE claims SET valid_to = ?1 \
                         WHERE src = ?2 AND rel_type = ?3 AND dst = ?4 AND valid_to IS NULL AND tombstoned = 0",
                        params![newer_vf.unwrap_or(0.0), src, rel_type, older_src],
                    );
                }
                continue; // Not a conflict — temporal succession handled
            }
            None => {
                // One or both missing time → severity from policy (default medium)
                reason_codes.push(reason_codes::MISSING_TEMPORAL.to_string());
                // If policy requires temporal evidence and it's missing, downgrade
                if policy_temporal_required {
                    priority = policy_missing_severity;
                } else {
                    priority = "medium";
                }
            }
        }

        let reason = format!(
            "Claim conflict: {} has different {} values: '{}' vs '{}'. Reasons: [{}]",
            src_canonical,
            rel_type,
            dst1_canonical,
            dst2_canonical,
            reason_codes.join(", ")
        );

        // Find source memory rids for the conflicting claims
        let (mem_a, mem_b) = {
            let conn = db.conn();
            let a = conn.query_row(
                "SELECT source_memory_rid FROM edges WHERE src = ?1 AND rel_type = ?2 AND dst = ?3 AND tombstoned = 0",
                params![src, rel_type, dst1],
                |row| row.get::<_, Option<String>>(0),
            ).ok().flatten();
            let b = conn.query_row(
                "SELECT source_memory_rid FROM edges WHERE src = ?1 AND rel_type = ?2 AND dst = ?3 AND tombstoned = 0",
                params![src, rel_type, dst2],
                |row| row.get::<_, Option<String>>(0),
            ).ok().flatten();
            (a, b)
        };

        // Use source memory rids if available, otherwise use src entity as fallback
        let rid_a = mem_a.unwrap_or_else(|| format!("claim:{}:{}:{}", src, rel_type, dst1));
        let rid_b = mem_b.unwrap_or_else(|| format!("claim:{}:{}:{}", src, rel_type, dst2));

        if !conflict_exists(db, &rid_a, &rid_b)? {
            let conflict = create_conflict(
                db,
                &ConflictType::IdentityFact,
                &rid_a,
                &rid_b,
                Some(&src_canonical),
                Some(rel_type),
                &reason,
            )?;

            // Override priority based on our temporal analysis
            {
                let conn = db.conn();
                conn.execute(
                    "UPDATE conflicts SET priority = ?1 WHERE conflict_id = ?2",
                    params![priority, conflict.conflict_id],
                )?;
            }

            conflicts.push(conflict);
        }
    }

    // RFC 006 Phase 4: polarity contradiction scan.
    // Find cases where the SAME (src, rel_type, dst) has both positive and
    // negative polarity claims — someone asserted X and someone denied X.
    if conflicts.len() < max_conflicts {
        let polarity_candidates: Vec<(
            String,
            String,
            String,
            String,
            Option<String>,
            Option<String>,
        )>;
        {
            let conn = db.conn();
            let mut stmt = conn.prepare(
                "SELECT e1.src, e1.rel_type, e1.dst, e1.namespace,
                        e1.source_memory_rid, e2.source_memory_rid
                 FROM edges e1
                 JOIN edges e2
                   ON e1.src = e2.src AND e1.rel_type = e2.rel_type AND e1.dst = e2.dst
                   AND e1.namespace = e2.namespace
                 WHERE e1.polarity = 1 AND e2.polarity = -1
                   AND e1.tombstoned = 0 AND e2.tombstoned = 0
                 LIMIT ?1",
            )?;
            let rows: Vec<_> = stmt
                .query_map(params![max_conflicts - conflicts.len()], |row| {
                    Ok((
                        row.get::<_, String>(0)?,
                        row.get::<_, String>(1)?,
                        row.get::<_, String>(2)?,
                        row.get::<_, String>(3)?,
                        row.get::<_, Option<String>>(4)?,
                        row.get::<_, Option<String>>(5)?,
                    ))
                })?
                .collect::<std::result::Result<Vec<_>, _>>()?;
            polarity_candidates = rows;
        }

        for (src, rel_type, dst, _ns, mem_a, mem_b) in &polarity_candidates {
            let rid_a = mem_a.as_deref().unwrap_or("unknown");
            let rid_b = mem_b.as_deref().unwrap_or("unknown");
            if rid_a != "unknown" && rid_b != "unknown" && !conflict_exists(db, rid_a, rid_b)? {
                let reason = format!(
                    "Polarity contradiction: '{}' has both positive and negative claims for {} → {}. Reasons: [{}]",
                    src, rel_type, dst, reason_codes::POLARITY_CONTRADICTION
                );
                let conflict = create_conflict(
                    db,
                    &ConflictType::IdentityFact,
                    rid_a,
                    rid_b,
                    Some(src),
                    Some(rel_type),
                    &reason,
                )?;
                // Polarity contradictions are always high priority
                {
                    let conn = db.conn();
                    conn.execute(
                        "UPDATE conflicts SET priority = 'high' WHERE conflict_id = ?1",
                        params![conflict.conflict_id],
                    )?;
                }
                conflicts.push(conflict);
            }
        }
    }

    Ok(conflicts)
}

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

    fn vec_seed(seed: f32, dim: usize) -> Vec<f32> {
        let raw: Vec<f32> = (0..dim).map(|i| (seed + i as f32) * 0.1).collect();
        let norm: f32 = raw.iter().map(|x| x * x).sum::<f32>().sqrt();
        raw.iter().map(|x| x / norm).collect()
    }

    fn empty_meta() -> serde_json::Value {
        serde_json::json!({})
    }

    #[test]
    fn test_create_conflict() {
        let db = YantrikDB::new(":memory:", 8).unwrap();
        let rid_a = db
            .record(
                "User likes coffee",
                "episodic",
                0.5,
                0.0,
                604800.0,
                &empty_meta(),
                &vec_seed(1.0, 8),
                "default",
                0.8,
                "general",
                "user",
                None,
            )
            .unwrap();
        let rid_b = db
            .record(
                "User likes tea",
                "episodic",
                0.5,
                0.0,
                604800.0,
                &empty_meta(),
                &vec_seed(2.0, 8),
                "default",
                0.8,
                "general",
                "user",
                None,
            )
            .unwrap();

        let conflict = create_conflict(
            &db,
            &ConflictType::Preference,
            &rid_a,
            &rid_b,
            Some("User"),
            Some("prefers"),
            "User has conflicting preference: coffee vs tea",
        )
        .unwrap();

        assert_eq!(conflict.status, "open");
        assert_eq!(conflict.conflict_type, "preference");
        assert_eq!(conflict.priority, "high");
        assert_eq!(conflict.memory_a, rid_a);
        assert_eq!(conflict.memory_b, rid_b);
    }

    #[test]
    fn test_conflict_dedup() {
        let db = YantrikDB::new(":memory:", 8).unwrap();
        let rid_a = db
            .record(
                "a",
                "episodic",
                0.5,
                0.0,
                604800.0,
                &empty_meta(),
                &vec_seed(1.0, 8),
                "default",
                0.8,
                "general",
                "user",
                None,
            )
            .unwrap();
        let rid_b = db
            .record(
                "b",
                "episodic",
                0.5,
                0.0,
                604800.0,
                &empty_meta(),
                &vec_seed(2.0, 8),
                "default",
                0.8,
                "general",
                "user",
                None,
            )
            .unwrap();

        assert!(!conflict_exists(&db, &rid_a, &rid_b).unwrap());
        create_conflict(
            &db,
            &ConflictType::Minor,
            &rid_a,
            &rid_b,
            None,
            None,
            "test",
        )
        .unwrap();
        assert!(conflict_exists(&db, &rid_a, &rid_b).unwrap());
        assert!(conflict_exists(&db, &rid_b, &rid_a).unwrap()); // reversed order
    }

    #[test]
    fn test_classify_conflict() {
        assert_eq!(classify_conflict("birthday"), ConflictType::IdentityFact);
        assert_eq!(classify_conflict("works_at"), ConflictType::IdentityFact);
        assert_eq!(classify_conflict("favorite"), ConflictType::Preference);
        assert_eq!(classify_conflict("prefers"), ConflictType::Preference);
        assert_eq!(classify_conflict("random_rel"), ConflictType::Minor);
    }

    #[test]
    fn test_scan_contradicting_edges() {
        let db = YantrikDB::new(":memory:", 8).unwrap();
        db.relate("User", "Google", "works_at", 1.0).unwrap();
        db.relate("User", "Meta", "works_at", 1.0).unwrap();

        let conflicts = scan_conflicts(&db).unwrap();
        assert!(!conflicts.is_empty());
        assert_eq!(conflicts[0].conflict_type, "identity_fact");
        assert_eq!(conflicts[0].entity.as_deref(), Some("User"));
    }

    #[test]
    fn test_scan_no_conflict_for_non_identity_edges() {
        let db = YantrikDB::new(":memory:", 8).unwrap();
        db.relate("User", "Alice", "friends_with", 1.0).unwrap();
        db.relate("User", "Bob", "friends_with", 1.0).unwrap();

        let conflicts = scan_conflicts(&db).unwrap();
        assert!(conflicts.is_empty());
    }

    #[test]
    fn test_conflict_type_default_priorities() {
        assert_eq!(ConflictType::IdentityFact.default_priority(), "critical");
        assert_eq!(ConflictType::Preference.default_priority(), "high");
        assert_eq!(ConflictType::Temporal.default_priority(), "high");
        assert_eq!(ConflictType::Consolidation.default_priority(), "medium");
        assert_eq!(ConflictType::Minor.default_priority(), "low");
    }
}