From 04b983459b7530c1536e9840ec3457530f0693fc Mon Sep 17 00:00:00 2001 From: Alex Clarke Date: Thu, 17 Sep 2026 13:16:05 -0600 Subject: [PATCH] feat(TASK-061): add rag.sync.* and mcp.server.* hook events plus COYOTE_ROLE - Register five new events (26 -> 31): rag.sync.started/completed/failed and mcp.server.connected/failed, whitelist-gated for agent global hooks like every non-escalation event - Bracket every real RAG document sync with rag.sync.* via pre-resolved RagSyncHooks snapshots carried into the build funnels; cache hits, pure loads, and attaches fire nothing, and a user abort mid-sync counts as a failure - Fire mcp.server.connected only on real spawns in McpFactory::acquire (reusing a live server is silent), marking respawns of a previously live key with COYOTE_MCP_RECONNECT; mcp.server.failed carries COYOTE_ERROR and COYOTE_MCP_AUTH_REQUIRED on auth failures. Boot-time registry connects deliberately fire nothing - Carry COYOTE_ROLE on session.* and turn.* events, resolved at fire time so mid-session role switches and temp roles are reflected --- src/config/agent.rs | 18 ++- src/config/mcp_factory.rs | 265 +++++++++++++++++++++++++++++++-- src/config/rag_cache.rs | 58 ++++++++ src/config/request_context.rs | 50 ++++++- src/function/agents.rs | 3 +- src/hooks.rs | 271 +++++++++++++++++++++++++++++++++- src/mcp/mod.rs | 4 + src/rag/mod.rs | 217 ++++++++++++++++++++++++++- src/repl/mod.rs | 137 +++++++++++++++-- 9 files changed, 986 insertions(+), 37 deletions(-) diff --git a/src/config/agent.rs b/src/config/agent.rs index 358d256..0451902 100644 --- a/src/config/agent.rs +++ b/src/config/agent.rs @@ -1,6 +1,6 @@ use super::*; -use crate::hooks::HooksMap; +use crate::hooks::{HooksMap, RagSyncHooks}; use crate::{ client::Model, config::memory, @@ -356,6 +356,16 @@ impl Agent { } }; + // Resolved from the config under construction: no context carries + // this agent yet, but its whitelist gate and hooks must still apply + // to the RAG builds below. + let rag_sync_hooks = RagSyncHooks::resolve_for_agent( + &app.hooks, + &agent_config.global_hooks, + &agent_config.hooks, + name, + ); + let rag = if rag_path.exists() { let key = RagKey::Agent(name.to_string()); let app_clone = app.clone(); @@ -388,6 +398,7 @@ impl Agent { let app_clone = app.clone(); let rag_path_clone = rag_path.clone(); let abort = abort_signal.clone(); + let sync_hooks = rag_sync_hooks.clone(); let rag = app_state .rag_cache .load_with(key, || async move { @@ -398,6 +409,7 @@ impl Agent { &document_paths, abort, true, + sync_hooks, ) .await }) @@ -421,6 +433,7 @@ impl Agent { &loaders, info_flag, abort_signal.clone(), + &rag_sync_hooks, ) .await? } @@ -1303,6 +1316,7 @@ async fn init_graph_rags( loaders: &HashMap, info_flag: bool, abort_signal: AbortSignal, + sync_hooks: &RagSyncHooks, ) -> Result>> { let mut rags = HashMap::new(); if info_flag { @@ -1379,6 +1393,7 @@ async fn init_graph_rags( let path_clone = rag_path.clone(); let name_clone = node_id.clone(); let abort = abort_signal.clone(); + let sync_hooks = sync_hooks.clone(); app_state .rag_cache .load_with(key, || async move { @@ -1389,6 +1404,7 @@ async fn init_graph_rags( &document_paths, &config, abort, + sync_hooks, ) .await }) diff --git a/src/config/mcp_factory.rs b/src/config/mcp_factory.rs index 23dba66..3405e0e 100644 --- a/src/config/mcp_factory.rs +++ b/src/config/mcp_factory.rs @@ -1,5 +1,6 @@ +use crate::hooks::McpServerHooks; use crate::mcp::{ - ConnectedServer, JsonField, McpAuthRequired, McpServer, McpTransportType, + ConnectedServer, HttpAuth, JsonField, McpAuthRequired, McpServer, McpTransportType, is_auth_required_error, resolve_http_auth, spawn_mcp_server, }; @@ -95,34 +96,91 @@ impl McpFactory { name: &str, spec: &McpServer, log_path: Option<&Path>, + hooks: &McpServerHooks, ) -> Result> { 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); } - let (auth, auth_reason) = resolve_http_auth(name, spec).await; - let handle = spawn_mcp_server(spec, log_path, auth).await.map_err(|e| { - if is_auth_required_error(&e) { - e.context(McpAuthRequired { - server: name.to_string(), - reason: auth_reason, - }) - } else { - e + // 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); + + let transport = transport_label(&spec.transport_type); + let handle = match spawn_server(name, spec, log_path).await { + Ok(handle) => handle, + Err(e) => { + hooks.fire_failed(name, transport, &e, is_auth_required_error(&e)); + return Err(e); } - })?; + }; self.insert_active(key, &handle); + hooks.fire_connected(name, transport, reconnect); Ok(handle) } } +fn transport_label(transport: &McpTransportType) -> &'static str { + match transport { + McpTransportType::Stdio => "stdio", + McpTransportType::Http => "http", + McpTransportType::Sse => "sse", + } +} + +async fn spawn_server( + name: &str, + spec: &McpServer, + log_path: Option<&Path>, +) -> Result> { + let (auth, auth_reason) = resolve_http_auth(name, spec).await; + spawn_transport(spec, log_path, auth).await.map_err(|e| { + if is_auth_required_error(&e) { + e.context(McpAuthRequired { + server: name.to_string(), + reason: auth_reason, + }) + } else { + e + } + }) +} + +async fn spawn_transport( + spec: &McpServer, + log_path: Option<&Path>, + auth: HttpAuth, +) -> Result> { + #[cfg(test)] + if let Some(result) = STUB_SPAWNS.lock().pop() { + return result; + } + spawn_mcp_server(spec, log_path, auth).await +} + +/// Injected spawn outcomes, popped instead of spawning a real transport: +/// nothing in a test environment speaks MCP over stdio or HTTP, so the +/// connect path is exercised with fixture handles. Empty means spawn for +/// real. +#[cfg(test)] +static STUB_SPAWNS: Mutex>>> = Mutex::new(Vec::new()); + #[cfg(test)] mod tests { use super::*; + use crate::config::test_fixtures::FixtureServer; + use crate::config::{AppConfig, AppState, RequestContext, WorkingMode}; + use crate::hooks::{HookDef, HooksMap, test_sink}; use crate::mcp::{JsonField, McpServer, McpTransportType}; + use anyhow::anyhow; use indexmap::IndexMap; + use rmcp::ServiceExt; + use serial_test::serial; use std::collections::HashMap; fn stdio_spec( @@ -321,4 +379,189 @@ mod tests { let map = factory.active.lock(); assert!(map.is_empty()); } + + /// Clears injected spawn outcomes on drop so a panicking test cannot + /// leak stubs into later tests. + struct StubSpawnsGuard; + + impl StubSpawnsGuard { + fn push(result: Result>) -> Self { + STUB_SPAWNS.lock().push(result); + StubSpawnsGuard + } + } + + impl Drop for StubSpawnsGuard { + fn drop(&mut self) { + STUB_SPAWNS.lock().clear(); + } + } + + /// An in-process client/server pair; the returned server half must stay + /// alive for the client handle to keep working. + async fn fixture_handle() -> (Arc, impl Sized) { + let (client_io, server_io) = tokio::io::duplex(4096); + let (server, client) = tokio::join!( + FixtureServer::default().serve(server_io), + ().serve(client_io) + ); + (Arc::new(client.unwrap()), server.unwrap()) + } + + fn ctx_with_mcp_hooks(marker: &str) -> RequestContext { + let mut hooks_map = HooksMap::default(); + for event in ["mcp.server.connected", "mcp.server.failed"] { + hooks_map.insert( + event.to_string(), + vec![HookDef { + name: format!("{marker}-{event}"), + command: "true".to_string(), + }], + ); + } + let mut app = AppState::test_default(); + app.config = Arc::new(AppConfig { + hooks: hooks_map, + ..Default::default() + }); + RequestContext::new(Arc::new(app), WorkingMode::Cmd) + } + + fn mcp_captures(marker: &str) -> Vec { + test_sink::snapshot() + .into_iter() + .filter(|capture| capture.hook_name.starts_with(marker)) + .collect() + } + + #[tokio::test] + #[serial] + async fn acquire_fires_connected_only_on_real_spawns_and_marks_reconnects() { + let _sink = test_sink::install(); + let marker = "mcp-spawn-k7d"; + let ctx = ctx_with_mcp_hooks(marker); + let hooks = McpServerHooks::resolve(&ctx); + let factory = McpFactory::default(); + let spec = stdio_spec("fixture-server-cmd", None, None); + + let (first, _first_server) = fixture_handle().await; + let stub = StubSpawnsGuard::push(Ok(first.clone())); + let connected = factory.acquire("srv", &spec, None, &hooks).await.unwrap(); + drop(stub); + assert!(Arc::ptr_eq(&connected, &first)); + let captures = mcp_captures(marker); + assert_eq!(captures.len(), 1, "{captures:?}"); + let capture = &captures[0]; + assert_eq!(capture.hook_name, format!("{marker}-mcp.server.connected")); + assert_eq!( + capture.envs.get("COYOTE_MCP_SERVER").map(String::as_str), + Some("srv") + ); + assert_eq!( + capture.envs.get("COYOTE_MCP_TRANSPORT").map(String::as_str), + Some("stdio") + ); + assert!( + !capture.envs.contains_key("COYOTE_MCP_RECONNECT"), + "a first connect must not be marked as a reconnect" + ); + + let reused = factory.acquire("srv", &spec, None, &hooks).await.unwrap(); + assert!(Arc::ptr_eq(&reused, &first)); + assert_eq!( + mcp_captures(marker).len(), + 1, + "reusing a live server must fire nothing" + ); + + // Every holder dropped: the weak in the factory dies, and the next + // acquire is a real spawn again — now marked as a reconnect. + drop(connected); + drop(reused); + drop(first); + let (second, _second_server) = fixture_handle().await; + let _stub = StubSpawnsGuard::push(Ok(second.clone())); + let respawned = factory.acquire("srv", &spec, None, &hooks).await.unwrap(); + assert!(Arc::ptr_eq(&respawned, &second)); + let captures = mcp_captures(marker); + assert_eq!(captures.len(), 2, "{captures:?}"); + assert_eq!( + captures[1] + .envs + .get("COYOTE_MCP_RECONNECT") + .map(String::as_str), + Some("true") + ); + } + + #[tokio::test] + #[serial] + async fn acquire_spawn_failure_fires_failed_with_the_error() { + let _sink = test_sink::install(); + let marker = "mcp-fail-v2q"; + let ctx = ctx_with_mcp_hooks(marker); + let hooks = McpServerHooks::resolve(&ctx); + let factory = McpFactory::default(); + // No stub: the nonexistent binary drives the real spawn-error path. + let spec = stdio_spec("coyote-nonexistent-mcp-binary-v2q", None, None); + + let result = factory.acquire("bad-srv", &spec, None, &hooks).await; + + assert!(result.is_err()); + let captures = mcp_captures(marker); + assert_eq!(captures.len(), 1, "{captures:?}"); + let capture = &captures[0]; + assert_eq!(capture.hook_name, format!("{marker}-mcp.server.failed")); + assert_eq!( + capture.envs.get("COYOTE_MCP_SERVER").map(String::as_str), + Some("bad-srv") + ); + assert_eq!( + capture.envs.get("COYOTE_MCP_TRANSPORT").map(String::as_str), + Some("stdio") + ); + assert!( + capture + .envs + .get("COYOTE_ERROR") + .is_some_and(|error| !error.is_empty()) + ); + assert!( + !capture.envs.contains_key("COYOTE_MCP_AUTH_REQUIRED"), + "a plain spawn failure must not be flagged as auth-required" + ); + } + + #[tokio::test] + #[serial] + async fn acquire_auth_failure_sets_the_auth_required_flag() { + let _sink = test_sink::install(); + let marker = "mcp-auth-h4n"; + let ctx = ctx_with_mcp_hooks(marker); + let hooks = McpServerHooks::resolve(&ctx); + let factory = McpFactory::default(); + let spec = stdio_spec("oauth-cmd", None, None); + let _stub = StubSpawnsGuard::push(Err(anyhow!("Auth required: no stored token"))); + + let err = factory + .acquire("oauth-srv", &spec, None, &hooks) + .await + .unwrap_err(); + + assert!( + err.downcast_ref::().is_some(), + "the auth error must be wrapped in McpAuthRequired context" + ); + let captures = mcp_captures(marker); + assert_eq!(captures.len(), 1, "{captures:?}"); + let capture = &captures[0]; + assert_eq!(capture.hook_name, format!("{marker}-mcp.server.failed")); + assert_eq!( + capture + .envs + .get("COYOTE_MCP_AUTH_REQUIRED") + .map(String::as_str), + Some("true") + ); + } } diff --git a/src/config/rag_cache.rs b/src/config/rag_cache.rs index 5a43eb6..73c1d3d 100644 --- a/src/config/rag_cache.rs +++ b/src/config/rag_cache.rs @@ -47,3 +47,61 @@ impl RagCache { Ok(arc) } } + +#[cfg(test)] +mod tests { + use super::*; + use crate::config::AppConfig; + use crate::hooks::test_sink; + use crate::rag::RagData; + use anyhow::anyhow; + use serial_test::serial; + use std::time::{SystemTime, UNIX_EPOCH}; + + /// A cache hit must never reach the loader: the loader closure is where + /// a real RAG build (and its `rag.sync.*` hook dispatch) happens, so + /// returning the cached entry is what keeps warm-cache lookups silent. + #[tokio::test] + #[serial] + async fn cache_hit_returns_the_cached_rag_without_running_the_loader() { + let _sink = test_sink::install(); + test_sink::drain(); + let unique = SystemTime::now() + .duration_since(UNIX_EPOCH) + .unwrap() + .as_nanos(); + let dir = std::env::temp_dir().join(format!("coyote-rag-cache-hit-{unique}")); + std::fs::create_dir_all(&dir).unwrap(); + let path = dir.join("kb.yaml"); + let data = RagData { + driver: "yaml".to_string(), + // The only embedding model resolvable in tests; see + // `seed_model_registries` in the request_context tests. + embedding_model: "test-seeded:test-embedder".to_string(), + chunk_size: 1000, + chunk_overlap: 100, + top_k: 5, + ..Default::default() + }; + std::fs::write(&path, serde_yaml::to_string(&data).unwrap()).unwrap(); + let rag = Arc::new(Rag::load(&AppConfig::default(), "kb", &path).await.unwrap()); + let _ = std::fs::remove_dir_all(&dir); + + let cache = RagCache::default(); + let key = RagKey::Named("kb".to_string()); + cache.insert(key.clone(), &rag); + + let loaded = cache + .load_with(key, || async { + Err(anyhow!("a cache hit must not rebuild")) + }) + .await + .unwrap(); + + assert!(Arc::ptr_eq(&loaded, &rag)); + assert!( + test_sink::drain().is_empty(), + "a cache hit must dispatch no hooks" + ); + } +} diff --git a/src/config/request_context.rs b/src/config/request_context.rs index 0f791a3..ae47738 100644 --- a/src/config/request_context.rs +++ b/src/config/request_context.rs @@ -1077,7 +1077,12 @@ impl RequestContext { // COYOTE_SESSION_ID and session-held role hooks still resolve; no // session means no event. if self.session.is_some() { - hooks::fire(HookEvent::SessionEnded, self, &[], None); + hooks::fire( + HookEvent::SessionEnded, + self, + &hooks::role_extras(self), + None, + ); } if let Some(mut session) = self.session.take() { let sessions_dir = self.sessions_dir(); @@ -4406,6 +4411,7 @@ impl RequestContext { if !server_ids.is_empty() { let app_ref = &self.app; + let mcp_hooks = hooks::McpServerHooks::resolve(self); let acquire_all = async { let mut handles = Vec::new(); let mut auth_required = Vec::new(); @@ -4413,7 +4419,7 @@ impl RequestContext { if let Some(spec) = mcp_config.mcp_servers.get(id) { match app_ref .mcp_factory - .acquire(id, spec, app_ref.mcp_log_path.as_deref()) + .acquire(id, spec, app_ref.mcp_log_path.as_deref(), &mcp_hooks) .await { Ok(handle) => handles.push((id.clone(), handle)), @@ -4738,9 +4744,19 @@ impl RequestContext { self.refresh_mcp_tool_filters(); self.init_agent_session_variables(new_session)?; if created_new_session { - hooks::fire(HookEvent::SessionStarted, self, &[], None); + hooks::fire( + HookEvent::SessionStarted, + self, + &hooks::role_extras(self), + None, + ); } else { - hooks::fire(HookEvent::SessionResumed, self, &[], None); + hooks::fire( + HookEvent::SessionResumed, + self, + &hooks::role_extras(self), + None, + ); } Ok(()) } @@ -5334,6 +5350,7 @@ impl RequestContext { let vault = self.app.vault.clone(); let rag_cache = self.rag_cache(); let working_mode = self.working_mode; + let sync_hooks = hooks::RagSyncHooks::resolve(self); let (rag, rag_key): (Arc, Option) = match rag { None => { @@ -5352,6 +5369,7 @@ impl RequestContext { &[], abort_signal.clone(), false, + sync_hooks, ) .await?, ), @@ -5373,8 +5391,16 @@ impl RequestContext { if working_mode.is_cmd() { bail!("Unknown RAG '{name}'"); } - Rag::init(&app, name, &rag_path, &[], abort_signal.clone(), true) - .await + Rag::init( + &app, + name, + &rag_path, + &[], + abort_signal.clone(), + true, + sync_hooks, + ) + .await } else { Rag::load_async(&app, &vault, name, &rag_path).await } @@ -5458,6 +5484,7 @@ impl RequestContext { false, &self.app.config, abort_signal, + hooks::RagSyncHooks::resolve(self), ) .await?; self.rag = Some(Arc::new(rag)); @@ -5488,8 +5515,15 @@ impl RequestContext { This will call the embedding API and may take a while.", rag.file_count() ); - rag.refresh_document_paths(&document_paths, true, true, &self.app.config, abort_signal) - .await?; + rag.refresh_document_paths( + &document_paths, + true, + true, + &self.app.config, + abort_signal, + hooks::RagSyncHooks::resolve(self), + ) + .await?; self.rag = Some(Arc::new(rag)); Ok(()) } diff --git a/src/function/agents.rs b/src/function/agents.rs index e1b92fe..24b2063 100644 --- a/src/function/agents.rs +++ b/src/function/agents.rs @@ -961,10 +961,11 @@ async fn populate_agent_mcp_runtime(ctx: &mut RequestContext, server_ids: &[Stri }) .unwrap_or_default(); + let mcp_hooks = hooks::McpServerHooks::resolve(ctx); for (id, spec) in server_specs { let handle = app .mcp_factory - .acquire(&id, &spec, app.mcp_log_path.as_deref()) + .acquire(&id, &spec, app.mcp_log_path.as_deref(), &mcp_hooks) .await?; ctx.tool_scope.mcp_runtime.insert(id, handle); } diff --git a/src/hooks.rs b/src/hooks.rs index efb19c6..7c76a0a 100644 --- a/src/hooks.rs +++ b/src/hooks.rs @@ -132,6 +132,11 @@ pub enum HookEvent { JobStarted, JobCompleted, JobFailed, + RagSyncStarted, + RagSyncCompleted, + RagSyncFailed, + McpServerConnected, + McpServerFailed, } impl HookEvent { @@ -163,10 +168,15 @@ impl HookEvent { HookEvent::JobStarted => "job.started", HookEvent::JobCompleted => "job.completed", HookEvent::JobFailed => "job.failed", + HookEvent::RagSyncStarted => "rag.sync.started", + HookEvent::RagSyncCompleted => "rag.sync.completed", + HookEvent::RagSyncFailed => "rag.sync.failed", + HookEvent::McpServerConnected => "mcp.server.connected", + HookEvent::McpServerFailed => "mcp.server.failed", } } - const ALL: [HookEvent; 26] = [ + const ALL: [HookEvent; 31] = [ HookEvent::TurnStarted, HookEvent::TurnCompleted, HookEvent::TurnInterrupted, @@ -193,6 +203,11 @@ impl HookEvent { HookEvent::JobStarted, HookEvent::JobCompleted, HookEvent::JobFailed, + HookEvent::RagSyncStarted, + HookEvent::RagSyncCompleted, + HookEvent::RagSyncFailed, + HookEvent::McpServerConnected, + HookEvent::McpServerFailed, ]; } @@ -208,6 +223,191 @@ pub struct ResolvedHook { pub cwd: PathBuf, } +/// Pre-resolved `rag.sync.*` hooks plus the names `base_envs_parts` needs, +/// carried by value into the context-free RAG build funnels (`Rag::init`, +/// `Rag::init_with_config`, `Rag::refresh_document_paths`). Resolution +/// happens where the caller's identity is known; dispatch happens inside the +/// funnel, bracketing the one real sync it performs — so cache hits, pure +/// loads, and attaches (which never enter a funnel) fire nothing. +#[derive(Debug, Clone, Default)] +pub struct RagSyncHooks { + started: Vec, + completed: Vec, + failed: Vec, + session_name: Option, + agent_name: Option, +} + +impl RagSyncHooks { + pub fn resolve(ctx: &RequestContext) -> Self { + Self { + started: ctx.resolved_hooks(HookEvent::RagSyncStarted), + completed: ctx.resolved_hooks(HookEvent::RagSyncCompleted), + failed: ctx.resolved_hooks(HookEvent::RagSyncFailed), + session_name: ctx + .session + .as_ref() + .map(|session| session.name().to_string()), + agent_name: ctx.agent.as_ref().map(|agent| agent.name().to_string()), + } + } + + /// Resolution for `Agent::init`, which builds agent and graph-node RAGs + /// before any context carries the agent: the whitelist gate and the + /// agent's own hooks come straight from the config being constructed. + /// No session or role exists at that point. + pub fn resolve_for_agent( + global_hooks: &HooksMap, + gate: &[String], + agent_hooks: &HooksMap, + agent_name: &str, + ) -> Self { + let resolve = |event| { + resolve_hooks( + event, + global_hooks, + Some((gate, agent_name)), + None, + Some((agent_hooks, agent_name)), + ) + }; + Self { + started: resolve(HookEvent::RagSyncStarted), + completed: resolve(HookEvent::RagSyncCompleted), + failed: resolve(HookEvent::RagSyncFailed), + session_name: None, + agent_name: Some(agent_name.to_string()), + } + } + + pub fn fire_started(&self, rag_name: &str, rag_path: &str) { + self.fire( + HookEvent::RagSyncStarted, + &self.started, + rag_name, + rag_path, + &[], + ); + } + + pub fn fire_completed(&self, rag_name: &str, rag_path: &str, doc_count: usize) { + self.fire( + HookEvent::RagSyncCompleted, + &self.completed, + rag_name, + rag_path, + &[("COYOTE_RAG_DOC_COUNT", doc_count.to_string())], + ); + } + + /// A user abort mid-sync counts as a failure: the funnel's sync came + /// back `Err` either way, and the knowledge base was not (re)built. + pub fn fire_failed(&self, rag_name: &str, rag_path: &str, error: &anyhow::Error) { + self.fire( + HookEvent::RagSyncFailed, + &self.failed, + rag_name, + rag_path, + &[("COYOTE_ERROR", format!("{error:#}"))], + ); + } + + fn fire( + &self, + event: HookEvent, + resolved: &[ResolvedHook], + rag_name: &str, + rag_path: &str, + extras: &[(&str, String)], + ) { + if resolved.is_empty() { + return; + } + let base_envs = base_envs_parts( + event, + self.session_name.as_deref(), + self.agent_name.as_deref(), + ); + let mut envs = vec![ + ("COYOTE_RAG_NAME", rag_name.to_string()), + ("COYOTE_RAG_PATH", rag_path.to_string()), + ]; + envs.extend(extras.iter().map(|(key, value)| (*key, value.clone()))); + fire_resolved(event, resolved.to_vec(), base_envs, &envs, None); + } +} + +/// Pre-resolved `mcp.server.*` hooks plus the names `base_envs_parts` needs, +/// carried into the context-free `McpFactory::acquire`. Dispatch happens at +/// the spawn itself — one event per real spawn — so handing back an +/// already-live server fires nothing. +#[derive(Debug, Clone, Default)] +pub struct McpServerHooks { + connected: Vec, + failed: Vec, + session_name: Option, + agent_name: Option, +} + +impl McpServerHooks { + pub fn resolve(ctx: &RequestContext) -> Self { + Self { + connected: ctx.resolved_hooks(HookEvent::McpServerConnected), + failed: ctx.resolved_hooks(HookEvent::McpServerFailed), + session_name: ctx + .session + .as_ref() + .map(|session| session.name().to_string()), + agent_name: ctx.agent.as_ref().map(|agent| agent.name().to_string()), + } + } + + /// `reconnect` marks a spawn for a server key that was live earlier in + /// this process; the variable is omitted entirely on a first connect. + pub fn fire_connected(&self, server: &str, transport: &str, reconnect: bool) { + let mut extras = vec![ + ("COYOTE_MCP_SERVER", server.to_string()), + ("COYOTE_MCP_TRANSPORT", transport.to_string()), + ]; + if reconnect { + extras.push(("COYOTE_MCP_RECONNECT", "true".to_string())); + } + self.fire(HookEvent::McpServerConnected, &self.connected, &extras); + } + + /// `auth_required` marks an `McpAuthRequired` failure; the variable is + /// omitted on every other error. + pub fn fire_failed( + &self, + server: &str, + transport: &str, + error: &anyhow::Error, + auth_required: bool, + ) { + let mut extras = vec![ + ("COYOTE_MCP_SERVER", server.to_string()), + ("COYOTE_MCP_TRANSPORT", transport.to_string()), + ("COYOTE_ERROR", format!("{error:#}")), + ]; + if auth_required { + extras.push(("COYOTE_MCP_AUTH_REQUIRED", "true".to_string())); + } + self.fire(HookEvent::McpServerFailed, &self.failed, &extras); + } + + fn fire(&self, event: HookEvent, resolved: &[ResolvedHook], extras: &[(&str, String)]) { + if resolved.is_empty() { + return; + } + let base_envs = base_envs_parts( + event, + self.session_name.as_deref(), + self.agent_name.as_deref(), + ); + fire_resolved(event, resolved.to_vec(), base_envs, extras, None); + } +} + impl RequestContext { /// Hooks that apply to this context for `event`, in firing order: global /// (whitelisted through `global_hooks` when an agent is active), then @@ -510,6 +710,21 @@ pub(crate) fn base_envs_parts( envs } +/// `COYOTE_ROLE` extras for the `session.*` and `turn.*` fire sites — the +/// only two families that carry the variable. Resolved at fire time, because +/// roles change mid-session (`.role`, `.exit role`, temp roles): the role +/// held directly on the context wins, then the name of a role a session has +/// absorbed. Derived roles have no name and report nothing. +pub fn role_extras(ctx: &RequestContext) -> Vec<(&'static str, String)> { + ctx.role + .as_ref() + .map(|role| role.name()) + .filter(|name| !name.is_empty()) + .or_else(|| ctx.session.as_ref().and_then(|session| session.role_name())) + .map(|name| vec![("COYOTE_ROLE", name.to_string())]) + .unwrap_or_default() +} + async fn write_payload_file(path: &Path, json: &str) -> std::io::Result<()> { let mut options = tokio::fs::OpenOptions::new(); options.write(true).create_new(true); @@ -1405,6 +1620,11 @@ mod tests { (HookEvent::JobStarted, "job.started"), (HookEvent::JobCompleted, "job.completed"), (HookEvent::JobFailed, "job.failed"), + (HookEvent::RagSyncStarted, "rag.sync.started"), + (HookEvent::RagSyncCompleted, "rag.sync.completed"), + (HookEvent::RagSyncFailed, "rag.sync.failed"), + (HookEvent::McpServerConnected, "mcp.server.connected"), + (HookEvent::McpServerFailed, "mcp.server.failed"), ]; for (event, name) in cases { assert_eq!(event.as_str(), name); @@ -1416,6 +1636,55 @@ mod tests { assert_eq!(HookEvent::ALL.len(), cases.len()); } + #[test] + fn rag_and_mcp_wildcard_whitelist_entries_are_valid() { + for event in [ + "rag.sync.started", + "rag.sync.completed", + "rag.sync.failed", + "mcp.server.connected", + "mcp.server.failed", + ] { + assert!( + entry_is_valid_wildcard(&format!("{event}.*")), + "'{event}.*' must validate as an .* wildcard" + ); + } + // Family prefixes are not events: like `tool.*`, these are partial + // globs that admit nothing. + assert!(!entry_is_valid_wildcard("rag.sync.*")); + assert!(!entry_is_valid_wildcard("mcp.server.*")); + } + + #[test] + fn rag_and_mcp_events_stay_behind_the_agent_whitelist_gate() { + let mut global = hooks_map("rag.sync.started", &[("watch", "cmd-rag")]); + global.extend(hooks_map("mcp.server.connected", &[("watch", "cmd-mcp")])); + + for event in [HookEvent::RagSyncStarted, HookEvent::McpServerConnected] { + let resolved = resolve_hooks(event, &global, Some((&[], "gated-agent")), None, None); + assert!( + resolved.is_empty(), + "{} must stay whitelist-gated for agents", + event.as_str() + ); + } + + let gate = vec![ + "rag.sync.started.*".to_string(), + "mcp.server.connected.*".to_string(), + ]; + for event in [HookEvent::RagSyncStarted, HookEvent::McpServerConnected] { + let resolved = resolve_hooks(event, &global, Some((&gate, "gated-agent")), None, None); + assert_eq!( + names(&resolved), + ["watch"], + "an `.*` whitelist entry must admit {}", + event.as_str() + ); + } + } + #[test] fn root_context_gets_all_global_hooks() { let mut global = hooks_map("turn.started", &[("a", "cmd-a"), ("b", "cmd-b")]); diff --git a/src/mcp/mod.rs b/src/mcp/mod.rs index 72f3930..fe3059a 100644 --- a/src/mcp/mod.rs +++ b/src/mcp/mod.rs @@ -401,6 +401,10 @@ impl McpRegistry { Ok(()) } + /// Boot-time connects fire no `mcp.server.*` hooks: this runs from + /// `AppState::new`, before any `RequestContext` exists to resolve hook + /// scopes against. Hook-observed connects go through + /// `McpFactory::acquire`. async fn start_server(&self, id: String) -> Result)>> { let spec = self .config diff --git a/src/rag/mod.rs b/src/rag/mod.rs index 6f5c627..5c171eb 100644 --- a/src/rag/mod.rs +++ b/src/rag/mod.rs @@ -13,6 +13,7 @@ mod splitter; use self::graph::{KnowledgeGraph, extract_entities}; use self::provider::RagProvider; use self::providers::{DuckDbProvider, QdrantProvider, YamlProvider}; +use crate::hooks::RagSyncHooks; use crate::sandbox::mcp_credentials; use crate::vault::{Vault, interpolate_secrets}; @@ -144,6 +145,7 @@ impl Rag { doc_paths: &[String], config: &RagInitConfig, abort_signal: AbortSignal, + sync_hooks: RagSyncHooks, ) -> Result { if doc_paths.is_empty() { bail!("Cannot build RAG knowledge base '{name}' with no documents"); @@ -154,12 +156,18 @@ impl Rag { let mut rag = Self::create(app, name, save_path, data).await?; let loaders = app.document_loaders.clone(); let (spinner, spinner_rx) = Spinner::create(""); - abortable_run_with_spinner_rx( + sync_hooks.fire_started(&rag.name, &rag.path); + let synced = abortable_run_with_spinner_rx( rag.sync_documents(doc_paths, true, false, loaders, Some(spinner)), spinner_rx, abort_signal, ) - .await?; + .await; + if let Err(err) = synced { + 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()); } @@ -261,6 +269,7 @@ impl Rag { doc_paths: &[String], abort_signal: AbortSignal, prompt_for_driver: bool, + sync_hooks: RagSyncHooks, ) -> Result { if !*IS_STDOUT_TERMINAL { bail!("Failed to init rag in non-interactive mode"); @@ -305,12 +314,18 @@ impl Rag { }; let loaders = app.document_loaders.clone(); let (spinner, spinner_rx) = Spinner::create(""); - abortable_run_with_spinner_rx( + sync_hooks.fire_started(&rag.name, &rag.path); + let synced = abortable_run_with_spinner_rx( rag.sync_documents(&paths, true, false, loaders, Some(spinner)), spinner_rx, abort_signal, ) - .await?; + .await; + if let Err(err) = synced { + 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()); } @@ -636,10 +651,12 @@ impl Rag { force_reingest: bool, app: &AppConfig, abort_signal: AbortSignal, + sync_hooks: RagSyncHooks, ) -> Result<()> { let loaders = app.document_loaders.clone(); let (spinner, spinner_rx) = Spinner::create(""); - abortable_run_with_spinner_rx( + sync_hooks.fire_started(&self.name, &self.path); + let synced = abortable_run_with_spinner_rx( self.sync_documents( document_paths, refresh, @@ -650,7 +667,12 @@ impl Rag { spinner_rx, abort_signal, ) - .await?; + .await; + if let Err(err) = synced { + 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); } @@ -2280,6 +2302,8 @@ where #[cfg(test)] mod tests { use super::*; + use crate::hooks::{HookDef, HooksMap, test_sink}; + use serial_test::serial; use std::path::PathBuf; use std::sync::atomic::{AtomicU64, Ordering}; use std::time::{SystemTime, UNIX_EPOCH}; @@ -2313,6 +2337,187 @@ mod tests { } } + fn rag_hooks_ctx(marker: &str) -> RequestContext { + let mut hooks_map = HooksMap::default(); + for event in ["rag.sync.started", "rag.sync.completed", "rag.sync.failed"] { + hooks_map.insert( + event.to_string(), + vec![HookDef { + name: format!("{marker}-{event}"), + command: "true".to_string(), + }], + ); + } + let mut app = AppState::test_default(); + app.config = Arc::new(AppConfig { + hooks: hooks_map, + ..Default::default() + }); + RequestContext::new(Arc::new(app), WorkingMode::Cmd) + } + + fn rag_captures(marker: &str) -> Vec { + test_sink::snapshot() + .into_iter() + .filter(|capture| capture.hook_name.starts_with(marker)) + .collect() + } + + fn buildable_data() -> RagData { + RagData { + driver: "yaml".to_string(), + // The only embedding model resolvable in tests: the model + // registry is a process-global OnceLock pre-seeded by + // `seed_model_registries` in the request_context tests. + embedding_model: "test-seeded:test-embedder".to_string(), + chunk_size: 1000, + chunk_overlap: 100, + top_k: 5, + ..Default::default() + } + } + + fn yaml_rag(name: &str, path: &Path, data: RagData) -> Rag { + Rag { + app_config: Arc::new(AppConfig::default()), + name: name.to_string(), + path: path.display().to_string(), + embedding_model: Model::new("openai", "text-embedding-3-small"), + bm25: data.build_bm25(), + provider: Box::new(YamlProvider::from_data(&data)), + node_to_docs: IndexMap::new(), + data, + last_sources: RwLock::new(None), + } + } + + #[tokio::test] + #[serial] + async fn refresh_document_paths_fires_started_then_completed_with_doc_count() { + let _sink = test_sink::install(); + let marker = "rag-sync-ok-v5k"; + let dir = TempDir::new("sync-hooks-ok"); + let doc = dir.path.join("doc.txt"); + fs::write(&doc, "hello rag hooks").unwrap(); + let doc_path = doc.display().to_string(); + + // The prepopulated file entry must use the exact string + // sync_documents resolves, so the sync sees an unchanged corpus and + // never needs an embeddings client. + let loaders = HashMap::new(); + let (_, _, _, _, local_paths) = resolve_paths(&loaders, std::slice::from_ref(&doc_path)) + .await + .unwrap(); + let resolved_path = local_paths.iter().next().unwrap().clone(); + + let mut data = buildable_data(); + data.next_file_id = 1; + data.files.insert( + 0, + RagFile { + hash: sha256("hello rag hooks"), + path: resolved_path, + documents: vec![], + }, + ); + let rag_path = dir.path.join("kb.yaml"); + let rag_path_display = rag_path.display().to_string(); + let mut rag = yaml_rag("kb", &rag_path, data); + let sync_hooks = RagSyncHooks::resolve(&rag_hooks_ctx(marker)); + + rag.refresh_document_paths( + &[doc_path], + false, + false, + &AppConfig::default(), + create_abort_signal(), + sync_hooks, + ) + .await + .unwrap(); + + 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[0].envs.get("COYOTE_RAG_NAME").map(String::as_str), + Some("kb") + ); + assert_eq!( + captures[0].envs.get("COYOTE_RAG_PATH").map(String::as_str), + Some(rag_path_display.as_str()) + ); + 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 refresh_document_paths_fires_started_then_failed_on_sync_error() { + let _sink = test_sink::install(); + let marker = "rag-sync-fail-b8m"; + let dir = TempDir::new("sync-hooks-fail"); + let missing = dir.path.join("missing.txt").display().to_string(); + let rag_path = dir.path.join("kb.yaml"); + let mut rag = yaml_rag("kb", &rag_path, buildable_data()); + let sync_hooks = RagSyncHooks::resolve(&rag_hooks_ctx(marker)); + + let result = rag + .refresh_document_paths( + &[missing], + false, + false, + &AppConfig::default(), + create_abort_signal(), + sync_hooks, + ) + .await; + + assert!(result.is_err(), "syncing a nonexistent document must fail"); + 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_eq!( + captures[1].envs.get("COYOTE_RAG_NAME").map(String::as_str), + Some("kb") + ); + 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() { + let _sink = test_sink::install(); + test_sink::drain(); + let dir = TempDir::new("sync-hooks-load"); + let path = dir.path.join("kb.yaml"); + fs::write(&path, serde_yaml::to_string(&buildable_data()).unwrap()).unwrap(); + + let rag = Rag::load(&AppConfig::default(), "kb", &path).await.unwrap(); + + assert_eq!(rag.file_count(), 0); + let captures = test_sink::drain(); + assert!( + captures.is_empty(), + "a pure load must dispatch no hooks: {captures:?}" + ); + } + #[test] fn attached_ragdata_serializes_driver_and_attached() { let mut data = RagData { diff --git a/src/repl/mod.rs b/src/repl/mod.rs index 58ab8b5..4e46b59 100644 --- a/src/repl/mod.rs +++ b/src/repl/mod.rs @@ -68,7 +68,7 @@ pub(crate) struct TurnBracket(()); impl TurnBracket { #[must_use] pub(crate) fn enter(ctx: &RequestContext) -> Self { - hooks::fire(HookEvent::TurnStarted, ctx, &[], None); + hooks::fire(HookEvent::TurnStarted, ctx, &hooks::role_extras(ctx), None); Self(()) } @@ -79,16 +79,25 @@ impl TurnBracket { result: Result, ) -> Result { if abort_signal.aborted_ctrlc() { - hooks::fire(HookEvent::TurnInterrupted, ctx, &[], None); + hooks::fire( + HookEvent::TurnInterrupted, + ctx, + &hooks::role_extras(ctx), + None, + ); } else { match &result { - Ok(_) => hooks::fire(HookEvent::TurnCompleted, ctx, &[], None), - Err(err) => hooks::fire( - HookEvent::TurnFailed, + Ok(_) => hooks::fire( + HookEvent::TurnCompleted, ctx, - &[("COYOTE_ERROR", format!("{err:#}"))], + &hooks::role_extras(ctx), None, ), + Err(err) => { + let mut extras = hooks::role_extras(ctx); + extras.push(("COYOTE_ERROR", format!("{err:#}"))); + hooks::fire(HookEvent::TurnFailed, ctx, &extras, None); + } } } result @@ -1143,7 +1152,12 @@ pub async fn run_repl_command( abort_signal.clone(), ) .await?; - hooks::fire(HookEvent::SessionCompressed, ctx, &[], None); + hooks::fire( + HookEvent::SessionCompressed, + ctx, + &hooks::role_extras(ctx), + None, + ); println!("✓ Successfully compressed the session."); } _ => { @@ -1697,7 +1711,12 @@ fn print_turn_divider() { async fn auto_compress_session(ctx: &mut RequestContext) { match ctx.compress_session().await { - Ok(()) => hooks::fire(HookEvent::SessionCompressed, ctx, &[], None), + Ok(()) => hooks::fire( + HookEvent::SessionCompressed, + ctx, + &hooks::role_extras(ctx), + None, + ), Err(err) => warn!("Failed to compress the session: {err}"), } } @@ -2018,7 +2037,7 @@ pub fn split_args_text(line: &str, is_win: bool) -> (Vec, &str) { mod tests { use super::*; use crate::client::{ClientConfig, Model}; - use crate::config::{AppState, RoleLike, Session, WorkingMode}; + use crate::config::{AppState, Role, RoleLike, Session, TEMP_ROLE_NAME, WorkingMode}; use crate::hooks::{HookDef, HooksMap, test_sink}; use anyhow::anyhow; use serial_test::serial; @@ -2165,6 +2184,64 @@ mod tests { assert_eq!(turn_counts(marker), (1, 0, 1, 0)); } + #[tokio::test] + #[serial] + async fn turn_events_carry_the_role_a_session_holds() { + let _sink = test_sink::install(); + let marker = "tb-role-m3x"; + let mut ctx = ctx_with_hooks(&TURN_EVENTS, marker); + // The role lives on the session (as after `.role` inside a session); + // fire-time resolution must find it there. + let mut session = Session::default(); + session.set_role(Role::new("dev-role", "Prompt")); + ctx.session = Some(session); + let abort_signal = create_abort_signal(); + + let bracket = TurnBracket::enter(&ctx); + let result: Result<()> = Ok(()); + bracket.finish(&ctx, &abort_signal, result).unwrap(); + + let captures: Vec<_> = test_sink::snapshot() + .into_iter() + .filter(|capture| capture.hook_name.starts_with(marker)) + .collect(); + assert_eq!(captures.len(), 2, "{captures:?}"); + for capture in &captures { + assert_eq!( + capture.envs.get("COYOTE_ROLE").map(String::as_str), + Some("dev-role"), + "{} must carry the session-held role", + capture.hook_name + ); + } + } + + #[tokio::test] + #[serial] + async fn turn_events_omit_the_role_var_when_no_role_is_active() { + let _sink = test_sink::install(); + let marker = "tb-norole-q8j"; + let ctx = ctx_with_hooks(&TURN_EVENTS, marker); + let abort_signal = create_abort_signal(); + + let bracket = TurnBracket::enter(&ctx); + let result: Result<()> = Ok(()); + bracket.finish(&ctx, &abort_signal, result).unwrap(); + + let captures: Vec<_> = test_sink::snapshot() + .into_iter() + .filter(|capture| capture.hook_name.starts_with(marker)) + .collect(); + assert_eq!(captures.len(), 2, "{captures:?}"); + for capture in &captures { + assert!( + !capture.envs.contains_key("COYOTE_ROLE"), + "{} must omit COYOTE_ROLE without an active role", + capture.hook_name + ); + } + } + /// Drives a deep REPL future to completion on a thread with extra stack /// headroom: nested `run_repl_command`/`compress_session` poll frames /// are deep in debug builds and overflow the default test-thread stack @@ -2312,6 +2389,48 @@ mod tests { ); } + #[test] + #[serial] + fn session_compressed_carries_the_active_role_name() { + let _sink = test_sink::install(); + let marker = "compress-role-t6b"; + run_async(async { + let mut ctx = compressible_ctx(marker); + // A temp role (`--prompt`) is a real role for COYOTE_ROLE. + ctx.role = Some(Role::new(TEMP_ROLE_NAME, "Session role prompt")); + auto_compress_session(&mut ctx).await; + }); + + let capture = test_sink::snapshot() + .into_iter() + .find(|capture| capture.hook_name == format!("{marker}-session.compressed")) + .expect("session.compressed capture"); + assert_eq!( + capture.envs.get("COYOTE_ROLE").map(String::as_str), + Some(TEMP_ROLE_NAME) + ); + } + + #[test] + #[serial] + fn session_compressed_omits_the_role_var_when_no_role_is_active() { + let _sink = test_sink::install(); + let marker = "compress-norole-p9c"; + run_async(async { + let mut ctx = compressible_ctx(marker); + auto_compress_session(&mut ctx).await; + }); + + let capture = test_sink::snapshot() + .into_iter() + .find(|capture| capture.hook_name == format!("{marker}-session.compressed")) + .expect("session.compressed capture"); + assert!( + !capture.envs.contains_key("COYOTE_ROLE"), + "session.compressed must omit COYOTE_ROLE without an active role" + ); + } + #[test] #[serial] fn auto_compress_failure_fires_no_session_compressed_and_swallows_error() {