diff --git a/crates/lore-core/migrations/0015_batch_fts.sql b/crates/lore-core/migrations/0015_batch_fts.sql new file mode 100644 index 0000000..8cae1ef --- /dev/null +++ b/crates/lore-core/migrations/0015_batch_fts.sql @@ -0,0 +1,14 @@ +-- Migration 0015: ingest indexes its new search rows itself, once per +-- transaction. +-- +-- FTS5 flushes its pending terms to disk at each statement that writes it. +-- Ingest inserts search rows one statement at a time, so the AFTER INSERT +-- trigger wrote one index segment per row. Ingest now inserts a transaction's +-- new rows into search_fts with one statement, in rowid order +-- (`crate::ingest`). +-- +-- The DELETE and UPDATE triggers stay. Every delete removes its rows in one +-- statement, so the DELETE trigger already flushes once per delete, and +-- nothing updates search_document. The existing index is already in sync, so +-- this migration changes only how later rows reach it. +DROP TRIGGER search_document_ai; diff --git a/crates/lore-core/src/ingest.rs b/crates/lore-core/src/ingest.rs index 5ae3613..786a695 100644 --- a/crates/lore-core/src/ingest.rs +++ b/crates/lore-core/src/ingest.rs @@ -110,6 +110,7 @@ fn persist_rows( "DELETE FROM agent_session WHERE id = ?1", params![session_id], )?; + let high_water = search_high_water(tx)?; // Deleting the logical session above cascades its source links. Preserve // links from copies/continued artifacts so replaying one source does not @@ -164,9 +165,35 @@ fn persist_rows( patch_blobs, RowBase::default(), )?; + index_search_rows(tx, high_water)?; Ok(session_id) } +/// The highest search row id. Rows inserted after reading it get higher ids, +/// since a new INTEGER PRIMARY KEY is one past the largest. Read it after the +/// transaction's deletes, which can free the largest id for reuse. +fn search_high_water(tx: &Connection) -> Result { + Ok(tx.query_row( + "SELECT coalesce(max(id), 0) FROM search_document", + [], + |r| r.get(0), + )?) +} + +/// Index the search rows above `high_water` with one statement, in rowid +/// order. FTS5 flushes at each statement that writes it, and ingest inserts +/// one search row per statement, so migration 0015 dropped the insert trigger +/// that indexed each row as it landed. A rowid below the last one also forces +/// a flush, hence the order. +fn index_search_rows(tx: &Connection, high_water: i64) -> Result<()> { + tx.execute( + "INSERT INTO search_fts(rowid, redacted_text) + SELECT id, redacted_text FROM search_document WHERE id > ?1 ORDER BY id", + [high_water], + )?; + Ok(()) +} + /// A native title (custom/ai) is distinct content, so it is scanned and /// indexed like any cleartext. A synthetic fallback title merely echoes the /// first user message — already scanned and indexed — so re-scanning or @@ -552,7 +579,8 @@ fn append_rows( )?; // Only a native title is projected, so only a new one changes the index. - if from.native_title != end.native_title { + let title_changed = from.native_title != end.native_title; + if title_changed { tx.execute( "DELETE FROM search_document WHERE session_id = ?1 AND source_kind = 'session' AND source_id = ?1 @@ -564,6 +592,9 @@ fn append_rows( WHERE source_kind = 'session' AND source_id = ?1 AND field = 'title'", [session_id], )?; + } + let high_water = search_high_water(tx)?; + if title_changed { project_title(tx, session_id, agent_id, parsed)?; } @@ -579,6 +610,7 @@ fn append_rows( file_events: from.file_event_count, }, )?; + index_search_rows(tx, high_water)?; for result in &append.earlier_results { let (seq, ordinal) = result.result_ref; diff --git a/crates/lore-core/src/storage/migrations.rs b/crates/lore-core/src/storage/migrations.rs index ff0b52b..ff22748 100644 --- a/crates/lore-core/src/storage/migrations.rs +++ b/crates/lore-core/src/storage/migrations.rs @@ -88,6 +88,11 @@ const MIGRATIONS: &[Migration] = &[ name: "ingest_resume", sql: include_str!("../../migrations/0014_ingest_resume.sql"), }, + Migration { + version: 15, + name: "batch_fts", + sql: include_str!("../../migrations/0015_batch_fts.sql"), + }, ]; /// How many migrations exist. Tests assert against this rather than a literal diff --git a/crates/lore-core/tests/fts_sync.rs b/crates/lore-core/tests/fts_sync.rs new file mode 100644 index 0000000..16680d8 --- /dev/null +++ b/crates/lore-core/tests/fts_sync.rs @@ -0,0 +1,303 @@ +//! Ingest writes the FTS index itself, once per transaction (migration 0015). +//! After every way rows reach or leave `search_document`, a search finds +//! exactly the rows it holds, and FTS5 agrees its index matches that table. +#![allow(clippy::expect_used, clippy::unwrap_used)] + +use lore_core::adapters::claude_code::ClaudeCodeAdapter; +use lore_core::forget::forget_session; +use lore_core::ingest::{ingest_file, ChangeClass, IngestOutcome}; +use lore_core::storage::blob::BlobStore; +use rusqlite::Connection; +use serde_json::json; + +/// The text of each row the index returns for `word`, in rowid order. `None` +/// is an index entry whose row is gone. +fn hits(conn: &Connection, word: &str) -> Vec> { + let mut stmt = conn + .prepare( + "SELECT search_document.redacted_text FROM search_fts + LEFT JOIN search_document ON search_document.id = search_fts.rowid + WHERE search_fts MATCH ?1 ORDER BY search_fts.rowid", + ) + .unwrap(); + let rows = stmt + .query_map([word], |r| r.get(0)) + .unwrap() + .collect::>>() + .unwrap(); + rows +} + +fn hit(text: &str) -> Vec> { + vec![Some(text.to_owned())] +} + +/// FTS5's integrity check. The `rank = 1` form also compares the index with +/// the external content table; the plain form only checks the index itself. +fn integrity(conn: &Connection) -> rusqlite::Result<()> { + conn.execute_batch( + "INSERT INTO search_fts(search_fts) VALUES('integrity-check'); + INSERT INTO search_fts(search_fts, rank) VALUES('integrity-check', 1);", + ) +} + +fn line(session: &str, event: serde_json::Value) -> String { + let mut v = json!({"sessionId": session, "cwd": "/repo/app", + "timestamp": "2026-08-10T10:00:00.000Z"}); + v.as_object_mut() + .unwrap() + .extend(event.as_object().unwrap().clone()); + format!("{v}\n") +} + +fn user(session: &str, uuid: &str, text: &str) -> String { + line( + session, + json!({"type": "user", "uuid": uuid, "parentUuid": null, + "message": {"role": "user", "content": text}}), + ) +} + +fn assistant(session: &str, uuid: &str, parent: &str, text: &str) -> String { + line( + session, + json!({"type": "assistant", "uuid": uuid, "parentUuid": parent, + "message": {"role": "assistant", "model": "claude-x", + "content": [{"type": "text", "text": text}]}}), + ) +} + +fn custom_title(session: &str, title: &str) -> String { + line( + session, + json!({"type": "custom-title", "customTitle": title}), + ) +} + +struct Archive { + _dir: tempfile::TempDir, + root: std::path::PathBuf, + blobs: BlobStore, + conn: Connection, +} + +impl Archive { + fn new() -> Self { + let dir = tempfile::tempdir().unwrap(); + let root = dir.path().to_path_buf(); + let blobs = BlobStore::open(root.join("blobs")).unwrap(); + let conn = lore_core::storage::open_in_memory().unwrap(); + conn.execute_batch( + "CREATE TEMP TABLE replaced (session_id TEXT); + CREATE TEMP TRIGGER log_replace AFTER DELETE ON agent_session + BEGIN INSERT INTO replaced VALUES (old.id); END;", + ) + .unwrap(); + Archive { + _dir: dir, + root, + blobs, + conn, + } + } + + /// Write `content` to `file` and ingest it. Returns the session id, the + /// change class, and whether ingest replaced an existing session. + fn ingest(&self, file: &str, content: &str) -> (String, ChangeClass, bool) { + let path = self.root.join(file); + std::fs::write(&path, content).unwrap(); + self.conn.execute_batch("DELETE FROM replaced").unwrap(); + let outcome = ingest_file(&self.conn, &ClaudeCodeAdapter::new(), &path, &self.blobs); + let replaced: i64 = self + .conn + .query_row("SELECT count(*) FROM replaced", [], |r| r.get(0)) + .unwrap(); + let (session_id, change) = match outcome.unwrap() { + IngestOutcome::Ingested { + session_id, change, .. + } => Some((session_id, change)), + IngestOutcome::Skipped => None, + } + .expect("a changed file must be ingested"); + (session_id, change, replaced > 0) + } +} + +#[test] +fn the_content_check_catches_a_row_missing_from_the_index() { + let archive = Archive::new(); + let (session, _, _) = archive.ingest("a.jsonl", &user("s-a", "u1", "Tune the quokka cache")); + integrity(&archive.conn).unwrap(); + archive + .conn + .execute( + "INSERT INTO search_document + (session_id, source_kind, source_id, field, redacted_text, created_at) + VALUES (?1, 'session', ?1, 'title', 'An unindexed wombat', 1)", + [&session], + ) + .unwrap(); + assert_eq!(hits(&archive.conn, "wombat"), Vec::>::new()); + assert!(integrity(&archive.conn).is_err()); +} + +#[test] +fn search_matches_the_archive_through_ingest_append_replace_and_forget() { + let archive = Archive::new(); + let first = user("s-a", "u1", "Tune the quokka cache"); + let second = assistant("s-a", "a1", "u1", "The wombat route is cold"); + let third = user("s-a", "u2", "Restart the numbat worker"); + + let (session, change, _) = archive.ingest("a.jsonl", &format!("{first}{second}")); + assert_eq!(change, ChangeClass::New); + assert_eq!(hits(&archive.conn, "quokka"), hit("Tune the quokka cache")); + assert_eq!( + hits(&archive.conn, "wombat"), + hit("The wombat route is cold") + ); + integrity(&archive.conn).unwrap(); + + let (appended, change, replaced) = + archive.ingest("a.jsonl", &format!("{first}{second}{third}")); + assert_eq!( + (appended.as_str(), change, replaced), + (session.as_str(), ChangeClass::Appended, false) + ); + assert_eq!(hits(&archive.conn, "quokka"), hit("Tune the quokka cache")); + assert_eq!( + hits(&archive.conn, "wombat"), + hit("The wombat route is cold") + ); + assert_eq!( + hits(&archive.conn, "numbat"), + hit("Restart the numbat worker") + ); + integrity(&archive.conn).unwrap(); + + let rewritten = user("s-a", "u1", "Tune the platypus cache"); + let (replaced_id, change, replaced) = + archive.ingest("a.jsonl", &format!("{rewritten}{second}{third}")); + assert_eq!( + (replaced_id.as_str(), change, replaced), + (session.as_str(), ChangeClass::Rewritten, true) + ); + assert_eq!(hits(&archive.conn, "quokka"), Vec::>::new()); + assert_eq!( + hits(&archive.conn, "platypus"), + hit("Tune the platypus cache") + ); + assert_eq!( + hits(&archive.conn, "wombat"), + hit("The wombat route is cold") + ); + assert_eq!( + hits(&archive.conn, "numbat"), + hit("Restart the numbat worker") + ); + integrity(&archive.conn).unwrap(); + + archive.ingest("b.jsonl", &user("s-b", "u1", "Plant the kiwi seeds")); + forget_session(&archive.conn, &archive.blobs, &session).unwrap(); + for word in ["platypus", "wombat", "numbat"] { + assert_eq!( + hits(&archive.conn, word), + Vec::>::new(), + "{word}" + ); + } + assert_eq!(hits(&archive.conn, "kiwi"), hit("Plant the kiwi seeds")); + integrity(&archive.conn).unwrap(); +} + +#[test] +fn a_source_rewritten_into_another_session_drops_the_old_sessions_rows() { + let archive = Archive::new(); + let (old, _, _) = archive.ingest("a.jsonl", &user("s-a", "u1", "Tune the quokka cache")); + let (new, change, _) = archive.ingest("a.jsonl", &user("s-z", "u1", "Tune the platypus cache")); + assert_ne!(new, old); + assert_eq!(change, ChangeClass::Rewritten); + assert_eq!(hits(&archive.conn, "quokka"), Vec::>::new()); + assert_eq!( + hits(&archive.conn, "platypus"), + hit("Tune the platypus cache") + ); + integrity(&archive.conn).unwrap(); +} + +/// The title is the archive's highest search row here, so deleting it frees +/// that id for the new title an append writes. +#[test] +fn an_appended_title_change_indexes_the_new_title() { + let archive = Archive::new(); + let thinking = line( + "s-t", + json!({"type": "assistant", "uuid": "a1", "parentUuid": null, + "message": {"role": "assistant", "model": "claude-x", + "content": [{"type": "thinking", "thinking": "hidden", "signature": "sig"}]}}), + ); + let head = format!("{}{thinking}", custom_title("s-t", "Quokka plans")); + archive.ingest("t.jsonl", &head); + assert_eq!(hits(&archive.conn, "quokka"), hit("Quokka plans")); + + let tail = format!( + "{}{}", + custom_title("s-t", "Numbat plans"), + user("s-t", "u1", "Feed the wombat") + ); + let (_, change, replaced) = archive.ingest("t.jsonl", &format!("{head}{tail}")); + assert_eq!((change, replaced), (ChangeClass::Appended, false)); + assert_eq!(hits(&archive.conn, "quokka"), Vec::>::new()); + assert_eq!(hits(&archive.conn, "numbat"), hit("Numbat plans")); + assert_eq!(hits(&archive.conn, "wombat"), hit("Feed the wombat")); + integrity(&archive.conn).unwrap(); +} + +/// A schema-14 archive is this schema plus the per-row insert trigger, with an +/// index already in sync. Migration 0015 must leave it in sync, and later +/// writes must keep it so. +#[test] +fn an_archive_indexed_by_the_insert_trigger_stays_in_sync_after_migration_0015() { + let archive = Archive::new(); + let quokka = user("s-a", "u1", "Tune the quokka cache"); + archive.ingest("a.jsonl", &quokka); + archive + .conn + .execute_batch( + "CREATE TRIGGER search_document_ai AFTER INSERT ON search_document BEGIN + INSERT INTO search_fts(rowid, redacted_text) VALUES (new.id, new.redacted_text); + END; + DELETE FROM schema_migrations WHERE version = 15; + INSERT INTO agent (id, display_name) VALUES ('old', 'Old'); + INSERT INTO agent_session (id, agent_id, dedupe_key) VALUES ('s-old', 'old', 'k'); + INSERT INTO search_document + (session_id, source_kind, source_id, field, redacted_text, created_at) + VALUES ('s-old', 'session', 's-old', 'title', 'Migrate the axolotl ledger', 1);", + ) + .unwrap(); + assert_eq!( + hits(&archive.conn, "axolotl"), + hit("Migrate the axolotl ledger") + ); + integrity(&archive.conn).unwrap(); + + lore_core::storage::migrations::run(&archive.conn).unwrap(); + assert_eq!(hits(&archive.conn, "quokka"), hit("Tune the quokka cache")); + assert_eq!( + hits(&archive.conn, "axolotl"), + hit("Migrate the axolotl ledger") + ); + integrity(&archive.conn).unwrap(); + + archive.ingest("b.jsonl", &user("s-b", "u1", "Plant the kiwi seeds")); + assert_eq!(hits(&archive.conn, "kiwi"), hit("Plant the kiwi seeds")); + let (_, change, _) = archive.ingest("a.jsonl", &user("s-a", "u1", "Tune the platypus cache")); + assert_eq!(change, ChangeClass::Rewritten); + assert_eq!(hits(&archive.conn, "quokka"), Vec::>::new()); + assert_eq!( + hits(&archive.conn, "platypus"), + hit("Tune the platypus cache") + ); + forget_session(&archive.conn, &archive.blobs, "s-old").unwrap(); + assert_eq!(hits(&archive.conn, "axolotl"), Vec::>::new()); + integrity(&archive.conn).unwrap(); +} diff --git a/crates/lore-core/tests/incremental_ingest.rs b/crates/lore-core/tests/incremental_ingest.rs index 4d8b649..b52aa0f 100644 --- a/crates/lore-core/tests/incremental_ingest.rs +++ b/crates/lore-core/tests/incremental_ingest.rs @@ -168,8 +168,11 @@ fn grow_and_compare( .unwrap(); kept.push(replaced == 0); } - live.execute_batch("INSERT INTO search_fts(search_fts) VALUES('integrity-check');") - .unwrap(); + // `rank = 1` also checks the index against search_document. + live.execute_batch( + "INSERT INTO search_fts(search_fts, rank) VALUES('integrity-check', 1);", + ) + .unwrap(); let fresh = lore_core::storage::open_in_memory().unwrap(); let fresh_id = session_id(ingest_file(&fresh, adapter, &path, &blobs).unwrap()); diff --git a/crates/lore-core/tests/schema.rs b/crates/lore-core/tests/schema.rs index 55059e2..83baaae 100644 --- a/crates/lore-core/tests/schema.rs +++ b/crates/lore-core/tests/schema.rs @@ -3,8 +3,6 @@ //! external-content FTS5 index tracks `search_document`. #![allow(clippy::unwrap_used, clippy::expect_used)] -use rusqlite::params; - fn db() -> rusqlite::Connection { lore_core::storage::open_in_memory().unwrap() } @@ -180,14 +178,16 @@ fn session_uniqueness_null_semantics() { } #[test] -fn fts_tracks_search_document() { +fn deleting_a_search_document_de_indexes_it() { let conn = db(); seed_session(&conn); - conn.execute( + // Ingest indexes new rows itself (migration 0015), as here. + conn.execute_batch( "INSERT INTO search_document (session_id, source_kind, source_id, field, redacted_text, created_at) - VALUES ('s1','message_part','p1','text','fix the stripe webhook signature',1)", - params![], + VALUES ('s1','message_part','p1','text','fix the stripe webhook signature',1); + INSERT INTO search_fts(rowid, redacted_text) + SELECT id, redacted_text FROM search_document;", ) .unwrap(); let hits: i64 = conn @@ -197,7 +197,7 @@ fn fts_tracks_search_document() { |r| r.get(0), ) .unwrap(); - assert_eq!(hits, 1, "inserting a search_document must index it in FTS5"); + assert_eq!(hits, 1); // Deleting the document must remove it from the FTS index too. conn.execute("DELETE FROM search_document WHERE source_id = 'p1'", [])