diff --git a/crates/tui/src/commands/groups/utility/mcp.rs b/crates/tui/src/commands/groups/utility/mcp.rs index dde80ef8ae..0c9a37372e 100644 --- a/crates/tui/src/commands/groups/utility/mcp.rs +++ b/crates/tui/src/commands/groups/utility/mcp.rs @@ -17,7 +17,7 @@ const CONTAINER_USE_SOURCE: &str = "https://github.com/dagger/container-use"; pub(in crate::commands) const COMMAND_INFO: CommandInfo = CommandInfo { name: "mcp", aliases: &[], - usage: "/mcp [init|import|import approve |import decline |recommendations|add recommended |add stdio [args...]|add http |enable |disable |remove |doctor|validate|restart|reload]", + usage: "/mcp [init|import|import approve |import decline |recommendations|add recommended |add stdio [args...]|add http |enable |disable |remove |retry |doctor|validate|restart|reload]", description_key: "cmd_mcp_description", }; @@ -81,6 +81,10 @@ fn mcp(presentation: &mut dyn CommandPresentationContext, args: Option<&str>) -> Ok(name) => CommandResult::action(AppAction::Mcp(McpUiAction::Logout { name })), Err(msg) => CommandResult::error(msg), }, + "retry" => match parse_name(parts.next(), "Usage: /mcp retry ") { + Ok(name) => CommandResult::action(AppAction::Mcp(McpUiAction::Retry { name })), + Err(msg) => CommandResult::error(msg), + }, "import" | "marketplace" | "sources" => { let sub = parts.next().unwrap_or("").to_ascii_lowercase(); match sub.as_str() { @@ -117,7 +121,7 @@ fn mcp(presentation: &mut dyn CommandPresentationContext, args: Option<&str>) -> CommandResult::action(AppAction::Mcp(McpUiAction::Reload)) } _ => CommandResult::error( - "Usage: /mcp [init|import|recommendations|add recommended |add stdio [args...]|add http |enable |disable |remove |login |logout |doctor|validate|restart|reload]", + "Usage: /mcp [init|import|recommendations|add recommended |add stdio [args...]|add http |enable |disable |remove |login |logout |retry |doctor|validate|restart|reload]", ), } } @@ -561,6 +565,12 @@ mod tests { if name == "remote" && scopes == vec!["tools/read".to_string(), "tools/write".to_string()] )); + + let retry = mcp(&mut FakePresentation, Some("retry remote")); + assert!(matches!( + retry.action, + Some(AppAction::Mcp(McpUiAction::Retry { name })) if name == "remote" + )); } #[test] diff --git a/crates/tui/src/core/engine.rs b/crates/tui/src/core/engine.rs index a2987d8786..a61db8fc7f 100644 --- a/crates/tui/src/core/engine.rs +++ b/crates/tui/src/core/engine.rs @@ -77,7 +77,8 @@ use super::authority::{ }; use super::events::{Event, TurnOutcomeStatus, TurnRoute}; use super::ops::{ - Op, ProviderRuntimeStatus, SessionSnapshot, USER_SHELL_TOOL_ID_PREFIX, UserInputProvenance, + McpManagerUpdate, Op, ProviderRuntimeStatus, SessionSnapshot, USER_SHELL_TOOL_ID_PREFIX, + UserInputProvenance, }; use super::session::Session; use super::tool_parser; @@ -689,6 +690,21 @@ impl EngineHandle { // === Engine === +/// Background MCP boot progress from the spawn-time connect task. +enum McpBootUpdate { + Progress { + generation: u64, + authority_errors: Arc>, + connection_errors: HashMap, + connecting: Vec, + }, + Finished { + generation: u64, + authority_errors: Arc>, + connection_errors: HashMap, + }, +} + /// The core engine that processes operations and emits events pub struct Engine { config: EngineConfig, @@ -722,6 +738,27 @@ pub struct Engine { /// transient `ToolContext` (#4475). file_read_tracker: SharedFileReadTracker, mcp_pool: Option>>, + /// Last connection diagnosis for each configured MCP server. + /// + /// Failed transports are intentionally absent from `McpPool::connections`, + /// so a later one-server retry cannot reconstruct sibling failures from + /// the pool alone. Keeping the diagnoses beside the engine-owned pool + /// lets every full manager snapshot remain truthful without reconnecting + /// unrelated servers. + mcp_connection_errors: HashMap, + /// True while the spawn-time concurrent connect pass is still running. + /// `mcp_tools` snapshots ready servers instead of waiting on optionals. + mcp_boot_in_flight: bool, + mcp_boot_rx: Option>, + mcp_boot_done: Option>, + /// Generation owned by the currently installed boot receiver. Terminal + /// cleanup is conditional on this exact value so an older pass can never + /// clear a newer receiver. + mcp_boot_generation: Option, + /// Monotonic generation for engine-authored MCP session snapshots. Boot + /// task updates retain their spawn generation so later passes can reject + /// only genuinely stale work. + mcp_event_generation: u64, /// Workspace-scoped immutable plugin catalogue and authority receipts. plugin_registry: Arc, api_provider: ApiProvider, @@ -976,6 +1013,8 @@ enum EngineRunInput { /// this wake an active goal waiting on background work stayed inert until /// the user typed something (morning-report continuation gap). ShellCompletionWake, + /// One MCP boot progress/settled update from the spawn-time connect task. + McpBootUpdate(McpBootUpdate), } impl SendMessageOutcome { @@ -1539,6 +1578,12 @@ impl Engine { shell_manager, file_read_tracker, mcp_pool: None, + mcp_connection_errors: HashMap::new(), + mcp_boot_in_flight: false, + mcp_boot_rx: None, + mcp_boot_done: None, + mcp_boot_generation: None, + mcp_event_generation: 0, plugin_registry, api_provider, api_provider_identity, @@ -2185,6 +2230,7 @@ impl Engine { .map(|op| EngineRunInput::Operation(Box::new(op))); } else { let shell_wake_armed = !host_managed_turns && self.idle_shell_wake_armed(); + let mcp_boot_armed = self.mcp_boot_rx.is_some(); tokio::select! { op = self.rx_op.recv() => { return op.map(|op| EngineRunInput::Operation(Box::new(op))); @@ -2200,6 +2246,17 @@ impl Engine { self.route_child_approval_decision(decision).await; } } + update = async { + match self.mcp_boot_rx.as_mut() { + Some(rx) => rx.recv().await, + None => None, + } + }, if mcp_boot_armed => { + match update { + Some(update) => return Some(EngineRunInput::McpBootUpdate(update)), + None => self.mcp_boot_rx = None, + } + } // Background shells have no completion channel, so an // idle engine polls only while a goal is active and a // background job is outstanding; the arm disarms itself @@ -2351,6 +2408,7 @@ impl Engine { // engine must wait for its host to claim and explicitly dispatch the // next turn so events cannot be attached to the wrong durable record. let host_managed_turns = self.host_managed_turns(); + self.start_mcp_session_boot().await; loop { let Some(input) = self.next_run_input(host_managed_turns).await else { @@ -2370,6 +2428,9 @@ impl Engine { EngineRunInput::SubAgentCompletion(completion) => { self.handle_idle_subagent_completion(completion).await; } + EngineRunInput::McpBootUpdate(update) => { + self.apply_mcp_boot_update(update).await; + } EngineRunInput::ShellCompletionWake => { self.handle_idle_shell_completion_wake().await; } @@ -2955,6 +3016,22 @@ impl Engine { let _ = tx.send(status); } } + Op::BootstrapMcp { tx } => { + let result = self.bootstrap_mcp_pool().await.map_err(|error| { + codewhale_config::persistence::redact_secrets(&format!("{error:#}")) + }); + if let Some(tx) = tx.lock().ok().and_then(|mut guard| guard.take()) { + let _ = tx.send(result); + } + } + Op::RetryMcpServer { name, tx } => { + let result = self.retry_mcp_server(&name).await.map_err(|error| { + codewhale_config::persistence::redact_secrets(&format!("{error:#}")) + }); + if let Some(tx) = tx.lock().ok().and_then(|mut guard| guard.take()) { + let _ = tx.send(result); + } + } Op::ReloadMcp { config_path, tx } => { let result = self.reload_mcp_pool(config_path).await.map_err(|error| { codewhale_config::persistence::redact_secrets(&format!("{error:#}")) @@ -5956,10 +6033,10 @@ impl Engine { Ok(pool) } - async fn reload_mcp_pool( - &mut self, - config_path: PathBuf, - ) -> anyhow::Result { + async fn reload_mcp_pool(&mut self, config_path: PathBuf) -> anyhow::Result { + if self.mcp_boot_in_flight { + self.wait_for_mcp_boot().await; + } let pool = self .ensure_mcp_pool() .await @@ -5980,30 +6057,410 @@ impl Engine { .map(|(name, error)| (name, crate::mcp::format_mcp_error_for_display(&error))) .collect::>(); self.session.mcp_config_path = config_path; - Ok(pool.manager_snapshot(&self.session.mcp_config_path, false, &errors)) + self.mcp_connection_errors = errors; + let snapshot = pool.manager_snapshot( + &self.session.mcp_config_path, + false, + &self.mcp_connection_errors, + ); + drop(pool); + let generation = self.next_mcp_event_generation(); + Ok(McpManagerUpdate { + snapshot, + generation, + }) + } + + async fn mcp_session_snapshot(&self) -> anyhow::Result { + let pool = self + .mcp_pool + .as_ref() + .ok_or_else(|| anyhow::anyhow!("MCP pool is not started"))?; + let pool = pool.lock().await; + Ok(pool.manager_snapshot( + &self.session.mcp_config_path, + false, + &self.mcp_connection_errors, + )) + } + + fn mcp_connecting_names(pool: &McpPool, errors: &HashMap) -> Vec { + let connected = pool.connected_servers(); + pool.enabled_server_names() + .into_iter() + .filter(|name| !connected.contains(&name.as_str()) && !errors.contains_key(name)) + .collect() + } + + fn next_mcp_event_generation(&mut self) -> u64 { + self.mcp_event_generation = self.mcp_event_generation.saturating_add(1); + self.mcp_event_generation + } + + fn replace_mcp_boot_errors( + &mut self, + authority_errors: &HashMap, + mut connection_errors: HashMap, + ) { + // Each update owns the ordinary connection diagnoses for this pass, + // so replacing the map drops stale transport errors. Reviewed-plugin + // authority failures are a separate, non-pending set and must remain + // visible throughout the pass; they win if a name ever overlaps. + connection_errors.extend(authority_errors.clone()); + self.mcp_connection_errors = connection_errors; + } + + fn finish_mcp_boot_generation(&mut self, generation: u64) -> bool { + if self.mcp_boot_generation != Some(generation) { + return false; + } + self.mcp_boot_generation = None; + self.mcp_boot_in_flight = false; + self.mcp_boot_rx = None; + self.mcp_boot_done = None; + true + } + + async fn emit_mcp_session_boot(&self, generation: u64, finished: bool) { + let Ok(snapshot) = self.mcp_session_snapshot().await else { + return; + }; + let connecting = if finished { + Vec::new() + } else if let Some(pool) = self.mcp_pool.as_ref() { + let pool = pool.lock().await; + Self::mcp_connecting_names(&pool, &self.mcp_connection_errors) + } else { + Vec::new() + }; + // Zero servers and nothing connecting is not a session-boot surface. + if snapshot.servers.is_empty() && connecting.is_empty() { + return; + } + let _ = self.tx_event.try_send(Event::McpSessionBoot { + generation, + snapshot, + connecting, + finished, + }); + } + + async fn apply_mcp_boot_update(&mut self, update: McpBootUpdate) { + match update { + McpBootUpdate::Progress { + generation, + authority_errors, + connection_errors, + connecting, + } => { + if self.mcp_boot_generation != Some(generation) { + return; + } + if generation < self.mcp_event_generation { + return; + } + self.mcp_event_generation = generation; + self.replace_mcp_boot_errors(&authority_errors, connection_errors); + if let Ok(snapshot) = self.mcp_session_snapshot().await { + let _ = self.tx_event.try_send(Event::McpSessionBoot { + generation, + snapshot, + connecting, + finished: false, + }); + } + } + McpBootUpdate::Finished { + generation, + authority_errors, + connection_errors, + } => { + if self.mcp_boot_generation != Some(generation) { + return; + } + if generation < self.mcp_event_generation { + self.finish_mcp_boot_generation(generation); + return; + } + self.mcp_event_generation = generation; + self.replace_mcp_boot_errors(&authority_errors, connection_errors); + self.finish_mcp_boot_generation(generation); + self.session.pending_prefix_change_reason = Some("mcp-session-boot".to_string()); + self.emit_mcp_session_boot(generation, true).await; + } + } + } + + async fn drain_mcp_boot_updates(&mut self) { + let receiver_generation = self.mcp_boot_generation; + let Some(mut rx) = self.mcp_boot_rx.take() else { + return; + }; + while let Ok(update) = rx.try_recv() { + // Apply without emitting until the last queued update so the UI + // sees one settled receipt rather than a burst. + match update { + McpBootUpdate::Progress { + generation, + authority_errors, + connection_errors, + connecting: _, + } => { + if self.mcp_boot_generation != Some(generation) { + continue; + } + if generation < self.mcp_event_generation { + continue; + } + self.mcp_event_generation = generation; + self.replace_mcp_boot_errors(&authority_errors, connection_errors); + } + McpBootUpdate::Finished { + generation, + authority_errors, + connection_errors, + } => { + if self.mcp_boot_generation != Some(generation) { + continue; + } + if generation < self.mcp_event_generation { + self.finish_mcp_boot_generation(generation); + break; + } + self.mcp_event_generation = generation; + self.replace_mcp_boot_errors(&authority_errors, connection_errors); + self.finish_mcp_boot_generation(generation); + self.session.pending_prefix_change_reason = + Some("mcp-session-boot".to_string()); + break; + } + } + } + if self.mcp_boot_in_flight && self.mcp_boot_generation == receiver_generation { + self.mcp_boot_rx = Some(rx); + } + } + + async fn wait_for_mcp_boot(&mut self) { + if let Some(rx) = self.mcp_boot_done.as_mut() { + while !*rx.borrow() { + if rx.changed().await.is_err() { + break; + } + } + } + self.drain_mcp_boot_updates().await; + } + + /// Start the concurrent connect pass without occupying the engine mailbox. + /// Optional servers never serialize the first model turn: `mcp_tools` + /// snapshots whatever is already ready. + async fn start_mcp_session_boot(&mut self) { + if !self.config.features.enabled(Feature::Mcp) { + return; + } + let pool = match self.ensure_mcp_pool().await { + Ok(pool) => pool, + Err(error) => { + tracing::debug!("MCP session boot skipped: {error}"); + return; + } + }; + + let (pending, auth_errors, timeouts, network_policy, catalog_generation) = { + let mut pool = pool.lock().await; + if let Err(error) = pool.reload_if_config_changed().await { + tracing::debug!( + "MCP session boot config reload failed: {}", + crate::mcp::format_mcp_error_for_display(&error) + ); + } + let (pending, auth_errors) = pool.collect_pending_connects(); + ( + pending, + auth_errors, + pool.connect_timeouts(), + pool.cloned_network_policy(), + pool.current_catalog_generation(), + ) + }; + + let authority_errors = auth_errors + .into_iter() + .map(|(name, error)| (name, crate::mcp::format_mcp_error_for_display(&error))) + .collect::>(); + self.mcp_connection_errors = authority_errors.clone(); + let authority_errors = Arc::new(authority_errors); + let generation = self.next_mcp_event_generation(); + + if pending.is_empty() { + self.mcp_boot_in_flight = false; + self.mcp_boot_generation = None; + self.emit_mcp_session_boot(generation, true).await; + return; + } + + self.mcp_boot_in_flight = true; + self.mcp_boot_generation = Some(generation); + let (progress_tx, progress_rx) = mpsc::unbounded_channel(); + let (done_tx, done_rx) = tokio::sync::watch::channel(false); + self.mcp_boot_rx = Some(progress_rx); + self.mcp_boot_done = Some(done_rx); + + self.emit_mcp_session_boot(generation, false).await; + + let pool_for_task = Arc::clone(&pool); + spawn_supervised( + "mcp-session-boot", + std::panic::Location::caller(), + async move { + let mut remaining: Vec = + pending.iter().map(|(name, _)| name.clone()).collect(); + let results = McpPool::connect_pending_concurrently( + pending, + timeouts, + network_policy, + catalog_generation, + ) + .await; + let mut connection_errors = HashMap::new(); + { + let mut pool = pool_for_task.lock().await; + for (name, result) in results { + remaining.retain(|pending_name| pending_name != &name); + match result { + Ok(connection) => pool.store_ready_connection(name, connection), + Err(error) => { + connection_errors + .insert(name, crate::mcp::format_mcp_error_for_display(&error)); + } + } + let _ = progress_tx.send(McpBootUpdate::Progress { + generation, + authority_errors: Arc::clone(&authority_errors), + connection_errors: connection_errors.clone(), + connecting: remaining.clone(), + }); + } + let mut required = Vec::new(); + pool.push_required_server_errors(&mut required); + for (name, error) in required { + connection_errors + .entry(name) + .or_insert_with(|| crate::mcp::format_mcp_error_for_display(&error)); + } + } + let _ = progress_tx.send(McpBootUpdate::Finished { + generation, + authority_errors, + connection_errors, + }); + let _ = done_tx.send(true); + }, + ); + } + + /// Connect the configured servers through the one engine-owned pool and + /// snapshot that exact pool for the boot UI. `connect_all` is bounded and + /// concurrent; already-ready connections are preserved, and unlike the + /// explicit reload path no config source is force-reloaded. + async fn bootstrap_mcp_pool(&mut self) -> anyhow::Result { + if self.mcp_pool.is_none() { + let _ = self.ensure_mcp_pool().await; + } + if self.mcp_boot_in_flight { + self.wait_for_mcp_boot().await; + } else if self.mcp_pool.is_some() && self.mcp_connection_errors.is_empty() { + // Tests and explicit `/mcp` callers may run before spawn-time boot + // has been scheduled; connect now without blocking later turns. + } + self.drain_mcp_boot_updates().await; + let snapshot = self.mcp_session_snapshot().await?; + let generation = self.next_mcp_event_generation(); + Ok(McpManagerUpdate { + snapshot, + generation, + }) + } + + async fn retry_mcp_server(&mut self, name: &str) -> anyhow::Result { + if self.mcp_boot_in_flight { + self.wait_for_mcp_boot().await; + } + let pool = self + .ensure_mcp_pool() + .await + .map_err(|error| anyhow::anyhow!(error.to_string()))?; + let mut pool = pool.lock().await; + match pool.retry_connection(name).await { + Ok(_) => { + self.mcp_connection_errors.remove(name); + } + Err(error) => { + self.mcp_connection_errors.insert( + name.to_string(), + crate::mcp::format_mcp_error_for_display(&error), + ); + } + } + let snapshot = pool.manager_snapshot( + &self.session.mcp_config_path, + false, + &self.mcp_connection_errors, + ); + self.mcp_connection_errors.retain(|server, _| { + snapshot + .servers + .iter() + .any(|configured| configured.name == *server) + }); + drop(pool); + let generation = self.next_mcp_event_generation(); + let _ = self.tx_event.try_send(Event::McpSessionBoot { + generation, + snapshot: snapshot.clone(), + connecting: Vec::new(), + finished: true, + }); + Ok(McpManagerUpdate { + snapshot, + generation, + }) } async fn mcp_tools(&mut self) -> Vec { let pool = match self.ensure_mcp_pool().await { Ok(pool) => pool, Err(err) => { - let _ = self.tx_event.send(Event::status(format!("{err:#}"))).await; + tracing::debug!("MCP tools unavailable: {err}"); return Vec::new(); } }; - let mut pool = pool.lock().await; - let errors = pool.connect_all().await; - for (server, err) in errors { - let _ = self - .tx_event - .send(Event::status(format!( - "Failed to connect MCP server '{server}': {err:#}" - ))) - .await; + if self.mcp_boot_in_flight { + // Optional servers are still connecting in the background. Snapshot + // currently-ready tools so the first LLM call is not serialized + // behind the slowest handshake. The catalog refreshes on a later + // turn once boot settles (KV-cache prefix re-pin: mcp-session-boot). + return pool.lock().await.to_api_tools(); } - pool.to_api_tools() + let mut pool = pool.lock().await; + let errors = pool.connect_all().await; + self.mcp_connection_errors = errors + .into_iter() + .map(|(server, error)| (server, crate::mcp::format_mcp_error_for_display(&error))) + .collect(); + // Failures stay on the session-boot snapshot, not as Status toasts. + drop(pool); + let generation = self.next_mcp_event_generation(); + self.emit_mcp_session_boot(generation, true).await; + self.mcp_pool + .as_ref() + .expect("pool exists") + .lock() + .await + .to_api_tools() } /// Handle a turn using the DeepSeek API. diff --git a/crates/tui/src/core/engine/handle.rs b/crates/tui/src/core/engine/handle.rs index e704cdbb2d..df7a7156fa 100644 --- a/crates/tui/src/core/engine/handle.rs +++ b/crates/tui/src/core/engine/handle.rs @@ -258,12 +258,45 @@ impl EngineHandle { .map_err(|_| anyhow::anyhow!("Engine dropped provider runtime status oneshot")) } + /// Run the bounded initial connection pass on the engine-owned MCP pool. + /// + /// The returned manager snapshot and every later tool call therefore see + /// the same connections and catalog generation. Unlike `reload_mcp`, this + /// does not force a config re-read or drop ready transports. Optional + /// servers are connected in the background at engine spawn; this waits + /// only if the caller explicitly asked for the settled receipt. + pub async fn bootstrap_mcp(&self) -> Result { + let (tx, rx) = tokio::sync::oneshot::channel(); + let tx = std::sync::Arc::new(std::sync::Mutex::new(Some(tx))); + self.send(Op::BootstrapMcp { tx }).await?; + rx.await + .map_err(|_| anyhow::anyhow!("Engine dropped MCP bootstrap oneshot"))? + .map_err(anyhow::Error::msg) + } + + /// Retry one failed server through the existing engine-owned pool. + pub async fn retry_mcp_server( + &self, + name: impl Into, + ) -> Result { + let (tx, rx) = tokio::sync::oneshot::channel(); + let tx = std::sync::Arc::new(std::sync::Mutex::new(Some(tx))); + self.send(Op::RetryMcpServer { + name: name.into(), + tx, + }) + .await?; + rx.await + .map_err(|_| anyhow::anyhow!("Engine dropped MCP retry oneshot"))? + .map_err(anyhow::Error::msg) + } + /// Force the engine-owned MCP pool to reload and reconnect, returning a /// snapshot from the exact live pool that supplies the next model turn. pub async fn reload_mcp( &self, config_path: std::path::PathBuf, - ) -> Result { + ) -> Result { let (tx, rx) = tokio::sync::oneshot::channel(); let tx = std::sync::Arc::new(std::sync::Mutex::new(Some(tx))); self.send(Op::ReloadMcp { config_path, tx }).await?; diff --git a/crates/tui/src/core/engine/tests.rs b/crates/tui/src/core/engine/tests.rs index fd3263fd5b..60e3a7783d 100644 --- a/crates/tui/src/core/engine/tests.rs +++ b/crates/tui/src/core/engine/tests.rs @@ -20158,7 +20158,8 @@ async fn reload_mcp_op_recovers_from_invalid_initial_config_in_process() { let snapshot = handle .reload_mcp(config_path.clone()) .await - .expect("fixed config reloads without restarting the engine"); + .expect("fixed config reloads without restarting the engine") + .snapshot; assert!(!snapshot.reload_required); assert_eq!(snapshot.servers.len(), 1); assert_eq!(snapshot.servers[0].name, "ready"); @@ -20173,7 +20174,8 @@ async fn reload_mcp_op_recovers_from_invalid_initial_config_in_process() { let alternate = handle .reload_mcp(alternate_path.clone()) .await - .expect("a changed config path replaces the engine pool in process"); + .expect("a changed config path replaces the engine pool in process") + .snapshot; assert_eq!(alternate.config_path, alternate_path); assert_eq!(alternate.servers.len(), 1); assert_eq!(alternate.servers[0].name, "alternate"); @@ -20182,6 +20184,184 @@ async fn reload_mcp_op_recovers_from_invalid_initial_config_in_process() { task.await.expect("engine task"); } +#[tokio::test] +async fn mcp_boot_updates_preserve_authority_errors_and_replace_ordinary_errors() { + let tmp = tempdir().expect("tempdir"); + let engine_config = EngineConfig { + workspace: tmp.path().to_path_buf(), + ..Default::default() + }; + let (mut engine, _handle) = Engine::new(engine_config, &Config::default()); + engine.mcp_event_generation = 3; + engine.mcp_boot_generation = Some(3); + engine.mcp_boot_in_flight = true; + engine.mcp_connection_errors = HashMap::from([( + "stale-transport".to_string(), + "obsolete connection failure".to_string(), + )]); + let authority_errors = Arc::new(HashMap::from([( + "revoked-plugin".to_string(), + "plugin authority revoked or changed".to_string(), + )])); + + engine + .apply_mcp_boot_update(McpBootUpdate::Progress { + generation: 3, + authority_errors: Arc::clone(&authority_errors), + connection_errors: HashMap::from([( + "current-transport".to_string(), + "current connection failure".to_string(), + )]), + connecting: Vec::new(), + }) + .await; + assert_eq!( + engine.mcp_connection_errors, + HashMap::from([ + ( + "revoked-plugin".to_string(), + "plugin authority revoked or changed".to_string(), + ), + ( + "current-transport".to_string(), + "current connection failure".to_string(), + ), + ]) + ); + assert!(!engine.mcp_connection_errors.contains_key("stale-transport")); + + engine.mcp_connection_errors.insert( + "stale-between-updates".to_string(), + "must not survive finished".to_string(), + ); + engine + .apply_mcp_boot_update(McpBootUpdate::Finished { + generation: 3, + authority_errors, + connection_errors: HashMap::from([( + "final-transport".to_string(), + "final connection failure".to_string(), + )]), + }) + .await; + assert_eq!( + engine.mcp_connection_errors, + HashMap::from([ + ( + "revoked-plugin".to_string(), + "plugin authority revoked or changed".to_string(), + ), + ( + "final-transport".to_string(), + "final connection failure".to_string(), + ), + ]) + ); +} + +#[tokio::test] +async fn stale_boot_finished_does_not_clear_a_newer_receiver() { + let tmp = tempdir().expect("tempdir"); + let engine_config = EngineConfig { + workspace: tmp.path().to_path_buf(), + ..Default::default() + }; + let (mut engine, _handle) = Engine::new(engine_config, &Config::default()); + let (_newer_tx, newer_rx) = tokio::sync::mpsc::unbounded_channel(); + engine.mcp_event_generation = 2; + engine.mcp_boot_generation = Some(2); + engine.mcp_boot_in_flight = true; + engine.mcp_boot_rx = Some(newer_rx); + + engine + .apply_mcp_boot_update(McpBootUpdate::Finished { + generation: 1, + authority_errors: Arc::new(HashMap::new()), + connection_errors: HashMap::new(), + }) + .await; + + assert_eq!(engine.mcp_boot_generation, Some(2)); + assert!(engine.mcp_boot_in_flight); + assert!(engine.mcp_boot_rx.is_some()); +} + +#[tokio::test] +async fn bootstrap_and_retry_mcp_use_the_engine_owned_pool() { + let tmp = tempdir().expect("tempdir"); + let workspace = tmp.path().join("workspace"); + std::fs::create_dir_all(&workspace).expect("workspace"); + let config_path = tmp.path().join("mcp.json"); + std::fs::write( + &config_path, + r#"{"servers":{"disabled":{"command":"node","disabled":true},"alpha":{"command":"codewhale-mcp-missing-alpha-9f8e7d6c"},"beta":{"command":"codewhale-mcp-missing-beta-9f8e7d6c"}}}"#, + ) + .expect("MCP config"); + let engine_config = EngineConfig { + workspace, + mcp_config_path: config_path.clone(), + ..Default::default() + }; + let (engine, handle) = Engine::new(engine_config, &Config::default()); + let task = tokio::spawn(async move { engine.run().await }); + + let boot_update = handle + .bootstrap_mcp() + .await + .expect("boot snapshots the engine pool"); + let boot_generation = boot_update.generation; + let boot = boot_update.snapshot; + assert_eq!(boot.config_path, config_path); + assert_eq!(boot.servers.len(), 3); + let disabled = boot + .servers + .iter() + .find(|server| server.name == "disabled") + .expect("disabled row"); + assert!(!disabled.enabled); + assert!(!disabled.connected); + let sibling_error = boot + .servers + .iter() + .find(|server| server.name == "beta") + .and_then(|server| server.error.clone()) + .expect("boot preserves the sibling connection diagnosis"); + + let retry_update = handle + .retry_mcp_server("alpha") + .await + .expect("a failed per-server retry still returns the live snapshot"); + assert!( + retry_update.generation > boot_generation, + "a direct retry needs a newer generation receipt than boot" + ); + let retry = retry_update.snapshot; + assert_eq!(retry.servers.len(), 3); + assert!( + retry + .servers + .iter() + .find(|server| server.name == "alpha") + .expect("retried row") + .error + .as_deref() + .is_some_and(|error| error.contains("alpha")), + "the named retry error must stay attached to its row" + ); + assert_eq!( + retry + .servers + .iter() + .find(|server| server.name == "beta") + .and_then(|server| server.error.as_ref()), + Some(&sibling_error), + "retrying one server must not erase a sibling diagnosis" + ); + + handle.send(Op::Shutdown).await.expect("shutdown"); + task.await.expect("engine task"); +} + #[tokio::test] async fn list_subagents_event_try_send_does_not_block_when_event_channel_full() { use tokio::sync::mpsc; diff --git a/crates/tui/src/core/events.rs b/crates/tui/src/core/events.rs index b1c7566455..ec6a5a9d55 100644 --- a/crates/tui/src/core/events.rs +++ b/crates/tui/src/core/events.rs @@ -445,6 +445,21 @@ pub enum Event { /// Status message for UI display Status { message: String }, + /// Session-owned MCP + plugin boot progress. + /// + /// Failures stay on this event (and therefore on the session page) until + /// retry succeeds. They are not `Status` toasts. `connecting` names the + /// enabled servers that have not settled yet; `finished` is the terminal + /// receipt for this boot pass. + McpSessionBoot { + /// Monotonic engine-owned event generation. A direct `/mcp` snapshot + /// may supersede one generation without suppressing later passes. + generation: u64, + snapshot: crate::mcp::McpManagerSnapshot, + connecting: Vec, + finished: bool, + }, + /// Rendered `/preview-request` manifest (#1004). /// /// The engine is the only authority that can rebuild the exact next-turn diff --git a/crates/tui/src/core/ops.rs b/crates/tui/src/core/ops.rs index 0927fc460d..f4264491f8 100644 --- a/crates/tui/src/core/ops.rs +++ b/crates/tui/src/core/ops.rs @@ -41,8 +41,24 @@ pub struct ProviderRuntimeStatus { pub active_provider_requests: usize, } +/// Engine-owned MCP snapshot plus the exact event generation it supersedes. +/// The TUI uses the receipt to reject already-queued boot events even when it +/// had not rendered that generation before the direct `/mcp` action. +#[derive(Debug, Clone)] +pub struct McpManagerUpdate { + pub snapshot: crate::mcp::McpManagerSnapshot, + pub generation: u64, +} + /// Result of rebuilding the engine-owned MCP pool in process. -pub type McpReloadResult = Result; +pub type McpReloadResult = Result; + +/// Result of the one-shot boot connection pass for the engine-owned MCP pool. +/// +/// This shares the reload result shape while remaining a separate operation: +/// boot may fill an empty live pool, but it must not force a config reload or +/// invalidate already-ready connections. +pub type McpBootstrapResult = Result; /// Origin of text being introduced as a user-role turn. /// @@ -294,6 +310,25 @@ pub enum Op { >, }, + /// Populate the engine-owned MCP pool once at UI boot and return a + /// snapshot from that exact pool. This is not a config reload and never + /// constructs a UI-owned discovery pool. Optional servers never block + /// the first model turn: that turn snapshots currently-ready tools. + BootstrapMcp { + tx: std::sync::Arc< + std::sync::Mutex>>, + >, + }, + + /// Retry one failed MCP server on the existing engine pool and return a + /// full snapshot. Ready siblings are never invalidated or reconnected. + RetryMcpServer { + name: String, + tx: std::sync::Arc< + std::sync::Mutex>>, + >, + }, + /// Force the engine-owned MCP config/catalog to reload and reconnect. /// The returned snapshot is taken from that same live pool. ReloadMcp { diff --git a/crates/tui/src/mcp.rs b/crates/tui/src/mcp.rs index 9ea94f7de9..024a2aebfa 100644 --- a/crates/tui/src/mcp.rs +++ b/crates/tui/src/mcp.rs @@ -2409,6 +2409,9 @@ pub struct McpPool { pub(crate) dynamic_servers: Arc>>, } +type McpPendingConnect = (String, McpServerConfig); +type McpConnectError = (String, anyhow::Error); + impl McpPool { /// Create a new pool with the given configuration pub fn new(config: McpConfig) -> Self { @@ -2514,6 +2517,18 @@ impl McpPool { self } + pub(crate) fn connect_timeouts(&self) -> McpTimeouts { + self.config.timeouts + } + + pub(crate) fn cloned_network_policy(&self) -> Option { + self.network_policy.clone() + } + + pub(crate) fn current_catalog_generation(&self) -> u64 { + self.catalog_generation.load(Ordering::SeqCst) + } + fn drop_connection(&mut self, server_name: &str, reason: &str) { if self.connections.remove(server_name).is_some() { tracing::debug!( @@ -2717,37 +2732,123 @@ impl McpPool { .await?; connection.catalog_generation = self.catalog_generation.load(Ordering::SeqCst); - self.connections.insert(server_name.to_string(), connection); + self.store_ready_connection(server_name.to_string(), connection); self.connections .get_mut(server_name) .ok_or_else(|| anyhow::anyhow!("Failed to store MCP connection for {server_name}")) } - /// Connect to all enabled servers, returning errors for failed connections - pub async fn connect_all(&mut self) -> Vec<(String, anyhow::Error)> { - let mut errors = Vec::new(); - // Reload before taking the configured-name snapshot. Previously the - // first call after adding a server captured the old names, then only - // noticed the config change inside `get_or_connect`, delaying the new - // server until a second turn. - if let Err(err) = self.reload_if_config_changed().await { - errors.push(("configuration".to_string(), err)); - return errors; + /// Retry exactly one server against the configuration already owned by + /// this pool. + /// + /// Unlike normal lazy tool dispatch, an explicit row retry must not notice + /// a concurrent config mtime and invalidate healthy siblings. Config edits + /// remain owned by the explicit reload path; this operation only replaces + /// the named transport. + pub async fn retry_connection(&mut self, server_name: &str) -> Result<&mut McpConnection> { + let plugin_source = self + .connections + .get(server_name) + .and_then(|connection| connection.config().reviewed_plugin.clone()) + .or_else(|| { + self.config + .servers + .get(server_name) + .and_then(|config| config.reviewed_plugin.clone()) + }); + if let Some(source) = plugin_source + && let Err(error) = source.validate_before_use(server_name, "use") + { + self.drop_connection(server_name, "plugin authority revoked or changed"); + return Err(error); } - let names: Vec = self + + self.drop_connection(server_name, "retry"); + + let server_config = self .config .servers - .keys() - .filter(|n| self.config.servers[*n].is_enabled()) + .get(server_name) .cloned() - .collect(); + .or_else(|| self.dynamic_servers.read().get(server_name).cloned()) + .ok_or_else(|| anyhow::anyhow!("Failed to find MCP server: {server_name}"))?; + + if !server_config.is_enabled() { + anyhow::bail!("Failed to connect MCP server '{server_name}': server is disabled"); + } + + let connection = McpConnection::connect_with_policy( + server_name.to_string(), + server_config, + &self.config.timeouts, + self.network_policy.as_ref(), + ) + .await?; + self.store_ready_connection(server_name.to_string(), connection); + self.connections + .get_mut(server_name) + .ok_or_else(|| anyhow::anyhow!("Failed to store MCP connection for {server_name}")) + } + + pub(crate) fn store_ready_connection(&mut self, name: String, mut connection: McpConnection) { + connection.catalog_generation = self.catalog_generation.load(Ordering::SeqCst); + self.connections.insert(name, connection); + } + /// Peak concurrent spawn+handshake attempts. Uncapped, a config full of + /// `npx` servers would start one node runtime per server at the same + /// instant — a memory spike on low-end machines the sequential loop never + /// produced. Eight keeps wall-clock wins (the connect timeout dominates) + /// while bounding peak memory. + const CONNECT_CONCURRENCY: usize = 8; + + /// Decide which enabled configured servers still need a handshake. + /// Dynamic runtime servers stay registered and connect via + /// [`Self::get_or_connect`]; `connect_all` has never spawned them. + pub(crate) fn collect_pending_connects( + &mut self, + ) -> (Vec, Vec) { + let names: Vec = self + .config + .servers + .iter() + .filter(|(_, server)| server.is_enabled()) + .map(|(name, _)| name.clone()) + .collect(); + let mut pending = Vec::new(); + let mut errors = Vec::new(); for name in names { - if let Err(e) = self.get_or_connect(&name).await { - errors.push((name, e)); + let Some(server_config) = self.config.servers.get(&name).cloned() else { + continue; + }; + + let plugin_source = self + .connections + .get(&name) + .and_then(|connection| connection.config().reviewed_plugin.clone()) + .or_else(|| server_config.reviewed_plugin.clone()); + if let Some(source) = plugin_source + && let Err(error) = source.validate_before_use(&name, "use") + { + self.drop_connection(&name, "plugin authority revoked or changed"); + errors.push((name, error)); + continue; } + + if self + .connections + .get(&name) + .is_some_and(McpConnection::is_ready) + { + continue; + } + self.drop_connection(&name, "reconnect"); + pending.push((name, server_config)); } + (pending, errors) + } + pub(crate) fn push_required_server_errors(&self, errors: &mut Vec) { for (name, server_cfg) in &self.config.servers { // Only stand in for a missing diagnosis. When the connect attempt // above already reported why this server failed, appending a @@ -2768,7 +2869,128 @@ impl McpPool { )); } } + } + + /// Handshake the pending servers concurrently without holding the pool + /// lock. Callers insert results under a short lock so a live turn can + /// snapshot ready tools while optional servers are still connecting. + pub(crate) async fn connect_pending_concurrently( + pending: Vec, + timeouts: McpTimeouts, + network_policy: Option, + catalog_generation: u64, + ) -> Vec<(String, Result)> { + if pending.is_empty() { + return Vec::new(); + } + let semaphore = std::sync::Arc::new(tokio::sync::Semaphore::new(Self::CONNECT_CONCURRENCY)); + let mut joins: tokio::task::JoinSet<(String, Result)> = + tokio::task::JoinSet::new(); + for (name, config) in pending { + let permit = semaphore.clone(); + let network_policy = network_policy.clone(); + joins.spawn(async move { + let _permit = permit.acquire_owned().await; + let connection = McpConnection::connect_with_policy( + name.clone(), + config, + &timeouts, + network_policy.as_ref(), + ) + .await + .map(|mut connection| { + connection.catalog_generation = catalog_generation; + connection + }); + (name, connection) + }); + } + + let mut results = Vec::new(); + while let Some(joined) = joins.join_next().await { + match joined { + Ok(result) => results.push(result), + // A panicked connect task loses its server name in the + // JoinError; attribute generically. The sequential loop + // would have propagated the panic and taken the whole + // pool down with it, so this is strictly better. + Err(join_error) => { + results.push(("connection task".to_string(), Err(join_error.into()))); + } + } + } + results + } + + /// Connect to all enabled servers, returning errors for failed connections. + /// + /// Servers connect **concurrently** (bounded by [`Self::CONNECT_CONCURRENCY`]). + /// This used to be a sequential loop over `get_or_connect`, so every + /// server paid the slowest server's spawn+handshake from its own budget: + /// with the default 10s connect timeout, N servers meant a worst case of + /// N×10s before the pool was usable. Each connection still gets its own + /// configured connect timeout; one wedged server can no longer serialize + /// the rest. + /// + /// Semantics preserved from the sequential loop: only configured servers + /// are connected (dynamic runtime entries stay registered), the config is + /// reloaded before the name snapshot (so a server added mid-session + /// connects on this call, not the next), plugin-authority revocation + /// drops the connection instead of silently reconnecting, and the + /// required-server sweep reports at most one error per name. Config edits + /// that land while the batch is in flight are reconciled by one retry + /// pass: a content change drops every connection the previous pass + /// inserted. + pub async fn connect_all(&mut self) -> Vec<(String, anyhow::Error)> { + let mut errors = Vec::new(); + // Reload before taking the configured-name snapshot. Previously the + // first call after adding a server captured the old names, then only + // noticed the config change inside `get_or_connect`, delaying the new + // server until a second turn. + if let Err(err) = self.reload_if_config_changed().await { + errors.push(("configuration".to_string(), err)); + return errors; + } + + for _pass in 0..2 { + let (pending, auth_errors) = self.collect_pending_connects(); + errors.extend(auth_errors); + if pending.is_empty() { + break; + } + + let results = Self::connect_pending_concurrently( + pending, + self.config.timeouts, + self.network_policy.clone(), + self.catalog_generation.load(Ordering::SeqCst), + ) + .await; + for (name, result) in results { + match result { + Ok(connection) => self.store_ready_connection(name, connection), + Err(error) => errors.push((name, error)), + } + } + + // Reconcile a config edit that landed mid-batch: a content + // change dropped every connection this pass inserted, so run one + // more pass against the new config and drop the stale pass's + // errors with it. + match self.reload_if_config_changed().await { + Ok(true) => { + errors.clear(); + continue; + } + Ok(false) => break, + Err(error) => { + errors.push(("configuration".to_string(), error)); + break; + } + } + } + self.push_required_server_errors(&mut errors); errors } diff --git a/crates/tui/src/runtime_threads.rs b/crates/tui/src/runtime_threads.rs index eee3e1cfa9..a34e15f750 100644 --- a/crates/tui/src/runtime_threads.rs +++ b/crates/tui/src/runtime_threads.rs @@ -8091,6 +8091,7 @@ impl RuntimeThreadManager { && matches!( &event, EngineEvent::Status { .. } + | EngineEvent::McpSessionBoot { .. } | EngineEvent::SessionUpdated { .. } | EngineEvent::AgentList { .. } | EngineEvent::AgentSpawned { .. } diff --git a/crates/tui/src/tui/app.rs b/crates/tui/src/tui/app.rs index 772e36f6cc..5d2e71015e 100644 --- a/crates/tui/src/tui/app.rs +++ b/crates/tui/src/tui/app.rs @@ -1798,6 +1798,16 @@ pub struct App { pub coordination_detail: Option, /// Last MCP manager/discovery snapshot shown in the UI. pub mcp_snapshot: Option, + /// True while the engine-owned MCP boot connection pass is in flight. + /// Configured rows render as connecting until its snapshot lands. + pub mcp_initializing: bool, + /// Latest engine-owned MCP event generation applied to the UI. + pub mcp_snapshot_generation: u64, + /// The direct snapshot from a successful `/mcp` action supersedes any + /// queued event at `mcp_snapshot_generation`, but not a later generation. + pub mcp_snapshot_generation_invalidated: bool, + /// Enabled servers that have not settled in the current boot pass. + pub mcp_connecting: Vec, /// Number of MCP servers declared in the user's config at app boot. /// Used by the footer chip (#502) so a count is visible even before /// the user runs `/mcp` for the first time. `0` hides the chip. diff --git a/crates/tui/src/tui/app/init.rs b/crates/tui/src/tui/app/init.rs index f0ffa1b2d7..9246b4ec12 100644 --- a/crates/tui/src/tui/app/init.rs +++ b/crates/tui/src/tui/app/init.rs @@ -661,13 +661,23 @@ impl App { Some(InitialInput::RemoteControl) => (String::new(), 0, false), _ => (String::new(), 0, false), }; - let mcp_configured_count = crate::mcp::load_config_with_workspace_and_plugins( - &mcp_config_path, - &workspace, - plugin_registry.as_ref(), - ) - .map(|cfg| cfg.servers.len()) - .unwrap_or(0); + let (mcp_configured_count, mcp_connecting) = + crate::mcp::load_config_with_workspace_and_plugins( + &mcp_config_path, + &workspace, + plugin_registry.as_ref(), + ) + .map(|cfg| { + let mut connecting = cfg + .servers + .iter() + .filter(|(_, server)| server.is_enabled()) + .map(|(name, _)| name.clone()) + .collect::>(); + connecting.sort(); + (cfg.servers.len(), connecting) + }) + .unwrap_or((0, Vec::new())); let mut hotbar_actions = HotbarActionRegistry::with_configured_routes( config, provider, @@ -959,6 +969,11 @@ impl App { }, coordination_detail: None, mcp_snapshot: None, + mcp_initializing: !mcp_connecting.is_empty() + && config.features().enabled(crate::features::Feature::Mcp), + mcp_snapshot_generation: 0, + mcp_snapshot_generation_invalidated: false, + mcp_connecting, // Read the MCP config once at boot to know how many servers // the user has declared. The footer chip uses this even when // no live snapshot is available (#502). Cheap (just reads diff --git a/crates/tui/src/tui/app/types.rs b/crates/tui/src/tui/app/types.rs index 85e77c9b43..e1616820db 100644 --- a/crates/tui/src/tui/app/types.rs +++ b/crates/tui/src/tui/app/types.rs @@ -1196,6 +1196,10 @@ pub enum McpUiAction { Logout { name: String, }, + /// Retry one failed/timed-out server through the engine-owned live pool. + Retry { + name: String, + }, /// List consent-gated external MCP import candidates with provenance. ImportList, /// Approve importing one discovered external server into user mcp.json. diff --git a/crates/tui/src/tui/mod.rs b/crates/tui/src/tui/mod.rs index 5a819b6ff9..cbca3d0ea2 100644 --- a/crates/tui/src/tui/mod.rs +++ b/crates/tui/src/tui/mod.rs @@ -83,6 +83,7 @@ pub mod prompt_suggestion; pub mod provider_picker; pub mod scrolling; pub mod selection; +pub mod session_boot; pub mod session_metrics; pub mod session_picker; pub mod settings_picker; diff --git a/crates/tui/src/tui/phase_strip.rs b/crates/tui/src/tui/phase_strip.rs index 941220bfa2..1556fa79c1 100644 --- a/crates/tui/src/tui/phase_strip.rs +++ b/crates/tui/src/tui/phase_strip.rs @@ -365,6 +365,31 @@ pub fn render_activity(area: Rect, buf: &mut Buffer, app: &mut App) { used += span_width(&detail); left.extend(detail); } + // MCP + plugin boot is a session-owned set. Surface it on the activity + // strip so a slow optional server cannot look like a hung turn. + if let Some(chip) = crate::tui::session_boot::activity_chip( + app, + available.saturating_sub(used + GROUP_GAP_WIDTH), + ) { + left.push(Span::raw(GROUP_GAP)); + used += GROUP_GAP_WIDTH + chip.width(); + let boot = crate::tui::session_boot::SessionBootSurface::from_app(app); + let ink = if boot.servers.iter().any(|row| { + matches!( + row.state, + crate::tui::session_boot::McpServerBootState::Failed + | crate::tui::session_boot::McpServerBootState::NeedsLogin + ) + }) { + ChromeInk::Failure + } else { + ChromeInk::Active + }; + left.push(Span::styled( + chip, + Style::default().fg(ink.color(&app.ui_theme)), + )); + } if let Some((text, ink)) = notice { left.push(Span::raw(GROUP_GAP)); left.push(Span::styled( @@ -1343,4 +1368,25 @@ mod tests { Some(crate::config::StatusItem::SessionMetrics) ); } + + #[test] + fn activity_band_names_connecting_mcp_servers() { + let mut app = test_app(); + app.ui_locale = crate::localization::Locale::En; + app.mcp_initializing = true; + app.mcp_configured_count = 4; + app.mcp_connecting = ["alpha", "beta", "gamma", "docs"] + .into_iter() + .map(str::to_string) + .collect(); + let text = activity_text(&mut app, 120); + assert!(text.contains("MCP"), "{text}"); + assert!(text.contains("4 connecting"), "{text}"); + assert!(text.contains("alpha"), "{text}"); + assert!(text.contains("docs"), "{text}"); + assert!( + !text.to_ascii_lowercase().contains("slack"), + "Slack is one server, not the chip: {text}" + ); + } } diff --git a/crates/tui/src/tui/session_boot.rs b/crates/tui/src/tui/session_boot.rs new file mode 100644 index 0000000000..ff7fabeb99 --- /dev/null +++ b/crates/tui/src/tui/session_boot.rs @@ -0,0 +1,875 @@ +//! Session-page MCP + plugin boot surface. +//! +//! Plugin discovery and every enabled MCP server boot as a **set**, not a +//! toast per name. The activity strip carries the compact pulse +//! (`MCP · 4 connecting`); the receipt under it keeps per-server outcomes +//! and next actions until retry succeeds. Slack is one server in that set. + +use std::borrow::Cow; + +use ratatui::{ + buffer::Buffer, + layout::Rect, + style::Style, + text::{Line, Span}, + widgets::{Block, Paragraph, Widget}, +}; +use unicode_width::UnicodeWidthStr; + +use crate::localization::{Locale, MessageId, tr}; +use crate::mcp::{McpManagerSnapshot, McpServerSnapshot}; +use crate::palette::ChromeInk; +use crate::plugins::PluginRegistry; +use crate::plugins::types::{PluginDiagnosticLevel, PluginTrustStatus}; +use crate::tui::app::App; + +const ITEM_SEPARATOR: &str = " · "; +const MAX_RECEIPT_ROWS: u16 = 6; +const MAX_NAMED_CHIPS: usize = 4; + +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum SessionBootPhase { + Hidden, + Booting, + Settled, +} + +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum McpServerBootState { + Connecting, + Connected, + Failed, + NeedsLogin, + Disabled, +} + +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum McpServerAction { + Retry, + Login, + Diagnose, + None, +} + +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct McpServerBootRow { + pub name: String, + pub state: McpServerBootState, + pub action: McpServerAction, +} + +#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)] +pub struct PluginBootSummary { + pub loaded: usize, + pub invalid: usize, + pub duplicate: usize, + pub needs_setup: usize, +} + +impl PluginBootSummary { + #[must_use] + pub fn is_quiet(self) -> bool { + self.loaded == 0 && self.invalid == 0 && self.duplicate == 0 && self.needs_setup == 0 + } + + #[must_use] + pub fn from_registry(registry: &PluginRegistry) -> Self { + let loaded = registry.list().len(); + let mut invalid = 0usize; + let mut duplicate = 0usize; + let mut needs_setup = 0usize; + for diagnostic in registry.diagnostics() { + match diagnostic.code { + "duplicate-root" | "name-conflict" => duplicate += 1, + _ if diagnostic.level == PluginDiagnosticLevel::Error => invalid += 1, + _ => {} + } + } + for plugin in registry.list() { + if plugin + .diagnostics + .iter() + .any(|diagnostic| diagnostic.level == PluginDiagnosticLevel::Error) + { + invalid += 1; + } else if matches!( + plugin.trust_status, + PluginTrustStatus::NeverReviewed | PluginTrustStatus::CapabilitiesChanged + ) { + needs_setup += 1; + } + } + Self { + loaded, + invalid, + duplicate, + needs_setup, + } + } +} + +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct SessionBootSurface { + pub phase: SessionBootPhase, + pub servers: Vec, + pub plugins: PluginBootSummary, + /// Enabled-server count used when names have not arrived yet, so the + /// first frame can still say `MCP · N connecting` instead of hiding. + unnamed_connecting: usize, +} + +impl SessionBootSurface { + #[must_use] + pub fn from_app(app: &App) -> Self { + Self::from_parts( + app.mcp_snapshot.as_ref(), + app.mcp_initializing, + &app.mcp_connecting, + app.mcp_configured_count, + PluginBootSummary::from_registry(app.plugin_registry.as_ref()), + ) + } + + #[must_use] + pub fn from_parts( + snapshot: Option<&McpManagerSnapshot>, + initializing: bool, + connecting: &[String], + configured_count: usize, + plugins: PluginBootSummary, + ) -> Self { + let servers = if let Some(snapshot) = snapshot { + snapshot + .servers + .iter() + .map(|server| row_from_snapshot(server, initializing, connecting)) + .collect() + } else if initializing { + let mut names = connecting.to_vec(); + names.sort(); + names + .into_iter() + .map(|name| McpServerBootRow { + name, + state: McpServerBootState::Connecting, + action: McpServerAction::None, + }) + .collect() + } else { + Vec::new() + }; + + let connecting_count = servers + .iter() + .filter(|row| row.state == McpServerBootState::Connecting) + .count(); + let unnamed_connecting = if connecting_count == 0 && initializing { + configured_count + } else { + 0 + }; + let phase = if servers.is_empty() && plugins.is_quiet() && unnamed_connecting == 0 { + SessionBootPhase::Hidden + } else if initializing || connecting_count > 0 || unnamed_connecting > 0 { + SessionBootPhase::Booting + } else { + SessionBootPhase::Settled + }; + + Self { + phase, + servers, + plugins, + unnamed_connecting, + } + } + + #[must_use] + pub fn is_hidden(&self) -> bool { + self.phase == SessionBootPhase::Hidden + } + + #[must_use] + pub fn activity_chip(&self, locale: Locale, budget: usize) -> Option { + if self.phase == SessionBootPhase::Hidden || budget == 0 { + return None; + } + let connecting: Vec<&str> = self + .servers + .iter() + .filter(|row| row.state == McpServerBootState::Connecting) + .map(|row| row.name.as_str()) + .collect(); + let failed = self + .servers + .iter() + .filter(|row| { + matches!( + row.state, + McpServerBootState::Failed | McpServerBootState::NeedsLogin + ) + }) + .count(); + let connected = self + .servers + .iter() + .filter(|row| row.state == McpServerBootState::Connected) + .count(); + + let mut candidates = Vec::new(); + if !connecting.is_empty() { + let count = connecting.len(); + let named = named_chip_line("MCP", count, "connecting", &connecting); + candidates.push(named); + candidates.push(format!("MCP{ITEM_SEPARATOR}{count} connecting")); + } else if failed > 0 { + candidates.push(format!( + "MCP{ITEM_SEPARATOR}{connected} {}{ITEM_SEPARATOR}{failed} {}", + tr(locale, MessageId::ExtensionsStateConnected), + tr(locale, MessageId::PhaseFailed) + )); + candidates.push(format!("MCP{ITEM_SEPARATOR}{failed} failed")); + } else if self.phase == SessionBootPhase::Booting { + let count = self.servers.len().max(self.unnamed_connecting); + if count > 0 { + candidates.push(format!("MCP{ITEM_SEPARATOR}{count} connecting")); + } + } + + candidates.into_iter().find(|line| line.width() <= budget) + } + + #[must_use] + pub fn receipt_lines(&self, locale: Locale, width: usize) -> Vec { + if self.phase == SessionBootPhase::Hidden || width == 0 { + return Vec::new(); + } + let mut lines = Vec::new(); + if let Some(plugin_line) = plugin_receipt_line(self.plugins, locale, width) { + lines.push(plugin_line); + } + + match self.phase { + SessionBootPhase::Hidden => {} + SessionBootPhase::Booting => { + let connecting: Vec<&str> = self + .servers + .iter() + .filter(|row| row.state == McpServerBootState::Connecting) + .map(|row| row.name.as_str()) + .collect(); + if connecting.is_empty() && self.servers.is_empty() { + if self.unnamed_connecting > 0 { + lines.push(format!( + "MCP{ITEM_SEPARATOR}{} connecting", + self.unnamed_connecting + )); + } + } else { + let count = if connecting.is_empty() { + self.servers.len() + } else { + connecting.len() + }; + let named = named_chip_line("MCP", count, "connecting", &connecting); + lines.push(truncate_to_width(&named, width)); + } + } + SessionBootPhase::Settled => { + if self.servers.len() == 1 { + lines.push(truncate_to_width( + &server_row_text(&self.servers[0], locale), + width, + )); + } else { + let mut remaining = + MAX_RECEIPT_ROWS.saturating_sub(lines.len() as u16) as usize; + if remaining == 0 { + return lines; + } + let notable: Vec<&McpServerBootRow> = self + .servers + .iter() + .filter(|row| { + matches!( + row.state, + McpServerBootState::Failed + | McpServerBootState::NeedsLogin + | McpServerBootState::Disabled + ) + }) + .collect(); + let connected = self + .servers + .iter() + .filter(|row| row.state == McpServerBootState::Connected) + .count(); + if notable.is_empty() { + if connected > 0 { + lines.push(truncate_to_width( + &format!( + "MCP{ITEM_SEPARATOR}{connected} {}", + tr(locale, MessageId::ExtensionsStateConnected) + ), + width, + )); + } + } else { + if connected > 0 && remaining > 1 { + lines.push(format!( + "MCP{ITEM_SEPARATOR}{connected} {}", + tr(locale, MessageId::ExtensionsStateConnected) + )); + remaining = remaining.saturating_sub(1); + } + let overflow = notable.len() > remaining; + let show = if overflow { + remaining.saturating_sub(1) + } else { + notable.len() + }; + for row in notable.iter().take(show) { + lines.push(truncate_to_width(&server_row_text(row, locale), width)); + } + let hidden = notable.len().saturating_sub(show); + if hidden > 0 { + lines.push(format!("+{hidden} more · /mcp")); + } + } + } + } + } + lines.truncate(MAX_RECEIPT_ROWS as usize); + lines + } + + #[must_use] + pub fn receipt_height(&self, locale: Locale, width: u16) -> u16 { + if self.is_hidden() { + return 0; + } + let lines = self.receipt_lines(locale, usize::from(width)); + (lines.len() as u16).min(MAX_RECEIPT_ROWS) + } +} + +fn row_from_snapshot( + server: &McpServerSnapshot, + initializing: bool, + connecting: &[String], +) -> McpServerBootRow { + let valid_name = server + .name + .chars() + .all(|ch| ch.is_ascii_alphanumeric() || matches!(ch, '-' | '_' | '.')); + if !server.enabled { + return McpServerBootRow { + name: server.name.clone(), + state: McpServerBootState::Disabled, + action: McpServerAction::None, + }; + } + if server.connected { + return McpServerBootRow { + name: server.name.clone(), + state: McpServerBootState::Connected, + action: McpServerAction::None, + }; + } + if let Some(error) = server.error.as_deref() { + if mcp_error_requires_login(error) { + return McpServerBootRow { + name: server.name.clone(), + state: McpServerBootState::NeedsLogin, + action: if valid_name { + McpServerAction::Login + } else { + McpServerAction::Diagnose + }, + }; + } + return McpServerBootRow { + name: server.name.clone(), + state: McpServerBootState::Failed, + action: if valid_name { + McpServerAction::Retry + } else { + McpServerAction::Diagnose + }, + }; + } + let connecting_now = initializing || connecting.iter().any(|name| name == &server.name); + McpServerBootRow { + name: server.name.clone(), + state: if connecting_now { + McpServerBootState::Connecting + } else { + McpServerBootState::Failed + }, + action: if connecting_now { + McpServerAction::None + } else if valid_name { + McpServerAction::Retry + } else { + McpServerAction::Diagnose + }, + } +} + +#[must_use] +pub fn mcp_error_requires_login(error: &str) -> bool { + let error = error.to_ascii_lowercase(); + error.contains("mcp login") + || error.contains("auth required") + || (error.contains("oauth") && error.contains("authenticat")) +} + +fn named_chip_line(kind: &str, count: usize, verb: &str, names: &[&str]) -> String { + let chips = names + .iter() + .take(MAX_NAMED_CHIPS) + .copied() + .collect::>(); + let extra = names.len().saturating_sub(chips.len()); + let mut line = format!("{kind}{ITEM_SEPARATOR}{count} {verb}"); + if !chips.is_empty() { + line.push_str(ITEM_SEPARATOR); + line.push_str(&chips.join(ITEM_SEPARATOR)); + if extra > 0 { + line.push_str(&format!("{ITEM_SEPARATOR}+{extra}")); + } + } + line +} + +fn server_row_text(row: &McpServerBootRow, locale: Locale) -> String { + let state = match row.state { + McpServerBootState::Connecting => Cow::Borrowed("connecting"), + McpServerBootState::Connected => tr(locale, MessageId::ExtensionsStateConnected), + McpServerBootState::Failed => tr(locale, MessageId::PhaseFailed), + McpServerBootState::NeedsLogin => Cow::Borrowed("needs login"), + McpServerBootState::Disabled => tr(locale, MessageId::HotbarSetupStatusDisabled), + }; + let action = match row.action { + McpServerAction::Retry => format!(" · /mcp retry {}", row.name), + McpServerAction::Login => format!(" · /mcp login {}", row.name), + McpServerAction::Diagnose => " · /mcp doctor".to_string(), + McpServerAction::None => String::new(), + }; + format!("{}{ITEM_SEPARATOR}{state}{action}", row.name) +} + +fn plugin_receipt_line(summary: PluginBootSummary, locale: Locale, width: usize) -> Option { + if summary.is_quiet() { + return None; + } + let mut parts = vec![format!( + "{}{ITEM_SEPARATOR}{} {}", + tr(locale, MessageId::ExtensionsTabPlugins), + summary.loaded, + "loaded" + )]; + if summary.invalid > 0 { + parts.push(format!( + "{} {}", + summary.invalid, + tr(locale, MessageId::ExtensionsStateInvalid) + )); + } + if summary.duplicate > 0 { + parts.push(format!("{} duplicate", summary.duplicate)); + } + if summary.needs_setup > 0 { + parts.push(format!("{} need setup", summary.needs_setup)); + } + Some(truncate_to_width(&parts.join(ITEM_SEPARATOR), width)) +} + +fn truncate_to_width(text: &str, width: usize) -> String { + crate::localization::truncate_to_width(text, width) +} + +/// Activity-strip chip for the current session boot set. +#[must_use] +pub fn activity_chip(app: &App, budget: usize) -> Option { + SessionBootSurface::from_app(app).activity_chip(app.ui_locale, budget) +} + +/// Rows the compact boot receipt wants above the activity band. +#[must_use] +pub fn receipt_height(app: &App, width: u16, budget: u16) -> u16 { + if budget == 0 { + return 0; + } + SessionBootSurface::from_app(app) + .receipt_height(app.ui_locale, width) + .min(budget) +} + +/// Paint the compact boot receipt. Text only: Reduced/Still skip any spin. +pub fn render(area: Rect, buf: &mut Buffer, app: &App) { + if area.width == 0 || area.height == 0 { + return; + } + let surface = SessionBootSurface::from_app(app); + let lines = surface.receipt_lines(app.ui_locale, usize::from(area.width)); + if lines.is_empty() { + return; + } + Block::default() + .style(Style::default().bg(app.ui_theme.surface_bg)) + .render(area, buf); + let ink = if surface.servers.iter().any(|row| { + matches!( + row.state, + McpServerBootState::Failed | McpServerBootState::NeedsLogin + ) + }) { + ChromeInk::Failure + } else if surface.phase == SessionBootPhase::Booting { + ChromeInk::Active + } else { + ChromeInk::Metadata + }; + let rendered: Vec> = lines + .into_iter() + .take(area.height as usize) + .map(|line| { + Line::from(Span::styled( + line, + Style::default().fg(ink.color(&app.ui_theme)), + )) + }) + .collect(); + Paragraph::new(rendered).render(area, buf); +} + +#[cfg(test)] +mod tests { + use super::*; + use crate::mcp::{McpManagerSnapshot, McpServerCapabilityMetadata, McpServerSnapshot}; + use std::path::PathBuf; + + fn server( + name: &str, + enabled: bool, + connected: bool, + error: Option<&str>, + ) -> McpServerSnapshot { + McpServerSnapshot { + name: name.to_string(), + enabled, + required: false, + transport: "stdio".to_string(), + command_or_url: format!("cmd-{name}"), + connect_timeout: 5, + execute_timeout: 5, + read_timeout: 5, + connected, + error: error.map(str::to_string), + capability_metadata: McpServerCapabilityMetadata::NotObserved, + tools: Vec::new(), + resources: Vec::new(), + prompts: Vec::new(), + } + } + + fn snapshot(servers: Vec) -> McpManagerSnapshot { + McpManagerSnapshot { + config_path: PathBuf::from("mcp.json"), + config_exists: true, + reload_required: false, + servers, + } + } + + #[test] + fn zero_servers_and_quiet_plugins_hide() { + let surface = + SessionBootSurface::from_parts(None, false, &[], 0, PluginBootSummary::default()); + assert_eq!(surface.phase, SessionBootPhase::Hidden); + assert!(surface.activity_chip(Locale::En, 80).is_none()); + assert!(surface.receipt_lines(Locale::En, 80).is_empty()); + assert_eq!(surface.receipt_height(Locale::En, 80), 0); + } + + #[test] + fn one_connecting_server_names_itself() { + let snap = snapshot(vec![server("alpha", true, false, None)]); + let surface = SessionBootSurface::from_parts( + Some(&snap), + true, + &["alpha".to_string()], + 1, + PluginBootSummary::default(), + ); + assert_eq!(surface.phase, SessionBootPhase::Booting); + assert_eq!(surface.servers.len(), 1); + assert_eq!(surface.servers[0].state, McpServerBootState::Connecting); + let chip = surface.activity_chip(Locale::En, 80).expect("chip"); + assert!(chip.contains("alpha"), "{chip}"); + assert!(!chip.to_ascii_lowercase().contains("slack"), "{chip}"); + let receipt = surface.receipt_lines(Locale::En, 80); + assert_eq!(receipt.len(), 1); + assert!(receipt[0].contains("alpha"), "{receipt:?}"); + } + + #[test] + fn many_connecting_servers_use_count_and_named_chips() { + let snap = snapshot(vec![ + server("alpha", true, false, None), + server("beta", true, false, None), + server("gamma", true, false, None), + server("docs", true, false, None), + ]); + let connecting = ["alpha", "beta", "gamma", "docs"] + .into_iter() + .map(str::to_string) + .collect::>(); + let surface = SessionBootSurface::from_parts( + Some(&snap), + true, + &connecting, + 4, + PluginBootSummary::default(), + ); + assert_eq!(surface.phase, SessionBootPhase::Booting); + let chip = surface.activity_chip(Locale::En, 80).expect("chip"); + assert!(chip.contains("4 connecting"), "{chip}"); + assert!(chip.contains("alpha"), "{chip}"); + assert!(chip.contains("docs"), "{chip}"); + assert!(!chip.to_ascii_lowercase().contains("slack"), "{chip}"); + let receipt = surface.receipt_lines(Locale::En, 80); + assert_eq!(receipt.len(), 1, "{receipt:?}"); + assert!(receipt[0].contains("4 connecting"), "{receipt:?}"); + } + + #[test] + fn settled_failures_keep_retry_and_login_on_the_row() { + let snap = snapshot(vec![ + server("alpha", true, true, None), + server("beta", true, false, Some("protocol negotiation timed out")), + server( + "gamma", + true, + false, + Some("MCP server 'gamma' requires OAuth authentication. Run `/mcp login gamma`"), + ), + server("docs", false, false, Some("disabled")), + ]); + let surface = SessionBootSurface::from_parts( + Some(&snap), + false, + &[], + 4, + PluginBootSummary { + loaded: 12, + invalid: 1, + duplicate: 2, + needs_setup: 0, + }, + ); + assert_eq!(surface.phase, SessionBootPhase::Settled); + assert_eq!( + surface + .servers + .iter() + .find(|row| row.name == "beta") + .map(|row| (row.state, row.action)), + Some((McpServerBootState::Failed, McpServerAction::Retry)) + ); + assert_eq!( + surface + .servers + .iter() + .find(|row| row.name == "gamma") + .map(|row| (row.state, row.action)), + Some((McpServerBootState::NeedsLogin, McpServerAction::Login)) + ); + let receipt = surface.receipt_lines(Locale::En, 100); + let joined = receipt.join("\n"); + assert!(joined.contains("Plugins"), "{joined}"); + assert!(joined.contains("12 loaded"), "{joined}"); + assert!(joined.contains("1 invalid"), "{joined}"); + assert!(joined.contains("2 duplicate"), "{joined}"); + assert!(joined.contains("/mcp retry beta"), "{joined}"); + assert!(joined.contains("/mcp login gamma"), "{joined}"); + assert!(!joined.contains("/mcp auth"), "{joined}"); + assert!(!joined.to_ascii_lowercase().contains("slack"), "{joined}"); + } + + #[test] + fn narrow_activity_budget_sheds_names_keeps_count() { + let snap = snapshot(vec![ + server("alpha", true, false, None), + server("beta", true, false, None), + server("gamma", true, false, None), + ]); + let connecting = ["alpha", "beta", "gamma"] + .into_iter() + .map(str::to_string) + .collect::>(); + let surface = SessionBootSurface::from_parts( + Some(&snap), + true, + &connecting, + 3, + PluginBootSummary::default(), + ); + let chip = surface.activity_chip(Locale::En, 22).expect("chip"); + assert_eq!(chip, "MCP · 3 connecting"); + } + + #[test] + fn first_frame_names_enabled_servers_before_a_snapshot_arrives() { + let connecting = ["gamma", "alpha", "docs"] + .into_iter() + .map(str::to_string) + .collect::>(); + let surface = SessionBootSurface::from_parts( + None, + true, + &connecting, + 3, + PluginBootSummary::default(), + ); + assert_eq!(surface.phase, SessionBootPhase::Booting); + assert_eq!( + surface + .servers + .iter() + .map(|row| row.name.as_str()) + .collect::>(), + vec!["alpha", "docs", "gamma"] + ); + let chip = surface.activity_chip(Locale::En, 80).expect("chip"); + assert!(chip.contains("3 connecting"), "{chip}"); + assert!(chip.contains("alpha"), "{chip}"); + assert!(chip.contains("gamma"), "{chip}"); + assert!(!chip.to_ascii_lowercase().contains("slack"), "{chip}"); + let receipt = surface.receipt_lines(Locale::En, 80); + assert_eq!(receipt.len(), 1, "{receipt:?}"); + assert!(receipt[0].contains("alpha"), "{receipt:?}"); + assert!(receipt[0].contains("docs"), "{receipt:?}"); + } + + #[test] + fn initializing_without_names_still_shows_the_count() { + let surface = + SessionBootSurface::from_parts(None, true, &[], 4, PluginBootSummary::default()); + assert_eq!(surface.phase, SessionBootPhase::Booting); + assert!(surface.servers.is_empty()); + assert_eq!( + surface.activity_chip(Locale::En, 80).as_deref(), + Some("MCP · 4 connecting") + ); + assert_eq!( + surface.receipt_lines(Locale::En, 80), + vec!["MCP · 4 connecting".to_string()] + ); + } + + #[test] + fn settled_single_server_keeps_the_name_and_next_action() { + let snap = snapshot(vec![server( + "alpha", + true, + false, + Some("protocol negotiation timed out"), + )]); + let surface = SessionBootSurface::from_parts( + Some(&snap), + false, + &[], + 1, + PluginBootSummary::default(), + ); + assert_eq!(surface.phase, SessionBootPhase::Settled); + assert_eq!( + surface.receipt_lines(Locale::En, 80), + vec!["alpha · failed · /mcp retry alpha".to_string()] + ); + } + + #[test] + fn settled_all_connected_collapses_to_the_count() { + let snap = snapshot(vec![ + server("alpha", true, true, None), + server("beta", true, true, None), + ]); + let surface = SessionBootSurface::from_parts( + Some(&snap), + false, + &[], + 2, + PluginBootSummary::default(), + ); + assert_eq!(surface.phase, SessionBootPhase::Settled); + assert_eq!( + surface.receipt_lines(Locale::En, 80), + vec!["MCP · 2 connected".to_string()] + ); + assert!(surface.activity_chip(Locale::En, 80).is_none()); + } + + #[test] + fn overflow_receipt_keeps_a_plus_more_row() { + let snap = snapshot( + (0..8) + .map(|i| { + server( + &format!("s{i}"), + true, + false, + Some("protocol negotiation timed out"), + ) + }) + .collect(), + ); + let surface = SessionBootSurface::from_parts( + Some(&snap), + false, + &[], + 8, + PluginBootSummary::default(), + ); + let receipt = surface.receipt_lines(Locale::En, 80); + assert_eq!(receipt.len(), 6, "{receipt:?}"); + assert!( + receipt.last().is_some_and(|line| line.contains("+3 more")), + "{receipt:?}" + ); + assert!( + receipt.iter().any(|line| line.contains("/mcp retry s0")), + "{receipt:?}" + ); + assert!(!receipt.join("\n").contains("/mcp auth"), "{receipt:?}"); + } + + #[test] + fn plugin_line_sits_beside_connecting_mcp_names() { + let connecting = ["alpha", "beta"] + .into_iter() + .map(str::to_string) + .collect::>(); + let surface = SessionBootSurface::from_parts( + None, + true, + &connecting, + 2, + PluginBootSummary { + loaded: 12, + invalid: 1, + duplicate: 2, + needs_setup: 0, + }, + ); + let receipt = surface.receipt_lines(Locale::En, 100); + let joined = receipt.join("\n"); + assert!(joined.contains("Plugins"), "{joined}"); + assert!(joined.contains("12 loaded"), "{joined}"); + assert!(joined.contains("alpha"), "{joined}"); + assert!(joined.contains("beta"), "{joined}"); + assert!(!joined.to_ascii_lowercase().contains("slack"), "{joined}"); + } +} diff --git a/crates/tui/src/tui/ui/event_loop.rs b/crates/tui/src/tui/ui/event_loop.rs index e8effe2dfe..a884a0aa62 100644 --- a/crates/tui/src/tui/ui/event_loop.rs +++ b/crates/tui/src/tui/ui/event_loop.rs @@ -2218,6 +2218,16 @@ pub(crate) async fn run_event_loop( .replace("{tools}", &tools); app.push_status_toast(message, StatusToastLevel::Warning, Some(12_000)); } + EngineEvent::McpSessionBoot { + generation, + snapshot, + connecting, + finished, + } => { + apply_mcp_session_boot_event( + app, generation, snapshot, connecting, finished, + ); + } EngineEvent::RequestManifestReady { rendered } => { // Typed manifest text, or the explicitly requested // base-prompt-only disclosure. Rendered as a system cell. @@ -5791,6 +5801,33 @@ pub(crate) async fn run_event_loop( } } +/// Apply one MCP session-boot event. Failures stay on the snapshot (and +/// therefore the session page) rather than as toast-only Status copy. +/// A direct `/mcp` snapshot invalidates only the event generation it +/// superseded. Older spawn-time updates cannot overwrite it, while a later +/// engine-authored generation can continue updating the live surface. +pub(crate) fn apply_mcp_session_boot_event( + app: &mut App, + generation: u64, + snapshot: crate::mcp::McpManagerSnapshot, + connecting: Vec, + finished: bool, +) { + if generation < app.mcp_snapshot_generation + || (generation == app.mcp_snapshot_generation && app.mcp_snapshot_generation_invalidated) + { + return; + } + app.mcp_snapshot_generation = generation; + app.mcp_snapshot_generation_invalidated = false; + app.mcp_configured_count = snapshot.servers.len(); + app.hotbar_actions.replace_mcp_tools(Some(&snapshot)); + app.mcp_snapshot = Some(snapshot); + app.mcp_connecting = connecting; + app.mcp_initializing = !finished; + app.needs_redraw = true; +} + pub(crate) async fn run_cache_warmup(app: &App, config: &Config) -> Result { let route = resolve_cache_replay_route(app, config)? .validate() @@ -5918,3 +5955,127 @@ async fn open_agents_register(app: &mut App, engine_handle: &EngineHandle) { let _ = engine_handle.send(Op::ListSubAgents).await; app.needs_redraw = true; } + +#[cfg(test)] +mod session_boot_event_tests { + use super::*; + use crate::mcp::{McpManagerSnapshot, McpServerCapabilityMetadata, McpServerSnapshot}; + use std::path::PathBuf; + + fn server(name: &str, connected: bool) -> McpServerSnapshot { + McpServerSnapshot { + name: name.to_string(), + enabled: true, + required: false, + transport: "stdio".to_string(), + command_or_url: format!("cmd-{name}"), + connect_timeout: 5, + execute_timeout: 5, + read_timeout: 5, + connected, + error: None, + capability_metadata: McpServerCapabilityMetadata::NotObserved, + tools: Vec::new(), + resources: Vec::new(), + prompts: Vec::new(), + } + } + + fn snapshot(servers: Vec) -> McpManagerSnapshot { + McpManagerSnapshot { + config_path: PathBuf::from("mcp.json"), + config_exists: true, + reload_required: false, + servers, + } + } + + fn test_app() -> App { + crate::test_support::test_app_with_options(crate::test_support::test_tui_options( + PathBuf::from("."), + )) + } + + #[test] + fn boot_event_names_every_connecting_server_on_the_app() { + let mut app = test_app(); + apply_mcp_session_boot_event( + &mut app, + 1, + snapshot(vec![server("alpha", false), server("beta", false)]), + vec!["alpha".into(), "beta".into()], + false, + ); + assert!(app.mcp_initializing); + assert_eq!(app.mcp_connecting, vec!["alpha", "beta"]); + assert_eq!(app.mcp_configured_count, 2); + let surface = crate::tui::session_boot::SessionBootSurface::from_app(&app); + let chip = surface + .activity_chip(crate::localization::Locale::En, 80) + .expect("chip"); + assert!(chip.contains("alpha"), "{chip}"); + assert!(chip.contains("beta"), "{chip}"); + assert!(!chip.to_ascii_lowercase().contains("slack"), "{chip}"); + } + + #[test] + fn direct_mcp_snapshot_rejects_an_unseen_older_boot_generation() { + let mut app = test_app(); + assert_eq!(app.mcp_snapshot_generation, 0); + app.mcp_snapshot = Some(snapshot(vec![server("direct", true)])); + // The direct engine response carries generation 2 even though the UI + // has not rendered queued boot generation 1 yet. + app.mcp_snapshot_generation = 2; + app.mcp_snapshot_generation_invalidated = true; + app.mcp_connecting = vec!["alpha".into()]; + apply_mcp_session_boot_event( + &mut app, + 1, + snapshot(vec![server("stale", true)]), + vec!["stale".into()], + true, + ); + assert_eq!(app.mcp_connecting, vec!["alpha"]); + assert_eq!( + app.mcp_snapshot + .as_ref() + .and_then(|snapshot| snapshot.servers.first()) + .map(|server| server.name.as_str()), + Some("direct") + ); + + // A queued event emitted by the direct operation itself is the same + // generation and cannot replace the already-applied response. + apply_mcp_session_boot_event( + &mut app, + 2, + snapshot(vec![server("same-pass", true)]), + Vec::new(), + true, + ); + assert_eq!( + app.mcp_snapshot + .as_ref() + .and_then(|snapshot| snapshot.servers.first()) + .map(|server| server.name.as_str()), + Some("direct") + ); + + apply_mcp_session_boot_event( + &mut app, + 3, + snapshot(vec![server("fresh", true)]), + Vec::new(), + true, + ); + assert_eq!(app.mcp_snapshot_generation, 3); + assert!(!app.mcp_snapshot_generation_invalidated); + assert_eq!( + app.mcp_snapshot + .as_ref() + .and_then(|snapshot| snapshot.servers.first()) + .map(|server| server.name.as_str()), + Some("fresh") + ); + } +} diff --git a/crates/tui/src/tui/ui/frame.rs b/crates/tui/src/tui/ui/frame.rs index ed8ffe8358..ac62174c27 100644 --- a/crates/tui/src/tui/ui/frame.rs +++ b/crates/tui/src/tui/ui/frame.rs @@ -902,8 +902,20 @@ pub(crate) fn render(f: &mut Frame, app: &mut App, _config: &Config) -> Option<( // up to three compact rows at the release floor. let preview_cap = if size.height >= 20 { 4 } else { 3 }; let preview_height = desired_preview_height.min(auxiliary_budget.min(preview_cap)); - let workflow_panel_height = - desired_workflow_panel_height.min(auxiliary_budget.saturating_sub(preview_height)); + let session_boot_height = if mini { + 0 + } else { + crate::tui::session_boot::receipt_height( + app, + shell_area.width, + auxiliary_budget.saturating_sub(preview_height), + ) + }; + let workflow_panel_height = desired_workflow_panel_height.min( + auxiliary_budget + .saturating_sub(preview_height) + .saturating_sub(session_boot_height), + ); // Two pinned bands bracket the composer and never trade places with // it: the activity band (transient phase pulse, notices, and the @@ -922,14 +934,16 @@ pub(crate) fn render(f: &mut Frame, app: &mut App, _config: &Config) -> Option<( Constraint::Length(workflow_panel_height), // Workflow panel (#4121) Constraint::Length(preview_height), // Pending input preview (0 if empty) Constraint::Length(indicator_height), // Background-work chip (#5286, 0 if idle) + Constraint::Length(session_boot_height), // MCP+plugin boot receipt (0 if quiet) Constraint::Length(activity_height), // Activity band above the composer Constraint::Length(composer_height), // Composer Constraint::Length(footer_height), // Identity band below the composer ]) .split(body_area); - let activity_slot = 5; - let composer_slot = 6; - let footer_slot = 7; + let session_boot_slot = 5; + let activity_slot = 6; + let composer_slot = 7; + let footer_slot = 8; let (work_chat_area, side_work_area) = if mini && !mini_cfg.keep_sidebar { // Mini mode without the side rail: the transcript takes the whole @@ -1048,6 +1062,11 @@ pub(crate) fn render(f: &mut Frame, app: &mut App, _config: &Config) -> Option<( crate::tui::background_indicator::render(body_chunks[4], buf, app, &pending_work); } + if session_boot_height > 0 { + let buf = f.buffer_mut(); + crate::tui::session_boot::render(body_chunks[session_boot_slot], buf, app); + } + // Render the pinned activity band (transient phase pulse, notices, // cost/metrics ledger). Its row is fixed above the composer in every // phase; only the text inside it changes. @@ -1155,6 +1174,13 @@ pub(crate) fn render(f: &mut Frame, app: &mut App, _config: &Config) -> Option<( column.paint_matching(work_chat_area, f.buffer_mut(), app.ui_theme.surface_bg); column.paint_matching(body_chunks[2], f.buffer_mut(), app.ui_theme.surface_bg); column.paint_matching(body_chunks[3], f.buffer_mut(), app.ui_theme.surface_bg); + if session_boot_height > 0 { + column.paint_matching( + body_chunks[session_boot_slot], + f.buffer_mut(), + app.ui_theme.surface_bg, + ); + } if activity_height > 0 { column.paint_matching( body_chunks[activity_slot], diff --git a/crates/tui/src/tui/ui/handlers.rs b/crates/tui/src/tui/ui/handlers.rs index 6d582e95b2..13c64c09be 100644 --- a/crates/tui/src/tui/ui/handlers.rs +++ b/crates/tui/src/tui/ui/handlers.rs @@ -413,6 +413,11 @@ pub(crate) async fn handle_mcp_ui_action( let mut changed = false; let mut message = None; let is_reload = matches!(&action, crate::tui::app::McpUiAction::Reload); + let retry_name = match &action { + crate::tui::app::McpUiAction::Retry { name } => Some(name.clone()), + _ => None, + }; + let snapshot_live_pool = matches!(&action, crate::tui::app::McpUiAction::Show); let discover = mcp_ui_action_refreshes_discovery(&action); let action_result = match action { @@ -550,6 +555,7 @@ pub(crate) async fn handle_mcp_ui_action( } } crate::tui::app::McpUiAction::Validate | crate::tui::app::McpUiAction::Reload => Ok(()), + crate::tui::app::McpUiAction::Retry { .. } => Ok(()), }; if let Err(err) = action_result { @@ -570,12 +576,22 @@ pub(crate) async fn handle_mcp_ui_action( // second, easy-to-miss reload step. The standalone reload action remains // the retry/compatibility path for externally edited configuration. let rebuild_live_pool = is_reload || changed; - let snapshot_result = if rebuild_live_pool { + let snapshot_result = if let Some(name) = retry_name.as_deref() { + engine_handle + .retry_mcp_server(name) + .await + .map(|update| (update.snapshot, Some(update.generation))) + } else if snapshot_live_pool { + engine_handle + .bootstrap_mcp() + .await + .map(|update| (update.snapshot, Some(update.generation))) + } else if rebuild_live_pool { match engine_handle.reload_mcp(path.clone()).await { - Ok(snapshot) => { + Ok(update) => { app.mcp_reload_required = false; - add_mcp_message(app, mcp_reload_summary(&snapshot)); - Ok(snapshot) + add_mcp_message(app, mcp_reload_summary(&update.snapshot)); + Ok((update.snapshot, Some(update.generation))) } Err(error) => { app.mcp_reload_required = true; @@ -594,6 +610,7 @@ pub(crate) async fn handle_mcp_ui_action( std::sync::Arc::clone(&app.plugin_registry), ) .await + .map(|snapshot| (snapshot, None)) } else { mcp::manager_snapshot_from_config_with_workspace_and_plugins( &path, @@ -601,10 +618,11 @@ pub(crate) async fn handle_mcp_ui_action( app.mcp_reload_required, app.plugin_registry.as_ref(), ) + .map(|snapshot| (snapshot, None)) }; match snapshot_result { - Ok(snapshot) => { + Ok((snapshot, generation)) => { if discover { add_mcp_message( app, @@ -615,12 +633,22 @@ pub(crate) async fn handle_mcp_ui_action( // snapshot so footers and panels reflect post-/mcp edits // (#502). app.mcp_configured_count = snapshot.servers.len(); + if let Some(generation) = generation { + app.mcp_snapshot_generation = generation; + app.mcp_snapshot_generation_invalidated = true; + } app.mcp_snapshot = Some(snapshot.clone()); + app.mcp_initializing = false; + app.mcp_connecting.clear(); // #2068: keep the hotbar's MCP-tool actions in sync with the tools // that are actually loaded; the hotbar never connects on its own. app.hotbar_actions.replace_mcp_tools(Some(&snapshot)); open_mcp_manager_pager(app, &snapshot); } + Err(err) if retry_name.is_some() => add_mcp_message( + app, + format!("MCP server retry failed; the live tool pool is unchanged: {err}"), + ), Err(err) if rebuild_live_pool => add_mcp_message( app, format!("MCP reload failed; the live tool pool is unchanged: {err}"), diff --git a/crates/tui/src/tui/ui/provider_routes.rs b/crates/tui/src/tui/ui/provider_routes.rs index f72bd49cb3..fa364ddabd 100644 --- a/crates/tui/src/tui/ui/provider_routes.rs +++ b/crates/tui/src/tui/ui/provider_routes.rs @@ -690,8 +690,7 @@ pub(crate) fn mcp_reload_summary(snapshot: &crate::mcp::McpManagerSnapshot) -> S pub(crate) fn mcp_ui_action_refreshes_discovery(action: &crate::tui::app::McpUiAction) -> bool { matches!( action, - crate::tui::app::McpUiAction::Show - | crate::tui::app::McpUiAction::Validate + crate::tui::app::McpUiAction::Validate | crate::tui::app::McpUiAction::Login { .. } | crate::tui::app::McpUiAction::Logout { .. } | crate::tui::app::McpUiAction::ImportList diff --git a/crates/tui/src/tui/ui/tests.rs b/crates/tui/src/tui/ui/tests.rs index ba2b1960c6..13d85240e6 100644 --- a/crates/tui/src/tui/ui/tests.rs +++ b/crates/tui/src/tui/ui/tests.rs @@ -1704,7 +1704,10 @@ fn resume_hint_omits_missing_session_id() { fn plain_mcp_show_refreshes_discovery_counts() { use crate::tui::app::McpUiAction; - assert!(mcp_ui_action_refreshes_discovery(&McpUiAction::Show)); + assert!( + !mcp_ui_action_refreshes_discovery(&McpUiAction::Show), + "plain /mcp snapshots the engine-owned live pool, not a UI discovery pool" + ); assert!(mcp_ui_action_refreshes_discovery(&McpUiAction::Validate)); assert!( !mcp_ui_action_refreshes_discovery(&McpUiAction::Reload), @@ -1765,7 +1768,12 @@ async fn mcp_enable_persists_and_applies_the_live_tool_pool_in_one_action() { Op::ReloadMcp { config_path, tx } => { assert_eq!(config_path, path); let sender = tx.lock().unwrap().take().expect("reload reply sender"); - sender.send(Ok(response_snapshot)).expect("reload reply"); + sender + .send(Ok(crate::core::ops::McpManagerUpdate { + snapshot: response_snapshot, + generation: 7, + })) + .expect("reload reply"); } other => panic!("unexpected op: {other:?}"), } @@ -1791,6 +1799,8 @@ async fn mcp_enable_persists_and_applies_the_live_tool_pool_in_one_action() { ); assert!(!app.mcp_reload_required); assert_eq!(app.mcp_snapshot.as_ref(), Some(&snapshot)); + assert_eq!(app.mcp_snapshot_generation, 7); + assert!(app.mcp_snapshot_generation_invalidated); assert!(app.history.iter().any(|cell| matches!( cell, HistoryCell::System { content } diff --git a/crates/tui/src/tui/views/extensions.rs b/crates/tui/src/tui/views/extensions.rs index 764c4688af..3c5ff9c125 100644 --- a/crates/tui/src/tui/views/extensions.rs +++ b/crates/tui/src/tui/views/extensions.rs @@ -5,6 +5,7 @@ //! database, installer, or network fetch of its own. Future actions emitted by //! this view must delegate to the existing command/mutation controllers. +use std::borrow::Cow; use std::cell::RefCell; use std::collections::BTreeSet; use std::fmt::Write as _; @@ -935,8 +936,13 @@ fn mcp_model(app: &App, locale: Locale) -> ExtensionsTabModel { .map(|server| server.enabled) .or_else(|| config.map(crate::mcp::McpServerConfig::is_enabled)) .unwrap_or(true); + let initializing = app.mcp_initializing + && enabled + && observed.is_none_or(|server| !server.connected && server.error.is_none()); let state = if !enabled { tr(locale, MessageId::HotbarSetupStatusDisabled) + } else if initializing { + Cow::Borrowed("connecting") } else if observed.is_some_and(|server| server.connected) { tr(locale, MessageId::ExtensionsStateConnected) } else if observed.is_some_and(|server| server.error.is_some()) { @@ -955,13 +961,18 @@ fn mcp_model(app: &App, locale: Locale) -> ExtensionsTabModel { observed.and_then(|server| server.error.as_deref()), oauth_capable, ); - let action = if crate::mcp::mcp_name_is_command_safe(&name) + let action = if initializing { + ExtensionAction::Status { + label: state.clone(), + } + } else if crate::mcp::mcp_name_is_command_safe(&name) || matches!( recovery, crate::mcp::McpRecoveryKind::Connect | crate::mcp::McpRecoveryKind::Reconnect | crate::mcp::McpRecoveryKind::Diagnose - ) { + ) + { ExtensionAction::Command { label: tr(locale, recovery.label_key()).into_owned(), command: recovery.slash_command(&name),