mentedb_server/
maintenance.rs

1//! Built-in maintenance sweep for self-hosted deployments.
2//!
3//! Runs the rule-based "overnight" jobs that keep memory healthy: consolidate
4//! near-duplicates, decay salience, forget cold memories, and evict the stale
5//! speculative cache. A self-hoster gets the same upkeep the cloud worker
6//! provides, either automatically on a timer (`--maintenance-interval-hours`)
7//! or one-shot via the `maintenance` subcommand for external cron, without
8//! writing any code. Per-turn fact extraction is handled separately by
9//! `--auto-extract`.
10
11use std::sync::Arc;
12use std::time::Duration;
13
14use mentedb::MenteDb;
15use mentedb::consolidation::archival::ArchivalDecision;
16use mentedb_extraction::{ExtractionConfig, ExtractionLlmJudge, HttpExtractionProvider};
17use tracing::{info, warn};
18
19/// Minimum cluster size for consolidation.
20pub(crate) const MIN_CLUSTER_SIZE: usize = 2;
21/// Recent trajectory topics to canonicalize per sweep. The learned topic cache
22/// makes already-labeled topics free, so this only spends LLM calls on new ones.
23const CANONICALIZE_LIMIT: usize = 32;
24/// Cap on LLM conflict checks per sweep. Pairs that already carry a conflict edge
25/// are skipped, so this only spends calls on unjudged high-similarity pairs.
26const CONFLICT_CHECK_LIMIT: usize = 32;
27/// Cosine similarity threshold for grouping memories to consolidate.
28pub(crate) const SIMILARITY_THRESHOLD: f32 = 0.85;
29/// Speculative cache entries older than this (24h) are evicted.
30const CACHE_MAX_AGE_US: u64 = 24 * 3600 * 1_000_000;
31/// Cap on deletions per sweep, so a single run can never mass-delete.
32pub(crate) const MAX_FORGETS_PER_SWEEP: usize = 500;
33
34/// What one maintenance sweep did.
35#[derive(Debug, Default)]
36pub struct MaintenanceReport {
37    pub consolidated: usize,
38    pub decayed: usize,
39    pub forgotten: usize,
40}
41
42/// Run one maintenance sweep synchronously. Each job is best-effort: a failure
43/// in one is logged and the remaining jobs still run.
44pub fn run_sweep(db: &MenteDb) -> MaintenanceReport {
45    let mut report = MaintenanceReport::default();
46
47    // 1. Consolidation: merge near-duplicate memories into one richer memory
48    //    (sources invalidated and linked, not deleted).
49    match db.find_consolidation_candidates(MIN_CLUSTER_SIZE, SIMILARITY_THRESHOLD) {
50        Ok(candidates) => {
51            for c in candidates {
52                if db.consolidate_cluster(&c.memories).is_ok() {
53                    report.consolidated += 1;
54                }
55            }
56        }
57        Err(e) => warn!("maintenance: consolidation failed: {e}"),
58    }
59
60    // 2. Decay: age salience so old, unused memories rank lower over time.
61    match db.apply_decay_global() {
62        Ok(n) => report.decayed = n,
63        Err(e) => warn!("maintenance: decay failed: {e}"),
64    }
65
66    // 3. Archival: forget cold memories flagged for deletion (capped per sweep).
67    match db.evaluate_archival_global() {
68        Ok(decisions) => {
69            for (id, decision) in decisions {
70                if report.forgotten >= MAX_FORGETS_PER_SWEEP {
71                    break;
72                }
73                if decision == ArchivalDecision::Delete && db.forget(id).is_ok() {
74                    report.forgotten += 1;
75                }
76            }
77        }
78        Err(e) => warn!("maintenance: archival failed: {e}"),
79    }
80
81    // 4. Evict stale speculative pre-assembly cache entries.
82    db.evict_stale_speculative(CACHE_MAX_AGE_US);
83
84    // 5. Persist everything to disk.
85    if let Err(e) = db.flush() {
86        warn!("maintenance: flush failed: {e}");
87    }
88
89    report
90}
91
92/// Give recent trajectory topics a stable semantic label via the configured
93/// LLM, so topic prediction and the speculative cache stop keying on raw
94/// message text. No-op when no LLM is configured. Async because the LLM call
95/// is; call it outside the sweep's blocking pool.
96pub async fn canonicalize_topics(db: &MenteDb, config: &Option<ExtractionConfig>) -> usize {
97    let Some(cfg) = config.clone() else {
98        return 0;
99    };
100    let provider = match HttpExtractionProvider::new(cfg) {
101        Ok(p) => p,
102        Err(e) => {
103            warn!("maintenance: topic canonicalization skipped, provider build failed: {e}");
104            return 0;
105        }
106    };
107    let judge = ExtractionLlmJudge::new(provider);
108    let n = db
109        .canonicalize_trajectory_topics(judge, CANONICALIZE_LIMIT)
110        .await;
111    if n > 0 {
112        info!(count = n, "maintenance: trajectory topics canonicalized");
113    }
114    n
115}
116
117/// Detect genuine contradictions and supersessions among stored memories via the
118/// configured LLM, recording them as the typed edges the Conflicts view shows.
119/// Unlike the old cosine-similarity heuristic (removed for ~0% precision), the
120/// judge reads both texts. No-op when no LLM is configured. Async because the
121/// LLM calls are; call it outside the sweep's blocking pool.
122pub async fn detect_conflicts(db: &MenteDb, config: &Option<ExtractionConfig>) -> usize {
123    let Some(cfg) = config.clone() else {
124        return 0;
125    };
126    let provider = match HttpExtractionProvider::new(cfg) {
127        Ok(p) => p,
128        Err(e) => {
129            warn!("maintenance: conflict detection skipped, provider build failed: {e}");
130            return 0;
131        }
132    };
133    let judge = ExtractionLlmJudge::new(provider);
134    let candidates = db.memory_ids();
135    let params = mentedb::llm_consolidation::ConflictDetectionParams::default();
136    match db
137        .detect_conflicts_with_llm(&candidates, judge, &params, CONFLICT_CHECK_LIMIT)
138        .await
139    {
140        Ok(n) => {
141            if n > 0 {
142                info!(count = n, "maintenance: conflicts detected");
143            }
144            n
145        }
146        Err(e) => {
147            warn!("maintenance: conflict detection failed: {e}");
148            0
149        }
150    }
151}
152
153/// Spawn a background task that runs a sweep every `interval_hours`. A value of
154/// 0 disables the scheduler (run the `maintenance` subcommand from cron instead).
155/// When an LLM is configured, each cycle also canonicalizes recent topics.
156pub fn spawn_scheduler(
157    databases: Arc<crate::databases::Databases>,
158    interval_hours: u64,
159    extraction_config: Option<ExtractionConfig>,
160) {
161    if interval_hours == 0 {
162        info!("maintenance scheduler disabled (--maintenance-interval-hours 0)");
163        return;
164    }
165    info!("maintenance scheduler enabled: sweep every {interval_hours}h");
166    tokio::spawn(async move {
167        let interval = Duration::from_secs(interval_hours * 3600);
168        loop {
169            tokio::time::sleep(interval).await;
170            // Every open database; idle ones closed by the registry are swept
171            // the next time they are open at sweep time.
172            for (name, db) in databases.open_handles() {
173                let db_sweep = Arc::clone(&db);
174                match tokio::task::spawn_blocking(move || run_sweep(&db_sweep)).await {
175                    Ok(r) => info!(
176                        db = %name,
177                        consolidated = r.consolidated,
178                        decayed = r.decayed,
179                        forgotten = r.forgotten,
180                        "maintenance sweep complete"
181                    ),
182                    Err(e) => warn!(db = %name, "maintenance sweep task panicked: {e}"),
183                }
184                canonicalize_topics(&db, &extraction_config).await;
185                detect_conflicts(&db, &extraction_config).await;
186            }
187        }
188    });
189}
190
191#[cfg(test)]
192mod tests {
193    use super::*;
194    use mentedb::prelude::*;
195    use mentedb_embedding::hash_provider::HashEmbeddingProvider;
196
197    #[test]
198    fn sweep_runs_on_a_populated_db_without_error() {
199        let path = std::env::temp_dir().join(format!("mentedb_maint_test_{}", std::process::id()));
200        let _ = std::fs::remove_dir_all(&path);
201        let db = MenteDb::open_with_embedder(&path, Box::new(HashEmbeddingProvider::new(256)))
202            .expect("open db");
203
204        let agent = AgentId::new();
205        for i in 0..5 {
206            let content = format!("fact number {i} about the project");
207            let emb = db.embed_text(&content).unwrap().unwrap();
208            db.store(MemoryNode::new(agent, MemoryType::Semantic, content, emb))
209                .unwrap();
210        }
211
212        // Must complete without panicking; counts are well-formed. On a small
213        // fresh DB nothing is necessarily removed, this is a smoke test that the
214        // full sweep (consolidate -> decay -> archival -> evict -> flush) runs.
215        let report = run_sweep(&db);
216        assert!(report.forgotten <= MAX_FORGETS_PER_SWEEP);
217
218        drop(db);
219        let _ = std::fs::remove_dir_all(&path);
220    }
221}
mentedb_server/maintenance.rs

mentedb_server/
maintenance.rs
