fix: honor pending auth before token redirect

src/agent/agent_loop.rs

@@ -7,9 +7,10 @@
 //! - `commands` - System commands and job handlers
 //! - `thread_ops` - Thread/session operations (user input, undo, approval, persistence)
 
-use std::sync::Arc;
+use std::sync::{Arc, LazyLock};
 
 use futures::StreamExt;
+use regex::Regex;
 use uuid::Uuid;
 
 use crate::agent::context_monitor::ContextMonitor;
@@ -62,6 +63,38 @@ pub(crate) fn truncate_for_preview(output: &str, max_chars: usize) -> String {
     }
 }
 
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+enum SensitiveChatCredential {
+    TelegramBotToken,
+}
+
+impl SensitiveChatCredential {
+    fn extension_name(self) -> &'static str {
+        match self {
+            Self::TelegramBotToken => "telegram",
+        }
+    }
+
+    fn redirect_message(self) -> &'static str {
+        match self {
+            Self::TelegramBotToken => {
+                "Telegram bot tokens can't be accepted in normal chat. Use the secure Telegram setup flow instead."
+            }
+        }
+    }
+}
+
+static TELEGRAM_BOT_TOKEN_RE: LazyLock<Regex> =
+    LazyLock::new(|| Regex::new(r"^\d{6,}:[A-Za-z0-9_-]{20,}$").expect("TELEGRAM_BOT_TOKEN_RE")); // safety: hardcoded literal
+
+fn detect_sensitive_chat_credential(content: &str) -> Option<SensitiveChatCredential> {
+    let trimmed = content.trim();
+    if TELEGRAM_BOT_TOKEN_RE.is_match(trimmed) {
+        return Some(SensitiveChatCredential::TelegramBotToken);
+    }
+    None
+}
+
 #[cfg(test)]
 fn resolve_routine_notification_user(metadata: &serde_json::Value) -> Option<String> {
     resolve_owner_scope_notification_user(
@@ -191,6 +224,28 @@ pub struct Agent {
 }
 
 impl Agent {
+    async fn intercept_sensitive_chat_credential(
+        &self,
+        message: &IncomingMessage,
+        credential: SensitiveChatCredential,
+    ) -> String {
+        let instructions = credential.redirect_message().to_string();
+        let _ = self
+            .channels
+            .send_status(
+                &message.channel,
+                crate::channels::StatusUpdate::AuthRequired {
+                    extension_name: credential.extension_name().to_string(),
+                    instructions: Some(instructions.clone()),
+                    auth_url: None,
+                    setup_url: None,
+                },
+                &message.metadata,
+            )
+            .await;
+        instructions
+    }
+
     pub(super) fn owner_id(&self) -> &str {
         if let Some(workspace) = self.deps.workspace.as_ref() {
             debug_assert_eq!(
@@ -1125,6 +1180,15 @@ impl Agent {
             }
         }
 
+        if let Submission::UserInput { ref content } = submission {
+            if let Some(credential) = detect_sensitive_chat_credential(content) {
+                return Ok(Some(
+                    self.intercept_sensitive_chat_credential(message, credential)
+                        .await,
+                ));
+            }
+        }
+
         tracing::trace!(
             "Received message from {} on {} ({} chars)",
             message.user_id,
@@ -1321,11 +1385,26 @@ impl Agent {
 #[cfg(test)]
 mod tests {
     use super::{
-        chat_tool_execution_metadata, resolve_routine_notification_user,
+        Agent, AgentDeps, SensitiveChatCredential, chat_tool_execution_metadata,
+        detect_sensitive_chat_credential, resolve_routine_notification_user,
         should_fallback_routine_notification, truncate_for_preview,
     };
+    use crate::agent::session::Thread;
     use crate::channels::IncomingMessage;
     use crate::error::ChannelError;
+    use crate::testing::{StubChannel, StubLlm};
+    use crate::{
+        agent::cost_guard::{CostGuard, CostGuardConfig},
+        channels::{ChannelManager, StatusUpdate},
+        config::{AgentConfig, SafetyConfig, SkillsConfig},
+        context::ContextManager,
+        hooks::HookRegistry,
+        safety::SafetyLayer,
+        tools::ToolRegistry,
+    };
+    use std::sync::Arc;
+    use std::time::Duration;
+    use uuid::Uuid;
 
     #[test]
     fn test_truncate_short_input() {
@@ -1483,4 +1562,183 @@ mod tests {
 
         assert!(should_fallback_routine_notification(&error)); // safety: test-only assertion
     }
+
+    #[test]
+    fn detects_telegram_bot_token_messages() {
+        let detected = detect_sensitive_chat_credential("123456789:AABBccDDeeFFgg_Test-Token");
+        assert_eq!(detected, Some(SensitiveChatCredential::TelegramBotToken));
+    }
+
+    #[test]
+    fn ignores_normal_telegram_setup_messages() {
+        let detected = detect_sensitive_chat_credential(
+            "Can you help me connect Telegram without sharing the token here?",
+        );
+        assert_eq!(detected, None);
+    }
+
+    async fn make_gateway_test_agent(
+        llm: Arc<StubLlm>,
+    ) -> (Agent, Arc<std::sync::Mutex<Vec<StatusUpdate>>>) {
+        let llm_provider: Arc<dyn crate::llm::LlmProvider> = llm;
+        let (stub, _sender) = StubChannel::new("gateway");
+        let statuses = stub.captured_statuses_handle();
+        let channel_manager = ChannelManager::new();
+        channel_manager.add(Box::new(stub)).await;
+
+        let deps = AgentDeps {
+            owner_id: "default".to_string(),
+            store: None,
+            llm: llm_provider,
+            cheap_llm: None,
+            safety: Arc::new(SafetyLayer::new(&SafetyConfig {
+                max_output_length: 100_000,
+                injection_check_enabled: false,
+            })),
+            tools: Arc::new(ToolRegistry::new()),
+            workspace: None,
+            extension_manager: None,
+            skill_registry: None,
+            skill_catalog: None,
+            skills_config: SkillsConfig::default(),
+            hooks: Arc::new(HookRegistry::new()),
+            cost_guard: Arc::new(CostGuard::new(CostGuardConfig::default())),
+            sse_tx: None,
+            http_interceptor: None,
+            transcription: None,
+            document_extraction: None,
+            sandbox_readiness: crate::agent::routine_engine::SandboxReadiness::DisabledByConfig,
+            builder: None,
+        };
+
+        let agent = Agent::new(
+            AgentConfig {
+                name: "test-agent".to_string(),
+                max_parallel_jobs: 1,
+                job_timeout: Duration::from_secs(60),
+                stuck_threshold: Duration::from_secs(60),
+                repair_check_interval: Duration::from_secs(30),
+                max_repair_attempts: 1,
+                use_planning: false,
+                session_idle_timeout: Duration::from_secs(300),
+                allow_local_tools: false,
+                max_cost_per_day_cents: None,
+                max_actions_per_hour: None,
+                max_tool_iterations: 5,
+                auto_approve_tools: false,
+                default_timezone: "UTC".to_string(),
+                max_tokens_per_job: 0,
+            },
+            deps,
+            Arc::new(channel_manager),
+            None,
+            None,
+            None,
+            Some(Arc::new(ContextManager::new(1))),
+            None,
+        );
+
+        (agent, statuses)
+    }
+
+    #[tokio::test]
+    async fn telegram_bot_token_messages_are_redirected_before_llm() {
+        let llm = Arc::new(StubLlm::new("this should never be used"));
+        let llm_handle = Arc::clone(&llm);
+        let (agent, statuses) = make_gateway_test_agent(llm).await;
+        let message = IncomingMessage::new(
+            "gateway",
+            "test-user",
+            "123456789:AABBccDDeeFFgg_Test-Token",
+        );
+
+        let response = agent
+            .handle_message(&message)
+            .await
+            .expect("handle_message");
+
+        assert_eq!(
+            response.as_deref(),
+            Some(
+                "Telegram bot tokens can't be accepted in normal chat. Use the secure Telegram setup flow instead."
+            )
+        );
+        assert_eq!(llm_handle.calls(), 0, "LLM should not see raw bot tokens");
+
+        let statuses = statuses.lock().expect("poisoned");
+        assert_eq!(statuses.len(), 1);
+        assert!(matches!(
+            &statuses[0],
+            StatusUpdate::AuthRequired {
+                extension_name,
+                instructions,
+                auth_url: None,
+                setup_url: None,
+            } if extension_name == "telegram"
+                && instructions.as_deref()
+                    == Some(
+                        "Telegram bot tokens can't be accepted in normal chat. Use the secure Telegram setup flow instead."
+                    )
+        ));
+    }
+
+    #[tokio::test]
+    async fn telegram_bot_token_messages_still_flow_through_pending_auth_mode() {
+        let llm = Arc::new(StubLlm::new("this should never be used"));
+        let llm_handle = Arc::clone(&llm);
+        let (agent, statuses) = make_gateway_test_agent(llm).await;
+        let thread_id = Uuid::new_v4();
+        let session = agent
+            .session_manager
+            .get_or_create_session("test-user")
+            .await;
+
+        {
+            let mut sess = session.lock().await;
+            let mut thread = Thread::with_id(thread_id, sess.id);
+            thread.enter_auth_mode("telegram".to_string());
+            sess.threads.insert(thread_id, thread);
+            sess.active_thread = Some(thread_id);
+        }
+
+        agent
+            .session_manager
+            .register_thread("test-user", "gateway", thread_id, Arc::clone(&session))
+            .await;
+
+        let message = IncomingMessage::new(
+            "gateway",
+            "test-user",
+            "123456789:AABBccDDeeFFgg_Test-Token",
+        )
+        .with_thread(thread_id.to_string());
+
+        let response = agent
+            .handle_message(&message)
+            .await
+            .expect("handle_message");
+
+        assert_eq!(
+            response.as_deref(),
+            Some("Extension manager not available."),
+            "pending auth should consume the token instead of treating it as normal chat"
+        );
+        assert_eq!(llm_handle.calls(), 0, "LLM should not see auth-mode tokens");
+
+        let statuses = statuses.lock().expect("poisoned");
+        assert!(
+            statuses.is_empty(),
+            "no redirect status should be emitted when auth mode consumes the token"
+        );
+
+        let sess = session.lock().await;
+        let pending_auth = sess
+            .threads
+            .get(&thread_id)
+            .and_then(|thread| thread.pending_auth.as_ref());
+        assert!(
+            pending_auth.is_none(),
+            "auth mode should be cleared after the token is processed"
+        );
+    }
 }

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/workspace/seeds/BOOTSTRAP.md

@@ -35,6 +35,10 @@ If they're interested, set it up right here using the extension tools:
 3. Use `tool_auth` to collect credentials (e.g. Telegram bot token from @BotFather)
 4. The channel will be hot-activated — no restart needed
 
+Never ask the user to paste tokens, passwords, API keys, or other secrets into
+normal chat. If an extension has a secure auth/setup flow, always use that flow
+and keep the secret out of the conversation history.
+
 Don't push if they're not interested — note their preference and move on.
 
 ## Step 3: Save What You Learned (MANDATORY after 3 user messages)