scone_core/
lib.rs

1//! Scone core: a local-first episodic + semantic memory engine.
2//!
3//! This crate performs no stdout I/O and touches the network only through
4//! provider traits. SQLite is the single source of truth (spec §5).
5
6pub mod auth;
7pub mod chunker;
8mod db;
9pub mod distill;
10pub mod embed;
11mod error;
12pub mod index;
13mod ingest;
14pub mod llm;
15mod portability;
16pub mod profile;
17mod recall;
18
19use std::path::{Path, PathBuf};
20
21use rusqlite::Connection;
22
23pub use distill::{ApplyReport, DistillReport, ProvenanceItem};
24pub use error::{Result, SconeError};
25pub use ingest::{IngestInput, IngestOutcome, ScanReport};
26pub use portability::ImportReport;
27pub use profile::Profile;
28pub use recall::{ContextPack, FactItem, RecallItem, RecallOpts};
29
30#[derive(Debug)]
31pub struct DoctorReport {
32    pub episodes: usize,
33    pub chunks: usize,
34    pub reembedded: usize,
35}
36
37impl std::fmt::Debug for Engine {
38    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
39        f.debug_struct("Engine")
40            .field("data_dir", &self.data_dir)
41            .field("embedder", &self.embedder.id())
42            .finish_non_exhaustive()
43    }
44}
45
46#[derive(Debug)]
47pub struct SpaceStatus {
48    pub name: String,
49    pub episodes: i64,
50    pub chunks: i64,
51    pub revision: i64,
52}
53
54#[derive(Debug)]
55pub struct StatusReport {
56    pub read_only: bool,
57    pub spaces: Vec<SpaceStatus>,
58    pub embedder_id: String,
59    pub embedder_dim: usize,
60    pub index_dirty: bool,
61    pub pending_distill: i64,
62    pub failed_distill: i64,
63    pub llm_id: Option<String>,
64}
65
66pub struct Engine {
67    conn: Connection,
68    data_dir: PathBuf,
69    embedder: Box<dyn embed::EmbeddingProvider>,
70    llm: Option<Box<dyn llm::LlmProvider>>,
71    fts: index::fts::FtsIndex,
72    vectors: index::vectors::VectorIndex,
73    /// Staged index writes awaiting a flush (flush-on-recall / on drop).
74    indexes_dirty: bool,
75    /// Chunking granularity for future ingests (tuning knob; benchmarked
76    /// sweeps live in memory/benchmarks.md).
77    chunk_target: usize,
78}
79
80impl Drop for Engine {
81    fn drop(&mut self) {
82        // Best-effort: a failed flush here is recovered by the catch-up
83        // reindex on the next open (meta 'indexed_max' high-water mark).
84        let _ = self.flush_indexes();
85    }
86}
87
88impl Engine {
89    /// Open (creating if needed) the engine rooted at `data_dir`.
90    ///
91    /// Refuses to open when the store was embedded with a different
92    /// provider (spec §9): switching embedders is an explicit
93    /// `doctor --rebuild`, never a silent re-index.
94    pub fn open(data_dir: &Path, embedder: Box<dyn embed::EmbeddingProvider>) -> Result<Engine> {
95        Self::open_inner(data_dir, embedder, false)
96    }
97
98    /// Open bypassing the embedder pin, for `doctor --rebuild` only.
99    pub fn open_for_repair(
100        data_dir: &Path,
101        embedder: Box<dyn embed::EmbeddingProvider>,
102    ) -> Result<Engine> {
103        Self::open_inner(data_dir, embedder, true)
104    }
105
106    fn open_inner(
107        data_dir: &Path,
108        embedder: Box<dyn embed::EmbeddingProvider>,
109        repair: bool,
110    ) -> Result<Engine> {
111        std::fs::create_dir_all(data_dir)?;
112        let conn = db::open(&data_dir.join("scone.db"))?;
113        if !repair {
114            let pinned: Option<String> = match conn.query_row(
115                "SELECT value FROM meta WHERE key = 'embedder_id'",
116                [],
117                |r| r.get(0),
118            ) {
119                Ok(v) => Some(v),
120                Err(rusqlite::Error::QueryReturnedNoRows) => None,
121                Err(e) => return Err(SconeError::Db(e)),
122            };
123            if let Some(pinned) = pinned
124                && !(pinned == embedder.id() && Self::pinned_dim(&conn)? == Some(embedder.dim()))
125            {
126                return Err(SconeError::Index(format!(
127                    "store is pinned to embedder {pinned}; opening with {} ({} dims)                      requires `scone doctor --rebuild`",
128                    embedder.id(),
129                    embedder.dim()
130                )));
131            }
132        }
133        let vectors_dir = data_dir.join("vectors");
134        let vectors = if repair {
135            index::vectors::VectorIndex::open_or_reset(&vectors_dir, embedder.dim())?
136        } else {
137            index::vectors::VectorIndex::open(&vectors_dir, embedder.dim())?
138        };
139        let fts = index::fts::FtsIndex::open(&data_dir.join("fts"))?;
140        let mut engine = Engine {
141            conn,
142            data_dir: data_dir.to_path_buf(),
143            embedder,
144            llm: None,
145            fts,
146            vectors,
147            indexes_dirty: false,
148            chunk_target: ingest::CHUNK_TARGET_BYTES,
149        };
150        if engine.fts.writable() {
151            engine.catch_up_indexes()?;
152        }
153        Ok(engine)
154    }
155
156    /// Set the chunking granularity (bytes) for future ingests.
157    pub fn set_chunk_target(&mut self, bytes: usize) {
158        self.chunk_target = bytes.max(64);
159    }
160
161    /// True when another scone process holds the index write lock: search
162    /// works from committed state; ingest/distill/doctor are refused with
163    /// typed errors until the other process exits.
164    pub fn is_read_only(&self) -> bool {
165        !self.fts.writable()
166    }
167
168    fn require_writable(&self) -> Result<()> {
169        if self.is_read_only() {
170            return Err(SconeError::InvalidInput(
171                "this store is read-only: another scone process holds the write lock".into(),
172            ));
173        }
174        Ok(())
175    }
176
177    /// Reindex any chunks written after the last successful flush — the
178    /// crash-recovery half of lazy flushing. Idempotent (both indexes
179    /// upsert), and cheap: only the tail beyond the high-water mark.
180    fn catch_up_indexes(&mut self) -> Result<()> {
181        let indexed_max: i64 = self
182            .get_meta("indexed_max")?
183            .and_then(|v| v.parse().ok())
184            .unwrap_or(0);
185        let chunk_max: i64 =
186            self.conn
187                .query_row("SELECT coalesce(max(id), 0) FROM chunks", [], |r| r.get(0))?;
188        if chunk_max <= indexed_max {
189            return Ok(());
190        }
191        let rows: Vec<(i64, i64, String, Option<Vec<u8>>)> = {
192            let mut stmt = self.conn.prepare(
193                "SELECT c.id, e.space_id, e.content, c.start_byte, c.end_byte, c.embedding
194                 FROM chunks c JOIN episodes e ON e.id = c.episode_id
195                 WHERE c.id > ?1 ORDER BY c.id",
196            )?;
197            let mapped = stmt.query_map([indexed_max], |r| {
198                let content: String = r.get(2)?;
199                let start: i64 = r.get(3)?;
200                let end: i64 = r.get(4)?;
201                let text = content
202                    .get(start as usize..end as usize)
203                    .unwrap_or_default()
204                    .to_owned();
205                Ok((r.get(0)?, r.get(1)?, text, r.get(5)?))
206            })?;
207            mapped.collect::<std::result::Result<Vec<_>, _>>()?
208        };
209        let dim = self.embedder.dim();
210        let fts_rows: Vec<(u64, u64, &str)> = rows
211            .iter()
212            .map(|(id, space, text, _)| (*id as u64, *space as u64, text.as_str()))
213            .collect();
214        self.fts.add(&fts_rows)?;
215        let mut vec_rows: Vec<(u64, Vec<f32>)> = Vec::new();
216        for (id, _, _, blob) in &rows {
217            if let Some(b) = blob
218                && b.len() == dim * 4
219            {
220                let v = b
221                    .as_chunks::<4>()
222                    .0
223                    .iter()
224                    .map(|c| f32::from_le_bytes(*c))
225                    .collect();
226                vec_rows.push((*id as u64, v));
227            }
228        }
229        let borrowed: Vec<(u64, &[f32])> =
230            vec_rows.iter().map(|(id, v)| (*id, v.as_slice())).collect();
231        self.vectors.add(&borrowed)?;
232        self.indexes_dirty = true;
233        self.flush_indexes()
234    }
235
236    /// Flush staged index writes; advances the high-water mark only after
237    /// both indexes are durable.
238    pub(crate) fn flush_indexes(&mut self) -> Result<()> {
239        if !self.indexes_dirty || self.is_read_only() {
240            return Ok(());
241        }
242        self.fts.commit()?;
243        self.vectors.flush()?;
244        let chunk_max: i64 =
245            self.conn
246                .query_row("SELECT coalesce(max(id), 0) FROM chunks", [], |r| r.get(0))?;
247        self.set_meta("indexed_max", &chunk_max.to_string())?;
248        self.indexes_dirty = false;
249        Ok(())
250    }
251
252    /// Attach or detach the semantic lane's LLM. `None` pauses lane 2
253    /// loudly; the episodic engine is unaffected (spec §9).
254    pub fn set_llm(&mut self, llm: Option<Box<dyn llm::LlmProvider>>) {
255        self.llm = llm;
256    }
257
258    pub fn has_llm(&self) -> bool {
259        self.llm.is_some()
260    }
261
262    /// Answer a question from rendered context via the configured LLM.
263    pub fn llm_answer(&self, question: &str, context: &str) -> Result<String> {
264        match &self.llm {
265            Some(llm) => llm.answer(question, context),
266            None => Err(SconeError::Llm("no LLM configured".into())),
267        }
268    }
269
270    fn pinned_dim(conn: &Connection) -> Result<Option<usize>> {
271        match conn.query_row(
272            "SELECT value FROM meta WHERE key = 'embedder_dim'",
273            [],
274            |r| r.get::<_, String>(0),
275        ) {
276            Ok(v) => Ok(v.parse().ok()),
277            Err(rusqlite::Error::QueryReturnedNoRows) => Ok(None),
278            Err(e) => Err(SconeError::Db(e)),
279        }
280    }
281
282    /// Rebuild every derived index from SQLite truth, re-embedding chunks
283    /// whose stored vectors no longer match the active embedder.
284    pub fn doctor_rebuild(&mut self) -> Result<DoctorReport> {
285        self.require_writable()?;
286        self.fts.wipe()?;
287        self.vectors.wipe()?;
288        // Slice chunk text in Rust: our offsets are bytes, and SQLite's
289        // substr() counts characters — mixing them corrupts multibyte text.
290        let rows: Vec<(i64, i64, String, Option<Vec<u8>>)> = {
291            let mut stmt = self.conn.prepare(
292                "SELECT c.id, e.space_id, e.content, c.start_byte, c.end_byte, c.embedding
293                 FROM chunks c JOIN episodes e ON e.id = c.episode_id
294                 ORDER BY c.id",
295            )?;
296            let mapped = stmt.query_map([], |r| {
297                let content: String = r.get(2)?;
298                let start: i64 = r.get(3)?;
299                let end: i64 = r.get(4)?;
300                let text = content
301                    .get(start as usize..end as usize)
302                    .unwrap_or_default()
303                    .to_owned();
304                Ok((r.get(0)?, r.get(1)?, text, r.get(5)?))
305            })?;
306            mapped.collect::<std::result::Result<Vec<_>, _>>()?
307        };
308        let dim = self.embedder.dim();
309        let mut reembedded = 0usize;
310        let mut fts_rows = Vec::with_capacity(rows.len());
311        let mut vec_data = Vec::with_capacity(rows.len());
312        for (chunk_id, space_id, text, blob) in &rows {
313            let vector = match blob {
314                Some(b) if b.len() == dim * 4 => b
315                    .as_chunks::<4>()
316                    .0
317                    .iter()
318                    .map(|c| f32::from_le_bytes(*c))
319                    .collect::<Vec<f32>>(),
320                _ => {
321                    let embedded = self.embedder.embed(&[text.as_str()])?;
322                    let v = embedded
323                        .into_iter()
324                        .next()
325                        .ok_or_else(|| SconeError::Embed("embedder returned no vector".into()))?;
326                    let new_blob: Vec<u8> = v.iter().flat_map(|f| f.to_le_bytes()).collect();
327                    self.conn.execute(
328                        "UPDATE chunks SET embedding = ?1 WHERE id = ?2",
329                        rusqlite::params![new_blob, chunk_id],
330                    )?;
331                    reembedded += 1;
332                    v
333                }
334            };
335            fts_rows.push((*chunk_id as u64, *space_id as u64, text.as_str()));
336            vec_data.push((*chunk_id as u64, vector));
337        }
338        self.fts.add(&fts_rows)?;
339        let vec_rows: Vec<(u64, &[f32])> =
340            vec_data.iter().map(|(id, v)| (*id, v.as_slice())).collect();
341        self.vectors.add(&vec_rows)?;
342        self.indexes_dirty = true;
343        self.flush_indexes()?;
344        self.set_meta("embedder_id", self.embedder.id())?;
345        self.set_meta("embedder_dim", &dim.to_string())?;
346        self.set_meta("index_dirty", "0")?;
347        let episodes: i64 = self
348            .conn
349            .query_row("SELECT count(*) FROM episodes", [], |r| r.get(0))?;
350        Ok(DoctorReport {
351            episodes: episodes as usize,
352            chunks: rows.len(),
353            reembedded,
354        })
355    }
356
357    pub(crate) fn set_meta(&self, key: &str, value: &str) -> Result<()> {
358        self.conn.execute(
359            "INSERT INTO meta (key, value) VALUES (?1, ?2)
360             ON CONFLICT (key) DO UPDATE SET value = excluded.value",
361            [key, value],
362        )?;
363        Ok(())
364    }
365
366    pub(crate) fn get_meta(&self, key: &str) -> Result<Option<String>> {
367        match self
368            .conn
369            .query_row("SELECT value FROM meta WHERE key = ?1", [key], |r| r.get(0))
370        {
371            Ok(v) => Ok(Some(v)),
372            Err(rusqlite::Error::QueryReturnedNoRows) => Ok(None),
373            Err(e) => Err(SconeError::Db(e)),
374        }
375    }
376
377    pub fn space_revision(&self, space: &auth::ScopedSpace) -> Result<i64> {
378        Ok(self.conn.query_row(
379            "SELECT revision FROM spaces WHERE id = ?1",
380            [space.id()],
381            |r| r.get(0),
382        )?)
383    }
384
385    /// Administrative overview of the whole store (single-user surface;
386    /// multi-user servers must scope what they expose of it).
387    pub fn status(&self) -> Result<StatusReport> {
388        let mut stmt = self.conn.prepare(
389            "SELECT s.name, s.revision,
390                    (SELECT count(*) FROM episodes e WHERE e.space_id = s.id),
391                    (SELECT count(*) FROM chunks c JOIN episodes e ON e.id = c.episode_id
392                     WHERE e.space_id = s.id)
393             FROM spaces s ORDER BY s.name",
394        )?;
395        let spaces = stmt
396            .query_map([], |r| {
397                Ok(SpaceStatus {
398                    name: r.get(0)?,
399                    revision: r.get(1)?,
400                    episodes: r.get(2)?,
401                    chunks: r.get(3)?,
402                })
403            })?
404            .collect::<std::result::Result<Vec<_>, _>>()?;
405        let (pending_distill, failed_distill) = self.conn.query_row(
406            "SELECT sum(state = 'pending'), sum(state = 'failed') FROM distill_queue",
407            [],
408            |r| {
409                Ok((
410                    r.get::<_, Option<i64>>(0)?.unwrap_or(0),
411                    r.get::<_, Option<i64>>(1)?.unwrap_or(0),
412                ))
413            },
414        )?;
415        Ok(StatusReport {
416            read_only: self.is_read_only(),
417            spaces,
418            embedder_id: self.embedder.id().to_owned(),
419            embedder_dim: self.embedder.dim(),
420            index_dirty: self.get_meta("index_dirty")?.as_deref() == Some("1"),
421            pending_distill,
422            failed_distill,
423            llm_id: self.llm.as_ref().map(|l| l.id().to_owned()),
424        })
425    }
426
427    /// Content and kind of one episode, straight from truth.
428    pub fn episode_content(&self, episode_id: i64) -> Result<(String, String)> {
429        self.conn
430            .query_row(
431                "SELECT content, kind FROM episodes WHERE id = ?1",
432                [episode_id],
433                |r| Ok((r.get(0)?, r.get(1)?)),
434            )
435            .map_err(|e| match e {
436                rusqlite::Error::QueryReturnedNoRows => {
437                    SconeError::NotFound(format!("episode {episode_id}"))
438                }
439                other => SconeError::Db(other),
440            })
441    }
442
443    pub fn data_dir(&self) -> &Path {
444        &self.data_dir
445    }
446
447    pub(crate) fn conn_mut(&mut self) -> &mut Connection {
448        &mut self.conn
449    }
450
451    pub fn schema_version(&self) -> Result<i64> {
452        let v: String = self.conn.query_row(
453            "SELECT value FROM meta WHERE key = 'schema_version'",
454            [],
455            |r| r.get(0),
456        )?;
457        v.parse()
458            .map_err(|_| SconeError::Index("schema_version is not a number".into()))
459    }
460}
scone_core/lib.rs

scone_core/
lib.rs
