scone_core/
ingest.rs

1//! Lane 1: episodic ingest (spec §6).
2//!
3//! Synchronous and offline-complete. One SQLite transaction is the unit of
4//! truth: episode, chunks, and embeddings commit together or not at all.
5//! Failures are typed values; nothing is ever deleted to handle one
6//! (memory/bugs.md P-2). Duplicates are a first-class outcome detected by
7//! UNIQUE(space_id, hash), never by matching error strings (P-5).
8
9use std::path::PathBuf;
10
11use crate::Engine;
12use crate::auth::ScopedSpace;
13use crate::chunker::chunk_text;
14use crate::error::{Result, SconeError};
15
16/// Default target chunk size in bytes (~250 tokens).
17pub(crate) const CHUNK_TARGET_BYTES: usize = 1000;
18
19const KINDS: [&str; 4] = ["note", "file", "conversation", "observation"];
20
21#[derive(Debug)]
22pub enum IngestInput {
23    Note { text: String },
24    File { path: PathBuf },
25}
26
27#[derive(Debug)]
28pub enum IngestOutcome {
29    Ingested { episode_id: i64, chunks: usize },
30    Deduplicated { episode_id: i64 },
31}
32
33impl Engine {
34    pub fn ingest(&mut self, space: &ScopedSpace, input: IngestInput) -> Result<IngestOutcome> {
35        let (kind, content, source) = match input {
36            IngestInput::Note { text } => ("note", text, None),
37            IngestInput::File { path } => {
38                let bytes = std::fs::read(&path)?;
39                let text = String::from_utf8(bytes).map_err(|_| {
40                    SconeError::InvalidInput(format!("{} is not valid UTF-8", path.display()))
41                })?;
42                ("file", text, Some(path.display().to_string()))
43            }
44        };
45        self.ingest_raw(space, kind, &content, source.as_deref(), None)
46    }
47
48    /// Import-grade ingest: explicit kind/source/created_at, same pipeline.
49    /// Returns the episode id and whether it was freshly stored.
50    pub(crate) fn import_episode(
51        &mut self,
52        space: &ScopedSpace,
53        kind: &str,
54        content: &str,
55        source: Option<&str>,
56        created_at: Option<&str>,
57    ) -> Result<(i64, bool)> {
58        match self.ingest_raw(space, kind, content, source, created_at)? {
59            IngestOutcome::Ingested { episode_id, .. } => Ok((episode_id, true)),
60            IngestOutcome::Deduplicated { episode_id } => Ok((episode_id, false)),
61        }
62    }
63
64    fn ingest_raw(
65        &mut self,
66        space: &ScopedSpace,
67        kind: &str,
68        content: &str,
69        source: Option<&str>,
70        created_at: Option<&str>,
71    ) -> Result<IngestOutcome> {
72        if !KINDS.contains(&kind) {
73            return Err(SconeError::InvalidInput(format!(
74                "kind must be one of {KINDS:?}, got {kind:?}"
75            )));
76        }
77        if content.trim().is_empty() {
78            return Err(SconeError::InvalidInput("content is empty".into()));
79        }
80        self.require_writable()?;
81
82        let hash = blake3::hash(content.as_bytes()).to_hex().to_string();
83        let spans = chunk_text(content, self.chunk_target);
84        let texts: Vec<&str> = spans.iter().map(|s| &content[s.start..s.end]).collect();
85        let embeddings = self.embedder.embed(&texts)?;
86
87        let tx = self.conn.transaction()?;
88        let inserted = tx.execute(
89            "INSERT INTO episodes (space_id, kind, content, hash, source, created_at)
90             VALUES (?1, ?2, ?3, ?4, ?5,
91                     COALESCE(?6, strftime('%Y-%m-%dT%H:%M:%fZ','now')))
92             ON CONFLICT (space_id, hash) DO NOTHING",
93            rusqlite::params![space.id(), kind, content, hash, source, created_at],
94        )?;
95        if inserted == 0 {
96            let episode_id = tx.query_row(
97                "SELECT id FROM episodes WHERE space_id = ?1 AND hash = ?2",
98                rusqlite::params![space.id(), hash],
99                |r| r.get(0),
100            )?;
101            // Nothing was written; the revision must not move (bugs.md P-8).
102            return Ok(IngestOutcome::Deduplicated { episode_id });
103        }
104        let episode_id = tx.last_insert_rowid();
105        let mut chunk_ids = Vec::with_capacity(spans.len());
106        {
107            let mut stmt = tx.prepare(
108                "INSERT INTO chunks (episode_id, pos, start_byte, end_byte, embedding)
109                 VALUES (?1, ?2, ?3, ?4, ?5)",
110            )?;
111            for (pos, (span, emb)) in spans.iter().zip(&embeddings).enumerate() {
112                let blob: Vec<u8> = emb.iter().flat_map(|f| f.to_le_bytes()).collect();
113                stmt.execute(rusqlite::params![
114                    episode_id,
115                    pos as i64,
116                    span.start as i64,
117                    span.end as i64,
118                    blob
119                ])?;
120                chunk_ids.push(tx.last_insert_rowid());
121            }
122        }
123        tx.execute(
124            "UPDATE spaces SET revision = revision + 1 WHERE id = ?1",
125            [space.id()],
126        )?;
127        // Lane 2 is asynchronous: enqueue for distillation, never block
128        // ingest on a model (spec §6).
129        tx.execute(
130            "INSERT OR IGNORE INTO distill_queue (episode_id) VALUES (?1)",
131            [episode_id],
132        )?;
133        // Pin the embedder identity on first write (spec §9); later opens
134        // with a different embedder are refused until doctor --rebuild.
135        tx.execute(
136            "INSERT OR IGNORE INTO meta (key, value) VALUES ('embedder_id', ?1)",
137            [self.embedder.id()],
138        )?;
139        tx.execute(
140            "INSERT OR IGNORE INTO meta (key, value) VALUES ('embedder_dim', ?1)",
141            [self.embedder.dim().to_string()],
142        )?;
143        tx.commit()?;
144
145        // Feed the derived indexes after truth commits. An index failure
146        // never fails the ingest: it marks the indexes dirty for
147        // `doctor --rebuild` and the status surface says so (spec §10).
148        let fts_rows: Vec<(u64, u64, &str)> = chunk_ids
149            .iter()
150            .zip(&texts)
151            .map(|(id, text)| (*id as u64, space.id() as u64, *text))
152            .collect();
153        let vec_rows: Vec<(u64, &[f32])> = chunk_ids
154            .iter()
155            .zip(&embeddings)
156            .map(|(id, emb)| (*id as u64, emb.as_slice()))
157            .collect();
158        let index_result = self
159            .fts
160            .add(&fts_rows)
161            .and_then(|()| self.vectors.add(&vec_rows));
162        match index_result {
163            Ok(()) => self.indexes_dirty = true,
164            Err(_) => self.set_meta("index_dirty", "1")?,
165        }
166
167        Ok(IngestOutcome::Ingested {
168            episode_id,
169            chunks: spans.len(),
170        })
171    }
172}
173
174/// Result of one directory scan.
175#[derive(Debug, Default)]
176pub struct ScanReport {
177    pub ingested: usize,
178    pub deduplicated: usize,
179    pub skipped: usize,
180}
181
182impl Engine {
183    /// Recursively ingest text files under `dir`. Hidden entries and
184    /// dependency/build directories are skipped; binary and oversized
185    /// files are counted in `skipped`, never silently dropped (spec §10).
186    /// Content-hash dedup makes rescans cheap and edits append-only.
187    pub fn ingest_directory(
188        &mut self,
189        space: &ScopedSpace,
190        dir: &std::path::Path,
191        max_file_bytes: u64,
192    ) -> Result<ScanReport> {
193        const SKIP_DIRS: [&str; 4] = ["node_modules", "target", ".git", "__pycache__"];
194        let mut report = ScanReport::default();
195        let mut stack = vec![dir.to_path_buf()];
196        while let Some(current) = stack.pop() {
197            for entry in std::fs::read_dir(&current)? {
198                let entry = entry?;
199                let path = entry.path();
200                let name = entry.file_name();
201                let name = name.to_string_lossy();
202                if name.starts_with('.') {
203                    continue;
204                }
205                let file_type = entry.file_type()?;
206                if file_type.is_dir() {
207                    if !SKIP_DIRS.contains(&name.as_ref()) {
208                        stack.push(path);
209                    }
210                    continue;
211                }
212                if !file_type.is_file() {
213                    continue;
214                }
215                if entry.metadata()?.len() > max_file_bytes {
216                    report.skipped += 1;
217                    continue;
218                }
219                match self.ingest(space, IngestInput::File { path }) {
220                    Ok(IngestOutcome::Ingested { .. }) => report.ingested += 1,
221                    Ok(IngestOutcome::Deduplicated { .. }) => report.deduplicated += 1,
222                    Err(SconeError::InvalidInput(_)) => report.skipped += 1,
223                    Err(other) => return Err(other),
224                }
225            }
226        }
227        Ok(report)
228    }
229}
scone_core/ingest.rs

scone_core/
ingest.rs
