diff --git a/src/memory/conversations/mod.rs b/src/memory/conversations/mod.rs
index de220cc..8035ec1 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 0a83c8f..6bcb859 100644
--- a/src/memory/conversations/store.rs
+++ b/src/memory/conversations/store.rs
@@ -4,24 +4,23 @@
 //! `threads/<hex(thread_id)>.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:
+//! 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.
 //!
-//! - `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 lock registry ([`locks`]) and deterministic-id idempotency were
+//! upstreamed from OpenHuman's host copy of this store.
 //!
-//! 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
+//! # 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 +43,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 +53,6 @@ pub(super) const THREADS_FILENAME: &str = "threads.jsonl";
 /// per-thread message JSONL files, named `<hex(thread_id)>.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<Mutex<()>> = 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 +64,26 @@ static CONVERSATION_STORE_LOCK: LazyLock<Mutex<()>> = 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<Mutex<HashMap<PathBuf, InvertedIndex>>> =
     LazyLock::new(|| Mutex::new(HashMap::new()));
 
@@ -104,13 +99,22 @@ 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<locks::StoreLocks>,
 }
 
 impl ConversationStore {
     /// Construct a store rooted at the given workspace directory.
+    ///
+    /// The conversation root is derived from `workspace_dir` alone
+    /// (`<workspace>/memory/conversations`, see `root_dir` in `store_index.rs`).
     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,
+        }
     }
 
     /// Construct a store rooted at the engine's configured workspace
@@ -118,6 +122,16 @@ impl ConversationStore {
     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
+    }
+
+    #[cfg(test)]
+    pub(super) fn thread_lock_count_for_test(&self) -> usize {
+        self.locks.thread_count()
+    }
 }
 
 /// One line in `threads.jsonl`. The append-only log is folded into the current
@@ -220,8 +234,13 @@ pub(super) fn normalize_labels(labels: Vec<String>) -> Vec<String> {
 }
 
 /// 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 +277,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<Option<ConversationMessage>, 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::<ConversationMessage>(&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<T>(path: &Path, value: &T) -> Result<(), String>
 where
@@ -284,9 +333,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<T>(path: &Path, values: &[T]) -> Result<(), String>
 where
     T: serde::Serialize,
@@ -381,6 +436,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<Option<usize>, 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<ConversationPurgeStats, String> {
     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 0000000..b4fa6ad
--- /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 ce44322..854178e 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<String> = {
-            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<R>(
+    /// 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<R>(
         &self,
         f: impl FnOnce(&mut InvertedIndex) -> R,
     ) -> Result<R, String> {
         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::<ConversationMessage>(&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<PathBuf, String> {
@@ -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<Vec<ConversationThread>, 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::<Vec<_>>();
+                (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<String, ThreadIndexEntry>) -> Vec<ConversationThread> {
         let mut threads: Vec<ConversationThread> = index
             .iter()
             .map(|(thread_id, entry)| {
@@ -231,7 +270,7 @@ impl ConversationStore {
                 .cmp(&timestamp_millis(&a.last_message_at))
                 .then_with(|| timestamp_millis(&b.created_at).cmp(&timestamp_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 0000000..651d81e
--- /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<HashMap<String, Weak<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<Option<Vec<(String, ConversationMessage)>>>,
+}
+
+impl StoreLocks {
+    pub(super) fn thread(&self, thread_id: &str) -> Arc<Mutex<()>> {
+        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<Mutex<HashMap<PathBuf, Weak<StoreLocks>>>> =
+    LazyLock::new(|| Mutex::new(HashMap::new()));
+
+pub(super) fn for_root(root: &Path) -> Arc<StoreLocks> {
+    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 1189cb1..215eca8 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<ConversationThread, String> {
-        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<Vec<ConversationThread>, 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<Vec<ConversationMessage>, 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<Vec<CrossThreadHit>, 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<ConversationMessage, String> {
-        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<ConversationThread, String> {
-        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<String>,
         updated_at: &str,
     ) -> Result<ConversationThread, String> {
-        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<ConversationMessage, String> {
-        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::<ConversationMessage>(&path)?;
         let mut updated: Option<ConversationMessage> = 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<Option<usize>, 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::<ConversationMessage>(&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<bool, String> {
-        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<ConversationPurgeStats, String> {
-        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 a47d879..8524fdc 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 4927ae2..41d0167 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::<Vec<_>>();
+    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 b9282a3..54c0847 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<char> {
+    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 67e6176..14b201d 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 e81ec16..6510c7b 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:<uuid>`, `<sender>:<uuid>`), can
+/// never be re-presented, and so must not pay for a duplicate lookup. The
+/// `agent:<uuid>` 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 50fee03..e6bee4e 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"
+    ));
+}
diff --git a/src/memory/health/types.rs b/src/memory/health/types.rs
index bacf72f..5ac28d2 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<PipelineFailure>,
 }
+
+#[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 0000000..2b32405
--- /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::<FailureCode>(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.<the code's own wire string>`.
+///
+/// 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::<FailureClass>(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());
+}
diff --git a/src/memory/ingest/extract/types_tests.rs b/src/memory/ingest/extract/types_tests.rs
index 869a62f..0558eb9 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::<MemoryIngestionConfig>(json!({})).is_err());
+}
diff --git a/src/memory/store/content/obsidian_registry_tests.rs b/src/memory/store/content/obsidian_registry_tests.rs
index c302d3a..7653d18 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:?}"
+    );
+}