fix(mcp): show startup progress as each server connects

crates/tui/src/core/engine.rs

@@ -6329,6 +6329,7 @@ impl Engine {
                 }
                 self.mcp_event_generation = generation;
                 self.replace_mcp_boot_errors(&authority_errors, connection_errors);
+                self.session.pending_prefix_change_reason = Some("mcp-session-boot".to_string());
                 if let Ok(snapshot) = self.mcp_session_snapshot().await {
                     let _ = self.tx_event.try_send(Event::McpSessionBoot {
                         generation,
@@ -6382,6 +6383,8 @@ impl Engine {
                     }
                     self.mcp_event_generation = generation;
                     self.replace_mcp_boot_errors(&authority_errors, connection_errors);
+                    self.session.pending_prefix_change_reason =
+                        Some("mcp-session-boot".to_string());
                 }
                 McpBootUpdate::Finished {
                     generation,
@@ -6484,33 +6487,54 @@ impl Engine {
             async move {
                 let mut remaining: Vec<String> =
                     pending.iter().map(|(name, _)| name.clone()).collect();
-                let results = McpPool::connect_pending_concurrently(
+                let mut connects = McpPool::spawn_pending_connects(
                     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) => {
-                                pool.note_connect_failure(&name, &error);
-                                connection_errors
-                                    .insert(name, crate::mcp::format_mcp_error_for_display(&error));
+                while let Some(joined) = connects.join_next().await {
+                    let (name, result) = joined
+                        .unwrap_or_else(|error| ("connection task".to_string(), Err(error.into())));
+                    remaining.retain(|pending_name| pending_name != &name);
+                    {
+                        let mut pool = pool_for_task.lock().await;
+                        // A turn may have reloaded the pool while these handshakes
+                        // were in flight. Never let their old authority or failures
+                        // overwrite the newly installed configuration.
+                        let reload = pool.reload_if_config_changed().await;
+                        if reload.is_err()
+                            || pool.current_catalog_generation() != catalog_generation
+                        {
+                            connects.abort_all();
+                            connection_errors.clear();
+                            if let Err(error) = reload {
+                                connection_errors.insert(
+                                    "configuration".to_string(),
+                                    crate::mcp::format_mcp_error_for_display(&error),
+                                );
                             }
+                            break;
                         }
-                        let _ = progress_tx.send(McpBootUpdate::Progress {
-                            generation,
-                            authority_errors: Arc::clone(&authority_errors),
-                            connection_errors: connection_errors.clone(),
-                            connecting: remaining.clone(),
+                        let result = result.and_then(|connection| {
+                            pool.store_ready_connection(name.clone(), connection)
                         });
+                        if let Err(error) = result {
+                            pool.note_connect_failure(&name, &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 pool = pool_for_task.lock().await;
                     let mut required = Vec::new();
                     pool.push_required_server_errors(&mut required);
                     for (name, error) in required {
@@ -6609,8 +6633,9 @@ impl Engine {
         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).
+            // behind the slowest handshake. Declare the refresh here as well
+            // as on Progress: a ready connection can precede its mailbox update.
+            self.session.pending_prefix_change_reason = Some("mcp-session-boot".to_string());
             return pool.lock().await.to_api_tools();
         }
 

crates/tui/src/core/engine/tests.rs

@@ -20754,6 +20754,148 @@ async fn reload_mcp_op_recovers_from_invalid_initial_config_in_process() {
     task.await.expect("engine task");
 }
 
+#[tokio::test]
+async fn mcp_boot_reports_ready_server_before_stalled_server_finishes() {
+    assert_incremental_mcp_boot(false).await;
+}
+
+#[tokio::test]
+async fn mcp_boot_does_not_restore_servers_removed_during_handshake() {
+    assert_incremental_mcp_boot(true).await;
+}
+
+async fn assert_incremental_mcp_boot(invalidate_config: bool) {
+    if std::process::Command::new("node")
+        .arg("--version")
+        .output()
+        .is_err()
+    {
+        tracing::warn!("skipping MCP stdio fixture because node is unavailable");
+        return;
+    }
+    let tmp = tempdir().expect("tempdir");
+    let server = tmp.path().join("server.mjs");
+    let release = tmp.path().join("release-slow");
+    std::fs::write(
+        &server,
+        r#"import fs from 'node:fs';
+import readline from 'node:readline';
+const lines = readline.createInterface({ input: process.stdin });
+lines.on('line', async (line) => {
+  const request = JSON.parse(line);
+  if (request.id === undefined) return;
+  if (process.argv[2] === 'slow' && request.method === 'initialize') {
+    while (!fs.existsSync(process.argv[3])) {
+      await new Promise(resolve => setTimeout(resolve, 10));
+    }
+  }
+  const result = request.method === 'initialize'
+    ? { protocolVersion: '2024-11-05', capabilities: { tools: {} },
+        serverInfo: { name: process.argv[2], version: '1' } }
+    : { tools: [{ name: 'ready', inputSchema: { type: 'object' } }] };
+  process.stdout.write(JSON.stringify({ jsonrpc: '2.0', id: request.id, result }) + '\n');
+});
+"#,
+    )
+    .expect("server fixture");
+    let config_path = tmp.path().join("mcp.json");
+    std::fs::write(
+        &config_path,
+        serde_json::to_vec(&serde_json::json!({
+            "timeouts": { "connect_timeout": 30 },
+            "servers": {
+                "fast": { "command": "node", "args": [server, "fast", release] },
+                "slow": { "command": "node", "args": [server, "slow", release] }
+            }
+        }))
+        .expect("config JSON"),
+    )
+    .expect("MCP config");
+    let (mut engine, handle) = Engine::new(
+        EngineConfig {
+            workspace: tmp.path().to_path_buf(),
+            mcp_config_path: config_path.clone(),
+            ..Default::default()
+        },
+        &Config::default(),
+    );
+    let pool = engine.ensure_mcp_pool().await.expect("engine pool");
+    let task = tokio::spawn(async move { engine.run().await });
+    let mut events = handle.rx_event.write().await;
+    let progress = tokio::time::timeout(Duration::from_secs(10), async {
+        while let Some(event) = events.recv().await {
+            if let Event::McpSessionBoot {
+                snapshot,
+                connecting,
+                finished: false,
+                ..
+            } = event
+                && connecting == ["slow"]
+            {
+                return snapshot;
+            }
+        }
+        panic!("engine event channel closed");
+    })
+    .await;
+    let ready_tools = pool.lock().await.to_api_tools();
+    if invalidate_config {
+        std::fs::write(
+            &config_path,
+            r#"{"servers":{"slow":{"command":"node","disabled":true}}}"#,
+        )
+        .expect("remove servers");
+        pool.lock()
+            .await
+            .reload_if_config_changed()
+            .await
+            .expect("reload config");
+    }
+    // Release and shut down even when testing the old batch-buffered behavior.
+    std::fs::write(&release, "continue").expect("release stalled fixture");
+    let finished = tokio::time::timeout(Duration::from_secs(10), async {
+        while let Some(event) = events.recv().await {
+            if let Event::McpSessionBoot {
+                snapshot,
+                finished: true,
+                ..
+            } = event
+            {
+                return snapshot;
+            }
+        }
+        panic!("engine event channel closed");
+    })
+    .await;
+    drop(events);
+    handle.send(Op::Shutdown).await.expect("shutdown");
+    task.await.expect("engine task");
+    let progress = progress.expect("fast server must be visible before slow server is released");
+    assert!(
+        progress
+            .servers
+            .iter()
+            .any(|row| row.name == "fast" && row.connected)
+    );
+    assert!(
+        progress
+            .servers
+            .iter()
+            .any(|row| row.name == "slow" && !row.connected)
+    );
+    assert!(ready_tools.iter().any(|tool| tool.name == "mcp_fast_ready"));
+    assert!(!ready_tools.iter().any(|tool| tool.name == "mcp_slow_ready"));
+    let finished = finished.expect("finished boot");
+    if invalidate_config {
+        assert_eq!(finished.servers.len(), 1);
+        assert!(!finished.servers[0].enabled);
+        assert!(!finished.servers[0].connected);
+        assert!(pool.lock().await.to_api_tools().is_empty());
+    } else {
+        assert!(finished.servers.iter().all(|row| row.connected));
+    }
+}
+
 #[tokio::test]
 async fn mcp_boot_updates_preserve_authority_errors_and_replace_ordinary_errors() {
     let tmp = tempdir().expect("tempdir");
@@ -20799,6 +20941,10 @@ async fn mcp_boot_updates_preserve_authority_errors_and_replace_ordinary_errors(
         ])
     );
     assert!(!engine.mcp_connection_errors.contains_key("stale-transport"));
+    assert_eq!(
+        engine.session.pending_prefix_change_reason.as_deref(),
+        Some("mcp-session-boot")
+    );
 
     engine.mcp_connection_errors.insert(
         "stale-between-updates".to_string(),
@@ -20829,6 +20975,42 @@ async fn mcp_boot_updates_preserve_authority_errors_and_replace_ordinary_errors(
     );
 }
 
+#[tokio::test]
+async fn mcp_boot_catalog_refresh_declares_prefix_before_mailbox_delivery() {
+    let tmp = tempdir().expect("tempdir");
+    let (mut engine, _handle) = Engine::new(
+        EngineConfig {
+            workspace: tmp.path().to_path_buf(),
+            ..Default::default()
+        },
+        &Config::default(),
+    );
+    engine.mcp_boot_generation = Some(1);
+    engine.mcp_boot_in_flight = true;
+    engine.session.pending_prefix_change_reason = None;
+    let _tools = engine.mcp_tools().await;
+    assert_eq!(
+        engine.session.pending_prefix_change_reason.as_deref(),
+        Some("mcp-session-boot")
+    );
+
+    engine.session.pending_prefix_change_reason = None;
+    let (tx, rx) = tokio::sync::mpsc::unbounded_channel();
+    engine.mcp_boot_rx = Some(rx);
+    tx.send(McpBootUpdate::Progress {
+        generation: 1,
+        authority_errors: Arc::new(HashMap::new()),
+        connection_errors: HashMap::new(),
+        connecting: vec!["slow".to_string()],
+    })
+    .expect("queue progress");
+    engine.drain_mcp_boot_updates().await;
+    assert_eq!(
+        engine.session.pending_prefix_change_reason.as_deref(),
+        Some("mcp-session-boot")
+    );
+}
+
 #[tokio::test]
 async fn stale_boot_finished_does_not_clear_a_newer_receiver() {
     let tmp = tempdir().expect("tempdir");