microclaw_engine/

memory_service.rs

use std::sync::Arc;

use anyhow::Result;
use tracing::info;

use crate::agent_engine::is_slash_command_text;
use crate::embedding::EmbeddingProvider;
use crate::internal::storage::db::{call_blocking, Database, KgTriple, Memory};
use crate::internal::storage::memory_quality;
use crate::memory_backend::MemoryBackend;
use crate::runtime::AppState;

pub struct ReflectorApplyOutcome {
    pub inserted: usize,
    pub updated: usize,
    pub skipped: usize,
    pub dedup_method: &'static str,
}

fn jaccard_similarity_ratio(a: &str, b: &str) -> f64 {
    use std::collections::HashSet;
    let a_words: HashSet<&str> = a.split_whitespace().collect();
    let b_words: HashSet<&str> = b.split_whitespace().collect();
    let intersection = a_words.intersection(&b_words).count();
    let union = a_words.len() + b_words.len() - intersection;
    if union == 0 {
        1.0
    } else {
        intersection as f64 / union as f64
    }
}

pub(crate) fn tokenize_for_relevance(text: &str) -> std::collections::HashSet<String> {
    let mut out = std::collections::HashSet::new();

    for token in text
        .split_whitespace()
        .map(|w| {
            w.chars()
                .filter(|c| c.is_alphanumeric())
                .collect::<String>()
                .to_lowercase()
        })
        .filter(|w| w.len() > 1)
    {
        out.insert(token);
    }

    let cjk_chars: Vec<char> = text.chars().filter(|c| is_cjk(*c)).collect();
    if cjk_chars.len() >= 2 {
        for pair in cjk_chars.windows(2) {
            let gram: String = pair.iter().collect();
            out.insert(gram);
        }
    } else if cjk_chars.len() == 1 {
        out.insert(cjk_chars[0].to_string());
    }

    out
}

fn score_relevance_with_cache(
    content: &str,
    query_tokens: &std::collections::HashSet<String>,
) -> usize {
    if query_tokens.is_empty() {
        return 0;
    }
    let content_tokens = tokenize_for_relevance(content);
    content_tokens
        .iter()
        .filter(|t| query_tokens.contains(*t))
        .count()
}

fn is_cjk(c: char) -> bool {
    matches!(
        c as u32,
        0x4E00..=0x9FFF
            | 0x3400..=0x4DBF
            | 0x20000..=0x2A6DF
            | 0x2A700..=0x2B73F
            | 0x2B740..=0x2B81F
            | 0x2B820..=0x2CEAF
            | 0xF900..=0xFAFF
            | 0x2F800..=0x2FA1F
    )
}

pub fn jaccard_similar(a: &str, b: &str, threshold: f64) -> bool {
    use std::collections::HashSet;
    let a_words: HashSet<&str> = a.split_whitespace().collect();
    let b_words: HashSet<&str> = b.split_whitespace().collect();
    let intersection = a_words.intersection(&b_words).count();
    let union = a_words.len() + b_words.len() - intersection;
    if union == 0 {
        return true;
    }
    intersection as f64 / union as f64 >= threshold
}

/// A minimal view of a memory for sleep-time consolidation.
#[derive(Debug, Clone)]
pub struct ConsolidationItem {
    pub id: i64,
    pub content: String,
    pub category: String,
}

/// Select near-duplicate memories to archive during a sleep-time consolidation pass.
///
/// `items` must be pre-sorted **best-first** (e.g. highest confidence / most recent
/// first): the first member of each duplicate group is kept and later same-category
/// near-duplicates are archived. PROFILE memories are always kept (they describe the
/// user's identity). Returns the ids to archive, capped at `max`.
pub fn select_duplicate_memories_to_archive(
    items: &[ConsolidationItem],
    threshold: f64,
    max: usize,
) -> Vec<i64> {
    let threshold = threshold.clamp(0.5, 1.0);
    let mut kept: Vec<&ConsolidationItem> = Vec::new();
    let mut archive: Vec<i64> = Vec::new();
    for it in items {
        if it.category == "PROFILE" {
            kept.push(it);
            continue;
        }
        let is_dup = kept.iter().any(|k| {
            k.category == it.category && jaccard_similar(&k.content, &it.content, threshold)
        });
        if is_dup {
            archive.push(it.id);
            if archive.len() >= max {
                break;
            }
        } else {
            kept.push(it);
        }
    }
    archive
}

fn should_merge_duplicate(
    existing: &Memory,
    incoming_content: &str,
    incoming_category: &str,
) -> bool {
    if existing.is_archived {
        return true;
    }
    if existing.content.eq_ignore_ascii_case(incoming_content) {
        return false;
    }
    if incoming_category == "PROFILE" && existing.category != "PROFILE" {
        return true;
    }
    incoming_content.len() > existing.content.len() + 8
}

fn is_corrective_action_item(content: &str) -> bool {
    let lower = content.to_ascii_lowercase();
    let trimmed = lower.trim();
    trimmed.starts_with("todo:")
        || trimmed.starts_with("todo ")
        || trimmed.contains(" ensure ")
        || trimmed.starts_with("ensure ")
}

fn looks_like_broken_behavior_fact(content: &str) -> bool {
    let lower = content.to_ascii_lowercase();
    let broken_cues = [
        "tool calls were broken",
        "typed tool calls as text",
        "posted as text",
        "authentication error",
        "auth fails",
        "not following instructions",
        "isn't following instructions",
        "failed",
        "broke ",
        "was broken",
        "error on",
    ];
    broken_cues.iter().any(|cue| lower.contains(cue))
}

pub fn should_skip_memory_poisoning_risk(content: &str) -> bool {
    looks_like_broken_behavior_fact(content) && !is_corrective_action_item(content)
}

#[cfg(feature = "sqlite-vec")]
pub async fn upsert_memory_embedding(
    state: &Arc<AppState>,
    memory_id: i64,
    content: &str,
) -> Result<(), ()> {
    let provider = match &state.embedding {
        Some(p) => p,
        None => return Ok(()),
    };
    upsert_memory_embedding_with_provider(state.db.clone(), provider, memory_id, content).await
}

#[cfg(feature = "sqlite-vec")]
async fn upsert_memory_embedding_with_provider(
    db: Arc<Database>,
    provider: &Arc<dyn EmbeddingProvider>,
    memory_id: i64,
    content: &str,
) -> Result<(), ()> {
    let model_name = provider.model().to_string();
    let embedding = provider.embed(content).await.map_err(|_| ())?;
    call_blocking(db, move |db| {
        db.upsert_memory_vec(memory_id, &embedding)?;
        db.update_memory_embedding_model(memory_id, &model_name)?;
        Ok(())
    })
    .await
    .map_err(|_| ())
}

pub(crate) async fn maybe_handle_explicit_memory_command(
    state: &AppState,
    chat_id: i64,
    override_prompt: Option<&str>,
    image_data: Option<(String, String)>,
) -> Result<Option<String>> {
    if override_prompt.is_some() || image_data.is_some() {
        return Ok(None);
    }

    let latest_user = call_blocking(state.db.clone(), move |db| {
        db.get_recent_messages(chat_id, 10)
    })
    .await?;
    let Some(last_user_text) = latest_user
        .into_iter()
        .rev()
        .find(|m| !m.is_from_bot && !is_slash_command_text(&m.content))
        .map(|m| m.content)
    else {
        return Ok(None);
    };

    let Some(explicit_content) = memory_quality::extract_explicit_memory_command(&last_user_text)
    else {
        return Ok(None);
    };
    if !memory_quality::memory_quality_ok(&explicit_content) {
        return Ok(Some(
            "I skipped saving that memory because it looked too vague. Please send a specific fact.".to_string(),
        ));
    }

    let existing = state
        .memory_backend
        .get_all_memories_for_chat(Some(chat_id))
        .await?;
    let explicit_topic = memory_quality::memory_topic_key(&explicit_content);
    if let Some(dup) = existing.iter().find(|m| {
        !m.is_archived
            && (m.content.eq_ignore_ascii_case(&explicit_content)
                || jaccard_similarity_ratio(&m.content, &explicit_content) >= 0.55)
    }) {
        let memory_id = dup.id;
        let content_for_update = explicit_content.clone();
        let _ = state
            .memory_backend
            .update_memory_with_metadata(
                memory_id,
                &content_for_update,
                "KNOWLEDGE",
                0.95,
                "explicit",
            )
            .await;
        return Ok(Some(format!(
            "Noted. Updated memory #{memory_id}: {explicit_content}"
        )));
    }

    if let Some(conflict) = existing.iter().find(|m| {
        !m.is_archived
            && m.category == "KNOWLEDGE"
            && memory_quality::memory_topic_key(&m.content) == explicit_topic
            && !m.content.eq_ignore_ascii_case(&explicit_content)
    }) {
        let from_id = conflict.id;
        let new_content = explicit_content.clone();
        let superseded_id = state
            .memory_backend
            .supersede_memory(
                from_id,
                &new_content,
                "KNOWLEDGE",
                "explicit_conflict",
                0.95,
                Some("explicit_topic_conflict"),
            )
            .await?;
        return Ok(Some(format!(
            "Noted. Superseded memory #{from_id} with #{superseded_id}: {explicit_content}"
        )));
    }

    let content_for_insert = explicit_content.clone();
    let inserted_id = state
        .memory_backend
        .insert_memory_with_metadata(
            Some(chat_id),
            &content_for_insert,
            "KNOWLEDGE",
            "explicit",
            0.95,
        )
        .await?;

    #[cfg(feature = "sqlite-vec")]
    {
        if let Some(provider) = &state.embedding {
            let _ = upsert_memory_embedding_with_provider(
                state.db.clone(),
                provider,
                inserted_id,
                &explicit_content,
            )
            .await;
        }
    }

    Ok(Some(format!(
        "Noted. Saved memory #{inserted_id}: {explicit_content}"
    )))
}

/// Sanitize a query string for memory retrieval.
///
/// AI agents sometimes prepend system prompts or tool definitions to user messages,
/// which destroys embedding quality for semantic search. This sanitizer extracts the
/// likely user intent by:
/// Effective ranking score for a memory: `confidence` × recency-decay.
/// PROFILE memories are exempted (they describe the user, not transient state)
/// and a non-positive `half_life_days` disables decay entirely.
fn effective_memory_score(
    m: &Memory,
    now: chrono::DateTime<chrono::Utc>,
    half_life_days: f64,
) -> f64 {
    if m.category == "PROFILE" || half_life_days <= 0.0 {
        return m.confidence;
    }
    let parsed = chrono::DateTime::parse_from_rfc3339(&m.last_seen_at)
        .or_else(|_| chrono::DateTime::parse_from_rfc3339(&m.updated_at));
    let last_seen = match parsed {
        Ok(t) => t.with_timezone(&chrono::Utc),
        Err(_) => return m.confidence,
    };
    let age_days = (now - last_seen).num_seconds().max(0) as f64 / 86_400.0;
    let decay = 0.5_f64.powf(age_days / half_life_days);
    m.confidence * decay
}

/// 1. Detecting and stripping common system prompt patterns
/// 2. Extracting the last meaningful sentence (most likely to be the actual query)
/// 3. Truncating overly long queries that are likely contaminated
fn sanitize_memory_query(raw: &str) -> String {
    let trimmed = raw.trim();
    if trimmed.is_empty() || trimmed.len() < 200 {
        // Short queries are unlikely to contain system prompt contamination
        return trimmed.to_string();
    }

    // Check for system prompt contamination markers
    let lower = trimmed.to_ascii_lowercase();
    let contamination_markers = [
        "you are a",
        "you are an",
        "your role is",
        "system prompt",
        "instructions:",
        "<system>",
        "tool_use",
        "tool_result",
        "[scheduler]:",
        "as an ai assistant",
    ];
    let is_contaminated = contamination_markers.iter().any(|m| lower.contains(m));

    if !is_contaminated {
        // Not contaminated, but still truncate if very long
        if trimmed.len() > 500 {
            return trimmed.chars().take(500).collect();
        }
        return trimmed.to_string();
    }

    // Strategy: extract the last meaningful sentence (tail extraction)
    // Split on sentence boundaries and take the last non-trivial one
    let sentences: Vec<&str> = trimmed
        .split(['.', '?', '!', '\n'])
        .map(|s| s.trim())
        .filter(|s| s.len() > 10)
        .collect();

    if let Some(last) = sentences.last() {
        return last.chars().take(300).collect();
    }

    // Fallback: take the last 200 chars
    let start = trimmed
        .char_indices()
        .rev()
        .nth(199)
        .map(|(i, _)| i)
        .unwrap_or(0);
    trimmed[start..].to_string()
}

/// Find which knowledge-graph entities a query mentions, to seed graph-augmented
/// retrieval. Uses case-insensitive substring matching (so multi-word entities
/// like "New York" are caught), prefers longer entities first (`entities` is
/// pre-sorted longest-first by the storage layer), and ignores 1-2 char entities
/// to avoid noise. Bounded to `max_seeds`.
fn extract_kg_seeds(query: &str, entities: &[String], max_seeds: usize) -> Vec<String> {
    let q = query.to_lowercase();
    if q.trim().is_empty() {
        return Vec::new();
    }
    let mut seeds: Vec<String> = Vec::new();
    let mut seen: std::collections::HashSet<String> = std::collections::HashSet::new();
    for ent in entities {
        if seeds.len() >= max_seeds {
            break;
        }
        let lc = ent.to_lowercase();
        if lc.chars().count() < 3 {
            continue;
        }
        if q.contains(&lc) && seen.insert(lc) {
            seeds.push(ent.clone());
        }
    }
    seeds
}

/// Render a connected-facts block from graph-expanded triples, skipping triples
/// whose meaning is already covered by an injected memory line so the block adds
/// genuinely new, multi-hop context rather than repeating L0-L2.
fn render_graph_section(triples: &[KgTriple], already_injected: &str) -> String {
    let injected_lc = already_injected.to_lowercase();
    let mut lines = String::new();
    for t in triples {
        // Cheap redundancy guard: if both endpoints already appear together in an
        // injected memory line, the relationship is probably already stated.
        let subj_lc = t.subject.to_lowercase();
        let obj_lc = t.object.to_lowercase();
        if injected_lc.contains(&subj_lc) && injected_lc.contains(&obj_lc) {
            continue;
        }
        lines.push_str(&format!("{} —[{}]→ {}\n", t.subject, t.predicate, t.object));
    }
    lines
}

/// Build structured memory context using a 4-layer memory stack:
///
/// - **L0 (Identity)**: PROFILE memories — always loaded first. These define who the user is.
///   Budget: up to 20% of total. Cost: ~100-200 tokens typically.
/// - **L1 (Essential)**: Highest-confidence, most-recently-seen memories across all categories.
///   These are the "essential story" — durable facts the agent should always know.
///   Budget: up to 30% of total. Cost: ~300-500 tokens.
/// - **L2 (Relevance)**: Query-relevant memories via semantic/keyword ranking.
///   Loaded based on what the user is currently asking about.
///   Budget: remaining tokens. Cost: variable.
/// - **Connected**: Graph-augmented facts reached by expanding the temporal
///   knowledge graph from entities the query mentions (1-2 hops). Surfaces
///   multi-hop context the flat memory layers miss. Bounded and query-gated.
/// - **L3 (Deep Search)**: Not injected here — available via `structured_memory_search` tool
///   for on-demand deep retrieval when the agent needs more context.
#[expect(clippy::too_many_arguments)]
pub(crate) async fn build_db_memory_context(
    memory_backend: &Arc<MemoryBackend>,
    db: &Arc<Database>,
    embedding: Option<&Arc<dyn EmbeddingProvider>>,
    chat_id: i64,
    query: &str,
    token_budget: usize,
    l0_identity_pct: usize,
    l1_essential_pct: usize,
    recency_half_life_days: f64,
    graph_recall_enabled: bool,
    graph_max_hops: usize,
    graph_max_triples: usize,
) -> String {
    let query = &sanitize_memory_query(query);
    let memories = match memory_backend.get_memories_for_context(chat_id, 100).await {
        Ok(m) => m,
        Err(_) => return String::new(),
    };

    if memories.is_empty() {
        return String::new();
    }

    let budget = token_budget.max(1);
    // Clamp percentages to sane range; remaining goes to L2 relevance
    let l0_pct = l0_identity_pct.min(50);
    let l1_pct = l1_essential_pct.min(50);
    let mut used_tokens = 0usize;
    let mut out = String::from("<structured_memories>\n");
    let mut injected_ids: std::collections::HashSet<i64> = std::collections::HashSet::new();

    // ── L0: Identity (PROFILE memories, up to l0_pct% of budget) ──
    let l0_budget = budget * l0_pct / 100;
    let mut profile_memories: Vec<&Memory> = memories
        .iter()
        .filter(|m| m.category == "PROFILE" && !m.is_archived)
        .collect();
    // Sort profiles by confidence desc, then recency
    profile_memories.sort_by(|a, b| {
        b.confidence
            .partial_cmp(&a.confidence)
            .unwrap_or(std::cmp::Ordering::Equal)
    });

    if !profile_memories.is_empty() {
        out.push_str("# Identity\n");
        for m in &profile_memories {
            let est = (m.content.len() / 4) + 10;
            if used_tokens + est > l0_budget && !injected_ids.is_empty() {
                break;
            }
            used_tokens += est;
            injected_ids.insert(m.id);
            let scope = if m.chat_id.is_none() {
                "global"
            } else {
                "chat"
            };
            out.push_str(&format!("[PROFILE] [{}] {}\n", scope, m.content));
        }
    }

    // ── L1: Essential Story (highest-confidence memories, up to l1_pct% of budget) ──
    let l1_budget = used_tokens + (budget * l1_pct / 100);
    let now = chrono::Utc::now();
    let mut essential: Vec<&Memory> = memories
        .iter()
        .filter(|m| !m.is_archived && !injected_ids.contains(&m.id))
        .collect();
    // Score by confidence × recency-decay (PROFILE memories don't decay; everyone
    // else loses half their weight per `recency_half_life_days`). The decay
    // pulls stale tactical facts down so durable knowledge floats.
    essential.sort_by(|a, b| {
        let sa = effective_memory_score(a, now, recency_half_life_days);
        let sb = effective_memory_score(b, now, recency_half_life_days);
        sb.partial_cmp(&sa).unwrap_or(std::cmp::Ordering::Equal)
    });

    let mut l1_count = 0usize;
    if !essential.is_empty() {
        out.push_str("# Essential\n");
        for m in &essential {
            let est = (m.content.len() / 4) + 10;
            if used_tokens + est > l1_budget {
                break;
            }
            used_tokens += est;
            injected_ids.insert(m.id);
            l1_count += 1;
            let scope = if m.chat_id.is_none() {
                "global"
            } else {
                "chat"
            };
            out.push_str(&format!("[{}] [{}] {}\n", m.category, scope, m.content));
        }
    }

    // ── L2: Relevance-ranked (query-dependent, fills remaining budget) ──
    // Build relevance-ordered list from memories not yet injected
    let remaining: Vec<&Memory> = memories
        .iter()
        .filter(|m| !injected_ids.contains(&m.id) && !m.is_archived)
        .collect();

    if !remaining.is_empty() {
        let mut relevance_ordered: Vec<&Memory> = Vec::new();

        #[cfg(feature = "sqlite-vec")]
        let mut retrieval_method = if memory_supports_local_semantic_ranking(memory_backend) {
            "keyword"
        } else {
            "provider"
        };
        #[cfg(not(feature = "sqlite-vec"))]
        let retrieval_method = "keyword";

        #[cfg(feature = "sqlite-vec")]
        {
            if let Some(provider) = embedding {
                if memory_supports_local_semantic_ranking(memory_backend)
                    && !query.trim().is_empty()
                {
                    if let Ok(query_vec) = provider.embed(query).await {
                        let knn_result = call_blocking(db.clone(), move |db| {
                            db.knn_memories(chat_id, &query_vec, 20)
                        })
                        .await;
                        if let Ok(knn_rows) = knn_result {
                            let by_id: std::collections::HashMap<i64, &&Memory> =
                                remaining.iter().map(|m| (m.id, m)).collect();
                            for (id, _) in knn_rows {
                                if let Some(mem) = by_id.get(&id) {
                                    relevance_ordered.push(**mem);
                                }
                            }
                            if !relevance_ordered.is_empty() {
                                retrieval_method = "knn";
                            }
                        }
                    }
                }
            }
        }

        #[cfg(not(feature = "sqlite-vec"))]
        {
            let _ = embedding;
        }

        if relevance_ordered.is_empty() {
            let query_tokens = tokenize_for_relevance(query);
            let mut scored: Vec<(usize, usize, &&Memory)> = remaining
                .iter()
                .enumerate()
                .map(|(idx, m)| {
                    (
                        score_relevance_with_cache(&m.content, &query_tokens),
                        idx,
                        m,
                    )
                })
                .collect();
            if !query.is_empty() {
                scored.sort_by(|a, b| b.0.cmp(&a.0).then_with(|| a.1.cmp(&b.1)));
            }
            relevance_ordered = scored.into_iter().map(|(_, _, m)| *m).collect();
        }

        let mut l2_count = 0usize;
        let mut l2_omitted = 0usize;
        if !relevance_ordered.is_empty() {
            out.push_str("# Relevant\n");
        }
        for (idx, m) in relevance_ordered.iter().enumerate() {
            let est = (m.content.len() / 4) + 10;
            if used_tokens + est > budget {
                l2_omitted = relevance_ordered.len().saturating_sub(idx);
                break;
            }
            used_tokens += est;
            injected_ids.insert(m.id);
            l2_count += 1;
            let scope = if m.chat_id.is_none() {
                "global"
            } else {
                "chat"
            };
            out.push_str(&format!("[{}] [{}] {}\n", m.category, scope, m.content));
        }

        if l2_omitted > 0 {
            out.push_str(&format!(
                "(+{l2_omitted} memories available via structured_memory_search tool)\n"
            ));
        }

        let _ = retrieval_method;
        let _ = l1_count;
        let _ = l2_count;
    }

    // ── Connected: graph-augmented retrieval over the temporal knowledge graph ──
    // Seed from entities the query mentions, expand a couple of hops over the KG,
    // and surface connected facts the flat L0-L2 layers miss (multi-hop context).
    // Local-only: no embeddings, no LLM, bounded by triple count and token budget.
    if graph_recall_enabled && graph_max_triples > 0 && !query.trim().is_empty() {
        let q = query.to_string();
        let entities = call_blocking(db.clone(), move |d| {
            d.kg_distinct_entities(Some(chat_id), 500)
        })
        .await
        .unwrap_or_default();
        let seeds = extract_kg_seeds(&q, &entities, 8);
        if !seeds.is_empty() {
            let hops = graph_max_hops.clamp(1, 3);
            let max_triples = graph_max_triples;
            let triples = call_blocking(db.clone(), move |d| {
                d.kg_neighborhood(Some(chat_id), &seeds, hops, max_triples)
            })
            .await
            .unwrap_or_default();
            if !triples.is_empty() {
                let section = render_graph_section(&triples, &out);
                let est = (section.len() / 4) + 12;
                if !section.trim().is_empty() && used_tokens + est <= budget {
                    out.push_str("# Connected\n");
                    out.push_str(&section);
                    used_tokens += est;
                }
            }
        }
    }

    out.push_str("</structured_memories>\n");

    let candidate_count = memories.len();
    let selected_count = injected_ids.len();
    let omitted = candidate_count.saturating_sub(selected_count);
    let retrieval_method = "layered";
    let retrieval_method_owned = retrieval_method.to_string();
    let _ = call_blocking(db.clone(), move |d| {
        d.log_memory_injection(
            chat_id,
            &retrieval_method_owned,
            candidate_count,
            selected_count,
            omitted,
            used_tokens,
        )
        .map(|_| ())
    })
    .await;
    info!(
        "Memory injection (4-layer): chat {} -> {}/{} memories (L0:identity + L1:essential + L2:relevant), tokens_est={}, omitted={}",
        chat_id, selected_count, candidate_count, used_tokens, omitted
    );
    out
}

pub async fn apply_reflector_extractions(
    state: &Arc<AppState>,
    chat_id: i64,
    existing: &[Memory],
    extracted: &[serde_json::Value],
) -> ReflectorApplyOutcome {
    let mut inserted = 0usize;
    let mut updated = 0usize;
    let mut skipped = 0usize;
    #[cfg(feature = "sqlite-vec")]
    let dedup_method = if state.embedding.is_some() {
        "semantic"
    } else {
        "jaccard"
    };
    #[cfg(not(feature = "sqlite-vec"))]
    let dedup_method = "jaccard";

    let mut seen_contents: Vec<(i64, String)> =
        existing.iter().map(|m| (m.id, m.content.clone())).collect();
    let existing_by_id: std::collections::HashMap<i64, &Memory> =
        existing.iter().map(|m| (m.id, m)).collect();
    let mut topic_latest: std::collections::HashMap<String, i64> = existing
        .iter()
        .filter(|m| !m.is_archived)
        .map(|m| (memory_quality::memory_topic_key(&m.content), m.id))
        .collect();

    for item in extracted {
        let content = match item.get("content").and_then(|v| v.as_str()) {
            Some(s) => s,
            None => continue,
        };
        let category = item
            .get("category")
            .and_then(|v| v.as_str())
            .unwrap_or("KNOWLEDGE")
            .to_ascii_uppercase();
        if !matches!(category.as_str(), "PROFILE" | "KNOWLEDGE" | "EVENT") {
            continue;
        }
        let content = match memory_quality::normalize_memory_content(content, 180) {
            Some(c) => c,
            None => continue,
        };
        // Strip credentials / PII before the row is persisted, deduped by
        // topic key, or shipped to the embedding model. Reflector-extracted
        // memories quote conversation content verbatim; without this gate a
        // user pasting an API key into chat would land that key in
        // long-lived memory and any downstream embedding store.
        let content = microclaw_core::redact::redact(&content);
        if should_skip_memory_poisoning_risk(&content) {
            skipped += 1;
            continue;
        }
        if !memory_quality::memory_quality_ok(&content) {
            continue;
        }

        let supersedes_id = item.get("supersedes_id").and_then(|v| v.as_i64());
        if let Some(sid) = supersedes_id {
            if existing.iter().any(|m| m.id == sid) {
                let content = content.to_string();
                let category = category.to_string();
                let db_content = content.clone();
                if state
                    .memory_backend
                    .update_memory_with_metadata(sid, &db_content, &category, 0.78, "reflector")
                    .await
                    .is_ok()
                {
                    updated += 1;
                    #[cfg(feature = "sqlite-vec")]
                    {
                        let _ = upsert_memory_embedding(state, sid, &content).await;
                    }
                    seen_contents.push((sid, content));
                }
                continue;
            }
        }

        let topic_key = memory_quality::memory_topic_key(&content);
        if let Some(prev_id) = topic_latest.get(&topic_key).copied() {
            if let Some(prev) = existing_by_id.get(&prev_id) {
                if !prev.content.eq_ignore_ascii_case(&content)
                    && !jaccard_similar(&prev.content, &content, 0.85)
                {
                    let new_content = content.to_string();
                    let new_category = category.to_string();
                    if let Ok(new_id) = state
                        .memory_backend
                        .supersede_memory(
                            prev_id,
                            &new_content,
                            &new_category,
                            "reflector_conflict",
                            0.74,
                            Some("topic_conflict"),
                        )
                        .await
                    {
                        updated += 1;
                        #[cfg(feature = "sqlite-vec")]
                        {
                            let _ = upsert_memory_embedding(state, new_id, &content).await;
                        }
                        topic_latest.insert(topic_key, new_id);
                        seen_contents.push((new_id, content));
                        continue;
                    }
                }
            }
        }

        let duplicate_id = {
            #[cfg(feature = "sqlite-vec")]
            {
                if let Some(provider) = &state.embedding {
                    if let Ok(query_vec) = provider.embed(&content).await {
                        let nearest = call_blocking(state.db.clone(), move |db| {
                            db.knn_memories(chat_id, &query_vec, 1)
                        })
                        .await
                        .ok()
                        .and_then(|rows| rows.first().copied());
                        nearest.and_then(|(id, dist)| if dist < 0.15 { Some(id) } else { None })
                    } else {
                        seen_contents
                            .iter()
                            .find(|(_, existing)| jaccard_similar(existing, &content, 0.5))
                            .map(|(id, _)| *id)
                    }
                } else {
                    seen_contents
                        .iter()
                        .find(|(_, existing)| jaccard_similar(existing, &content, 0.5))
                        .map(|(id, _)| *id)
                }
            }
            #[cfg(not(feature = "sqlite-vec"))]
            {
                seen_contents
                    .iter()
                    .find(|(_, existing)| jaccard_similar(existing, &content, 0.5))
                    .map(|(id, _)| *id)
            }
        };
        if let Some(dup_id) = duplicate_id {
            if let Some(existing_mem) = existing_by_id.get(&dup_id) {
                if should_merge_duplicate(existing_mem, &content, &category) {
                    let update_content = content.to_string();
                    let update_category = category.to_string();
                    if state
                        .memory_backend
                        .update_memory_with_metadata(
                            dup_id,
                            &update_content,
                            &update_category,
                            0.70,
                            "reflector",
                        )
                        .await
                        .is_ok()
                    {
                        updated += 1;
                    } else {
                        skipped += 1;
                    }
                } else {
                    let _ = state
                        .memory_backend
                        .touch_memory_last_seen(dup_id, Some(0.55))
                        .await;
                    skipped += 1;
                }
            } else {
                skipped += 1;
            }
            continue;
        }

        let content = content.to_string();
        let db_content = content.clone();
        let category = category.to_string();
        let inserted_id = state
            .memory_backend
            .insert_memory_with_metadata(Some(chat_id), &db_content, &category, "reflector", 0.68)
            .await
            .ok();
        if let Some(memory_id) = inserted_id {
            inserted += 1;
            #[cfg(feature = "sqlite-vec")]
            {
                let _ = upsert_memory_embedding(state, memory_id, &content).await;
            }
            #[cfg(not(feature = "sqlite-vec"))]
            let _ = memory_id;
            seen_contents.push((memory_id, content));
            topic_latest.insert(topic_key, memory_id);
        }
    }

    ReflectorApplyOutcome {
        inserted,
        updated,
        skipped,
        dedup_method,
    }
}

#[cfg(feature = "sqlite-vec")]
pub(crate) fn memory_supports_local_semantic_ranking(memory_backend: &MemoryBackend) -> bool {
    memory_backend.supports_local_semantic_ranking()
}

#[cfg(test)]
mod recency_tests {
    use super::effective_memory_score;
    use crate::internal::storage::db::Memory;

    fn mk(category: &str, last_seen: &str, confidence: f64) -> Memory {
        Memory {
            id: 1,
            chat_id: Some(1),
            content: "x".into(),
            category: category.into(),
            created_at: last_seen.into(),
            updated_at: last_seen.into(),
            embedding_model: None,
            confidence,
            source: "test".into(),
            last_seen_at: last_seen.into(),
            is_archived: false,
            archived_at: None,
            expires_at: None,
        }
    }

    #[test]
    fn profile_memories_are_immune_to_decay() {
        let now = chrono::Utc::now();
        let stale = (now - chrono::Duration::days(365)).to_rfc3339();
        let m = mk("PROFILE", &stale, 0.9);
        assert!((effective_memory_score(&m, now, 30.0) - 0.9).abs() < 1e-9);
    }

    #[test]
    fn knowledge_memories_decay_by_half_per_half_life() {
        let now = chrono::Utc::now();
        let half_life = 30.0;
        let one_half = (now - chrono::Duration::days(30)).to_rfc3339();
        let m = mk("KNOWLEDGE", &one_half, 1.0);
        let s = effective_memory_score(&m, now, half_life);
        assert!((s - 0.5).abs() < 0.01, "expected ~0.5, got {s}");
    }

    #[test]
    fn zero_half_life_disables_decay() {
        let now = chrono::Utc::now();
        let stale = (now - chrono::Duration::days(365)).to_rfc3339();
        let m = mk("EVENT", &stale, 0.7);
        assert!((effective_memory_score(&m, now, 0.0) - 0.7).abs() < 1e-9);
    }
}

#[cfg(test)]
mod consolidation_tests {
    use super::{select_duplicate_memories_to_archive, ConsolidationItem};

    fn item(id: i64, content: &str, category: &str) -> ConsolidationItem {
        ConsolidationItem {
            id,
            content: content.into(),
            category: category.into(),
        }
    }

    #[test]
    fn archives_near_duplicate_keeping_first() {
        let items = vec![
            item(
                1,
                "user prefers concise replies and bullet points",
                "KNOWLEDGE",
            ),
            item(
                2,
                "user prefers concise replies and bullet points please",
                "KNOWLEDGE",
            ),
            item(3, "user lives in Berlin", "KNOWLEDGE"),
        ];
        let archived = select_duplicate_memories_to_archive(&items, 0.7, 20);
        assert_eq!(
            archived,
            vec![2],
            "only the later near-duplicate is archived"
        );
    }

    #[test]
    fn never_archives_profile() {
        let items = vec![
            item(1, "name is Alex", "PROFILE"),
            item(2, "name is Alex", "PROFILE"),
        ];
        let archived = select_duplicate_memories_to_archive(&items, 0.7, 20);
        assert!(archived.is_empty(), "PROFILE memories are always kept");
    }

    #[test]
    fn different_categories_are_not_duplicates() {
        let items = vec![
            item(1, "shipping the release today", "EVENT"),
            item(2, "shipping the release today", "KNOWLEDGE"),
        ];
        let archived = select_duplicate_memories_to_archive(&items, 0.7, 20);
        assert!(archived.is_empty());
    }

    #[test]
    fn respects_max_cap() {
        let items = vec![
            item(1, "alpha beta gamma delta", "KNOWLEDGE"),
            item(2, "alpha beta gamma delta", "KNOWLEDGE"),
            item(3, "alpha beta gamma delta", "KNOWLEDGE"),
        ];
        let archived = select_duplicate_memories_to_archive(&items, 0.7, 1);
        assert_eq!(archived.len(), 1, "cap limits archives per pass");
    }

    #[test]
    fn distinct_memories_are_kept() {
        let items = vec![
            item(1, "user is a rust developer", "KNOWLEDGE"),
            item(2, "user enjoys hiking on weekends", "KNOWLEDGE"),
        ];
        let archived = select_duplicate_memories_to_archive(&items, 0.82, 20);
        assert!(archived.is_empty());
    }
}

#[cfg(test)]
mod graph_recall_tests {
    use super::{extract_kg_seeds, render_graph_section};
    use crate::internal::storage::db::KgTriple;

    fn triple(subject: &str, predicate: &str, object: &str) -> KgTriple {
        KgTriple {
            id: 1,
            subject: subject.into(),
            predicate: predicate.into(),
            object: object.into(),
            chat_id: Some(1),
            valid_from: "2026-01-01T00:00:00Z".into(),
            valid_to: None,
            confidence: 0.9,
            source: "test".into(),
            source_memory_id: None,
            created_at: "2026-01-01T00:00:00Z".into(),
        }
    }

    #[test]
    fn seeds_match_multiword_entities_case_insensitively() {
        let entities = vec![
            "New York".to_string(),
            "Acme".to_string(),
            "Go".to_string(), // too short → ignored
        ];
        let seeds = extract_kg_seeds("Does acme still have an office in new york?", &entities, 8);
        assert!(seeds.contains(&"New York".to_string()));
        assert!(seeds.contains(&"Acme".to_string()));
        assert!(!seeds.iter().any(|s| s == "Go"));
    }

    #[test]
    fn empty_query_yields_no_seeds() {
        let entities = vec!["Acme".to_string()];
        assert!(extract_kg_seeds("   ", &entities, 8).is_empty());
    }

    #[test]
    fn graph_section_skips_already_stated_relationships() {
        let triples = vec![
            triple("Acme", "located_in", "Berlin"),
            triple("Bob", "manages", "Acme"),
        ];
        // The first relationship is already spelled out in an injected memory; the
        // second (Bob→Acme) is new and should survive.
        let injected = "[KNOWLEDGE] [chat] Acme is located_in Berlin\n";
        let section = render_graph_section(&triples, injected);
        assert!(!section.contains("Berlin"));
        assert!(section.contains("Bob"));
        assert!(section.contains("Acme"));
    }
}