fix: orphaned tool_results and parallel tool_result merging in Worker

src/agent/agent_loop.rs

@@ -560,6 +560,8 @@ impl Agent {
                     // Spawn notification forwarder (mirrors heartbeat pattern)
                     let channels = self.channels.clone();
                     let extension_manager = self.deps.extension_manager.clone();
+                    let session_mgr_for_notify = self.session_manager.clone();
+                    let owner_id_for_notify = self.owner_id().to_string();
                     tokio::spawn(async move {
                         while let Some(response) = notify_rx.recv().await {
                             let notify_channel = response
@@ -587,6 +589,38 @@ impl Agent {
                                 continue;
                             };
 
+                            // Inject notification into the user's Telegram thread so
+                            // that when the user replies, the LLM has context about
+                            // what was notified.
+                            {
+                                let broadcast_channel = notify_channel.as_deref().unwrap_or("telegram");
+                                let (session, thread_id) = session_mgr_for_notify
+                                    .resolve_thread(&owner_id_for_notify, broadcast_channel, None)
+                                    .await;
+                                let mut sess = session.lock().await;
+                                if let Some(thread) = sess.threads.get_mut(&thread_id) {
+                                    let routine_name = response.metadata
+                                        .get("routine_name")
+                                        .and_then(|v| v.as_str())
+                                        .unwrap_or("routine");
+                                    let notification_text = format!(
+                                        "[Notification from '{}'] {}",
+                                        routine_name,
+                                        response.content
+                                    );
+                                    thread.start_turn(&notification_text);
+                                    thread.complete_turn(format!(
+                                        "I sent this notification to you via {}.",
+                                        broadcast_channel
+                                    ));
+                                    tracing::debug!(
+                                        routine = %routine_name,
+                                        thread_id = %thread_id,
+                                        "Injected routine notification into user thread for context"
+                                    );
+                                }
+                            }
+
                             // Try the configured channel first, fall back to
                             // broadcasting on all channels.
                             let targeted_ok = if let Some(ref channel) = notify_channel {

src/agent/session_manager.rs

@@ -116,14 +116,33 @@ impl SessionManager {
             external_thread_id: external_thread_id.map(String::from),
         };
 
+        const THREAD_IDLE_TIMEOUT: std::time::Duration = std::time::Duration::from_secs(30 * 60);
+
         // Check if we have a mapping
         {
             let thread_map = self.thread_map.read().await;
             if let Some(&thread_id) = thread_map.get(&key) {
-                // Verify thread still exists in session
                 let sess = session.lock().await;
-                if sess.threads.contains_key(&thread_id) {
-                    return (Arc::clone(&session), thread_id);
+                if let Some(thread) = sess.threads.get(&thread_id) {
+                    if external_thread_id.is_none() {
+                        let idle_duration = chrono::Utc::now()
+                            .signed_duration_since(thread.updated_at);
+                        if idle_duration > chrono::Duration::from_std(THREAD_IDLE_TIMEOUT)
+                            .unwrap_or(chrono::Duration::seconds(1800))
+                        {
+                            tracing::info!(
+                                thread_id = %thread_id,
+                                idle_secs = idle_duration.num_seconds(),
+                                "Thread idle timeout exceeded, creating new thread"
+                            );
+                            drop(sess);
+                            drop(thread_map);
+                        } else {
+                            return (Arc::clone(&session), thread_id);
+                        }
+                    } else {
+                        return (Arc::clone(&session), thread_id);
+                    }
                 }
             }
         }

src/app.rs

@@ -520,7 +520,7 @@ impl AppBuilder {
                                             server_name,
                                             e
                                         );
-                                        return;
+                                        return None;
                                     }
                                 };
 
@@ -537,13 +537,15 @@ impl AppBuilder {
                                                     tool_count,
                                                     server_name
                                                 );
+                                                Some((server_name, std::sync::Arc::new(client)))
                                             }
                                             Err(e) => {
                                                 tracing::warn!(
                                                     "Failed to create tools from MCP server '{}': {}",
                                                     server_name,
                                                     e
                                                 );
+                                                None
                                             }
                                         }
                                     }
@@ -565,16 +567,23 @@ impl AppBuilder {
                                                 e
                                             );
                                         }
+                                        None
                                     }
                                 }
                             });
                         }
 
+                        let mut startup_clients = Vec::new();
                         while let Some(result) = join_set.join_next().await {
-                            if let Err(e) = result {
-                                tracing::warn!("MCP server loading task panicked: {}", e);
+                            match result {
+                                Ok(Some(client_pair)) => startup_clients.push(client_pair),
+                                Ok(None) => {}
+                                Err(e) => {
+                                    tracing::warn!("MCP server loading task panicked: {}", e);
+                                }
                             }
                         }
+                        return startup_clients;
                     }
                     Err(e) => {
                         if matches!(
@@ -592,10 +601,11 @@ impl AppBuilder {
                         }
                     }
                 }
+                Vec::new()
             }
         };
 
-        let (dev_loaded_tool_names, _) = tokio::join!(wasm_tools_future, mcp_servers_future);
+        let (dev_loaded_tool_names, startup_mcp_clients) = tokio::join!(wasm_tools_future, mcp_servers_future);
 
         // Load registry catalog entries for extension discovery
         let mut catalog_entries = match crate::registry::RegistryCatalog::load_or_embedded() {
@@ -657,6 +667,17 @@ impl AppBuilder {
             ));
             tools.register_extension_tools(Arc::clone(&manager));
             tracing::debug!("Extension manager initialized with in-chat discovery tools");
+
+            if !startup_mcp_clients.is_empty() {
+                tracing::info!(
+                    "Injecting {} startup MCP client(s) into extension manager",
+                    startup_mcp_clients.len()
+                );
+                for (name, client) in startup_mcp_clients {
+                    manager.inject_mcp_client(name, client).await;
+                }
+            }
+
             Some(manager)
         };
 

src/extensions/manager.rs

@@ -911,6 +911,11 @@ impl ExtensionManager {
         Ok(results)
     }
 
+    /// Inject a pre-created MCP client (from startup loading) into the manager.
+    pub async fn inject_mcp_client(&self, name: String, client: std::sync::Arc<crate::tools::mcp::McpClient>) {
+        self.mcp_clients.write().await.insert(name, client);
+    }
+
     /// Install an extension by name (from registry) or by explicit URL.
     pub async fn install(
         &self,