diff --git a/src/openhuman/memory_conversations/README.md b/src/openhuman/memory_conversations/README.md deleted file mode 100644 index d98ab0306..000000000 --- a/src/openhuman/memory_conversations/README.md +++ /dev/null @@ -1,34 +0,0 @@ -# conversations - -Workspace-backed conversation thread/message storage. Lives at -`/memory/conversations/` as plain JSONL — easy to inspect, -recover, and back up. Used by the desktop UI for chat threads and by -non-web channel adapters (Slack, Telegram, …) so all surfaces share one -persistence path. - -## Files - -- **`mod.rs`** — re-exports the public surface - (`ConversationStore`, `ConversationThread`, `ConversationMessage`, - `CreateConversationThread`, `ConversationMessagePatch`, - `ConversationPurgeStats`, free-function shims, and - `register_conversation_persistence_subscriber`). -- **`types.rs`** — wire/storage structs: thread metadata, message - records, create requests, partial-update patches. -- **`store.rs`** — `ConversationStore` plus free-function shims. - Thread metadata is appended to `threads.jsonl` (upsert/delete log); - messages live in `threads/.jsonl`. A process-wide mutex - serialises every on-disk mutation. -- **`bus.rs`** — `EventHandler` that mirrors inbound `DomainEvent` - channel messages into the store, so non-web providers persist - alongside UI-driven threads. -- **`store_tests.rs`** — unit tests covering upsert, append, label/ - title updates, deletion, and purge. - -## Where it fits - -Sits next to the unified memory store but is intentionally separate: -the conversation log is append-only chat history with no embeddings or -graph relations. Ingestion into the searchable memory tree happens via -`tree/` and the per-provider ingestion modules (e.g. `slack_ingestion/`) -— this folder only owns durable transcript storage. diff --git a/src/openhuman/memory_conversations/inverted_index.rs b/src/openhuman/memory_conversations/inverted_index.rs deleted file mode 100644 index 10986b937..000000000 --- a/src/openhuman/memory_conversations/inverted_index.rs +++ /dev/null @@ -1,670 +0,0 @@ -//! In-memory inverted index for `search_cross_thread_messages`. -//! -//! ## Architecture (v1) -//! -//! ```text -//! ┌───────────────┐ -//! query "cat" ───▶ │ Phase 1: │ posting-list intersection -//! │ ngram lookup │ on character n-grams -//! └──────┬────────┘ -//! │ candidate doc ids per term -//! ┌──────▼────────┐ -//! │ Phase 2: │ exact substring verify on -//! │ verify+score │ normalized content, score -//! └──────┬────────┘ by `matched_terms / total_terms` -//! │ -//! ▼ -//! Vec -//! ``` -//! -//! ## Ownership choices for scale -//! -//! Conversation corpora can grow to hundreds of thousands of messages per -//! workspace. The data structures here are picked to keep the resident-set -//! size predictable at that scale: -//! -//! - **`thread_id` and `role` are interned `Arc`** (see -//! `intern_thread_id` / `intern_role`). A workspace with one thread of -//! N messages would otherwise store N copies of the same thread id; a -//! role string only ever takes two distinct values in practice. The -//! interner amortises both to a single heap allocation per distinct -//! value, plus one `Arc` clone per `DocEntry`. -//! - **Posting-map keys are `Box`** (16 bytes) rather than `String` -//! (24 bytes). Saves 8 bytes per distinct ngram in the corpus — at -//! ~17k Latin trigrams plus CJK bigrams that adds up. -//! - **Posting lists are still `BTreeSet`** for ergonomic ordered -//! iteration. The Phase 1 intersection is performed against a -//! single-allocation `Vec` accumulator via a two-pointer -//! sort-merge (no per-iteration `BTreeSet` rebuilds), so the BTreeSet -//! shape only affects insertion and removal, not query latency. -//! Roaring Bitmaps + FST + LSM segments are the long-term destination -//! (Gemini Deep Research write-up); we defer that until corpus sizes -//! justify the complexity. -//! - **Whole index lives in RAM**, rebuilt from JSONL on first access in -//! the process. The JSONL files remain the source of truth. -//! - **Scoring matches the previous linear scan**: -//! `score = matched_terms / total_terms` with a `created_at` tiebreaker. -//! - **Pathological query short-circuit**: if Phase 1 produces a -//! candidate set larger than `LARGE_CANDIDATE_LIMIT` for any term, the -//! index returns recency-ordered hits without running Phase 2. This -//! genuinely caps tail latency — the check fires *before* the -//! substring-verification loop, not after. - -use std::collections::{BTreeSet, HashMap}; -use std::sync::Arc; - -use super::tokenize::{ngrams, normalize}; -use super::types::{ConversationMessage, CrossThreadHit}; - -/// Minimum byte length for a query term to be considered. Matches the -/// historical behaviour of `search_cross_thread_messages` so existing -/// callers (and tests) see no change. Single-byte ASCII tokens like "a" -/// or "is" are filtered out; a single CJK character (3 bytes in UTF-8) -/// passes through. -const MIN_TERM_BYTES: usize = 3; - -/// When Phase 1 returns more than this many candidates we skip Phase 2 -/// verification and fall back to a pure recency-ranked truncation. This -/// is the mitigation for the "user types `e`" pathological case. -const LARGE_CANDIDATE_LIMIT: usize = 10_000; - -/// One indexed message. Carries enough state to (a) reconstruct a -/// `CrossThreadHit` without re-reading JSONL on the hot path and (b) -/// verify Phase 1 candidates by exact substring match on the normalized -/// form. -/// -/// `thread_id` and `role` are `Arc` because they repeat heavily -/// across messages (N messages per thread → N references to the same -/// thread id; only ~2 distinct role values across the entire corpus). -/// `message_id`, `content`, `content_normalized` and `created_at` are -/// per-message unique so they stay as `String`. -#[derive(Debug, Clone)] -struct DocEntry { - thread_id: Arc, - message_id: String, - role: Arc, - content: String, // original, returned verbatim in hits - content_normalized: String, // for Phase 2 substring verification - created_at: String, -} - -/// In-memory trigram/bigram inverted index over conversation messages. -/// -/// Documents are addressed by a dense `u32` doc-id assigned in insertion -/// order. Deletes leave tombstones (`docs[i] = None`) rather than shifting -/// the array, so posting-list integers stay valid without rebuilding. -#[derive(Debug, Default)] -pub(crate) struct InvertedIndex { - /// `ngram -> sorted set of doc-ids`. BTreeSet so per-doc removals - /// are O(log n) and iteration is in sorted order (drives the - /// sort-merge intersect in `candidates_for_term`). Keys are - /// `Box` to shave 8 bytes per entry vs `String`. - postings: HashMap, BTreeSet>, - /// Tombstoned: `docs[i] == None` means the message was deleted. We - /// keep the slot so existing doc-ids in posting lists stay valid. - docs: Vec>, - /// Reverse lookup for incremental removal: `(thread_id, message_id)` - /// → `doc_id`. Letting us drop a single message without re-walking - /// the corpus. - by_message: HashMap<(String, String), u32>, - /// Interner pools. Keep a single `Arc` per distinct thread id - /// and role so every `DocEntry` referencing them can hold a cheap - /// 16-byte `Arc` clone instead of a 24-byte `String`. - thread_id_pool: HashMap>, - role_pool: HashMap>, -} - -impl InvertedIndex { - pub fn new() -> Self { - Self::default() - } - - /// Insert one message. Takes the message by value so the caller's - /// owned strings can be moved into the index without an internal - /// clone of each field. If the (thread, message_id) pair is already - /// in the index this is a no-op (messages are append-only in the - /// store, so duplicate IDs indicate a corrupt JSONL — silently - /// ignore rather than panic). - pub fn insert(&mut self, thread_id: &str, msg: ConversationMessage) { - let ConversationMessage { - id, - content, - sender, - created_at, - message_type: _, - extra_metadata: _, - } = msg; - - let key = (thread_id.to_string(), id.clone()); - if self.by_message.contains_key(&key) { - return; - } - let normalized = normalize(&content); - let doc_id = self.docs.len() as u32; - for ngram in ngrams(&normalized) { - if let Some(posting) = self.postings.get_mut(ngram) { - posting.insert(doc_id); - } else { - let mut set = BTreeSet::new(); - set.insert(doc_id); - self.postings.insert(ngram.into(), set); - } - } - let thread_arc = self.intern_thread_id(thread_id); - let role_arc = self.intern_role(&sender); - self.docs.push(Some(DocEntry { - thread_id: thread_arc, - message_id: id, - role: role_arc, - content, - content_normalized: normalized, - created_at, - })); - self.by_message.insert(key, doc_id); - } - - /// Drop every document belonging to a thread. Used by - /// `delete_thread` and during full purge. - pub fn remove_thread(&mut self, thread_id: &str) { - let to_remove: Vec = self - .by_message - .iter() - .filter(|((t, _), _)| t == thread_id) - .map(|(_, id)| *id) - .collect(); - for doc_id in to_remove { - self.remove_doc(doc_id); - } - self.thread_id_pool.remove(thread_id); - } - - /// Reset the index to its empty state. Cheaper than dropping and - /// re-allocating when a workspace is being rebuilt. - pub fn clear(&mut self) { - self.postings.clear(); - self.docs.clear(); - self.by_message.clear(); - self.thread_id_pool.clear(); - self.role_pool.clear(); - } - - fn remove_doc(&mut self, doc_id: u32) { - let idx = doc_id as usize; - let Some(entry) = self.docs.get_mut(idx).and_then(|slot| slot.take()) else { - return; - }; - self.by_message - .remove(&(entry.thread_id.to_string(), entry.message_id.clone())); - // Remove doc_id from every posting list referencing it. We re- - // tokenize the normalized content rather than tracking the - // per-doc ngram set; tokenization is allocation-free now that - // `ngrams` returns borrowed slices, so this stays cheap. - for ngram in ngrams(&entry.content_normalized) { - if let Some(posting) = self.postings.get_mut(ngram) { - posting.remove(&doc_id); - if posting.is_empty() { - self.postings.remove(ngram); - } - } - } - } - - /// The Phase 1 + Phase 2 query pipeline. Mirrors the contract of - /// `ConversationStore::search_cross_thread_messages` so the store - /// method can be a thin shim. - pub fn search( - &self, - query: &str, - limit: usize, - exclude_thread_id: Option<&str>, - ) -> Vec { - if limit == 0 { - return Vec::new(); - } - let query_lower = normalize(query); - // Filter terms by raw byte length (matches the historical - // 3-byte threshold; single CJK chars are 3 bytes and pass). - let terms: Vec = query_lower - .split_whitespace() - .filter(|t| t.len() >= MIN_TERM_BYTES) - .map(|s| s.to_string()) - .collect(); - if terms.is_empty() { - return Vec::new(); - } - - // Phase 1: collect candidate doc-ids per term. Short-circuit to - // recency-only ordering if any single term's candidate set - // already exceeds the pathological threshold — this is the cap - // on tail latency, and it must fire BEFORE we run the substring - // verification loop. - let mut per_term: Vec> = Vec::with_capacity(terms.len()); - for term in &terms { - let candidates = match self.candidates_for_term(term) { - Some(v) => v, - None => self - .docs - .iter() - .enumerate() - .filter_map(|(i, slot)| slot.as_ref().map(|_| i as u32)) - .collect::>(), - }; - if candidates.len() > LARGE_CANDIDATE_LIMIT { - return self.recency_fallback(exclude_thread_id, limit); - } - per_term.push(candidates); - } - - // Phase 2: verify each candidate by exact substring match. - // Count distinct terms per doc for the score. - let mut hit_counts: HashMap = HashMap::new(); - for (term, candidates) in terms.iter().zip(per_term) { - for doc_id in candidates { - let Some(entry) = self.docs[doc_id as usize].as_ref() else { - continue; - }; - if exclude_thread_id == Some(entry.thread_id.as_ref()) { - continue; - } - if entry.content_normalized.contains(term.as_str()) { - *hit_counts.entry(doc_id).or_insert(0) += 1; - } - } - } - - let total_terms = terms.len() as f64; - // Rank on cheap keys first — the match count and a borrowed `created_at` - // — then materialize the heavy CrossThreadHit (which clones the KB-sized - // `content`) only for the `limit` survivors. Phase 2 can leave thousands - // of candidates in `hit_counts` while callers ask for 3-10 results, so - // cloning every candidate's content before truncating is ~99% wasted. - // Ranking by `matched` (usize) is order-equivalent to ranking by - // `score = matched / total_terms` since `total_terms` is a positive - // constant, so the returned order is unchanged. - let mut ranked: Vec<(u32, usize, &str)> = hit_counts - .into_iter() - .map(|(doc_id, matched)| { - let entry = self.docs[doc_id as usize] - .as_ref() - .expect("doc_id from hit_counts must be live"); - (doc_id, matched, entry.created_at.as_str()) - }) - .collect(); - ranked.sort_by(|a, b| b.1.cmp(&a.1).then_with(|| b.2.cmp(a.2))); - ranked.truncate(limit); - - ranked - .into_iter() - .map(|(doc_id, matched, _)| { - let entry = self.docs[doc_id as usize] - .as_ref() - .expect("doc_id from hit_counts must be live"); - CrossThreadHit { - thread_id: entry.thread_id.to_string(), - message_id: entry.message_id.clone(), - role: entry.role.to_string(), - content: entry.content.clone(), - created_at: entry.created_at.clone(), - score: matched as f64 / total_terms, - } - }) - .collect() - } - - /// Build the Phase 1 candidate set for one query term. - /// - /// Returns `Some(vec)` containing the sorted intersection of posting - /// lists for every ngram of `term`. If `term` is too short to - /// produce any ngram (e.g. a single CJK char of length 1) returns - /// `None` so the caller can fall back to a linear scan. - /// - /// The intersect is a two-pointer sort-merge over the already-sorted - /// posting lists: `acc` is rewritten in place once per remaining - /// ngram, allocating zero intermediate sets. - fn candidates_for_term(&self, term: &str) -> Option> { - let term_ngrams = ngrams(term); - if term_ngrams.is_empty() { - return None; - } - let mut iter = term_ngrams.iter(); - let first = iter.next().expect("non-empty by check above"); - let mut acc: Vec = match self.postings.get(*first) { - Some(p) => p.iter().copied().collect(), - None => return Some(Vec::new()), - }; - for ng in iter { - if acc.is_empty() { - return Some(acc); - } - match self.postings.get(*ng) { - Some(p) => intersect_sorted_with_btreeset(&mut acc, p), - None => return Some(Vec::new()), - } - } - Some(acc) - } - - fn intern_thread_id(&mut self, thread_id: &str) -> Arc { - if let Some(existing) = self.thread_id_pool.get(thread_id) { - return Arc::clone(existing); - } - let arc: Arc = Arc::from(thread_id); - self.thread_id_pool - .insert(thread_id.to_string(), Arc::clone(&arc)); - arc - } - - fn intern_role(&mut self, role: &str) -> Arc { - if let Some(existing) = self.role_pool.get(role) { - return Arc::clone(existing); - } - let arc: Arc = Arc::from(role); - self.role_pool.insert(role.to_string(), Arc::clone(&arc)); - arc - } - - fn recency_fallback( - &self, - exclude_thread_id: Option<&str>, - limit: usize, - ) -> Vec { - // Rank the whole corpus by recency on a borrowed `created_at`, then - // clone the heavy fields only for the `limit` newest survivors — this - // fallback fires when a term matches >10k docs, so cloning every doc's - // content before truncating could allocate hundreds of MB needlessly. - let mut ranked: Vec<(usize, &str)> = self - .docs - .iter() - .enumerate() - .filter_map(|(i, slot)| slot.as_ref().map(|entry| (i, entry))) - .filter(|(_, entry)| exclude_thread_id != Some(entry.thread_id.as_ref())) - .map(|(i, entry)| (i, entry.created_at.as_str())) - .collect(); - ranked.sort_by(|a, b| b.1.cmp(a.1)); - ranked.truncate(limit); - - ranked - .into_iter() - .map(|(i, _)| { - let entry = self.docs[i] - .as_ref() - .expect("index from a live doc slot must stay live"); - CrossThreadHit { - thread_id: entry.thread_id.to_string(), - message_id: entry.message_id.clone(), - role: entry.role.to_string(), - content: entry.content.clone(), - created_at: entry.created_at.clone(), - // Score 0.0 signals "matched via recency fallback only" — - // documented in the function rustdoc above. Callers - // sorting by `(score desc, created_at desc)` still see - // the newest entries first. - score: 0.0, - } - }) - .collect() - } -} - -/// Two-pointer sort-merge intersect. `acc` and `other` are both sorted -/// ascending; on return `acc` contains only the elements present in -/// both, in the same sorted order. Runs in O(|acc| + |other|) with zero -/// allocations. -fn intersect_sorted_with_btreeset(acc: &mut Vec, other: &BTreeSet) { - let mut other_iter = other.iter().copied().peekable(); - let mut write = 0usize; - for read in 0..acc.len() { - let target = acc[read]; - // Advance `other_iter` past everything strictly less than the - // current `target`. After this loop the next peeked value is - // either equal to `target` (keep) or strictly greater (drop). - while let Some(&o) = other_iter.peek() { - if o < target { - other_iter.next(); - } else { - break; - } - } - if other_iter.peek().copied() == Some(target) { - acc[write] = target; - write += 1; - other_iter.next(); - } - } - acc.truncate(write); -} - -#[cfg(test)] -mod tests { - use serde_json::json; - - use super::*; - - fn msg(id: &str, content: &str, created: &str) -> ConversationMessage { - ConversationMessage { - id: id.to_string(), - content: content.to_string(), - message_type: "text".to_string(), - extra_metadata: json!({}), - sender: "user".to_string(), - created_at: created.to_string(), - } - } - - #[test] - fn substring_inside_word_matches() { - // Canonical substring-inside-word case: querying "cat" must find - // "concatenate" — a token-boundary tokenizer would miss this. - let mut idx = InvertedIndex::new(); - idx.insert( - "t1", - msg("m1", "concatenate the strings", "2026-04-10T10:00:00Z"), - ); - - let hits = idx.search("cat", 10, None); - assert_eq!(hits.len(), 1); - assert_eq!(hits[0].message_id, "m1"); - } - - #[test] - fn polish_diacritics_normalized_both_sides() { - let mut idx = InvertedIndex::new(); - idx.insert( - "t1", - msg("m1", "Lecę dziś do Krakowa", "2026-04-10T10:00:00Z"), - ); - // Query with no diacritics finds content with diacritics. - let hits = idx.search("krakow", 10, None); - assert_eq!(hits.len(), 1, "krakow should match Krakowa"); - } - - #[test] - fn japanese_bigram_match() { - let mut idx = InvertedIndex::new(); - idx.insert( - "t1", - msg("m1", "東京タワーが見える", "2026-04-10T10:00:00Z"), - ); - let hits = idx.search("東京", 10, None); - assert_eq!(hits.len(), 1, "two-char CJK query should match"); - } - - #[test] - fn arabic_harakat_stripped() { - let mut idx = InvertedIndex::new(); - // "wrote" with full vocalization vs bare consonants. - idx.insert("t1", msg("m1", "كَتَبَ الطالب", "2026-04-10T10:00:00Z")); - // The bare-consonant form should still find the vocalized one. - let hits = idx.search("كتب", 10, None); - assert_eq!(hits.len(), 1, "harakat stripping should equalize forms"); - } - - #[test] - fn excludes_active_thread() { - let mut idx = InvertedIndex::new(); - idx.insert( - "active", - msg("ma", "postgres deploy", "2026-04-10T10:00:00Z"), - ); - idx.insert( - "other", - msg("mo", "postgres deploy", "2026-04-10T10:01:00Z"), - ); - let hits = idx.search("postgres", 10, Some("active")); - assert_eq!(hits.len(), 1); - assert_eq!(hits[0].thread_id, "other"); - } - - #[test] - fn empty_query_returns_empty() { - let mut idx = InvertedIndex::new(); - idx.insert("t", msg("m", "anything here", "2026-04-10T10:00:00Z")); - assert!(idx.search("", 10, None).is_empty()); - } - - #[test] - fn short_terms_only_returns_empty() { - // Mirrors the legacy `search_cross_thread_messages_skips_short_terms` - // behaviour — terms < 3 bytes are dropped. - let mut idx = InvertedIndex::new(); - idx.insert("t", msg("m", "Postgres", "2026-04-10T10:00:00Z")); - assert!(idx.search("a is on", 10, None).is_empty()); - } - - #[test] - fn limit_zero_returns_empty() { - let mut idx = InvertedIndex::new(); - idx.insert("t", msg("m", "Postgres", "2026-04-10T10:00:00Z")); - assert!(idx.search("postgres", 0, None).is_empty()); - } - - #[test] - fn score_matches_legacy_semantics() { - // 5-term query, 2 substring matches → score = 0.4. - let mut idx = InvertedIndex::new(); - idx.insert( - "t1", - msg( - "m1", - "Remember: my project is called Phoenix and uses Go and PostgreSQL.", - "2026-04-10T10:00:00Z", - ), - ); - let hits = idx.search("What database does my project use", 10, None); - assert_eq!(hits.len(), 1); - // "project" + "use" (substring of "uses") → 2 of 5 terms. - assert!( - (hits[0].score - 0.4).abs() < 1e-9, - "score = {}", - hits[0].score - ); - } - - #[test] - fn ranks_by_score_then_recency_before_truncating() { - // More matches than `limit`, so truncation must keep the top-ranked - // hits by (score desc, created_at desc) — this pins that ranking still - // happens before the result set is cut, not after. - let mut idx = InvertedIndex::new(); - // Both terms → score 1.0, but oldest. - idx.insert( - "t1", - msg("both", "alpha beta gamma", "2026-04-10T10:00:00Z"), - ); - // One term → score 0.5, newest of the 0.5 group. - idx.insert("t1", msg("newest", "alpha delta", "2026-04-10T10:03:00Z")); - // One term → score 0.5, middle. - idx.insert("t1", msg("middle", "alpha epsilon", "2026-04-10T10:02:00Z")); - // One term → score 0.5, oldest of the 0.5 group. - idx.insert("t1", msg("oldest", "beta zeta", "2026-04-10T10:01:00Z")); - - let hits = idx.search("alpha beta", 2, None); - assert_eq!(hits.len(), 2, "must respect the limit"); - // Highest score wins outright; the recency tiebreak then picks the - // newest of the equal-score remainder. "middle"/"oldest" are dropped. - assert_eq!(hits[0].message_id, "both"); - assert!( - (hits[0].score - 1.0).abs() < 1e-9, - "score = {}", - hits[0].score - ); - assert_eq!(hits[1].message_id, "newest"); - assert!( - (hits[1].score - 0.5).abs() < 1e-9, - "score = {}", - hits[1].score - ); - } - - #[test] - fn remove_thread_drops_all_messages() { - let mut idx = InvertedIndex::new(); - idx.insert("t1", msg("m1", "postgres deploy", "2026-04-10T10:00:00Z")); - idx.insert("t1", msg("m2", "postgres backup", "2026-04-10T10:01:00Z")); - idx.insert("t2", msg("m3", "postgres replica", "2026-04-10T10:02:00Z")); - assert_eq!(idx.search("postgres", 10, None).len(), 3); - idx.remove_thread("t1"); - let hits = idx.search("postgres", 10, None); - assert_eq!(hits.len(), 1); - assert_eq!(hits[0].thread_id, "t2"); - } - - #[test] - fn duplicate_message_id_is_idempotent() { - let mut idx = InvertedIndex::new(); - let m = msg("m1", "postgres deploy", "2026-04-10T10:00:00Z"); - idx.insert("t1", m.clone()); - idx.insert("t1", m); // dup — should be ignored - let hits = idx.search("postgres", 10, None); - assert_eq!(hits.len(), 1, "duplicate insert must not duplicate hits"); - } - - #[test] - fn pathological_query_short_circuits_to_recency() { - // Build a corpus big enough to trip LARGE_CANDIDATE_LIMIT. - let mut idx = InvertedIndex::new(); - let n = LARGE_CANDIDATE_LIMIT + 50; - for i in 0..n { - // "common" appears in every doc, so the trigram "com" hits - // all of them. - let created = format!("2026-04-10T10:{:02}:{:02}Z", i / 60, i % 60); - idx.insert("bulk", msg(&format!("m{i}"), "common payload", &created)); - } - let hits = idx.search("common", 5, None); - assert_eq!(hits.len(), 5, "fallback must still respect limit"); - // Score 0.0 is the recency-fallback marker. - assert!(hits.iter().all(|h| h.score == 0.0)); - } - - #[test] - fn intern_pool_dedupes_repeated_thread_id_and_role() { - // Smoke-test the Arc interning: many messages on one - // thread must share a single Arc backing the thread_id. - let mut idx = InvertedIndex::new(); - for i in 0..5 { - let ts = format!("2026-04-10T10:00:{:02}Z", i); - idx.insert("shared-thread", msg(&format!("m{i}"), "payload", &ts)); - } - assert_eq!(idx.thread_id_pool.len(), 1); - assert_eq!(idx.role_pool.len(), 1); - // After removing the thread, the pool entry is dropped too. - idx.remove_thread("shared-thread"); - assert_eq!(idx.thread_id_pool.len(), 0); - } - - #[test] - fn intersect_sorted_with_btreeset_basic() { - let other: BTreeSet = [2, 4, 5, 9].into_iter().collect(); - let mut acc: Vec = vec![1, 2, 3, 5, 7, 9]; - intersect_sorted_with_btreeset(&mut acc, &other); - assert_eq!(acc, vec![2, 5, 9]); - } - - #[test] - fn intersect_sorted_with_btreeset_empty_other() { - let other: BTreeSet = BTreeSet::new(); - let mut acc: Vec = vec![1, 2, 3]; - intersect_sorted_with_btreeset(&mut acc, &other); - assert!(acc.is_empty()); - } -} diff --git a/src/openhuman/memory_conversations/mod.rs b/src/openhuman/memory_conversations/mod.rs index 43a551008..abcb8afdb 100644 --- a/src/openhuman/memory_conversations/mod.rs +++ b/src/openhuman/memory_conversations/mod.rs @@ -1,27 +1,25 @@ -//! Workspace-backed conversation thread/message storage for the desktop UI. +//! Workspace-backed conversation thread/message storage for the desktop UI — +//! thin host shim over `tinycortex::memory::conversations` (W7). //! -//! Conversations are stored as JSONL files under `/memory/conversations/`. -//! Thread metadata is append-only in `threads.jsonl`; each thread's messages live -//! in a dedicated JSONL file for straightforward inspection and recovery. +//! Conversations are stored as JSONL files under the workspace (thread metadata +//! append-only in `threads.jsonl`; each thread's messages in a dedicated JSONL +//! file). The store / inverted-index / tokenizer / types engine is the crate's +//! (a byte-identical port, incl. the D1 rank-before-materialize fix); this +//! module re-exports that surface so the ~30 host consumers +//! (`openhuman::memory` re-exports it as `memory::conversations`, plus jsonrpc, +//! agent orchestration, agent_memory, threads, channels) keep their import paths +//! and identical `Result<_, String>` / on-disk behaviour unchanged. //! -//! This module was split out of `openhuman::memory` into the top-level -//! `openhuman::memory_conversations` namespace so the high-level memory policy -//! layer does not also own UI thread persistence. `openhuman::memory` re-exports -//! this module as `memory::conversations` during the migration. +//! Host-retained: [`bus`] — the `core::event_bus` persistence subscriber that +//! bridges typed channel events onto the crate store (the crate abstracts the +//! bus behind its own `ConversationEventBus` trait; the host wires the real one). mod bus; -mod inverted_index; -mod store; -mod tokenize; -mod types; pub use bus::register_conversation_persistence_subscriber; -pub use store::{ +pub use tinycortex::memory::conversations::{ append_message, delete_thread, ensure_thread, get_messages, list_threads, purge_threads, - update_message, update_thread_labels, update_thread_title, ConversationPurgeStats, - ConversationStore, -}; -pub use types::{ - ConversationMessage, ConversationMessagePatch, ConversationThread, CreateConversationThread, - CrossThreadHit, + update_message, update_thread_labels, update_thread_title, ConversationMessage, + ConversationMessagePatch, ConversationPurgeStats, ConversationStore, ConversationThread, + CreateConversationThread, CrossThreadHit, }; diff --git a/src/openhuman/memory_conversations/store.rs b/src/openhuman/memory_conversations/store.rs deleted file mode 100644 index 1dbc4a527..000000000 --- a/src/openhuman/memory_conversations/store.rs +++ /dev/null @@ -1,1064 +0,0 @@ -//! JSONL-backed thread and message store. Thread metadata lives in -//! `threads.jsonl` (append-only upsert/delete log); each thread's messages -//! are appended to a per-thread JSONL file under `threads/.jsonl`. -//! -//! All on-disk mutations serialise through a single process-wide mutex so -//! concurrent RPC handlers don't interleave writes. - -use std::collections::{BTreeMap, HashMap}; -use std::fs::{self, File, OpenOptions}; -use std::hash::{Hash, Hasher}; -use std::io::{BufRead, BufReader, Write}; -use std::path::{Path, PathBuf}; - -use log::{debug, warn}; -use once_cell::sync::Lazy; -use parking_lot::Mutex; -use tempfile::NamedTempFile; - -use super::inverted_index::InvertedIndex; -use super::types::{ - ConversationMessage, ConversationMessagePatch, ConversationThread, CreateConversationThread, - CrossThreadHit, -}; - -const LOG_PREFIX: &str = "[memory:conversations]"; -const THREADS_FILENAME: &str = "threads.jsonl"; -const THREAD_MESSAGES_DIR: &str = "threads"; -static CONVERSATION_STORE_LOCK: Lazy> = Lazy::new(|| Mutex::new(())); - -/// Per-workspace inverted index cache. Keyed by the workspace's -/// `memory/conversations` root so multiple `ConversationStore` clones -/// pointing at the same workspace share one index. The cache outlives -/// individual store handles (which are cloneable PathBuf wrappers); it -/// is bounded by the number of distinct workspaces a single process -/// touches, which in practice is one. Tests using `TempDir` paths leave -/// behind dead entries when the dir is removed — acceptable for an -/// in-process cache. -/// -/// # Lock ordering -/// -/// When BOTH `CONVERSATION_STORE_LOCK` and `CONVERSATION_INDEX_CACHE` -/// must be held simultaneously, `CONVERSATION_STORE_LOCK` MUST be -/// acquired first. This applies to `append_message` (writes JSONL then -/// updates the warm index) and `with_index` (caller holds the outer -/// lock, then takes the cache lock to run the search closure). -/// -/// `prime_index_if_cold` minimises shared locking. It may hold both -/// locks only momentarily, and always in the `CONVERSATION_STORE_LOCK` -/// → `CONVERSATION_INDEX_CACHE` order above: while holding the outer -/// lock to snapshot live thread IDs via `thread_index_unlocked` -/// (header-only, no per-thread I/O) it re-checks the cache once. It then -/// releases `CONVERSATION_STORE_LOCK` before reading per-thread JSONL -/// content (no lock held) and finally acquires `CONVERSATION_INDEX_CACHE` -/// alone to insert the built index. It never holds both across the slow -/// JSONL walk, and neither operation calls back into a function that -/// would acquire the other lock. -/// -/// `list_threads_unlocked` MUST NOT be used inside the locked snapshot — -/// it calls `measure_messages_unlocked` per legacy thread (no Stats -/// history), which reads every per-thread JSONL file and appends a -/// `Stats` entry to `threads.jsonl`, reintroducing the multi-second -/// stall under the outer lock that this design was built to avoid. -static CONVERSATION_INDEX_CACHE: Lazy>> = - Lazy::new(|| Mutex::new(HashMap::new())); - -fn redact_title_for_log(title: &str) -> String { - let mut hasher = std::collections::hash_map::DefaultHasher::new(); - title.hash(&mut hasher); - format!( - "", - title.chars().count(), - hasher.finish() - ) -} - -/// Counts returned by [`purge_threads`] — how much was deleted. -#[derive(Debug, Clone, Copy, Default)] -pub struct ConversationPurgeStats { - pub thread_count: usize, - pub message_count: usize, -} - -/// Workspace-rooted handle that reads and writes the JSONL conversation log. -#[derive(Debug, Clone)] -pub struct ConversationStore { - workspace_dir: PathBuf, -} - -#[derive(Debug, Clone, serde::Serialize, serde::Deserialize)] -#[serde(tag = "op", rename_all = "snake_case")] -enum ThreadLogEntry { - Upsert { - thread_id: String, - title: String, - created_at: String, - updated_at: String, - #[serde(default, skip_serializing_if = "Option::is_none")] - parent_thread_id: Option, - #[serde(default, skip_serializing_if = "Option::is_none")] - labels: Option>, - #[serde(default, skip_serializing_if = "Option::is_none")] - personality_id: Option, - }, - Delete { - thread_id: String, - deleted_at: String, - }, - /// Single message appended to a thread. Increments `message_count` by 1 - /// and overwrites `last_message_at`. Emitted by `append_message` to keep - /// list_threads O(threads.jsonl) instead of O(total messages). - MessageAppended { - thread_id: String, - last_message_at: String, - }, - /// Absolute stat snapshot — overrides the running count + timestamp. - /// Used to backfill legacy threads whose messages were written before - /// `MessageAppended` existed. - Stats { - thread_id: String, - message_count: usize, - last_message_at: String, - }, -} - -impl ConversationStore { - /// Construct a store rooted at the given workspace directory. - pub fn new(workspace_dir: PathBuf) -> Self { - Self { workspace_dir } - } - - /// Create or update a thread, appending an `Upsert` entry to `threads.jsonl`. - pub fn ensure_thread( - &self, - request: CreateConversationThread, - ) -> Result { - let _guard = CONVERSATION_STORE_LOCK.lock(); - let root = self.ensure_root()?; - let threads_path = root.join(THREADS_FILENAME); - let now = request.created_at.clone(); - let labels = request.labels.clone().map(normalize_labels); - append_jsonl( - &threads_path, - &ThreadLogEntry::Upsert { - thread_id: request.id.clone(), - title: request.title.clone(), - created_at: request.created_at.clone(), - updated_at: now, - parent_thread_id: request.parent_thread_id.clone(), - labels, - personality_id: request.personality_id.clone(), - }, - )?; - debug!( - "{LOG_PREFIX} ensured thread id={} path={}", - request.id, - threads_path.display() - ); - self.thread_summary_unlocked(&request.id)? - .ok_or_else(|| format!("thread {} missing after ensure", request.id)) - } - - /// List all live threads (folding the upsert/delete log). - pub fn list_threads(&self) -> Result, String> { - let _guard = CONVERSATION_STORE_LOCK.lock(); - self.list_threads_unlocked() - } - - /// Read every persisted message for a thread in append order. - pub fn get_messages(&self, thread_id: &str) -> Result, String> { - let _guard = CONVERSATION_STORE_LOCK.lock(); - if !self.thread_exists_unlocked(thread_id)? { - return Ok(Vec::new()); - } - let path = self.thread_messages_path(thread_id); - if !path.exists() { - return Ok(Vec::new()); - } - read_jsonl::(&path) - } - - /// Substring-match messages across **every** thread in the workspace, - /// optionally excluding one thread (the active chat). Returns up to - /// `limit` of the most-recent matching messages, newest first. - /// - /// Workspace scope is enforced by the store's `workspace_dir` — one - /// workspace dir per user — so this helper cannot cross that - /// boundary. Issue #1505: the conversational durable-fact pipeline - /// (`learning::transcript_ingest`) is async and batched, so cross- - /// chat continuity needs a direct cross-thread reader to surface - /// context the user shared in chat A when they ask a dependent - /// question in chat B. - /// - /// Backed by an in-memory trigram/CJK-bigram inverted index - /// (`super::inverted_index`). The legacy implementation walked every - /// JSONL file and did `content.to_lowercase().contains(term)` per - /// message, which is O(threads × messages × content_len). The index - /// turns that into O(|posting lists|) for typical queries while - /// preserving the previous scoring contract - /// (`score = matched_terms / total_terms`, recency tiebreak). - /// - /// # Lock strategy (issue #2849) - /// - /// **Fast path (warm cache):** acquires only `CONVERSATION_INDEX_CACHE` - /// — no outer store lock — and returns immediately. - /// - /// **Cold path (first access):** snapshots the thread list under - /// `CONVERSATION_STORE_LOCK` (brief), then releases it before reading - /// JSONL files to build the inverted index. This avoids blocking - /// `append_message` / `get_messages` / `list_threads` during the - /// potentially-long rebuild. JSONL files are append-only, so a - /// concurrent write during the rebuild may mean the rebuilt index - /// misses that one message. It is *not* re-read later; subsequent - /// `append_message` calls only index their own (new) messages once - /// the cache is warm. The missed message therefore stays absent - /// until the cache is evicted and rebuilt — an accepted tradeoff - /// for issue #2849. - pub fn search_cross_thread_messages( - &self, - query: &str, - limit: usize, - exclude_thread_id: Option<&str>, - ) -> Result, String> { - // Warm the index outside the outer lock so concurrent - // append_message / get_messages calls are not stalled during the - // cold JSONL rebuild (which can take seconds on large workspaces). - // After this returns the cache entry is guaranteed to exist, so - // with_index will not trigger a second rebuild. - self.prime_index_if_cold()?; - - let _guard = CONVERSATION_STORE_LOCK.lock(); - self.with_index(|idx| idx.search(query, limit, exclude_thread_id)) - } - - /// If no index entry exists for this workspace, snapshot the live thread - /// IDs under `CONVERSATION_STORE_LOCK` (fast — reads only - /// `threads.jsonl`, no per-thread I/O), release that lock, read all - /// per-thread JSONL files with no lock held (safe — append-only), then - /// insert the built index into `CONVERSATION_INDEX_CACHE` using - /// `entry().or_insert()` so a concurrent prime that finished first wins - /// and ours is discarded. - /// - /// After this call returns, `with_index` will always find a warm entry - /// and will not re-enter `populate_index_unlocked`. - fn prime_index_if_cold(&self) -> Result<(), String> { - let key = self.root_dir(); - // Fast path: already warm — one tiny lock acquisition and out. - if CONVERSATION_INDEX_CACHE.lock().contains_key(&key) { - return Ok(()); - } - // Snapshot live thread IDs while holding the outer lock. - // `thread_index_unlocked` reads only `threads.jsonl` (header-only, - // O(threads), no per-thread file I/O) — the lock is released - // immediately after, so the slow content reads below never block - // concurrent writers. - // - // Do NOT call `list_threads_unlocked` here. For workspaces where - // any thread has no `MessageAppended`/`Stats` history (common before - // the Stats log was introduced), `list_threads_unlocked` triggers - // `measure_messages_unlocked` + a `Stats` append per thread — all - // under `CONVERSATION_STORE_LOCK` — reintroducing the multi-second - // stall this function is designed to avoid. - let thread_ids: Vec = { - let _guard = CONVERSATION_STORE_LOCK.lock(); - // Re-check after acquiring: a concurrent prime may have just - // finished while we waited for the outer lock. - if CONVERSATION_INDEX_CACHE.lock().contains_key(&key) { - return Ok(()); - } - self.thread_index_unlocked()?.into_keys().collect() - }; - // Build the index with no locks held. The per-thread JSONL files are - // append-only so reads are safe without synchronisation. The worst - // case is a message appended during this window: append_message sees a - // still-cold cache (we have not inserted yet) and skips its index - // update, so that specific message stays absent from the in-memory - // index until the next cold rebuild (e.g. a process restart re-reads - // JSONL). Later appends index only their own messages, not the raced - // one. This is the accepted tradeoff documented in issue #2849. - let mut idx = InvertedIndex::new(); - for thread_id in &thread_ids { - let path = self.thread_messages_path(thread_id); - if !path.exists() { - continue; - } - match read_jsonl::(&path) { - Ok(messages) => { - for msg in messages { - idx.insert(thread_id, msg); - } - } - Err(err) => { - tracing::warn!( - "{LOG_PREFIX} index prime skipped unreadable file path={} error={}", - path.display(), - err - ); - } - } - } - // Insert only if the key is still absent — a concurrent prime that - // finished first wins; ours is discarded. Log the discard so - // duplicate-build churn under load is diagnosable. - { - let mut cache = CONVERSATION_INDEX_CACHE.lock(); - if cache.contains_key(&key) { - debug!( - "{LOG_PREFIX} discarded freshly-built index; concurrent prime won workspace={}", - key.display() - ); - } else { - cache.insert(key.clone(), idx); - debug!( - "{LOG_PREFIX} inverted index primed workspace={}", - key.display() - ); - } - } - Ok(()) - } - - /// Acquire the cached inverted index for this workspace (building it - /// from JSONL on first access) and run `f` against it. Caller MUST - /// hold `CONVERSATION_STORE_LOCK` for the duration of the closure. - /// - /// In the normal path the index has already been warmed by - /// `prime_index_if_cold`, so the cold-build branch here is a safety net - /// for any future callers that bypass the priming step. - fn with_index(&self, f: impl FnOnce(&mut InvertedIndex) -> R) -> Result { - let key = self.root_dir(); - let mut cache = CONVERSATION_INDEX_CACHE.lock(); - if !cache.contains_key(&key) { - let mut idx = InvertedIndex::new(); - self.populate_index_unlocked(&mut idx)?; - cache.insert(key.clone(), idx); - } - let idx = cache.get_mut(&key).expect("inserted above if absent"); - Ok(f(idx)) - } - - /// Walk every per-thread JSONL file in the workspace and insert each - /// message into `idx`. Used as the fallback cold-build path inside - /// `with_index`; `prime_index_if_cold` handles the normal first-access - /// case outside the outer lock. The JSONL files are the source of truth - /// so a rebuild after a process crash is always safe. - fn populate_index_unlocked(&self, idx: &mut InvertedIndex) -> Result<(), String> { - // Caller (`with_index`) already holds `CONVERSATION_STORE_LOCK`, so we - // must NOT re-acquire it here — `parking_lot::Mutex` is not reentrant - // and doing so would deadlock. Use the `_unlocked` thread reader - // directly. - // - // `list_threads_unlocked` already handles a fresh workspace: - // `ensure_root` creates the directory + threads log if missing, and - // `read_jsonl` returns an empty Vec for an empty file. Anything that - // still bubbles up here is a real filesystem/setup failure and must - // propagate — silently returning Ok would mask it and make search - // appear to return zero results for an undiagnosed reason. - let threads = self.list_threads_unlocked()?; - for thread in threads { - let path = self.thread_messages_path(&thread.id); - if !path.exists() { - continue; - } - let messages = match read_jsonl::(&path) { - Ok(m) => m, - Err(err) => { - tracing::warn!( - "{LOG_PREFIX} index build skipped unreadable file path={} error={}", - path.display(), - err - ); - continue; - } - }; - for msg in messages { - idx.insert(&thread.id, msg); - } - } - debug!( - "{LOG_PREFIX} inverted index populated workspace={}", - self.root_dir().display() - ); - Ok(()) - } - - /// Append a message to the thread's JSONL file. Errors if the thread is missing. - pub fn append_message( - &self, - thread_id: &str, - message: ConversationMessage, - ) -> Result { - let _guard = CONVERSATION_STORE_LOCK.lock(); - if !self.thread_exists_unlocked(thread_id)? { - return Err(format!("thread {} not found", thread_id)); - } - let path = self.thread_messages_path(thread_id); - if let Some(parent) = path.parent() { - fs::create_dir_all(parent) - .map_err(|e| format!("create conversation dir {}: {e}", parent.display()))?; - } - append_jsonl(&path, &message)?; - // Bump the threads-log stat trail so subsequent `list_threads` - // calls can compute (message_count, last_message_at) without - // re-reading this file. - let threads_path = self.root_dir().join(THREADS_FILENAME); - append_jsonl( - &threads_path, - &ThreadLogEntry::MessageAppended { - thread_id: thread_id.to_string(), - last_message_at: message.created_at.clone(), - }, - )?; - // Keep the inverted index in sync. We only update if the index - // has already been materialized for this workspace — otherwise - // the next search will lazily rebuild and pick up this message - // anyway, and we avoid paying the rebuild cost on a write path. - // `insert` takes the message by value so it can move owned - // fields straight into its `DocEntry`; we clone here only when - // the cache is actually warm, paying for one extra owned copy - // (instead of cloning each field inside `insert`). - { - let mut cache = CONVERSATION_INDEX_CACHE.lock(); - if let Some(idx) = cache.get_mut(&self.root_dir()) { - idx.insert(thread_id, message.clone()); - } - } - debug!( - "{LOG_PREFIX} appended message thread_id={} message_id={} path={}", - thread_id, - message.id, - path.display() - ); - Ok(message) - } - - /// Rewrite the thread title via a new `Upsert` log entry, preserving labels. - pub fn update_thread_title( - &self, - thread_id: &str, - title: &str, - updated_at: &str, - ) -> Result { - let _guard = CONVERSATION_STORE_LOCK.lock(); - let index = self.thread_index_unlocked()?; - let entry = index - .get(thread_id) - .ok_or_else(|| format!("thread {} not found", thread_id))?; - let threads_path = self.ensure_root()?.join(THREADS_FILENAME); - append_jsonl( - &threads_path, - &ThreadLogEntry::Upsert { - thread_id: thread_id.to_string(), - title: title.to_string(), - created_at: entry.created_at.clone(), - updated_at: updated_at.to_string(), - parent_thread_id: entry.parent_thread_id.clone(), - labels: Some(entry.labels.clone()), - personality_id: entry.personality_id.clone(), - }, - )?; - debug!( - "{LOG_PREFIX} updated thread title id={} title={} path={}", - thread_id, - redact_title_for_log(title), - threads_path.display() - ); - self.thread_summary_unlocked(thread_id)? - .ok_or_else(|| format!("thread {} missing after title update", thread_id)) - } - - /// Replace the label set on a thread via a new `Upsert` log entry. - pub fn update_thread_labels( - &self, - thread_id: &str, - labels: Vec, - updated_at: &str, - ) -> Result { - let _guard = CONVERSATION_STORE_LOCK.lock(); - let index = self.thread_index_unlocked()?; - let entry = index - .get(thread_id) - .ok_or_else(|| format!("thread {} not found", thread_id))?; - let threads_path = self.ensure_root()?.join(THREADS_FILENAME); - let labels = normalize_labels(labels); - append_jsonl( - &threads_path, - &ThreadLogEntry::Upsert { - thread_id: thread_id.to_string(), - title: entry.title.clone(), - created_at: entry.created_at.clone(), - updated_at: updated_at.to_string(), - parent_thread_id: entry.parent_thread_id.clone(), - labels: Some(labels), - personality_id: entry.personality_id.clone(), - }, - )?; - debug!( - "{LOG_PREFIX} updated thread labels id={} path={}", - thread_id, - threads_path.display() - ); - self.thread_summary_unlocked(thread_id)? - .ok_or_else(|| format!("thread {} missing after labels update", thread_id)) - } - - /// Apply a patch to one message and rewrite the thread's JSONL file in place. - pub fn update_message( - &self, - thread_id: &str, - message_id: &str, - patch: ConversationMessagePatch, - ) -> Result { - let _guard = CONVERSATION_STORE_LOCK.lock(); - let path = self.thread_messages_path(thread_id); - let mut messages = read_jsonl::(&path)?; - let mut updated: Option = None; - for message in &mut messages { - if message.id == message_id { - if let Some(extra_metadata) = patch.extra_metadata.clone() { - message.extra_metadata = extra_metadata; - } - updated = Some(message.clone()); - break; - } - } - let updated = updated - .ok_or_else(|| format!("message {} not found in thread {}", message_id, thread_id))?; - rewrite_jsonl(&path, &messages)?; - debug!( - "{LOG_PREFIX} updated message thread_id={} message_id={} path={}", - thread_id, - message_id, - path.display() - ); - Ok(updated) - } - - /// Append a `Delete` entry and remove the thread's messages file. Returns - /// `false` if the thread did not exist. - pub fn delete_thread(&self, thread_id: &str, deleted_at: &str) -> Result { - let _guard = CONVERSATION_STORE_LOCK.lock(); - if !self.thread_exists_unlocked(thread_id)? { - return Ok(false); - } - let root = self.ensure_root()?; - let threads_path = root.join(THREADS_FILENAME); - append_jsonl( - &threads_path, - &ThreadLogEntry::Delete { - thread_id: thread_id.to_string(), - deleted_at: deleted_at.to_string(), - }, - )?; - let messages_path = self.thread_messages_path(thread_id); - match fs::remove_file(&messages_path) { - Ok(()) => {} - Err(error) if error.kind() == std::io::ErrorKind::NotFound => {} - Err(error) => { - return Err(format!( - "delete conversation messages {}: {error}", - messages_path.display() - )); - } - } - // Drop every indexed message for this thread so future searches - // don't surface stale content. - { - let mut cache = CONVERSATION_INDEX_CACHE.lock(); - if let Some(idx) = cache.get_mut(&self.root_dir()) { - idx.remove_thread(thread_id); - } - } - debug!( - "{LOG_PREFIX} deleted thread id={} path={}", - thread_id, - messages_path.display() - ); - Ok(true) - } - - /// Wipe the entire conversation directory and re-create an empty layout. - pub fn purge_threads(&self) -> Result { - let _guard = CONVERSATION_STORE_LOCK.lock(); - let stats = self.purge_stats_unlocked()?; - let root = self.root_dir(); - if root.exists() { - fs::remove_dir_all(&root) - .map_err(|e| format!("remove conversation dir {}: {e}", root.display()))?; - } - self.ensure_root()?; - // Drop the cached inverted index — the workspace is now empty, - // and any next search will lazily rebuild from the (now empty) - // JSONL tree. - { - let mut cache = CONVERSATION_INDEX_CACHE.lock(); - cache.remove(&root); - } - debug!( - "{LOG_PREFIX} purged threads={} messages={} root={}", - stats.thread_count, - stats.message_count, - root.display() - ); - Ok(stats) - } - - fn ensure_root(&self) -> Result { - let root = self.root_dir(); - let threads_dir = root.join(THREAD_MESSAGES_DIR); - fs::create_dir_all(&threads_dir) - .map_err(|e| format!("create conversation dir {}: {e}", threads_dir.display()))?; - let threads_file = root.join(THREADS_FILENAME); - if !threads_file.exists() { - File::create(&threads_file) - .map_err(|e| format!("create threads log {}: {e}", threads_file.display()))?; - } - Ok(root) - } - - fn root_dir(&self) -> PathBuf { - self.workspace_dir.join("memory").join("conversations") - } - - fn thread_messages_path(&self, thread_id: &str) -> PathBuf { - self.root_dir() - .join(THREAD_MESSAGES_DIR) - .join(format!("{}.jsonl", hex::encode(thread_id.as_bytes()))) - } - - fn list_threads_unlocked(&self) -> Result, String> { - let mut index = self.thread_index_unlocked()?; - // Backfill stats for any thread with no MessageAppended/Stats history - // yet (legacy data). The slow per-thread file read happens at most - // once per thread — we persist a `Stats` snapshot so subsequent - // list_threads calls hit the fast path. - let needs_backfill: Vec = index - .iter() - .filter_map(|(id, entry)| { - if entry.message_count.is_none() { - Some(id.clone()) - } else { - None - } - }) - .collect(); - if !needs_backfill.is_empty() { - let threads_path = self.ensure_root()?.join(THREADS_FILENAME); - for thread_id in &needs_backfill { - let (count, last_message_at) = self.measure_messages_unlocked(thread_id)?; - // Treat created_at as last_message_at when there are no - // messages — keeps the sort key meaningful and matches the - // pre-refactor semantics. - let resolved_last = last_message_at.unwrap_or_else(|| { - index - .get(thread_id) - .map(|e| e.created_at.clone()) - .unwrap_or_default() - }); - append_jsonl( - &threads_path, - &ThreadLogEntry::Stats { - thread_id: thread_id.clone(), - message_count: count, - last_message_at: resolved_last.clone(), - }, - )?; - if let Some(entry) = index.get_mut(thread_id) { - entry.message_count = Some(count); - entry.last_message_at = Some(resolved_last); - } - debug!( - "{LOG_PREFIX} backfilled stats thread_id={} count={}", - thread_id, count - ); - } - } - - let mut threads: Vec = index - .iter() - .map(|(thread_id, entry)| { - let message_count = entry.message_count.unwrap_or(0); - let last_message_at = entry - .last_message_at - .clone() - .unwrap_or_else(|| entry.created_at.clone()); - ConversationThread { - id: thread_id.clone(), - title: entry.title.clone(), - chat_id: None, - is_active: true, - message_count, - last_message_at, - created_at: entry.created_at.clone(), - parent_thread_id: entry.parent_thread_id.clone(), - labels: normalize_labels(entry.labels.clone()), - personality_id: entry.personality_id.clone(), - } - }) - .collect(); - threads.sort_by(|a, b| { - b.last_message_at - .cmp(&a.last_message_at) - .then_with(|| b.created_at.cmp(&a.created_at)) - }); - Ok(threads) - } - - /// Count messages and find the newest timestamp by reading the - /// per-thread JSONL file. Slow path — used only when the threads-log - /// stat history is missing (legacy data) so we can write a one-time - /// `Stats` snapshot. - fn measure_messages_unlocked( - &self, - thread_id: &str, - ) -> Result<(usize, Option), String> { - let path = self.thread_messages_path(thread_id); - if !path.exists() { - return Ok((0, None)); - } - let messages = read_jsonl::(&path)?; - let count = messages.len(); - let last = messages.last().map(|m| m.created_at.clone()); - Ok((count, last)) - } - - fn thread_summary_unlocked( - &self, - thread_id: &str, - ) -> Result, String> { - let index = self.thread_index_unlocked()?; - let entry = match index.get(thread_id) { - Some(entry) => entry, - None => return Ok(None), - }; - // Prefer the index-tracked stats (cheap). Fall back to a single - // per-thread file read for legacy threads with no stat history — - // list_threads is responsible for permanently backfilling those. - let (message_count, last_message_at) = - match (entry.message_count, entry.last_message_at.as_ref()) { - (Some(count), Some(last_at)) => (count, last_at.clone()), - _ => { - let (count, last_at) = self.measure_messages_unlocked(thread_id)?; - (count, last_at.unwrap_or_else(|| entry.created_at.clone())) - } - }; - Ok(Some(ConversationThread { - id: thread_id.to_string(), - title: entry.title.clone(), - chat_id: None, - is_active: true, - message_count, - last_message_at, - created_at: entry.created_at.clone(), - parent_thread_id: entry.parent_thread_id.clone(), - labels: normalize_labels(entry.labels.clone()), - personality_id: entry.personality_id.clone(), - })) - } - - fn thread_exists_unlocked(&self, thread_id: &str) -> Result { - Ok(self.thread_index_unlocked()?.contains_key(thread_id)) - } - - fn thread_index_unlocked(&self) -> Result, String> { - self.ensure_root()?; - let path = self.root_dir().join(THREADS_FILENAME); - let mut index: BTreeMap = BTreeMap::new(); - for entry in read_jsonl::(&path)? { - match entry { - ThreadLogEntry::Upsert { - thread_id, - title, - created_at, - parent_thread_id, - labels, - personality_id, - .. - } => { - let ( - created_at_value, - parent_thread_id_value, - labels_value, - message_count_value, - last_message_at_value, - personality_id_value, - ) = match index.get(&thread_id) { - Some(existing) => ( - existing.created_at.clone(), - parent_thread_id.or_else(|| existing.parent_thread_id.clone()), - labels - .map(normalize_labels) - .unwrap_or_else(|| existing.labels.clone()), - existing.message_count, - existing.last_message_at.clone(), - personality_id.or_else(|| existing.personality_id.clone()), - ), - None => { - let inferred = labels - .map(normalize_labels) - .unwrap_or_else(|| infer_labels(&thread_id)); - ( - created_at, - parent_thread_id, - inferred, - None, - None, - personality_id, - ) - } - }; - index.insert( - thread_id, - ThreadIndexEntry { - title, - created_at: created_at_value, - parent_thread_id: parent_thread_id_value, - labels: labels_value, - message_count: message_count_value, - last_message_at: last_message_at_value, - personality_id: personality_id_value, - }, - ); - } - ThreadLogEntry::Delete { thread_id, .. } => { - index.remove(&thread_id); - } - ThreadLogEntry::MessageAppended { - thread_id, - last_message_at, - } => { - if let Some(entry) = index.get_mut(&thread_id) { - // Increment from a known baseline. If we have no - // baseline yet (legacy thread with messages but no - // Stats snapshot), leave count as `None` so the - // backfill path in `list_threads_unlocked` can do - // the one-shot file read instead of producing a - // wrong "1" here. - if let Some(count) = entry.message_count.as_mut() { - *count += 1; - } - entry.last_message_at = Some(last_message_at); - } - } - ThreadLogEntry::Stats { - thread_id, - message_count, - last_message_at, - } => { - if let Some(entry) = index.get_mut(&thread_id) { - entry.message_count = Some(message_count); - entry.last_message_at = Some(last_message_at); - } - } - } - } - Ok(index) - } - - fn purge_stats_unlocked(&self) -> Result { - let threads = self.list_threads_unlocked()?; - let message_count = threads.iter().map(|thread| thread.message_count).sum(); - Ok(ConversationPurgeStats { - thread_count: threads.len(), - message_count, - }) - } -} - -#[derive(Debug, Clone)] -struct ThreadIndexEntry { - title: String, - created_at: String, - parent_thread_id: Option, - labels: Vec, - /// Folded message count. `None` means we have no `MessageAppended` / - /// `Stats` history for this thread yet (legacy data) — `list_threads` - /// backfills by doing a one-shot read of the per-thread messages file. - message_count: Option, - /// Timestamp of the newest message, or `None` if unknown (legacy). - last_message_at: Option, - personality_id: Option, -} - -fn infer_labels(thread_id: &str) -> Vec { - if thread_id == "proactive:morning_briefing" { - vec!["briefing".to_string()] - } else if thread_id.starts_with("proactive:") { - vec!["notification".to_string()] - } else { - vec!["general".to_string()] - } -} - -fn normalize_labels(labels: Vec) -> Vec { - let mut normalized = Vec::with_capacity(labels.len()); - for label in labels { - let next = match label.as_str() { - "work" => "general".to_string(), - "from_reflection" | "subconscious_tick" => "subconscious".to_string(), - "agent-task" | "worker" => "tasks".to_string(), - _ => label, - }; - if !normalized.contains(&next) { - normalized.push(next); - } - } - normalized -} - -fn read_jsonl(path: &Path) -> Result, String> -where - T: for<'de> serde::Deserialize<'de>, -{ - if !path.exists() { - return Ok(Vec::new()); - } - let file = File::open(path).map_err(|e| format!("open {}: {e}", path.display()))?; - let reader = BufReader::new(file); - let mut items = Vec::new(); - for (line_no, line) in reader.lines().enumerate() { - let line = - line.map_err(|e| format!("read {} line {}: {e}", path.display(), line_no + 1))?; - let trimmed = line.trim(); - if trimmed.is_empty() { - continue; - } - match serde_json::from_str::(trimmed) { - Ok(value) => items.push(value), - Err(error) => { - warn!( - "{LOG_PREFIX} skipping invalid jsonl line path={} line={} error={}", - path.display(), - line_no + 1, - error - ); - } - } - } - Ok(items) -} - -fn append_jsonl(path: &Path, value: &T) -> Result<(), String> -where - T: serde::Serialize, -{ - let parent = path - .parent() - .ok_or_else(|| format!("resolve parent dir for {}", path.display()))?; - fs::create_dir_all(parent) - .map_err(|e| format!("create jsonl dir {}: {e}", parent.display()))?; - let mut file = OpenOptions::new() - .create(true) - .append(true) - .open(path) - .map_err(|e| format!("open {} for append: {e}", path.display()))?; - let line = serde_json::to_string(value) - .map_err(|e| format!("serialize jsonl line for {}: {e}", path.display()))?; - writeln!(file, "{line}").map_err(|e| format!("write {}: {e}", path.display()))?; - file.sync_all() - .map_err(|e| format!("sync {}: {e}", path.display()))?; - Ok(()) -} - -fn rewrite_jsonl(path: &Path, values: &[T]) -> Result<(), String> -where - T: serde::Serialize, -{ - let parent = path - .parent() - .ok_or_else(|| format!("resolve parent dir for {}", path.display()))?; - fs::create_dir_all(parent) - .map_err(|e| format!("create jsonl dir {}: {e}", parent.display()))?; - let mut temp = NamedTempFile::new_in(parent) - .map_err(|e| format!("create temp jsonl in {}: {e}", parent.display()))?; - for value in values { - let line = serde_json::to_string(value) - .map_err(|e| format!("serialize jsonl line for {}: {e}", path.display()))?; - writeln!(temp, "{line}") - .map_err(|e| format!("write temp jsonl for {}: {e}", path.display()))?; - } - temp.as_file_mut() - .sync_all() - .map_err(|e| format!("sync temp jsonl for {}: {e}", path.display()))?; - temp.persist(path) - .map_err(|e| format!("persist {}: {}", path.display(), e.error))?; - Ok(()) -} - -/// Free-function shim around [`ConversationStore::ensure_thread`]. -pub fn ensure_thread( - workspace_dir: PathBuf, - request: CreateConversationThread, -) -> Result { - ConversationStore::new(workspace_dir).ensure_thread(request) -} - -/// Free-function shim around [`ConversationStore::list_threads`]. -pub fn list_threads(workspace_dir: PathBuf) -> Result, String> { - ConversationStore::new(workspace_dir).list_threads() -} - -/// Free-function shim around [`ConversationStore::get_messages`]. -pub fn get_messages( - workspace_dir: PathBuf, - thread_id: &str, -) -> Result, String> { - ConversationStore::new(workspace_dir).get_messages(thread_id) -} - -/// Free-function shim around [`ConversationStore::append_message`]. -pub fn append_message( - workspace_dir: PathBuf, - thread_id: &str, - message: ConversationMessage, -) -> Result { - ConversationStore::new(workspace_dir).append_message(thread_id, message) -} - -/// Free-function shim around [`ConversationStore::update_thread_title`]. -pub fn update_thread_title( - workspace_dir: PathBuf, - thread_id: &str, - title: &str, - updated_at: &str, -) -> Result { - ConversationStore::new(workspace_dir).update_thread_title(thread_id, title, updated_at) -} - -/// Free-function shim around [`ConversationStore::update_thread_labels`]. -pub fn update_thread_labels( - workspace_dir: PathBuf, - thread_id: &str, - labels: Vec, - updated_at: &str, -) -> Result { - ConversationStore::new(workspace_dir).update_thread_labels(thread_id, labels, updated_at) -} - -/// Free-function shim around [`ConversationStore::update_message`]. -pub fn update_message( - workspace_dir: PathBuf, - thread_id: &str, - message_id: &str, - patch: ConversationMessagePatch, -) -> Result { - ConversationStore::new(workspace_dir).update_message(thread_id, message_id, patch) -} - -/// Free-function shim around [`ConversationStore::purge_threads`]. -pub fn purge_threads(workspace_dir: PathBuf) -> Result { - ConversationStore::new(workspace_dir).purge_threads() -} - -/// Free-function shim around [`ConversationStore::delete_thread`]. -pub fn delete_thread( - workspace_dir: PathBuf, - thread_id: &str, - deleted_at: &str, -) -> Result { - ConversationStore::new(workspace_dir).delete_thread(thread_id, deleted_at) -} - -#[cfg(test)] -#[path = "store_tests.rs"] -mod tests; diff --git a/src/openhuman/memory_conversations/store_tests.rs b/src/openhuman/memory_conversations/store_tests.rs deleted file mode 100644 index 0c7b77a66..000000000 --- a/src/openhuman/memory_conversations/store_tests.rs +++ /dev/null @@ -1,1322 +0,0 @@ -//! Unit tests for the JSONL-backed [`ConversationStore`], exercising thread -//! upsert, message append, label/title updates, deletion and purge semantics. - -use tempfile::TempDir; - -use super::*; -use serde_json::json; - -fn make_store() -> (TempDir, ConversationStore) { - let temp = TempDir::new().expect("tempdir"); - let store = ConversationStore::new(temp.path().to_path_buf()); - (temp, store) -} - -#[test] -fn store_roundtrips_threads_and_messages() { - let (_temp, store) = make_store(); - let created_at = "2026-04-10T12:00:00Z".to_string(); - let thread = store - .ensure_thread(CreateConversationThread { - parent_thread_id: None, - id: "default-thread".to_string(), - title: "Conversation".to_string(), - created_at: created_at.clone(), - labels: None, - personality_id: None, - }) - .expect("ensure thread"); - assert_eq!(thread.message_count, 0); - - store - .append_message( - "default-thread", - ConversationMessage { - id: "m1".to_string(), - content: "hello".to_string(), - message_type: "text".to_string(), - extra_metadata: json!({}), - sender: "user".to_string(), - created_at: "2026-04-10T12:01:00Z".to_string(), - }, - ) - .expect("append message"); - - let threads = store.list_threads().expect("list threads"); - assert_eq!(threads.len(), 1); - assert_eq!(threads[0].message_count, 1); - assert_eq!(threads[0].last_message_at, "2026-04-10T12:01:00Z"); - - let messages = store.get_messages("default-thread").expect("get messages"); - assert_eq!(messages.len(), 1); - assert_eq!(messages[0].content, "hello"); -} - -#[test] -fn get_messages_for_new_empty_thread_returns_empty_list() { - let (_temp, store) = make_store(); - store - .ensure_thread(CreateConversationThread { - parent_thread_id: None, - id: "empty-thread".to_string(), - title: "Conversation".to_string(), - created_at: "2026-04-10T12:00:00Z".to_string(), - labels: None, - personality_id: None, - }) - .expect("ensure thread"); - - let messages = store.get_messages("empty-thread").expect("get messages"); - assert!(messages.is_empty()); -} - -#[test] -fn store_updates_message_metadata() { - let (_temp, store) = make_store(); - store - .ensure_thread(CreateConversationThread { - parent_thread_id: None, - id: "default-thread".to_string(), - title: "Conversation".to_string(), - created_at: "2026-04-10T12:00:00Z".to_string(), - labels: None, - personality_id: None, - }) - .expect("ensure thread"); - store - .append_message( - "default-thread", - ConversationMessage { - id: "m1".to_string(), - content: "hello".to_string(), - message_type: "text".to_string(), - extra_metadata: json!({}), - sender: "user".to_string(), - created_at: "2026-04-10T12:01:00Z".to_string(), - }, - ) - .expect("append message"); - - let updated = store - .update_message( - "default-thread", - "m1", - ConversationMessagePatch { - extra_metadata: Some(json!({ "myReactions": ["👍"] })), - }, - ) - .expect("update message"); - - assert_eq!(updated.extra_metadata, json!({ "myReactions": ["👍"] })); - let messages = store.get_messages("default-thread").expect("get messages"); - assert_eq!(messages[0].extra_metadata, json!({ "myReactions": ["👍"] })); -} - -#[test] -fn purge_removes_threads_and_messages() { - let (_temp, store) = make_store(); - store - .ensure_thread(CreateConversationThread { - parent_thread_id: None, - id: "default-thread".to_string(), - title: "Conversation".to_string(), - created_at: "2026-04-10T12:00:00Z".to_string(), - labels: None, - personality_id: None, - }) - .expect("ensure thread"); - store - .append_message( - "default-thread", - ConversationMessage { - id: "m1".to_string(), - content: "hello".to_string(), - message_type: "text".to_string(), - extra_metadata: json!({}), - sender: "user".to_string(), - created_at: "2026-04-10T12:01:00Z".to_string(), - }, - ) - .expect("append message"); - - let stats = store.purge_threads().expect("purge"); - assert_eq!(stats.thread_count, 1); - assert_eq!(stats.message_count, 1); - assert!(store.list_threads().expect("list threads").is_empty()); -} - -#[test] -fn ensure_thread_is_idempotent() { - let (_temp, store) = make_store(); - let req = CreateConversationThread { - parent_thread_id: None, - id: "t1".to_string(), - title: "Thread".to_string(), - created_at: "2026-04-10T12:00:00Z".to_string(), - labels: None, - personality_id: None, - }; - store.ensure_thread(req.clone()).unwrap(); - store.ensure_thread(req).unwrap(); - let threads = store.list_threads().unwrap(); - assert_eq!(threads.len(), 1); -} - -#[test] -fn delete_thread_removes_thread_and_messages() { - let (_temp, store) = make_store(); - store - .ensure_thread(CreateConversationThread { - parent_thread_id: None, - id: "t1".to_string(), - title: "Thread".to_string(), - created_at: "2026-04-10T12:00:00Z".to_string(), - labels: None, - personality_id: None, - }) - .unwrap(); - store - .append_message( - "t1", - ConversationMessage { - id: "m1".to_string(), - content: "msg".to_string(), - message_type: "text".to_string(), - extra_metadata: json!({}), - sender: "user".to_string(), - created_at: "2026-04-10T12:01:00Z".to_string(), - }, - ) - .unwrap(); - store.delete_thread("t1", "2026-04-10T12:02:00Z").unwrap(); - let threads = store.list_threads().unwrap(); - assert!(threads.is_empty()); -} - -#[test] -fn delete_nonexistent_thread_is_ok() { - let (_temp, store) = make_store(); - // Should not error - store - .delete_thread("nonexistent", "2026-04-10T12:00:00Z") - .unwrap(); -} - -#[test] -fn get_messages_empty_thread() { - let (_temp, store) = make_store(); - store - .ensure_thread(CreateConversationThread { - parent_thread_id: None, - id: "t1".to_string(), - title: "Empty".to_string(), - created_at: "2026-04-10T12:00:00Z".to_string(), - labels: None, - personality_id: None, - }) - .unwrap(); - let messages = store.get_messages("t1").unwrap(); - assert!(messages.is_empty()); -} - -#[test] -fn get_messages_nonexistent_thread() { - let (_temp, store) = make_store(); - let messages = store.get_messages("nonexistent").unwrap(); - assert!(messages.is_empty()); -} - -#[test] -fn multiple_threads_and_messages() { - let (_temp, store) = make_store(); - for i in 0..3 { - store - .ensure_thread(CreateConversationThread { - parent_thread_id: None, - id: format!("t{i}"), - title: format!("Thread {i}"), - created_at: format!("2026-04-10T12:0{i}:00Z"), - labels: None, - personality_id: None, - }) - .unwrap(); - store - .append_message( - &format!("t{i}"), - ConversationMessage { - id: format!("m{i}"), - content: format!("msg {i}"), - message_type: "text".to_string(), - extra_metadata: json!({}), - sender: "user".to_string(), - created_at: format!("2026-04-10T12:0{i}:30Z"), - }, - ) - .unwrap(); - } - let threads = store.list_threads().unwrap(); - assert_eq!(threads.len(), 3); -} - -#[test] -fn purge_on_empty_store() { - let (_temp, store) = make_store(); - let stats = store.purge_threads().unwrap(); - assert_eq!(stats.thread_count, 0); - assert_eq!(stats.message_count, 0); -} - -#[test] -fn update_message_nonexistent_returns_error() { - let (_temp, store) = make_store(); - store - .ensure_thread(CreateConversationThread { - parent_thread_id: None, - id: "t1".to_string(), - title: "Thread".to_string(), - created_at: "2026-04-10T12:00:00Z".to_string(), - labels: None, - personality_id: None, - }) - .unwrap(); - let result = store.update_message( - "t1", - "nonexistent", - ConversationMessagePatch { - extra_metadata: Some(json!({})), - }, - ); - assert!(result.is_err()); -} - -#[test] -fn update_thread_title_persists_latest_title() { - let (_temp, store) = make_store(); - store - .ensure_thread(CreateConversationThread { - parent_thread_id: None, - id: "t1".to_string(), - title: "Chat Apr 10 12:00 PM".to_string(), - created_at: "2026-04-10T12:00:00Z".to_string(), - labels: None, - personality_id: None, - }) - .unwrap(); - - let updated = store - .update_thread_title("t1", "Invoice follow-up", "2026-04-10T12:03:00Z") - .unwrap(); - - assert_eq!(updated.title, "Invoice follow-up"); - let threads = store.list_threads().unwrap(); - assert_eq!(threads[0].title, "Invoice follow-up"); - assert_eq!(threads[0].created_at, "2026-04-10T12:00:00Z"); -} - -#[test] -fn store_handles_labels_and_inference() { - let (_temp, store) = make_store(); - - // 1. Explicit labels on ensure - store - .ensure_thread(CreateConversationThread { - parent_thread_id: None, - id: "t1".to_string(), - title: "Thread 1".to_string(), - created_at: "2026-04-10T12:00:00Z".to_string(), - labels: Some(vec!["custom".to_string()]), - personality_id: None, - }) - .unwrap(); - - // 2. Inferred labels for morning briefing - store - .ensure_thread(CreateConversationThread { - parent_thread_id: None, - id: "proactive:morning_briefing".to_string(), - title: "Morning Briefing".to_string(), - created_at: "2026-04-10T12:00:00Z".to_string(), - labels: None, - personality_id: None, - }) - .unwrap(); - - // 3. Inferred labels for other proactive - store - .ensure_thread(CreateConversationThread { - parent_thread_id: None, - id: "proactive:system".to_string(), - title: "System Notification".to_string(), - created_at: "2026-04-10T12:00:00Z".to_string(), - labels: None, - personality_id: None, - }) - .unwrap(); - - // 4. Default inferred labels (general) - store - .ensure_thread(CreateConversationThread { - parent_thread_id: None, - id: "user-thread".to_string(), - title: "User Chat".to_string(), - created_at: "2026-04-10T12:00:00Z".to_string(), - labels: None, - personality_id: None, - }) - .unwrap(); - - // 5. Legacy explicit labels normalize into their canonical buckets. - store - .ensure_thread(CreateConversationThread { - parent_thread_id: None, - id: "legacy-work-thread".to_string(), - title: "Legacy Work Chat".to_string(), - created_at: "2026-04-10T12:00:00Z".to_string(), - labels: Some(vec![ - "work".to_string(), - "urgent".to_string(), - "work".to_string(), - ]), - personality_id: None, - }) - .unwrap(); - store - .ensure_thread(CreateConversationThread { - parent_thread_id: None, - id: "legacy-subconscious-thread".to_string(), - title: "Legacy Subconscious Chat".to_string(), - created_at: "2026-04-10T12:00:00Z".to_string(), - labels: Some(vec![ - "from_reflection".to_string(), - "subconscious_tick".to_string(), - ]), - personality_id: None, - }) - .unwrap(); - store - .ensure_thread(CreateConversationThread { - parent_thread_id: None, - id: "legacy-task-thread".to_string(), - title: "Legacy Task Chat".to_string(), - created_at: "2026-04-10T12:00:00Z".to_string(), - labels: Some(vec!["agent-task".to_string(), "worker".to_string()]), - personality_id: None, - }) - .unwrap(); - - let threads = store.list_threads().unwrap(); - { - let t1 = threads.iter().find(|t| t.id == "t1").unwrap(); - assert_eq!(t1.labels, vec!["custom"]); - } - { - let mb = threads - .iter() - .find(|t| t.id == "proactive:morning_briefing") - .unwrap(); - assert_eq!(mb.labels, vec!["briefing"]); - } - { - let sys = threads.iter().find(|t| t.id == "proactive:system").unwrap(); - assert_eq!(sys.labels, vec!["notification"]); - } - { - let user = threads.iter().find(|t| t.id == "user-thread").unwrap(); - assert_eq!(user.labels, vec!["general"]); - } - { - let legacy = threads - .iter() - .find(|t| t.id == "legacy-work-thread") - .unwrap(); - assert_eq!(legacy.labels, vec!["general", "urgent"]); - } - { - let legacy = threads - .iter() - .find(|t| t.id == "legacy-subconscious-thread") - .unwrap(); - assert_eq!(legacy.labels, vec!["subconscious"]); - } - { - let legacy = threads - .iter() - .find(|t| t.id == "legacy-task-thread") - .unwrap(); - assert_eq!(legacy.labels, vec!["tasks"]); - } - - // 6. Update labels - store - .update_thread_labels("t1", vec!["updated".to_string()], "2026-04-10T12:05:00Z") - .unwrap(); - let threads = store.list_threads().unwrap(); - { - let t1 = threads.iter().find(|t| t.id == "t1").unwrap(); - assert_eq!(t1.labels, vec!["updated"]); - } - - // 7. Title update preserves labels - store - .update_thread_title("t1", "New Title", "2026-04-10T12:06:00Z") - .unwrap(); - let threads = store.list_threads().unwrap(); - { - let t1 = threads.iter().find(|t| t.id == "t1").unwrap(); - assert_eq!(t1.labels, vec!["updated"]); - assert_eq!(t1.title, "New Title"); - } -} - -#[test] -fn conversation_store_new() { - let tmp = TempDir::new().unwrap(); - let store = ConversationStore::new(tmp.path().to_path_buf()); - let threads = store.list_threads().unwrap(); - assert!(threads.is_empty()); -} - -#[test] -fn conversation_purge_stats_default() { - let stats = ConversationPurgeStats::default(); - assert_eq!(stats.thread_count, 0); - assert_eq!(stats.message_count, 0); -} - -#[test] -fn list_threads_does_not_read_per_thread_files_after_first_call() { - // After the first list_threads (which may backfill), deleting every - // per-thread messages file must leave count + last_message_at intact — - // proving the slow path is no longer on the hot loop. - let (temp, store) = make_store(); - store - .ensure_thread(CreateConversationThread { - parent_thread_id: None, - id: "t1".to_string(), - title: "T1".to_string(), - created_at: "2026-04-10T12:00:00Z".to_string(), - labels: None, - personality_id: None, - }) - .unwrap(); - for i in 0..3 { - store - .append_message( - "t1", - ConversationMessage { - id: format!("m{i}"), - content: format!("hi {i}"), - message_type: "text".to_string(), - extra_metadata: json!({}), - sender: "user".to_string(), - created_at: format!("2026-04-10T12:0{}:00Z", i + 1), - }, - ) - .unwrap(); - } - // Warm-up: list_threads folds the MessageAppended entries. - let _ = store.list_threads().unwrap(); - - // Now blow away the per-thread JSONL. If list_threads still reads it, - // the count would drop to 0. If our index-only path works, the cached - // (3, latest_ts) survives. - let messages_dir = temp - .path() - .join("memory") - .join("conversations") - .join("threads"); - let entries: Vec<_> = std::fs::read_dir(&messages_dir) - .unwrap() - .filter_map(Result::ok) - .collect(); - for entry in entries { - std::fs::remove_file(entry.path()).unwrap(); - } - - let threads = store.list_threads().unwrap(); - assert_eq!(threads.len(), 1); - assert_eq!(threads[0].message_count, 3); - assert_eq!(threads[0].last_message_at, "2026-04-10T12:03:00Z"); -} - -#[test] -fn backfill_writes_stats_snapshot_for_legacy_threads() { - // Simulate legacy data: write only an Upsert entry (no MessageAppended) - // plus a per-thread messages file. The first list_threads must backfill. - let (temp, store) = make_store(); - let conversations_dir = temp.path().join("memory").join("conversations"); - std::fs::create_dir_all(conversations_dir.join("threads")).unwrap(); - - let threads_log = conversations_dir.join("threads.jsonl"); - let upsert = serde_json::json!({ - "op": "upsert", - "thread_id": "legacy-1", - "title": "Legacy", - "created_at": "2026-04-10T08:00:00Z", - "updated_at": "2026-04-10T08:00:00Z", - }); - std::fs::write(&threads_log, format!("{}\n", upsert)).unwrap(); - - // Write 2 messages directly to the per-thread file (no MessageAppended - // entries — this is what pre-upgrade data looks like). - let messages_file = conversations_dir - .join("threads") - .join(format!("{}.jsonl", hex::encode("legacy-1".as_bytes()))); - let m1 = serde_json::json!({ - "id": "m1", "content": "a", "type": "text", - "extraMetadata": {}, "sender": "user", - "createdAt": "2026-04-10T09:00:00Z", - }); - let m2 = serde_json::json!({ - "id": "m2", "content": "b", "type": "text", - "extraMetadata": {}, "sender": "user", - "createdAt": "2026-04-10T09:05:00Z", - }); - std::fs::write(&messages_file, format!("{m1}\n{m2}\n")).unwrap(); - - let threads = store.list_threads().unwrap(); - assert_eq!(threads.len(), 1); - assert_eq!(threads[0].message_count, 2); - assert_eq!(threads[0].last_message_at, "2026-04-10T09:05:00Z"); - - // The backfill should have appended a Stats entry — check the log - // contents now contain "op":"stats" for legacy-1. - let log = std::fs::read_to_string(&threads_log).unwrap(); - assert!( - log.contains("\"op\":\"stats\"") && log.contains("legacy-1"), - "expected backfilled Stats entry in threads.jsonl, got:\n{log}", - ); - - // Second call: blow away the messages file. Stats from the log keep - // count + last_message_at correct without re-reading. - std::fs::remove_file(&messages_file).unwrap(); - let threads2 = store.list_threads().unwrap(); - assert_eq!(threads2[0].message_count, 2); - assert_eq!(threads2[0].last_message_at, "2026-04-10T09:05:00Z"); -} - -#[test] -fn legacy_log_without_stats_still_parses() { - // Old on-disk format (only Upsert + Delete variants) must still load - // without errors after the enum gained MessageAppended + Stats. - let (temp, store) = make_store(); - let conversations_dir = temp.path().join("memory").join("conversations"); - std::fs::create_dir_all(conversations_dir.join("threads")).unwrap(); - let threads_log = conversations_dir.join("threads.jsonl"); - let upsert = serde_json::json!({ - "op": "upsert", - "thread_id": "old", - "title": "Old", - "created_at": "2026-04-10T08:00:00Z", - "updated_at": "2026-04-10T08:00:00Z", - }); - std::fs::write(&threads_log, format!("{}\n", upsert)).unwrap(); - - let threads = store.list_threads().unwrap(); - assert_eq!(threads.len(), 1); - assert_eq!(threads[0].id, "old"); - assert_eq!(threads[0].message_count, 0); - // No messages → last_message_at falls back to created_at. - assert_eq!(threads[0].last_message_at, "2026-04-10T08:00:00Z"); -} - -#[test] -fn delete_thread_clears_stats_from_index() { - let (_temp, store) = make_store(); - store - .ensure_thread(CreateConversationThread { - parent_thread_id: None, - id: "doomed".to_string(), - title: "Doomed".to_string(), - created_at: "2026-04-10T12:00:00Z".to_string(), - labels: None, - personality_id: None, - }) - .unwrap(); - store - .append_message( - "doomed", - ConversationMessage { - id: "m1".to_string(), - content: "x".to_string(), - message_type: "text".to_string(), - extra_metadata: json!({}), - sender: "user".to_string(), - created_at: "2026-04-10T12:01:00Z".to_string(), - }, - ) - .unwrap(); - assert_eq!(store.list_threads().unwrap().len(), 1); - - store - .delete_thread("doomed", "2026-04-10T12:02:00Z") - .unwrap(); - assert!(store.list_threads().unwrap().is_empty()); -} - -#[test] -fn search_cross_thread_messages_finds_hits_outside_excluded_thread() { - let (_temp, store) = make_store(); - - // Chat A — durable fact lives here. - store - .ensure_thread(CreateConversationThread { - parent_thread_id: None, - id: "thread-a".to_string(), - title: "Chat A".to_string(), - created_at: "2026-04-10T12:00:00Z".to_string(), - labels: None, - personality_id: None, - }) - .unwrap(); - store - .append_message( - "thread-a", - ConversationMessage { - id: "m-a-1".to_string(), - content: "Remember: my project is called Phoenix and uses Go and PostgreSQL." - .to_string(), - message_type: "text".to_string(), - extra_metadata: json!({}), - sender: "user".to_string(), - created_at: "2026-04-10T12:01:00Z".to_string(), - }, - ) - .unwrap(); - - // Chat B — active chat, asking dependent question. Should be excluded - // so its own text doesn't echo back into [Cross-chat context]. - store - .ensure_thread(CreateConversationThread { - parent_thread_id: None, - id: "thread-b".to_string(), - title: "Chat B".to_string(), - created_at: "2026-04-10T13:00:00Z".to_string(), - labels: None, - personality_id: None, - }) - .unwrap(); - store - .append_message( - "thread-b", - ConversationMessage { - id: "m-b-1".to_string(), - content: "What database does my project use?".to_string(), - message_type: "text".to_string(), - extra_metadata: json!({}), - sender: "user".to_string(), - created_at: "2026-04-10T13:01:00Z".to_string(), - }, - ) - .unwrap(); - - let hits = store - .search_cross_thread_messages("What database does my project use", 10, Some("thread-b")) - .expect("cross-thread search"); - - assert_eq!(hits.len(), 1, "exactly one cross-thread hit"); - let hit = &hits[0]; - assert_eq!(hit.thread_id, "thread-a"); - assert!(hit.content.contains("PostgreSQL")); - assert!(hit.score > 0.0); -} - -#[test] -fn search_cross_thread_messages_excludes_active_thread() { - let (_temp, store) = make_store(); - - // Single thread — the only matching message lives in the thread we're - // about to exclude. Expect zero hits (don't echo same-chat history). - store - .ensure_thread(CreateConversationThread { - parent_thread_id: None, - id: "thread-only".to_string(), - title: "Only".to_string(), - created_at: "2026-04-10T12:00:00Z".to_string(), - labels: None, - personality_id: None, - }) - .unwrap(); - store - .append_message( - "thread-only", - ConversationMessage { - id: "m-1".to_string(), - content: "PostgreSQL deployment running on staging".to_string(), - message_type: "text".to_string(), - extra_metadata: json!({}), - sender: "user".to_string(), - created_at: "2026-04-10T12:01:00Z".to_string(), - }, - ) - .unwrap(); - - let hits = store - .search_cross_thread_messages("PostgreSQL deployment staging", 10, Some("thread-only")) - .expect("cross-thread search"); - assert!( - hits.is_empty(), - "active thread must not echo into cross-chat" - ); - - // Sanity: without exclude, the hit is returned. - let hits_no_exclude = store - .search_cross_thread_messages("PostgreSQL deployment staging", 10, None) - .expect("cross-thread search"); - assert_eq!(hits_no_exclude.len(), 1); -} - -#[test] -fn search_cross_thread_messages_skips_short_terms_and_empty_queries() { - let (_temp, store) = make_store(); - store - .ensure_thread(CreateConversationThread { - parent_thread_id: None, - id: "t".to_string(), - title: "T".to_string(), - created_at: "2026-04-10T12:00:00Z".to_string(), - labels: None, - personality_id: None, - }) - .unwrap(); - store - .append_message( - "t", - ConversationMessage { - id: "m".to_string(), - content: "Postgres".to_string(), - message_type: "text".to_string(), - extra_metadata: json!({}), - sender: "user".to_string(), - created_at: "2026-04-10T12:01:00Z".to_string(), - }, - ) - .unwrap(); - - // All terms < 3 chars → empty - assert!(store - .search_cross_thread_messages("a is on", 10, None) - .unwrap() - .is_empty()); - // Empty query → empty - assert!(store - .search_cross_thread_messages("", 10, None) - .unwrap() - .is_empty()); -} - -#[test] -fn search_cross_thread_messages_finds_polish_substring_without_diacritics() { - let (_temp, store) = make_store(); - store - .ensure_thread(CreateConversationThread { - parent_thread_id: None, - id: "thread-pl".to_string(), - title: "PL".to_string(), - created_at: "2026-04-10T12:00:00Z".to_string(), - labels: None, - personality_id: None, - }) - .unwrap(); - store - .append_message( - "thread-pl", - ConversationMessage { - id: "m1".to_string(), - content: "Lecę w piątek do Łodzi a potem Krakowa".to_string(), - message_type: "text".to_string(), - extra_metadata: json!({}), - sender: "user".to_string(), - created_at: "2026-04-10T12:01:00Z".to_string(), - }, - ) - .unwrap(); - - // Query without diacritics should still find content with them. - let hits = store - .search_cross_thread_messages("Lodzi", 10, None) - .expect("cross-thread search"); - assert_eq!(hits.len(), 1, "ł-fold should match Łodzi via lodzi"); - - let hits = store - .search_cross_thread_messages("krakow", 10, None) - .expect("cross-thread search"); - assert_eq!(hits.len(), 1, "diacritic strip should match Krakowa"); -} - -#[test] -fn search_cross_thread_messages_finds_japanese_bigram_match() { - let (_temp, store) = make_store(); - store - .ensure_thread(CreateConversationThread { - parent_thread_id: None, - id: "thread-jp".to_string(), - title: "JP".to_string(), - created_at: "2026-04-10T12:00:00Z".to_string(), - labels: None, - personality_id: None, - }) - .unwrap(); - store - .append_message( - "thread-jp", - ConversationMessage { - id: "m1".to_string(), - content: "明日東京に行きます".to_string(), // "Tomorrow I'm going to Tokyo" - message_type: "text".to_string(), - extra_metadata: json!({}), - sender: "user".to_string(), - created_at: "2026-04-10T12:01:00Z".to_string(), - }, - ) - .unwrap(); - - let hits = store - .search_cross_thread_messages("東京", 10, None) - .expect("cross-thread search"); - assert_eq!(hits.len(), 1, "CJK bigram lookup should find 東京"); - assert_eq!(hits[0].message_id, "m1"); -} - -#[test] -fn search_cross_thread_messages_rebuilds_index_from_jsonl_after_reopen() { - // First store handle writes messages, second handle (simulating - // process restart on the same workspace dir) must lazy-rebuild the - // index from JSONL and still answer search queries. - let temp = TempDir::new().expect("tempdir"); - let workspace = temp.path().to_path_buf(); - { - let store = ConversationStore::new(workspace.clone()); - store - .ensure_thread(CreateConversationThread { - parent_thread_id: None, - id: "thread-x".to_string(), - title: "X".to_string(), - created_at: "2026-04-10T12:00:00Z".to_string(), - labels: None, - personality_id: None, - }) - .unwrap(); - store - .append_message( - "thread-x", - ConversationMessage { - id: "m1".to_string(), - content: "persisted across reopen — checksum kitten".to_string(), - message_type: "text".to_string(), - extra_metadata: json!({}), - sender: "user".to_string(), - created_at: "2026-04-10T12:01:00Z".to_string(), - }, - ) - .unwrap(); - } - // The cache key is per-workspace path; this TempDir was never seen - // before, so a fresh store handle will trigger a lazy rebuild. - let reopened = ConversationStore::new(workspace); - let hits = reopened - .search_cross_thread_messages("kitten", 10, None) - .expect("cross-thread search"); - assert_eq!(hits.len(), 1, "reopened store must rebuild index from disk"); -} - -#[test] -fn update_thread_labels_missing_thread_returns_error() { - let (_temp, store) = make_store(); - let err = store - .update_thread_labels("missing", vec!["work".into()], "2026-04-10T12:05:00Z") - .unwrap_err(); - assert!(err.contains("thread missing not found")); -} - -#[test] -fn cold_search_does_not_serialize_on_outer_lock() { - // Issue #2849: verify that a cold-cache search releases the store - // lock before the JSONL rebuild, so concurrent writes aren't blocked. - let (_temp, store) = make_store(); - - // Seed a thread with a message so the search has something to find. - store - .ensure_thread(CreateConversationThread { - id: "t1".to_string(), - title: "test thread".to_string(), - created_at: "2026-01-01T00:00:00Z".to_string(), - parent_thread_id: None, - labels: None, - personality_id: None, - }) - .unwrap(); - store - .append_message( - "t1", - ConversationMessage { - id: "m1".to_string(), - content: "hello world".to_string(), - message_type: "text".to_string(), - extra_metadata: json!({}), - sender: "user".to_string(), - created_at: "2026-01-01T00:00:00Z".to_string(), - }, - ) - .unwrap(); - - // Evict any warm cache so the next search triggers a cold rebuild. - { - let mut cache = CONVERSATION_INDEX_CACHE.lock(); - cache.remove(&store.root_dir()); - } - - // Spawn a thread that tries to append a message while a cold search - // is (conceptually) running. In the old code this would deadlock or - // serialize behind the full rebuild; in the fixed code the store lock - // is released after the thread-list snapshot and the append succeeds - // concurrently. - let store2 = store.clone(); - let writer = std::thread::spawn(move || { - store2 - .append_message( - "t1", - ConversationMessage { - id: "m2".to_string(), - content: "concurrent write".to_string(), - message_type: "text".to_string(), - extra_metadata: json!({}), - sender: "assistant".to_string(), - created_at: "2026-01-01T00:00:01Z".to_string(), - }, - ) - .unwrap(); - }); - - // Run the cold search — should not deadlock. - let results = store - .search_cross_thread_messages("hello", 10, None) - .unwrap(); - assert!(!results.is_empty(), "search should find seeded message"); - - // The concurrent write must also succeed. - writer.join().expect("concurrent write must not deadlock"); -} - -#[test] -fn read_jsonl_skips_invalid_lines_but_keeps_valid_ones() { - let tmp = TempDir::new().unwrap(); - let path = tmp.path().join("messages.jsonl"); - std::fs::write( - &path, - concat!( - "{\"id\":\"m1\",\"content\":\"ok\",\"type\":\"text\",\"extraMetadata\":{},\"sender\":\"user\",\"createdAt\":\"2026-04-10T12:00:00Z\"}\n", - "{not valid json}\n", - "{\"id\":\"m2\",\"content\":\"ok2\",\"type\":\"text\",\"extraMetadata\":{},\"sender\":\"agent\",\"createdAt\":\"2026-04-10T12:01:00Z\"}\n" - ), - ) - .unwrap(); - - let messages: Vec = read_jsonl(&path).expect("read jsonl"); - assert_eq!(messages.len(), 2); - assert_eq!(messages[0].id, "m1"); - assert_eq!(messages[1].id, "m2"); -} - -// ── concurrency: search cold rebuild must not block concurrent append ──────── - -/// Regression test for issue #2849. -/// -/// Before the fix, `search_cross_thread_messages` held `CONVERSATION_STORE_LOCK` -/// for the entire cold index rebuild, stalling every concurrent -/// `append_message` call for as long as the rebuild took. The fix -/// moves the rebuild outside the outer lock (`prime_index_if_cold`), so -/// an append in flight during a cold rebuild acquires the outer lock -/// independently and completes promptly. -/// -/// The test seeds a fresh workspace (cold cache), races a search against -/// an append using a barrier, and asserts the append finishes within a -/// generous timeout that would be violated if the two operations were -/// serialised through the outer lock. -#[test] -fn search_cold_rebuild_does_not_block_concurrent_append() { - use std::sync::{mpsc, Arc, Barrier}; - use std::thread; - use std::time::Duration; - - let ts = "2026-04-10T12:00:00Z".to_string(); - - // Fresh TempDir → path never seen by the process-level cache → cold. - // Note: append_message only updates an *existing* cache entry; it never - // inserts one, so the cache stays cold until the first search call. - let temp = TempDir::new().unwrap(); - let store = ConversationStore::new(temp.path().to_path_buf()); - - store - .ensure_thread(CreateConversationThread { - parent_thread_id: None, - id: "t1".to_string(), - title: "Rebuild thread".to_string(), - created_at: ts.clone(), - labels: None, - personality_id: None, - }) - .unwrap(); - - // Seed enough messages to give the rebuild real work. - for i in 0..200_usize { - store - .append_message( - "t1", - ConversationMessage { - id: format!("seed-{i}"), - content: format!("seed message {i} for cold rebuild test"), - message_type: "text".to_string(), - extra_metadata: serde_json::json!({}), - sender: "user".to_string(), - created_at: ts.clone(), - }, - ) - .unwrap(); - } - - let store_search = store.clone(); - let store_append = store.clone(); - - // Both threads start at the same time. - let barrier = Arc::new(Barrier::new(2)); - let b_search = Arc::clone(&barrier); - let b_append = Arc::clone(&barrier); - - let search_handle = thread::spawn(move || { - b_search.wait(); - store_search.search_cross_thread_messages("seed message", 5, None) - }); - - let (tx, rx) = mpsc::channel(); - thread::spawn(move || { - b_append.wait(); - let result = store_append.append_message( - "t1", - ConversationMessage { - id: "concurrent-append".to_string(), - content: "written during cold rebuild".to_string(), - message_type: "text".to_string(), - extra_metadata: serde_json::json!({}), - sender: "user".to_string(), - created_at: ts, - }, - ); - let _ = tx.send(result); - }); - - // append_message must complete even if the rebuild is in progress. On the - // old code this blocked for the full rebuild duration; on fixed code the - // two operations proceed concurrently. The 30 s budget tolerates a slow CI - // runner — a genuine deadlock never completes, so a regression still fails. - let append_result = rx - .recv_timeout(Duration::from_secs(30)) - .expect("append_message did not complete within 30 s — likely blocked by cold rebuild"); - assert!( - append_result.is_ok(), - "append failed: {:?}", - append_result.err() - ); - - let search_result = search_handle.join().expect("search thread panicked"); - assert!( - search_result.is_ok(), - "search failed: {:?}", - search_result.err() - ); -} - -// ── legacy workspace (pre-Stats backfill path) ─────────────────────────────── - -/// Regression test for issue #2849 (backfill path). -/// -/// A workspace where `threads.jsonl` contains only `Upsert` entries with no -/// `MessageAppended` / `Stats` history is a "pre-Stats" workspace — common for -/// data written before the Stats log was introduced. When -/// `list_threads_unlocked` encounters such threads it calls -/// `measure_messages_unlocked` per thread and appends a `Stats` entry to -/// `threads.jsonl`, all while holding `CONVERSATION_STORE_LOCK`. -/// -/// `prime_index_if_cold` must NOT call `list_threads_unlocked`. It uses -/// `thread_index_unlocked` (header-only, no per-thread I/O) to snapshot -/// thread IDs under the lock, then reads per-thread JSONL content outside -/// the lock. This test verifies that a cold search on such a workspace -/// still finds the correct messages, and that the former blocking code path -/// is no longer reachable from `prime_index_if_cold`. -#[test] -fn prime_index_cold_build_works_on_legacy_workspace_without_stats() { - let temp = TempDir::new().unwrap(); - let store = ConversationStore::new(temp.path().to_path_buf()); - let ts = "2023-06-01T00:00:00Z".to_string(); - - // Bootstrap a legacy workspace by writing directly to the JSONL files — - // bypassing ensure_thread / append_message so no MessageAppended or Stats - // entries end up in threads.jsonl. This is exactly the shape produced by - // versions of the code predating the Stats log. - let root = store.root_dir(); - std::fs::create_dir_all(root.join(THREAD_MESSAGES_DIR)).unwrap(); - - // Write an Upsert-only threads.jsonl (no MessageAppended / Stats entries). - append_jsonl( - &root.join(THREADS_FILENAME), - &ThreadLogEntry::Upsert { - thread_id: "legacy-t1".to_string(), - title: "Legacy Thread".to_string(), - created_at: ts.clone(), - updated_at: ts.clone(), - parent_thread_id: None, - labels: None, - personality_id: None, - }, - ) - .unwrap(); - - // Write messages directly to the per-thread JSONL file, bypassing - // append_message so message_count stays None in the index. - let msg_path = store.thread_messages_path("legacy-t1"); - for i in 0..3_usize { - append_jsonl( - &msg_path, - &ConversationMessage { - id: format!("lm{i}"), - content: format!("legacy kitten message {i}"), - message_type: "text".to_string(), - extra_metadata: serde_json::json!({}), - sender: "user".to_string(), - created_at: ts.clone(), - }, - ) - .unwrap(); - } - - // Cold build on a pre-Stats workspace must index all messages without - // triggering measure_messages_unlocked under CONVERSATION_STORE_LOCK. - let hits = store - .search_cross_thread_messages("kitten", 10, None) - .expect("search on legacy workspace"); - assert_eq!( - hits.len(), - 3, - "all three legacy messages must be found via cold build" - ); - assert!( - hits.iter().any(|h| h.message_id == "lm0"), - "lm0 must be in results" - ); -} - -/// Extends the concurrent-append test to the legacy (no-Stats) workspace shape. -/// -/// Before the fix, `prime_index_if_cold` called `list_threads_unlocked` under -/// the outer lock; for pre-Stats workspaces this triggered a slow -/// `measure_messages_unlocked` + `Stats` append per thread — stalling any -/// concurrent `append_message`. After the fix, `thread_index_unlocked` is -/// used instead (header-only) so the append proceeds concurrently. -#[test] -fn legacy_workspace_cold_rebuild_does_not_block_concurrent_append() { - use std::sync::{mpsc, Arc, Barrier}; - use std::thread; - use std::time::Duration; - - let temp = TempDir::new().unwrap(); - let store = ConversationStore::new(temp.path().to_path_buf()); - let ts = "2023-06-01T00:00:00Z".to_string(); - - // Build a pre-Stats workspace with many threads to make the rebuild - // measurable (each thread has a per-thread JSONL file but no Stats entry). - let root = store.root_dir(); - std::fs::create_dir_all(root.join(THREAD_MESSAGES_DIR)).unwrap(); - - for t in 0..20_usize { - let tid = format!("legacy-t{t}"); - append_jsonl( - &root.join(THREADS_FILENAME), - &ThreadLogEntry::Upsert { - thread_id: tid.clone(), - title: format!("Legacy {t}"), - created_at: ts.clone(), - updated_at: ts.clone(), - parent_thread_id: None, - labels: None, - personality_id: None, - }, - ) - .unwrap(); - let msg_path = store.thread_messages_path(&tid); - for m in 0..50_usize { - append_jsonl( - &msg_path, - &ConversationMessage { - id: format!("lm-{t}-{m}"), - content: format!("legacy content thread {t} message {m}"), - message_type: "text".to_string(), - extra_metadata: serde_json::json!({}), - sender: "user".to_string(), - created_at: ts.clone(), - }, - ) - .unwrap(); - } - } - - // Also need a thread that append_message can target — create it properly - // so it exists in threads.jsonl (still Upsert-only, no Stats). - append_jsonl( - &root.join(THREADS_FILENAME), - &ThreadLogEntry::Upsert { - thread_id: "append-target".to_string(), - title: "Append Target".to_string(), - created_at: ts.clone(), - updated_at: ts.clone(), - parent_thread_id: None, - labels: None, - personality_id: None, - }, - ) - .unwrap(); - - let store_search = store.clone(); - let store_append = store.clone(); - - let barrier = Arc::new(Barrier::new(2)); - let b_search = Arc::clone(&barrier); - let b_append = Arc::clone(&barrier); - - let search_handle = thread::spawn(move || { - b_search.wait(); - store_search.search_cross_thread_messages("legacy content", 5, None) - }); - - let (tx, rx) = mpsc::channel(); - thread::spawn(move || { - b_append.wait(); - let result = store_append.append_message( - "append-target", - ConversationMessage { - id: "concurrent-legacy-append".to_string(), - content: "written during legacy cold rebuild".to_string(), - message_type: "text".to_string(), - extra_metadata: serde_json::json!({}), - sender: "user".to_string(), - created_at: ts, - }, - ); - let _ = tx.send(result); - }); - - let append_result = rx - .recv_timeout(Duration::from_secs(30)) - .expect("append_message blocked — legacy workspace cold rebuild held STORE_LOCK too long"); - assert!( - append_result.is_ok(), - "append failed: {:?}", - append_result.err() - ); - - let search_result = search_handle.join().expect("search thread panicked"); - assert!( - search_result.is_ok(), - "search failed: {:?}", - search_result.err() - ); -} diff --git a/src/openhuman/memory_conversations/tokenize.rs b/src/openhuman/memory_conversations/tokenize.rs deleted file mode 100644 index da7ba3523..000000000 --- a/src/openhuman/memory_conversations/tokenize.rs +++ /dev/null @@ -1,283 +0,0 @@ -//! Multilingual normalization + character n-gram generation for the -//! cross-thread search inverted index. -//! -//! ## Why character n-grams (not word tokens)? -//! -//! The cross-thread search must find substrings *inside* words — querying -//! `cat` should return messages containing `concatenate` or `Kotlin`. A -//! whitespace/word-boundary tokenizer fundamentally cannot do that, and -//! it also breaks down for CJK scripts that have no whitespace at all. -//! Character n-grams sidestep both problems. -//! -//! ## Why a hybrid trigram + CJK-bigram scheme? -//! -//! Trigrams strike a good balance between recall and dictionary size for -//! alphabetic scripts (~26³ ≈ 17k Latin trigrams). For CJK scripts the -//! alphabet is tens of thousands of characters, so character trigrams -//! explode the dictionary while character bigrams stay tractable. The -//! Gemini Deep Research write-up (see PR description) flagged this as the -//! single most important multilingual mitigation. We therefore generate: -//! -//! - **bigrams** for contiguous runs of CJK characters (Han / Hiragana / -//! Katakana / Hangul), -//! - **trigrams** for everything else. -//! -//! Tokens from both schemes coexist in the same posting map. As long as -//! query-time tokenization runs the *same* code path, lookups stay -//! consistent. -//! -//! ## Normalization pipeline -//! -//! Implemented in `normalize()` as a single iterator chain. The order -//! matters — strip-marks must run on the decomposed form, lowercase must -//! run on stripped code points, and the final NFKC re-compose unifies -//! compatibility variants for byte-stable indexing/querying: -//! -//! 1. **NFKD** — decompose so combining marks become standalone code points -//! (Polish ą → `a` + ̨, Arabic kataba+harakat → base letters + marks). -//! 2. **Strip combining marks** — uses `canonical_combining_class` to -//! drop diacritics across all scripts (Polish ą→a, ć→c; Arabic harakat; -//! Hebrew niqqud; combining tone marks; etc.) without needing -//! per-language tables. -//! 3. **Lowercase** — Unicode-aware case folding for cross-alphabet -//! case insensitivity. -//! 4. **NFKC** — re-compose to canonical form (and unify compatibility -//! characters: half/full-width CJK variants, Arabic presentation forms, -//! ligatures) so byte equality lines up at lookup time. -//! 5. **Non-decomposing fold** (`fold_non_decomposing`) — small per-letter -//! table for decorated letters NFKD leaves untouched (Polish ł, German -//! ß, Norwegian ø, Icelandic þ/ð, Latin æ/œ, Turkish ı, Croatian đ, -//! Maltese ħ, Sami ŋ). - -use unicode_normalization::char::canonical_combining_class; -use unicode_normalization::UnicodeNormalization; - -/// Normalize a piece of text for indexing or querying. Idempotent: running -/// the output through `normalize` again yields the same string. -/// -/// After the standard NFKD + strip-combining-marks pass we additionally -/// fold a small set of letters that *look* decorated but don't decompose -/// canonically (so NFKD alone leaves them unchanged). Polish ł/Ł is the -/// motivating case — a Polish user typing `lacka` reasonably expects to -/// find `łącka`. Same idea for German ß, Norwegian ø, Icelandic þ/ð and -/// Latin æ. -pub fn normalize(text: &str) -> String { - let stripped: String = text - .nfkd() // decompose so combining marks become standalone code points - .filter(|c| canonical_combining_class(*c) == 0) - .flat_map(char::to_lowercase) - .nfkc() // re-compose to canonical form for downstream byte equality - .collect(); - fold_non_decomposing(&stripped) -} - -/// Apply per-letter folds for non-decomposing "decorated" letters that -/// NFKD leaves untouched. Run only after lowercase + NFKD so we don't -/// need uppercase entries. -fn fold_non_decomposing(s: &str) -> String { - let mut out = String::with_capacity(s.len()); - for c in s.chars() { - match c { - 'ł' => out.push('l'), - 'ø' => out.push('o'), - 'ß' => out.push_str("ss"), - 'æ' => out.push_str("ae"), - 'œ' => out.push_str("oe"), - 'þ' => out.push_str("th"), - 'ð' => out.push('d'), - 'đ' => out.push('d'), - 'ħ' => out.push('h'), - 'ı' => out.push('i'), // Turkish dotless i - 'ŋ' => out.push('n'), - other => out.push(other), - } - } - out -} - -/// Returns true for code points that should be tokenized as CJK bigrams. -/// -/// Covers Han ideographs (CJK Unified + Ext A + Compatibility), Japanese -/// kana (Hiragana, Katakana), and Hangul (Jamo + precomposed syllables). -/// CJK punctuation and symbols (U+3000..=U+303F) are intentionally -/// excluded — they should be treated as token delimiters, not content. -pub fn is_cjk(c: char) -> bool { - matches!( - c as u32, - 0x3040..=0x309F // Hiragana - | 0x30A0..=0x30FF // Katakana - | 0x3400..=0x4DBF // CJK Unified Ideographs Extension A - | 0x4E00..=0x9FFF // CJK Unified Ideographs - | 0xF900..=0xFAFF // CJK Compatibility Ideographs - | 0x1100..=0x11FF // Hangul Jamo - | 0xAC00..=0xD7AF // Hangul Syllables - ) -} - -/// Tokenize already-normalized text into character n-grams as borrowed -/// slices into `normalized`. -/// -/// Returning `&str` (instead of owned `String`s) keeps the hot search -/// path allocation-free: query-time ngram extraction only needs to look -/// up posting-list keys, never to insert. On the insert side, the index -/// allocates a fresh key only when an ngram is brand-new to the corpus — -/// see `InvertedIndex::insert`. -/// -/// - CJK runs (≥2 chars) → bigrams. -/// - Non-CJK runs (≥3 chars) → trigrams. -/// - Runs shorter than the relevant n are dropped (they cannot be -/// substring-matched against any document containing them anyway, so -/// the Phase 2 verification will catch them via the linear fallback in -/// `InvertedIndex::search`). -/// -/// Word boundaries inside a run do NOT split the n-gram window — we -/// deliberately want substring matches that span punctuation. -pub fn ngrams(normalized: &str) -> Vec<&str> { - let mut out = Vec::new(); - // Capture (byte_offset, is_cjk) per char. Byte offsets let us slice - // `normalized` directly to return `&str` views; the cjk flag drives - // the script-class run partitioning below. - let chars: Vec<(usize, bool)> = normalized - .char_indices() - .map(|(b, c)| (b, is_cjk(c))) - .collect(); - if chars.is_empty() { - return out; - } - let end_byte = normalized.len(); - - // Walk contiguous runs of "same script class" (CJK vs non-CJK) and - // emit the appropriate n-gram size for each run. - let mut i = 0; - while i < chars.len() { - let cjk = chars[i].1; - let mut j = i + 1; - while j < chars.len() && chars[j].1 == cjk { - j += 1; - } - let n = if cjk { 2 } else { 3 }; - if j - i >= n { - for k in i..=j - n { - let start = chars[k].0; - let end = if k + n < chars.len() { - chars[k + n].0 - } else { - end_byte - }; - out.push(&normalized[start..end]); - } - } - i = j; - } - out -} - -#[cfg(test)] -mod tests { - use super::*; - - #[test] - fn normalize_lowercases_ascii() { - assert_eq!(normalize("Hello World"), "hello world"); - } - - #[test] - fn normalize_strips_polish_diacritics() { - // Polish: Kraków, żółć, łąka, plus a Spanish parity case. - // ł/Ł are *not* canonically decomposable (no combining ogonek - // form) so we fold them manually via `fold_non_decomposing`. - assert_eq!(normalize("Kraków"), "krakow"); - assert_eq!(normalize("żółć"), "zolc"); - assert_eq!(normalize("łąka"), "laka"); - assert_eq!(normalize("Mañana"), "manana"); - } - - #[test] - fn normalize_folds_non_decomposing_letters() { - // Letters that NFKD leaves untouched but a user typing without - // the diacritic would still expect to find. - assert_eq!(normalize("Łódź"), "lodz"); - assert_eq!(normalize("Straße"), "strasse"); - assert_eq!(normalize("Bjørn"), "bjorn"); - assert_eq!(normalize("Þórr"), "thorr"); - } - - #[test] - fn normalize_strips_arabic_harakat() { - // Arabic: word "kataba" (he wrote) with harakat marks vs without - let with_marks = "كَتَبَ"; - let without_marks = "كتب"; - assert_eq!(normalize(with_marks), normalize(without_marks)); - } - - #[test] - fn normalize_unifies_cjk_halfwidth_fullwidth() { - // NFKC maps half-width katakana to full-width. - let halfwidth = "ｶﾀｶﾅ"; // half-width - let fullwidth = "カタカナ"; // full-width - assert_eq!(normalize(halfwidth), normalize(fullwidth)); - } - - #[test] - fn normalize_is_idempotent() { - let s = "Café — 東京 — żółć"; - let once = normalize(s); - let twice = normalize(&once); - assert_eq!(once, twice); - } - - #[test] - fn ngrams_emits_trigrams_for_latin() { - let g = ngrams("kitten"); - assert_eq!(g, vec!["kit", "itt", "tte", "ten"]); - } - - #[test] - fn ngrams_emits_bigrams_for_cjk() { - // 日本語 → 日本, 本語 - let g = ngrams("日本語"); - assert_eq!(g, vec!["日本", "本語"]); - } - - #[test] - fn ngrams_mixed_script_splits_at_boundary() { - // "東京tokyo" → CJK run [東京] gives bigram "東京", - // Latin run [tokyo] gives trigrams tok, oky, kyo. - let g = ngrams("東京tokyo"); - assert_eq!(g, vec!["東京", "tok", "oky", "kyo"]); - } - - #[test] - fn ngrams_drops_runs_too_short() { - // "ab東" → Latin run [ab] is only 2 chars → dropped; CJK run [東] - // is only 1 char → dropped. Empty result. - let g = ngrams("ab東"); - assert!(g.is_empty(), "got {:?}", g); - } - - #[test] - fn ngrams_substring_inside_word_is_indexable() { - // After normalize, "Concatenate" → "concatenate" → trigrams include - // "cat". This is the canonical substring-inside-word scenario that - // motivates the character-n-gram scheme over word tokenization. - let normalized = normalize("Concatenate"); - let g = ngrams(&normalized); - assert!(g.contains(&"cat"), "trigrams: {:?}", g); - } - - #[test] - fn ngrams_empty_input_returns_empty() { - assert!(ngrams("").is_empty()); - } - - #[test] - fn is_cjk_classifies_common_scripts() { - assert!(is_cjk('東')); - assert!(is_cjk('あ')); // hiragana - assert!(is_cjk('カ')); // katakana - assert!(is_cjk('한')); // hangul syllable - assert!(!is_cjk('a')); - assert!(!is_cjk('ą')); - assert!(!is_cjk('，')); // CJK punctuation — intentionally NOT cjk - } -} diff --git a/src/openhuman/memory_conversations/types.rs b/src/openhuman/memory_conversations/types.rs deleted file mode 100644 index c7dd11f5d..000000000 --- a/src/openhuman/memory_conversations/types.rs +++ /dev/null @@ -1,143 +0,0 @@ -//! Wire/storage types for the workspace-backed conversation store: threads, -//! messages, create requests, and partial-update patches. - -use serde::{Deserialize, Serialize}; -use serde_json::Value; - -/// A persisted conversation thread, mirroring one entry in `threads.jsonl`. -#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)] -#[serde(rename_all = "camelCase")] -pub struct ConversationThread { - pub id: String, - pub title: String, - #[serde(default, skip_serializing_if = "Option::is_none")] - pub chat_id: Option, - pub is_active: bool, - pub message_count: usize, - pub last_message_at: String, - pub created_at: String, - #[serde(default, skip_serializing_if = "Option::is_none")] - pub parent_thread_id: Option, - #[serde(default)] - pub labels: Vec, - #[serde(default, skip_serializing_if = "Option::is_none")] - pub personality_id: Option, -} - -/// A single message appended to a thread's JSONL log. -#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)] -#[serde(rename_all = "camelCase")] -pub struct ConversationMessage { - pub id: String, - pub content: String, - #[serde(rename = "type")] - pub message_type: String, - #[serde(default)] - pub extra_metadata: Value, - pub sender: String, - pub created_at: String, -} - -/// Input payload to create-or-update a thread via [`super::ensure_thread`]. -#[derive(Debug, Clone, Serialize, Deserialize)] -#[serde(rename_all = "camelCase")] -pub struct CreateConversationThread { - pub id: String, - pub title: String, - pub created_at: String, - #[serde(default)] - pub parent_thread_id: Option, - #[serde(default)] - pub labels: Option>, - #[serde(default)] - pub personality_id: Option, -} - -/// Partial update to apply to a stored message (e.g. rewriting `extraMetadata`). -#[derive(Debug, Clone, Serialize, Deserialize, Default)] -#[serde(rename_all = "camelCase")] -pub struct ConversationMessagePatch { - #[serde(default)] - pub extra_metadata: Option, -} - -/// A single match returned by -/// [`super::store::ConversationStore::search_cross_thread_messages`]. Carries -/// the source `thread_id` so the caller can render provenance into the -/// `[Cross-chat context]` block (issue #1505). -#[derive(Debug, Clone, PartialEq)] -pub struct CrossThreadHit { - pub thread_id: String, - pub message_id: String, - pub role: String, - pub content: String, - pub created_at: String, - pub score: f64, -} - -#[cfg(test)] -mod tests { - use super::*; - use serde_json::json; - - #[test] - fn conversation_thread_serde_uses_camel_case_and_defaults_labels() { - let raw = json!({ - "id": "thread-1", - "title": "Memory", - "chatId": 42, - "isActive": true, - "messageCount": 3, - "lastMessageAt": "2026-05-24T08:00:00Z", - "createdAt": "2026-05-24T07:00:00Z", - "parentThreadId": "parent-1" - }); - - let thread: ConversationThread = serde_json::from_value(raw).unwrap(); - assert_eq!(thread.chat_id, Some(42)); - assert_eq!(thread.parent_thread_id.as_deref(), Some("parent-1")); - assert!(thread.labels.is_empty(), "labels should default to []"); - - let encoded = serde_json::to_value(&thread).unwrap(); - assert_eq!(encoded["chatId"], json!(42)); - assert_eq!(encoded["parentThreadId"], json!("parent-1")); - assert!(encoded.get("chat_id").is_none()); - assert!(encoded.get("parent_thread_id").is_none()); - } - - #[test] - fn conversation_message_patch_defaults_to_no_changes() { - let patch: ConversationMessagePatch = serde_json::from_value(json!({})).unwrap(); - assert!(patch.extra_metadata.is_none()); - - let patch_with_metadata: ConversationMessagePatch = - serde_json::from_value(json!({"extraMetadata": {"source": "mock"}})).unwrap(); - assert_eq!( - patch_with_metadata.extra_metadata, - Some(json!({"source": "mock"})) - ); - } - - #[test] - fn create_thread_optional_fields_roundtrip() { - let create = CreateConversationThread { - id: "thread-2".into(), - title: "Thread".into(), - created_at: "2026-05-24T08:00:00Z".into(), - parent_thread_id: None, - labels: Some(vec!["important".into(), "memory".into()]), - personality_id: None, - }; - - let encoded = serde_json::to_value(&create).unwrap(); - assert_eq!(encoded["labels"], json!(["important", "memory"])); - assert_eq!(encoded["parentThreadId"], Value::Null); - - let decoded: CreateConversationThread = serde_json::from_value(encoded).unwrap(); - assert_eq!( - decoded.labels, - Some(vec!["important".to_string(), "memory".to_string()]) - ); - assert!(decoded.parent_thread_id.is_none()); - } -}