diff --git a/src/config/mcp_factory.rs b/src/config/mcp_factory.rs index 3405e0e..aaa4b14 100644 --- a/src/config/mcp_factory.rs +++ b/src/config/mcp_factory.rs @@ -81,11 +81,6 @@ pub struct McpFactory { } impl McpFactory { - pub fn try_get_active(&self, key: &McpServerKey) -> Option> { - let map = self.active.lock(); - map.get(key).and_then(|weak| weak.upgrade()) - } - pub fn insert_active(&self, key: McpServerKey, handle: &Arc) { let mut map = self.active.lock(); map.insert(key, Arc::downgrade(handle)); @@ -101,15 +96,21 @@ impl McpFactory { let key = McpServerKey::from_spec(name, spec); // Reuse of a live server fires no hooks; only a real spawn below - // reports anything. - if let Some(existing) = self.try_get_active(&key) { - return Ok(existing); - } - - // The live probe failed, so an entry still present for this key is a - // dead weak: the key was connected earlier in this process and the - // spawn below is a reconnect. - let reconnect = self.active.lock().contains_key(&key); + // reports anything. The live probe and the reconnect read share one + // lock section so a racing first connect that inserts between them + // cannot be misreported as a reconnect. The spawn await below stays + // outside the lock, so two racing first connects may still both + // spawn (no singleflight); the loser then reports a reconnect. + let reconnect = { + let map = self.active.lock(); + if let Some(existing) = map.get(&key).and_then(|weak| weak.upgrade()) { + return Ok(existing); + } + // The live probe failed, so an entry still present for this key + // is a dead weak: the key was connected earlier in this process + // and the spawn below is a reconnect. + map.contains_key(&key) + }; let transport = transport_label(&spec.transport_type); let handle = match spawn_server(name, spec, log_path).await { @@ -357,22 +358,6 @@ mod tests { } } - #[test] - fn factory_try_get_active_returns_none_when_empty() { - let factory = McpFactory::default(); - let spec = stdio_spec("cmd", None, None); - let key = McpServerKey::from_spec("s", &spec); - assert!(factory.try_get_active(&key).is_none()); - } - - #[test] - fn factory_try_get_active_returns_none_for_unknown_key() { - let factory = McpFactory::default(); - let spec = stdio_spec("cmd", None, None); - let key = McpServerKey::from_spec("s", &spec); - assert!(factory.try_get_active(&key).is_none()); - } - #[test] fn factory_default_has_empty_active_map() { let factory = McpFactory::default(); diff --git a/src/rag/mod.rs b/src/rag/mod.rs index 5c171eb..129962d 100644 --- a/src/rag/mod.rs +++ b/src/rag/mod.rs @@ -167,10 +167,17 @@ impl Rag { sync_hooks.fire_failed(&rag.name, &rag.path, &err); return Err(err); } - sync_hooks.fire_completed(&rag.name, &rag.path, rag.file_count()); - if rag.save()? { - println!("✓ Saved RAG to '{}'.", save_path.display()); + // rag.sync.completed promises a saved knowledge base, so it fires + // only after a successful save; a save error is a sync failure. + match rag.save() { + Ok(true) => println!("✓ Saved RAG to '{}'.", save_path.display()), + Ok(false) => {} + Err(err) => { + sync_hooks.fire_failed(&rag.name, &rag.path, &err); + return Err(err); + } } + sync_hooks.fire_completed(&rag.name, &rag.path, rag.file_count()); Ok(rag) } @@ -325,10 +332,17 @@ impl Rag { sync_hooks.fire_failed(&rag.name, &rag.path, &err); return Err(err); } - sync_hooks.fire_completed(&rag.name, &rag.path, rag.file_count()); - if rag.save()? { - println!("✓ Saved RAG to '{}'.", save_path.display()); + // rag.sync.completed promises a saved knowledge base, so it fires + // only after a successful save; a save error is a sync failure. + match rag.save() { + Ok(true) => println!("✓ Saved RAG to '{}'.", save_path.display()), + Ok(false) => {} + Err(err) => { + sync_hooks.fire_failed(&rag.name, &rag.path, &err); + return Err(err); + } } + sync_hooks.fire_completed(&rag.name, &rag.path, rag.file_count()); Ok(rag) } @@ -672,10 +686,17 @@ impl Rag { sync_hooks.fire_failed(&self.name, &self.path, &err); return Err(err); } - sync_hooks.fire_completed(&self.name, &self.path, self.file_count()); - if self.save()? { - println!("✓ Saved rag to '{}'.", self.path); + // rag.sync.completed promises a saved knowledge base, so it fires + // only after a successful save; a save error is a sync failure. + match self.save() { + Ok(true) => println!("✓ Saved rag to '{}'.", self.path), + Ok(false) => {} + Err(err) => { + sync_hooks.fire_failed(&self.name, &self.path, &err); + return Err(err); + } } + sync_hooks.fire_completed(&self.name, &self.path, self.file_count()); Ok(()) } @@ -2377,6 +2398,32 @@ mod tests { } } + /// An AppConfig whose client list can construct the "test-seeded" + /// embeddings client: init_with_config instantiates the client even + /// when a sync has nothing to embed. + fn seeded_client_app() -> AppConfig { + let mut client = ClientConfig::default(); + if let ClientConfig::OpenAIConfig(config) = &mut client { + config.name = Some("test-seeded".to_string()); + let mut embedder = ModelData::new("test-embedder"); + embedder.model_type = "embedding".to_string(); + config.models = vec![embedder]; + } + AppConfig { + clients: vec![client], + ..Default::default() + } + } + + fn init_config() -> RagInitConfig { + RagInitConfig { + embedding_model: Some("test-seeded:test-embedder".to_string()), + chunk_size: Some(1000), + chunk_overlap: Some(100), + ..Default::default() + } + } + fn yaml_rag(name: &str, path: &Path, data: RagData) -> Rag { Rag { app_config: Arc::new(AppConfig::default()), @@ -2499,6 +2546,133 @@ mod tests { ); } + #[tokio::test] + #[serial] + async fn init_with_config_fires_started_then_completed_after_save() { + let _sink = test_sink::install(); + let marker = "rag-init-ok-q7t"; + let dir = TempDir::new("init-hooks-ok"); + // An empty document syncs without embedding anything, so the build + // needs no live embeddings endpoint. + let doc = dir.path.join("doc.txt"); + fs::write(&doc, "").unwrap(); + let save_path = dir.path.join("kb.yaml"); + let sync_hooks = RagSyncHooks::resolve(&rag_hooks_ctx(marker)); + + let rag = Rag::init_with_config( + &seeded_client_app(), + "kb-init", + &save_path, + &[doc.display().to_string()], + &init_config(), + create_abort_signal(), + sync_hooks, + ) + .await + .unwrap(); + + assert_eq!(rag.file_count(), 1); + assert!( + save_path.exists(), + "completed must only fire after the knowledge base was saved" + ); + let captures = rag_captures(marker); + assert_eq!(captures.len(), 2, "{captures:?}"); + assert_eq!(captures[0].hook_name, format!("{marker}-rag.sync.started")); + assert_eq!( + captures[1].hook_name, + format!("{marker}-rag.sync.completed") + ); + assert_eq!( + captures[1] + .envs + .get("COYOTE_RAG_DOC_COUNT") + .map(String::as_str), + Some("1") + ); + } + + #[tokio::test] + #[serial] + async fn init_with_config_fires_failed_when_save_fails() { + let _sink = test_sink::install(); + let marker = "rag-init-save-fail-w2j"; + let dir = TempDir::new("init-hooks-save-fail"); + let doc = dir.path.join("doc.txt"); + fs::write(&doc, "").unwrap(); + // A regular file where save() expects a parent directory makes the + // save fail after a successful sync. + let blocker = dir.path.join("blocker"); + fs::write(&blocker, "not a directory").unwrap(); + let save_path = blocker.join("kb.yaml"); + let sync_hooks = RagSyncHooks::resolve(&rag_hooks_ctx(marker)); + + let result = Rag::init_with_config( + &seeded_client_app(), + "kb-init", + &save_path, + &[doc.display().to_string()], + &init_config(), + create_abort_signal(), + sync_hooks, + ) + .await; + + assert!( + result.is_err(), + "an unwritable save path must fail the init" + ); + let captures = rag_captures(marker); + assert_eq!(captures.len(), 2, "{captures:?}"); + assert_eq!(captures[0].hook_name, format!("{marker}-rag.sync.started")); + assert_eq!(captures[1].hook_name, format!("{marker}-rag.sync.failed")); + assert!( + captures[1] + .envs + .get("COYOTE_ERROR") + .is_some_and(|error| !error.is_empty()) + ); + } + + // The user-abort arm in abortable_run_with_spinner_rx only exists when + // stdout is a terminal, so a unit test cannot reach it; its + // bail!("Aborted.") surfaces through the same `synced` Err arm this + // sync error drives. + #[tokio::test] + #[serial] + async fn init_with_config_fires_failed_on_sync_error() { + let _sink = test_sink::install(); + let marker = "rag-init-fail-d4n"; + let dir = TempDir::new("init-hooks-fail"); + let missing = dir.path.join("missing.txt").display().to_string(); + let save_path = dir.path.join("kb.yaml"); + let sync_hooks = RagSyncHooks::resolve(&rag_hooks_ctx(marker)); + + let result = Rag::init_with_config( + &seeded_client_app(), + "kb-init", + &save_path, + &[missing], + &init_config(), + create_abort_signal(), + sync_hooks, + ) + .await; + + assert!(result.is_err(), "syncing a nonexistent document must fail"); + assert!(!save_path.exists()); + let captures = rag_captures(marker); + assert_eq!(captures.len(), 2, "{captures:?}"); + assert_eq!(captures[0].hook_name, format!("{marker}-rag.sync.started")); + assert_eq!(captures[1].hook_name, format!("{marker}-rag.sync.failed")); + assert!( + captures[1] + .envs + .get("COYOTE_ERROR") + .is_some_and(|error| !error.is_empty()) + ); + } + #[tokio::test] #[serial] async fn pure_load_fires_no_rag_sync_events() {