From 3bc6667243d0750955919702beb826ad468fceed Mon Sep 17 00:00:00 2001 From: Jonathan Davies Date: Mon, 17 Aug 2026 21:49:56 +0000 Subject: [PATCH 1/5] refactor(store): split store.rs into a store/ module directory The SQLite backend has been split across a store/ folder, with each file grouped by an individual concern. --- crates/icm-store/src/store.rs | 8759 --------------------- crates/icm-store/src/store/facts.rs | 244 + crates/icm-store/src/store/feedback.rs | 302 + crates/icm-store/src/store/maintenance.rs | 309 + crates/icm-store/src/store/memoir.rs | 654 ++ crates/icm-store/src/store/memory.rs | 842 ++ crates/icm-store/src/store/mod.rs | 1302 +++ crates/icm-store/src/store/patterns.rs | 391 + crates/icm-store/src/store/tests.rs | 4251 ++++++++++ crates/icm-store/src/store/transcript.rs | 509 ++ 10 files changed, 8804 insertions(+), 8759 deletions(-) delete mode 100644 crates/icm-store/src/store.rs create mode 100644 crates/icm-store/src/store/facts.rs create mode 100644 crates/icm-store/src/store/feedback.rs create mode 100644 crates/icm-store/src/store/maintenance.rs create mode 100644 crates/icm-store/src/store/memoir.rs create mode 100644 crates/icm-store/src/store/memory.rs create mode 100644 crates/icm-store/src/store/mod.rs create mode 100644 crates/icm-store/src/store/patterns.rs create mode 100644 crates/icm-store/src/store/tests.rs create mode 100644 crates/icm-store/src/store/transcript.rs diff --git a/crates/icm-store/src/store.rs b/crates/icm-store/src/store.rs deleted file mode 100644 index 2929df56..00000000 --- a/crates/icm-store/src/store.rs +++ /dev/null @@ -1,8759 +0,0 @@ -use std::collections::{HashMap, HashSet, VecDeque}; -use std::num::NonZeroUsize; -use std::path::Path; -use std::sync::{Mutex, Once}; - -use chrono::{DateTime, Utc}; -use lru::LruCache; -use rusqlite::{ffi::sqlite3_auto_extension, params, Connection}; -use sha2::{Digest, Sha256}; -use zerocopy::IntoBytes; - -use icm_core::{ - Concept, ConceptLink, Embedder, Fact, FactsStats, FactsStore, Feedback, FeedbackStats, - FeedbackStore, IcmError, IcmResult, Importance, Label, Memoir, MemoirStats, MemoirStore, - Memory, MemorySource, MemoryStore, Message, PatternCluster, Relation, Role, Session, - StoreStats, TopicHealth, TranscriptHit, TranscriptStats, TranscriptStore, -}; - -use crate::schema::init_db_with_dims; - -/// Convert rusqlite::Error to IcmError::Database -pub(crate) fn db_err(e: rusqlite::Error) -> IcmError { - IcmError::Database(e.to_string()) -} - -/// True when a rusqlite error is a "no such table" (a legacy DB missing an -/// FTS shadow table we optionally rebuild — issue #313). -fn is_missing_table(e: &rusqlite::Error) -> bool { - e.to_string().contains("no such table") -} - -/// FTS5 shadow tables maintained by ICM, checked/rebuilt during repair (#313). -const FTS_TABLES: [&str; 4] = [ - "memories_fts", - "concepts_fts", - "feedback_fts", - "messages_fts", -]; - -// Shared public row types live in `crate::common` so all backends can be -// compiled into one binary without colliding definitions (issue #301). -pub use crate::common::{CodeArea, HookEvent, HookEventInsert, HookStatsRow, PendingRow}; - -/// Collect mapped rows into a Vec, converting rusqlite errors. -fn collect_rows( - rows: rusqlite::MappedRows<'_, impl FnMut(&rusqlite::Row<'_>) -> rusqlite::Result>, -) -> IcmResult> { - rows.collect::, _>>().map_err(db_err) -} - -static SQLITE_VEC_INIT: Once = Once::new(); - -fn ensure_sqlite_vec() { - SQLITE_VEC_INIT.call_once(|| unsafe { - #[allow(clippy::missing_transmute_annotations)] - sqlite3_auto_extension(Some(std::mem::transmute( - sqlite_vec::sqlite3_vec_init as *const (), - ))); - }); -} - -/// URI-encode a filesystem path for a SQLite `file:` URI so a backslash on -/// Windows or a `?`/`#`/`%` in a pathological filename can't break the parser. -fn encode_sqlite_uri_path(path: &Path) -> String { - path.to_string_lossy() - .chars() - .map(|c| match c { - '?' | '#' | '%' => format!("%{:02X}", c as u32), - // Normalize Windows backslashes; SQLite URIs accept "/". - '\\' => "/".into(), - other => other.to_string(), - }) - .collect() -} - -/// Open `path` strictly read-only. When `immutable` is set, add the -/// `immutable=1` URI flag — SQLite then assumes the file never changes and -/// touches no `-shm`/`-wal` sidecars, which is required on a `chmod -w` -/// parent directory (issue #263) but serves a permanently stale snapshot and -/// eventually reports spurious `SQLITE_CORRUPT` on a live DB (issue #319). -/// Plain `mode=ro` (immutable = false) is WAL-aware and sees committed -/// writes, at the cost of needing a writable directory for the sidecars. -fn open_readonly_uri(path: &Path, immutable: bool) -> IcmResult { - let encoded = encode_sqlite_uri_path(path); - let uri = if immutable { - format!("file:{encoded}?mode=ro&immutable=1") - } else { - format!("file:{encoded}?mode=ro") - }; - Connection::open_with_flags( - uri, - rusqlite::OpenFlags::SQLITE_OPEN_READ_ONLY - | rusqlite::OpenFlags::SQLITE_OPEN_URI - | rusqlite::OpenFlags::SQLITE_OPEN_NO_MUTEX, - ) - .map_err(|e| IcmError::Database(format!("cannot open database read-only: {e}"))) -} - -/// Open a long-lived read-only connection (issue #319). -/// -/// Prefer a normal WAL-aware `mode=ro` connection: it respects locking and -/// sees writes committed after it opened — the actual deployment model for -/// `icm --read-only serve`, where hooks keep writing the same DB. Fall back to -/// `immutable=1` only when the live open can't even read the DB, e.g. a -/// `chmod -w` sandbox where SQLite can't create the `-shm` sidecar for a -/// WAL-mode file (issue #263). The read probe is essential: on such a -/// directory the open may *succeed* yet the first real read fails, so opening -/// alone is not a sufficient signal. -fn open_readonly_connection(path: &Path) -> IcmResult { - if let Ok(conn) = open_readonly_uri(path, false) { - // Give a momentarily-locked writer time to release before deciding - // the live open is unusable. - let _ = conn.execute_batch("PRAGMA busy_timeout=30000;"); - // Exercise a real table read (touches the WAL/-shm path) — `SELECT 1` - // would not. - if conn - .query_row("SELECT count(*) FROM sqlite_master", [], |_| Ok(())) - .is_ok() - { - return Ok(conn); - } - } - // Live open unusable (e.g. a read-only sandbox dir, #263). Fall back to an - // immutable snapshot — but warn, because a long-lived reader on this - // connection will NOT see subsequent writes (the #319 staleness tradeoff). - tracing::warn!( - path = %path.display(), - "read-only DB opened immutable (sandbox fallback): writes committed after \ - this point will not be visible until the connection is reopened" - ); - open_readonly_uri(path, true) -} - -/// In-process LRU cache size for hot memories. Each entry is one -/// fully-hydrated `Memory` (incl. optional 384×f32 embedding ≈ 1.5KB), -/// so 256 entries cap RAM at ~400KB worst case. Helps long-running -/// processes (`icm serve`, TUI) where the same memories are read -/// repeatedly; zero benefit in one-shot CLI invocations beyond the -/// single recall flow. -const MEMORY_CACHE_CAP: usize = 256; - -pub struct SqliteStore { - conn: Connection, - cache: Mutex>, - /// `true` when opened through [`Self::open_readonly`]. Read-like - /// methods that would otherwise dirty the DB (auto-decay, - /// `update_access`) check this and skip silently; mutation methods - /// (`store`, `update`, `delete`, etc.) check this and return - /// `IcmError::ReadOnly`. Issue #263. - readonly: bool, -} - -impl SqliteStore { - pub fn new(path: &Path) -> IcmResult { - Self::with_dims(path, icm_core::DEFAULT_EMBEDDING_DIMS) - } - - /// Open an existing database in read-only mode (issue #263). - /// - /// Differences vs [`Self::with_dims`]: - /// - The parent directory is NOT created. - /// - The connection is opened with `SQLITE_OPEN_READ_ONLY` — SQLite - /// itself refuses any DDL/DML that the application might miss. - /// - No `PRAGMA journal_mode=WAL` (WAL requires writable access). - /// - No `init_db_with_dims` (schema migration would mutate the DB). - /// - /// Returns an error if the file is absent (caller may want to fall - /// through to writable mode then). Use [`std::path::Path::exists`] - /// at the call site if you need a missing-DB fast path. - pub fn open_readonly(path: &Path) -> IcmResult { - ensure_sqlite_vec(); - if !path.exists() { - return Err(IcmError::NotFound(format!( - "database not found at {}", - path.display() - ))); - } - let conn = open_readonly_connection(path)?; - // foreign_keys is a no-op for reads; busy_timeout is still useful - // when another writer holds the file. - conn.execute_batch("PRAGMA foreign_keys=ON; PRAGMA busy_timeout=30000;") - .map_err(db_err)?; - Ok(Self { - conn, - cache: Mutex::new(new_cache()), - readonly: true, - }) - } - - /// Open an existing database for maintenance — integrity check and - /// repair (issue #313). - /// - /// Writable (so `REINDEX` and FTS `'rebuild'` can run) but, unlike - /// [`Self::with_dims`], it deliberately does NOT: - /// - run `init_db_with_dims` — schema migration would fail on, or mutate, - /// a corrupt DB before it can even be inspected; - /// - switch `journal_mode` — a damaged file's on-disk format is left - /// exactly as found so recovery reasons about the real state. - /// - /// Returns [`IcmError::NotFound`] when the file is absent. - pub fn open_maintenance(path: &Path) -> IcmResult { - ensure_sqlite_vec(); - if !path.exists() { - return Err(IcmError::NotFound(format!( - "database not found at {}", - path.display() - ))); - } - let conn = Connection::open(path) - .map_err(|e| IcmError::Database(format!("cannot open database: {e}")))?; - conn.execute_batch("PRAGMA busy_timeout=30000;") - .map_err(db_err)?; - Ok(Self { - conn, - cache: Mutex::new(new_cache()), - readonly: false, - }) - } - - /// Check database integrity (issue #313) and return a list of problems. - /// A healthy database yields exactly `["ok"]`; a damaged one yields one - /// entry per problem. - /// - /// Two complementary checks run: - /// 1. `PRAGMA integrity_check` — structural b-tree / page validation. - /// This is what caught the shadow-table and index damage reported in - /// the incident (`btreeInitPage`, `wrong # of entries in index …`). - /// 2. FTS5 `'integrity-check'` per shadow table — validates each FTS - /// index's internal structure, complementing the structural pass for - /// damage confined to the FTS shadow tables. - /// - /// This never returns `Err`: even a failure to *run* a check (e.g. an FTS - /// vtable too damaged to instantiate) is recorded as a problem, so the - /// caller — `icm doctor` / `icm repair` — always gets a usable verdict on - /// a badly corrupt database instead of a propagated error. - pub fn integrity_check(&self) -> IcmResult> { - let mut problems = Vec::new(); - - // 1. Structural check. Record a run failure as a problem instead of - // aborting the whole verdict. - match self.run_integrity_pragma() { - Ok(lines) => problems.extend(lines.into_iter().filter(|l| l != "ok")), - Err(e) => problems.push(format!("integrity_check pragma failed: {e}")), - } - - // 2. Per-FTS-table consistency. - for table in FTS_TABLES { - // `rank = 1` makes FTS5 verify the index against the *content* - // table, not just its own internal structure. Without it, an - // index that is stale or out of step with the base table (e.g. - // an interrupted write) is reported as healthy. Requires - // SQLite ≥ 3.37 (bundled rusqlite is well past that). - let sql = format!("INSERT INTO {table}({table}, rank) VALUES('integrity-check', 1);"); - match self.conn.execute_batch(&sql) { - Ok(()) => {} - Err(e) if is_missing_table(&e) => {} // legacy DB without this table - Err(e) => problems.push(format!("fts5 {table}: {e}")), - } - } - - if problems.is_empty() { - Ok(vec!["ok".to_string()]) - } else { - Ok(problems) - } - } - - /// Structural-only integrity check (`PRAGMA integrity_check`), safe on a - /// read-only connection (issue #313 follow-up). Unlike - /// [`Self::integrity_check`] it does NOT run the FTS5 `'integrity-check'` - /// (which is an `INSERT` and needs a writable connection), so it never - /// mutates the DB or triggers a WAL checkpoint. Used by the read-only - /// inspection paths (`icm doctor`, `icm repair --dry-run`). A healthy DB - /// yields `["ok"]`. Never returns `Err`. - pub fn integrity_check_structural(&self) -> IcmResult> { - let problems: Vec = match self.run_integrity_pragma() { - Ok(lines) => lines.into_iter().filter(|l| l != "ok").collect(), - Err(e) => vec![format!("integrity_check pragma failed: {e}")], - }; - if problems.is_empty() { - Ok(vec!["ok".to_string()]) - } else { - Ok(problems) - } - } - - /// Run `PRAGMA integrity_check` and collect its result rows. - fn run_integrity_pragma(&self) -> IcmResult> { - let mut stmt = self - .conn - .prepare("PRAGMA integrity_check") - .map_err(db_err)?; - let rows = stmt - .query_map([], |row| row.get::<_, String>(0)) - .map_err(db_err)?; - let mut out = Vec::new(); - for r in rows { - out.push(r.map_err(db_err)?); - } - Ok(out) - } - - /// Rebuild the FTS5 shadow tables from their content tables and `REINDEX` - /// every b-tree index (issue #313). This repairs the most common - /// corruption class — damaged indexes / FTS shadow tables with intact - /// base tables — without touching row data. - /// - /// Best-effort by design: a shadow table too damaged for `'rebuild'` to - /// even instantiate is skipped rather than aborting the whole repair, and - /// `REINDEX` failure is tolerated too. The caller re-runs - /// [`Self::integrity_check`] afterwards and reports any damage that - /// survived, so nothing is silently claimed fixed. Returns the FTS tables - /// that were successfully rebuilt. - pub fn rebuild_search_indexes(&self) -> IcmResult> { - let mut rebuilt = Vec::new(); - for table in FTS_TABLES { - let sql = format!("INSERT INTO {table}({table}) VALUES('rebuild');"); - // A missing (legacy DB) or too-corrupt-to-instantiate shadow table - // is skipped rather than aborting; the post-repair integrity check - // surfaces any table that could not be restored. - if self.conn.execute_batch(&sql).is_ok() { - rebuilt.push(table.to_string()); - } - } - // May fail on a badly damaged b-tree; the post-repair integrity check - // reports whatever REINDEX could not fix. - let _ = self.conn.execute_batch("REINDEX;"); - Ok(rebuilt) - } - - /// True when the store was opened read-only (issue #263). Read-like - /// methods skip side-effect mutations; write methods return - /// `IcmError::ReadOnly`. - #[must_use] - pub fn is_readonly(&self) -> bool { - self.readonly - } - - /// Peek `icm_metadata.embedding_dims` without running any schema - /// migration. Returns `Ok(None)` when the DB file is absent, the - /// metadata table doesn't exist (legacy DB), or the row is missing. - /// - /// Use this *before* calling [`Self::with_dims`] when running in a - /// mode that must not trigger a destructive vector recreate — most - /// notably the `--no-embeddings` path (issue #267): if the caller - /// has no embedder loaded, `with_dims` would otherwise fall back to - /// `DEFAULT_EMBEDDING_DIMS`, mismatch the stored value, and silently - /// DROP `vec_memories` while NULL-ing every `memories.embedding`. - pub fn read_stored_embedding_dims(path: &Path) -> IcmResult> { - if !path.exists() { - return Ok(None); - } - // Open strictly immutable so this helper survives a `chmod -w` - // sandbox (issue #263 interaction). `SQLITE_OPEN_READ_ONLY` - // alone is NOT enough — SQLite still tries to create/update - // the `-shm` / `-wal` companion files for any WAL-mode DB, - // which fails when the parent directory is non-writable. - // The `immutable=1` URI flag tells SQLite the file will not - // change during the connection's lifetime and stops it from - // touching WAL infrastructure entirely. This is a one-shot probe - // (not a long-lived connection), so the staleness that #319 fixes - // for `open_readonly` does not apply here. - let conn = open_readonly_uri(path, true)?; - // Probe for the metadata table — legacy DBs predate it. - let has_table: bool = conn - .query_row( - "SELECT COUNT(*) > 0 FROM sqlite_master - WHERE type = 'table' AND name = 'icm_metadata'", - [], - |row| row.get(0), - ) - .map_err(db_err)?; - if !has_table { - return Ok(None); - } - let row: Option = conn - .query_row( - "SELECT value FROM icm_metadata WHERE key = 'embedding_dims'", - [], - |row| row.get(0), - ) - .optional() - .map_err(db_err)?; - Ok(row.and_then(|s| s.parse().ok())) - } - - /// Open or create a store with a specific embedding dimension. - pub fn with_dims(path: &Path, embedding_dims: usize) -> IcmResult { - ensure_sqlite_vec(); - if let Some(parent) = path.parent() { - std::fs::create_dir_all(parent) - .map_err(|e| IcmError::Database(format!("cannot create db directory: {e}")))?; - } - let conn = Connection::open(path) - .map_err(|e| IcmError::Database(format!("cannot open database: {e}")))?; - // Schema/PRAGMA setup races with other processes opening the same - // brand-new DB simultaneously (found via real concurrent testing: - // 10 processes opening one fresh DB, several hung, others errored, - // zero succeeded). Both the WAL-mode switch (needs a brief - // exclusive lock to convert a fresh file — busy_timeout must be - // set first in the same batch, or this statement itself has no - // timeout active yet) and init_db_with_dims's schema creation - // (BEGIN IMMEDIATE-wrapped in schema.rs, but SQLite's FTS5 - // virtual-table module can still surface a transient error on the - // loser even so) are retried together here: whatever the winner - // already committed, a fresh attempt's PRAGMA + existence checks - // correctly see and no-op past it. Jittered, not just linear, - // backoff: a fixed schedule lets many racing processes retry in - // near-lockstep and collide again and again. - let mut last_err = None; - for attempt in 0..40u32 { - if attempt > 0 { - let base_ms = (attempt as u64).min(20) * 15; - let jitter_ms = std::time::SystemTime::now() - .duration_since(std::time::UNIX_EPOCH) - .map(|d| u64::from(d.subsec_nanos()) % 40) - .unwrap_or(0); - std::thread::sleep(std::time::Duration::from_millis(base_ms + jitter_ms)); - } - // A short busy_timeout during this retry loop, not the normal - // 30s: 30s is meant to tolerate *ordinary* write contention - // during real use (e.g. a hook write racing a consolidate), - // but stacked with up to 40 outer attempts here it turns into - // a potentially multi-minute worst case under real multi- - // process contention (measured: several real `icm` processes - // hung well past 60s with the 30s inner timeout) — the outer - // jittered loop is what actually provides the robustness here, - // so the inner SQLite-level wait only needs to be long enough - // to smooth over a single competing transaction, not to be a - // retry mechanism in its own right. Restored to 30s below once - // the schema is confirmed present. - let attempt_result = conn - .execute_batch( - "PRAGMA busy_timeout=1000; PRAGMA journal_mode=WAL; PRAGMA foreign_keys=ON;", - ) - .map_err(db_err) - .and_then(|()| init_db_with_dims(&conn, embedding_dims)); - match attempt_result { - Ok(()) => { - last_err = None; - break; - } - Err(e) => { - let msg = e.to_string(); - let transient = msg.contains("vtable constructor failed") - || msg.contains("already exists") - || msg.contains("database is locked") - || msg.contains("database is busy"); - last_err = Some(e); - if !transient { - break; - } - } - } - } - if let Some(e) = last_err { - return Err(e); - } - conn.execute_batch("PRAGMA busy_timeout=30000;") - .map_err(db_err)?; - - Ok(Self { - conn, - cache: Mutex::new(new_cache()), - readonly: false, - }) - } - - /// Apply decay if more than 24 hours since last decay. - /// Called automatically on recall to avoid manual `icm decay` cron. - /// - /// No-op when the store is read-only (issue #263): recall must work - /// against a DB the process cannot write to, and the bookkeeping - /// writes here would otherwise abort the whole read with - /// "attempt to write a readonly database". - pub fn maybe_auto_decay(&self) -> IcmResult<()> { - if self.readonly { - return Ok(()); - } - let now = Utc::now(); - let now_str = now.to_rfc3339(); - - // Audit finding: this used to apply a flat 0.95 step whenever >= 1 - // day had passed, regardless of HOW MANY days had actually passed — - // a machine touched once a week decayed at 0.95/week (≈0.993/day) - // instead of the documented 0.95/day. Read the previous timestamp - // first so the step can be `0.95 ^ elapsed_days` (compounded), - // matching the documented per-day rate regardless of gaps between - // calls. (The 0.95 base itself stays hardcoded here — wiring the - // CLI's configurable `decay_rate` through to this crate is a - // separate, out-of-scope change.) - let last_decay_at: Option = self - .conn - .query_row( - "SELECT value FROM icm_metadata WHERE key = 'last_decay_at'", - [], - |row| row.get(0), - ) - .optional() - .map_err(db_err)?; - let elapsed_days = last_decay_at - .as_deref() - .and_then(|prev| prev.parse::>().ok()) - .map(|prev| (now - prev).num_seconds() as f64 / 86_400.0) - .filter(|d| d.is_finite() && *d > 0.0) - .unwrap_or(1.0); // first run ever: preserve the historical single-step behavior - - // Atomic check-and-update: only one caller wins the race. (A narrow - // window between the read above and this claim could let a losing - // racer's `elapsed_days` be computed from a slightly stale - // timestamp, but only one claim ever succeeds — a day or two of - // imprecision in a decay RATE is harmless, not worth a stricter - // compare-and-swap loop.) - let changed = self - .conn - .execute( - "INSERT INTO icm_metadata (key, value) VALUES ('last_decay_at', ?1) - ON CONFLICT(key) DO UPDATE SET value = ?1 - WHERE value IS NULL OR julianday(?1) - julianday(value) >= 1.0", - params![now_str], - ) - .map_err(db_err)?; - - if changed > 0 { - let factor = 0.95_f64.powf(elapsed_days) as f32; - self.apply_decay(factor)?; - } - - Ok(()) - } - - /// Atomically increment the hook call counter and return the new value. - pub fn increment_hook_counter(&self) -> IcmResult { - let count: usize = self - .conn - .query_row( - "INSERT INTO icm_metadata (key, value) VALUES ('hook_counter', '1') - ON CONFLICT(key) DO UPDATE SET value = CAST(CAST(value AS INTEGER) + 1 AS TEXT) - RETURNING CAST(value AS INTEGER)", - [], - |row| row.get(0), - ) - .map_err(db_err)?; - Ok(count) - } - - /// Reset the hook call counter to 0. - pub fn reset_hook_counter(&self) -> IcmResult<()> { - self.conn - .execute( - "INSERT INTO icm_metadata (key, value) VALUES ('hook_counter', '0') - ON CONFLICT(key) DO UPDATE SET value = '0'", - [], - ) - .map_err(db_err)?; - Ok(()) - } - - // ── Async extraction queue ───────────────────────────────────────── - // - // Row tuple shape: `(id, project, tool_name, raw_output, captured_at)` - // - // When `[extraction.summarizer].provider` is set to something other - // than `"none"`, PostToolUse hooks INSERT raw tool output here in - // ~50ms (no embedder load) and a worker (`icm extract-pending` or - // the SessionEnd async fork) dequeues batches and runs the LLM CLI. - - /// Enqueue raw tool output for later LLM extraction. Returns the - /// generated row id so the caller can correlate logs. - pub fn enqueue_pending_extraction( - &self, - project: &str, - tool_name: &str, - raw_output: &str, - ) -> IcmResult { - let id = ulid::Ulid::new().to_string(); - let now = chrono::Utc::now().to_rfc3339(); - self.conn - .execute( - "INSERT INTO pending_extractions (id, project, tool_name, raw_output, captured_at) - VALUES (?1, ?2, ?3, ?4, ?5)", - rusqlite::params![id, project, tool_name, raw_output, now], - ) - .map_err(db_err)?; - Ok(id) - } - - /// Pop up to `limit` oldest pending rows. Caller is expected to call - /// `delete_pending_extractions` after successful processing. - pub fn list_pending_extractions(&self, limit: usize) -> IcmResult> { - let mut stmt = self - .conn - .prepare( - "SELECT id, project, tool_name, raw_output, captured_at - FROM pending_extractions - ORDER BY captured_at ASC - LIMIT ?1", - ) - .map_err(db_err)?; - let rows = stmt - .query_map([limit as i64], |row| { - Ok(( - row.get::<_, String>(0)?, - row.get::<_, String>(1)?, - row.get::<_, String>(2)?, - row.get::<_, String>(3)?, - row.get::<_, String>(4)?, - )) - }) - .map_err(db_err)? - .collect::, _>>() - .map_err(db_err)?; - Ok(rows) - } - - /// Delete pending rows by id. Used after a worker has processed them. - pub fn delete_pending_extractions(&self, ids: &[String]) -> IcmResult { - if ids.is_empty() { - return Ok(0); - } - let placeholders = ids.iter().map(|_| "?").collect::>().join(","); - let sql = format!("DELETE FROM pending_extractions WHERE id IN ({placeholders})"); - let params: Vec<&dyn rusqlite::ToSql> = - ids.iter().map(|s| s as &dyn rusqlite::ToSql).collect(); - let n = self.conn.execute(&sql, params.as_slice()).map_err(db_err)?; - Ok(n) - } - - /// Total rows currently waiting in the queue. Used by `icm doctor`. - pub fn pending_extraction_count(&self) -> IcmResult { - let n: i64 = self - .conn - .query_row("SELECT COUNT(*) FROM pending_extractions", [], |r| r.get(0)) - .map_err(db_err)?; - Ok(n as usize) - } - - // ── Code areas (auto-captured file edits — issue #196) ──────────── - // - // `cmd_hook_post` calls `upsert_code_area` whenever the upstream - // tool was Edit / Write / MultiEdit / NotebookEdit. Same project + - // file_path => touch_count++ via ON CONFLICT. - - /// Insert or refresh a row for `(project, file_path)`. On conflict - /// bumps `touch_count`, updates `last_touched_at`, refreshes - /// `session_id` / `tool_name`, and only overwrites `description` if - /// the caller passes `Some` (so the most recent meaningful hint - /// wins without clobbering an existing one with `None`). - pub fn upsert_code_area( - &self, - project: &str, - file_path: &str, - description: Option<&str>, - session_id: Option<&str>, - tool_name: Option<&str>, - ) -> IcmResult<()> { - let now = chrono::Utc::now().to_rfc3339(); - self.conn - .execute( - "INSERT INTO code_areas (project, file_path, description, - session_id, tool_name, touch_count, - first_touched_at, last_touched_at) - VALUES (?1, ?2, ?3, ?4, ?5, 1, ?6, ?6) - ON CONFLICT(project, file_path) DO UPDATE SET - touch_count = touch_count + 1, - last_touched_at = excluded.last_touched_at, - session_id = COALESCE(excluded.session_id, session_id), - tool_name = COALESCE(excluded.tool_name, tool_name), - description = COALESCE(excluded.description, description)", - rusqlite::params![project, file_path, description, session_id, tool_name, now], - ) - .map_err(db_err)?; - Ok(()) - } - - /// List code areas, optionally filtered by project / file_path / - /// since timestamp. `limit` caps the result count (use `usize::MAX` - /// to disable). Ordered by `last_touched_at DESC` so the freshest - /// edits come first. - pub fn list_code_areas( - &self, - project: Option<&str>, - in_file: Option<&str>, - since: Option>, - limit: usize, - ) -> IcmResult> { - let mut sql = String::from( - "SELECT id, project, file_path, description, session_id, tool_name, - touch_count, first_touched_at, last_touched_at - FROM code_areas - WHERE 1=1", - ); - let mut params: Vec> = Vec::new(); - if let Some(p) = project { - sql.push_str(" AND project = ?"); - params.push(Box::new(p.to_string())); - } - if let Some(f) = in_file { - // Match either an exact file_path or a path that ends with - // the provided fragment so users can pass a short suffix. - sql.push_str(" AND (file_path = ? OR file_path LIKE ?)"); - params.push(Box::new(f.to_string())); - params.push(Box::new(format!("%/{f}"))); - } - if let Some(t) = since { - sql.push_str(" AND last_touched_at >= ?"); - params.push(Box::new(t.to_rfc3339())); - } - sql.push_str(" ORDER BY last_touched_at DESC LIMIT ?"); - params.push(Box::new(limit as i64)); - - let mut stmt = self.conn.prepare(&sql).map_err(db_err)?; - let param_refs: Vec<&dyn rusqlite::ToSql> = params - .iter() - .map(|p| p.as_ref() as &dyn rusqlite::ToSql) - .collect(); - let rows = stmt - .query_map(param_refs.as_slice(), |row| { - let first: String = row.get(7)?; - let last: String = row.get(8)?; - Ok(CodeArea { - id: row.get(0)?, - project: row.get(1)?, - file_path: row.get(2)?, - description: row.get(3)?, - session_id: row.get(4)?, - tool_name: row.get(5)?, - touch_count: row.get(6)?, - first_touched_at: DateTime::parse_from_rfc3339(&first) - .map(|d| d.with_timezone(&Utc)) - .unwrap_or_else(|_| Utc::now()), - last_touched_at: DateTime::parse_from_rfc3339(&last) - .map(|d| d.with_timezone(&Utc)) - .unwrap_or_else(|_| Utc::now()), - }) - }) - .map_err(db_err)? - .collect::, _>>() - .map_err(db_err)?; - Ok(rows) - } - - /// Total rows in `code_areas`. Cheap; used by stats / doctor. - pub fn code_area_count(&self) -> IcmResult { - let n: i64 = self - .conn - .query_row("SELECT COUNT(*) FROM code_areas", [], |r| r.get(0)) - .map_err(db_err)?; - Ok(n as usize) - } - - // ── Hook telemetry ───────────────────────────────────────────────── - // - // Every `icm hook ` fire writes one row to `hook_events`. Read - // back via `hook_events_recent` / `hook_stats`. Inserts are designed - // to be cheap (single statement, no FTS) so they stay well under the - // <50ms async-path budget. - - /// Append one hook telemetry row. Errors are swallowed by callers in - /// hook paths (logging must never block the user), but tests can - /// inspect the `Result`. - pub fn record_hook_event(&self, ev: &HookEventInsert) -> IcmResult { - let now = chrono::Utc::now().to_rfc3339(); - self.conn - .execute( - "INSERT INTO hook_events - (ts, event, project, session_id, tool_name, - duration_ms, exit_code, payload_size, note) - VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9)", - rusqlite::params![ - now, - ev.event, - ev.project, - ev.session_id, - ev.tool_name, - ev.duration_ms, - ev.exit_code, - ev.payload_size, - ev.note, - ], - ) - .map_err(db_err)?; - Ok(self.conn.last_insert_rowid()) - } - - /// Most recent `limit` hook events, newest first. Optional `event` - /// filter (e.g. `Some("end")` to see only SessionEnd hooks). - pub fn hook_events_recent( - &self, - limit: usize, - event_filter: Option<&str>, - ) -> IcmResult> { - let limit_i64 = limit as i64; - let row_to_event = |row: &rusqlite::Row<'_>| -> rusqlite::Result { - let ts_str: String = row.get(1)?; - let ts = chrono::DateTime::parse_from_rfc3339(&ts_str) - .map(|t| t.with_timezone(&Utc)) - .unwrap_or_else(|_| Utc::now()); - Ok(HookEvent { - id: row.get(0)?, - ts, - event: row.get(2)?, - project: row.get(3)?, - session_id: row.get(4)?, - tool_name: row.get(5)?, - duration_ms: row.get(6)?, - exit_code: row.get(7)?, - payload_size: row.get(8)?, - note: row.get(9)?, - }) - }; - match event_filter { - Some(e) => { - let mut stmt = self - .conn - .prepare( - "SELECT id, ts, event, project, session_id, tool_name, - duration_ms, exit_code, payload_size, note - FROM hook_events - WHERE event = ?1 - ORDER BY id DESC - LIMIT ?2", - ) - .map_err(db_err)?; - let rows = stmt - .query_map(rusqlite::params![e, limit_i64], row_to_event) - .map_err(db_err)?; - collect_rows(rows) - } - None => { - let mut stmt = self - .conn - .prepare( - "SELECT id, ts, event, project, session_id, tool_name, - duration_ms, exit_code, payload_size, note - FROM hook_events - ORDER BY id DESC - LIMIT ?1", - ) - .map_err(db_err)?; - let rows = stmt - .query_map(rusqlite::params![limit_i64], row_to_event) - .map_err(db_err)?; - collect_rows(rows) - } - } - } - - /// Aggregate counts and latency percentiles per event type, over a - /// time window starting `since` (RFC3339). Used by `icm hook-stats`. - pub fn hook_stats(&self, since_rfc3339: &str) -> IcmResult> { - // Pull each event type and compute percentiles in Rust — SQLite - // has no native percentile function and the row count is small - // enough (~1k/day worst case) that an in-process sort is fine. - let mut stmt = self - .conn - .prepare( - "SELECT event, duration_ms, exit_code - FROM hook_events - WHERE ts >= ?1 - ORDER BY event", - ) - .map_err(db_err)?; - let rows = stmt - .query_map([since_rfc3339], |r| { - Ok(( - r.get::<_, String>(0)?, - r.get::<_, Option>(1)?, - r.get::<_, i32>(2)?, - )) - }) - .map_err(db_err)?; - let mut by_event: std::collections::BTreeMap, i32)>> = - std::collections::BTreeMap::new(); - for r in rows { - let (ev, dur, exit) = r.map_err(db_err)?; - by_event.entry(ev).or_default().push((dur, exit)); - } - let mut out = Vec::with_capacity(by_event.len()); - for (event, mut items) in by_event { - let count = items.len() as i64; - let error_count = items.iter().filter(|(_, e)| *e != 0).count() as i64; - let mut durations: Vec = items.iter().filter_map(|(d, _)| *d).collect(); - durations.sort_unstable(); - let avg = if durations.is_empty() { - 0.0 - } else { - durations.iter().sum::() as f64 / durations.len() as f64 - }; - let p = |q: f64| -> i64 { - if durations.is_empty() { - 0 - } else { - let idx = ((durations.len() as f64 - 1.0) * q).round() as usize; - durations[idx.min(durations.len() - 1)] - } - }; - out.push(HookStatsRow { - event, - count, - error_count, - avg_duration_ms: avg, - p50_duration_ms: p(0.50), - p99_duration_ms: p(0.99), - }); - // Avoid clippy 'unused variable' on items after move - let _ = &mut items; - } - Ok(out) - } - - /// Delete hook telemetry rows older than `cutoff_rfc3339`. Used by an - /// optional retention pass (`icm hook-log --prune-older-than ...`). - pub fn prune_hook_events(&self, cutoff_rfc3339: &str) -> IcmResult { - let n = self - .conn - .execute( - "DELETE FROM hook_events WHERE ts < ?1", - rusqlite::params![cutoff_rfc3339], - ) - .map_err(db_err)?; - Ok(n) - } - - /// Total rows currently in `hook_events`. Used by tests and `icm doctor`. - pub fn hook_event_count(&self) -> IcmResult { - let n: i64 = self - .conn - .query_row("SELECT COUNT(*) FROM hook_events", [], |r| r.get(0)) - .map_err(db_err)?; - Ok(n as usize) - } - - pub fn in_memory() -> IcmResult { - Self::in_memory_with_dims(icm_core::DEFAULT_EMBEDDING_DIMS) - } - - /// Open an in-memory store with a specific embedding dimension. - /// Useful for tests that exercise the dim-migration / dim-drift paths. - pub fn in_memory_with_dims(embedding_dims: usize) -> IcmResult { - ensure_sqlite_vec(); - let conn = Connection::open_in_memory() - .map_err(|e| IcmError::Database(format!("cannot open in-memory db: {e}")))?; - conn.execute_batch("PRAGMA foreign_keys=ON; PRAGMA busy_timeout=30000;") - .map_err(db_err)?; - init_db_with_dims(&conn, embedding_dims)?; - Ok(Self { - conn, - cache: Mutex::new(new_cache()), - readonly: false, - }) - } - - fn cache_get(&self, id: &str) -> Option { - self.cache.lock().ok().and_then(|mut c| c.get(id).cloned()) - } - - fn cache_put(&self, m: &Memory) { - if let Ok(mut c) = self.cache.lock() { - c.put(m.id.clone(), m.clone()); - } - } - - fn cache_invalidate(&self, id: &str) { - if let Ok(mut c) = self.cache.lock() { - c.pop(id); - } - } - - fn cache_invalidate_many(&self, ids: &[&str]) { - if let Ok(mut c) = self.cache.lock() { - for id in ids { - c.pop(*id); - } - } - } - - fn cache_clear(&self) { - if let Ok(mut c) = self.cache.lock() { - c.clear(); - } - } -} - -fn new_cache() -> LruCache { - let cap = NonZeroUsize::new(MEMORY_CACHE_CAP) - .expect("MEMORY_CACHE_CAP must be non-zero — see store.rs"); - LruCache::new(cap) -} - -// --------------------------------------------------------------------------- -// Memory helpers -// --------------------------------------------------------------------------- - -fn source_type(source: &MemorySource) -> &'static str { - match source { - MemorySource::ClaudeCode { .. } => "claude_code", - MemorySource::Conversation { .. } => "conversation", - MemorySource::Manual => "manual", - } -} - -fn source_data(source: &MemorySource) -> Option { - match source { - MemorySource::Manual => None, - other => serde_json::to_string(other).ok(), - } -} - -fn parse_source(source_type_str: &str, source_data_str: Option) -> MemorySource { - match source_type_str { - "manual" => MemorySource::Manual, - _ => source_data_str - .and_then(|d| serde_json::from_str(&d).ok()) - .unwrap_or(MemorySource::Manual), - } -} - -fn embedding_to_blob(embedding: &[f32]) -> Vec { - embedding.as_bytes().to_vec() -} - -fn blob_to_embedding(blob: &[u8]) -> Vec { - if !blob.len().is_multiple_of(4) { - tracing::warn!( - blob_size = blob.len(), - "embedding blob size not divisible by 4, truncating" - ); - } - blob.chunks_exact(4) - .map(|c| f32::from_le_bytes([c[0], c[1], c[2], c[3]])) - .collect() -} - -fn row_to_memory(row: &rusqlite::Row) -> rusqlite::Result { - // Column order: id(0), created_at(1), updated_at(2), last_accessed(3), - // access_count(4), weight(5), topic(6), summary(7), raw_excerpt(8), - // keywords(9), importance(10), source_type(11), source_data(12), - // related_ids(13), embedding(14) - let keywords_json: String = row.get::<_, Option>(9)?.unwrap_or_default(); - let keywords: Vec = serde_json::from_str(&keywords_json).unwrap_or_default(); - - let importance_str: String = row.get(10)?; - let importance = importance_str.parse().unwrap_or(Importance::Medium); - - let source_type_str: String = row.get(11)?; - let source_data_str: Option = row.get(12)?; - let source = parse_source(&source_type_str, source_data_str); - - let related_json: String = row.get::<_, Option>(13)?.unwrap_or_default(); - let related_ids: Vec = serde_json::from_str(&related_json).unwrap_or_default(); - - let embedding: Option> = row - .get::<_, Option>>(14)? - .map(|b| blob_to_embedding(&b)); - - let created_at_str: String = row.get(1)?; - let updated_at_str: String = row.get::<_, Option>(2)?.unwrap_or_default(); - let last_accessed_str: String = row.get(3)?; - - let created_at = parse_dt(&created_at_str); - - Ok(Memory { - id: row.get(0)?, - created_at, - updated_at: if updated_at_str.is_empty() { - created_at - } else { - parse_dt(&updated_at_str) - }, - last_accessed: parse_dt(&last_accessed_str), - access_count: row.get::<_, u32>(4)?, - weight: row.get(5)?, - topic: row.get(6)?, - summary: row.get(7)?, - raw_excerpt: row.get(8)?, - keywords, - importance, - source, - related_ids, - embedding, - scope: icm_core::Scope::User, // default for existing local memories - }) -} - -const SELECT_COLS: &str = "id, created_at, updated_at, last_accessed, access_count, weight, \ - topic, summary, raw_excerpt, keywords, \ - importance, source_type, source_data, related_ids, embedding"; - -/// Sanitize a query string for FTS5 MATCH. -/// -/// FTS5 treats characters like `-`, `*`, `"`, `:`, `^`, `+`, `~` as operators. -/// A query like `"sqlite-vec"` makes FTS5 interpret `-` as NOT and `vec` as a -/// column name, causing "no such column: vec". -/// -/// Escape `%`, `_`, and the escape character itself so a keyword can be -/// safely wrapped in a `%...%` LIKE pattern. Pair with `ESCAPE '\'` in the -/// SQL — without it, a keyword containing `%` matches every row and `_` -/// matches any single character (audit finding). -fn escape_like_wildcards(s: &str) -> String { - s.replace('\\', "\\\\") - .replace('%', "\\%") - .replace('_', "\\_") -} - -/// Cap on auxiliary metadata (transcript sessions/messages) - best-effort -/// truncation, not rejection, matching MAX_MESSAGE_BYTES's rationale. -const MAX_METADATA_BYTES: usize = 8 * 1024; - -/// Truncate `s` to at most `max` bytes without splitting a UTF-8 char. -fn truncate_at_char_boundary(s: &str, max: usize) -> &str { - if s.len() <= max { - return s; - } - let mut cut = max; - while !s.is_char_boundary(cut) { - cut -= 1; - } - &s[..cut] -} - -/// This function strips special chars and wraps each token in double quotes. -fn sanitize_fts_query(query: &str) -> String { - // Limit input length to prevent abuse (UTF-8 safe truncation) - let query = if query.len() > 10_000 { - let mut end = 10_000; - while end > 0 && !query.is_char_boundary(end) { - end -= 1; - } - &query[..end] - } else { - query - }; - - // Replace FTS5 operator chars with spaces, then quote each resulting token. - // FTS5 tokenizer (unicode61) splits on `-` too, so we must keep tokens separate. - let cleaned: String = query - .chars() - .map(|c| { - if matches!( - c, - '-' | '*' | '"' | '(' | ')' | '{' | '}' | ':' | '^' | '+' | '~' | '\\' - ) { - ' ' - } else { - c - } - }) - .collect(); - - let tokens: Vec = cleaned - .split_whitespace() - .filter(|w| !w.is_empty()) - .take(100) // Limit token count to prevent excessive query complexity - .map(|w| { - // Strip any remaining quotes from tokens before wrapping in quotes - let stripped = w.replace('"', ""); - format!("\"{stripped}\"") - }) - .collect(); - tokens.join(" ") -} - -/// Whether `e` is FTS5 rejecting a malformed MATCH query (e.g. "hello AND", -/// unbalanced parens) rather than a genuine database error. Used by -/// `search_transcripts` to degrade to "no results" instead of surfacing a -/// raw sqlite error, without pre-sanitizing the query text away from valid -/// FTS5 syntax (which callers rely on — see -/// `test_transcript_search_fts5_boolean_and_phrase`). -fn is_fts5_syntax_error(e: &rusqlite::Error) -> bool { - matches!( - e, - rusqlite::Error::SqliteFailure(_, Some(msg)) if msg.contains("fts5: syntax error") - ) -} - -// --------------------------------------------------------------------------- -// MemoryStore impl -// --------------------------------------------------------------------------- - -/// Maximum byte length of a stored summary. Audit finding: a transcript -/// containing a 1 MB unbroken text block landed as a single memory whose -/// summary was the full 1 MB blob. Caps the cost of a single bad write -/// (memory bloat, embedding compute, FTS5 index growth) to a generous -/// but bounded 64 KB. -const MAX_SUMMARY_BYTES: usize = 64 * 1024; - -/// Maximum byte length of a stored topic. Topics surface in `icm -/// topics` listings and as the routing key for project filters; a -/// thousand-byte topic is always a bug, never legitimate user input. -const MAX_TOPIC_BYTES: usize = 256; - -/// Validate and normalize a `Memory` before insertion. Trims topic -/// whitespace and rejects inputs that we know corrupt or break the -/// store: -/// -/// - Empty or whitespace-only `topic` / `summary` — these would surface -/// as blank rows in `icm topics` / `icm list` and pollute the FTS5 -/// index without conveying information. -/// - NUL byte (`\0`) in `topic` or `summary` — libsql binds text via a -/// NUL-terminated C string, so anything past the first `\0` is -/// silently dropped. Rather than silently truncate, refuse the -/// write so the caller knows. -/// - Newline / CR / tab in `topic` — these break the `icm topics` -/// tabular layout and could enable display-spoofing of topic names -/// (e.g. a topic that visually overlaps another in TUI/log output). -/// Allowed in `summary` since it's free-form prose. -/// - `topic` longer than `MAX_TOPIC_BYTES` or `summary` longer than -/// `MAX_SUMMARY_BYTES` — see the constant docs for rationale. -fn validate_and_normalize(mut memory: Memory) -> IcmResult { - memory.topic = memory.topic.trim().to_string(); - validate_fields(&memory.topic, &memory.summary)?; - Ok(memory) -} - -/// The borrowed core of [`validate_and_normalize`], shared with `update()` -/// (audit finding: the update path previously bypassed every size/content -/// check, so oversized or NUL-carrying payloads could enter the store by -/// storing small then updating big). -fn validate_fields(topic: &str, summary: &str) -> IcmResult<()> { - if topic.is_empty() { - return Err(IcmError::InvalidInput("topic cannot be empty".into())); - } - if summary.trim().is_empty() { - return Err(IcmError::InvalidInput("summary cannot be empty".into())); - } - if topic.contains('\0') { - return Err(IcmError::InvalidInput( - "topic must not contain NUL bytes".into(), - )); - } - if summary.contains('\0') { - return Err(IcmError::InvalidInput( - "summary must not contain NUL bytes".into(), - )); - } - if topic.contains(['\n', '\r', '\t']) { - return Err(IcmError::InvalidInput( - "topic must not contain newline / CR / tab characters".into(), - )); - } - if topic.len() > MAX_TOPIC_BYTES { - return Err(IcmError::InvalidInput(format!( - "topic exceeds {} bytes", - MAX_TOPIC_BYTES - ))); - } - if summary.len() > MAX_SUMMARY_BYTES { - return Err(IcmError::InvalidInput(format!( - "summary exceeds {} bytes", - MAX_SUMMARY_BYTES - ))); - } - Ok(()) -} - -/// Local total order on `Importance` (Critical > High > Medium > Low). -/// `Importance` does not implement `Ord` because the project did not -/// want to imply a globally meaningful ordering across all uses -/// (e.g. presentation, filtering). For the dedup-merge path we *do* -/// want to take the maximum so re-storing with a higher priority -/// upgrades the existing row. -fn importance_rank(i: Importance) -> u8 { - match i { - Importance::Critical => 4, - Importance::High => 3, - Importance::Medium => 2, - Importance::Low => 1, - } -} - -/// Return the higher-priority importance. Used by the dedup path so -/// `store(...)` semantics are "re-store with critical upgrades, never -/// downgrades". -fn max_importance(a: Importance, b: Importance) -> Importance { - if importance_rank(a) >= importance_rank(b) { - a - } else { - b - } -} - -/// SHA-256 over the normalized `(topic, summary)` pair, hex-encoded. -/// Normalization: trim + lowercase + collapse whitespace runs to single -/// spaces. Topic and summary are joined by `\0` to prevent boundary -/// ambiguity (e.g. `"a"|"bc"` vs `"ab"|"c"` would otherwise hash the -/// same). Used by the dedup `INSERT OR IGNORE` path. -pub(crate) fn summary_hash(topic: &str, summary: &str) -> String { - let topic_n = topic.trim().to_lowercase(); - let summary_n: String = summary - .split_whitespace() - .collect::>() - .join(" ") - .to_lowercase(); - let mut h = Sha256::new(); - h.update(topic_n.as_bytes()); - h.update(b"\0"); - h.update(summary_n.as_bytes()); - format!("{:x}", h.finalize()) -} - -impl SqliteStore { - /// Insert a memory into the database without transaction management. - /// Callers are responsible for wrapping this in a transaction. - /// - /// Dedup contract: an INSERT that collides with an existing memory on - /// `(topic, summary_hash)` is silently ignored, and the **existing** - /// row's id is returned. The caller's `memory.id` is forgotten in - /// that case. This keeps `store(...)` idempotent: writing the same - /// fact 100× ends up with one row, not 100. - fn store_inner(&self, memory: &Memory) -> IcmResult { - let keywords_json = serde_json::to_string(&memory.keywords)?; - let related_json = serde_json::to_string(&memory.related_ids)?; - let st = source_type(&memory.source); - let sd = source_data(&memory.source); - let emb_blob = memory.embedding.as_deref().map(embedding_to_blob); - let hash = summary_hash(&memory.topic, &memory.summary); - - let inserted = self - .conn - .execute( - "INSERT OR IGNORE INTO memories (id, created_at, updated_at, last_accessed, access_count, weight, - topic, summary, raw_excerpt, keywords, - importance, source_type, source_data, related_ids, embedding, summary_hash) - VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9, ?10, ?11, ?12, ?13, ?14, ?15, ?16)", - params![ - memory.id, - memory.created_at.to_rfc3339(), - memory.updated_at.to_rfc3339(), - memory.last_accessed.to_rfc3339(), - memory.access_count, - memory.weight, - memory.topic, - memory.summary, - memory.raw_excerpt, - keywords_json, - memory.importance.to_string(), - st, - sd, - related_json, - emb_blob, - hash, - ], - ) - .map_err(db_err)?; - - if inserted == 0 { - // Dedup hit: a row with the same (topic, summary_hash) - // already exists. Audit #185 H2: the previous behaviour - // returned the existing id and silently dropped the - // caller's importance / keywords / raw_excerpt. So - // running `icm store -t T -c "X" -i medium` then `icm - // store -t T -c "X" -i critical` left the importance at - // medium without warning the user. - // - // New behaviour: merge the caller's metadata into the - // existing row before returning the id. - // - importance: take the max (critical > high > medium > - // low). Re-storing with a *higher* priority upgrades. - // Re-storing with a *lower* priority is a no-op so a - // careless write can't downgrade an already-flagged - // critical memory. - // - keywords: union, preserving existing order then - // appending new ones not already present. - // - raw_excerpt: prefer the new value if non-None, - // otherwise keep existing. - // - updated_at: bumped whenever any field actually changed. - let (existing_id, existing_importance_str, existing_keywords_json, existing_raw): ( - String, - String, - String, - Option, - ) = self - .conn - .query_row( - // Audit finding: `summary_hash` already encodes the topic - // (Rust `to_lowercase()`, full Unicode) as part of the - // hash input — an additional `LOWER(topic) = LOWER(?)` - // comparison here used SQLite's built-in `LOWER()`, - // which is ASCII-only and does not fold e.g. 'É' → 'é'. - // For an all-caps accented topic like "DÉCISIONS" that - // mismatch meant this SELECT could fail to find the row - // the `INSERT OR IGNORE` conflict was already about, - // even though `summary_hash` alone uniquely identifies - // it. `summary_hash` is sufficient on its own. - "SELECT id, importance, keywords, raw_excerpt FROM memories - WHERE summary_hash = ?1", - params![hash], - |row| Ok((row.get(0)?, row.get(1)?, row.get(2)?, row.get(3)?)), - ) - .map_err(db_err)?; - - let existing_importance: Importance = existing_importance_str - .parse() - .unwrap_or(Importance::Medium); - let merged_importance = max_importance(existing_importance, memory.importance); - - let existing_keywords: Vec = - serde_json::from_str(&existing_keywords_json).unwrap_or_default(); - let mut merged_keywords = existing_keywords.clone(); - for kw in &memory.keywords { - if !merged_keywords.contains(kw) { - merged_keywords.push(kw.clone()); - } - } - - let merged_raw = memory.raw_excerpt.clone().or(existing_raw.clone()); - - let importance_changed = merged_importance != existing_importance; - let keywords_changed = merged_keywords != existing_keywords; - let raw_changed = merged_raw != existing_raw; - if importance_changed || keywords_changed || raw_changed { - let merged_keywords_json = serde_json::to_string(&merged_keywords)?; - self.conn - .execute( - "UPDATE memories - SET importance = ?1, keywords = ?2, raw_excerpt = ?3, updated_at = ?4 - WHERE id = ?5", - params![ - merged_importance.to_string(), - merged_keywords_json, - merged_raw, - Utc::now().to_rfc3339(), - existing_id, - ], - ) - .map_err(db_err)?; - self.cache_invalidate(&existing_id); - } - - tracing::debug!( - topic = %memory.topic, - existing = %existing_id, - attempted = %memory.id, - imp_changed = importance_changed, - kw_changed = keywords_changed, - raw_changed = raw_changed, - "store: dedup'd duplicate memory (metadata merged)" - ); - return Ok(existing_id); - } - - // Sync to vec_memories for KNN search (only on a fresh insert). - if let Some(ref blob) = emb_blob { - self.conn - .execute( - "INSERT INTO vec_memories (memory_id, embedding) VALUES (?1, ?2)", - params![memory.id, blob], - ) - .map_err(db_err)?; - } - - Ok(memory.id.clone()) - } -} - -impl MemoryStore for SqliteStore { - fn store(&self, memory: Memory) -> IcmResult { - let memory = validate_and_normalize(memory)?; - - self.conn - .execute_batch("BEGIN IMMEDIATE;") - .map_err(db_err)?; - - match self.store_inner(&memory) { - Ok(id) => { - self.conn.execute_batch("COMMIT;").map_err(db_err)?; - Ok(id) - } - Err(e) => { - let _ = self.conn.execute_batch("ROLLBACK;"); - Err(e) - } - } - } - - fn get(&self, id: &str) -> IcmResult> { - if let Some(m) = self.cache_get(id) { - return Ok(Some(m)); - } - - let mut stmt = self - .conn - .prepare(&format!("SELECT {SELECT_COLS} FROM memories WHERE id = ?1")) - .map_err(db_err)?; - - let result = stmt - .query_row(params![id], row_to_memory) - .optional() - .map_err(db_err)?; - - if let Some(ref m) = result { - self.cache_put(m); - } - Ok(result) - } - - fn update(&self, memory: &Memory) -> IcmResult<()> { - // Same constraints as `store()` — without this, oversized or - // NUL-carrying payloads could bypass validation by storing small - // then updating big (audit finding). - validate_fields(&memory.topic, &memory.summary)?; - - let keywords_json = serde_json::to_string(&memory.keywords)?; - let related_json = serde_json::to_string(&memory.related_ids)?; - let st = source_type(&memory.source); - let sd = source_data(&memory.source); - let emb_blob = memory.embedding.as_deref().map(embedding_to_blob); - - // Recompute summary_hash on update — topic or summary may have - // changed, and the partial unique index on (topic, summary_hash) - // would otherwise reflect stale state. - let hash = summary_hash(&memory.topic, &memory.summary); - - // memories + vec_memories must move together: a failure between the - // row update and the vector sync would leave a memory invisible to - // (or stale in) vector search (audit finding — same pattern as - // `store()` / `consolidate_topic`). - self.conn - .execute_batch("BEGIN IMMEDIATE;") - .map_err(db_err)?; - - let result: IcmResult<()> = (|| { - let changed = self - .conn - .execute( - "UPDATE memories SET - updated_at = ?2, last_accessed = ?3, access_count = ?4, weight = ?5, - topic = ?6, summary = ?7, raw_excerpt = ?8, keywords = ?9, - importance = ?10, source_type = ?11, source_data = ?12, related_ids = ?13, - embedding = ?14, summary_hash = ?15 - WHERE id = ?1", - params![ - memory.id, - memory.updated_at.to_rfc3339(), - memory.last_accessed.to_rfc3339(), - memory.access_count, - memory.weight, - memory.topic, - memory.summary, - memory.raw_excerpt, - keywords_json, - memory.importance.to_string(), - st, - sd, - related_json, - emb_blob, - hash, - ], - ) - .map_err(db_err)?; - - if changed == 0 { - return Err(IcmError::NotFound(memory.id.clone())); - } - - // Sync vec_memories: always delete old, re-insert if embedding exists - self.conn - .execute( - "DELETE FROM vec_memories WHERE memory_id = ?1", - params![memory.id], - ) - .map_err(db_err)?; - if let Some(ref blob) = emb_blob { - self.conn - .execute( - "INSERT INTO vec_memories (memory_id, embedding) VALUES (?1, ?2)", - params![memory.id, blob], - ) - .map_err(db_err)?; - } - Ok(()) - })(); - - match result { - Ok(()) => { - self.conn.execute_batch("COMMIT;").map_err(db_err)?; - self.cache_invalidate(&memory.id); - Ok(()) - } - Err(e) => { - let _ = self.conn.execute_batch("ROLLBACK;"); - Err(e) - } - } - } - - fn delete(&self, id: &str) -> IcmResult<()> { - // Both deletes in one transaction so a failure can't strand an - // orphaned vector or a memory whose vector is gone (audit finding). - self.conn - .execute_batch("BEGIN IMMEDIATE;") - .map_err(db_err)?; - - let result: IcmResult<()> = (|| { - self.conn - .execute("DELETE FROM vec_memories WHERE memory_id = ?1", params![id]) - .map_err(db_err)?; - - let changed = self - .conn - .execute("DELETE FROM memories WHERE id = ?1", params![id]) - .map_err(db_err)?; - - if changed == 0 { - return Err(IcmError::NotFound(id.to_string())); - } - - // Manual-testing finding: deleting a memory left it as a - // dangling entry in every other memory's `related_ids` - // (auto-link back-references) forever — `expand_with_neighbors` - // tolerates the miss silently, but each stale id still spends a - // slot out of the caller's `max_neighbors` budget instead of - // surfacing a real, live neighbor, and any external consumer of - // the JSON export sees a reference to nothing. Strip the - // deleted id from every `related_ids` array that mentions it. - // The `LIKE` clause is a cheap prefilter (only rows that could - // possibly match do the JSON rewrite); it matches the - // JSON-quoted form specifically so a ULID that happens to be a - // literal substring of another can't cause a false hit. - self.conn - .execute( - "UPDATE memories - SET related_ids = ( - SELECT COALESCE(json_group_array(value), '[]') - FROM json_each(memories.related_ids) - WHERE value != ?1 - ) - WHERE related_ids LIKE '%\"' || ?1 || '\"%'", - params![id], - ) - .map_err(db_err)?; - - Ok(()) - })(); - - match result { - Ok(()) => { - self.conn.execute_batch("COMMIT;").map_err(db_err)?; - // The related_ids cleanup above can touch an arbitrary - // number of other rows, not just `id` — clear the whole - // cache rather than tracking which ones, so a cached - // neighbor's `related_ids` can't keep serving the - // just-deleted id after this returns. - self.cache_clear(); - Ok(()) - } - Err(e) => { - let _ = self.conn.execute_batch("ROLLBACK;"); - Err(e) - } - } - } - - fn search_by_keywords(&self, keywords: &[&str], limit: usize) -> IcmResult> { - if keywords.is_empty() { - return Ok(Vec::new()); - } - - // Cap keywords to avoid massive SQL generation - let keywords = &keywords[..keywords.len().min(50)]; - let limit = limit.min(100); - - // Audit finding: a keyword containing `%` or `_` was interpolated - // straight into the LIKE pattern unescaped. `%` matches every row - // (`"100%"` as a keyword becomes the pattern `%100%%%`, which - // degrades to "contains 100" at best and can blow up matching); - // `_` matches any single character (`"snake_case"` also matches - // "snakeXcase"). Both are plausible keywords coming from an LLM via - // MCP. Escape them and declare the escape character explicitly. - let where_parts: Vec = (0..keywords.len()) - .map(|i| { - let p = i + 1; - format!( - "(keywords LIKE ?{p} ESCAPE '\\' OR summary LIKE ?{p} ESCAPE '\\' \ - OR topic LIKE ?{p} ESCAPE '\\')" - ) - }) - .collect(); - let where_clause = where_parts.join(" OR "); - - let query = format!( - "SELECT {SELECT_COLS} FROM memories WHERE {where_clause} ORDER BY weight DESC LIMIT ?{}", - keywords.len() + 1 - ); - - let mut stmt = self.conn.prepare(&query).map_err(db_err)?; - - let mut param_values: Vec> = keywords - .iter() - .map(|k| { - Box::new(format!("%{}%", escape_like_wildcards(k))) - as Box - }) - .collect(); - param_values.push(Box::new(limit as i64)); - - let params_ref: Vec<&dyn rusqlite::types::ToSql> = - param_values.iter().map(|p| p.as_ref()).collect(); - - let rows = stmt - .query_map(params_ref.as_slice(), row_to_memory) - .map_err(db_err)?; - - collect_rows(rows) - } - - fn search_fts(&self, query: &str, limit: usize) -> IcmResult> { - let limit = limit.min(100); - let sanitized = sanitize_fts_query(query); - if sanitized.is_empty() { - return Ok(Vec::new()); - } - - let sql = format!( - "SELECT {SELECT_COLS} FROM memories - WHERE id IN ( - SELECT id FROM memories_fts WHERE memories_fts MATCH ?1 - ) - ORDER BY weight DESC - LIMIT ?2" - ); - - let mut stmt = self.conn.prepare(&sql).map_err(db_err)?; - - let rows = stmt - .query_map(params![sanitized, limit as i64], row_to_memory) - .map_err(db_err)?; - - collect_rows(rows) - } - - fn search_by_embedding( - &self, - embedding: &[f32], - limit: usize, - ) -> IcmResult> { - let query_blob = embedding_to_blob(embedding); - - // KNN query on vec0 virtual table (requires LIMIT in the query itself) - let mut knn_stmt = self - .conn - .prepare( - "SELECT memory_id, distance - FROM vec_memories - WHERE embedding MATCH ?1 - ORDER BY distance - LIMIT ?2", - ) - .map_err(db_err)?; - - let knn_rows: Vec<(String, f32)> = knn_stmt - .query_map(params![query_blob, limit as i64], |row| { - Ok((row.get::<_, String>(0)?, row.get::<_, f32>(1)?)) - }) - .map_err(db_err)? - .filter_map(|r| r.ok()) - .collect(); - - if knn_rows.is_empty() { - return Ok(Vec::new()); - } - - // Batch fetch all memories in one query - let placeholders: Vec = (1..=knn_rows.len()).map(|i| format!("?{i}")).collect(); - let sql = format!( - "SELECT {SELECT_COLS} FROM memories WHERE id IN ({})", - placeholders.join(", ") - ); - let mut stmt = self.conn.prepare(&sql).map_err(db_err)?; - - let ids: Vec<&str> = knn_rows.iter().map(|(id, _)| id.as_str()).collect(); - let params: Vec<&dyn rusqlite::types::ToSql> = ids - .iter() - .map(|id| id as &dyn rusqlite::types::ToSql) - .collect(); - - let rows = stmt.query_map(&*params, row_to_memory).map_err(db_err)?; - - let mut memory_map: std::collections::HashMap = HashMap::new(); - for row in rows.flatten() { - memory_map.insert(row.id.clone(), row); - } - - // Reassemble in KNN order with similarity scores - let results: Vec<(Memory, f32)> = knn_rows - .into_iter() - .filter_map(|(id, distance)| memory_map.remove(&id).map(|mem| (mem, 1.0 - distance))) - .collect(); - - Ok(results) - } - - fn search_hybrid( - &self, - query: &str, - embedding: &[f32], - limit: usize, - ) -> IcmResult> { - let limit = limit.min(1000); - let pool_size = limit * 4; - let sanitized = sanitize_fts_query(query); - - // 1. Get FTS results with rank scores - let fts_sql = - "SELECT m.id, m.created_at, m.updated_at, m.last_accessed, m.access_count, m.weight, \ - m.topic, m.summary, m.raw_excerpt, m.keywords, \ - m.importance, m.source_type, m.source_data, m.related_ids, m.embedding, \ - fts.rank \ - FROM memories_fts fts \ - JOIN memories m ON m.id = fts.id \ - WHERE memories_fts MATCH ?1 \ - ORDER BY fts.rank \ - LIMIT ?2"; - - let mut fts_scores: HashMap = HashMap::with_capacity(pool_size); - let mut all_memories: HashMap = HashMap::with_capacity(pool_size); - - if !sanitized.is_empty() { - if let Ok(mut stmt) = self.conn.prepare(fts_sql) { - if let Ok(rows) = stmt.query_map(params![sanitized, pool_size as i64], |row| { - let memory = row_to_memory(row)?; - let rank: f32 = row.get(15)?; - Ok((memory, rank)) - }) { - for row in rows.flatten() { - let (memory, rank) = row; - // FTS5 bm25 rank is <= 0, MORE negative = MORE - // relevant. `1.0 / (1.0 + |rank|)` inverted this: it - // DECREASES as relevance increases (audit finding, - // proven wrong e.g. rank=-4.83 (strong match) scored - // 0.17 while rank=-0.2 (weak match) scored 0.83). - // `|rank| / (1.0 + |rank|)` keeps the same bounded - // [0,1) shape but is correctly monotonically - // INCREASING in relevance. - let score = rank.abs() / (1.0 + rank.abs()); - fts_scores.insert(memory.id.clone(), score); - all_memories.insert(memory.id.clone(), memory); - } - } - } - } // sanitized.is_empty() - - // 2. Get vector results - let vec_results = self.search_by_embedding(embedding, pool_size)?; - let mut vec_scores: HashMap = HashMap::with_capacity(pool_size); - for (memory, similarity) in vec_results { - vec_scores.insert(memory.id.clone(), similarity); - all_memories.entry(memory.id.clone()).or_insert(memory); - } - - // 3. Combine scores: 30% FTS + 70% vector - let keys: Vec = all_memories.keys().cloned().collect(); - let mut scored: Vec<(String, f32)> = Vec::with_capacity(keys.len()); - for id in keys { - let fts_score = fts_scores.get(&id).copied().unwrap_or(0.0); - let vec_score = vec_scores.get(&id).copied().unwrap_or(0.0); - let combined = 0.3 * fts_score + 0.7 * vec_score; - scored.push((id, combined)); - } - - // Sort by combined score descending - scored.sort_by(|a, b| b.1.partial_cmp(&a.1).unwrap_or(std::cmp::Ordering::Equal)); - scored.truncate(limit); - - let results: Vec<(Memory, f32)> = scored - .into_iter() - .filter_map(|(id, score)| all_memories.remove(&id).map(|mem| (mem, score))) - .collect(); - - Ok(results) - } - - fn update_access(&self, id: &str) -> IcmResult<()> { - // Read-only short-circuit (issue #263): callers of recall expect - // this to be best-effort bookkeeping, not a hard precondition. - // Skipping silently lets `icm recall` work against a DB the - // process cannot write to. - if self.readonly { - return Ok(()); - } - let now = Utc::now().to_rfc3339(); - let changed = self - .conn - .execute( - "UPDATE memories SET last_accessed = ?1, access_count = access_count + 1 WHERE id = ?2", - params![now, id], - ) - .map_err(db_err)?; - - if changed == 0 { - return Err(IcmError::NotFound(id.to_string())); - } - self.cache_invalidate(id); - Ok(()) - } - - fn batch_update_access(&self, ids: &[&str]) -> IcmResult { - if ids.is_empty() { - return Ok(0); - } - if self.readonly { - // Same rationale as `update_access` (issue #263). - return Ok(0); - } - let now = Utc::now().to_rfc3339(); - let placeholders: Vec = (2..=ids.len() + 1).map(|i| format!("?{i}")).collect(); - let sql = format!( - "UPDATE memories SET last_accessed = ?1, access_count = access_count + 1 WHERE id IN ({})", - placeholders.join(", ") - ); - let mut params_vec: Vec> = - Vec::with_capacity(ids.len() + 1); - params_vec.push(Box::new(now)); - for id in ids { - params_vec.push(Box::new(id.to_string())); - } - let refs: Vec<&dyn rusqlite::types::ToSql> = - params_vec.iter().map(|p| p.as_ref()).collect(); - let changed = self.conn.execute(&sql, refs.as_slice()).map_err(db_err)?; - self.cache_invalidate_many(ids); - Ok(changed) - } - - fn apply_decay(&self, decay_factor: f32) -> IcmResult { - if self.readonly { - return Err(IcmError::ReadOnly("apply_decay".into())); - } - // Access-aware decay: frequently accessed memories decay slower. - // decay = base_rate * importance_multiplier / (1 + min(access_count, 5) * 0.1) - // - // Audit #185 H7: the access-count term used to be uncapped - // (`1 + access_count * 0.1`). A memory with `access_count=100` - // got a 11x slowdown on its decay, which made it effectively - // immune to pruning even at low importance. Anyone (or any - // bench loop, or any benign hook-driven recall pattern) that - // touched a memory many times pinned it near the top of the - // ranking forever — the same gaming class as the M01 issue - // the maintainer flagged earlier. - // - // Cap at 5 accesses → max 1.5x slowdown (33%). That preserves - // the original intent ("useful memories decay a bit slower") - // without giving any single memory infinite decay immunity. - // Critical-importance memories still skip decay entirely. - // - // Importance multipliers: - // critical: never decays (filtered by WHERE clause) - // high: 0.5x decay (half speed) - // medium: 1.0x decay (normal) - // low: 2.0x decay (double speed) - // Audit finding: for `low` importance (2x multiplier) with a - // low-access memory, the multiplier `1.0 - (1.0-f)*mult/denom` goes - // NEGATIVE once `f < 0.5` — `icm decay --factor 0.4` (accepted by - // the CLI's own `[0.0, 1.0)` validation) drove low-importance - // weights negative, putting them last in every `ORDER BY weight - // DESC` and prunable on the next pass. `MAX(0.0, ...)` clamps the - // multiplier at the SQL layer so weight can never go negative - // regardless of the caller (CLI, MCP, or any future direct caller - // that bypasses the CLI's own boundary check). - let changed = self - .conn - .execute( - "UPDATE memories SET weight = weight * MAX(0.0, - 1.0 - (1.0 - ?1) * - CASE importance - WHEN 'high' THEN 0.5 - WHEN 'low' THEN 2.0 - ELSE 1.0 - END - / (1.0 + MIN(access_count, 5) * 0.1) - ) - WHERE importance != 'critical'", - params![decay_factor], - ) - .map_err(db_err)?; - - // Decay touches every non-critical row's weight; can't selectively - // invalidate without re-reading rows, so just nuke the cache. - self.cache_clear(); - Ok(changed) - } - - fn prune(&self, weight_threshold: f32) -> IcmResult { - // Never prune critical or high importance memories. Both deletes in - // one transaction, and the vec_memories error is propagated instead - // of swallowed — a partial prune would leave orphaned vectors that - // keep matching KNN search for rows that no longer exist (audit - // finding). - self.conn - .execute_batch("BEGIN IMMEDIATE;") - .map_err(db_err)?; - - let result: IcmResult = (|| { - self.conn.execute( - "DELETE FROM vec_memories WHERE memory_id IN ( - SELECT id FROM memories WHERE weight < ?1 AND importance NOT IN ('critical', 'high') - )", - params![weight_threshold], - ) - .map_err(db_err)?; - - self.conn - .execute( - "DELETE FROM memories WHERE weight < ?1 AND importance NOT IN ('critical', 'high')", - params![weight_threshold], - ) - .map_err(db_err) - })(); - - match result { - Ok(changed) => { - self.conn.execute_batch("COMMIT;").map_err(db_err)?; - if changed > 0 { - self.cache_clear(); - } - Ok(changed) - } - Err(e) => { - let _ = self.conn.execute_batch("ROLLBACK;"); - Err(e) - } - } - } - - fn get_by_topic(&self, topic: &str) -> IcmResult> { - let mut stmt = self - .conn - .prepare(&format!( - "SELECT {SELECT_COLS} FROM memories WHERE topic = ?1 ORDER BY weight DESC LIMIT 500" - )) - .map_err(db_err)?; - - let rows = stmt - .query_map(params![topic], row_to_memory) - .map_err(db_err)?; - - collect_rows(rows) - } - - fn list_all(&self) -> IcmResult> { - let mut stmt = self - .conn - .prepare(&format!( - "SELECT {SELECT_COLS} FROM memories ORDER BY weight DESC LIMIT 10000" - )) - .map_err(db_err)?; - - let rows = stmt.query_map([], row_to_memory).map_err(db_err)?; - collect_rows(rows) - } - - fn list_topics(&self) -> IcmResult> { - let mut stmt = self - .conn - .prepare("SELECT topic, COUNT(*) FROM memories GROUP BY topic ORDER BY topic") - .map_err(db_err)?; - - let rows = stmt - .query_map([], |row| { - Ok((row.get::<_, String>(0)?, row.get::<_, usize>(1)?)) - }) - .map_err(db_err)?; - - collect_rows(rows) - } - - fn consolidate_topic(&self, topic: &str, consolidated: Memory) -> IcmResult<()> { - // The consolidated memory goes through the same validation as any - // other write — MCP `icm_memory_consolidate` passes a caller-provided - // summary that previously bypassed every size/content check. - let consolidated = validate_and_normalize(consolidated)?; - - self.conn - .execute_batch("BEGIN IMMEDIATE;") - .map_err(db_err)?; - - // Manual-testing finding: captured before the delete below, since - // afterward the rows (and thus this query) are gone. Used to clean - // up any *other* memory's related_ids that pointed at these — - // same dangling-reference bug already fixed for the single-id - // `delete`, reachable here too since this is a second, separate - // bulk-delete code path. - let deleted_ids: Vec = { - let mut stmt = self - .conn - .prepare("SELECT id FROM memories WHERE topic = ?1 AND importance != 'critical'") - .map_err(db_err)?; - let rows = stmt - .query_map(params![topic], |row| row.get::<_, String>(0)) - .map_err(db_err)?; - rows.collect::>>().map_err(db_err)? - }; - - // `critical` memories are never deleted — same contract as - // `apply_decay` and `prune`. Consolidation replaces the expendable - // tail of a topic, not its "never forget" entries (audit finding: - // this DELETE previously wiped critical memories too). - // Clean vec_memories for entries about to be deleted - if let Err(e) = self.conn.execute( - "DELETE FROM vec_memories WHERE memory_id IN ( - SELECT id FROM memories WHERE topic = ?1 AND importance != 'critical' - )", - params![topic], - ) { - tracing::warn!(topic, error = %e, "consolidate_topic: rolling back after vec_memories delete failed"); - let _ = self.conn.execute_batch("ROLLBACK;"); - return Err(IcmError::Database(e.to_string())); - } - - if let Err(e) = self.conn.execute( - "DELETE FROM memories WHERE topic = ?1 AND importance != 'critical'", - params![topic], - ) { - tracing::warn!(topic, error = %e, "consolidate_topic: rolling back after memories delete failed"); - let _ = self.conn.execute_batch("ROLLBACK;"); - return Err(IcmError::Database(e.to_string())); - } - - if !deleted_ids.is_empty() { - let ids_json = serde_json::to_string(&deleted_ids).map_err(IcmError::from)?; - if let Err(e) = self.conn.execute( - "UPDATE memories - SET related_ids = ( - SELECT COALESCE(json_group_array(value), '[]') - FROM json_each(memories.related_ids) - WHERE value NOT IN (SELECT value FROM json_each(?1)) - ) - WHERE EXISTS ( - SELECT 1 FROM json_each(memories.related_ids) - WHERE value IN (SELECT value FROM json_each(?1)) - )", - params![ids_json], - ) { - tracing::warn!(topic, error = %e, "consolidate_topic: rolling back after related_ids cleanup failed"); - let _ = self.conn.execute_batch("ROLLBACK;"); - return Err(IcmError::Database(e.to_string())); - } - } - - if let Err(e) = self.store_inner(&consolidated) { - tracing::warn!(topic, error = %e, "consolidate_topic: rolling back after store failed"); - let _ = self.conn.execute_batch("ROLLBACK;"); - return Err(e); - } - - // Rebuild FTS index to eliminate any ghost entries from the external - // content table. This guarantees search results stay consistent after - // bulk deletes (fixes #44). - if let Err(e) = self - .conn - .execute_batch("INSERT INTO memories_fts(memories_fts) VALUES('rebuild');") - { - tracing::warn!(topic, error = %e, "consolidate_topic: rolling back after FTS rebuild failed"); - let _ = self.conn.execute_batch("ROLLBACK;"); - return Err(IcmError::Database(e.to_string())); - } - - self.conn.execute_batch("COMMIT;").map_err(db_err)?; - // Bulk delete + re-insert touches arbitrarily many cached entries. - self.cache_clear(); - Ok(()) - } - - fn count(&self) -> IcmResult { - self.conn - .query_row("SELECT COUNT(*) FROM memories", [], |row| { - row.get::<_, usize>(0) - }) - .map_err(|e| IcmError::Database(e.to_string())) - } - - fn count_by_topic(&self, topic: &str) -> IcmResult { - self.conn - .query_row( - "SELECT COUNT(*) FROM memories WHERE topic = ?1", - params![topic], - |row| row.get::<_, usize>(0), - ) - .map_err(|e| IcmError::Database(e.to_string())) - } - - fn topic_health(&self, topic: &str) -> IcmResult { - let row = self - .conn - .query_row( - "SELECT - COUNT(*), - AVG(weight), - AVG(CAST(access_count AS REAL)), - MIN(created_at), - MAX(created_at), - MAX(last_accessed), - SUM(CASE WHEN weight < 0.5 - AND julianday('now') - julianday(last_accessed) > 14 - THEN 1 ELSE 0 END) - FROM memories WHERE topic = ?1", - params![topic], - |row| { - Ok(( - row.get::<_, usize>(0)?, - row.get::<_, f32>(1)?, - row.get::<_, f32>(2)?, - row.get::<_, Option>(3)?, - row.get::<_, Option>(4)?, - row.get::<_, Option>(5)?, - row.get::<_, usize>(6)?, - )) - }, - ) - .map_err(db_err)?; - - let ( - entry_count, - avg_weight, - avg_access, - oldest_str, - newest_str, - last_accessed_str, - stale_count, - ) = row; - - if entry_count == 0 { - return Err(IcmError::NotFound(format!("topic: {topic}"))); - } - - let parse_dt = |s: &str| -> Option> { - match DateTime::parse_from_rfc3339(s) { - Ok(d) => Some(d.with_timezone(&Utc)), - Err(e) => { - tracing::warn!("invalid timestamp '{}': {}", s, e); - None - } - } - }; - - Ok(TopicHealth { - topic: topic.to_string(), - entry_count, - avg_weight, - avg_access_count: avg_access, - oldest: oldest_str.as_deref().and_then(parse_dt), - newest: newest_str.as_deref().and_then(parse_dt), - last_accessed: last_accessed_str.as_deref().and_then(parse_dt), - needs_consolidation: entry_count > 5, - stale_count, - }) - } - - fn stats(&self) -> IcmResult { - let (total_memories, total_topics, avg_weight, oldest_str, newest_str): ( - usize, - usize, - f32, - Option, - Option, - ) = self - .conn - .query_row( - "SELECT COUNT(*), COUNT(DISTINCT topic), COALESCE(AVG(weight), 0.0), \ - MIN(created_at), MAX(created_at) FROM memories", - [], - |row| { - Ok(( - row.get(0)?, - row.get(1)?, - row.get(2)?, - row.get(3)?, - row.get(4)?, - )) - }, - ) - .map_err(db_err)?; - - let oldest_memory = oldest_str - .and_then(|s| DateTime::parse_from_rfc3339(&s).ok()) - .map(|d| d.with_timezone(&Utc)); - let newest_memory = newest_str - .and_then(|s| DateTime::parse_from_rfc3339(&s).ok()) - .map(|d| d.with_timezone(&Utc)); - - Ok(StoreStats { - total_memories, - total_topics, - avg_weight, - oldest_memory, - newest_memory, - }) - } -} - -// --------------------------------------------------------------------------- -// Memoir / Concept helpers -// --------------------------------------------------------------------------- - -fn parse_dt(s: &str) -> DateTime { - DateTime::parse_from_rfc3339(s) - .map(|d| d.with_timezone(&Utc)) - .unwrap_or_else(|_| Utc::now()) -} - -fn row_to_memoir(row: &rusqlite::Row) -> rusqlite::Result { - Ok(Memoir { - id: row.get(0)?, - name: row.get(1)?, - description: row.get(2)?, - created_at: parse_dt(&row.get::<_, String>(3)?), - updated_at: parse_dt(&row.get::<_, String>(4)?), - consolidation_threshold: row.get::<_, u32>(5)?, - }) -} - -const MEMOIR_COLS: &str = "id, name, description, created_at, updated_at, consolidation_threshold"; - -fn row_to_concept(row: &rusqlite::Row) -> rusqlite::Result { - let labels_json: String = row.get(4)?; - let labels: Vec