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

scone_core/
lib.rs
