From a56fec7ebcabf24ae39977415c735398eff6784a Mon Sep 17 00:00:00 2001 From: Henry Park Date: Tue, 7 Apr 2026 20:05:45 -0700 Subject: [PATCH] perf: fix multi-tenant inference latency (per-conversation locking + workspace indexing) (#2127) --- .env.example | 2 +- .../src/runtime/conversation.rs | 440 +++++++++++++----- src/config/database.rs | 2 +- src/workspace/mod.rs | 120 +++++ 4 files changed, 438 insertions(+), 126 deletions(-) diff --git a/.env.example b/.env.example index 2395fee70b..f81b1d6976 100644 --- a/.env.example +++ b/.env.example @@ -1,6 +1,6 @@ # Database Configuration DATABASE_URL=postgres://localhost/ironclaw -DATABASE_POOL_SIZE=10 +DATABASE_POOL_SIZE=30 # multi-tenant default; reduce to 5-10 for single-user or low-resource deployments # LLM Provider # LLM_BACKEND=nearai # default diff --git a/crates/ironclaw_engine/src/runtime/conversation.rs b/crates/ironclaw_engine/src/runtime/conversation.rs index bbc50c61cc..e797d94820 100644 --- a/crates/ironclaw_engine/src/runtime/conversation.rs +++ b/crates/ironclaw_engine/src/runtime/conversation.rs @@ -8,7 +8,7 @@ use std::collections::HashMap; use std::sync::Arc; -use tokio::sync::RwLock; +use tokio::sync::{Mutex, RwLock}; use tracing::debug; use crate::runtime::manager::ThreadManager; @@ -20,6 +20,7 @@ use crate::types::message::ThreadMessage; use crate::types::project::ProjectId; use crate::types::thread::{ThreadConfig, ThreadId, ThreadState, ThreadType}; +#[derive(Clone, Copy)] enum ActiveForeground { Running(ThreadId), Resumable(ThreadId), @@ -31,10 +32,23 @@ enum ActiveForeground { /// 1. Spawn a new foreground thread for the message /// 2. Inject the message into an existing active thread /// 3. Create a new conversation if none exists for this channel+user +/// +/// ## Locking strategy +/// +/// `conversations` is a *directory*: the global `RwLock` is held only for +/// HashMap lookups/inserts and is never held across an `.await`. Each +/// `ConversationSurface` is wrapped in a `tokio::sync::Mutex` so concurrent +/// messages to *different* conversations run fully in parallel. +/// +/// **Lock ordering invariant:** NEVER hold the global `RwLock` and a +/// per-conversation `Mutex` simultaneously. `get_conversation_lock()` enforces +/// this — it drops the read guard before returning the `Arc>`. pub struct ConversationManager { thread_manager: Arc, store: Arc, - conversations: RwLock>, + // LOCK ORDER: when acquiring both write locks, always take `conversations` before + // `channel_user_index`. Reversing this order will deadlock under concurrent access. + conversations: RwLock>>>, /// Maps (channel, user_id) → conversation ID for lookup. channel_user_index: RwLock>, } @@ -49,22 +63,47 @@ impl ConversationManager { } } + /// Get the per-conversation lock. Holds the global RwLock only briefly + /// (HashMap lookup), then releases it. Returns Err if the conversation + /// does not exist. + async fn get_conversation_lock( + &self, + conversation_id: ConversationId, + ) -> Result>, EngineError> { + let map = self.conversations.read().await; + map.get(&conversation_id) + .map(Arc::clone) + .ok_or_else(|| EngineError::Store { + reason: format!("conversation {conversation_id} not found"), + }) + } // RwLockReadGuard dropped here + /// Restore persisted conversations for a user into the in-memory index. pub async fn bootstrap_user(&self, user_id: &str) -> Result { let conversations = self.store.list_conversations(user_id).await?; - let count = conversations.len(); let mut convs = self.conversations.write().await; let mut index = self.channel_user_index.write().await; + let mut inserted = 0usize; for conversation in conversations { + if convs.contains_key(&conversation.id) { + // Still upsert the index — it may be missing if a prior + // get_or_create_conversation inserted the conv but then rolled + // back the index entry on a failed save_conversation. + index + .entry((conversation.channel.clone(), conversation.user_id.clone())) + .or_insert(conversation.id); + continue; + } index.insert( (conversation.channel.clone(), conversation.user_id.clone()), conversation.id, ); - convs.insert(conversation.id, conversation); + convs.insert(conversation.id, Arc::new(Mutex::new(conversation))); + inserted += 1; } - Ok(count) + Ok(inserted) } /// Get or create a conversation for a channel+user pair. @@ -93,20 +132,48 @@ impl ConversationManager { let conv_id = conv.id; let mut convs = self.conversations.write().await; let mut index = self.channel_user_index.write().await; - convs.insert(conv_id, conv); + // Double-check: another task may have inserted while we did I/O. + if let Some(existing_id) = index.get(&key) { + return Ok(*existing_id); + } + convs.insert(conv_id, Arc::new(Mutex::new(conv))); index.insert(key, conv_id); return Ok(conv_id); } - // Create new conversation + // Create new conversation. let conv = ConversationSurface::new(channel, user_id); let conv_id = conv.id; - let mut convs = self.conversations.write().await; - let mut index = self.channel_user_index.write().await; - convs.insert(conv_id, conv.clone()); - index.insert(key, conv_id); - self.store.save_conversation(&conv).await?; + { + let mut convs = self.conversations.write().await; + let mut index = self.channel_user_index.write().await; + // Double-check: another task may have inserted while we did I/O. + if let Some(existing_id) = index.get(&key) { + return Ok(*existing_id); + } + convs.insert(conv_id, Arc::new(Mutex::new(conv.clone()))); + index.insert(key.clone(), conv_id); + } // write locks released before the async save + + if let Err(e) = self.store.save_conversation(&conv).await { + // Known limitation: a concurrent caller that observed the new conv_id via the + // double-check fast path (between our insert and this rollback) will hold a + // now-deleted, never-persisted ConversationId. This race requires simultaneous + // first-time logins from the same user+channel AND a store write failure — it + // is unlikely in practice and accepted as a structural trade-off of optimistic + // in-memory caching with async persistence. The alternative (holding write + // locks across the async save) would re-introduce cross-tenant serialization. + // Roll back the in-memory insertion so the next caller does not + // receive an unpersisted ConversationId. + let mut convs = self.conversations.write().await; + let mut index = self.channel_user_index.write().await; + convs.remove(&conv_id); + index.remove(&key); + return Err(EngineError::Store { + reason: e.to_string(), + }); + } debug!(conversation_id = %conv_id, channel, user_id, "created conversation"); Ok(conv_id) @@ -118,6 +185,10 @@ impl ConversationManager { /// injected into it. Otherwise, a new foreground thread is spawned. /// /// Returns the thread ID that is handling the message. + /// + /// The per-conversation `Mutex` is held for the entire operation — from + /// the active-thread check through `save_conversation`. This eliminates + /// the TOCTOU double-spawn window present in the old 5-phase split. pub async fn handle_user_message( &self, conversation_id: ConversationId, @@ -126,10 +197,8 @@ impl ConversationManager { user_id: &str, thread_config: ThreadConfig, ) -> Result { - let mut convs = self.conversations.write().await; - let conv = convs.get_mut(&conversation_id).ok_or(EngineError::Store { - reason: format!("conversation {conversation_id} not found"), - })?; + let conv_arc = self.get_conversation_lock(conversation_id).await?; + let mut conv = conv_arc.lock().await; // Tenant isolation: verify the requesting user owns this conversation. if conv.user_id != user_id { @@ -139,13 +208,17 @@ impl ConversationManager { }); } - // Record the user entry - conv.add_entry(ConversationEntry::user(content)); + // Snapshot what find_active_foreground needs before the async calls. + // NOTE: do NOT add the user entry yet — it will be added after the thread + // operation succeeds to avoid orphaned entries if the async op fails. + let active_thread_ids = conv.active_threads.clone(); + let channel_name = conv.channel.clone(); - // Check for an active foreground thread - let active_foreground = self.find_active_foreground(conv).await; + // Async I/O to find the active foreground thread — allowed here because + // we hold a tokio::sync::Mutex (not std::sync::Mutex). + let active_foreground = self.find_active_foreground(&active_thread_ids).await; - match active_foreground { + let thread_id = match active_foreground { Some(ActiveForeground::Running(thread_id)) => { debug!( conversation_id = %conversation_id, @@ -155,8 +228,7 @@ impl ConversationManager { self.thread_manager .inject_message(thread_id, user_id, ThreadMessage::user(content)) .await?; - self.store.save_conversation(conv).await?; - Ok(thread_id) + thread_id } Some(ActiveForeground::Resumable(thread_id)) => { debug!( @@ -173,18 +245,15 @@ impl ConversationManager { None, ) .await?; - conv.add_entry(ConversationEntry::system_for_thread( - thread_id, - "Thread resumed", - )); - self.store.save_conversation(conv).await?; - Ok(thread_id) + thread_id } None => { - // Build conversation history from prior entries for context continuity + // Build conversation history from prior entries for context continuity. + // Clone here (None branch only) — inject/resume paths don't need history, + // so deferring avoids an O(entries) allocation on those fast paths. let history = build_history_from_entries(&conv.entries); - // Spawn new foreground thread with conversation history + // Spawn new foreground thread with conversation history. let thread_id = self .thread_manager .spawn_thread_with_history( @@ -201,31 +270,52 @@ impl ConversationManager { // Store the base channel name in thread metadata so the // orchestrator can populate `source_channel` in the execution // context (used by mission_create to default notify_channels). - let base_channel = conv - .channel + let base_channel = channel_name .split(':') .next() - .unwrap_or(&conv.channel) + .unwrap_or(&channel_name) .to_string(); self.thread_manager .set_thread_metadata(thread_id, "source_channel", &base_channel) .await; + thread_id + } + }; + + // Final in-memory mutations under the already-held per-conv Mutex. + // The user entry is added here — after the thread operation succeeded — to + // prevent orphaned entries if inject_message/resume_thread/spawn_thread_with_history + // returned an error above. + conv.add_entry(ConversationEntry::user(content)); + match active_foreground { + Some(ActiveForeground::Running(_)) => { + // No additional in-memory mutation needed beyond the user entry above. + } + Some(ActiveForeground::Resumable(_)) => { + conv.add_entry(ConversationEntry::system_for_thread( + thread_id, + "Thread resumed", + )); + } + None => { conv.track_thread(thread_id); conv.add_entry(ConversationEntry::system_for_thread( thread_id, "Thread started", )); - self.store.save_conversation(conv).await?; - debug!( conversation_id = %conversation_id, thread_id = %thread_id, "spawned new foreground thread" ); - Ok(thread_id) } } + + // Persist outside the global RwLock (per-conv Mutex is still held). + self.store.save_conversation(&conv).await?; + + Ok(thread_id) } /// Record a thread's outcome in its conversation. @@ -235,50 +325,53 @@ impl ConversationManager { thread_id: ThreadId, outcome: &ThreadOutcome, ) -> Result<(), EngineError> { - let mut convs = self.conversations.write().await; - if let Some(conv) = convs.get_mut(&conversation_id) { - match outcome { - ThreadOutcome::Completed { response } => { - if let Some(text) = response { - conv.add_entry(ConversationEntry::agent(thread_id, text)); - } - conv.untrack_thread(thread_id); - } - ThreadOutcome::Stopped => { - conv.add_entry(ConversationEntry::system_for_thread( - thread_id, - "Thread stopped", - )); - conv.untrack_thread(thread_id); - } - ThreadOutcome::MaxIterations => { - conv.add_entry(ConversationEntry::system_for_thread( - thread_id, - "Thread reached max iterations", - )); - conv.untrack_thread(thread_id); - } - ThreadOutcome::Failed { error } => { - conv.add_entry(ConversationEntry::system_for_thread( - thread_id, - format!("Thread failed: {error}"), - )); - conv.untrack_thread(thread_id); - } - ThreadOutcome::GatePaused { - gate_name, - action_name, - .. - } => { - conv.add_entry(ConversationEntry::system_for_thread( - thread_id, - format!("Gate '{gate_name}' paused execution of action: {action_name}"), - )); - // Thread stays active — waiting for gate resolution + let conv_arc = self.get_conversation_lock(conversation_id).await?; + let mut conv = conv_arc.lock().await; + match outcome { + ThreadOutcome::Completed { response } => { + if let Some(text) = response { + conv.add_entry(ConversationEntry::agent(thread_id, text)); } + conv.untrack_thread(thread_id); + } + ThreadOutcome::Stopped => { + conv.add_entry(ConversationEntry::system_for_thread( + thread_id, + "Thread stopped", + )); + conv.untrack_thread(thread_id); + } + ThreadOutcome::MaxIterations => { + conv.add_entry(ConversationEntry::system_for_thread( + thread_id, + "Thread reached max iterations", + )); + conv.untrack_thread(thread_id); + } + ThreadOutcome::Failed { error } => { + conv.add_entry(ConversationEntry::system_for_thread( + thread_id, + format!("Thread failed: {error}"), + )); + conv.untrack_thread(thread_id); + } + ThreadOutcome::GatePaused { + gate_name, + action_name, + .. + } => { + conv.add_entry(ConversationEntry::system_for_thread( + thread_id, + format!("Gate '{gate_name}' paused execution of action: {action_name}"), + )); + // Thread stays active — waiting for gate resolution } - self.store.save_conversation(conv).await?; } + // Known limitation: if save_conversation fails, the in-memory mutations (add_entry, + // untrack_thread) are already applied but not persisted. Memory and DB diverge until + // the next successful save. Rolling back would require snapshotting the prior state, + // which is not implemented here — accepted as a low-probability failure mode. + self.store.save_conversation(&conv).await?; Ok(()) } @@ -291,21 +384,20 @@ impl ConversationManager { conversation_id: ConversationId, user_id: &str, ) -> Result<(), EngineError> { - let mut convs = self.conversations.write().await; - if let Some(conv) = convs.get_mut(&conversation_id) { - // Tenant isolation: verify ownership. - if conv.user_id != user_id { - return Err(EngineError::AccessDenied { - user_id: user_id.to_string(), - entity: format!("conversation {conversation_id}"), - }); - } - conv.active_threads.clear(); - conv.entries.clear(); - conv.updated_at = chrono::Utc::now(); - self.store.save_conversation(conv).await?; - debug!(conversation_id = %conversation_id, "cleared conversation"); + let conv_arc = self.get_conversation_lock(conversation_id).await?; + let mut conv = conv_arc.lock().await; + // Tenant isolation: verify ownership. + if conv.user_id != user_id { + return Err(EngineError::AccessDenied { + user_id: user_id.to_string(), + entity: format!("conversation {conversation_id}"), + }); } + conv.active_threads.clear(); + conv.entries.clear(); + conv.updated_at = chrono::Utc::now(); + self.store.save_conversation(&conv).await?; + debug!(conversation_id = %conversation_id, "cleared conversation"); Ok(()) } @@ -314,23 +406,47 @@ impl ConversationManager { &self, conversation_id: ConversationId, ) -> Option { - let convs = self.conversations.read().await; - convs.get(&conversation_id).cloned() + let arc = { + let convs = self.conversations.read().await; + convs.get(&conversation_id).map(Arc::clone) + }?; + Some(arc.lock().await.clone()) } - /// List all conversations for a user. + /// Returns conversations for the given user. + /// + /// Uses `channel_user_index` to pre-filter by user before acquiring any + /// per-conversation locks, keeping lock scope minimal. This is a best-effort + /// snapshot: each conversation is locked and read individually, so concurrent + /// mutations between locks may be partially visible. pub async fn list_conversations(&self, user_id: &str) -> Vec { - let convs = self.conversations.read().await; - convs - .values() - .filter(|c| c.user_id == user_id) - .cloned() - .collect() + let arcs: Vec>> = { + let convs = self.conversations.read().await; + let index = self.channel_user_index.read().await; + index + .iter() + .filter(|((_, uid), _)| uid == user_id) + .filter_map(|(_, id)| convs.get(id).cloned()) + .collect() + }; + let mut result = Vec::with_capacity(arcs.len()); + for arc in arcs { + result.push(arc.lock().await.clone()); + } + result } - /// Find an active foreground thread in a conversation. - async fn find_active_foreground(&self, conv: &ConversationSurface) -> Option { - for &tid in &conv.active_threads { + /// Find an active foreground thread given a snapshot of active thread IDs. + /// + /// Accepts a plain slice rather than a `&ConversationSurface` so callers + /// can drop the conversations write lock before invoking this method — + /// it performs async I/O (is_running, load_thread) that must not be held + /// under any lock. + async fn find_active_foreground( + &self, + active_thread_ids: &[ThreadId], + ) -> Option { + for &tid in active_thread_ids { if self.thread_manager.is_running(tid).await { return Some(ActiveForeground::Running(tid)); } @@ -343,27 +459,34 @@ impl ConversationManager { } None } + + /// Test helper: track a thread in a conversation without accessing the + /// internal HashMap directly. + #[cfg(test)] + pub async fn track_thread_in_conversation(&self, conv_id: ConversationId, thread_id: ThreadId) { + let arc = self + .get_conversation_lock(conv_id) + .await + .expect("conversation exists in test"); + arc.lock().await.track_thread(thread_id); + } } /// Build ThreadMessage history from conversation entries. /// /// Converts user and agent entries into ThreadMessages so a new thread /// inherits context from prior turns in the same conversation. +/// +/// The caller passes a snapshot taken *before* the current user message was +/// appended, so all entries here are prior-turn history — include them all. +/// System entries (thread lifecycle notifications) are skipped as they are not +/// useful LLM context. fn build_history_from_entries( entries: &[ConversationEntry], ) -> Vec { use crate::types::conversation::EntrySender; - // Skip the last entry (it's the current user message, added by the caller - // before this function runs). Also skip system entries (thread lifecycle - // notifications aren't useful as LLM context). - let history_entries = if entries.len() > 1 { - &entries[..entries.len() - 1] // safety: slice index on Vec, not a string — no UTF-8 concern - } else { - return Vec::new(); - }; - - history_entries + entries .iter() .filter_map(|entry| match &entry.sender { EntrySender::User => Some(crate::types::message::ThreadMessage::user(&entry.content)), @@ -706,11 +829,7 @@ mod tests { .unwrap(); store.save_thread(&thread).await.unwrap(); - { - let mut convs = cm.conversations.write().await; - let conv = convs.get_mut(&conv_id).unwrap(); - conv.track_thread(thread.id); - } + cm.track_thread_in_conversation(conv_id, thread.id).await; let resumed = cm .handle_user_message( @@ -735,11 +854,7 @@ mod tests { let tid = ThreadId::new(); // Manually track a thread - { - let mut convs = cm.conversations.write().await; - let conv = convs.get_mut(&conv_id).unwrap(); - conv.track_thread(tid); - } + cm.track_thread_in_conversation(conv_id, tid).await; // Record completion cm.record_thread_outcome( @@ -843,4 +958,81 @@ mod tests { assert!(conv.entries.is_empty()); assert!(conv.active_threads.is_empty()); } + + #[tokio::test] + async fn concurrent_handle_user_message_spawns_one_thread() { + // T1: Two concurrent handle_user_message calls on the same conversation + // must serialize — only ONE new thread should be spawned. + let (_, cm) = make_conv_manager(); + let conv_id = cm.get_or_create_conversation("web", "user1").await.unwrap(); + let project = ProjectId::new(); + let cm = Arc::new(cm); + + let cm1 = Arc::clone(&cm); + let cm2 = Arc::clone(&cm); + + let t1 = tokio::spawn(async move { + cm1.handle_user_message( + conv_id, + "message one", + project, + "user1", + ThreadConfig::default(), + ) + .await + }); + let t2 = tokio::spawn(async move { + cm2.handle_user_message( + conv_id, + "message two", + project, + "user1", + ThreadConfig::default(), + ) + .await + }); + + let r1 = t1.await.unwrap(); + let r2 = t2.await.unwrap(); + + // Both calls must succeed. + assert!(r1.is_ok(), "first handle_user_message failed: {r1:?}"); + assert!(r2.is_ok(), "second handle_user_message failed: {r2:?}"); + + // The per-conv Mutex serializes the two calls. The second call sees the + // first thread as Running (or the same thread ID if inject_message is used), + // so at most one NEW thread should exist in active_threads. + let conv = cm.get_conversation(conv_id).await.unwrap(); + assert_eq!( + conv.active_threads.len(), + 1, + "expected exactly 1 active thread, got {}: {:?}", + conv.active_threads.len(), + conv.active_threads + ); + } + + #[tokio::test] + async fn record_thread_outcome_unknown_conv_returns_err() { + // T4: After C1 fix, record_thread_outcome with an unknown ConversationId + // must return Err, not silently succeed. + let (_, cm) = make_conv_manager(); + let unknown_conv_id = ConversationId::new(); + let tid = ThreadId::new(); + + let result = cm + .record_thread_outcome( + unknown_conv_id, + tid, + &ThreadOutcome::Completed { + response: Some("irrelevant".into()), + }, + ) + .await; + + assert!( + result.is_err(), + "expected Err for unknown conversation, got Ok" + ); + } } diff --git a/src/config/database.rs b/src/config/database.rs index 55d8baea7f..cdb753921c 100644 --- a/src/config/database.rs +++ b/src/config/database.rs @@ -130,7 +130,7 @@ impl DatabaseConfig { hint: "Run 'ironclaw onboard' or set DATABASE_URL environment variable".to_string(), })?; - let pool_size = parse_optional_env("DATABASE_POOL_SIZE", 10)?; + let pool_size = parse_optional_env("DATABASE_POOL_SIZE", 30)?; let ssl_mode: SslMode = if let Some(s) = optional_env("DATABASE_SSLMODE")? { s.parse().map_err(|e| ConfigError::InvalidValue { diff --git a/src/workspace/mod.rs b/src/workspace/mod.rs index 199c9bf738..5c617f98d0 100644 --- a/src/workspace/mod.rs +++ b/src/workspace/mod.rs @@ -112,6 +112,38 @@ fn is_system_prompt_file(path: &str) -> bool { .any(|p| path.eq_ignore_ascii_case(p)) } +/// Returns `true` for engine runtime state paths that should never be chunked +/// or indexed for FTS/vector search. +/// +/// Covered prefixes / paths (all machine-generated blobs, not semantic docs): +/// - `engine/.runtime/` — execution-state blobs (threads, steps, events, leases, +/// conversations, compacted summaries) written by the bridge on every turn. +/// - `engine/projects/` — project and mission JSON files serialised on every +/// state mutation (e.g. `engine/projects/{slug}/project.json`, +/// `engine/projects/{slug}/missions/{slug}/mission.json`). +/// - `engine/orchestrator/failures.json` — orchestrator failure-tracker blob, +/// updated at engine-turn frequency. +/// +/// Semantic content that is intentionally KEPT indexed: +/// - `engine/knowledge/` — summaries, lessons, plans, specs, notes. +/// - `engine/orchestrator/v{N}.py` — versioned orchestrator code. +/// - `engine/orchestrator/*.md` — prompt overlays. +/// +/// Indexing the excluded paths floods the DB connection pool under +/// multi-tenant load. +fn is_engine_runtime_path(path: &str) -> bool { + // normalize_path() does not resolve '..' segments — this guard is + // load-bearing. Without it, `engine/.runtime/../knowledge/foo.md` + // would pass the starts_with check but refer to a semantic document. + !path.contains("..") + && (path.starts_with("engine/.runtime/") + || path.starts_with("engine/projects/") + || path == "engine/orchestrator/failures.json" + // Auto-generated per-workspace README — regenerated at engine-turn + // frequency; should not accumulate version rows. + || path == "engine/README.md") +} + /// Shared sanitizer instance — avoids rebuilding Aho-Corasick + regexes on every write. static SANITIZER: std::sync::LazyLock = std::sync::LazyLock::new(Sanitizer::new); @@ -1016,6 +1048,33 @@ impl Workspace { .get_or_create_document_by_path(&self.user_id, self.agent_id, &path) .await?; + // Engine runtime state files are execution-state blobs, not semantic + // documents. Skip the resolve_metadata DB query and all + // chunking/embedding work for them entirely. + if is_engine_runtime_path(&path) { + // One-time cleanup: delete any chunks that were created before this + // guard existed. This is a no-op once the document has no chunks, + // and prevents stale chunks from polluting search results or + // consuming storage indefinitely. + // Fail-open: chunk deletion failure must not block state writes. + let _ = self.storage.delete_chunks(doc.id).await; + + if doc.content == content { + return Ok(doc); + } + let skip_meta = DocumentMetadata { + skip_indexing: Some(true), + skip_versioning: Some(true), + ..Default::default() + }; + // Fail-open: versioning failures must not block state writes. + let _ = self + .maybe_save_version(doc.id, &doc.content, &skip_meta, Some(&self.user_id)) + .await; + self.storage.update_document(doc.id, content).await?; + return self.storage.get_document_by_id(doc.id).await; + } + // Short-circuit when content is unchanged: skip versioning and update, // but still reindex so metadata-driven flags (e.g. skip_indexing toggled // via the memory_write metadata param) take effect immediately. @@ -2860,4 +2919,65 @@ mod versioning_tests { assert_eq!(result.document.content, "hello world"); } + + // T2: engine/projects/ paths skip FTS/vector indexing; engine/knowledge/ paths do not. + #[tokio::test] + async fn engine_projects_path_skips_indexing_but_knowledge_does_not() { + let (ws, _dir) = create_test_workspace().await; + + // Write to an engine/projects/ path — should skip chunking entirely. + ws.write( + "engine/projects/test-proj--abc12345/project.json", + r#"{"id":"abc12345","name":"test-proj"}"#, + ) + .await + .unwrap(); + + // No chunks should exist for this document. + let chunks = ws + .storage + .get_chunks_without_embeddings("test_version", None, 100) + .await + .unwrap(); + assert!( + chunks.is_empty(), + "engine/projects/ write must not produce any chunks, got: {chunks:?}" + ); + + // Write to an engine/knowledge/ path — should be indexed normally. + ws.write( + "engine/knowledge/lessons/lesson-one--abc12345.md", + "This is a lesson learned from the last run.", + ) + .await + .unwrap(); + + // At least one chunk should now exist for the knowledge document. + let chunks = ws + .storage + .get_chunks_without_embeddings("test_version", None, 100) + .await + .unwrap(); + assert!( + !chunks.is_empty(), + "engine/knowledge/ write must produce chunks for FTS/vector indexing" + ); + } + + // T3: writes to engine/.runtime/ paths produce zero version rows. + #[tokio::test] + async fn runtime_path_writes_produce_no_versions() { + let (ws, _dir) = create_test_workspace().await; + + let path = "engine/.runtime/threads/test-thread.json"; + let doc = ws.write(path, "v1").await.unwrap(); + ws.write(path, "v2").await.unwrap(); + + let versions = ws.list_versions(doc.id, 50).await.unwrap(); + assert_eq!( + versions.len(), + 0, + "runtime path writes must not accumulate version rows, got: {versions:?}" + ); + } }