fix(agent): surface active extension state to the LLM

src/agent/dispatcher.rs

@@ -131,6 +131,18 @@ impl Agent {
             }
         }
 
+        if let Some(extension_manager) = self.deps.extension_manager.as_ref() {
+            match extension_manager.llm_extension_state_summary().await {
+                Ok(Some(summary)) => {
+                    reasoning = reasoning.with_extension_state_summary(summary);
+                }
+                Ok(None) => {}
+                Err(e) => {
+                    tracing::debug!("Could not load extension state summary: {}", e);
+                }
+            }
+        }
+
         if let Some(prompt) = system_prompt {
             reasoning = reasoning.with_system_prompt(prompt);
         }

src/agent/session_manager.rs

@@ -102,9 +102,6 @@ impl SessionManager {
     /// Resolve an external thread ID to an internal thread.
     ///
     /// Returns the session and thread ID. Creates both if they don't exist.
-    ///
-    /// Uses a single read-lock acquisition for both the key lookup and the UUID
-    /// adoption check to reduce contention under concurrent approval load.
     pub async fn resolve_thread(
         &self,
         user_id: &str,
@@ -119,57 +116,51 @@ impl SessionManager {
             external_thread_id: external_thread_id.map(String::from),
         };
 
-        // Parse UUID once outside the lock (if applicable)
-        let ext_uuid = external_thread_id.and_then(|s| Uuid::parse_str(s).ok());
-
-        // Single read lock for both the key lookup and UUID adoption check
-        let adoptable_uuid = {
+        // Check if we have a mapping
+        {
             let thread_map = self.thread_map.read().await;
-
-            // Fast path: exact key match
             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);
                 }
             }
+        }
 
-            // UUID adoption check (still under the same read lock).
-            // If external_thread_id is a valid UUID not mapped elsewhere,
-            // it may be a thread created by chat_new_thread_handler or
-            // hydrated from DB that we can adopt.
-            if let Some(ext_uuid) = ext_uuid {
-                let mapped_elsewhere = thread_map.values().any(|&v| v == ext_uuid);
-                if !mapped_elsewhere {
-                    Some(ext_uuid)
-                } else {
-                    None
+        // Check if external_thread_id is itself a known thread UUID that
+        // exists in the session but was never registered in the thread_map
+        // (e.g. created by chat_new_thread_handler or hydrated from DB).
+        // We only adopt it if no thread_map entry maps to this UUID —
+        // otherwise it belongs to a different channel scope.
+        if let Some(ext_tid) = external_thread_id
+            && let Ok(ext_uuid) = Uuid::parse_str(ext_tid)
+        {
+            let thread_map = self.thread_map.read().await;
+            let mapped_elsewhere = thread_map.values().any(|&v| v == ext_uuid);
+            drop(thread_map);
+
+            if !mapped_elsewhere {
+                let sess = session.lock().await;
+                if sess.threads.contains_key(&ext_uuid) {
+                    drop(sess);
+
+                    let mut thread_map = self.thread_map.write().await;
+                    // Re-check after acquiring write lock to prevent race condition
+                    // where another task mapped this UUID between our read and write.
+                    if !thread_map.values().any(|&v| v == ext_uuid) {
+                        thread_map.insert(key, ext_uuid);
+                        drop(thread_map);
+                        // Ensure undo manager exists
+                        let mut undo_managers = self.undo_managers.write().await;
+                        undo_managers
+                            .entry(ext_uuid)
+                            .or_insert_with(|| Arc::new(Mutex::new(UndoManager::new())));
+                        return (session, ext_uuid);
+                    }
+                    // If it was mapped elsewhere while we were unlocked, fall through
+                    // to create a new thread, preserving channel isolation.
                 }
-            } else {
-                None
-            }
-        }; // Single read lock dropped here
-
-        // If we found an adoptable UUID, verify it exists in session and acquire write lock
-        if let Some(ext_uuid) = adoptable_uuid {
-            let sess = session.lock().await;
-            if sess.threads.contains_key(&ext_uuid) {
-                drop(sess);
-
-                let mut thread_map = self.thread_map.write().await;
-                // Re-check after acquiring write lock to prevent race condition
-                // where another task mapped this UUID between our read and write.
-                if !thread_map.values().any(|&v| v == ext_uuid) {
-                    thread_map.insert(key, ext_uuid);
-                    drop(thread_map);
-                    // Ensure undo manager exists
-                    let mut undo_managers = self.undo_managers.write().await;
-                    undo_managers
-                        .entry(ext_uuid)
-                        .or_insert_with(|| Arc::new(Mutex::new(UndoManager::new())));
-                    return (session, ext_uuid);
-                }
-                // If mapped elsewhere while unlocked, fall through to create new thread
             }
         }
 
@@ -918,44 +909,6 @@ mod tests {
         }
     }
 
-    #[tokio::test]
-    async fn test_resolve_thread_consolidates_read_path() {
-        // Verify that resolve_thread still correctly handles:
-        // 1. Fast path: key exists in thread_map
-        // 2. UUID adoption: external_thread_id is a UUID in session but not in map
-        // 3. New thread: neither path matches
-        use crate::agent::session::Thread;
-
-        let manager = SessionManager::new();
-
-        // Case 1: Normal resolution creates thread and maps it
-        let (session1, tid1) = manager
-            .resolve_thread("user1", "chan1", Some("ext-1"))
-            .await;
-        // Resolving again with same key should return same thread (fast path)
-        let (_, tid1_again) = manager
-            .resolve_thread("user1", "chan1", Some("ext-1"))
-            .await;
-        assert_eq!(tid1, tid1_again);
-
-        // Case 2: UUID adoption - insert a thread directly into session
-        let adopted_id = Uuid::new_v4();
-        {
-            let mut sess = session1.lock().await;
-            let thread = Thread::with_id(adopted_id, sess.id);
-            sess.threads.insert(adopted_id, thread);
-        }
-        // Resolve with the UUID as external_thread_id -- should adopt it
-        let (_, resolved) = manager
-            .resolve_thread("user1", "chan1", Some(&adopted_id.to_string()))
-            .await;
-        assert_eq!(resolved, adopted_id);
-
-        // Case 3: Different channel gets different thread
-        let (_, tid2) = manager.resolve_thread("user1", "chan2", None).await;
-        assert_ne!(tid1, tid2);
-    }
-
     #[tokio::test]
     async fn test_resolve_thread_finds_existing_session_thread_by_uuid() {
         use crate::agent::session::{Session, Thread};

src/extensions/manager.rs

@@ -1498,6 +1498,52 @@ impl ExtensionManager {
         Ok(extensions)
     }
 
+    /// Build a compact, deterministic extension snapshot for LLM prompt context.
+    pub async fn llm_extension_state_summary(&self) -> Result<Option<String>, ExtensionError> {
+        let mut extensions = self.list(None, false).await?;
+        extensions.sort_by(|a, b| {
+            llm_extension_sort_key(a.kind)
+                .cmp(&llm_extension_sort_key(b.kind))
+                .then_with(|| a.name.cmp(&b.name))
+        });
+
+        let mut channels = Vec::new();
+        let mut tools = Vec::new();
+        let mut servers = Vec::new();
+
+        for extension in extensions {
+            let owner_bound = matches!(extension.kind, ExtensionKind::WasmChannel)
+                && self.has_wasm_channel_owner_binding(&extension.name).await;
+            if !(extension.active || extension.authenticated || owner_bound) {
+                continue;
+            }
+
+            let item = llm_extension_summary_item(&extension, owner_bound);
+            match extension.kind {
+                ExtensionKind::WasmChannel | ExtensionKind::ChannelRelay => channels.push(item),
+                ExtensionKind::WasmTool => tools.push(item),
+                ExtensionKind::McpServer => servers.push(item),
+            }
+        }
+
+        let mut lines = Vec::new();
+        if !channels.is_empty() {
+            lines.push(format!("- Channels: {}", channels.join("; ")));
+        }
+        if !tools.is_empty() {
+            lines.push(format!("- Tools: {}", tools.join("; ")));
+        }
+        if !servers.is_empty() {
+            lines.push(format!("- MCP servers: {}", servers.join("; ")));
+        }
+
+        if lines.is_empty() {
+            Ok(None)
+        } else {
+            Ok(Some(lines.join("\n")))
+        }
+    }
+
     /// Remove an installed extension.
     pub async fn remove(&self, name: &str) -> Result<String, ExtensionError> {
         Self::validate_extension_name(name)?;
@@ -5613,6 +5659,40 @@ fn combine_install_errors(
     }
 }
 
+fn llm_extension_sort_key(kind: ExtensionKind) -> u8 {
+    match kind {
+        ExtensionKind::WasmChannel | ExtensionKind::ChannelRelay => 0,
+        ExtensionKind::WasmTool => 1,
+        ExtensionKind::McpServer => 2,
+    }
+}
+
+fn llm_extension_summary_item(extension: &InstalledExtension, owner_bound: bool) -> String {
+    let mut states = Vec::new();
+    if extension.authenticated {
+        states.push("authenticated".to_string());
+    }
+    if extension.active {
+        states.push("active".to_string());
+    } else if extension.authenticated {
+        states.push("inactive".to_string());
+    }
+    if owner_bound {
+        states.push("owner-bound".to_string());
+    }
+    if !extension.tools.is_empty() {
+        let mut tool_names = extension.tools.clone();
+        tool_names.sort();
+        states.push(format!("tools: {}", tool_names.join(", ")));
+    }
+
+    if states.is_empty() {
+        extension.name.clone()
+    } else {
+        format!("{} ({})", extension.name, states.join(", "))
+    }
+}
+
 #[cfg(test)]
 mod tests {
     use std::fmt::Debug;
@@ -6571,6 +6651,142 @@ mod tests {
         )
     }
 
+    #[cfg(feature = "libsql")]
+    #[tokio::test]
+    async fn test_llm_extension_state_summary_reports_active_owner_bound_telegram()
+    -> Result<(), String> {
+        let dir = tempfile::tempdir().map_err(|err| format!("temp dir: {err}"))?;
+        let channels_dir = dir.path().join("channels");
+        std::fs::create_dir_all(&channels_dir).map_err(|err| format!("channels dir: {err}"))?;
+        std::fs::write(channels_dir.join("telegram.wasm"), b"mock")
+            .map_err(|err| format!("write wasm: {err}"))?;
+        std::fs::write(
+            channels_dir.join("telegram.capabilities.json"),
+            serde_json::to_vec(&serde_json::json!({
+                "type": "channel",
+                "name": "telegram",
+                "setup": {
+                    "required_secrets": [
+                        {
+                            "name": "telegram_bot_token",
+                            "prompt": "Enter your Telegram Bot API token (from @BotFather)",
+                            "optional": false
+                        }
+                    ]
+                },
+                "capabilities": {
+                    "channel": {
+                        "allowed_paths": ["/webhook/telegram"]
+                    }
+                },
+                "config": {
+                    "owner_id": null
+                }
+            }))
+            .map_err(|err| format!("serialize capabilities: {err}"))?,
+        )
+        .map_err(|err| format!("write capabilities: {err}"))?;
+
+        let (db, _db_tmp) = crate::testing::test_db().await;
+        let manager = {
+            use crate::secrets::{InMemorySecretsStore, SecretsCrypto};
+            use crate::testing::credentials::TEST_CRYPTO_KEY;
+            use crate::tools::ToolRegistry;
+            use crate::tools::mcp::process::McpProcessManager;
+            use crate::tools::mcp::session::McpSessionManager;
+
+            let master_key = secrecy::SecretString::from(TEST_CRYPTO_KEY.to_string());
+            let crypto = Arc::new(
+                SecretsCrypto::new(master_key)
+                    .map_err(|err| format!("failed to construct test crypto: {err}"))?,
+            );
+
+            ExtensionManager::new(
+                Arc::new(McpSessionManager::new()),
+                Arc::new(McpProcessManager::new()),
+                Arc::new(InMemorySecretsStore::new(crypto)),
+                Arc::new(ToolRegistry::new()),
+                None,
+                None,
+                dir.path().join("tools"),
+                channels_dir.clone(),
+                None,
+                "test".to_string(),
+                Some(db),
+                Vec::new(),
+            )
+        };
+
+        let channel_manager = Arc::new(ChannelManager::new());
+        let runtime = Arc::new(
+            WasmChannelRuntime::new(WasmChannelRuntimeConfig::for_testing())
+                .map_err(|err| format!("runtime: {err}"))?,
+        );
+        let pairing_store = Arc::new(PairingStore::with_base_dir(
+            dir.path().join("pairing-state"),
+        ));
+        let router = Arc::new(WasmChannelRouter::new());
+        manager
+            .set_channel_runtime(
+                Arc::clone(&channel_manager),
+                Arc::clone(&runtime),
+                Arc::clone(&pairing_store),
+                Arc::clone(&router),
+                std::collections::HashMap::new(),
+            )
+            .await;
+        manager
+            .set_test_wasm_channel_loader(Arc::new({
+                let runtime = Arc::clone(&runtime);
+                let pairing_store = Arc::clone(&pairing_store);
+                move |name| {
+                    Ok(make_test_loaded_channel(
+                        Arc::clone(&runtime),
+                        name,
+                        Arc::clone(&pairing_store),
+                    ))
+                }
+            }))
+            .await;
+        manager
+            .set_test_telegram_binding_resolver(Arc::new(|_token, existing_owner_id| {
+                if existing_owner_id.is_some() {
+                    return Err(ExtensionError::Other(
+                        "owner binding should be derived during setup".to_string(),
+                    ));
+                }
+                Ok(TelegramBindingResult::Bound(TelegramBindingData {
+                    owner_id: 424242,
+                    bot_username: Some("test_hot_bot".to_string()),
+                    binding_state: TelegramOwnerBindingState::VerifiedNow,
+                }))
+            }))
+            .await;
+
+        manager
+            .configure(
+                "telegram",
+                &std::collections::HashMap::from([(
+                    "telegram_bot_token".to_string(),
+                    "123456789:ABCdefGhI".to_string(),
+                )]),
+                &std::collections::HashMap::new(),
+            )
+            .await
+            .map_err(|err| format!("configure succeeds: {err}"))?;
+
+        let summary = manager
+            .llm_extension_state_summary()
+            .await
+            .map_err(|err| format!("summary: {err}"))?
+            .ok_or_else(|| "expected extension summary".to_string())?;
+
+        require(
+            summary.contains("- Channels: telegram (authenticated, active, owner-bound)"),
+            format!("unexpected summary: {summary}"),
+        )
+    }
+
     #[tokio::test]
     async fn test_telegram_hot_activation_returns_verification_challenge_before_binding()
     -> Result<(), String> {

src/llm/reasoning.rs

@@ -353,6 +353,8 @@ pub struct Reasoning {
     workspace_system_prompt: Option<String>,
     /// Optional skill context block to inject into system prompt.
     skill_context: Option<String>,
+    /// Optional snapshot of connected/active extensions for the current user.
+    extension_state_summary: Option<String>,
     /// Channel name (e.g. "discord", "telegram") for formatting hints.
     channel: Option<String>,
     /// Model name for runtime context.
@@ -371,6 +373,7 @@ impl Reasoning {
             llm,
             workspace_system_prompt: None,
             skill_context: None,
+            extension_state_summary: None,
             channel: None,
             model_name: None,
             is_group_chat: false,
@@ -400,6 +403,14 @@ impl Reasoning {
         self
     }
 
+    /// Set extension runtime context to inject into the system prompt.
+    pub fn with_extension_state_summary(mut self, summary: String) -> Self {
+        if !summary.is_empty() {
+            self.extension_state_summary = Some(summary);
+        }
+        self
+    }
+
     /// Set the channel name for channel-specific formatting hints.
     pub fn with_channel(mut self, channel: impl Into<String>) -> Self {
         let ch = channel.into();
@@ -932,21 +943,54 @@ Example:
     }
 
     fn build_extensions_section_for_tools(&self, tools: &[ToolDefinition]) -> String {
-        // Only include when the extension management tools are available
-        let has_ext_tools = tools.iter().any(|t| t.name == "tool_search");
-        if !has_ext_tools {
+        let has_search = tools.iter().any(|t| t.name == "tool_search");
+        let has_list = tools.iter().any(|t| t.name == "tool_list");
+        let has_info = tools.iter().any(|t| t.name == "extension_info");
+
+        if self.extension_state_summary.is_none() && !has_search && !has_list && !has_info {
             return String::new();
         }
 
-        "\n\n## Extensions\n\
-         You can search, install, and activate extensions to add new capabilities:\n\
-         - **Channels** (Telegram, Slack, Discord) — messaging integrations. \
-         When users ask about connecting a messaging platform, search for it as a channel.\n\
-         - **Tools** — sandboxed functions that extend your abilities.\n\
-         - **MCP servers** — external API integrations via the Model Context Protocol.\n\n\
-         Use `tool_search` to find extensions by name. Refer to them by their kind \
-         (channel, tool, or server) — not as \"MCP server\" generically."
-            .to_string()
+        let mut blocks = Vec::new();
+        if let Some(ref summary) = self.extension_state_summary {
+            blocks.push(format!(
+                "Current extension state for this user:\n{}",
+                summary
+            ));
+        }
+
+        if has_search || has_list || has_info {
+            let mut guidance = String::from(
+                "You can search, install, and activate extensions to add new capabilities:\n\
+                 - **Channels** (Telegram, Slack, Discord) — messaging integrations. \
+                 When users ask about connecting a messaging platform, search for it as a channel.\n\
+                 - **Tools** — sandboxed functions that extend your abilities.\n\
+                 - **MCP servers** — external API integrations via the Model Context Protocol.",
+            );
+
+            if has_list {
+                guidance.push_str(
+                    "\n\nBefore telling the user to connect, activate, or re-enable an extension, \
+                     inspect the current state with `tool_list`.",
+                );
+            }
+            if has_info {
+                guidance.push_str(
+                    "\nUse `extension_info` when you need deeper compatibility or runtime details \
+                     for an installed extension.",
+                );
+            }
+            if has_search {
+                guidance.push_str(
+                    "\nUse `tool_search` to find extensions by name. Refer to them by their kind \
+                     (channel, tool, or server) — not as \"MCP server\" generically.",
+                );
+            }
+
+            blocks.push(guidance);
+        }
+
+        format!("\n\n## Extensions\n{}", blocks.join("\n\n"))
     }
 
     fn build_channel_section(&self) -> String {
@@ -2285,6 +2329,42 @@ That's my plan."#;
         );
     }
 
+    #[test]
+    fn test_system_prompt_includes_extension_runtime_summary() {
+        let reasoning = make_test_reasoning().with_extension_state_summary(
+            "- Channels: telegram (authenticated, active, owner-bound)".to_string(),
+        );
+
+        let prompt = reasoning.build_system_prompt_with_tools(&[]);
+        assert!(
+            prompt.contains("## Extensions"),
+            "Prompt should contain an Extensions section when runtime state is present"
+        );
+        assert!(
+            prompt.contains("telegram (authenticated, active, owner-bound)"),
+            "Prompt should include the injected extension runtime summary"
+        );
+    }
+
+    #[test]
+    fn test_system_prompt_extension_guidance_prefers_inspection_before_reconnect() {
+        let reasoning = make_test_reasoning();
+        let prompt = reasoning.build_system_prompt_with_tools(&make_tools(&[
+            "tool_search",
+            "tool_list",
+            "extension_info",
+        ]));
+
+        assert!(
+            prompt.contains("inspect the current state with `tool_list`"),
+            "Prompt should direct the model to inspect installed/active state first"
+        );
+        assert!(
+            prompt.contains("Use `extension_info` when you need deeper compatibility"),
+            "Prompt should mention extension_info for deeper extension details"
+        );
+    }
+
     // ---- plan/evaluate bypass clean_response (Bug #564-2) ----
 
     #[test]