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    // Judging reads stored memories, so it runs on the reader model.
101    let provider = match HttpExtractionProvider::new(cfg.reader_config()) {
102        Ok(p) => p,
103        Err(e) => {
104            warn!("maintenance: topic canonicalization skipped, provider build failed: {e}");
105            return 0;
106        }
107    };
108    let judge = ExtractionLlmJudge::new(provider);
109    let n = db
110        .canonicalize_trajectory_topics(judge, CANONICALIZE_LIMIT)
111        .await;
112    if n > 0 {
113        info!(count = n, "maintenance: trajectory topics canonicalized");
114    }
115    n
116}
117
118/// Detect genuine contradictions and supersessions among stored memories via the
119/// configured LLM, recording them as the typed edges the Conflicts view shows.
120/// Unlike the old cosine-similarity heuristic (removed for ~0% precision), the
121/// judge reads both texts. No-op when no LLM is configured. Async because the
122/// LLM calls are; call it outside the sweep's blocking pool.
123pub async fn detect_conflicts(db: &MenteDb, config: &Option<ExtractionConfig>) -> usize {
124    let Some(cfg) = config.clone() else {
125        return 0;
126    };
127    // Judging reads stored memories, so it runs on the reader model.
128    let provider = match HttpExtractionProvider::new(cfg.reader_config()) {
129        Ok(p) => p,
130        Err(e) => {
131            warn!("maintenance: conflict detection skipped, provider build failed: {e}");
132            return 0;
133        }
134    };
135    let judge = ExtractionLlmJudge::new(provider);
136    let candidates = db.memory_ids();
137    let params = mentedb::llm_consolidation::ConflictDetectionParams::default();
138    match db
139        .detect_conflicts_with_llm(&candidates, judge, &params, CONFLICT_CHECK_LIMIT)
140        .await
141    {
142        Ok(n) => {
143            if n > 0 {
144                info!(count = n, "maintenance: conflicts detected");
145            }
146            n
147        }
148        Err(e) => {
149            warn!("maintenance: conflict detection failed: {e}");
150            0
151        }
152    }
153}
154
155/// Spawn a background task that runs a sweep every `interval_hours`. A value of
156/// 0 disables the scheduler (run the `maintenance` subcommand from cron instead).
157/// When an LLM is configured, each cycle also canonicalizes recent topics.
158pub fn spawn_scheduler(
159    databases: Arc<crate::databases::Databases>,
160    interval_hours: u64,
161    extraction_config: Option<ExtractionConfig>,
162) {
163    if interval_hours == 0 {
164        info!("maintenance scheduler disabled (--maintenance-interval-hours 0)");
165        return;
166    }
167    info!("maintenance scheduler enabled: sweep every {interval_hours}h");
168    tokio::spawn(async move {
169        let interval = Duration::from_secs(interval_hours * 3600);
170        loop {
171            tokio::time::sleep(interval).await;
172            // Every open database; idle ones closed by the registry are swept
173            // the next time they are open at sweep time.
174            for (name, db) in databases.open_handles() {
175                let db_sweep = Arc::clone(&db);
176                match tokio::task::spawn_blocking(move || run_sweep(&db_sweep)).await {
177                    Ok(r) => info!(
178                        db = %name,
179                        consolidated = r.consolidated,
180                        decayed = r.decayed,
181                        forgotten = r.forgotten,
182                        "maintenance sweep complete"
183                    ),
184                    Err(e) => warn!(db = %name, "maintenance sweep task panicked: {e}"),
185                }
186                canonicalize_topics(&db, &extraction_config).await;
187                detect_conflicts(&db, &extraction_config).await;
188            }
189        }
190    });
191}
192
193#[cfg(test)]
194mod tests {
195    use super::*;
196    use mentedb::prelude::*;
197    use mentedb_embedding::hash_provider::HashEmbeddingProvider;
198
199    #[test]
200    fn sweep_runs_on_a_populated_db_without_error() {
201        let path = std::env::temp_dir().join(format!("mentedb_maint_test_{}", std::process::id()));
202        let _ = std::fs::remove_dir_all(&path);
203        let db = MenteDb::open_with_embedder(&path, Box::new(HashEmbeddingProvider::new(256)))
204            .expect("open db");
205
206        let agent = AgentId::new();
207        for i in 0..5 {
208            let content = format!("fact number {i} about the project");
209            let emb = db.embed_text(&content).unwrap().unwrap();
210            db.store(MemoryNode::new(agent, MemoryType::Semantic, content, emb))
211                .unwrap();
212        }
213
214        // Must complete without panicking; counts are well-formed. On a small
215        // fresh DB nothing is necessarily removed, this is a smoke test that the
216        // full sweep (consolidate -> decay -> archival -> evict -> flush) runs.
217        let report = run_sweep(&db);
218        assert!(report.forgotten <= MAX_FORGETS_PER_SWEEP);
219
220        drop(db);
221        let _ = std::fs::remove_dir_all(&path);
222    }
223}
mentedb_server/maintenance.rs

mentedb_server/
maintenance.rs
