From 01f77ad1613b4f4298326b49da94803b93ddcb3a Mon Sep 17 00:00:00 2001 From: Steven Enamakel Date: Tue, 29 Sep 2026 20:56:16 +0300 Subject: [PATCH 1/5] feat(conversations): replace global mutex with per-resource locks for finer-grained concurrency Replace the single process-wide mutex that serialized all on-disk mutations with a lock hierarchy scoped to the conversation root, shared metadata, and individual threads. This allows concurrent operations on disjoint threads to proceed without blocking each other, while preserving the existing lock ordering guarantees for operations that touch multiple resources. The change also introduces idempotent message appends for deterministically-generated IDs, a `delete_messages_from` operation for truncating thread message logs, and a coordinated thread listing that repairs metadata under per-thread locks to avoid long stalls. Auto-committed-on: dragonfly Co-authored-by: Medulla --- src/memory/conversations/store.rs | 167 +++++++--- .../conversations/store_concurrency_tests.rs | 104 ++++++ src/memory/conversations/store_index.rs | 197 +++++++----- src/memory/conversations/store_locks.rs | 137 ++++++++ src/memory/conversations/store_ops.rs | 250 +++++++++++---- src/memory/conversations/store_tests.rs | 129 ++++++++ src/memory/conversations/store_tests_late.rs | 295 ++++++++++++++++-- src/memory/conversations/tokenize.rs | 28 +- src/memory/conversations/tokenize_tests.rs | 11 + src/memory/conversations/types.rs | 45 +++ src/memory/conversations/types_tests.rs | 21 ++ 11 files changed, 1173 insertions(+), 211 deletions(-) create mode 100644 src/memory/conversations/store_concurrency_tests.rs create mode 100644 src/memory/conversations/store_locks.rs diff --git a/src/memory/conversations/store.rs b/src/memory/conversations/store.rs index 0a83c8f4..506821b0 100644 --- a/src/memory/conversations/store.rs +++ b/src/memory/conversations/store.rs @@ -4,24 +4,37 @@ //! `threads/.jsonl` so arbitrary provider ids remain //! filesystem-safe. //! -//! All on-disk mutations serialise through a single process-wide mutex so -//! concurrent RPC handlers don't interleave writes. +//! On-disk mutations synchronize at the narrowest safe scope: lifecycle per +//! conversation root, shared metadata per root, and messages per thread. //! -//! Ported from OpenHuman's `memory_conversations::store`. The behaviour is -//! preserved; the only mechanical changes are dependency substitutions that -//! keep this crate's `Cargo.toml` untouched: +//! This file is the store as it came back from the memory engine (#5560), +//! unchanged apart from the one constructor described below. The three +//! dependency substitutions the round trip introduced — [`std::sync::LazyLock`] +//! for the statics, the local [`hex_encode`] for per-thread filenames, and the +//! hand-rolled temp-write in [`rewrite_jsonl`] — are all kept: they produce the +//! bytes and the paths that every existing transcript already lives at, and a +//! move must not rewrite those. See [`super`]'s module docs for the full +//! accounting of what the trip cost. //! -//! - `once_cell::sync::Lazy` → `std::sync::LazyLock` for the process-wide -//! statics. -//! - `hex::encode` → the local [`hex_encode`] helper for per-thread filenames. -//! - `tempfile::NamedTempFile` (a dev-only dependency here) → a write-to-temp + -//! atomic-rename in [`rewrite_jsonl`]. -//! - OpenHuman's `log`/`tracing` diagnostics are dropped (this crate has no -//! logging facade wired up). +//! # The one thing that did change: `from_config` //! -//! To respect the repo's 500-line-per-file limit the `impl ConversationStore` -//! is split across two child modules — [`ops`] (the public CRUD + search API) -//! and [`index`] (private thread-folding and inverted-index helpers). Both are +//! The engine's [`ConversationStore`] carried a second constructor, +//! `from_config(&MemoryConfig)`, whose whole body was +//! `Self::new(config.workspace.clone())`. It is gone rather than translated, +//! and that is the point of the move: it was this module's *only* coupling to +//! the rest of the engine, and re-expressing it here would either drag +//! `MemoryConfig` back in or add a second name for a constructor that already +//! exists. [`ConversationStore::new`] takes the workspace directory, every +//! caller in this host already has one, and no caller ever used `from_config` +//! — so nothing needed a translation and the derived on-disk root +//! (`/memory/conversations`, see `ConversationStore::root_dir` in +//! `store_index.rs`) is byte-identical either way. +//! +//! # File split +//! +//! To respect the repo's file-size limit the `impl ConversationStore` is split +//! across two child modules — [`ops`] (the public CRUD + search API) and +//! [`index`] (private thread-folding and inverted-index helpers). Both are //! descendant modules of `store`, so they share access to the private statics, //! constants, log-entry enum, and JSONL helpers defined here. @@ -44,6 +57,8 @@ mod ops; #[path = "store_index.rs"] mod index; +#[path = "store_locks.rs"] +mod locks; /// Filename of the append-only thread metadata log, relative to the /// `memory/conversations` root. @@ -52,10 +67,6 @@ pub(super) const THREADS_FILENAME: &str = "threads.jsonl"; /// per-thread message JSONL files, named `.jsonl`. pub(super) const THREAD_MESSAGES_DIR: &str = "threads"; -/// Serialises every on-disk mutation so concurrent handlers can't interleave -/// writes to `threads.jsonl` or the per-thread message logs. -static CONVERSATION_STORE_LOCK: LazyLock> = LazyLock::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 @@ -67,28 +78,26 @@ static CONVERSATION_STORE_LOCK: LazyLock> = LazyLock::new(|| Mutex::ne /// /// # 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). +/// Every operation first takes the root lifecycle lock. Message operations +/// then take their per-thread lock and briefly take the metadata lock when +/// they must inspect or append `threads.jsonl`. When metadata and the index +/// cache are both needed, metadata is acquired first. No code may acquire a +/// thread or metadata lock while holding `CONVERSATION_INDEX_CACHE`. /// /// `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` +/// metadata and index locks only momentarily, and always in the metadata +/// → `CONVERSATION_INDEX_CACHE` order above: while holding metadata +/// 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. +/// releases metadata before reading each transcript under that thread's own +/// lock and finally acquires `CONVERSATION_INDEX_CACHE` alone to insert the +/// built index. It never holds both across the slow JSONL walk. /// /// `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. +/// stall under the shared metadata lock that this design was built to avoid. static CONVERSATION_INDEX_CACHE: LazyLock>> = LazyLock::new(|| Mutex::new(HashMap::new())); @@ -104,19 +113,39 @@ pub struct ConversationPurgeStats { /// Workspace-rooted handle that reads and writes the JSONL conversation log. #[derive(Debug, Clone)] pub struct ConversationStore { - workspace_dir: PathBuf, + root_dir: PathBuf, + locks: std::sync::Arc, } impl ConversationStore { /// Construct a store rooted at the given workspace directory. + /// + /// This is the only constructor. The conversation root is derived from + /// `workspace_dir` alone (`/memory/conversations`, see + /// `root_dir` in `store_index.rs`), so the caller's workspace is the whole + /// input and there is nothing for a config type to add. pub fn new(workspace_dir: PathBuf) -> Self { - Self { workspace_dir } + let root = locks::normalized_root(&workspace_dir.join("memory").join("conversations")); + let locks = locks::for_root(&root); + Self { + root_dir: root, + locks, + } + } + + #[cfg(test)] + pub(super) fn lock_identity_for_test(&self) -> usize { + std::sync::Arc::as_ptr(&self.locks) as usize } - /// Construct a store rooted at the engine's configured workspace - /// ([`MemoryConfig::workspace`](crate::memory::config::MemoryConfig)). - pub fn from_config(config: &crate::memory::config::MemoryConfig) -> Self { - Self::new(config.workspace.clone()) + #[cfg(test)] + pub(super) fn thread_lock_identity_for_test(&self, thread_id: &str) -> usize { + std::sync::Arc::as_ptr(&self.locks.thread(thread_id)) as usize + } + + #[cfg(test)] + pub(super) fn thread_lock_count_for_test(&self) -> usize { + self.locks.thread_count() } } @@ -220,8 +249,13 @@ pub(super) fn normalize_labels(labels: Vec) -> Vec { } /// Lowercase hex-encode bytes — used to derive a filesystem-safe per-thread -/// messages filename from an arbitrary thread id. Replaces OpenHuman's use of -/// the `hex` crate, which is not a dependency of this crate. +/// messages filename from an arbitrary thread id. +/// +/// This exists because the memory engine had no `hex` dependency; this crate +/// does, so `hex::encode` would work here. It is kept anyway: this is the +/// function that decides which file a user's transcript is read from and +/// written to, and swapping it is only provably safe, never *obviously* safe. +/// The three lines are cheaper than the argument. pub(super) fn hex_encode(bytes: &[u8]) -> String { const HEX: [u8; 16] = *b"0123456789abcdef"; let mut out = String::with_capacity(bytes.len() * 2); @@ -258,6 +292,36 @@ where Ok(items) } +/// Find one message in a thread's JSONL log by id, without materializing the +/// whole transcript. +/// +/// Only the lines whose raw text carries the quoted id are deserialized, so a +/// lookup costs one parse rather than one per stored message; a line that +/// merely quotes the id inside its own content is rejected by the `id` check. +/// Mirrors [`read_jsonl`]'s tolerance of blank and corrupt lines. +pub(super) fn find_message_by_id( + path: &Path, + id: &str, +) -> Result, String> { + if !path.exists() { + return Ok(None); + } + let needle = serde_json::to_string(id).map_err(|e| format!("encode message id {id}: {e}"))?; + let file = File::open(path).map_err(|e| format!("open {}: {e}", path.display()))?; + for (line_no, line) in BufReader::new(file).lines().enumerate() { + let line = + line.map_err(|e| format!("read {} line {}: {e}", path.display(), line_no + 1))?; + if !line.contains(&needle) { + continue; + } + match serde_json::from_str::(&line) { + Ok(message) if message.id == id => return Ok(Some(message)), + _ => continue, + } + } + Ok(None) +} + /// Append one serialized value as a JSONL line, fsync'd before returning. pub(super) fn append_jsonl(path: &Path, value: &T) -> Result<(), String> where @@ -284,9 +348,15 @@ where /// Atomically rewrite `path` with `values`, one JSON object per line. /// /// Writes to a sibling temp file then renames over the target so a crash -/// mid-write never leaves a partially-written transcript. Replaces -/// OpenHuman's `tempfile::NamedTempFile`, which is a dev-only dependency in -/// this crate. +/// mid-write never leaves a partially-written transcript. +/// +/// This hand-rolls what OpenHuman originally got from +/// `tempfile::NamedTempFile`, because the memory engine carried `tempfile` as +/// a dev-dependency only. This crate has it in full, so the substitution is no +/// longer forced — but reverting it would change the temp file's name, its +/// permissions, and which side deletes it when a write fails. That is a change +/// to the crash-safety path for a user's transcript, and it belongs in a +/// change that is about that, not in a module move. pub(super) fn rewrite_jsonl(path: &Path, values: &[T]) -> Result<(), String> where T: serde::Serialize, @@ -381,6 +451,15 @@ pub fn update_message( ConversationStore::new(workspace_dir).update_message(thread_id, message_id, patch) } +/// Free-function shim around [`ConversationStore::delete_messages_from`]. +pub fn delete_messages_from( + workspace_dir: PathBuf, + thread_id: &str, + message_id: &str, +) -> Result, String> { + ConversationStore::new(workspace_dir).delete_messages_from(thread_id, message_id) +} + /// Free-function shim around [`ConversationStore::purge_threads`]. pub fn purge_threads(workspace_dir: PathBuf) -> Result { ConversationStore::new(workspace_dir).purge_threads() diff --git a/src/memory/conversations/store_concurrency_tests.rs b/src/memory/conversations/store_concurrency_tests.rs new file mode 100644 index 00000000..b4fa6ad2 --- /dev/null +++ b/src/memory/conversations/store_concurrency_tests.rs @@ -0,0 +1,104 @@ +use super::*; + +#[test] +fn one_hundred_agent_threads_append_without_loss_or_corruption() { + use std::sync::{Arc, Barrier}; + + let temp = TempDir::new().unwrap(); + let store = ConversationStore::new(temp.path().to_path_buf()); + let created_at = "2026-09-10T00:00:00Z".to_string(); + + for index in 0..100 { + store + .ensure_thread(CreateConversationThread { + parent_thread_id: None, + id: format!("agent-{index}"), + title: format!("Agent {index}"), + created_at: created_at.clone(), + labels: None, + personality_id: None, + }) + .unwrap(); + } + + let barrier = Arc::new(Barrier::new(101)); + let mut writers = Vec::with_capacity(100); + for index in 0..100 { + let store = store.clone(); + let barrier = Arc::clone(&barrier); + let created_at = created_at.clone(); + writers.push(std::thread::spawn(move || { + barrier.wait(); + store.append_message( + &format!("agent-{index}"), + ConversationMessage { + id: format!("message-{index}"), + content: format!("reply from agent {index}"), + message_type: "text".to_string(), + extra_metadata: serde_json::json!({}), + sender: "assistant".to_string(), + created_at, + }, + ) + })); + } + barrier.wait(); + + for writer in writers { + writer.join().expect("writer panicked").expect("append"); + } + + let threads = store.list_threads().unwrap(); + assert_eq!(threads.len(), 100); + assert!(threads.iter().all(|thread| thread.message_count == 1)); + for index in 0..100 { + let messages = store.get_messages(&format!("agent-{index}")).unwrap(); + assert_eq!(messages.len(), 1); + assert_eq!(messages[0].id, format!("message-{index}")); + } +} + +#[test] +fn one_unreadable_transcript_does_not_stop_other_repairs() { + let temp = TempDir::new().unwrap(); + let store = ConversationStore::new(temp.path().to_path_buf()); + let created_at = "2026-09-10T00:00:00Z".to_string(); + + for id in ["a-unreadable", "z-readable"] { + store + .ensure_thread(CreateConversationThread { + parent_thread_id: None, + id: id.to_string(), + title: id.to_string(), + created_at: created_at.clone(), + labels: None, + personality_id: None, + }) + .unwrap(); + } + + std::fs::create_dir_all(store.thread_messages_path("a-unreadable")).unwrap(); + append_jsonl( + &store.thread_messages_path("z-readable"), + &ConversationMessage { + id: "message".to_string(), + content: "recover me".to_string(), + message_type: "text".to_string(), + extra_metadata: serde_json::json!({}), + sender: "assistant".to_string(), + created_at, + }, + ) + .unwrap(); + + let threads = store.list_threads().unwrap(); + let readable = threads + .iter() + .find(|thread| thread.id == "z-readable") + .unwrap(); + assert_eq!(readable.message_count, 1); + + let folded = store.thread_index_unlocked().unwrap(); + assert_eq!(folded["z-readable"].message_count, Some(1)); + assert_eq!(folded["a-unreadable"].message_count, None); +} diff --git a/src/memory/conversations/store_index.rs b/src/memory/conversations/store_index.rs index ce443220..854178e1 100644 --- a/src/memory/conversations/store_index.rs +++ b/src/memory/conversations/store_index.rs @@ -4,7 +4,7 @@ //! API in `store_ops.rs` (and the unit tests) can call it, but it stays out of //! the crate's public surface. -use std::collections::BTreeMap; +use std::collections::{BTreeMap, HashSet}; use std::fs::{self, File}; use std::path::PathBuf; @@ -13,52 +13,59 @@ use super::super::types::{ConversationMessage, ConversationThread}; use super::{ append_jsonl, hex_encode, infer_labels, normalize_labels, read_jsonl, ConversationPurgeStats, ConversationStore, ThreadIndexEntry, ThreadLogEntry, CONVERSATION_INDEX_CACHE, - CONVERSATION_STORE_LOCK, THREADS_FILENAME, THREAD_MESSAGES_DIR, + THREADS_FILENAME, THREAD_MESSAGES_DIR, }; impl ConversationStore { - /// 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. + /// If no index entry exists for this workspace, serialize cold builders, + /// start a short-lived append journal, snapshot the live thread IDs under + /// the root metadata lock, release it, and read every JSONL file under its + /// per-thread lock. Publication folds in every append journaled during the + /// scan while holding metadata, so it cannot publish stale and never needs + /// to retry under sustained write traffic. /// /// After this call returns, `with_index` will always find a warm entry and /// will not re-enter `populate_index_unlocked`. pub(super) fn prime_index_if_cold(&self) -> Result<(), String> { + self.prime_index_if_cold_with_hook(|| {}) + } + + /// `after_scan` is a deterministic test seam for mutations that land + /// after file reads but before publication. Production always passes a + /// no-op closure through [`Self::prime_index_if_cold`]. + pub(super) fn prime_index_if_cold_with_hook( + &self, + mut after_scan: impl FnMut(), + ) -> 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 _build = self.locks.index_build.lock(); + if CONVERSATION_INDEX_CACHE.lock().contains_key(&key) { + return Ok(()); + } + self.locks.begin_index_build(); + + // This is header-only O(threads) work. Do not use + // `list_threads_unlocked`: legacy workspaces can make that measure and + // append stats for every thread while metadata is held. 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(()); + let _metadata = self.locks.metadata.lock(); + match self.thread_index_unlocked() { + Ok(index) => index.into_keys().collect(), + Err(error) => { + self.locks.cancel_index_build(); + return Err(error); + } } - 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. A message - // appended during this window stays absent from the in-memory index - // until the next cold rebuild — the accepted tradeoff for issue #2849. + let mut idx = InvertedIndex::new(); for thread_id in &thread_ids { + let thread_lock = self.locks.thread(thread_id); + let _thread = thread_lock.lock(); let path = self.thread_messages_path(thread_id); if !path.exists() { continue; @@ -69,64 +76,36 @@ impl ConversationStore { } } } - // Insert only if the key is still absent — a concurrent prime that - // finished first wins; ours is discarded. - { - let mut cache = CONVERSATION_INDEX_CACHE.lock(); - cache.entry(key).or_insert(idx); + after_scan(); + + // Append finalization takes metadata too. Therefore every append is + // either already in the journal drained here, or waits until after + // publication and updates the now-warm cache directly. + let _metadata = self.locks.metadata.lock(); + for (thread_id, message) in self.locks.finish_index_build() { + idx.insert(&thread_id, message); } + CONVERSATION_INDEX_CACHE.lock().insert(key, idx); 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. - pub(super) fn with_index( + /// Acquire an index that the caller has already warmed with + /// [`Self::prime_index_if_cold`] and run `f` against it. The only + /// production caller, `search_cross_thread_messages`, holds the root's + /// lifecycle read guard across both calls, so purge cannot remove the + /// entry between priming and access. + pub(super) fn with_primed_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"); + let idx = cache + .get_mut(&key) + .ok_or_else(|| "conversation index missing after required prime".to_string())?; 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. - pub(super) 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. - 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(_) => continue, - }; - for msg in messages { - idx.insert(&thread.id, msg); - } - } - Ok(()) - } - /// Ensure the `memory/conversations` directory tree (and an empty /// `threads.jsonl`) exists, returning the conversation root. pub(super) fn ensure_root(&self) -> Result { @@ -144,7 +123,7 @@ impl ConversationStore { /// Absolute path to this workspace's `memory/conversations` root. pub(super) fn root_dir(&self) -> PathBuf { - self.workspace_dir.join("memory").join("conversations") + self.root_dir.clone() } /// Absolute path to a thread's per-thread messages JSONL file. The thread @@ -204,6 +183,66 @@ impl ConversationStore { } } + Ok(Self::threads_from_index(index)) + } + + /// Fold and repair thread metadata without inverting the lock order used + /// by message mutations. Recovery reads take the target thread lock before + /// metadata; a newly-created thread discovered between passes is handled + /// by the next iteration. + pub(super) fn list_threads_coordinated(&self) -> Result, String> { + let mut unreadable = HashSet::new(); + loop { + let (index, missing) = { + let _metadata = self.locks.metadata.lock(); + let index = self.thread_index_unlocked()?; + let missing = index + .iter() + .filter(|(_, entry)| { + entry.message_count.is_none() || entry.last_message_at.is_none() + }) + .filter(|(thread_id, _)| !unreadable.contains(*thread_id)) + .map(|(thread_id, _)| thread_id.clone()) + .collect::>(); + (index, missing) + }; + if missing.is_empty() { + return Ok(Self::threads_from_index(index)); + } + + for thread_id in missing { + let thread_lock = self.locks.thread(&thread_id); + let _thread = thread_lock.lock(); + let _metadata = self.locks.metadata.lock(); + let index = self.thread_index_unlocked()?; + let Some(entry) = index.get(&thread_id) else { + continue; + }; + if entry.message_count.is_some() && entry.last_message_at.is_some() { + continue; + } + let Ok((count, last_message_at)) = self.measure_messages_unlocked(&thread_id) + else { + // Quarantine this thread for this invocation so it neither + // blocks repairs for later threads nor causes the outer + // loop to retry it forever. A future list call retries it. + unreadable.insert(thread_id); + continue; + }; + let resolved_last = last_message_at.unwrap_or_else(|| entry.created_at.clone()); + append_jsonl( + &self.ensure_root()?.join(THREADS_FILENAME), + &ThreadLogEntry::Stats { + thread_id, + message_count: count, + last_message_at: resolved_last, + }, + )?; + } + } + } + + fn threads_from_index(index: BTreeMap) -> Vec { let mut threads: Vec = index .iter() .map(|(thread_id, entry)| { @@ -231,7 +270,7 @@ impl ConversationStore { .cmp(×tamp_millis(&a.last_message_at)) .then_with(|| timestamp_millis(&b.created_at).cmp(×tamp_millis(&a.created_at))) }); - Ok(threads) + threads } /// Count messages and find the newest timestamp by reading the per-thread diff --git a/src/memory/conversations/store_locks.rs b/src/memory/conversations/store_locks.rs new file mode 100644 index 00000000..651d81e5 --- /dev/null +++ b/src/memory/conversations/store_locks.rs @@ -0,0 +1,137 @@ +//! Lock registry for the JSONL conversation store. +//! +//! A root owns shared metadata (`threads.jsonl`) and many independent message +//! files. Keeping those synchronization scopes separate lets unrelated agent +//! sessions write their message files concurrently while preserving atomic +//! metadata appends and purge semantics. + +use std::collections::HashMap; +use std::path::{Path, PathBuf}; +use std::sync::{Arc, LazyLock, Weak}; + +use parking_lot::{Mutex, RwLock}; + +use super::super::types::ConversationMessage; + +#[derive(Debug, Default)] +pub(super) struct StoreLocks { + /// Ordinary operations take a read guard; purge takes the write guard. + pub(super) lifecycle: RwLock<()>, + /// Serializes reads and appends of the root's shared `threads.jsonl`. + pub(super) metadata: Mutex<()>, + /// Only one cold scan may construct this root's in-memory index. + pub(super) index_build: Mutex<()>, + threads: Mutex>>>, + /// Appends completed while a cold scan is in flight. `None` means no scan + /// is active, so the warm-cache path alone owns index maintenance. + pending_index_appends: Mutex>>, +} + +impl StoreLocks { + pub(super) fn thread(&self, thread_id: &str) -> Arc> { + let mut locks = self.threads.lock(); + locks.retain(|_, lock| lock.strong_count() > 0); + if let Some(lock) = locks.get(thread_id).and_then(Weak::upgrade) { + return lock; + } + let lock = Arc::new(Mutex::new(())); + locks.insert(thread_id.to_string(), Arc::downgrade(&lock)); + lock + } + + pub(super) fn begin_index_build(&self) { + let previous = self.pending_index_appends.lock().replace(Vec::new()); + debug_assert!(previous.is_none(), "index builds must be serialized"); + } + + pub(super) fn record_index_append(&self, thread_id: &str, message: &ConversationMessage) { + if let Some(pending) = self.pending_index_appends.lock().as_mut() { + pending.push((thread_id.to_string(), message.clone())); + } + } + + pub(super) fn finish_index_build(&self) -> Vec<(String, ConversationMessage)> { + self.pending_index_appends.lock().take().unwrap_or_default() + } + + pub(super) fn cancel_index_build(&self) { + self.pending_index_appends.lock().take(); + } + + /// Call only while holding the lifecycle write guard. + pub(super) fn remove_thread(&self, thread_id: &str) { + self.threads.lock().remove(thread_id); + } + + /// Call only while holding the lifecycle write guard. + pub(super) fn clear_threads(&self) { + self.threads.lock().clear(); + } + + #[cfg(test)] + pub(super) fn thread_count(&self) -> usize { + self.threads + .lock() + .values() + .filter(|lock| lock.strong_count() > 0) + .count() + } +} + +/// Separate `ConversationStore::new` calls for the same root must coordinate. +/// Weak entries avoid retaining one lock set for every temporary workspace a +/// long-running process has ever touched. +static ROOTS: LazyLock>>> = + LazyLock::new(|| Mutex::new(HashMap::new())); + +pub(super) fn for_root(root: &Path) -> Arc { + let root = normalized_root(root); + let mut roots = ROOTS.lock(); + // A process may open many ephemeral workspaces over its lifetime. The + // weak value avoids retaining each lock set; pruning dead values here also + // prevents their path keys from making the registry itself grow forever. + roots.retain(|_, locks| locks.strong_count() > 0); + if let Some(existing) = roots.get(&root).and_then(Weak::upgrade) { + return existing; + } + let locks = Arc::new(StoreLocks::default()); + roots.insert(root, Arc::downgrade(&locks)); + locks +} + +/// Resolve aliases even before the conversation directory itself exists. +/// Canonicalizing the nearest existing ancestor handles symlinks and `..`; +/// the missing suffix is then appended without touching the filesystem. +pub(super) fn normalized_root(root: &Path) -> PathBuf { + let absolute; + let root = if root.is_absolute() { + root + } else { + absolute = std::env::current_dir() + .map(|cwd| cwd.join(root)) + .unwrap_or_else(|_| root.to_path_buf()); + &absolute + }; + if let Ok(canonical) = root.canonicalize() { + return canonical; + } + + let mut suffix = Vec::new(); + let mut ancestor = root; + loop { + if let Ok(canonical) = ancestor.canonicalize() { + return suffix + .iter() + .rev() + .fold(canonical, |path, component| path.join(component)); + } + let Some(name) = ancestor.file_name() else { + return root.to_path_buf(); + }; + suffix.push(name.to_os_string()); + let Some(parent) = ancestor.parent() else { + return root.to_path_buf(); + }; + ancestor = parent; + } +} diff --git a/src/memory/conversations/store_ops.rs b/src/memory/conversations/store_ops.rs index 1189cb1d..215eca8e 100644 --- a/src/memory/conversations/store_ops.rs +++ b/src/memory/conversations/store_ops.rs @@ -6,12 +6,12 @@ use std::fs; use super::super::types::{ - ConversationMessage, ConversationMessagePatch, ConversationThread, CreateConversationThread, - CrossThreadHit, + is_deterministic_message_id, ConversationMessage, ConversationMessagePatch, ConversationThread, + CreateConversationThread, CrossThreadHit, }; use super::{ - append_jsonl, normalize_labels, read_jsonl, rewrite_jsonl, ConversationPurgeStats, - ConversationStore, ThreadLogEntry, CONVERSATION_INDEX_CACHE, CONVERSATION_STORE_LOCK, + append_jsonl, find_message_by_id, normalize_labels, read_jsonl, rewrite_jsonl, + ConversationPurgeStats, ConversationStore, ThreadLogEntry, CONVERSATION_INDEX_CACHE, THREADS_FILENAME, }; @@ -21,7 +21,10 @@ impl ConversationStore { &self, request: CreateConversationThread, ) -> Result { - let _guard = CONVERSATION_STORE_LOCK.lock(); + let _lifecycle = self.locks.lifecycle.read(); + let thread_lock = self.locks.thread(&request.id); + let _thread = thread_lock.lock(); + let _metadata = self.locks.metadata.lock(); let root = self.ensure_root()?; let threads_path = root.join(THREADS_FILENAME); let now = request.created_at.clone(); @@ -44,15 +47,20 @@ impl ConversationStore { /// 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() + let _lifecycle = self.locks.lifecycle.read(); + self.list_threads_coordinated() } /// 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 _lifecycle = self.locks.lifecycle.read(); + let thread_lock = self.locks.thread(thread_id); + let _thread = thread_lock.lock(); + { + let _metadata = self.locks.metadata.lock(); + if !self.thread_exists_unlocked(thread_id)? { + return Ok(Vec::new()); + } } let path = self.thread_messages_path(thread_id); if !path.exists() { @@ -82,31 +90,29 @@ impl ConversationStore { /// /// # Lock strategy (issue #2849) /// - /// **Fast path (warm cache):** acquires only `CONVERSATION_INDEX_CACHE` - /// — no outer store lock — and returns immediately. + /// **Fast path (warm cache):** acquires the root lifecycle read guard and + /// `CONVERSATION_INDEX_CACHE`, with no metadata or thread lock. /// /// **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 other store - /// operations 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 until the cache is evicted and - /// rebuilt — an accepted tradeoff for issue #2849. + /// the root metadata lock (brief), then releases it before reading each + /// JSONL file under its per-thread lock. This avoids blocking unrelated + /// threads during the potentially-long rebuild. Appends completed during + /// that scan are journaled and folded into the index atomically at + /// publication. 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 + // Warm the index without the metadata lock so concurrent // append_message / get_messages calls are not stalled during the // cold JSONL rebuild. After this returns the cache entry is // guaranteed to exist, so with_index will not trigger a second // rebuild. + let _lifecycle = self.locks.lifecycle.read(); self.prime_index_if_cold()?; - - let _guard = CONVERSATION_STORE_LOCK.lock(); - self.with_index(|idx| idx.search(query, limit, exclude_thread_id)) + self.with_primed_index(|idx| idx.search(query, limit, exclude_thread_id)) } /// Append a message to the thread's JSONL file. Errors if the thread is missing. @@ -115,16 +121,42 @@ impl ConversationStore { /// row, then a compact `MessageAppended` stat entry. Thread reads reconcile /// that stat trail against the message file, repairing a crash between the /// two appends. + /// + /// Idempotent for the ids the core mints deterministically + /// ([`is_deterministic_message_id`]): when the thread already holds a row + /// with that id, nothing is written (no message row, no stat bump, no index + /// insert) and the stored row is returned exactly as a fresh append would + /// return its input. Two writers can legitimately persist the same reply — + /// background delivery and the client that + /// also persists the `chat_done` it announced (#5933) — and a thread must + /// never carry two messages under one id (the frontend keys React and + /// assistant-ui resources by it). + /// + /// The lookup is deliberately narrow. Every other id in the store is + /// UUID-fresh by construction and cannot be re-presented, so it must not + /// pay to have that verified: a lookup on *every* append would put a scan + /// of the thread's transcript on every hot write and make growing a thread + /// quadratic. 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 _lifecycle = self.locks.lifecycle.read(); + let thread_lock = self.locks.thread(thread_id); + let _thread = thread_lock.lock(); + { + let _metadata = self.locks.metadata.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 is_deterministic_message_id(&message.id) { + if let Some(existing) = find_message_by_id(&path, &message.id)? { + return Ok(existing); + } + } if let Some(parent) = path.parent() { fs::create_dir_all(parent) .map_err(|e| format!("create conversation dir {}: {e}", parent.display()))?; @@ -133,23 +165,25 @@ impl ConversationStore { // 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. { + let _metadata = self.locks.metadata.lock(); + // The transcript row is already durable. Publish it to an active + // cold-build journal and any warm cache before the derived stats + // append, which may fail independently. + self.locks.record_index_append(thread_id, &message); let mut cache = CONVERSATION_INDEX_CACHE.lock(); if let Some(idx) = cache.get_mut(&self.root_dir()) { idx.insert(thread_id, message.clone()); } + drop(cache); + 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(), + }, + )?; } Ok(message) } @@ -161,7 +195,10 @@ impl ConversationStore { title: &str, updated_at: &str, ) -> Result { - let _guard = CONVERSATION_STORE_LOCK.lock(); + let _lifecycle = self.locks.lifecycle.read(); + let thread_lock = self.locks.thread(thread_id); + let _thread = thread_lock.lock(); + let _metadata = self.locks.metadata.lock(); let index = self.thread_index_unlocked()?; let entry = index .get(thread_id) @@ -190,7 +227,10 @@ impl ConversationStore { labels: Vec, updated_at: &str, ) -> Result { - let _guard = CONVERSATION_STORE_LOCK.lock(); + let _lifecycle = self.locks.lifecycle.read(); + let thread_lock = self.locks.thread(thread_id); + let _thread = thread_lock.lock(); + let _metadata = self.locks.metadata.lock(); let index = self.thread_index_unlocked()?; let entry = index .get(thread_id) @@ -220,7 +260,9 @@ impl ConversationStore { message_id: &str, patch: ConversationMessagePatch, ) -> Result { - let _guard = CONVERSATION_STORE_LOCK.lock(); + let _lifecycle = self.locks.lifecycle.read(); + let thread_lock = self.locks.thread(thread_id); + let _thread = thread_lock.lock(); let path = self.thread_messages_path(thread_id); let mut messages = read_jsonl::(&path)?; let mut updated: Option = None; @@ -239,33 +281,112 @@ impl ConversationStore { Ok(updated) } + /// Truncate a thread's message log at `message_id`: drop that message and + /// every message after it (append order == chronological order), keeping + /// everything before it. Backs `threads.edit_message` / `threads.regenerate` + /// (edit/regenerate rewrite the tail of a conversation, never the middle). + /// + /// Returns the number of messages removed, or `Ok(None)` if `message_id` + /// is not present in the thread (a stale/unknown cut point — the caller + /// should treat this as "nothing to truncate", not silently drop the + /// whole log). + /// + /// Evicts the thread from the cross-thread search index the same way + /// [`Self::delete_thread`] does: the index has no per-message removal, so + /// the conservative move is to drop the whole thread's postings rather + /// than search a stale truncated message back into a hit. The next + /// cross-thread search that touches this thread re-primes it from the + /// (now-truncated) file on disk. + pub fn delete_messages_from( + &self, + thread_id: &str, + message_id: &str, + ) -> Result, String> { + let _lifecycle = self.locks.lifecycle.read(); + let thread_lock = self.locks.thread(thread_id); + let _thread = thread_lock.lock(); + let path = self.thread_messages_path(thread_id); + let messages = read_jsonl::(&path)?; + let Some(cut_at) = messages.iter().position(|m| m.id == message_id) else { + return Ok(None); + }; + let removed = messages.len() - cut_at; + let kept = &messages[..cut_at]; + rewrite_jsonl(&path, kept)?; + // The compact stat trail in `threads.jsonl` (`MessageAppended`/ + // `Stats`) only ever grows via `append_message`'s increment — it has + // no notion of a truncation. Append an authoritative `Stats` snapshot + // now so `list_threads`'s `message_count`/`last_message_at` reflect + // the post-truncation file immediately, instead of staying + // overcounted until this thread is next quarantined as unreadable + // and rescanned (which never happens on its own — see + // `list_threads_coordinated`, which only remeasures a `None` count). + let last_message_at = kept.last().map(|m| m.created_at.clone()); + { + let _metadata = self.locks.metadata.lock(); + let resolved_last = match last_message_at { + Some(ts) => ts, + None => self + .thread_summary_unlocked(thread_id)? + .map(|t| t.created_at) + .unwrap_or_default(), + }; + append_jsonl( + &self.ensure_root()?.join(THREADS_FILENAME), + &ThreadLogEntry::Stats { + thread_id: thread_id.to_string(), + message_count: kept.len(), + last_message_at: resolved_last, + }, + )?; + } + { + let mut cache = CONVERSATION_INDEX_CACHE.lock(); + if let Some(idx) = cache.get_mut(&self.root_dir()) { + idx.remove_thread(thread_id); + } + } + Ok(Some(removed)) + } + /// 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() - )); + // Deletion also evicts the thread's lock entry. Exclusive lifecycle + // ownership prevents a new operation from retaining the old lock + // while the registry entry is replaced. + let _lifecycle = self.locks.lifecycle.write(); + let thread_lock = self.locks.thread(thread_id); + let _thread = thread_lock.lock(); + { + let _metadata = self.locks.metadata.lock(); + if !self.thread_exists_unlocked(thread_id)? { + self.locks.remove_thread(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); + let remove_result = match fs::remove_file(&messages_path) { + Ok(()) => Ok(()), + Err(error) if error.kind() == std::io::ErrorKind::NotFound => Ok(()), + Err(error) => Err(format!( + "delete conversation messages {}: {error}", + messages_path.display() + )), + }; + // Evict on every path after the tombstone is durable, including a + // filesystem deletion error. The lifecycle write guard prevents a new + // operation from observing a replacement lock before this one drops. + self.locks.remove_thread(thread_id); // Drop every indexed message for this thread so future searches // don't surface stale content. { @@ -274,12 +395,14 @@ impl ConversationStore { idx.remove_thread(thread_id); } } + remove_result?; 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 _lifecycle = self.locks.lifecycle.write(); + let _metadata = self.locks.metadata.lock(); let stats = self.purge_stats_unlocked()?; let root = self.root_dir(); if root.exists() { @@ -293,6 +416,7 @@ impl ConversationStore { let mut cache = CONVERSATION_INDEX_CACHE.lock(); cache.remove(&root); } + self.locks.clear_threads(); Ok(stats) } } diff --git a/src/memory/conversations/store_tests.rs b/src/memory/conversations/store_tests.rs index a47d879f..8524fdc1 100644 --- a/src/memory/conversations/store_tests.rs +++ b/src/memory/conversations/store_tests.rs @@ -52,6 +52,135 @@ fn store_roundtrips_threads_and_messages() { assert_eq!(messages[0].content, "hello"); } +#[test] +fn append_message_is_idempotent_by_message_id() { + let (_temp, store) = make_store(); + store + .ensure_thread(CreateConversationThread { + parent_thread_id: None, + id: "t".to_string(), + title: "Conversation".to_string(), + created_at: "2026-04-10T12:00:00Z".to_string(), + labels: None, + personality_id: None, + }) + .expect("ensure thread"); + let first = ConversationMessage { + id: "agent:run-1".to_string(), + content: "first".to_string(), + message_type: "text".to_string(), + extra_metadata: json!({}), + sender: "agent".to_string(), + created_at: "2026-04-10T12:01:00Z".to_string(), + }; + store.append_message("t", first.clone()).expect("append"); + + // A second writer racing for the same id (the client persisting the + // `chat_done` an autonomous run already persisted itself — #5933). + let returned = store + .append_message( + "t", + ConversationMessage { + content: "second".to_string(), + created_at: "2026-04-10T12:02:00Z".to_string(), + ..first + }, + ) + .expect("append again"); + + // The stored row wins, and is what the second writer gets back. + assert_eq!(returned.content, "first"); + assert_eq!(returned.created_at, "2026-04-10T12:01:00Z"); + let messages = store.get_messages("t").expect("get messages"); + assert_eq!(messages.len(), 1, "one id, one row"); + assert_eq!(messages[0].content, "first"); + // The no-op append did not bump the stat trail either. + let threads = store.list_threads().expect("list threads"); + assert_eq!(threads[0].message_count, 1); + assert_eq!(threads[0].last_message_at, "2026-04-10T12:01:00Z"); +} + +#[test] +fn append_message_does_not_dedupe_client_generated_ids() { + // The idempotency lookup is scoped to the ids the core mints + // deterministically. Client-generated ids are UUID-fresh per message, so + // paying a transcript scan to verify that on every append would make the + // per-thread write path quadratic — the store takes them at face value + // instead. + let (_temp, store) = make_store(); + store + .ensure_thread(CreateConversationThread { + parent_thread_id: None, + id: "t".to_string(), + title: "Conversation".to_string(), + created_at: "2026-04-10T12:00:00Z".to_string(), + labels: None, + personality_id: None, + }) + .expect("ensure thread"); + let message = ConversationMessage { + id: "user:5f1d0c3e-1f8b-4c1a-9c2e-2a7b6d4e8f90".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(), + }; + store.append_message("t", message.clone()).expect("append"); + store.append_message("t", message).expect("append again"); + + assert_eq!(store.get_messages("t").expect("get messages").len(), 2); +} + +#[test] +fn append_message_idempotency_ignores_an_id_quoted_inside_content() { + // The lookup narrows candidate lines by raw text before parsing them; a + // message that merely *quotes* another message's id must not be mistaken + // for that message and swallow the real append. + let (_temp, store) = make_store(); + store + .ensure_thread(CreateConversationThread { + parent_thread_id: None, + id: "t".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( + "t", + ConversationMessage { + id: "user:1".to_string(), + content: "agent:run-9".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 quoting message"); + let stored = store + .append_message( + "t", + ConversationMessage { + id: "agent:run-9".to_string(), + content: "the real reply".to_string(), + message_type: "text".to_string(), + extra_metadata: json!({}), + sender: "agent".to_string(), + created_at: "2026-04-10T12:02:00Z".to_string(), + }, + ) + .expect("append reply"); + + assert_eq!(stored.content, "the real reply"); + let messages = store.get_messages("t").expect("get messages"); + assert_eq!(messages.len(), 2); + assert_eq!(messages[1].id, "agent:run-9"); +} + #[test] fn get_messages_for_new_empty_thread_returns_empty_list() { let (_temp, store) = make_store(); diff --git a/src/memory/conversations/store_tests_late.rs b/src/memory/conversations/store_tests_late.rs index 4927ae2d..41d0167b 100644 --- a/src/memory/conversations/store_tests_late.rs +++ b/src/memory/conversations/store_tests_late.rs @@ -1,4 +1,5 @@ use super::*; +use std::sync::Arc; #[test] fn search_cross_thread_messages_finds_japanese_bigram_match() { @@ -174,6 +175,75 @@ fn read_jsonl_skips_invalid_lines_but_keeps_valid_ones() { assert_eq!(messages[1].id, "m2"); } +#[test] +fn one_hundred_agent_threads_use_independent_message_locks() { + let temp = TempDir::new().unwrap(); + let store = ConversationStore::new(temp.path().to_path_buf()); + let root_lock = store.lock_identity_for_test(); + let mut message_locks = std::collections::HashSet::new(); + let mut retained_locks = Vec::new(); + + for index in 0..100 { + let clone = ConversationStore::new(temp.path().to_path_buf()); + assert_eq!(clone.lock_identity_for_test(), root_lock); + let lock = clone.locks.thread(&format!("agent-{index}")); + assert!(message_locks.insert(Arc::as_ptr(&lock) as usize)); + retained_locks.push(lock); + } + + assert_eq!(message_locks.len(), 100); +} + +#[test] +fn equivalent_workspace_paths_share_the_same_lock_registry_entry() { + let temp = TempDir::new().unwrap(); + let child = temp.path().join("child"); + std::fs::create_dir(&child).unwrap(); + + let direct = ConversationStore::new(temp.path().to_path_buf()); + let dotted = ConversationStore::new(child.join("..")); + + assert_eq!( + direct.lock_identity_for_test(), + dotted.lock_identity_for_test() + ); + assert_eq!(direct.root_dir(), dotted.root_dir()); +} + +#[test] +fn nonexistent_relative_and_absolute_workspaces_share_store_identity() { + let relative = PathBuf::from(format!("target/store-alias-{}", uuid::Uuid::new_v4())); + let absolute = std::env::current_dir().unwrap().join(&relative); + + let relative_store = ConversationStore::new(relative); + let absolute_store = ConversationStore::new(absolute); + + assert_eq!( + relative_store.lock_identity_for_test(), + absolute_store.lock_identity_for_test() + ); + assert_eq!(relative_store.root_dir(), absolute_store.root_dir()); +} + +#[cfg(unix)] +#[test] +fn symlinked_workspace_paths_share_the_same_lock_registry_entry() { + let temp = TempDir::new().unwrap(); + let workspace = temp.path().join("workspace"); + let alias = temp.path().join("alias"); + std::fs::create_dir(&workspace).unwrap(); + std::os::unix::fs::symlink(&workspace, &alias).unwrap(); + + let direct = ConversationStore::new(workspace); + let linked = ConversationStore::new(alias); + + assert_eq!( + direct.lock_identity_for_test(), + linked.lock_identity_for_test() + ); + assert_eq!(direct.root_dir(), linked.root_dir()); +} + // ── concurrency: search cold rebuild must not block concurrent append ──────── /// Regression test for issue #2849. @@ -181,17 +251,17 @@ fn read_jsonl_skips_invalid_lines_but_keeps_valid_ones() { /// 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. +/// moved the rebuild outside that lock (`prime_index_if_cold`). The current +/// store additionally takes only the target thread lock while reading each +/// transcript, so unrelated appends complete independently. /// /// 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. +/// serialized through one shared lock. #[test] fn search_cold_rebuild_does_not_block_concurrent_append() { - use std::sync::{mpsc, Arc, Barrier}; + use std::sync::mpsc; use std::thread; use std::time::Duration; @@ -213,6 +283,16 @@ fn search_cold_rebuild_does_not_block_concurrent_append() { personality_id: None, }) .unwrap(); + store + .ensure_thread(CreateConversationThread { + parent_thread_id: None, + id: "t2".to_string(), + title: "Concurrent append target".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 { @@ -234,21 +314,24 @@ fn search_cold_rebuild_does_not_block_concurrent_append() { 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 (scanned_tx, scanned_rx) = mpsc::channel(); + let (release_tx, release_rx) = mpsc::channel(); let search_handle = thread::spawn(move || { - b_search.wait(); - store_search.search_cross_thread_messages("seed message", 5, None) + let _lifecycle = store_search.locks.lifecycle.read(); + store_search.prime_index_if_cold_with_hook(|| { + scanned_tx.send(()).unwrap(); + release_rx.recv().unwrap(); + })?; + store_search.with_primed_index(|idx| idx.search("seed message", 5, None)) }); + scanned_rx + .recv_timeout(Duration::from_secs(5)) + .expect("cold rebuild did not reach the post-scan test seam"); let (tx, rx) = mpsc::channel(); thread::spawn(move || { - b_append.wait(); let result = store_append.append_message( - "t1", + "t2", ConversationMessage { id: "concurrent-append".to_string(), content: "written during cold rebuild".to_string(), @@ -261,18 +344,18 @@ fn search_cold_rebuild_does_not_block_concurrent_append() { 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. + // The unrelated append must finish while publication is deliberately + // paused. Releasing the rebuild first would let root-wide serialization + // pass this test eventually and prove nothing about overlap. let append_result = rx - .recv_timeout(Duration::from_secs(30)) - .expect("append_message did not complete within 30 s — likely blocked by cold rebuild"); + .recv_timeout(Duration::from_secs(5)) + .expect("unrelated append blocked behind a cold rebuild"); assert!( append_result.is_ok(), "append failed: {:?}", append_result.err() ); + release_tx.send(()).unwrap(); let search_result = search_handle.join().expect("search thread panicked"); assert!( @@ -280,6 +363,114 @@ fn search_cold_rebuild_does_not_block_concurrent_append() { "search failed: {:?}", search_result.err() ); + let appended = store + .search_cross_thread_messages("written during cold rebuild", 10, None) + .unwrap(); + assert!( + appended + .iter() + .any(|hit| hit.message_id == "concurrent-append"), + "completed append was omitted from index: {appended:?}" + ); +} + +#[test] +fn delete_during_cold_prime_cannot_republish_stale_messages() { + use std::sync::mpsc; + use std::thread; + use std::time::Duration; + + let (temp, store) = make_store(); + store + .ensure_thread(CreateConversationThread { + parent_thread_id: None, + id: "t1".to_string(), + title: "Delete race".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: "hello from a soon-deleted thread".to_string(), + message_type: "text".to_string(), + extra_metadata: serde_json::json!({}), + sender: "user".to_string(), + created_at: "2026-04-10T12:01:00Z".to_string(), + }, + ) + .unwrap(); + let store_prime = store.clone(); + let store_delete = store.clone(); + let root = store.root_dir(); + CONVERSATION_INDEX_CACHE.lock().remove(&root); + + let (scanned_tx, scanned_rx) = mpsc::channel(); + let (release_tx, release_rx) = mpsc::channel(); + let prime = thread::spawn(move || { + let _lifecycle = store_prime.locks.lifecycle.read(); + store_prime.prime_index_if_cold_with_hook(|| { + scanned_tx.send(()).unwrap(); + release_rx.recv().unwrap(); + }) + }); + scanned_rx + .recv_timeout(Duration::from_secs(5)) + .expect("prime did not reach post-scan seam"); + + let (delete_tx, delete_rx) = mpsc::channel(); + thread::spawn(move || { + let result = store_delete.delete_thread("t1", "2026-04-10T12:02:00Z"); + delete_tx.send(result).unwrap(); + }); + assert!( + delete_rx.recv_timeout(Duration::from_millis(100)).is_err(), + "delete must wait for the cold prime's lifecycle read guard" + ); + release_tx.send(()).unwrap(); + prime.join().unwrap().unwrap(); + assert!(delete_rx + .recv_timeout(Duration::from_secs(5)) + .unwrap() + .unwrap()); + + let hits = store + .search_cross_thread_messages("hello", 10, None) + .unwrap(); + assert!(hits.is_empty(), "deleted content was republished: {hits:?}"); + drop(temp); +} + +#[test] +fn delete_and_purge_evict_historical_thread_locks() { + let (_temp, store) = make_store(); + store + .ensure_thread(CreateConversationThread { + parent_thread_id: None, + id: "t1".to_string(), + title: "Eviction".to_string(), + created_at: "2026-04-10T12:00:00Z".to_string(), + labels: None, + personality_id: None, + }) + .unwrap(); + let t1_lock = store.locks.thread("t1"); + assert_eq!(store.thread_lock_count_for_test(), 1); + assert!(store.delete_thread("t1", "2026-04-10T12:02:00Z").unwrap()); + assert_eq!(store.thread_lock_count_for_test(), 0); + drop(t1_lock); + + let historical_locks = (0..100) + .map(|index| store.locks.thread(&format!("historical-{index}"))) + .collect::>(); + assert_eq!(store.thread_lock_count_for_test(), 100); + store.purge_threads().unwrap(); + assert_eq!(store.thread_lock_count_for_test(), 0); + drop(historical_locks); } // ── legacy workspace (pre-Stats backfill path) ─────────────────────────────── @@ -291,12 +482,13 @@ fn search_cold_rebuild_does_not_block_concurrent_append() { /// 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`. +/// `threads.jsonl`, formerly while holding `CONVERSATION_STORE_LOCK` and now +/// while holding the root's metadata 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 +/// thread IDs under the metadata lock, then reads each JSONL file under its +/// per-thread 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] @@ -346,7 +538,7 @@ fn prime_index_cold_build_works_on_legacy_workspace_without_stats() { } // Cold build on a pre-Stats workspace must index all messages without - // triggering measure_messages_unlocked under CONVERSATION_STORE_LOCK. + // triggering measure_messages_unlocked under the metadata lock. let hits = store .search_cross_thread_messages("kitten", 10, None) .expect("search on legacy workspace"); @@ -476,3 +668,58 @@ fn legacy_workspace_cold_rebuild_does_not_block_concurrent_append() { search_result.err() ); } + +#[test] +fn delete_error_still_evicts_tombstoned_thread_from_warm_index() { + let (_temp, store) = make_store(); + store + .ensure_thread(CreateConversationThread { + parent_thread_id: None, + id: "delete-error".to_string(), + title: "Delete error".to_string(), + created_at: "2026-04-10T12:00:00Z".to_string(), + labels: None, + personality_id: None, + }) + .unwrap(); + store + .append_message( + "delete-error", + ConversationMessage { + id: "indexed-before-delete-error".to_string(), + content: "must disappear after tombstone".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 + .search_cross_thread_messages("disappear after tombstone", 10, None) + .unwrap() + .len(), + 1 + ); + + // A directory at the transcript path makes remove_file fail after the + // metadata tombstone has already become durable. + let transcript = store.thread_messages_path("delete-error"); + std::fs::remove_file(&transcript).unwrap(); + std::fs::create_dir(&transcript).unwrap(); + assert!(store + .delete_thread("delete-error", "2026-04-10T12:02:00Z") + .is_err()); + + let hits = store + .search_cross_thread_messages("disappear after tombstone", 10, None) + .unwrap(); + assert!( + hits.is_empty(), + "tombstoned thread remained indexed: {hits:?}" + ); +} + +#[path = "store_concurrency_tests.rs"] +mod concurrency; diff --git a/src/memory/conversations/tokenize.rs b/src/memory/conversations/tokenize.rs index b9282a34..54c08472 100644 --- a/src/memory/conversations/tokenize.rs +++ b/src/memory/conversations/tokenize.rs @@ -43,7 +43,10 @@ //! 4. **Non-decomposing fold** — small per-letter table for decorated letters //! that have no canonical base (Polish ł, German ß, Norwegian ø, Icelandic //! þ/ð, Latin æ/œ, Turkish ı, Croatian đ, Maltese ħ, Sami ŋ). -//! 5. **Half-width → full-width katakana** — unifies the half-width and +//! 5. **Full-width ASCII → ASCII** — folds the full-width variants +//! (`ＡＢＣ` → `abc`, U+FF01..=U+FF5E, plus the ideographic space) so an +//! ASCII query retrieves full-width Latin content, matching NFKC. +//! 6. **Half-width → full-width katakana** — unifies the half-width and //! full-width kana forms so byte equality lines up at lookup time. //! //! The result is idempotent: re-running `normalize` on its own output is a @@ -66,6 +69,8 @@ pub fn normalize(text: &str) -> String { out.push_str(folded); } else if let Some(base) = strip_latin_diacritic(c) { out.push(base); + } else if let Some(ascii) = fullwidth_to_ascii(c) { + out.push(ascii); } else if let Some(full) = halfwidth_to_fullwidth(c) { out.push(full); } else { @@ -75,6 +80,27 @@ pub fn normalize(text: &str) -> String { out } +/// Full-width ASCII variants (U+FF01..=U+FF5E) → their ASCII originals, plus +/// the ideographic space (U+3000) → an ordinary space. +/// +/// These are NFKC compatibility folds — `ＡＢＣ` normalizes to `ABC` under +/// NFKC — and the pipeline this port reproduces ran NFKD→NFKC, so without +/// this arm an ASCII query could not retrieve indexed full-width Latin +/// content (common in CJK text, where full-width forms are produced by the +/// IME). The offset is uniform: full-width `！` (U+FF01) through `～` +/// (U+FF5E) sit exactly `0xFEE0` above `!`..=`~`. Lowercasing has already +/// happened by the time this runs, and `char::to_lowercase` maps full-width +/// `Ａ`→`ａ` within the full-width block, so the subtraction lands on the +/// lowercase ASCII letter directly. Disjoint from the half-width katakana +/// range (U+FF61..) handled below. +fn fullwidth_to_ascii(c: char) -> Option { + match c { + '\u{FF01}'..='\u{FF5E}' => char::from_u32(c as u32 - 0xFEE0), + '\u{3000}' => Some(' '), + _ => None, + } +} + /// Per-letter folds for "decorated" letters that have no canonical /// decomposition (so stripping combining marks alone leaves them unchanged). /// Polish ł/Ł is the motivating case — a Polish user typing `lacka` diff --git a/src/memory/conversations/tokenize_tests.rs b/src/memory/conversations/tokenize_tests.rs index 67e61760..14b201de 100644 --- a/src/memory/conversations/tokenize_tests.rs +++ b/src/memory/conversations/tokenize_tests.rs @@ -106,3 +106,14 @@ fn is_cjk_classifies_common_scripts() { assert!(!is_cjk('ą')); assert!(!is_cjk('，')); // CJK punctuation — intentionally NOT cjk } + +#[test] +fn normalize_folds_fullwidth_ascii() { + // NFKC folds the full-width ASCII variants; the port must too, or an + // ASCII query cannot retrieve indexed full-width Latin content. + assert_eq!(normalize("ＡＢＣ"), "abc"); + assert_eq!(normalize("ｈｅｌｌｏ　ｗｏｒｌｄ"), "hello world"); + assert_eq!(normalize("１２３！"), "123!"); + // Idempotent: the fold lands on plain ASCII, which passes through. + assert_eq!(normalize(&normalize("ＡＢＣ")), "abc"); +} diff --git a/src/memory/conversations/types.rs b/src/memory/conversations/types.rs index e81ec16e..6510c7b3 100644 --- a/src/memory/conversations/types.rs +++ b/src/memory/conversations/types.rs @@ -115,6 +115,51 @@ pub struct CrossThreadHit { pub score: f64, } +/// Prefix of the message ids the core mints **deterministically** rather than +/// from a fresh UUID. +/// +/// Only such an id can be presented to the store twice by two different +/// writers, so it is also the marker +/// [`is_deterministic_message_id`] keys the store's idempotency lookup on. +pub const DETERMINISTIC_MESSAGE_ID_PREFIX: &str = "agent:"; + +/// The id an autonomous run's closing reply is stored under. +/// +/// Two writers legitimately persist that one reply — the core's +/// background-delivery path and the client that also persists the +/// `chat_done` the run announces (`ChatRuntimeProvider` mirrors this shape for +/// `client_id: "system"` turns) — so both must derive the same id and the store +/// must collapse the second write onto the first (#5933). +pub fn run_reply_message_id(run_id: &str) -> String { + format!("{DETERMINISTIC_MESSAGE_ID_PREFIX}{run_id}") +} + +/// Whether `id` is one the core mints deterministically, i.e. one a second +/// writer can legitimately present again. +/// +/// This is what buys back the constant-time append path: every other id in the +/// store is UUID-fresh by construction (`user:`, `:`), can +/// never be re-presented, and so must not pay for a duplicate lookup. The +/// `agent:` ids the subagent/worker-thread writers mint do match — they +/// pay for a lookup they can never hit, which is one cheap scan of a +/// two-message worker transcript. +pub fn is_deterministic_message_id(id: &str) -> bool { + id.starts_with(DETERMINISTIC_MESSAGE_ID_PREFIX) +} + +/// The run/request id [`run_reply_message_id`] minted `id` from, or `None` +/// when `id` is not a deterministic reply id (see +/// [`is_deterministic_message_id`]). +/// +/// Backs `threads.edit_message` / `threads.regenerate`: an assistant reply's +/// store id is the one place the conversation-store id space and the +/// model-facing transcript's `request_id` space provably correlate, so +/// recovering the run id from the store id is how a UI message id resolves +/// to a transcript cut point. +pub fn reply_run_id(id: &str) -> Option<&str> { + id.strip_prefix(DETERMINISTIC_MESSAGE_ID_PREFIX) +} + #[cfg(test)] #[path = "types_tests.rs"] mod tests; diff --git a/src/memory/conversations/types_tests.rs b/src/memory/conversations/types_tests.rs index 50fee030..e6bee4ea 100644 --- a/src/memory/conversations/types_tests.rs +++ b/src/memory/conversations/types_tests.rs @@ -63,3 +63,24 @@ fn create_thread_optional_fields_roundtrip() { ); assert!(decoded.parent_thread_id.is_none()); } + +#[test] +fn run_reply_id_is_deterministic_and_recognised_as_such() { + // The background-delivery producer and the predicate the store + // gates its idempotency lookup on must agree, or the two writers of an + // autonomous reply stop collapsing onto one row (#5933). + assert_eq!(run_reply_message_id("run-7"), "agent:run-7"); + assert!(is_deterministic_message_id(&run_reply_message_id("run-7"))); +} + +#[test] +fn client_generated_ids_are_not_deterministic() { + // These are UUID-fresh per message, so they can never be re-presented and + // must keep the constant-time append path. + assert!(!is_deterministic_message_id( + "user:5f1d0c3e-1f8b-4c1a-9c2e-2a7b6d4e8f90" + )); + assert!(!is_deterministic_message_id( + "assistant:5f1d0c3e-1f8b-4c1a-9c2e-2a7b6d4e8f90" + )); +} From 58d9a150f3aefc4d1f226e5887326758faf587f0 Mon Sep 17 00:00:00 2001 From: Steven Enamakel Date: Tue, 29 Sep 2026 20:57:23 +0300 Subject: [PATCH 2/5] chore: files changed src/memory/conversations/mod.rs,src/memory/conversations/store.rs,src/memory/st Auto-committed-on: dragonfly Co-authored-by: Medulla --- src/memory/conversations/mod.rs | 10 ++--- src/memory/conversations/store.rs | 40 +++++++------------ .../store/content/obsidian_registry_tests.rs | 24 +++++++++++ 3 files changed, 44 insertions(+), 30 deletions(-) diff --git a/src/memory/conversations/mod.rs b/src/memory/conversations/mod.rs index de220cca..8035ec11 100644 --- a/src/memory/conversations/mod.rs +++ b/src/memory/conversations/mod.rs @@ -43,11 +43,11 @@ pub use bus::{ ConversationEventBus, ConversationPersistenceSubscriber, }; pub use store::{ - append_message, delete_thread, ensure_thread, get_messages, list_threads, purge_threads, - update_message, update_thread_labels, update_thread_title, ConversationPurgeStats, - ConversationStore, + append_message, delete_messages_from, 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, + is_deterministic_message_id, reply_run_id, run_reply_message_id, ConversationMessage, + ConversationMessagePatch, ConversationThread, CreateConversationThread, CrossThreadHit, }; diff --git a/src/memory/conversations/store.rs b/src/memory/conversations/store.rs index 506821b0..cd0d632b 100644 --- a/src/memory/conversations/store.rs +++ b/src/memory/conversations/store.rs @@ -7,28 +7,14 @@ //! On-disk mutations synchronize at the narrowest safe scope: lifecycle per //! conversation root, shared metadata per root, and messages per thread. //! -//! This file is the store as it came back from the memory engine (#5560), -//! unchanged apart from the one constructor described below. The three -//! dependency substitutions the round trip introduced — [`std::sync::LazyLock`] -//! for the statics, the local [`hex_encode`] for per-thread filenames, and the -//! hand-rolled temp-write in [`rewrite_jsonl`] — are all kept: they produce the -//! bytes and the paths that every existing transcript already lives at, and a -//! move must not rewrite those. See [`super`]'s module docs for the full -//! accounting of what the trip cost. +//! Dependency substitutions that keep this crate's `Cargo.toml` untouched: +//! [`std::sync::LazyLock`] for the statics, the local [`hex_encode`] for +//! per-thread filenames, and the hand-rolled temp-write in [`rewrite_jsonl`]. +//! They produce the bytes and paths every existing transcript already lives +//! at, so they must not change. //! -//! # The one thing that did change: `from_config` -//! -//! The engine's [`ConversationStore`] carried a second constructor, -//! `from_config(&MemoryConfig)`, whose whole body was -//! `Self::new(config.workspace.clone())`. It is gone rather than translated, -//! and that is the point of the move: it was this module's *only* coupling to -//! the rest of the engine, and re-expressing it here would either drag -//! `MemoryConfig` back in or add a second name for a constructor that already -//! exists. [`ConversationStore::new`] takes the workspace directory, every -//! caller in this host already has one, and no caller ever used `from_config` -//! — so nothing needed a translation and the derived on-disk root -//! (`/memory/conversations`, see `ConversationStore::root_dir` in -//! `store_index.rs`) is byte-identical either way. +//! The lock registry ([`locks`]) and deterministic-id idempotency were +//! upstreamed from OpenHuman's host copy of this store. //! //! # File split //! @@ -120,10 +106,8 @@ pub struct ConversationStore { impl ConversationStore { /// Construct a store rooted at the given workspace directory. /// - /// This is the only constructor. The conversation root is derived from - /// `workspace_dir` alone (`/memory/conversations`, see - /// `root_dir` in `store_index.rs`), so the caller's workspace is the whole - /// input and there is nothing for a config type to add. + /// The conversation root is derived from `workspace_dir` alone + /// (`/memory/conversations`, see `root_dir` in `store_index.rs`). pub fn new(workspace_dir: PathBuf) -> Self { let root = locks::normalized_root(&workspace_dir.join("memory").join("conversations")); let locks = locks::for_root(&root); @@ -133,6 +117,12 @@ impl ConversationStore { } } + /// Construct a store rooted at the engine's configured workspace + /// ([`MemoryConfig::workspace`](crate::memory::config::MemoryConfig)). + pub fn from_config(config: &crate::memory::config::MemoryConfig) -> Self { + Self::new(config.workspace.clone()) + } + #[cfg(test)] pub(super) fn lock_identity_for_test(&self) -> usize { std::sync::Arc::as_ptr(&self.locks) as usize diff --git a/src/memory/store/content/obsidian_registry_tests.rs b/src/memory/store/content/obsidian_registry_tests.rs index c302d3ab..7653d18b 100644 --- a/src/memory/store/content/obsidian_registry_tests.rs +++ b/src/memory/store/content/obsidian_registry_tests.rs @@ -153,3 +153,27 @@ fn second_candidate_wins_when_first_missing() { let real = write_config(tmp.path(), &[root.to_str().unwrap()]); assert!(registration_in_files(&root, &[missing, real]).registered); } + +/// The override arm is the host's own reason for keeping this code: a user +/// with a non-standard Obsidian install points at their config dir and the +/// probe has to look there *first*. Both spellings are accepted because users +/// cannot tell whether the path should end in `obsidian/`. +#[test] +fn extra_config_dir_is_probed_before_the_standard_locations() { + let tmp = tempfile::tempdir().unwrap(); + let files = candidate_config_files(Some(tmp.path())); + assert_eq!(files[0], tmp.path().join("obsidian.json")); + assert_eq!(files[1], tmp.path().join("obsidian").join("obsidian.json")); +} + +/// No override means no override-derived candidates — the probe must not +/// invent a relative `./obsidian.json`, which would make the answer depend on +/// the process working directory. +#[test] +fn no_extra_config_dir_adds_no_relative_candidate() { + let files = candidate_config_files(None); + assert!( + files.iter().all(|p| p.is_absolute()), + "every candidate must be absolute; got {files:?}" + ); +} From c2cc9815e4b03e1fa1a4118ee97b4d41f4ae3345 Mon Sep 17 00:00:00 2001 From: Steven Enamakel Date: Tue, 29 Sep 2026 20:57:50 +0300 Subject: [PATCH 3/5] test(extract): add serialisation and deserialisation tests for ingestion types Adds two tests to pin the JSON wire format of `MemoryIngestionResult` and `MemoryIngestionConfig`, ensuring camelCase keys and required fields are preserved and that backward compatibility with snake_case mode values is maintained. Auto-committed-on: dragonfly Co-authored-by: Medulla --- src/memory/ingest/extract/types_tests.rs | 56 ++++++++++++++++++++++++ 1 file changed, 56 insertions(+) diff --git a/src/memory/ingest/extract/types_tests.rs b/src/memory/ingest/extract/types_tests.rs index 869a62ff..0558eb9c 100644 --- a/src/memory/ingest/extract/types_tests.rs +++ b/src/memory/ingest/extract/types_tests.rs @@ -36,3 +36,59 @@ fn memory_ingestion_request_defaults_config_when_absent() { assert_eq!(request.config.model_name, DEFAULT_MEMORY_EXTRACTION_MODEL); assert_eq!(request.config.extraction_mode, ExtractionMode::Sentence); } + +/// Pins the `MemoryIngestionResult` JSON literal-for-literal so the wire +/// shape (camelCase keys, field set) can never drift. +#[test] +fn ingestion_result_serialises_the_engine_shape() { + let result = MemoryIngestionResult { + document_id: "doc-1".into(), + namespace: "global".into(), + model_name: "driver-managed".into(), + extraction_mode: "driver-managed".into(), + chunk_count: 0, + entity_count: 0, + relation_count: 0, + preference_count: 0, + decision_count: 0, + tags: vec!["t".into()], + entities: Vec::new(), + relations: Vec::new(), + }; + let value = serde_json::to_value(&result).expect("serialise"); + assert_eq!( + value, + json!({ + "documentId": "doc-1", + "namespace": "global", + "modelName": "driver-managed", + "extractionMode": "driver-managed", + "chunkCount": 0, + "entityCount": 0, + "relationCount": 0, + "preferenceCount": 0, + "decisionCount": 0, + "tags": ["t"], + "entities": [], + "relations": [] + }) + ); +} + +/// A config a prior engine accepted must still deserialize: `modelName` +/// required, everything else defaulted, including the mode's `snake_case` +/// spelling. +#[test] +fn ingestion_config_minimal_json_defaults_and_requires_model_name() { + let config: MemoryIngestionConfig = + serde_json::from_value(json!({ "modelName": "heuristic-only" })) + .expect("minimal config deserialises"); + assert_eq!(config.extraction_mode, ExtractionMode::Sentence); + assert!((config.entity_threshold - 0.45).abs() < f32::EPSILON); + assert!((config.relation_threshold - 0.30).abs() < f32::EPSILON); + assert!((config.adjacency_threshold - 0.50).abs() < f32::EPSILON); + assert_eq!(config.batch_size, 16); + let chunk: ExtractionMode = serde_json::from_value(json!("chunk")).expect("snake_case mode"); + assert_eq!(chunk, ExtractionMode::Chunk); + assert!(serde_json::from_value::(json!({})).is_err()); +} From fdf72aca2257b0412bdb0905afa71264f43dedfb Mon Sep 17 00:00:00 2001 From: Steven Enamakel Date: Tue, 29 Sep 2026 20:58:37 +0300 Subject: [PATCH 4/5] feat(memory): add test module for health types Include a test module for the memory health types by conditionally compiling and linking the types_tests.rs file under the `#[cfg(test)]` attribute. This enables unit tests to be written and run for the types defined in this module without affecting production builds. Auto-committed-on: dragonfly Co-authored-by: Medulla --- src/memory/health/types.rs | 4 + src/memory/health/types_tests.rs | 389 +++++++++++++++++++++++++++++++ 2 files changed, 393 insertions(+) create mode 100644 src/memory/health/types_tests.rs diff --git a/src/memory/health/types.rs b/src/memory/health/types.rs index bacf72f5..5ac28d28 100644 --- a/src/memory/health/types.rs +++ b/src/memory/health/types.rs @@ -105,3 +105,7 @@ pub struct DegradedState { #[serde(default, skip_serializing_if = "Option::is_none")] pub cause: Option, } + +#[cfg(test)] +#[path = "types_tests.rs"] +mod tests; diff --git a/src/memory/health/types_tests.rs b/src/memory/health/types_tests.rs new file mode 100644 index 00000000..2b324052 --- /dev/null +++ b/src/memory/health/types_tests.rs @@ -0,0 +1,389 @@ +//! Wire pins for the health taxonomy types. +//! +//! These assert **bytes**, not behaviour: every literal is the JSON the +//! `FailureCode` / `FailureClass` / `PipelineFailure` / `DegradedState` types +//! emit over the memory RPC surface. The table [`CODES`] is the whole +//! vocabulary and `every_variant_is_in_the_table` stops it rotting: a new +//! variant is a compile error in its `match`, not a silently unpinned row. + +use super::*; + +/// `(variant, wire string, class, remediation key)` for every [`FailureCode`]. +/// +/// Order matches the enum's declaration order, which is also the serde +/// discriminant order — not that anything depends on it, since every +/// representation here is by name. +const CODES: &[(FailureCode, &str, FailureClass, &str)] = &[ + ( + FailureCode::BudgetExhausted, + "budget_exhausted", + FailureClass::Unrecoverable, + "memory.health.remediation.budget_exhausted", + ), + ( + FailureCode::AuthMissing, + "auth_missing", + FailureClass::Unrecoverable, + "memory.health.remediation.auth_missing", + ), + ( + FailureCode::AuthInvalid, + "auth_invalid", + FailureClass::Unrecoverable, + "memory.health.remediation.auth_invalid", + ), + ( + FailureCode::EmbeddingsUnconfigured, + "embeddings_unconfigured", + FailureClass::Unrecoverable, + "memory.health.remediation.embeddings_unconfigured", + ), + ( + FailureCode::EmbeddingDimMismatch, + "embedding_dim_mismatch", + FailureClass::Unrecoverable, + "memory.health.remediation.embedding_dim_mismatch", + ), + ( + FailureCode::LocalModelUnavailable, + "local_model_unavailable", + FailureClass::Transient, + "memory.health.remediation.local_model_unavailable", + ), + ( + FailureCode::ExtractionTimeout, + "extraction_timeout", + FailureClass::Transient, + "memory.health.remediation.extraction_timeout", + ), + ( + FailureCode::SummarizerUnavailable, + "summarizer_unavailable", + FailureClass::Unrecoverable, + "memory.health.remediation.summarizer_unavailable", + ), + ( + FailureCode::EmptyInputRefused, + "empty_input_refused", + FailureClass::Unrecoverable, + "memory.health.remediation.empty_input_refused", + ), + ( + FailureCode::StorageUnavailable, + "storage_unavailable", + FailureClass::Unrecoverable, + "memory.health.remediation.storage_unavailable", + ), + ( + FailureCode::Transient, + "transient", + FailureClass::Transient, + "memory.health.remediation.transient", + ), +]; + +/// A variant added to [`FailureCode`] must be added to [`CODES`] too, or its +/// spelling, class and remediation key go unpinned. +/// +/// The `match` is what enforces it: adding a variant fails to compile here +/// first, and the length assertion then fails until the table catches up. +/// Without this, every other test in this file would keep passing over a +/// vocabulary it no longer covers. +#[test] +fn every_variant_is_in_the_table() { + fn position(code: FailureCode) -> usize { + match code { + FailureCode::BudgetExhausted => 0, + FailureCode::AuthMissing => 1, + FailureCode::AuthInvalid => 2, + FailureCode::EmbeddingsUnconfigured => 3, + FailureCode::EmbeddingDimMismatch => 4, + FailureCode::LocalModelUnavailable => 5, + FailureCode::ExtractionTimeout => 6, + FailureCode::SummarizerUnavailable => 7, + FailureCode::EmptyInputRefused => 8, + FailureCode::StorageUnavailable => 9, + FailureCode::Transient => 10, + } + } + + assert_eq!(CODES.len(), 11, "the table must cover every variant"); + for (index, (code, ..)) in CODES.iter().enumerate() { + assert_eq!(position(*code), index, "table out of order at {index}"); + } +} + +/// The whole vocabulary, one row at a time: the wire string, the serde +/// spelling, the derived class, the remediation key, and the round trip back. +/// +/// `as_str` and the serde representation are asserted to agree — they are two +/// independent tables (an explicit `match` and `#[serde(rename_all)]`), read by +/// different consumers, and nothing but this makes them the same words. +#[test] +fn every_code_serialises_and_parses_to_its_pinned_spelling() { + for (code, wire, class, remediation_key) in CODES { + assert_eq!(code.as_str(), *wire, "as_str for {code:?}"); + assert_eq!( + serde_json::to_value(code).unwrap(), + serde_json::json!(wire), + "serde spelling for {code:?}" + ); + assert_eq!( + serde_json::from_value::(serde_json::json!(wire)).unwrap(), + *code, + "serde round trip for {code:?}" + ); + assert_eq!( + FailureCode::from_str(wire), + Some(*code), + "from_str for {code:?}" + ); + assert_eq!(code.class(), *class, "class for {code:?}"); + assert_eq!( + code.remediation_key(), + *remediation_key, + "remediation key for {code:?}" + ); + } +} + +/// The remediation keys are a frontend contract, so the *shape* is pinned too: +/// every key is `memory.health.remediation.`. +/// +/// Asserted as a relation rather than as eleven more literals, because that is +/// the property a locale file relies on — a key that drifts away from its code +/// resolves to nothing and the panel renders a blank remediation. +#[test] +fn every_remediation_key_is_derived_from_its_code() { + for (code, wire, _, remediation_key) in CODES { + assert_eq!( + *remediation_key, + format!("memory.health.remediation.{wire}"), + "remediation key shape for {code:?}" + ); + } +} + +/// A code this build has no variant for is `None`, not a nearest guess. +#[test] +fn an_unknown_code_does_not_parse() { + assert_eq!( + FailureCode::from_str("a_cause_this_build_never_heard_of"), + None + ); + assert_eq!(FailureCode::from_str(""), None); + // Casing is not normalised: the wire is snake_case and only snake_case. + assert_eq!(FailureCode::from_str("Auth_Missing"), None); + assert_eq!(FailureCode::from_str("AuthMissing"), None); +} + +/// Both `FailureClass` spellings, in both directions. +#[test] +fn failure_class_serialises_to_its_pinned_spelling() { + for (class, wire) in [ + (FailureClass::Transient, "transient"), + (FailureClass::Unrecoverable, "unrecoverable"), + ] { + assert_eq!(class.as_str(), wire); + assert_eq!( + serde_json::to_value(class).unwrap(), + serde_json::json!(wire) + ); + assert_eq!( + serde_json::from_value::(serde_json::json!(wire)).unwrap(), + class + ); + } +} + +/// `PipelineFailure`'s object shape: three fields always, `detail` only when +/// present. Field names are asserted as an exact object, so a rename cannot +/// pass by adding a field the assertion does not mention. +#[test] +fn pipeline_failure_object_shape_is_unchanged() { + let bare = PipelineFailure::new(FailureCode::EmbeddingsUnconfigured); + assert_eq!( + serde_json::to_value(&bare).unwrap(), + serde_json::json!({ + "code": "embeddings_unconfigured", + "class": "unrecoverable", + "remediation_key": "memory.health.remediation.embeddings_unconfigured", + }), + "detail is omitted when None, never emitted as null" + ); + + let detailed = PipelineFailure::new(FailureCode::Transient).with_detail("boom"); + assert_eq!( + serde_json::to_value(&detailed).unwrap(), + serde_json::json!({ + "code": "transient", + "class": "transient", + "remediation_key": "memory.health.remediation.transient", + "detail": "boom", + }) + ); +} + +/// The deserialisation tolerance the wire has always had: `detail` may be +/// absent. Older payloads (and the driver's own, when it attaches none) omit +/// it, so `#[serde(default)]` is load-bearing rather than decoration. +#[test] +fn pipeline_failure_deserialises_without_a_detail() { + let parsed: PipelineFailure = serde_json::from_str( + r#"{"code":"auth_missing","class":"unrecoverable","remediation_key":"k"}"#, + ) + .expect("detail is optional"); + assert_eq!(parsed.code, FailureCode::AuthMissing); + assert_eq!(parsed.class, FailureClass::Unrecoverable); + assert_eq!(parsed.remediation_key, "k"); + assert_eq!(parsed.detail, None); +} + +/// `new` derives the class and the remediation key from the code, and +/// `is_unrecoverable` reads the class rather than re-deriving from the code — +/// which is what lets `report`/`rpc` override a class the driver stated +/// without the two disagreeing. +#[test] +fn new_derives_class_and_key_and_is_unrecoverable_reads_the_class() { + for (code, _, class, remediation_key) in CODES { + let failure = PipelineFailure::new(*code); + assert_eq!(failure.class, *class); + assert_eq!(failure.remediation_key, *remediation_key); + assert_eq!(failure.detail, None); + assert_eq!( + failure.is_unrecoverable(), + *class == FailureClass::Unrecoverable + ); + } + + let mut overridden = PipelineFailure::new(FailureCode::BudgetExhausted); + assert!(overridden.is_unrecoverable()); + overridden.class = FailureClass::Transient; + assert!(!overridden.is_unrecoverable()); +} + +/// `with_detail` caps at 200 **characters** and marks the cut with `…`. +/// +/// Characters, not bytes: a byte slice would panic on a multi-byte boundary, +/// and the detail can carry a provider's non-ASCII error body. +#[test] +fn with_detail_truncates_by_characters_and_marks_the_cut() { + let short = PipelineFailure::new(FailureCode::Transient).with_detail("x"); + assert_eq!(short.detail.as_deref(), Some("x")); + + let exact = PipelineFailure::new(FailureCode::Transient).with_detail("a".repeat(200)); + assert_eq!(exact.detail.as_deref().unwrap().chars().count(), 200); + assert!(!exact.detail.as_deref().unwrap().ends_with('…')); + + let long = PipelineFailure::new(FailureCode::Transient).with_detail("a".repeat(250)); + let detail = long.detail.as_deref().unwrap(); + assert_eq!(detail.chars().count(), 201, "200 kept plus the ellipsis"); + assert!(detail.ends_with('…')); + + // Multi-byte input must not be sliced mid-codepoint. + let wide = PipelineFailure::new(FailureCode::Transient).with_detail("é".repeat(250)); + let detail = wide.detail.as_deref().unwrap(); + assert_eq!(detail.chars().count(), 201); + assert!(detail.starts_with('é')); +} + +/// The `Display` rendering that reaches logs: `code (class)`, with the detail +/// appended only when there is one. +#[test] +fn pipeline_failure_display_is_unchanged() { + assert_eq!( + PipelineFailure::new(FailureCode::AuthMissing).to_string(), + "auth_missing (unrecoverable)" + ); + assert_eq!( + PipelineFailure::new(FailureCode::AuthMissing) + .with_detail("boom") + .to_string(), + "auth_missing (unrecoverable): boom" + ); +} + +/// `DegradedState`'s object shape: the three flags always present (`storage` +/// has `#[serde(default)]` and deliberately **no** `skip_serializing_if`), +/// `cause` only when known. +#[test] +fn degraded_state_object_shape_is_unchanged() { + assert_eq!( + serde_json::to_value(DegradedState::default()).unwrap(), + serde_json::json!({ + "semantic_recall": false, + "structure": false, + "storage": false, + }), + "all three flags are emitted; cause is omitted when None" + ); + + assert_eq!( + serde_json::to_value(DegradedState { + semantic_recall: true, + structure: true, + storage: true, + cause: Some(PipelineFailure::new(FailureCode::AuthInvalid).with_detail("d")), + }) + .unwrap(), + serde_json::json!({ + "semantic_recall": true, + "structure": true, + "storage": true, + "cause": { + "code": "auth_invalid", + "class": "unrecoverable", + "remediation_key": "memory.health.remediation.auth_invalid", + "detail": "d", + }, + }) + ); +} + +/// An older client's payload — no `storage`, no `cause` — still deserialises, +/// with `storage` defaulting to `false`. That backward compatibility is the +/// whole reason `storage` carries `#[serde(default)]`. +#[test] +fn degraded_state_deserialises_a_payload_without_storage_or_cause() { + let parsed: DegradedState = + serde_json::from_str(r#"{"semantic_recall":true,"structure":false}"#) + .expect("storage and cause are both optional"); + assert_eq!( + parsed, + DegradedState { + semantic_recall: true, + structure: false, + storage: false, + cause: None, + } + ); +} + +/// `is_degraded` is the OR of the three flags and ignores `cause` — a cause +/// with no flag set is a report about something already cleared, not a +/// degradation. +#[test] +fn is_degraded_is_the_or_of_the_three_flags() { + assert!(!DegradedState::default().is_degraded()); + for state in [ + DegradedState { + semantic_recall: true, + ..Default::default() + }, + DegradedState { + structure: true, + ..Default::default() + }, + DegradedState { + storage: true, + ..Default::default() + }, + ] { + assert!(state.is_degraded()); + } + assert!(!DegradedState { + cause: Some(PipelineFailure::new(FailureCode::Transient)), + ..Default::default() + } + .is_degraded()); +} From a778750335fe4b48dd7bb4aae3ca4365cf2d7e5a Mon Sep 17 00:00:00 2001 From: Steven Enamakel Date: Tue, 29 Sep 2026 21:00:55 +0300 Subject: [PATCH 5/5] chore(store): remove unused test helper Removed the `thread_lock_identity_for_test` method from `ConversationStore`, as it was no longer used by any test code. The method exposed internal lock pointer identities that were only needed during an earlier debugging phase. Auto-committed-on: dragonfly Co-authored-by: Medulla --- src/memory/conversations/store.rs | 5 ----- 1 file changed, 5 deletions(-) diff --git a/src/memory/conversations/store.rs b/src/memory/conversations/store.rs index cd0d632b..6bcb8599 100644 --- a/src/memory/conversations/store.rs +++ b/src/memory/conversations/store.rs @@ -128,11 +128,6 @@ impl ConversationStore { std::sync::Arc::as_ptr(&self.locks) as usize } - #[cfg(test)] - pub(super) fn thread_lock_identity_for_test(&self, thread_id: &str) -> usize { - std::sync::Arc::as_ptr(&self.locks.thread(thread_id)) as usize - } - #[cfg(test)] pub(super) fn thread_lock_count_for_test(&self) -> usize { self.locks.thread_count()