innate_core/storage/
mod.rs

1//! SQLite storage layer.
2//!
3//! Replaces sqlite-vec virtual tables with ordinary BLOB columns + pure-Rust
4//! cosine similarity, keeping the schema otherwise aligned with v4.5.x.
5
6use std::cell::{Cell, RefCell};
7use std::collections::HashMap;
8use std::path::{Path, PathBuf};
9
10use rusqlite::{params, Connection, Row};
11use serde_json::Value;
12
13use crate::errors::{InnateError, Result};
14use crate::utils::{dot_product, l2_normalize, unpack_embedding};
15
16mod chunks;
17mod evolution;
18mod meta;
19pub mod metrics;
20mod raw;
21mod traces;
22
23const EXPECTED_SCHEMA_VERSION: &str = "4.21";
24
25// Embedded SQL schema — no external files needed.
26const SCHEMA_SQL: &str = include_str!("../schema.sql");
27
28type VectorEntries = Vec<(String, Vec<f32>)>;
29type VectorCache = RefCell<Option<VectorEntries>>;
30
31/// A single dependency edge: `(dst, kind, dst_lib)`.
32pub type DepEdge = (String, String, Option<String>);
33
34pub struct Storage {
35    pub db_path: PathBuf,
36    conn: Connection,
37    pub content_dim: usize,
38    pub trigger_dim: usize,
39    /// Pre-parsed in-memory caches for vector search; None = cold (not loaded or invalidated).
40    vec_content_cache: VectorCache,
41    vec_trigger_cache: VectorCache,
42    /// Last observed vector revision. Only vector writes advance this value.
43    vector_cache_revision: Cell<Option<i64>>,
44}
45
46#[derive(Debug, Clone, PartialEq, Eq)]
47pub struct EvolveRequestClaim {
48    pub id: String,
49    pub reason: String,
50}
51
52impl Storage {
53    pub fn open(db_path: impl AsRef<Path>, content_dim: usize, trigger_dim: usize) -> Result<Self> {
54        let db_path = db_path.as_ref().to_path_buf();
55        if let Some(parent) = db_path.parent() {
56            std::fs::create_dir_all(parent)?;
57        }
58        let conn = Connection::open(&db_path)?;
59        configure_pragmas(&conn)?;
60        let mut s = Self {
61            db_path,
62            conn,
63            content_dim,
64            trigger_dim,
65            vec_content_cache: RefCell::new(None),
66            vec_trigger_cache: RefCell::new(None),
67            vector_cache_revision: Cell::new(None),
68        };
69        s.init_schema()?;
70        Ok(s)
71    }
72
73    pub fn open_readonly(db_path: impl AsRef<Path>) -> Result<Self> {
74        let db_path = db_path.as_ref().to_path_buf();
75        let conn = Connection::open_with_flags(
76            &db_path,
77            rusqlite::OpenFlags::SQLITE_OPEN_READ_ONLY | rusqlite::OpenFlags::SQLITE_OPEN_NO_MUTEX,
78        )?;
79        conn.pragma_update(None, "query_only", "ON")?;
80        conn.pragma_update(None, "foreign_keys", "ON")?;
81        let s = Self {
82            db_path,
83            conn,
84            content_dim: 1024,
85            trigger_dim: 256,
86            vec_content_cache: RefCell::new(None),
87            vec_trigger_cache: RefCell::new(None),
88            vector_cache_revision: Cell::new(None),
89        };
90        Ok(s)
91    }
92
93    fn init_schema(&mut self) -> Result<()> {
94        let has_meta: bool = self.conn.query_row(
95            "SELECT count(*) FROM sqlite_master WHERE type='table' AND name='meta'",
96            [],
97            |r| r.get::<_, i64>(0),
98        )? > 0;
99
100        if !has_meta {
101            // Wrap schema creation in a transaction for atomicity.
102            self.conn.execute_batch("BEGIN IMMEDIATE")?;
103            let r = self.conn.execute_batch(SCHEMA_SQL);
104            if r.is_ok() {
105                self.conn.execute_batch("COMMIT")?;
106            } else {
107                let _ = self.conn.execute_batch("ROLLBACK");
108                r?;
109            }
110            return Ok(());
111        }
112
113        let current: Option<String> = self
114            .conn
115            .query_row(
116                "SELECT value FROM meta WHERE key='schema_version'",
117                [],
118                |r| r.get(0),
119            )
120            .optional()?;
121
122        let current = current
123            .ok_or_else(|| InnateError::Other("meta table missing schema_version".into()))?;
124
125        let cur = ver_tuple(&current);
126        let exp = ver_tuple(EXPECTED_SCHEMA_VERSION);
127
128        match cur.cmp(&exp) {
129            std::cmp::Ordering::Equal => Ok(()),
130            std::cmp::Ordering::Greater => {
131                // Forward-compat: newer schema, warn but allow.
132                eprintln!(
133                    "[innate] warning: db schema {current} > expected {EXPECTED_SCHEMA_VERSION}"
134                );
135                Ok(())
136            }
137            std::cmp::Ordering::Less => {
138                // Delegate to the proper migration chain which handles all steps atomically.
139                let applied = crate::migrate::run_migrations(&self.db_path)?;
140                if !applied.is_empty() {
141                    eprintln!("[innate] auto-migrated: {}", applied.join(", "));
142                }
143                Ok(())
144            }
145        }
146    }
147
148    // ------------------------------------------------------------------
149    // Transactions
150    // ------------------------------------------------------------------
151
152    pub fn begin_immediate(&self) -> Result<()> {
153        self.conn.execute_batch("BEGIN IMMEDIATE")?;
154        Ok(())
155    }
156
157    pub fn commit(&self) -> Result<()> {
158        self.conn.execute_batch("COMMIT")?;
159        Ok(())
160    }
161
162    pub fn rollback(&self) -> Result<()> {
163        self.conn.execute_batch("ROLLBACK")?;
164        // In-place cache upserts from the aborted transaction may not have
165        // persisted; drop caches so the next search reloads committed state.
166        self.invalidate_vector_caches();
167        Ok(())
168    }
169
170    // ------------------------------------------------------------------
171}
172
173// ------------------------------------------------------------------
174// Row types
175// ------------------------------------------------------------------
176
177#[derive(Debug, Default, Clone)]
178pub struct ChunkRow {
179    pub id: String,
180    pub skill_name: Option<String>,
181    pub seq: i64,
182    pub content: String,
183    pub trigger_desc: Option<String>,
184    pub anti_trigger_desc: Option<String>,
185    pub content_hash: String,
186    pub token_count: Option<i64>,
187    pub origin: String,
188    pub source: Option<String>,
189    pub agent: Option<String>,
190    pub maturity: Option<String>,
191    pub related_ids: Option<String>,
192    pub protected: i64,
193    pub state: String,
194    pub state_reason: Option<String>,
195    pub state_updated_at: Option<String>,
196    pub confidence: f64,
197    pub confidence_reason: Option<String>,
198    pub version: i64,
199    pub distilled_from: Option<String>,
200    pub distill_provider: Option<String>,
201    pub distill_model: Option<String>,
202    pub distill_prompt_version: Option<String>,
203    pub parent_id: Option<String>,
204    pub selected_count: i64,
205    pub used_count: i64,
206    pub used_success_count: i64,
207    pub success_trace_ids_count: i64,
208    pub last_success_at: Option<String>,
209    pub last_agg_ts: Option<String>,
210    pub embed_version: i64,
211    pub created_at: String,
212    pub updated_at: String,
213    pub last_used_at: Option<String>,
214}
215
216#[derive(Debug, Default)]
217pub struct EpisodicLogRow {
218    pub id: String,
219    pub trace_id: String,
220    pub lib_id: String,
221    pub ts: String,
222    pub query: Option<String>,
223    pub recall_snapshot: Option<String>,
224    pub output: Option<String>,
225    pub output_summary: Option<String>,
226    pub outcome: Option<String>,
227    pub event_source: String,
228    pub agent: Option<String>,
229    pub task_state: String,
230    pub completed_at: Option<String>,
231    pub usage_state: String,
232    pub used_ids: Option<String>,
233    pub used_attribution: Option<String>,
234    pub used_complete: bool,
235    pub context_key: Option<String>,
236    pub nomination: Option<String>,
237    pub priority: i64,
238    pub distill_state: String,
239    pub distill_note: Option<String>,
240}
241
242// ------------------------------------------------------------------
243// Helpers
244// ------------------------------------------------------------------
245
246fn configure_pragmas(conn: &Connection) -> Result<()> {
247    conn.execute_batch(
248        "PRAGMA journal_mode=WAL;
249         PRAGMA foreign_keys=ON;
250         PRAGMA synchronous=NORMAL;
251         PRAGMA cache_size=-65536;
252         PRAGMA mmap_size=268435456;
253         PRAGMA busy_timeout=5000;
254         PRAGMA temp_store=memory;",
255    )?;
256    // Validate WAL mode was accepted (some VFS/filesystems silently downgrade).
257    let mode: String = conn.query_row("PRAGMA journal_mode", [], |r| r.get(0))?;
258    if mode != "wal" {
259        return Err(crate::errors::InnateError::Other(format!(
260            "WAL mode required but got '{mode}'; check filesystem support"
261        )));
262    }
263    Ok(())
264}
265
266fn ver_tuple(v: &str) -> (u32, u32, u32) {
267    let parts: Vec<u32> = v.split('.').filter_map(|s| s.parse().ok()).collect();
268    (
269        parts.first().copied().unwrap_or(0),
270        parts.get(1).copied().unwrap_or(0),
271        parts.get(2).copied().unwrap_or(0),
272    )
273}
274
275/// Convert a rusqlite Row to serde_json::Value using column names from statement.
276fn row_to_json_with_names(row: &Row, names: &[String]) -> rusqlite::Result<Value> {
277    let mut map = serde_json::Map::new();
278    for (i, name) in names.iter().enumerate() {
279        let v = row_value_at(row, i);
280        map.insert(name.clone(), v);
281    }
282    Ok(Value::Object(map))
283}
284
285fn row_to_json(row: &Row) -> rusqlite::Result<Value> {
286    let count = row.as_ref().column_count();
287    let mut map = serde_json::Map::new();
288    for i in 0..count {
289        let name = row.as_ref().column_name(i)?.to_string();
290        let v = row_value_at(row, i);
291        map.insert(name, v);
292    }
293    Ok(Value::Object(map))
294}
295
296fn row_value_at(row: &Row, i: usize) -> Value {
297    // Try types in preference order
298    if let Ok(v) = row.get::<_, Option<String>>(i) {
299        return v.map(Value::String).unwrap_or(Value::Null);
300    }
301    if let Ok(v) = row.get::<_, Option<i64>>(i) {
302        return v.map(|n| Value::Number(n.into())).unwrap_or(Value::Null);
303    }
304    if let Ok(v) = row.get::<_, Option<f64>>(i) {
305        return v
306            .and_then(serde_json::Number::from_f64)
307            .map(Value::Number)
308            .unwrap_or(Value::Null);
309    }
310    Value::Null
311}
312
313trait OptionalExt<T> {
314    fn optional(self) -> rusqlite::Result<Option<T>>;
315}
316impl<T> OptionalExt<T> for rusqlite::Result<T> {
317    fn optional(self) -> rusqlite::Result<Option<T>> {
318        match self {
319            Ok(v) => Ok(Some(v)),
320            Err(rusqlite::Error::QueryReturnedNoRows) => Ok(None),
321            Err(e) => Err(e),
322        }
323    }
324}
innate_core/storage/mod.rs

innate_core/storage/
mod.rs
