mentedb/

process_turn.rs

//! Unified `process_turn` orchestration.
//!
//! This module provides a single `MenteDb::process_turn()` method that
//! implements the full conversation-turn pipeline used by MCP servers,
//! cloud APIs, and any other client. The steps are:
//!
//! 1. Embed query + retrieve context (speculative cache → hybrid search)
//! 2. Check pain signals
//! 3. Store episodic memory (with write inference)
//! 4. Detect actions, proactive recall, corrections, sentiment
//! 5. Detect phantoms + stream contradiction check
//! 6. Update trajectory + ghost memories
//! 7. Update speculative cache
//! 8. Extract facts + link edges
//! 9. Auto-maintenance (decay / archival / consolidation on intervals)
//!
//! LLM-powered features (entity resolution, topic canonicalization,
//! contradiction verification, memory extraction) are **not** included
//! here — they are client-specific and should be layered on top of the
//! `ProcessTurnResult` by the caller.

use mentedb_cognitive::stream::StreamAlert;
use mentedb_cognitive::trajectory::{DecisionState, TrajectoryNode};
use mentedb_consolidation::FactExtractor;
use mentedb_context::{DeltaTracker, ScoredMemory};
use mentedb_core::edge::EdgeType;
use mentedb_core::memory::MemoryType;
use mentedb_core::types::{AgentId, MemoryId, Timestamp, UserId};
use mentedb_core::{MemoryEdge, MemoryNode};
use tracing::{debug, info, warn};
use uuid::Uuid;

use crate::MenteDb;

/// Input for a single conversation turn.
#[derive(Debug, Clone)]
pub struct ProcessTurnInput {
    /// The user's message text.
    pub user_message: String,
    /// The assistant's response (if available).
    pub assistant_response: Option<String>,
    /// Monotonically increasing turn number.
    pub turn_id: u64,
    /// Optional project/workspace scope for memory tagging.
    pub project_context: Option<String>,
    /// Agent UUID (defaults to nil if not provided).
    pub agent_id: Option<Uuid>,
    /// End-user UUID, orthogonal to `agent_id` (defaults to nil if not
    /// provided). The stored turn is tagged with both owners, and internal
    /// recalls are scoped to this user AND the agent so another user's private
    /// memories can never surface or be linked.
    pub user_id: Option<Uuid>,
    /// Originating session, tagged onto stored turns so injection recall
    /// can exclude memories already present in that session's context.
    pub session_id: Option<String>,
}

/// A detected action from the conversation.
#[derive(Debug, Clone)]
pub struct DetectedAction {
    pub action_type: String,
    pub detail: String,
}

/// A pain warning matched against the current context.
#[derive(Debug, Clone)]
pub struct PainWarning {
    pub signal_id: MemoryId,
    pub intensity: f32,
    pub description: String,
}

/// A proactively recalled memory relevant to detected actions.
#[derive(Debug, Clone)]
pub struct ProactiveRecall {
    pub memory_id: MemoryId,
    pub content: String,
    pub relevance: f32,
    pub action_type: String,
}

/// Result of processing a conversation turn.
#[derive(Debug, Clone)]
pub struct ProcessTurnResult {
    /// Retrieved context memories (scored and ranked).
    pub context: Vec<ScoredMemory>,
    /// IDs of newly stored memories this turn.
    pub stored_ids: Vec<MemoryId>,
    /// The episodic memory ID for this turn.
    pub episodic_id: Option<MemoryId>,
    /// Pain warnings triggered by the user message.
    pub pain_warnings: Vec<PainWarning>,
    /// Whether context came from the speculative cache.
    pub cache_hit: bool,
    /// Number of write-inference actions applied.
    pub inference_actions: u32,
    /// Detected actions in the conversation.
    pub detected_actions: Vec<DetectedAction>,
    /// Proactively recalled memories.
    pub proactive_recalls: Vec<ProactiveRecall>,
    /// Auto-correction memory ID if a correction was detected.
    pub correction_id: Option<MemoryId>,
    /// Sentiment score (-1.0 to 1.0).
    pub sentiment: f32,
    /// Number of phantom (knowledge gap) entities detected.
    pub phantom_count: usize,
    /// Number of stream contradictions detected.
    pub contradiction_count: usize,
    /// Predicted next topics from trajectory analysis.
    pub predicted_topics: Vec<String>,
    /// Number of facts extracted and linked.
    pub facts_extracted: usize,
    /// Number of edges created from fact extraction.
    pub edges_created: u32,
    /// Whether enrichment is pending (caller should invoke enrichment pipeline).
    pub enrichment_pending: bool,
    /// Delta: memory IDs added since last turn.
    pub delta_added: Vec<MemoryId>,
    /// Delta: memory IDs removed since last turn.
    pub delta_removed: Vec<MemoryId>,
}

/// The read half of a turn: what the caller needs to build its next prompt.
/// Produced by [`MenteDb::recall_turn_context`]; feed `context` to
/// [`MenteDb::ingest_turn`] so analysis sees what the caller saw.
#[derive(Debug, Clone)]
pub struct RecallContextResult {
    /// Retrieved context memories (scored and ranked).
    pub context: Vec<ScoredMemory>,
    /// Pain warnings triggered by the user message.
    pub pain_warnings: Vec<PainWarning>,
    /// Whether context came from the speculative cache.
    pub cache_hit: bool,
    /// Delta: memory IDs added since last turn.
    pub delta_added: Vec<MemoryId>,
    /// Delta: memory IDs removed since last turn.
    pub delta_removed: Vec<MemoryId>,
}

/// The write half of a turn: everything [`MenteDb::ingest_turn`] produced
/// after the caller's response no longer depends on it.
#[derive(Debug, Clone)]
pub struct IngestTurnResult {
    /// IDs of newly stored memories this turn.
    pub stored_ids: Vec<MemoryId>,
    /// The episodic memory ID for this turn.
    pub episodic_id: Option<MemoryId>,
    /// Number of write-inference actions applied.
    pub inference_actions: u32,
    /// Detected actions in the conversation.
    pub detected_actions: Vec<DetectedAction>,
    /// Proactively recalled memories.
    pub proactive_recalls: Vec<ProactiveRecall>,
    /// Auto-correction memory ID if a correction was detected.
    pub correction_id: Option<MemoryId>,
    /// Sentiment score (-1.0 to 1.0).
    pub sentiment: f32,
    /// Number of phantom (knowledge gap) entities detected.
    pub phantom_count: usize,
    /// Number of stream contradictions detected.
    pub contradiction_count: usize,
    /// Predicted next topics from trajectory analysis.
    pub predicted_topics: Vec<String>,
    /// Number of facts extracted and linked.
    pub facts_extracted: usize,
    /// Number of edges created from fact extraction.
    pub edges_created: u32,
    /// Whether enrichment is pending (caller should invoke enrichment pipeline).
    pub enrichment_pending: bool,
}

/// Recompose the split halves into the classic single-call result, so
/// `process_turn` keeps its exact shape for existing callers.
fn compose_turn_result(recall: RecallContextResult, ingest: IngestTurnResult) -> ProcessTurnResult {
    ProcessTurnResult {
        context: recall.context,
        stored_ids: ingest.stored_ids,
        episodic_id: ingest.episodic_id,
        pain_warnings: recall.pain_warnings,
        cache_hit: recall.cache_hit,
        inference_actions: ingest.inference_actions,
        detected_actions: ingest.detected_actions,
        proactive_recalls: ingest.proactive_recalls,
        correction_id: ingest.correction_id,
        sentiment: ingest.sentiment,
        phantom_count: ingest.phantom_count,
        contradiction_count: ingest.contradiction_count,
        predicted_topics: ingest.predicted_topics,
        facts_extracted: ingest.facts_extracted,
        edges_created: ingest.edges_created,
        enrichment_pending: ingest.enrichment_pending,
        delta_added: recall.delta_added,
        delta_removed: recall.delta_removed,
    }
}

// ── Action detection keywords ──

const ACTION_KEYWORDS: &[(&str, &str)] = &[
    ("deploy", "deployment"),
    ("release", "release"),
    ("migrate", "migration"),
    ("refactor", "refactoring"),
    ("debug", "debugging"),
    ("install", "installation"),
    ("configure", "configuration"),
    ("test", "testing"),
    ("build", "building"),
    ("commit", "version_control"),
    ("merge", "version_control"),
    ("review", "code_review"),
];

// ── Correction indicators ──

const CORRECTION_INDICATORS: &[&str] = &[
    "actually",
    "correction",
    "i was wrong",
    "that's not right",
    "let me correct",
    "i meant",
    "not quite",
    "to clarify",
    "i misspoke",
    "update:",
    "scratch that",
];

// ── Speculation indicators (for ghost memories) ──

const SPECULATION_INDICATORS: &[&str] = &[
    "might be",
    "probably",
    "seems like",
    "i think",
    "looks like",
    "considering",
    "planning to",
    "thinking about",
    "maybe",
];

// ── Sentiment keywords ──

const POSITIVE_WORDS: &[&str] = &[
    "great",
    "excellent",
    "perfect",
    "thanks",
    "awesome",
    "love",
    "good",
    "nice",
    "wonderful",
    "fantastic",
    "helpful",
    "amazing",
    "brilliant",
    "works",
    "solved",
];

const NEGATIVE_WORDS: &[&str] = &[
    "bad",
    "terrible",
    "awful",
    "hate",
    "wrong",
    "broken",
    "fail",
    "error",
    "bug",
    "frustrating",
    "annoying",
    "useless",
    "horrible",
    "worst",
    "disappointed",
];

impl MenteDb {
    /// Process a single conversation turn through the full cognitive pipeline.
    ///
    /// This is the unified entry point that replaces the duplicated orchestration
    /// in MCP and platform. It handles:
    /// - Context retrieval (speculative cache → hybrid search)
    /// - Pain signal checking
    /// - Episodic memory storage (with automatic write inference)
    /// - Action detection, proactive recall, corrections, sentiment
    /// - Phantom detection + stream contradiction checking
    /// - Trajectory tracking + ghost memory creation
    /// - Speculative cache pre-assembly
    /// - Fact extraction + edge linking
    /// - Auto-maintenance (decay / archival / consolidation)
    ///
    /// LLM-powered features (entity resolution, topic canonicalization,
    /// Bedrock extraction, etc.) should be layered on top by the caller.
    pub fn process_turn(
        &self,
        input: &ProcessTurnInput,
        delta_tracker: &mut DeltaTracker,
    ) -> crate::MenteResult<ProcessTurnResult> {
        let recall = self.recall_turn_context(input, delta_tracker)?;
        let ingest = self.ingest_turn(input, &recall.context)?;
        Ok(compose_turn_result(recall, ingest))
    }

    /// The read half of a turn: everything the caller needs to build its next
    /// prompt (recalled context, pain warnings, delta), and nothing else. The
    /// write half is [`MenteDb::ingest_turn`]; callers on a latency budget
    /// respond after this returns and run ingestion behind the response.
    /// `process_turn` composes the two halves and is behavior-identical to the
    /// original single-call pipeline.
    pub fn recall_turn_context(
        &self,
        input: &ProcessTurnInput,
        delta_tracker: &mut DeltaTracker,
    ) -> crate::MenteResult<RecallContextResult> {
        // §1: Embed query
        let query_embedding = self.embed_or_empty(&input.user_message)?;

        // §1b: Try speculative cache, fall back to hybrid search
        let (context, current_ids, cache_hit) = self.retrieve_context(
            &input.user_message,
            &query_embedding,
            input.agent_id.map(AgentId),
            input.user_id.map(UserId),
            delta_tracker,
        )?;

        // §2: Pain signals
        let pain_warnings = self.check_pain_signals(&input.user_message);

        // §2b: Delta tracking
        let delta = delta_tracker.compute_delta(&current_ids, &delta_tracker.last_served.clone());
        delta_tracker.update(&current_ids);

        Ok(RecallContextResult {
            context,
            pain_warnings,
            cache_hit,
            delta_added: delta.added,
            delta_removed: delta.removed,
        })
    }

    /// The write half of a turn: store, reconcile, analyze, maintain. Takes
    /// the context recalled by [`MenteDb::recall_turn_context`] so phantom and
    /// stream analysis see exactly what the caller saw, without re-running
    /// recall. Safe to run after the caller's response has already been sent;
    /// the single-writer discipline is the caller's (hold the account write
    /// lock while this runs, as with any write).
    pub fn ingest_turn(
        &self,
        input: &ProcessTurnInput,
        context: &[ScoredMemory],
    ) -> crate::MenteResult<IngestTurnResult> {
        let agent_id = AgentId(input.agent_id.unwrap_or(Uuid::nil()));
        let user_id = UserId(input.user_id.unwrap_or(Uuid::nil()));
        let assistant_resp = input.assistant_response.as_deref().unwrap_or("");
        let conversation = format!(
            "User: {}\nAssistant: {}",
            input.user_message, assistant_resp
        );

        // §3: Store episodic turn (write inference runs automatically inside store)
        let (stored_ids, episodic_id) = self.store_episodic(
            &conversation,
            agent_id,
            user_id,
            &input.project_context,
            &input.session_id,
        )?;

        // Scope internal recalls to the turn's owner on BOTH axes so cross-owner
        // inference, fact linking, and proactive recall cannot surface or link
        // another user's or agent's private memories. None on an axis preserves
        // global behavior for that axis.
        let scope_agent = input.agent_id.map(AgentId);
        let scope_user = input.user_id.map(UserId);

        // §4: Write inference on the episodic memory
        let inference_actions = if let Some(eid) = episodic_id {
            self.run_explicit_write_inference(eid, scope_agent, scope_user)?
        } else {
            0
        };

        // §5: Extract facts + link edges
        let (facts_extracted, edges_created) = if let Some(eid) = episodic_id {
            self.extract_and_link_facts(eid, scope_agent, scope_user)?
        } else {
            (0, 0)
        };

        // §6: Detect actions
        let combined_text = format!("{} {}", input.user_message, assistant_resp);
        let detected_actions = detect_actions(&combined_text);

        // §7: Proactive recall
        let proactive_recalls =
            self.proactive_recall(&detected_actions, scope_agent, scope_user)?;

        // §8: Auto-detect corrections (tags the episodic turn, never mints a node)
        let correction_id = self.auto_detect_correction(&input.user_message, episodic_id)?;

        // §9: Sentiment analysis
        let sentiment = analyze_sentiment(&input.user_message);

        // §10: Phantom detection + stream contradiction check
        let (phantom_count, contradiction_count) = self.detect_phantoms_and_check_stream(
            &conversation,
            assistant_resp,
            context,
            input.turn_id,
        );

        // §11: Trajectory + ghost memories
        let predicted_topics = self.update_trajectory(
            input,
            agent_id,
            user_id,
            &stored_ids,
            &detected_actions,
            &combined_text,
        )?;

        // §12: Speculative cache pre-assembly
        self.update_speculative_cache_from_predictions(&predicted_topics);

        // §13: Auto-maintenance
        self.maybe_run_maintenance(input.turn_id);

        Ok(IngestTurnResult {
            stored_ids,
            episodic_id,
            inference_actions,
            detected_actions,
            proactive_recalls,
            correction_id,
            sentiment,
            phantom_count,
            contradiction_count,
            predicted_topics,
            facts_extracted,
            edges_created,
            enrichment_pending: self.needs_enrichment(),
        })
    }

    // ── Private helpers ──

    fn retrieve_context(
        &self,
        user_message: &str,
        query_embedding: &[f32],
        agent: Option<AgentId>,
        user: Option<UserId>,
        _delta_tracker: &DeltaTracker,
    ) -> crate::MenteResult<(Vec<ScoredMemory>, Vec<MemoryId>, bool)> {
        // Try speculative cache first
        if let Some(entry) = self.try_speculative_hit(user_message, Some(query_embedding)) {
            let matched: Vec<ScoredMemory> = entry
                .memory_ids
                .iter()
                .filter_map(|id| self.get_memory(*id).ok())
                .filter(|m| {
                    crate::agent_visible(m.agent_id, agent) && crate::user_visible(m.user_id, user)
                })
                .map(|m| ScoredMemory {
                    memory: m,
                    score: 0.9,
                })
                .collect();
            if matched.len() >= entry.memory_ids.len() / 2 {
                let ids: Vec<MemoryId> = matched.iter().map(|sm| sm.memory.id).collect();
                return Ok((matched, ids, true));
            }
        }

        // Hybrid search fallback
        let now = now_us();
        let results = self.recall_hybrid_scoped_at_mode(
            query_embedding,
            Some(user_message),
            10,
            now,
            None,
            false,
            None,
            agent,
            user,
            None,
        )?;
        let mut scored: Vec<ScoredMemory> = results
            .iter()
            .filter_map(|(mid, score)| {
                self.get_memory(*mid).ok().map(|m| ScoredMemory {
                    memory: m,
                    score: *score,
                })
            })
            .collect();

        // Append always-scoped memories not already recalled. Fetch them from the
        // tag index (scope:always is capped near-zero), loading only the handful
        // of matches. The old path scanned every page and called load_memory for
        // every memory on each turn, which thrashed the buffer pool against disk
        // on large accounts and dominated turn latency.
        let hybrid_ids: std::collections::HashSet<MemoryId> =
            scored.iter().map(|sm| sm.memory.id).collect();
        for mid in self.index.bitmap.query_tag("scope:always") {
            if hybrid_ids.contains(&mid) {
                continue;
            }
            if let Ok(mem) = self.get_memory(mid)
                && crate::agent_visible(mem.agent_id, agent)
                && crate::user_visible(mem.user_id, user)
            {
                scored.push(ScoredMemory {
                    memory: mem,
                    score: 0.85,
                });
            }
        }

        let ids: Vec<MemoryId> = scored.iter().map(|sm| sm.memory.id).collect();
        Ok((scored, ids, false))
    }

    fn check_pain_signals(&self, user_message: &str) -> Vec<PainWarning> {
        let context_words: Vec<String> = user_message
            .split_whitespace()
            .map(|w| w.to_lowercase())
            .collect();
        let all_signals = self.all_pain_signals();
        let mut scored: Vec<_> = all_signals
            .iter()
            .filter_map(|s| {
                let matched = s
                    .trigger_keywords
                    .iter()
                    .filter(|trigger| {
                        context_words
                            .iter()
                            .any(|ctx| ctx.contains(&trigger.to_lowercase()))
                    })
                    .count();
                if matched > 0 {
                    let relevance = matched as f32 / s.trigger_keywords.len().max(1) as f32;
                    let score = s.intensity * relevance;
                    Some((s, score))
                } else {
                    None
                }
            })
            .collect();
        scored.sort_by(|a, b| b.1.partial_cmp(&a.1).unwrap_or(std::cmp::Ordering::Equal));
        scored
            .iter()
            .map(|(s, _)| PainWarning {
                signal_id: s.id,
                intensity: s.intensity,
                description: s.description.clone(),
            })
            .collect()
    }

    /// Embed text, or return an empty embedding when no provider is configured.
    ///
    /// An empty embedding degrades honestly: store() skips the vector index,
    /// write inference skips similarity checks, and retrieval falls back to
    /// keyword (BM25) matching. A zero vector would instead pollute the HNSW
    /// index and silently make every cosine similarity zero.
    fn embed_or_empty(&self, text: &str) -> crate::MenteResult<Vec<f32>> {
        match self.embed_text(text)? {
            Some(embedding) => Ok(embedding),
            None => {
                static NO_EMBEDDER_WARN: std::sync::Once = std::sync::Once::new();
                NO_EMBEDDER_WARN.call_once(|| {
                    warn!(
                        "no embedding provider configured: semantic search, auto-linking, and \
                         contradiction detection are DISABLED; retrieval falls back to keyword \
                         (BM25) matching only. Configure a provider via open_with_embedder()."
                    );
                });
                Ok(Vec::new())
            }
        }
    }

    fn store_episodic(
        &self,
        conversation: &str,
        agent_id: AgentId,
        user_id: UserId,
        project_context: &Option<String>,
        session_id: &Option<String>,
    ) -> crate::MenteResult<(Vec<MemoryId>, Option<MemoryId>)> {
        let embedding = self.embed_or_empty(conversation)?;
        let mut node = MemoryNode::new(
            agent_id,
            MemoryType::Episodic,
            conversation.to_string(),
            embedding,
        )
        .with_user_id(user_id);
        node.tags.push("turn".to_string());
        if let Some(ctx) = project_context {
            node.tags.push(format!("scope:project:{}", ctx));
        }
        if let Some(session) = session_id {
            node.tags.push(format!("session:{}", session));
        }
        let id = node.id;
        self.store(node)?;
        Ok((vec![id], Some(id)))
    }

    fn run_explicit_write_inference(
        &self,
        id: MemoryId,
        agent: Option<AgentId>,
        user: Option<UserId>,
    ) -> crate::MenteResult<u32> {
        // The turn's memory can legitimately be gone by now: paraphrase dedup
        // can skip the store, and write inference inside store can supersede
        // or forget it (a correction resolving against an earlier fact). A
        // missing target ends this step, never the whole turn.
        let Ok(target) = self.get_memory(id) else {
            return Ok(0);
        };
        // Scope to the turn's owner on BOTH axes: contradiction and dedup edges
        // must not form across users or agents. Nil owned (shared/global)
        // memories remain in scope.
        let similar_ids = self
            .recall_hybrid_scoped_at_mode(
                &target.embedding,
                None,
                50,
                now_us(),
                None,
                false,
                None,
                agent,
                user,
                None,
            )
            .unwrap_or_default();
        let existing: Vec<MemoryNode> = similar_ids
            .iter()
            .filter_map(|(mid, _)| {
                if *mid != id {
                    self.get_memory(*mid).ok()
                } else {
                    None
                }
            })
            .collect();

        let engine = crate::WriteInferenceEngine::new();
        let actions = engine.infer_on_write(&target, &existing, &[]);
        let now = now_us();
        let mut applied = 0u32;

        for action in &actions {
            match action {
                crate::InferredAction::FlagContradiction {
                    existing: ex, new, ..
                } => {
                    let edge = MemoryEdge {
                        source: *new,
                        target: *ex,
                        edge_type: EdgeType::Contradicts,
                        weight: 1.0,
                        created_at: now,
                        valid_from: None,
                        valid_until: None,
                        label: None,
                    };
                    if let Err(e) = self.relate(edge) {
                        tracing::warn!("failed to create inferred edge: {e}");
                    }
                    applied += 1;
                }
                crate::InferredAction::MarkObsolete {
                    memory,
                    superseded_by,
                } => {
                    let edge = MemoryEdge {
                        source: *superseded_by,
                        target: *memory,
                        edge_type: EdgeType::Supersedes,
                        weight: 1.0,
                        created_at: now,
                        valid_from: None,
                        valid_until: None,
                        label: None,
                    };
                    if let Err(e) = self.relate(edge) {
                        tracing::warn!("failed to create inferred edge: {e}");
                    }
                    applied += 1;
                }
                crate::InferredAction::CreateEdge {
                    source,
                    target,
                    edge_type,
                    weight,
                } => {
                    let edge = MemoryEdge {
                        source: *source,
                        target: *target,
                        edge_type: *edge_type,
                        weight: *weight,
                        created_at: now,
                        valid_from: None,
                        valid_until: None,
                        label: None,
                    };
                    if let Err(e) = self.relate(edge) {
                        tracing::warn!("failed to create inferred edge: {e}");
                    }
                    applied += 1;
                }
                crate::InferredAction::UpdateConfidence {
                    memory,
                    new_confidence,
                } => {
                    if let Ok(mut mem) = self.get_memory(*memory) {
                        mem.confidence = *new_confidence;
                        let _ = self.store(mem);
                        applied += 1;
                    }
                }
                crate::InferredAction::InvalidateMemory {
                    memory,
                    valid_until,
                    ..
                } => {
                    if let Ok(mut mem) = self.get_memory(*memory) {
                        mem.valid_until = Some(*valid_until);
                        let _ = self.store(mem);
                        applied += 1;
                    }
                }
                crate::InferredAction::UpdateContent {
                    memory,
                    new_content,
                    ..
                } => {
                    if let Ok(mut mem) = self.get_memory(*memory) {
                        mem.content = new_content.clone();
                        let _ = self.store(mem);
                        applied += 1;
                    }
                }
                crate::InferredAction::PropagateBeliefChange { .. } => {
                    applied += 1;
                }
                crate::InferredAction::DeduplicateExact { duplicate, keeper } => {
                    // Byte-identical re-save: invalidate the duplicate and link it
                    // to the survivor with a Derived edge (dedup lineage), not a
                    // Supersedes edge that would read as a conflict.
                    if let Ok(mut mem) = self.get_memory(*duplicate) {
                        mem.valid_until = Some(now);
                        let _ = self.store(mem);
                    }
                    let edge = MemoryEdge {
                        source: *keeper,
                        target: *duplicate,
                        edge_type: EdgeType::Derived,
                        weight: 1.0,
                        created_at: now,
                        valid_from: None,
                        valid_until: None,
                        label: None,
                    };
                    if let Err(e) = self.relate(edge) {
                        tracing::warn!("failed to create dedup edge: {e}");
                    }
                    applied += 1;
                }
            }
        }
        Ok(applied)
    }

    fn extract_and_link_facts(
        &self,
        id: MemoryId,
        agent: Option<AgentId>,
        user: Option<UserId>,
    ) -> crate::MenteResult<(usize, u32)> {
        // Same tolerance as run_explicit_write_inference: the turn's memory
        // can have been deduped, superseded, or forgotten between store and
        // this step; a missing target ends this step, never the whole turn.
        let Ok(target) = self.get_memory(id) else {
            return Ok((0, 0));
        };
        let extractor = FactExtractor::new();
        let facts = extractor.extract_facts(&target);
        if facts.is_empty() {
            return Ok((0, 0));
        }

        // Scope to the turn's owner on BOTH axes: "related" edges must not link
        // one user's or agent's memory to another's. Nil owned (shared/global)
        // memories remain in scope.
        let similar_ids = self
            .recall_hybrid_scoped_at_mode(
                &target.embedding,
                None,
                50,
                now_us(),
                None,
                false,
                None,
                agent,
                user,
                None,
            )
            .unwrap_or_default();
        let nearby: Vec<MemoryNode> = similar_ids
            .iter()
            .filter_map(|(mid, _)| {
                if *mid != id {
                    self.get_memory(*mid).ok()
                } else {
                    None
                }
            })
            .collect();

        let now = now_us();
        let mut edges_created = 0u32;
        for fact in &facts {
            for other in &nearby {
                if other.content.contains(&fact.subject) || other.content.contains(&fact.object) {
                    let edge = MemoryEdge {
                        source: id,
                        target: other.id,
                        edge_type: EdgeType::Related,
                        weight: 0.5,
                        created_at: now,
                        valid_from: None,
                        valid_until: None,
                        label: None,
                    };
                    if let Err(e) = self.relate(edge) {
                        tracing::warn!("failed to create inferred edge: {e}");
                    }
                    edges_created += 1;
                }
            }
        }
        Ok((facts.len(), edges_created))
    }

    fn proactive_recall(
        &self,
        actions: &[DetectedAction],
        agent: Option<AgentId>,
        user: Option<UserId>,
    ) -> crate::MenteResult<Vec<ProactiveRecall>> {
        let mut recalls = Vec::new();
        for action in actions {
            let search_query = format!("{} {}", action.action_type, action.detail);
            let Some(emb) = self.embed_text(&search_query)? else {
                continue;
            };
            // Scope to the turn's owner on BOTH axes so proactive recall never
            // surfaces another user's or agent's private memories (nil
            // owned/global still pass).
            let results = self
                .recall_hybrid_scoped_at_mode(
                    &emb,
                    None,
                    3,
                    now_us(),
                    None,
                    false,
                    None,
                    agent,
                    user,
                    None,
                )
                .unwrap_or_default();
            for (mid, score) in &results {
                if let Ok(mem) = self.get_memory(*mid) {
                    recalls.push(ProactiveRecall {
                        memory_id: *mid,
                        content: mem.content.clone(),
                        relevance: *score,
                        action_type: action.action_type.clone(),
                    });
                }
            }
        }
        Ok(recalls)
    }

    /// Detect a correction signal and tag the stored episodic turn with it.
    ///
    /// Only the user message is scanned: correction indicators in assistant
    /// output ("actually", "to clarify", ...) are ordinary prose, and scanning
    /// them caused unrelated user messages to be recorded as corrections.
    /// The signal is preserved as a tag on the episodic turn rather than a
    /// fabricated semantic node, because a verbatim copy of the user message
    /// is working material, not distilled knowledge; distilled correction
    /// memories come from fact extraction.
    fn auto_detect_correction(
        &self,
        user_message: &str,
        episodic_id: Option<MemoryId>,
    ) -> crate::MenteResult<Option<MemoryId>> {
        let lowered = user_message.to_lowercase();
        let is_correction = CORRECTION_INDICATORS
            .iter()
            .any(|ind| lowered.contains(ind));
        if !is_correction {
            return Ok(None);
        }
        let Some(eid) = episodic_id else {
            return Ok(None);
        };
        let pid = {
            let pm = self.page_map.read();
            match pm.get(&eid) {
                Some(p) => *p,
                None => return Ok(None),
            }
        };
        let Ok(mut node) = self.storage.load_memory(pid) else {
            return Ok(None);
        };
        if !node.tags.iter().any(|t| t == "auto-correction") {
            node.tags.push("auto-correction".to_string());
            self.update_node(pid, &node)?;
        }
        Ok(Some(eid))
    }

    fn detect_phantoms_and_check_stream(
        &self,
        conversation: &str,
        assistant_resp: &str,
        context: &[ScoredMemory],
        turn_id: u64,
    ) -> (usize, usize) {
        // Phantom detection
        let known: Vec<String> = context
            .iter()
            .flat_map(|sm| {
                sm.memory
                    .content
                    .split_whitespace()
                    .map(|w| w.to_lowercase())
            })
            .collect();
        let phantoms = self.detect_phantoms(conversation, &known, turn_id);
        let phantom_count = phantoms.len();

        // Stream contradiction check
        let known_facts: Vec<(MemoryId, String)> = context
            .iter()
            .map(|sm| (sm.memory.id, sm.memory.content.clone()))
            .collect();
        let contradiction_count = if !known_facts.is_empty() {
            self.feed_stream_token(assistant_resp);
            let alerts = self.check_stream_alerts(&known_facts);
            alerts
                .iter()
                .filter(|a| matches!(a, StreamAlert::Contradiction { .. }))
                .count()
        } else {
            0
        };

        (phantom_count, contradiction_count)
    }

    fn update_trajectory(
        &self,
        input: &ProcessTurnInput,
        agent_id: AgentId,
        user_id: UserId,
        stored_ids: &[MemoryId],
        detected_actions: &[DetectedAction],
        combined_text: &str,
    ) -> crate::MenteResult<Vec<String>> {
        let decision_state = if stored_ids.is_empty() {
            DecisionState::Investigating
        } else {
            DecisionState::Completed
        };

        let raw_topic = if input.user_message.chars().count() > 100 {
            format!(
                "{}...",
                input.user_message.chars().take(100).collect::<String>()
            )
        } else {
            input.user_message.clone()
        };

        let topic_embedding = self.embed_or_empty(&raw_topic)?;

        let node = TrajectoryNode {
            turn_id: input.turn_id,
            topic_summary: raw_topic,
            topic_embedding,
            decision_state,
            open_questions: Vec::new(),
            timestamp: now_us(),
        };
        self.record_trajectory_turn(node);
        let predictions = self.predict_next_topics();

        // Ghost memory: store speculative content at low confidence
        let has_speculation = SPECULATION_INDICATORS
            .iter()
            .any(|s| combined_text.contains(s));
        if has_speculation && !detected_actions.is_empty() {
            let ghost_content = format!(
                "Unconfirmed: {}",
                input.user_message.chars().take(300).collect::<String>()
            );
            if let Ok(Some(ghost_emb)) = self.embed_text(&ghost_content) {
                let mut ghost_node =
                    MemoryNode::new(agent_id, MemoryType::Semantic, ghost_content, ghost_emb)
                        .with_user_id(user_id);
                ghost_node.confidence = 0.3;
                ghost_node.tags = vec!["ghost-memory".to_string(), "unconfirmed".to_string()];
                if let Some(ctx) = &input.project_context {
                    ghost_node.tags.push(format!("scope:project:{}", ctx));
                }
                let _ = self.store(ghost_node);
            }
        }

        Ok(predictions)
    }

    fn update_speculative_cache_from_predictions(&self, predictions: &[String]) {
        if predictions.is_empty() {
            debug!("speculative cache: no predictions this turn");
            return;
        }
        // Each predicted topic either becomes an entry or is dropped at one
        // step; count which, so an empty cache in production says why instead
        // of failing silently.
        #[derive(Default)]
        struct Outcome {
            built: usize,
            no_embedder: usize,
            embed_failed: usize,
            recall_failed: usize,
            nothing_similar: usize,
            unloadable: usize,
        }
        let outcome = std::cell::RefCell::new(Outcome::default());
        let predictions_owned = predictions.to_vec();
        self.pre_assemble_speculative(predictions_owned, |topic| {
            let topic_emb = match self.embed_text(topic) {
                Ok(Some(e)) => e,
                Ok(None) => {
                    outcome.borrow_mut().no_embedder += 1;
                    return None;
                }
                Err(e) => {
                    warn!(error = %e, "speculative cache: topic embed failed");
                    outcome.borrow_mut().embed_failed += 1;
                    return None;
                }
            };
            let similar_ids = match self.recall_similar(&topic_emb, 5) {
                Ok(ids) => ids,
                Err(e) => {
                    warn!(error = %e, "speculative cache: recall failed");
                    outcome.borrow_mut().recall_failed += 1;
                    return None;
                }
            };
            if similar_ids.is_empty() {
                outcome.borrow_mut().nothing_similar += 1;
                return None;
            }
            let mut context_parts = Vec::new();
            let mut memory_ids = Vec::new();
            for (mid, _) in &similar_ids {
                if let Ok(mem) = self.get_memory(*mid) {
                    context_parts.push(mem.content.clone());
                    memory_ids.push(*mid);
                }
            }
            if memory_ids.is_empty() {
                outcome.borrow_mut().unloadable += 1;
                return None;
            }
            outcome.borrow_mut().built += 1;
            // Keep the topic embedding so a later query matches the entry by
            // meaning; without it the cache falls back to keyword overlap.
            Some((context_parts.join("\n---\n"), memory_ids, Some(topic_emb)))
        });
        let o = outcome.into_inner();
        info!(
            predictions = predictions.len(),
            built = o.built,
            no_embedder = o.no_embedder,
            embed_failed = o.embed_failed,
            recall_failed = o.recall_failed,
            nothing_similar = o.nothing_similar,
            unloadable = o.unloadable,
            enabled = self.cognitive_config.speculative_cache,
            cache_size = self.speculative_cache_stats().cache_size,
            "speculative cache pre-assembly"
        );
    }

    fn maybe_run_maintenance(&self, turn_id: u64) {
        if turn_id == 0 {
            return;
        }

        // Every 50 turns: salience decay
        if turn_id.is_multiple_of(50) {
            match self.apply_decay_global() {
                Ok(updated) => {
                    debug!(turn_id, updated, "auto-maintenance: decay applied");
                }
                Err(e) => {
                    warn!(turn_id, error = %e, "auto-maintenance: decay failed");
                }
            }
        }

        // Every 100 turns: archival evaluation
        if turn_id.is_multiple_of(100) {
            match self.evaluate_archival_global() {
                Ok(decisions) => {
                    let mut archived = 0u64;
                    for (id, decision) in &decisions {
                        if matches!(
                            decision,
                            crate::ArchivalDecision::Delete | crate::ArchivalDecision::Archive
                        ) {
                            let _ = self.forget(*id);
                            archived += 1;
                        }
                    }
                    debug!(turn_id, archived, "auto-maintenance: archival evaluated");
                }
                Err(e) => {
                    warn!(turn_id, error = %e, "auto-maintenance: archival failed");
                }
            }
        }

        // Every 200 turns: consolidation
        if turn_id.is_multiple_of(200) {
            match self.find_consolidation_candidates(2, 0.85) {
                Ok(candidates) => {
                    for candidate in &candidates {
                        let _ = self.consolidate_cluster(&candidate.memories);
                    }
                    debug!(
                        turn_id,
                        clusters = candidates.len(),
                        "auto-maintenance: consolidation"
                    );
                }
                Err(e) => {
                    warn!(turn_id, error = %e, "auto-maintenance: consolidation failed");
                }
            }
        }

        // Enrichment trigger: mark pending when interval is reached
        let enrichment = &self.cognitive_config.enrichment_config;
        if enrichment.enabled && enrichment.trigger_interval > 0 {
            let last = self.last_enrichment_turn();
            let turns_since = turn_id.saturating_sub(last);
            if turns_since >= enrichment.trigger_interval && !self.needs_enrichment() {
                *self.enrichment_pending.write() = true;
                debug!(
                    turn_id,
                    last_enrichment = last,
                    turns_since,
                    "enrichment trigger: marked pending"
                );
            }
        }
    }
}

// ── Free functions ──

fn detect_actions(combined_text: &str) -> Vec<DetectedAction> {
    let lower = combined_text.to_lowercase();
    ACTION_KEYWORDS
        .iter()
        .filter(|(kw, _)| lower.contains(kw))
        .map(|(kw, action_type)| {
            let detail = extract_sentence_containing(&lower, kw);
            DetectedAction {
                action_type: action_type.to_string(),
                detail,
            }
        })
        .collect()
}

fn analyze_sentiment(text: &str) -> f32 {
    let lower = text.to_lowercase();
    let words: Vec<&str> = lower.split_whitespace().collect();
    let total = words.len().max(1) as f32;

    let positive = words
        .iter()
        .filter(|w| POSITIVE_WORDS.iter().any(|p| w.contains(p)))
        .count() as f32;
    let negative = words
        .iter()
        .filter(|w| NEGATIVE_WORDS.iter().any(|n| w.contains(n)))
        .count() as f32;

    ((positive - negative) / total).clamp(-1.0, 1.0)
}

fn extract_sentence_containing(text: &str, keyword: &str) -> String {
    text.split('.')
        .find(|s| s.contains(keyword))
        .unwrap_or(keyword)
        .trim()
        .chars()
        .take(200)
        .collect()
}

fn now_us() -> Timestamp {
    std::time::SystemTime::now()
        .duration_since(std::time::UNIX_EPOCH)
        .unwrap_or_default()
        .as_micros() as u64
}