diff --git a/src/memory/chunks/embeddings_query.rs b/src/memory/chunks/embeddings_query.rs index 09a3b8d..cdac334 100644 --- a/src/memory/chunks/embeddings_query.rs +++ b/src/memory/chunks/embeddings_query.rs @@ -96,8 +96,10 @@ fn embedding_from_blob(bytes: &[u8], dim: i64, label: &str) -> Result = bytes - .chunks_exact(4) - .map(|c| f32::from_le_bytes([c[0], c[1], c[2], c[3]])) + .as_chunks::<4>() + .0 + .iter() + .map(|c| f32::from_le_bytes(*c)) .collect(); if floats.len() != dim as usize { anyhow::bail!( diff --git a/src/memory/chunks/migrations.rs b/src/memory/chunks/migrations.rs index 25fe6da..43d1aed 100644 --- a/src/memory/chunks/migrations.rs +++ b/src/memory/chunks/migrations.rs @@ -81,8 +81,10 @@ pub(super) fn migrate_legacy_embeddings_to_sidecar( continue; } let vec: Vec = blob - .chunks_exact(4) - .map(|c| f32::from_le_bytes([c[0], c[1], c[2], c[3]])) + .as_chunks::<4>() + .0 + .iter() + .map(|c| f32::from_le_bytes(*c)) .collect(); if is_chunk { set_chunk_embedding_for_signature_tx(&tx, &id, &sig, &vec)?; diff --git a/src/memory/chunks/mod.rs b/src/memory/chunks/mod.rs index 819459f..56419b6 100644 --- a/src/memory/chunks/mod.rs +++ b/src/memory/chunks/mod.rs @@ -76,9 +76,15 @@ use tinycortex_api::chunks as types; #[path = "store_conn_tests.rs"] mod store_conn_tests; #[cfg(test)] +#[path = "store_delete_tests.rs"] +mod store_delete_tests; +#[cfg(test)] #[path = "store_embed_tests.rs"] mod store_embed_tests; #[cfg(test)] +#[path = "store_list_tests.rs"] +mod store_list_tests; +#[cfg(test)] #[path = "store_tests.rs"] mod store_tests; @@ -127,10 +133,13 @@ pub use store::{ }; pub(crate) use store_delete::remove_unreferenced_content_files; pub use store_delete::{ - delete_chunks_by_owner, delete_chunks_by_source, delete_chunks_by_source_prefix, - delete_orphaned_source_tree, + delete_chunk_by_id, delete_chunks_by_owner, delete_chunks_by_source, + delete_chunks_by_source_prefix, delete_orphaned_source_tree, purge_all, +}; +pub use store_list::{ + count_chunks_matching, list_chunk_details, list_chunks, source_totals, ChunkDetailRow, + ListChunksQuery, SourceTotal, }; -pub use store_list::{list_chunks, ListChunksQuery}; pub use store_sources::{get_chunk_lifecycle_status_tx, set_chunk_lifecycle_status_tx}; // ── Shared internal constants / helpers ───────────────────────────────────── diff --git a/src/memory/chunks/store.rs b/src/memory/chunks/store.rs index 681b8d3..37ef2f8 100644 --- a/src/memory/chunks/store.rs +++ b/src/memory/chunks/store.rs @@ -390,13 +390,21 @@ pub fn extraction_coverage(config: &MemoryConfig) -> Result { /// negative (defensive against a hand-edited or otherwise corrupted DB — this /// silently coerces rather than erroring, since a negative count/seq has no /// valid interpretation but also isn't worth failing the whole read over). +/// `tags_json` is treated the same way, and deliberately so: a value that does +/// not deserialize as `Vec` decodes to no tags, with a warning, rather +/// than failing. Tags are metadata *about* a chunk, so losing them must not +/// lose the chunk — and because this decoder is shared by every listing, the +/// strict reading meant a single corrupted row took out an entire page for +/// every reader, which is the failure a caller can neither see past nor work +/// around. The value can only be malformed if something bypassed this module's +/// own writer, which always stores `serde_json::to_string`. +/// /// `partial_message` is never persisted (it's a transient chunker signal) and /// always decodes to `false`. /// /// # Errors -/// Returns `Err` if `source_kind` fails [`SourceKind::parse`], `tags_json` -/// fails to deserialize as `Vec`, or any of the three timestamp -/// columns fails [`ms_to_utc`] (out-of-range milliseconds). +/// Returns `Err` if `source_kind` fails [`SourceKind::parse`], or any of the +/// three timestamp columns fails [`ms_to_utc`] (out-of-range milliseconds). pub(super) fn row_to_chunk(row: &rusqlite::Row<'_>) -> rusqlite::Result { let id: String = row.get(0)?; let source_kind_s: String = row.get(1)?; @@ -419,9 +427,13 @@ pub(super) fn row_to_chunk(row: &rusqlite::Row<'_>) -> rusqlite::Result { let timestamp = ms_to_utc(ts_ms)?; let time_range = (ms_to_utc(trs_ms)?, ms_to_utc(tre_ms)?); let created_at = ms_to_utc(created_ms)?; - let tags: Vec = serde_json::from_str(&tags_json).map_err(|e| { - rusqlite::Error::FromSqlConversionFailure(8, rusqlite::types::Type::Text, Box::new(e)) - })?; + let tags: Vec = serde_json::from_str(&tags_json).unwrap_or_else(|e| { + log::warn!( + "[memory::chunks] chunk {id}: tags_json did not decode as a string list \ + ({e}); reading the chunk with no tags rather than failing the query" + ); + Vec::new() + }); Ok(Chunk { id, diff --git a/src/memory/chunks/store_delete.rs b/src/memory/chunks/store_delete.rs index d89ed74..10c9f6b 100644 --- a/src/memory/chunks/store_delete.rs +++ b/src/memory/chunks/store_delete.rs @@ -1,6 +1,7 @@ -//! Chunk deletion by source / source-prefix / owner, with cascade cleanup of -//! the dependent score / entity-index / embedding side tables, the source -//! ingest gate, and any on-disk content files. +//! Chunk deletion by source / source-prefix / owner / id, and the whole-tier +//! purge, with cascade cleanup of the dependent score / entity-index / +//! embedding side tables, the source ingest gate, and any on-disk content +//! files. //! //! Unlike OpenHuman, this slice does **not** cascade into summary trees (that //! subsystem is not ported here); it deletes only the chunk-owned rows and @@ -16,7 +17,7 @@ //! scopes must be processed after the database transaction commits. use anyhow::{Context, Result}; -use rusqlite::params; +use rusqlite::{params, OptionalExtension, Transaction}; use std::collections::HashSet; use super::connection::with_connection; @@ -82,17 +83,58 @@ pub fn delete_chunks_by_owner( delete_chunks_by_source_filter(config, source_kind, DeleteFilter::Owner(owner)) } +/// Delete one chunk by its id, with the same dependent-row, ingest-gate and +/// content-file cleanup the source-scoped deletes perform. +/// +/// Returns `1` when the chunk existed and `0` when it did not — deleting an +/// unknown id is not an error, matching the idempotence of the other deletes. +/// +/// The stored source kind is read first because the shared implementation +/// needs it to decide whether the source's ingest gate and source-scoped +/// summary tree have just been orphaned. It cannot go stale between the read +/// and the delete: a chunk id is derived from its source kind, so a row whose +/// kind changed is a row whose id changed. +/// +/// # Errors +/// Returns `Err` if the kind lookup fails, if the stored kind is not one this +/// build knows (a hand-edited DB), or for the reasons +/// `delete_chunks_by_source_filter` documents. +pub fn delete_chunk_by_id(config: &MemoryConfig, chunk_id: &str) -> Result { + let stored_kind = with_connection(config, |conn| { + let stored = conn + .query_row( + "SELECT source_kind FROM mem_tree_chunks WHERE id = ?1", + params![chunk_id], + |row| row.get::<_, String>(0), + ) + .optional() + .context("Failed to read the source kind of the chunk to delete")?; + Ok(stored) + })?; + let Some(stored_kind) = stored_kind else { + return Ok(0); + }; + let source_kind = SourceKind::parse(&stored_kind).map_err(|error| { + anyhow::anyhow!("Chunk {chunk_id} has an unusable source kind: {error}") + })?; + delete_chunks_by_source_filter(config, source_kind, DeleteFilter::Chunk(chunk_id)) +} + #[derive(Clone, Copy)] enum DeleteFilter<'a> { ExactSource(&'a str), SourcePrefix(&'a str), Owner(&'a str), + Chunk(&'a str), } impl<'a> DeleteFilter<'a> { fn value(self) -> &'a str { match self { - Self::ExactSource(value) | Self::SourcePrefix(value) | Self::Owner(value) => value, + Self::ExactSource(value) + | Self::SourcePrefix(value) + | Self::Owner(value) + | Self::Chunk(value) => value, } } @@ -101,12 +143,32 @@ impl<'a> DeleteFilter<'a> { Self::ExactSource(_) => "source_id = ?2", Self::SourcePrefix(_) => "substr(source_id, 1, length(?2)) = ?2", Self::Owner(_) => "owner = ?2", + Self::Chunk(_) => "id = ?2", + } + } + + /// The `source_kind = ?1 AND` prefix the source-scoped filters select on, + /// and nothing for the by-id filter. + /// + /// `id` is the primary key, so a kind ANDed onto it can never narrow the + /// match — it can only make the whole delete silently select nothing when + /// the caller's kind and the stored one disagree. `?1` stays bound either + /// way: SQLite sizes a statement's parameter list by the highest index it + /// names, so an unused `?1` beside a used `?2` is legal, and all four + /// filters keep the one binding site below. + fn source_kind_clause(self) -> &'static str { + match self { + Self::ExactSource(_) | Self::SourcePrefix(_) | Self::Owner(_) => { + "source_kind = ?1 AND " + } + Self::Chunk(_) => "", } } } /// Shared implementation behind [`delete_chunks_by_source`], -/// [`delete_chunks_by_source_prefix`], and [`delete_chunks_by_owner`]. +/// [`delete_chunks_by_source_prefix`], [`delete_chunks_by_owner`], and +/// [`delete_chunk_by_id`]. /// /// Selects matching rows into a temporary SQLite table, then deletes each /// dependent table with one set-based statement before removing the chunk @@ -146,7 +208,8 @@ fn delete_chunks_by_source_filter( (id, source_id, path_scope, content_path, raw_refs_json) SELECT id, source_id, path_scope, content_path, raw_refs_json FROM mem_tree_chunks - WHERE source_kind = ?1 AND {}", + WHERE {}{}", + filter.source_kind_clause(), filter.chunk_predicate() ); tx.execute( @@ -277,7 +340,12 @@ fn delete_chunks_by_source_filter( params![source_kind.as_str(), prefix], )?; } - DeleteFilter::Owner(_) => {} + // Neither an owner nor a single chunk id names a source, so + // neither may remove an ingest gate on its own — chunks of that + // source owned by someone else, or simply other chunks, can still + // be there. The orphan sweep above already takes the gate in the + // case where they are not. + DeleteFilter::Owner(_) | DeleteFilter::Chunk(_) => {} } for scope in &deleted_tree_scopes { @@ -363,6 +431,130 @@ pub fn delete_orphaned_source_tree( Ok(cascaded) } +/// Every table [`purge_all`] empties before `mem_tree_chunks`, in the order it +/// empties them. +/// +/// The order is load-bearing. `PRAGMA foreign_keys` is ON for every connection +/// this module hands out, so a parent whose children are still present cannot +/// be deleted: `mem_tree_summaries` and `mem_tree_buffers` go before the +/// `mem_tree_trees` rows they reference. The four embedding/tombstone sidecars +/// are listed explicitly even though their foreign keys declare +/// `ON DELETE CASCADE` — the same belt-and-braces `purge_global_topic_trees` +/// uses, so this keeps emptying them if a future schema revision drops a +/// cascade. +const PURGE_TABLES_BEFORE_CHUNKS: &[&str] = &[ + "mem_tree_score", + "mem_tree_entity_index", + "mem_tree_entity_edges", + "mem_tree_entity_hotness", + "mem_tree_jobs", + "mem_tree_buffers", + "mem_tree_summary_embeddings", + "mem_tree_summary_reembed_skipped", + "mem_tree_summaries", + "mem_tree_trees", + "mem_tree_chunk_embeddings", + "mem_tree_chunk_reembed_skipped", +]; + +/// Empty the chunk tier: every chunk, every row keyed off one, the summary +/// trees built from them, the ingest gates that would otherwise refuse to +/// re-ingest anything, and the on-disk bodies they pointed at. +/// +/// Returns the total number of rows removed, summed across every table this +/// purge empties — deliberately NOT the chunk-row count that +/// [`delete_chunks_by_source`] and its siblings return. +/// +/// The unit differs from its siblings on purpose, and the reason is the caller +/// rather than this module: the only consumer is a whole-store wipe that has +/// always reported a cross-table sum, so returning chunk rows here would shrink +/// a number a user already reads without anything having changed about what was +/// forgotten. A scoped delete answers "how many chunks did this reach", which is +/// the caller's own unit; a purge answers "how much did emptying the store +/// remove", which is every row it touched. +/// +/// The consequence is that the number moves when this purge learns about a new +/// table. That is accepted: a table added here is a table that was previously +/// left behind, so the count rising is the fix being visible rather than a +/// meaning change. +/// +/// # What it deliberately does not clear +/// +/// `mcp_writes`, the audit trail of tool-driven writes. It records that a +/// write happened, not what memory holds, and discarding an audit log is a +/// decision for whoever wants it discarded rather than a side effect of +/// emptying the store. +/// +/// Summary trees are cleared here, unlike everywhere else in this module: a +/// tree built out of chunks that no longer exist is not memory this store can +/// explain, and the module's rule about not cascading into the summary tier is +/// about *which chunks* a scoped delete may reach beyond, not about leaving a +/// whole-store purge half done. +/// +/// # Errors +/// Returns `Err` if any `DELETE`, the path collection, or the commit fails, in +/// which case the whole purge rolls back and the store is exactly as it was. +/// Content-file removal runs after the commit and is best-effort: a filesystem +/// failure there leaves orphaned files but neither fails nor undoes the +/// database side. +pub fn purge_all(config: &MemoryConfig) -> Result { + let mut content_paths = Vec::new(); + let deleted = with_connection(config, |conn| { + let tx = conn.unchecked_transaction()?; + collect_purge_content_paths(&tx, &mut content_paths)?; + let mut total = 0usize; + for table in PURGE_TABLES_BEFORE_CHUNKS { + total += tx.execute(&format!("DELETE FROM {table}"), [])?; + } + total += tx.execute("DELETE FROM mem_tree_chunks", [])?; + total += tx.execute("DELETE FROM mem_tree_ingested_sources", [])?; + tx.commit()?; + Ok(total) + })?; + remove_chunk_content_files(config, &content_paths); + Ok(deleted) +} + +/// Collect every on-disk path [`purge_all`] is about to orphan: chunk and +/// summary bodies in the content vault, plus the raw archive files chunks +/// point at. +/// +/// Gathered inside the transaction and removed only after it commits, the same +/// order `delete_chunks_by_source_filter` uses, so a purge that rolls back +/// never leaves a deleted file behind a surviving row. Unlike that function +/// there is no reachability question to answer first — nothing survives a +/// purge, so every path found here is unreferenced by definition. +/// +/// A `raw_refs_json` value that does not parse is skipped rather than failing +/// the purge. Its file is then orphaned on disk, which is exactly what its +/// unreadable pointer row already made it. +fn collect_purge_content_paths(tx: &Transaction<'_>, out: &mut Vec) -> Result<()> { + for sql in [ + "SELECT content_path FROM mem_tree_chunks + WHERE content_path IS NOT NULL AND content_path != ''", + "SELECT content_path FROM mem_tree_summaries + WHERE content_path IS NOT NULL AND content_path != ''", + ] { + let mut stmt = tx.prepare(sql)?; + let rows = stmt.query_map([], |row| row.get::<_, String>(0))?; + for row in rows { + out.push(row.context("Failed to read a content path to purge")?); + } + } + let mut stmt = tx.prepare( + "SELECT raw_refs_json FROM mem_tree_chunks + WHERE raw_refs_json IS NOT NULL AND raw_refs_json != ''", + )?; + let rows = stmt.query_map([], |row| row.get::<_, String>(0))?; + for row in rows { + let json = row.context("Failed to read a raw-ref pointer to purge")?; + if let Ok(refs) = serde_json::from_str::>(&json) { + out.extend(refs.into_iter().map(|raw_ref| raw_ref.path)); + } + } + Ok(()) +} + /// Best-effort removal of on-disk chunk content files, with strict sandboxing: /// a `content_path` that escapes the content root (via `..`, an absolute path, /// or a symlink pointing outside) is refused rather than followed. diff --git a/src/memory/chunks/store_delete_tests.rs b/src/memory/chunks/store_delete_tests.rs new file mode 100644 index 0000000..58fe112 --- /dev/null +++ b/src/memory/chunks/store_delete_tests.rs @@ -0,0 +1,290 @@ +//! Unit tests for the deletes `super::store_delete` owns beyond the +//! source-scoped ones already covered in `store_tests`: the by-id delete and +//! the whole-tier purge. + +use super::types::{chunk_id, Chunk, Metadata, SourceKind, SourceRef}; +use super::{ + count_chunks, delete_chunk_by_id, get_chunk, is_source_ingested, purge_all, upsert_chunks, + with_connection, +}; +use crate::memory::config::MemoryConfig; +use chrono::{TimeZone, Utc}; +use rusqlite::params; +use tempfile::TempDir; + +fn test_config() -> (TempDir, MemoryConfig) { + let tmp = TempDir::new().expect("tempdir"); + let cfg = MemoryConfig::new(tmp.path()); + (tmp, cfg) +} + +fn chunk_with(source_id: &str, seq: u32, ts_ms: i64, content: &str) -> Chunk { + let ts = Utc.timestamp_millis_opt(ts_ms).unwrap(); + Chunk { + id: chunk_id(SourceKind::Chat, source_id, seq, content), + content: content.to_string(), + metadata: Metadata { + source_kind: SourceKind::Chat, + source_id: source_id.to_string(), + owner: "alice@example.com".to_string(), + timestamp: ts, + time_range: (ts, ts), + tags: vec!["eng".into()], + source_ref: Some(SourceRef::new(format!("slack://{source_id}/{seq}"))), + path_scope: None, + }, + token_count: 12, + seq_in_source: seq, + created_at: ts, + partial_message: false, + } +} + +fn count_rows(cfg: &MemoryConfig, table: &str) -> i64 { + with_connection(cfg, |conn| { + Ok( + conn.query_row(&format!("SELECT COUNT(*) FROM {table}"), [], |row| { + row.get(0) + })?, + ) + }) + .unwrap() +} + +fn index_entity(cfg: &MemoryConfig, node_id: &str) { + with_connection(cfg, |conn| { + conn.execute( + "INSERT INTO mem_tree_entity_index ( + entity_id, node_id, node_kind, entity_kind, surface, score, timestamp_ms + ) VALUES ('entity:a', ?1, 'leaf', 'org', 'Acme', 0.9, 1700000000000)", + params![node_id], + )?; + Ok(()) + }) + .unwrap(); +} + +#[test] +fn delete_chunk_by_id_takes_one_chunk_and_leaves_the_source() { + let (_tmp, cfg) = test_config(); + let target = chunk_with("slack:#eng", 0, 2_000, "target"); + let sibling = chunk_with("slack:#eng", 1, 1_000, "sibling"); + upsert_chunks(&cfg, &[target.clone(), sibling.clone()]).unwrap(); + index_entity(&cfg, &target.id); + index_entity(&cfg, &sibling.id); + with_connection(&cfg, |conn| { + conn.execute( + "INSERT INTO mem_tree_ingested_sources (source_kind, source_id, ingested_at_ms) + VALUES ('chat', 'slack:#eng', 1700000000000)", + [], + )?; + Ok(()) + }) + .unwrap(); + + assert_eq!(delete_chunk_by_id(&cfg, &target.id).unwrap(), 1); + assert!(get_chunk(&cfg, &target.id).unwrap().is_none()); + assert!(get_chunk(&cfg, &sibling.id).unwrap().is_some()); + assert_eq!(count_rows(&cfg, "mem_tree_entity_index"), 1); + assert!( + is_source_ingested(&cfg, SourceKind::Chat, "slack:#eng").unwrap(), + "a source with surviving chunks keeps its ingest gate" + ); + + // Idempotent, and an unknown id is not an error. + assert_eq!(delete_chunk_by_id(&cfg, &target.id).unwrap(), 0); + assert_eq!(delete_chunk_by_id(&cfg, "no-such-chunk").unwrap(), 0); +} + +/// Seed one row in every table `purge_all` empties, so a table name that does +/// not exist in the schema fails the test rather than passing silently. +fn seed_full_tier(cfg: &MemoryConfig, chunk: &Chunk) { + with_connection(cfg, |conn| { + let tx = conn.unchecked_transaction()?; + tx.execute( + "INSERT INTO mem_tree_score ( + chunk_id, total, token_count_signal, unique_words_signal, metadata_weight, + source_weight, interaction_weight, entity_density, dropped, reason, computed_at_ms + ) VALUES (?1, 1.0, 1.0, 1.0, 1.0, 1.0, 1.0, 1.0, 0, NULL, 1700000000000)", + params![chunk.id], + )?; + tx.execute( + "INSERT INTO mem_tree_entity_index ( + entity_id, node_id, node_kind, entity_kind, surface, score, timestamp_ms + ) VALUES ('entity:a', ?1, 'leaf', 'org', 'Acme', 0.9, 1700000000000)", + params![chunk.id], + )?; + tx.execute( + "INSERT INTO mem_tree_entity_edges (entity_a, entity_b, weight, updated_ms) + VALUES ('entity:a', 'entity:b', 1, 1700000000000)", + [], + )?; + tx.execute( + "INSERT INTO mem_tree_entity_hotness (entity_id, last_updated_ms) + VALUES ('entity:a', 1700000000000)", + [], + )?; + tx.execute( + "INSERT INTO mem_tree_jobs (id, kind, payload_json, available_at_ms, created_at_ms) + VALUES ('job-1', 'score', '{}', 1700000000000, 1700000000000)", + [], + )?; + tx.execute( + "INSERT INTO mem_tree_trees (id, kind, scope, max_level, created_at_ms) + VALUES ('tree-1', 'source', 'slack:#eng', 0, 1700000000000)", + [], + )?; + tx.execute( + "INSERT INTO mem_tree_buffers (tree_id, level, updated_at_ms) + VALUES ('tree-1', 0, 1700000000000)", + [], + )?; + tx.execute( + "INSERT INTO mem_tree_summaries ( + id, tree_id, tree_kind, level, content, token_count, + time_range_start_ms, time_range_end_ms, sealed_at_ms + ) VALUES ('sum-1', 'tree-1', 'source', 1, 'summary', 3, + 1700000000000, 1700000000000, 1700000000000)", + [], + )?; + tx.execute( + "INSERT INTO mem_tree_summary_embeddings ( + summary_id, model_signature, vector, dim, created_at + ) VALUES ('sum-1', 'test/model@3', ?1, 3, 1700000000.0)", + params![vec![1_u8, 2, 3]], + )?; + tx.execute( + "INSERT INTO mem_tree_summary_reembed_skipped ( + summary_id, model_signature, reason, skipped_at_ms + ) VALUES ('sum-1', 'test/model@3', 'terminal', 1700000000000)", + [], + )?; + tx.execute( + "INSERT INTO mem_tree_chunk_embeddings ( + chunk_id, model_signature, vector, dim, created_at + ) VALUES (?1, 'test/model@3', ?2, 3, 1700000000.0)", + params![chunk.id, vec![1_u8, 2, 3]], + )?; + tx.execute( + "INSERT INTO mem_tree_chunk_reembed_skipped ( + chunk_id, model_signature, reason, skipped_at_ms + ) VALUES (?1, 'test/model@3', 'terminal', 1700000000000)", + params![chunk.id], + )?; + tx.execute( + "INSERT INTO mem_tree_ingested_sources (source_kind, source_id, ingested_at_ms) + VALUES ('chat', 'slack:#eng', 1700000000000)", + [], + )?; + tx.commit()?; + Ok(()) + }) + .unwrap(); +} + +const PURGED_TABLES: [&str; 14] = [ + "mem_tree_score", + "mem_tree_entity_index", + "mem_tree_entity_edges", + "mem_tree_entity_hotness", + "mem_tree_jobs", + "mem_tree_buffers", + "mem_tree_summary_embeddings", + "mem_tree_summary_reembed_skipped", + "mem_tree_summaries", + "mem_tree_trees", + "mem_tree_chunk_embeddings", + "mem_tree_chunk_reembed_skipped", + "mem_tree_chunks", + "mem_tree_ingested_sources", +]; + +/// A purge that cannot finish must not half-finish. A child table with a +/// restricting foreign key onto `mem_tree_chunks` makes the chunk delete — the +/// thirteenth of fourteen statements — fail, so every table emptied before it +/// has to come back. +#[test] +fn purge_all_rolls_back_when_one_delete_fails() { + let (_tmp, cfg) = test_config(); + let chunk = chunk_with("slack:#eng", 0, 1_000, "body"); + upsert_chunks(&cfg, std::slice::from_ref(&chunk)).unwrap(); + seed_full_tier(&cfg, &chunk); + + let before: Vec = PURGED_TABLES + .iter() + .map(|table| count_rows(&cfg, table)) + .collect(); + assert!(before.iter().all(|count| *count == 1)); + + with_connection(&cfg, |conn| { + conn.execute_batch( + "CREATE TABLE purge_block ( + chunk_id TEXT NOT NULL REFERENCES mem_tree_chunks(id) + );", + )?; + conn.execute( + "INSERT INTO purge_block (chunk_id) VALUES (?1)", + params![chunk.id], + )?; + Ok(()) + }) + .unwrap(); + + assert!( + purge_all(&cfg).is_err(), + "the blocked chunk delete must surface as an error" + ); + let after: Vec = PURGED_TABLES + .iter() + .map(|table| count_rows(&cfg, table)) + .collect(); + assert_eq!( + after, before, + "a failed purge must leave every table intact" + ); + + with_connection(&cfg, |conn| { + conn.execute_batch("DROP TABLE purge_block;")?; + Ok(()) + }) + .unwrap(); + + assert_eq!( + purge_all(&cfg).unwrap(), + PURGED_TABLES.len(), + "the count is every row removed, and each table above holds exactly one" + ); + for table in PURGED_TABLES { + assert_eq!(count_rows(&cfg, table), 0, "{table} must be empty"); + } + assert_eq!(count_chunks(&cfg).unwrap(), 0); + // Idempotent: purging an already-empty store is not an error. + assert_eq!(purge_all(&cfg).unwrap(), 0); +} + +#[test] +fn purge_all_leaves_the_write_audit_trail_alone() { + let (_tmp, cfg) = test_config(); + let chunk = chunk_with("slack:#eng", 0, 1_000, "body"); + upsert_chunks(&cfg, std::slice::from_ref(&chunk)).unwrap(); + with_connection(&cfg, |conn| { + conn.execute( + "INSERT INTO mcp_writes (timestamp_ms, client_info, tool_name, success) + VALUES (1700000000000, 'test-client', 'remember', 1)", + [], + )?; + Ok(()) + }) + .unwrap(); + + let expected: i64 = PURGED_TABLES + .iter() + .map(|table| count_rows(&cfg, table)) + .sum(); + assert_eq!(purge_all(&cfg).unwrap() as i64, expected); + assert_eq!( + count_rows(&cfg, "mcp_writes"), + 1, + "an audit record of a write is not the memory it wrote" + ); +} diff --git a/src/memory/chunks/store_list.rs b/src/memory/chunks/store_list.rs index 78447fa..dea9dce 100644 --- a/src/memory/chunks/store_list.rs +++ b/src/memory/chunks/store_list.rs @@ -1,4 +1,6 @@ -//! Filtered and paginated chunk listing. +//! Filtered chunk listing and the reads that must agree with it: the count of +//! what it matched, the detail view of the same page, and the per-source +//! rollup over the same scope. use anyhow::{Context, Result}; @@ -11,6 +13,22 @@ const DEFAULT_LIST_LIMIT: usize = 100; const MAX_LIST_LIMIT: usize = 10_000; /// Optional filters and pagination for [`list_chunks`]. +/// +/// Every field is a filter and they compose with `AND`; the default matches +/// everything the store holds, bounded by the store's own row cap. +/// +/// # An empty list means "unfiltered", not "match nothing" +/// +/// The `Vec` fields are absent-by-default, exactly like the `Option` ones: an +/// empty `ids` does not restrict the id, it declines to. That is forced by +/// [`Default`] — a query that left every list empty would otherwise match no +/// rows at all — and it is the reading a caller deserialising a partial query +/// off the wire gets for free. +/// +/// The cost is real and worth naming: a caller that *computed* a candidate set +/// and got nothing must not hand the empty set to this struct expecting an +/// empty page. It has to short-circuit and skip the query, because the query +/// will happily return everything. #[derive(Debug, Default, Clone)] pub struct ListChunksQuery { /// Exact source-kind filter. @@ -31,6 +49,37 @@ pub struct ListChunksQuery { pub source_scope: Option>, /// Exclude lifecycle-dropped chunks. pub exclude_dropped: bool, + /// Exact chunk ids to keep. + pub ids: Vec, + /// Allowed source kinds. Composes with — does not replace — `source_kind`; + /// setting both keeps only rows satisfying each, which is empty whenever + /// the scalar is not one of the listed kinds. + pub source_kinds: Vec, + /// Allowed logical source ids. Composes with `source_id` on the same terms + /// as `source_kinds` does with `source_kind`. + pub source_ids: Vec, + /// Keep only chunks the entity index links to one of these canonical + /// entity ids. + pub entity_ids: Vec, + /// Keep only chunks the entity index links to an entity of one of these + /// kinds. + /// + /// Independent of `entity_ids`: setting both asks for a chunk carrying + /// *some* listed entity and *some* entity of a listed kind, not for one + /// index row satisfying both. Each predicate therefore means exactly one + /// thing whether or not the other is set, which is worth more than the + /// narrower joint reading — a caller that wants the joint one already + /// knows the kind of the ids it is asking about. + pub entity_kinds: Vec, + /// Substring the stored `content` column must contain, matched literally + /// — `%` and `_` in the value are text, not wildcards — and + /// case-insensitively for ASCII, which is what SQLite's `LIKE` does. + /// + /// This searches the SQLite column, which for a staged chunk holds only a + /// ≤500-character preview of a body that lives in the content vault. A + /// match is therefore proof the text is in the chunk; a miss is not proof + /// it is absent. + pub content_contains: Option, } /// List chunks matching all supplied filters in deterministic newest-first @@ -42,41 +91,9 @@ pub fn list_chunks(config: &MemoryConfig, query: &ListChunksQuery) -> Result> = Vec::new(); - if let Some(kind) = query.source_kind { - sql.push_str(" AND source_kind = ?"); - bound.push(Box::new(kind.as_str().to_string())); - } - for (clause, value) in [ - (" AND source_id = ?", query.source_id.as_ref()), - (" AND owner = ?", query.owner.as_ref()), - ] { - if let Some(value) = value { - sql.push_str(clause); - bound.push(Box::new(value.clone())); - } - } - if let Some(value) = query.since_ms { - sql.push_str(" AND timestamp_ms >= ?"); - bound.push(Box::new(value)); - } - if let Some(value) = query.until_ms { - sql.push_str(" AND timestamp_ms <= ?"); - bound.push(Box::new(value)); - } - if query.exclude_dropped { - sql.push_str(" AND lifecycle_status != ?"); - bound.push(Box::new(CHUNK_STATUS_DROPPED.to_string())); - } - append_source_scope(&mut sql, &mut bound, query.source_scope.as_ref()); - sql.push_str(" ORDER BY timestamp_ms DESC, seq_in_source ASC, id ASC LIMIT ? OFFSET ?"); - bound.push(Box::new(normalized_limit(query.limit))); - bound.push(Box::new( - i64::try_from(query.offset.unwrap_or(0)).unwrap_or(i64::MAX), - )); - let params = bound - .iter() - .map(|value| value.as_ref() as &dyn rusqlite::ToSql) - .collect::>(); + append_filters(&mut sql, &mut bound, query)?; + append_page(&mut sql, &mut bound, query); + let params = as_params(&bound); conn.prepare(&sql)? .query_map(params.as_slice(), row_to_chunk)? .collect::>>() @@ -84,6 +101,382 @@ pub fn list_chunks(config: &MemoryConfig, query: &ListChunksQuery) -> Result, + /// Lifecycle state — `admitted`, `sealed`, `dropped`, … + /// + /// `Option` for the decode, not for the column. The column arrived as an + /// additive `ALTER TABLE` and is declared `TEXT NOT NULL DEFAULT + /// 'admitted'` (`connection.rs`), so SQLite backfilled every pre-existing + /// row and rejects a writer that omits it — a NULL cannot be produced by + /// this schema. What the `Option` buys is that a row which somehow read + /// back empty does not fail the whole page, and that the contract type + /// this maps onto can be answered by a driver with no lifecycle concept at + /// all. + /// + /// This matters for `exclude_dropped`, which filters with + /// `lifecycle_status != 'dropped'`: under SQLite a NULL there would compare + /// NULL and silently drop the row from the page *and* from the count. It + /// does not need a `COALESCE` guard, and the reason is the `NOT NULL` above + /// rather than luck. + pub lifecycle_status: Option, + /// Whether an embedding vector exists for this chunk in **any** signature. + /// + /// Not scoped to a model on purpose — this answers "has this been embedded + /// at all", which is the question an inspection view asks. Whether a + /// *particular* space holds it is + /// [`get_chunk_embedding_for_signature`](super::get_chunk_embedding_for_signature). + pub has_embedding: bool, +} + +/// The same page [`list_chunks`] returns, with the three per-chunk facts an +/// inspection view shows beside each row. +/// +/// One `SELECT`, not four reads per row: `content_path`, `lifecycle_status` +/// and the embedding existence test are all answerable from the chunk row and +/// one correlated `EXISTS`, and a caller rendering a page of them out of +/// process would otherwise pay a round trip per fact per row. +/// +/// Filters, order and page bounds come from the same builders [`list_chunks`] +/// uses, so switching a caller between the two views changes what it learns +/// about each row and nothing about which rows it gets. +/// +/// # Errors +/// Returns an error when SQLite preparation, execution, or row decoding fails. +pub fn list_chunk_details( + config: &MemoryConfig, + query: &ListChunksQuery, +) -> Result> { + with_connection(config, |conn| { + let mut sql = format!( + "SELECT {SELECT_COLUMNS}, + content_path, + lifecycle_status, + EXISTS (SELECT 1 FROM mem_tree_chunk_embeddings e + WHERE e.chunk_id = mem_tree_chunks.id) + FROM mem_tree_chunks WHERE 1=1" + ); + let mut bound: Vec> = Vec::new(); + append_filters(&mut sql, &mut bound, query)?; + append_page(&mut sql, &mut bound, query); + let params = as_params(&bound); + conn.prepare(&sql)? + .query_map(params.as_slice(), |row| { + Ok(ChunkDetailRow { + chunk: row_to_chunk(row)?, + content_path: row + .get::<_, Option>(DETAIL_CONTENT_PATH_COLUMN)? + .filter(|path| !path.is_empty()), + lifecycle_status: row.get(DETAIL_LIFECYCLE_STATUS_COLUMN)?, + has_embedding: row.get::<_, i64>(DETAIL_HAS_EMBEDDING_COLUMN)? != 0, + }) + })? + .collect::>>() + .context("Failed to collect chunk details") + }) +} + +// The three columns `list_chunk_details` selects past `SELECT_COLUMNS`, whose +// fourteen columns `row_to_chunk` consumes as `0..=13`. Named rather than +// spelled inline so adding a column to `SELECT_COLUMNS` moves them in one +// place instead of silently shifting what each `row.get` reads. +const DETAIL_CONTENT_PATH_COLUMN: usize = 14; +const DETAIL_LIFECYCLE_STATUS_COLUMN: usize = 15; +const DETAIL_HAS_EMBEDDING_COLUMN: usize = 16; + +/// How many rows [`list_chunks`] would match for `query`, ignoring its `limit` +/// and `offset`. +/// +/// The predicate is built by the same private filter builder the listing uses, +/// so the two cannot drift apart. That matters more than the saving: a total that +/// disagrees with the page beside it sends a caller paging towards rows that +/// are not there, which is worse than having no total at all. +/// +/// [`count_chunks`](super::count_chunks) answers a different question — every +/// row in the table, no filters — and is not a substitute for this. +/// +/// # Errors +/// Returns an error when SQLite preparation or execution fails. +pub fn count_chunks_matching(config: &MemoryConfig, query: &ListChunksQuery) -> Result { + with_connection(config, |conn| { + let mut sql = String::from("SELECT COUNT(*) FROM mem_tree_chunks WHERE 1=1"); + let mut bound: Vec> = Vec::new(); + append_filters(&mut sql, &mut bound, query)?; + let params = as_params(&bound); + let total: i64 = conn + .query_row(&sql, params.as_slice(), |row| row.get(0)) + .context("Failed to count chunks")?; + Ok(total.max(0) as u64) + }) +} + +/// What the store holds under one `(source_kind, source_id)` pair. +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct SourceTotal { + /// The source kind the chunks were ingested under. + pub source_kind: SourceKind, + /// The logical source id, verbatim as stored. + pub source_id: String, + /// Chunks currently held under this pair, at any lifecycle status. + pub chunk_count: u64, + /// The newest source time among them, in epoch milliseconds. + pub last_timestamp_ms: i64, +} + +/// Every source the store holds chunks for, most-recently-active first. +/// +/// The rollup a "what is in my memory" view needs, answered by one grouped +/// query instead of a listing the caller counts itself: the counts here are +/// over *all* matching rows, where a listing would only ever see one page of +/// them. +/// +/// `scope` is the same memory-source allowlist [`ListChunksQuery::source_scope`] +/// carries and is applied by the same builder, so a scoped caller's totals +/// cover exactly the chunks its scoped listing can reach. `limit` shares the +/// listing's clamp — it bounds returned *groups*, not chunks. +/// +/// Ordering breaks ties down to the group key so the result is stable across +/// calls; without that a caller polling this would watch equally-recent +/// sources swap places for no reason. +/// +/// # Errors +/// Returns an error when SQLite preparation or execution fails, or when a +/// stored `source_kind` is not one this build knows (a hand-edited DB — the +/// same condition that already fails [`list_chunks`]). +pub fn source_totals( + config: &MemoryConfig, + limit: Option, + scope: Option<&std::collections::HashSet>, +) -> Result> { + with_connection(config, |conn| { + let mut sql = String::from( + "SELECT source_kind, source_id, COUNT(*) AS chunk_count, + MAX(timestamp_ms) AS last_timestamp_ms + FROM mem_tree_chunks WHERE 1=1", + ); + let mut bound: Vec> = Vec::new(); + append_source_scope(&mut sql, &mut bound, scope); + sql.push_str( + " GROUP BY source_kind, source_id + ORDER BY last_timestamp_ms DESC, chunk_count DESC, + source_kind ASC, source_id ASC + LIMIT ?", + ); + bound.push(Box::new(normalized_limit(limit))); + let params = as_params(&bound); + conn.prepare(&sql)? + .query_map(params.as_slice(), |row| { + let stored_kind: String = row.get(0)?; + let source_kind = SourceKind::parse(&stored_kind).map_err(|error| { + rusqlite::Error::FromSqlConversionFailure( + 0, + rusqlite::types::Type::Text, + error.into(), + ) + })?; + let chunk_count: i64 = row.get(2)?; + Ok(SourceTotal { + source_kind, + source_id: row.get(1)?, + chunk_count: chunk_count.max(0) as u64, + last_timestamp_ms: row.get(3)?, + }) + })? + .collect::>>() + .context("Failed to collect source totals") + }) +} + +/// Append every `WHERE` term the query asks for, binding in the same order. +/// +/// Shared by [`list_chunks`], [`list_chunk_details`] and +/// [`count_chunks_matching`]: everything that decides *which* rows match lives +/// here, and nothing that decides how many of them are returned does. Ordering +/// and the page bounds stay in [`append_page`] because the count deliberately +/// ignores them — `limit` is a property of the page, not of the result set. +/// +/// # Errors +/// Returns an error when a list predicate's values cannot be encoded as the +/// JSON array they are bound as. +fn append_filters( + sql: &mut String, + bound: &mut Vec>, + query: &ListChunksQuery, +) -> Result<()> { + if let Some(kind) = query.source_kind { + sql.push_str(" AND source_kind = ?"); + bound.push(Box::new(kind.as_str().to_string())); + } + for (clause, value) in [ + (" AND source_id = ?", query.source_id.as_ref()), + (" AND owner = ?", query.owner.as_ref()), + ] { + if let Some(value) = value { + sql.push_str(clause); + bound.push(Box::new(value.clone())); + } + } + if let Some(value) = query.since_ms { + sql.push_str(" AND timestamp_ms >= ?"); + bound.push(Box::new(value)); + } + if let Some(value) = query.until_ms { + sql.push_str(" AND timestamp_ms <= ?"); + bound.push(Box::new(value)); + } + append_in_list(sql, bound, "id", &query.ids)?; + let source_kinds = query + .source_kinds + .iter() + .map(|kind| kind.as_str().to_string()) + .collect::>(); + append_in_list(sql, bound, "source_kind", &source_kinds)?; + append_in_list(sql, bound, "source_id", &query.source_ids)?; + append_entity_exists(sql, bound, "entity_id", &query.entity_ids)?; + append_entity_exists(sql, bound, "entity_kind", &query.entity_kinds)?; + if let Some(value) = query.content_contains.as_deref() { + sql.push_str(" AND content LIKE ? ESCAPE '\\'"); + bound.push(Box::new(like_contains_pattern(value))); + } + if query.exclude_dropped { + sql.push_str(" AND lifecycle_status != ?"); + bound.push(Box::new(CHUNK_STATUS_DROPPED.to_string())); + } + append_source_scope(sql, bound, query.source_scope.as_ref()); + Ok(()) +} + +/// Append the deterministic order and the page bounds. +/// +/// Separate from [`append_filters`] so [`count_chunks_matching`] cannot pick +/// them up, and shared by the two listings so the same query cannot present +/// the same rows in two different orders depending on which view asked. +fn append_page( + sql: &mut String, + bound: &mut Vec>, + query: &ListChunksQuery, +) { + sql.push_str(" ORDER BY timestamp_ms DESC, seq_in_source ASC, id ASC LIMIT ? OFFSET ?"); + bound.push(Box::new(normalized_limit(query.limit))); + bound.push(Box::new( + i64::try_from(query.offset.unwrap_or(0)).unwrap_or(i64::MAX), + )); +} + +/// Append `AND IN (…)` for a non-empty list, binding the whole list +/// as a single JSON array parameter. +/// +/// One parameter however long the list, which is what makes this usable from a +/// filter builder at all. [`get_chunks_batch`](super::get_chunks_batch) binds +/// one placeholder per id and windows the ids to stay under SQLite's +/// bound-parameter ceiling; it can window because it issues one statement per +/// window and merges the results into a map. A filtered listing cannot — the +/// order, `LIMIT` and `OFFSET` are properties of the whole result set, so +/// splitting the statement splits the page and the total stops matching it. +/// Binding through `json_each` removes the ceiling instead of managing it, and +/// is already what the source-scope gate below relies on. +/// +/// # Errors +/// Returns an error when `values` cannot be encoded as a JSON array. +fn append_in_list( + sql: &mut String, + bound: &mut Vec>, + column: &str, + values: &[String], +) -> Result<()> { + let Some(json) = json_list(values)? else { + return Ok(()); + }; + sql.push_str(&format!( + " AND {column} IN (SELECT value FROM json_each(?))" + )); + bound.push(Box::new(json)); + Ok(()) +} + +/// Append `AND EXISTS (…)` over the entity index for a non-empty list. +/// +/// `EXISTS`, not a join: a chunk carrying three of the listed entities has +/// three index rows, and a join would return it three times. That would be a +/// duplicate page, and — worse — a total from [`count_chunks_matching`] three +/// times the number of chunks that actually match, which is exactly the +/// page/total disagreement the shared builder exists to prevent. `EXISTS` +/// cannot multiply rows, so neither read needs a `SELECT COUNT(*) FROM (…)` +/// wrapper or a `DISTINCT` to undo the damage. +/// +/// # Errors +/// Returns an error when `values` cannot be encoded as a JSON array. +fn append_entity_exists( + sql: &mut String, + bound: &mut Vec>, + column: &str, + values: &[String], +) -> Result<()> { + let Some(json) = json_list(values)? else { + return Ok(()); + }; + sql.push_str(&format!( + " AND EXISTS (SELECT 1 FROM mem_tree_entity_index ei + WHERE ei.node_id = mem_tree_chunks.id + AND ei.{column} IN (SELECT value FROM json_each(?)))" + )); + bound.push(Box::new(json)); + Ok(()) +} + +/// Encode a filter list as the JSON array `json_each` reads, or `None` when +/// the list is empty and the predicate is therefore not asked for. +/// +/// # Errors +/// Returns an error when the values cannot be encoded. +fn json_list(values: &[String]) -> Result> { + if values.is_empty() { + return Ok(None); + } + serde_json::to_string(values) + .map(Some) + .context("Failed to encode chunk filter list") +} + +/// Build the `LIKE` operand for "contains `value`", escaping the wildcards so +/// the caller's text is matched literally. +/// +/// `%`, `_` and the escape character itself are escaped with `\`, paired with +/// `ESCAPE '\'` at the call site. Without this a source id or body fragment +/// containing `_` silently matches any character in that position — the same +/// literalness `delete_chunks_by_source_prefix` gets by filtering in Rust +/// rather than with `LIKE`. +fn like_contains_pattern(value: &str) -> String { + let mut pattern = String::with_capacity(value.len() + 2); + pattern.push('%'); + for character in value.chars() { + if matches!(character, '\\' | '%' | '_') { + pattern.push('\\'); + } + pattern.push(character); + } + pattern.push('%'); + pattern +} + fn append_source_scope( sql: &mut String, bound: &mut Vec>, @@ -111,6 +504,14 @@ fn append_source_scope( sql.push(')'); } +/// Borrow the boxed bindings as the slice `rusqlite` takes. +fn as_params(bound: &[Box]) -> Vec<&dyn rusqlite::ToSql> { + bound + .iter() + .map(|value| value.as_ref() as &dyn rusqlite::ToSql) + .collect() +} + fn normalized_limit(requested: Option) -> i64 { let limit = requested .unwrap_or(DEFAULT_LIST_LIMIT) diff --git a/src/memory/chunks/store_list_tests.rs b/src/memory/chunks/store_list_tests.rs new file mode 100644 index 0000000..29440db --- /dev/null +++ b/src/memory/chunks/store_list_tests.rs @@ -0,0 +1,432 @@ +//! Unit tests for the filtered listing surface (`super::store_list`): the +//! predicates every read shares, the detail view over a page, and the +//! per-source rollup. + +use super::types::{chunk_id, Chunk, Metadata, SourceKind, SourceRef}; +use super::{ + count_chunks_matching, list_chunk_details, list_chunks, source_totals, upsert_chunks, + with_connection, ListChunksQuery, +}; +use crate::memory::config::MemoryConfig; +use chrono::{TimeZone, Utc}; +use rusqlite::params; +use std::collections::HashSet; +use tempfile::TempDir; + +fn test_config() -> (TempDir, MemoryConfig) { + let tmp = TempDir::new().expect("tempdir"); + let cfg = MemoryConfig::new(tmp.path()); + (tmp, cfg) +} + +fn chunk_with( + source_kind: SourceKind, + source_id: &str, + seq: u32, + ts_ms: i64, + content: &str, +) -> Chunk { + let ts = Utc.timestamp_millis_opt(ts_ms).unwrap(); + Chunk { + id: chunk_id(source_kind, source_id, seq, content), + content: content.to_string(), + metadata: Metadata { + source_kind, + source_id: source_id.to_string(), + owner: "alice@example.com".to_string(), + timestamp: ts, + time_range: (ts, ts), + tags: vec!["eng".into()], + source_ref: Some(SourceRef::new(format!("slack://{source_id}/{seq}"))), + path_scope: None, + }, + token_count: 12, + seq_in_source: seq, + created_at: ts, + partial_message: false, + } +} + +/// Write one `mem_tree_entity_index` row directly. The scorer owns this table +/// and is not in the chunk-store test path, so the tests seed it themselves. +fn index_entity(cfg: &MemoryConfig, node_id: &str, entity_id: &str, entity_kind: &str) { + with_connection(cfg, |conn| { + conn.execute( + "INSERT INTO mem_tree_entity_index ( + entity_id, node_id, node_kind, entity_kind, surface, score, timestamp_ms + ) VALUES (?1, ?2, 'leaf', ?3, 'surface', 0.9, 1700000000000)", + params![entity_id, node_id, entity_kind], + )?; + Ok(()) + }) + .unwrap(); +} + +fn ids_of(chunks: &[Chunk]) -> Vec { + chunks.iter().map(|chunk| chunk.id.clone()).collect() +} + +// ── The six list predicates ───────────────────────────────────────────────── + +/// Each of the six list predicates is built from one near-miss chunk that +/// differs from the wanted chunk in exactly that dimension and matches it in +/// every other. The full query must return the one chunk; dropping any single +/// predicate must let exactly its own near-miss back in — which is what proves +/// the predicate does work, rather than being a clause that matches nothing. +#[test] +fn every_list_predicate_composes_with_and() { + let (_tmp, cfg) = test_config(); + + let wanted = chunk_with(SourceKind::Chat, "slack:#eng", 0, 6_000, "quarterly plan"); + let other_kind = chunk_with(SourceKind::Email, "slack:#eng", 1, 5_000, "quarterly plan"); + let other_source = chunk_with(SourceKind::Chat, "slack:#ops", 2, 4_000, "quarterly plan"); + let other_entity = chunk_with(SourceKind::Chat, "slack:#eng", 3, 3_000, "quarterly plan"); + let other_entity_kind = chunk_with(SourceKind::Chat, "slack:#eng", 4, 2_000, "quarterly plan"); + let other_content = chunk_with(SourceKind::Chat, "slack:#eng", 5, 1_000, "annual plan"); + let unlisted_id = chunk_with(SourceKind::Chat, "slack:#eng", 6, 7_000, "quarterly plan"); + + let all = [ + wanted.clone(), + other_kind.clone(), + other_source.clone(), + other_entity.clone(), + other_entity_kind.clone(), + other_content.clone(), + unlisted_id.clone(), + ]; + upsert_chunks(&cfg, &all).unwrap(); + + for chunk in [ + &wanted, + &other_kind, + &other_source, + &other_content, + &unlisted_id, + ] { + index_entity(&cfg, &chunk.id, "entity:acme", "org"); + } + // The wanted entity, of the wrong kind: `entity_ids` accepts this chunk and + // `entity_kinds` is the only predicate that may reject it. + index_entity(&cfg, &other_entity_kind.id, "entity:acme", "person"); + // The wanted kind, under the wrong entity: the mirror image. + index_entity(&cfg, &other_entity.id, "entity:other", "org"); + + let full = ListChunksQuery { + ids: ids_of(&[ + wanted.clone(), + other_kind.clone(), + other_source.clone(), + other_entity.clone(), + other_entity_kind.clone(), + other_content.clone(), + ]), + source_kinds: vec![SourceKind::Chat], + source_ids: vec!["slack:#eng".to_string()], + entity_ids: vec!["entity:acme".to_string()], + entity_kinds: vec!["org".to_string()], + content_contains: Some("quarterly".to_string()), + ..Default::default() + }; + + let rows = list_chunks(&cfg, &full).unwrap(); + assert_eq!(ids_of(&rows), vec![wanted.id.clone()]); + assert_eq!(count_chunks_matching(&cfg, &full).unwrap(), 1); + let details = list_chunk_details(&cfg, &full).unwrap(); + assert_eq!(details.len(), 1); + assert_eq!(details[0].chunk.id, wanted.id); + + let readmits = |dropped: &str, query: ListChunksQuery, readmitted: &Chunk| { + let mut got = ids_of(&list_chunks(&cfg, &query).unwrap()); + got.sort(); + let mut expected = vec![wanted.id.clone(), readmitted.id.clone()]; + expected.sort(); + assert_eq!( + got, expected, + "dropping {dropped} must readmit exactly its own near-miss" + ); + assert_eq!( + count_chunks_matching(&cfg, &query).unwrap(), + 2, + "the total must follow the page when {dropped} is dropped" + ); + }; + + readmits( + "ids", + ListChunksQuery { + ids: Vec::new(), + ..full.clone() + }, + &unlisted_id, + ); + readmits( + "source_kinds", + ListChunksQuery { + source_kinds: Vec::new(), + ..full.clone() + }, + &other_kind, + ); + readmits( + "source_ids", + ListChunksQuery { + source_ids: Vec::new(), + ..full.clone() + }, + &other_source, + ); + readmits( + "entity_ids", + ListChunksQuery { + entity_ids: Vec::new(), + ..full.clone() + }, + &other_entity, + ); + readmits( + "entity_kinds", + ListChunksQuery { + entity_kinds: Vec::new(), + ..full.clone() + }, + &other_entity_kind, + ); + readmits( + "content_contains", + ListChunksQuery { + content_contains: None, + ..full.clone() + }, + &other_content, + ); +} + +/// A chunk carrying several matching entities is still one chunk. A join would +/// return it once per index row and inflate the total with it; `EXISTS` cannot. +#[test] +fn entity_predicates_never_multiply_a_chunk() { + let (_tmp, cfg) = test_config(); + let chunk = chunk_with(SourceKind::Chat, "slack:#eng", 0, 1_000, "three entities"); + let single = chunk_with(SourceKind::Chat, "slack:#eng", 1, 2_000, "one entity"); + upsert_chunks(&cfg, &[chunk.clone(), single.clone()]).unwrap(); + for entity in ["entity:a", "entity:b", "entity:c"] { + index_entity(&cfg, &chunk.id, entity, "org"); + } + index_entity(&cfg, &single.id, "entity:a", "org"); + + let query = ListChunksQuery { + entity_ids: vec!["entity:a".into(), "entity:b".into(), "entity:c".into()], + entity_kinds: vec!["org".into()], + ..Default::default() + }; + let rows = list_chunks(&cfg, &query).unwrap(); + assert_eq!(rows.len(), 2, "two chunks, not four index rows"); + assert_eq!(rows.iter().filter(|row| row.id == chunk.id).count(), 1); + assert_eq!(count_chunks_matching(&cfg, &query).unwrap(), 2); + assert_eq!(list_chunk_details(&cfg, &query).unwrap().len(), 2); +} + +/// `_` and `%` inside `content_contains` are text the chunk must contain, not +/// wildcards. Without the `ESCAPE` clause each of these would also match its +/// look-alike. +#[test] +fn content_contains_treats_like_metacharacters_literally() { + let (_tmp, cfg) = test_config(); + let underscore = chunk_with(SourceKind::Chat, "s", 0, 4_000, "release a_b notes"); + let lookalike = chunk_with(SourceKind::Chat, "s", 1, 3_000, "release axb notes"); + let percent = chunk_with(SourceKind::Chat, "s", 2, 2_000, "growth 100% target"); + let no_percent = chunk_with(SourceKind::Chat, "s", 3, 1_000, "growth 1000 target"); + upsert_chunks( + &cfg, + &[ + underscore.clone(), + lookalike.clone(), + percent.clone(), + no_percent.clone(), + ], + ) + .unwrap(); + + for (needle, expected) in [("a_b", &underscore), ("100%", &percent)] { + let query = ListChunksQuery { + content_contains: Some(needle.to_string()), + ..Default::default() + }; + assert_eq!( + ids_of(&list_chunks(&cfg, &query).unwrap()), + vec![expected.id.clone()], + "{needle} must match literally" + ); + assert_eq!(count_chunks_matching(&cfg, &query).unwrap(), 1); + } + + // The escape character itself is escaped too, so a body containing it is + // matched rather than being read as the start of an escape sequence. + let backslash = chunk_with(SourceKind::Chat, "s", 4, 5_000, r"path\to\file"); + upsert_chunks(&cfg, std::slice::from_ref(&backslash)).unwrap(); + let query = ListChunksQuery { + content_contains: Some(r"\to".to_string()), + ..Default::default() + }; + assert_eq!( + ids_of(&list_chunks(&cfg, &query).unwrap()), + vec![backslash.id.clone()] + ); +} + +// ── The detail view and the source rollup ─────────────────────────────────── + +#[test] +fn chunk_details_report_stored_facts_without_reading_the_vault() { + let (_tmp, cfg) = test_config(); + let staged = chunk_with(SourceKind::Chat, "slack:#eng", 0, 2_000, "staged body"); + let inline = chunk_with(SourceKind::Chat, "slack:#eng", 1, 1_000, "inline body"); + upsert_chunks(&cfg, &[staged.clone(), inline.clone()]).unwrap(); + + with_connection(&cfg, |conn| { + // A path that does not exist on disk: the detail row must still carry + // it, which is what proves nothing here opens the vault file. + conn.execute( + "UPDATE mem_tree_chunks + SET content_path = 'chat/never-written.md', lifecycle_status = 'sealed' + WHERE id = ?1", + params![staged.id], + )?; + // An empty string is neither NULL nor a path, and reads back as neither. + conn.execute( + "UPDATE mem_tree_chunks SET content_path = '' WHERE id = ?1", + params![inline.id], + )?; + conn.execute( + "INSERT INTO mem_tree_chunk_embeddings ( + chunk_id, model_signature, vector, dim, created_at + ) VALUES (?1, 'test/model@3', ?2, 3, 1700000000.0)", + params![staged.id, vec![1_u8, 2, 3]], + )?; + Ok(()) + }) + .unwrap(); + + let details = list_chunk_details(&cfg, &ListChunksQuery::default()).unwrap(); + assert_eq!(details.len(), 2); + + let staged_row = &details[0]; + assert_eq!(staged_row.chunk.id, staged.id); + assert_eq!( + staged_row.content_path.as_deref(), + Some("chat/never-written.md") + ); + assert_eq!(staged_row.lifecycle_status.as_deref(), Some("sealed")); + assert!(staged_row.has_embedding); + + let inline_row = &details[1]; + assert_eq!(inline_row.chunk.id, inline.id); + assert_eq!(inline_row.content_path, None); + assert_eq!(inline_row.lifecycle_status.as_deref(), Some("admitted")); + assert!(!inline_row.has_embedding); +} + +#[test] +fn source_totals_group_by_source_and_honour_the_scope() { + let (_tmp, cfg) = test_config(); + let mut chunks = vec![ + chunk_with(SourceKind::Chat, "slack:#eng", 0, 1_000, "a"), + chunk_with(SourceKind::Chat, "slack:#eng", 1, 9_000, "b"), + chunk_with(SourceKind::Email, "gmail:t-1", 0, 5_000, "c"), + ]; + for chunk in &mut chunks { + chunk.metadata.tags = vec!["memory_sources".to_string()]; + } + upsert_chunks(&cfg, &chunks).unwrap(); + + let totals = source_totals(&cfg, None, None).unwrap(); + assert_eq!(totals.len(), 2); + assert_eq!(totals[0].source_kind, SourceKind::Chat); + assert_eq!(totals[0].source_id, "slack:#eng"); + assert_eq!(totals[0].chunk_count, 2); + assert_eq!(totals[0].last_timestamp_ms, 9_000); + assert_eq!(totals[1].source_kind, SourceKind::Email); + assert_eq!(totals[1].chunk_count, 1); + assert_eq!(totals[1].last_timestamp_ms, 5_000); + + let scoped = + source_totals(&cfg, None, Some(&HashSet::from(["gmail:t-1".to_string()]))).unwrap(); + assert_eq!(scoped.len(), 1); + assert_eq!(scoped[0].source_id, "gmail:t-1"); + + // `limit` bounds groups, not chunks. + assert_eq!(source_totals(&cfg, Some(1), None).unwrap().len(), 1); +} + +/// `exclude_dropped` filters with `lifecycle_status != 'dropped'`, and under +/// SQLite a NULL in that column would compare NULL — the row would vanish from +/// the page *and* from the count, silently, for a chunk that was never dropped. +/// +/// It cannot happen, and this pins why rather than trusting it: the column is +/// declared `TEXT NOT NULL DEFAULT 'admitted'` as an additive `ALTER TABLE`, so +/// an insert that names every other column still lands with a value. The test +/// writes a row through raw SQL that deliberately omits `lifecycle_status` — +/// the "writer that bypassed it" case — and asserts the row is stored +/// `admitted` and survives the filter. +#[test] +fn exclude_dropped_keeps_a_row_whose_lifecycle_was_never_written() { + let (_tmp, cfg) = test_config(); + let seed = chunk_with(SourceKind::Chat, "slack:#eng", 0, 1_000, "seed"); + upsert_chunks(&cfg, std::slice::from_ref(&seed)).expect("seed"); + + let bypassed = chunk_id(SourceKind::Chat, "slack:#eng", 1, "bypassed"); + with_connection(&cfg, |conn| { + // The stored encoding of `source_kind` is read back off the seeded row + // rather than spelled out, so this test cannot drift if that encoding + // ever changes. + let kind: String = conn.query_row( + "SELECT source_kind FROM mem_tree_chunks WHERE id = ?1", + params![seed.id], + |row| row.get(0), + )?; + conn.execute( + "INSERT INTO mem_tree_chunks ( + id, source_kind, source_id, owner, + timestamp_ms, time_range_start_ms, time_range_end_ms, + tags_json, content, token_count, seq_in_source, created_at_ms + ) VALUES (?1, ?2, 'slack:#eng', 'alice@example.com', 2000, 2000, 2000, + '[]', 'bypassed', 1, 1, 2000)", + params![bypassed, kind], + )?; + Ok(()) + }) + .expect("insert without a lifecycle_status"); + + let stored: Option = with_connection(&cfg, |conn| { + Ok(conn.query_row( + "SELECT lifecycle_status FROM mem_tree_chunks WHERE id = ?1", + params![bypassed], + |row| row.get(0), + )?) + }) + .expect("read the stored lifecycle"); + assert_eq!( + stored.as_deref(), + Some("admitted"), + "the NOT NULL default is what makes the filter safe" + ); + + let query = ListChunksQuery { + exclude_dropped: true, + ..Default::default() + }; + let ids: HashSet = list_chunks(&cfg, &query) + .expect("list") + .into_iter() + .map(|chunk| chunk.id) + .collect(); + assert!( + ids.contains(&bypassed), + "a row that was never dropped must survive exclude_dropped" + ); + assert_eq!( + count_chunks_matching(&cfg, &query).expect("count"), + ids.len() as u64, + "the count must agree with the page it labels" + ); +} diff --git a/src/memory/persona/pipeline.rs b/src/memory/persona/pipeline.rs index 076b645..63fd0bb 100644 --- a/src/memory/persona/pipeline.rs +++ b/src/memory/persona/pipeline.rs @@ -254,9 +254,12 @@ impl Pipeline<'_> { let asks = self.persona.asks(); let mut bodies = BTreeMap::new(); - let mut state = ReduceState::default(); - // Reconstruct verbatim directives from the persisted store. - state.directives = super::compile::read_directives(self.config); + // Directives are reconstructed verbatim from the persisted store; the + // rest of the reduction starts empty. + let mut state = ReduceState { + directives: super::compile::read_directives(self.config), + ..Default::default() + }; for facet in PersonaFacet::ALL { let factory = TreeFactory::flavoured(facet.tree_scope(), asks.ask(facet)); let tree = factory.get_or_create(self.config)?; diff --git a/src/memory/persona/readers/instruction_tests.rs b/src/memory/persona/readers/instruction_tests.rs index 8da68ae..52ef02e 100644 --- a/src/memory/persona/readers/instruction_tests.rs +++ b/src/memory/persona/readers/instruction_tests.rs @@ -82,7 +82,7 @@ fn discover_matches_names_and_repo_scope() { let global = dir.path().join("global-CLAUDE.md"); std::fs::write(&global, "- global rule").unwrap(); - let found = discover(&[dir.path().to_path_buf()], &[global.clone()]); + let found = discover(&[dir.path().to_path_buf()], std::slice::from_ref(&global)); let names: Vec = found .iter() .map(|f| f.path.file_name().unwrap().to_string_lossy().to_string()) diff --git a/src/memory/persona/retrieve.rs b/src/memory/persona/retrieve.rs index 4e16053..9dee918 100644 --- a/src/memory/persona/retrieve.rs +++ b/src/memory/persona/retrieve.rs @@ -166,7 +166,7 @@ impl PersonaRetriever { let mut scored: Vec<(f32, &ObsDoc)> = self .docs .iter() - .filter(|d| facet.map_or(true, |f| d.facet == f)) + .filter(|d| facet.is_none_or(|f| d.facet == f)) .filter_map(|doc| { let bm25 = self.bm25(&q_terms, doc, n); if bm25 <= 0.0 { diff --git a/src/memory/score/embed.rs b/src/memory/score/embed.rs index 02a9e60..b48d449 100644 --- a/src/memory/score/embed.rs +++ b/src/memory/score/embed.rs @@ -119,8 +119,10 @@ pub fn unpack_embedding(b: &[u8]) -> Result> { ); } let floats: Vec = b - .chunks_exact(4) - .map(|c| f32::from_le_bytes([c[0], c[1], c[2], c[3]])) + .as_chunks::<4>() + .0 + .iter() + .map(|c| f32::from_le_bytes(*c)) .collect(); if floats.len() != EMBEDDING_DIM { anyhow::bail!( diff --git a/src/memory/store/vectors/store.rs b/src/memory/store/vectors/store.rs index 8500461..bb69208 100644 --- a/src/memory/store/vectors/store.rs +++ b/src/memory/store/vectors/store.rs @@ -434,11 +434,10 @@ pub fn bytes_to_vec(bytes: &[u8]) -> anyhow::Result> { bytes.len() ); Ok(bytes - .chunks_exact(4) - .map(|chunk| { - let arr: [u8; 4] = chunk.try_into().unwrap_or([0; 4]); - f32::from_le_bytes(arr) - }) + .as_chunks::<4>() + .0 + .iter() + .map(|chunk| f32::from_le_bytes(*chunk)) .collect()) } diff --git a/src/memory/tree/store/common.rs b/src/memory/tree/store/common.rs index cabdde6..53f01e4 100644 --- a/src/memory/tree/store/common.rs +++ b/src/memory/tree/store/common.rs @@ -38,8 +38,10 @@ pub(crate) fn decode_signature_blob( anyhow::bail!("{label} blob length {} not a multiple of 4", bytes.len()); } let floats: Vec = bytes - .chunks_exact(4) - .map(|c| f32::from_le_bytes([c[0], c[1], c[2], c[3]])) + .as_chunks::<4>() + .0 + .iter() + .map(|c| f32::from_le_bytes(*c)) .collect(); if floats.len() != dim as usize { anyhow::bail!(