kimetsu_brain/
maintenance.rs

1//! Brain maintenance: prune, compact, projection rebuild, lock clearing.
2//! Split out of `project.rs` (v2.5.1); re-exported by [`crate::project`].
3
4use std::fs;
5use std::path::Path;
6
7use kimetsu_core::KimetsuResult;
8use kimetsu_core::paths::ProjectPaths;
9use rusqlite::params;
10
11use crate::lock::ProjectLock;
12use crate::project::*;
13use crate::projector;
14use crate::trace::{self};
15
16/// MP-6: bulk prune of memories whose outcome-attribution data says they
17/// are net-negative. Selection rules:
18///   use_count >= min_uses
19///   usefulness_score / use_count <= max_ratio
20///   invalidated_at IS NULL
21///   scope filter optional
22///
23/// `apply = false` is the default at the CLI layer so the user sees
24/// what would be touched before any writes. `apply = true` invalidates
25/// each match via the existing `invalidate_memory` path so every
26/// removal still emits a canonical `memory.invalidated` event.
27#[derive(Debug, Clone)]
28pub struct PruneOptions {
29    pub scope: Option<String>,
30    pub min_uses: u32,
31    pub max_ratio: f32,
32    pub apply: bool,
33}
34
35impl Default for PruneOptions {
36    fn default() -> Self {
37        Self {
38            scope: None,
39            min_uses: 3,
40            max_ratio: -0.2,
41            apply: false,
42        }
43    }
44}
45
46#[derive(Debug, Clone)]
47pub struct PruneCandidate {
48    pub memory_id: String,
49    pub scope: String,
50    pub kind: String,
51    pub use_count: u32,
52    pub usefulness_score: f32,
53    pub text: String,
54}
55
56#[derive(Debug, Clone, Default)]
57pub struct PruneSummary {
58    pub candidates: Vec<PruneCandidate>,
59    pub invalidated: u32,
60    pub failed: u32,
61}
62
63pub fn prune_low_usefulness(start: &Path, opts: PruneOptions) -> KimetsuResult<PruneSummary> {
64    let min_uses = opts.min_uses.max(1) as i64;
65
66    let candidates = {
67        let (_paths, _config, conn) = load_project(start)?;
68        let (sql, scope_param): (&str, Option<String>) = if let Some(scope) = opts.scope.as_deref()
69        {
70            (
71                "
72                SELECT memory_id, scope, kind, text, use_count, usefulness_score
73                FROM memories
74                WHERE invalidated_at IS NULL
75                  AND superseded_by IS NULL
76                  AND use_count >= ?1
77                  AND (usefulness_score / CAST(use_count AS REAL)) <= ?2
78                  AND lower(scope) = lower(?3)
79                ORDER BY (usefulness_score / CAST(use_count AS REAL)) ASC
80                ",
81                Some(scope.to_string()),
82            )
83        } else {
84            (
85                "
86                SELECT memory_id, scope, kind, text, use_count, usefulness_score
87                FROM memories
88                WHERE invalidated_at IS NULL
89                  AND superseded_by IS NULL
90                  AND use_count >= ?1
91                  AND (usefulness_score / CAST(use_count AS REAL)) <= ?2
92                ORDER BY (usefulness_score / CAST(use_count AS REAL)) ASC
93                ",
94                None,
95            )
96        };
97        let mut stmt = conn.prepare(sql)?;
98        let max_ratio = opts.max_ratio as f64;
99        let mut found: Vec<PruneCandidate> = if let Some(scope) = scope_param {
100            stmt.query_map(params![min_uses, max_ratio, scope], |row| {
101                Ok(PruneCandidate {
102                    memory_id: row.get(0)?,
103                    scope: row.get(1)?,
104                    kind: row.get(2)?,
105                    text: row.get(3)?,
106                    use_count: row.get(4)?,
107                    usefulness_score: row.get::<_, f64>(5)? as f32,
108                })
109            })?
110            .collect::<Result<Vec<_>, _>>()?
111        } else {
112            stmt.query_map(params![min_uses, max_ratio], |row| {
113                Ok(PruneCandidate {
114                    memory_id: row.get(0)?,
115                    scope: row.get(1)?,
116                    kind: row.get(2)?,
117                    text: row.get(3)?,
118                    use_count: row.get(4)?,
119                    usefulness_score: row.get::<_, f64>(5)? as f32,
120                })
121            })?
122            .collect::<Result<Vec<_>, _>>()?
123        };
124        // Stable tie-break: lowest ratio first, then highest use_count
125        // first (penalize the long-running underperformers).
126        found.sort_by(|a, b| {
127            let ra = a.usefulness_score as f64 / a.use_count.max(1) as f64;
128            let rb = b.usefulness_score as f64 / b.use_count.max(1) as f64;
129            ra.partial_cmp(&rb)
130                .unwrap_or(std::cmp::Ordering::Equal)
131                .then_with(|| b.use_count.cmp(&a.use_count))
132        });
133        found
134    };
135
136    let mut summary = PruneSummary {
137        candidates: candidates.clone(),
138        invalidated: 0,
139        failed: 0,
140    };
141    if !opts.apply {
142        return Ok(summary);
143    }
144
145    for candidate in &candidates {
146        let ratio = candidate.usefulness_score / candidate.use_count.max(1) as f32;
147        let reason = format!(
148            "pruned_by_usefulness ratio={:+.2} use_count={}",
149            ratio, candidate.use_count
150        );
151        match invalidate_memory(start, &candidate.memory_id, Some(&reason)) {
152            Ok(()) => summary.invalidated += 1,
153            Err(_) => summary.failed += 1,
154        }
155    }
156    Ok(summary)
157}
158
159pub fn rebuild_projection(start: &Path, from_traces: bool) -> KimetsuResult<usize> {
160    let (paths, _config, conn) = load_project(start)?;
161    let _lock = ProjectLock::acquire(&paths, "brain rebuild", None)?;
162
163    // Explicit legacy import: rebuild from on-disk trace.jsonl files (inserts
164    // any events missing from the table via OR IGNORE, then projects).
165    if from_traces {
166        let events = trace::read_all_traces(&paths)?;
167        projector::rebuild(&conn, &events)?;
168        return Ok(events.len());
169    }
170
171    // Auto-fallback: a brain whose events table was wiped by a pre-W1.1 rebuild
172    // still has its history only in trace.jsonl. If the table is empty but
173    // traces exist, import them first, then proceed.
174    let event_count: i64 = conn.query_row("SELECT COUNT(*) FROM events", [], |r| r.get(0))?;
175    if event_count == 0 {
176        let events = trace::read_all_traces(&paths)?;
177        if !events.is_empty() {
178            eprintln!(
179                "[kimetsu] events table empty; importing {} event(s) from legacy traces",
180                events.len()
181            );
182            projector::rebuild(&conn, &events)?;
183            return Ok(events.len());
184        }
185    }
186
187    // Normal path: replay the durable events table in place.
188    projector::rebuild_in_place(&conn)
189}
190
191pub fn clear_lock(start: &Path) -> KimetsuResult<bool> {
192    let paths = ProjectPaths::discover(start)?;
193    crate::lock::clear_force(&paths)
194}
195
196// ── Q8: brain compact ────────────────────────────────────────────────────────
197
198/// Report returned by [`compact_brain`] describing what was freed.
199#[derive(Debug, Clone, serde::Serialize)]
200pub struct CompactReport {
201    /// brain.db file size in bytes before compaction.
202    pub bytes_before: u64,
203    /// brain.db file size in bytes after compaction (WAL checkpointed first).
204    pub bytes_after: u64,
205    /// Number of events deleted by `--trim-events-older-than` (0 when not requested).
206    pub events_trimmed: u64,
207    /// Number of invalidated memory rows purged (0 when not requested).
208    pub invalidated_memories_purged: u64,
209}
210
211/// Reclaim dead space in brain.db.
212///
213/// 1. Acquires the project lock (same as `rebuild_projection`).
214/// 2. Optionally purges invalidated memory rows (`purge_invalidated`).
215/// 3. Optionally trims old events (`trim_events_older_than`).
216/// 4. Runs `VACUUM` (outside any transaction) to rebuild the file in-place.
217/// 5. Checkpoints the WAL before measuring `bytes_after` so the measurement
218///    reflects the on-disk file, not the shadow WAL.
219pub fn compact_brain(
220    start: &Path,
221    trim_events_older_than: Option<std::time::Duration>,
222    purge_invalidated: bool,
223) -> KimetsuResult<CompactReport> {
224    let (paths, _config, conn) = load_project(start)?;
225    let _lock = ProjectLock::acquire(&paths, "brain compact", None)?;
226
227    // Step 2: record bytes_before.
228    let bytes_before = fs::metadata(&paths.brain_db).map(|m| m.len()).unwrap_or(0);
229
230    // Step 3: purge invalidated memories (optional, gated by caller).
231    let invalidated_memories_purged = if purge_invalidated {
232        let count: i64 = conn.query_row(
233            "SELECT COUNT(*) FROM memories WHERE invalidated_at IS NOT NULL",
234            [],
235            |r| r.get(0),
236        )?;
237        conn.execute_batch(
238            "DELETE FROM memories_fts WHERE memory_id IN (
239                 SELECT memory_id FROM memories WHERE invalidated_at IS NOT NULL
240             );
241             DELETE FROM memories WHERE invalidated_at IS NOT NULL;",
242        )?;
243        count as u64
244    } else {
245        0
246    };
247
248    // Step 4: trim old events (optional, gated by caller).
249    let events_trimmed = if let Some(dur) = trim_events_older_than {
250        // Compute the cutoff as an RFC 3339 string (UTC) so it compares
251        // correctly against the TEXT `ts` column.
252        let cutoff_secs = dur.as_secs();
253        let now_unix = std::time::SystemTime::now()
254            .duration_since(std::time::UNIX_EPOCH)
255            .map(|d| d.as_secs())
256            .unwrap_or(0);
257        let cutoff_unix = now_unix.saturating_sub(cutoff_secs);
258        // Format as a naive UTC RFC 3339 string (matches the stored format).
259        let cutoff_rfc3339 = {
260            let secs = cutoff_unix as i64;
261            // Use the `time` crate (already a dependency of projector.rs).
262            use time::OffsetDateTime;
263            use time::format_description::well_known::Rfc3339;
264            OffsetDateTime::from_unix_timestamp(secs)
265                .map_err(|e| format!("compact_brain: invalid cutoff timestamp: {e}"))?
266                .format(&Rfc3339)
267                .map_err(|e| format!("compact_brain: failed to format cutoff: {e}"))?
268        };
269        let count: i64 = conn.query_row(
270            "SELECT COUNT(*) FROM events WHERE ts < ?1",
271            rusqlite::params![cutoff_rfc3339],
272            |r| r.get(0),
273        )?;
274        conn.execute(
275            "DELETE FROM events WHERE ts < ?1",
276            rusqlite::params![cutoff_rfc3339],
277        )?;
278        count as u64
279    } else {
280        0
281    };
282
283    // Step 5: VACUUM — must run outside any active transaction.
284    // `rusqlite::Connection` does not hold an implicit transaction here so
285    // execute_batch is safe.
286    conn.execute_batch("VACUUM;")?;
287
288    // Step 6: Checkpoint the WAL so bytes_after reflects the real file size
289    // (on systems without WAL mode this is a no-op).
290    conn.execute_batch("PRAGMA wal_checkpoint(TRUNCATE);")?;
291
292    let bytes_after = fs::metadata(&paths.brain_db).map(|m| m.len()).unwrap_or(0);
293
294    Ok(CompactReport {
295        bytes_before,
296        bytes_after,
297        events_trimmed,
298        invalidated_memories_purged,
299    })
300}
kimetsu_brain/maintenance.rs

kimetsu_brain/
maintenance.rs
