mirror of
https://github.com/tinyhumansai/openhuman.git
synced 2026-07-27 21:08:00 +00:00
refactor(tinycortex): W7 — shim memory_conversations over the crate (#4787)
This commit is contained in:
Steven Enamakel
@@ -1,34 +0,0 @@
|
||||
# conversations
|
||||
|
||||
Workspace-backed conversation thread/message storage. Lives at
|
||||
`<workspace>/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/<thread_id>.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.
|
||||
@@ -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<Hit>
|
||||
//! ```
|
||||
//!
|
||||
//! ## 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<str>`** (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<str>`** (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<u32>`** for ergonomic ordered
|
||||
//! iteration. The Phase 1 intersection is performed against a
|
||||
//! single-allocation `Vec<u32>` 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<str>` 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<str>,
|
||||
message_id: String,
|
||||
role: Arc<str>,
|
||||
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<str>` to shave 8 bytes per entry vs `String`.
|
||||
postings: HashMap<Box<str>, BTreeSet<u32>>,
|
||||
/// Tombstoned: `docs[i] == None` means the message was deleted. We
|
||||
/// keep the slot so existing doc-ids in posting lists stay valid.
|
||||
docs: Vec<Option<DocEntry>>,
|
||||
/// 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<str>` 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<String, Arc<str>>,
|
||||
role_pool: HashMap<String, Arc<str>>,
|
||||
}
|
||||
|
||||
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<u32> = 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<CrossThreadHit> {
|
||||
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<String> = 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<u32>> = 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::<Vec<u32>>(),
|
||||
};
|
||||
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<u32, usize> = 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<Vec<u32>> {
|
||||
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<u32> = 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<str> {
|
||||
if let Some(existing) = self.thread_id_pool.get(thread_id) {
|
||||
return Arc::clone(existing);
|
||||
}
|
||||
let arc: Arc<str> = 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<str> {
|
||||
if let Some(existing) = self.role_pool.get(role) {
|
||||
return Arc::clone(existing);
|
||||
}
|
||||
let arc: Arc<str> = 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<CrossThreadHit> {
|
||||
// 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<u32>, other: &BTreeSet<u32>) {
|
||||
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<str> 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<u32> = [2, 4, 5, 9].into_iter().collect();
|
||||
let mut acc: Vec<u32> = 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<u32> = BTreeSet::new();
|
||||
let mut acc: Vec<u32> = vec![1, 2, 3];
|
||||
intersect_sorted_with_btreeset(&mut acc, &other);
|
||||
assert!(acc.is_empty());
|
||||
}
|
||||
}
|
||||
@@ -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 `<workspace>/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,
|
||||
};
|
||||
|
||||
File diff suppressed because it is too large
Load Diff
File diff suppressed because it is too large
Load Diff
@@ -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
|
||||
}
|
||||
}
|
||||
@@ -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<i64>,
|
||||
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<String>,
|
||||
#[serde(default)]
|
||||
pub labels: Vec<String>,
|
||||
#[serde(default, skip_serializing_if = "Option::is_none")]
|
||||
pub personality_id: Option<String>,
|
||||
}
|
||||
|
||||
/// 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<String>,
|
||||
#[serde(default)]
|
||||
pub labels: Option<Vec<String>>,
|
||||
#[serde(default)]
|
||||
pub personality_id: Option<String>,
|
||||
}
|
||||
|
||||
/// 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<Value>,
|
||||
}
|
||||
|
||||
/// 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());
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user