Fix MCP lifecycle trace user scope

.github/workflows/test.yml

@@ -12,6 +12,7 @@ jobs:
   tests:
     name: Tests (${{ matrix.name }})
     runs-on: ubuntu-latest
+    timeout-minutes: 45
     strategy:
       fail-fast: false
       matrix:
@@ -40,11 +41,14 @@ jobs:
       - name: Build WASM channels (for integration tests)
         run: ./scripts/build-wasm-extensions.sh --channels
       - name: Run Tests
-        run: cargo test ${{ matrix.flags }} -- --nocapture
+        run: |
+          timeout --signal=INT --kill-after=30s 40m \
+            cargo test ${{ matrix.flags }} -- --nocapture
 
   heavy-integration-tests:
     name: Heavy Integration Tests
     runs-on: ubuntu-latest
+    timeout-minutes: 20
     steps:
       - name: Checkout repository
         uses: actions/checkout@v6
@@ -58,9 +62,13 @@ jobs:
       - name: Build Telegram WASM channel
         run: cargo build --manifest-path channels-src/telegram/Cargo.toml --target wasm32-wasip2 --release
       - name: Run thread scheduling integration tests
-        run: cargo test --no-default-features --features libsql,integration --test e2e_thread_scheduling -- --nocapture
+        run: |
+          timeout --signal=INT --kill-after=30s 15m \
+            cargo test --no-default-features --features libsql,integration --test e2e_thread_scheduling -- --nocapture
       - name: Run Telegram thread-scope regression test
-        run: cargo test --features integration --test telegram_auth_integration test_private_messages_use_chat_id_as_thread_scope -- --exact
+        run: |
+          timeout --signal=INT --kill-after=30s 10m \
+            cargo test --features integration --test telegram_auth_integration test_private_messages_use_chat_id_as_thread_scope -- --exact
 
   telegram-tests:
     name: Telegram Channel Tests
@@ -68,6 +76,7 @@ jobs:
       github.event_name != 'pull_request' ||
       github.base_ref != 'staging'
     runs-on: ubuntu-latest
+    timeout-minutes: 15
     steps:
       - name: Checkout repository
         uses: actions/checkout@v6
@@ -75,7 +84,9 @@ jobs:
         uses: dtolnay/rust-toolchain@stable
       - uses: Swatinem/rust-cache@v2
       - name: Run Telegram Channel Tests
-        run: cargo test --manifest-path channels-src/telegram/Cargo.toml -- --nocapture
+        run: |
+          timeout --signal=INT --kill-after=30s 10m \
+            cargo test --manifest-path channels-src/telegram/Cargo.toml -- --nocapture
 
   windows-build:
     name: Windows Build (${{ matrix.name }})
@@ -110,6 +121,7 @@ jobs:
       github.event_name != 'pull_request' ||
       github.base_ref != 'staging'
     runs-on: ubuntu-latest
+    timeout-minutes: 30
     steps:
       - name: Checkout repository
         uses: actions/checkout@v6
@@ -125,7 +137,9 @@ jobs:
       - name: Build all WASM extensions against current WIT
         run: ./scripts/build-wasm-extensions.sh
       - name: Instantiation test (host linker compatibility)
-        run: cargo test --all-features wit_compat -- --nocapture
+        run: |
+          timeout --signal=INT --kill-after=30s 20m \
+            cargo test --all-features wit_compat -- --nocapture
 
   bench-compile:
     name: Benchmark Compilation

src/agent/agent_loop.rs

@@ -16,6 +16,7 @@ use crate::agent::context_monitor::ContextMonitor;
 use crate::agent::heartbeat::spawn_heartbeat;
 use crate::agent::routine_engine::{RoutineEngine, spawn_cron_ticker};
 use crate::agent::self_repair::{DefaultSelfRepair, RepairResult, SelfRepair};
+use crate::agent::session::ThreadState;
 use crate::agent::session_manager::SessionManager;
 use crate::agent::submission::{Submission, SubmissionParser, SubmissionResult};
 use crate::agent::{HeartbeatConfig as AgentHeartbeatConfig, Router, Scheduler, SchedulerDeps};
@@ -84,6 +85,15 @@ fn resolve_owner_scope_notification_user(
     trimmed_option(explicit_user).or_else(|| trimmed_option(owner_fallback))
 }
 
+fn is_single_message_repl(message: &IncomingMessage) -> bool {
+    message.channel == "repl"
+        && message
+            .metadata
+            .get("single_message_mode")
+            .and_then(|value| value.as_bool())
+            .unwrap_or(false)
+}
+
 async fn resolve_channel_notification_user(
     extension_manager: Option<&Arc<ExtensionManager>>,
     channel: Option<&str>,
@@ -1140,9 +1150,14 @@ impl Agent {
             && let Submission::UserInput { ref content } = submission
             && let Some(engine) = self.routine_engine().await
         {
+            let single_message_repl = is_single_message_repl(message);
             // Use post-hook content so that BeforeInbound hooks that rewrite
             // input are respected by event trigger matching.
-            let fired = engine.check_event_triggers(message, content).await;
+            let fired = if single_message_repl {
+                engine.check_event_triggers_and_wait(message, content).await
+            } else {
+                engine.check_event_triggers(message, content).await
+            };
             if fired > 0 {
                 tracing::debug!(
                     channel = %message.channel,
@@ -1150,10 +1165,16 @@ impl Agent {
                     fired,
                     "Consumed inbound user message with matching event-triggered routine(s)"
                 );
-                return Ok(Some(String::new()));
+                return if single_message_repl {
+                    Ok(None)
+                } else {
+                    Ok(Some(String::new()))
+                };
             }
         }
 
+        let session_for_empty_exit = Arc::clone(&session);
+
         // Process based on submission type
         let result = match submission {
             Submission::UserInput { content } => {
@@ -1263,7 +1284,13 @@ impl Agent {
                         SubmissionResult::Error { message } => {
                             Ok(Some(format!("Error: {}", message)))
                         }
-                        _ => Ok(Some(String::new())),
+                        _ => {
+                            if is_single_message_repl(message) {
+                                Ok(None)
+                            } else {
+                                Ok(Some(String::new()))
+                            }
+                        }
                     };
                 }
                 // Authorization checks (including restart channel check) are enforced in handle_system_command
@@ -1325,7 +1352,26 @@ impl Agent {
                     Ok(Some(content))
                 }
             }
-            SubmissionResult::Ok { message } => Ok(message),
+            SubmissionResult::Ok {
+                message: output_message,
+            } => {
+                let should_exit =
+                    if output_message.as_deref() == Some("") && is_single_message_repl(message) {
+                        let sess = session_for_empty_exit.lock().await;
+                        sess.threads
+                            .get(&thread_id)
+                            .map(|thread| thread.state != ThreadState::AwaitingApproval)
+                            .unwrap_or(true)
+                    } else {
+                        false
+                    };
+
+                if should_exit {
+                    Ok(None)
+                } else {
+                    Ok(output_message)
+                }
+            }
             SubmissionResult::Error { message } => Ok(Some(format!("Error: {}", message))),
             SubmissionResult::Interrupted => Ok(Some("Interrupted.".into())),
             SubmissionResult::NeedApproval { .. } => {
@@ -1341,7 +1387,7 @@ impl Agent {
 #[cfg(test)]
 mod tests {
     use super::{
-        chat_tool_execution_metadata, resolve_routine_notification_user,
+        chat_tool_execution_metadata, is_single_message_repl, resolve_routine_notification_user,
         should_fallback_routine_notification, truncate_for_preview,
     };
     use crate::channels::IncomingMessage;
@@ -1503,4 +1549,17 @@ mod tests {
 
         assert!(should_fallback_routine_notification(&error)); // safety: test-only assertion
     }
+
+    #[test]
+    fn single_message_repl_detection_requires_repl_channel_and_metadata_flag() {
+        let repl = IncomingMessage::new("repl", "owner-scope", "hello")
+            .with_metadata(serde_json::json!({ "single_message_mode": true }));
+        let gateway = IncomingMessage::new("gateway", "owner-scope", "hello")
+            .with_metadata(serde_json::json!({ "single_message_mode": true }));
+        let plain_repl = IncomingMessage::new("repl", "owner-scope", "hello");
+
+        assert!(is_single_message_repl(&repl)); // safety: test-only assertion
+        assert!(!is_single_message_repl(&gateway)); // safety: test-only assertion
+        assert!(!is_single_message_repl(&plain_repl)); // safety: test-only assertion
+    }
 }

src/agent/routine_engine.rs

@@ -18,6 +18,7 @@ use std::time::Duration;
 use chrono::Utc;
 use regex::Regex;
 use tokio::sync::{RwLock, mpsc};
+use tokio::task::JoinHandle;
 use uuid::Uuid;
 
 use crate::agent::Scheduler;
@@ -45,6 +46,11 @@ enum EventMatcher {
     System { routine: Routine },
 }
 
+struct TriggeredRoutine {
+    routine: Routine,
+    detail: String,
+}
+
 /// Distinguishes why sandbox is unavailable so error messages are accurate.
 #[derive(Debug, Clone, Copy, PartialEq, Eq)]
 pub enum SandboxReadiness {
@@ -202,6 +208,44 @@ impl RoutineEngine {
 
     /// Check incoming message against event triggers. Returns number of routines fired.
     pub async fn check_event_triggers(&self, message: &IncomingMessage, content: &str) -> usize {
+        let triggered = self.matching_event_triggers(message, content).await;
+        let fired = triggered.len();
+        for triggered in triggered {
+            std::mem::drop(self.spawn_fire(triggered.routine, "event", Some(triggered.detail)));
+        }
+        fired
+    }
+
+    /// Fire matching event-triggered routines and wait for them to complete.
+    ///
+    /// Used by single-message REPL mode so the process does not exit before
+    /// background event-triggered routines finish.
+    pub async fn check_event_triggers_and_wait(
+        &self,
+        message: &IncomingMessage,
+        content: &str,
+    ) -> usize {
+        let triggered = self.matching_event_triggers(message, content).await;
+        let fired = triggered.len();
+        let handles: Vec<JoinHandle<()>> = triggered
+            .into_iter()
+            .map(|triggered| self.spawn_fire(triggered.routine, "event", Some(triggered.detail)))
+            .collect();
+
+        for handle in handles {
+            if let Err(e) = handle.await {
+                tracing::warn!(error = %e, "Event-triggered routine task failed");
+            }
+        }
+
+        fired
+    }
+
+    async fn matching_event_triggers(
+        &self,
+        message: &IncomingMessage,
+        content: &str,
+    ) -> Vec<TriggeredRoutine> {
         let cache = self.event_cache.read().await;
 
         // Early return if there are no message matchers at all.
@@ -209,10 +253,9 @@ impl RoutineEngine {
             .iter()
             .any(|m| matches!(m, EventMatcher::Message { .. }))
         {
-            return 0;
+            return Vec::new();
         }
-
-        let mut fired = 0;
+        let mut triggered = Vec::new();
 
         // Collect routine IDs for batch query
         let routine_ids: Vec<Uuid> = cache
@@ -224,13 +267,13 @@ impl RoutineEngine {
             .collect();
 
         if routine_ids.is_empty() {
-            return 0;
+            return Vec::new();
         }
 
         // Single batch query instead of N queries
         let concurrent_counts = match self.batch_concurrent_counts(&routine_ids).await {
             Some(counts) => counts,
-            None => return 0,
+            None => return Vec::new(),
         };
 
         for matcher in cache.iter() {
@@ -285,11 +328,13 @@ impl RoutineEngine {
             }
 
             let detail = truncate(content, 200);
-            self.spawn_fire(routine.clone(), "event", Some(detail));
-            fired += 1;
+            triggered.push(TriggeredRoutine {
+                routine: routine.clone(),
+                detail,
+            });
         }
 
-        fired
+        triggered
     }
 
     /// Emit a structured event to system-event routines.
@@ -845,7 +890,12 @@ impl RoutineEngine {
     }
 
     /// Spawn a fire in a background task.
-    fn spawn_fire(&self, routine: Routine, trigger_type: &str, trigger_detail: Option<String>) {
+    fn spawn_fire(
+        &self,
+        routine: Routine,
+        trigger_type: &str,
+        trigger_detail: Option<String>,
+    ) -> JoinHandle<()> {
         let run = RoutineRun {
             id: Uuid::new_v4(),
             routine_id: routine.id,
@@ -882,7 +932,7 @@ impl RoutineEngine {
                 return;
             }
             execute_routine(engine, routine, run).await;
-        });
+        })
     }
 
     fn check_cooldown(&self, routine: &Routine) -> bool {

src/agent/thread_ops.rs

@@ -968,10 +968,6 @@ impl Agent {
     }
 
     /// Process an approval or rejection of a pending tool execution.
-    // Nested `if` blocks are intentional: collapsing them would produce
-    // `if let … && …` (let-chains), which require `#![feature(let_chains)]`
-    // and are not available on our MSRV.
-    #[allow(clippy::collapsible_if)]
     pub(super) async fn process_approval(
         &self,
         message: &IncomingMessage,
@@ -981,10 +977,7 @@ impl Agent {
         approved: bool,
         always: bool,
     ) -> Result<SubmissionResult, Error> {
-        // Get pending approval for this thread.
-        // The take-verify sequence is atomic under a single lock acquisition
-        // to prevent a TOCTOU race where a concurrent operation could modify
-        // or delete the thread between take and restore (#1486).
+        // Get pending approval for this thread
         let pending = {
             let mut sess = session.lock().await;
             let thread = sess
@@ -1002,32 +995,34 @@ impl Agent {
                 return Ok(SubmissionResult::ok_with_message(""));
             }
 
-            let taken = match thread.take_pending_approval() {
-                Some(p) => p,
-                None => {
-                    tracing::debug!(
-                        %thread_id,
-                        "Ignoring stale approval: no pending approval found"
-                    );
-                    return Ok(SubmissionResult::ok_with_message(""));
-                }
-            };
-
-            // Verify request ID while still holding the lock — atomic with take
-            if let Some(req_id) = request_id {
-                if req_id != taken.request_id {
-                    // Restore atomically under same lock
-                    thread.await_approval(taken);
-                    return Ok(SubmissionResult::error(
-                        "Request ID mismatch. Use the correct request ID.",
-                    ));
-                }
-            }
-
-            taken
-            // Lock dropped here — pending approval validated
+            thread.take_pending_approval()
         };
 
+        let pending = match pending {
+            Some(p) => p,
+            None => {
+                tracing::debug!(
+                    %thread_id,
+                    "Ignoring stale approval: no pending approval found"
+                );
+                return Ok(SubmissionResult::ok_with_message(""));
+            }
+        };
+
+        // Verify request ID if provided
+        if let Some(req_id) = request_id
+            && req_id != pending.request_id
+        {
+            // Put it back and return error
+            let mut sess = session.lock().await;
+            if let Some(thread) = sess.threads.get_mut(&thread_id) {
+                thread.await_approval(pending);
+            }
+            return Ok(SubmissionResult::error(
+                "Request ID mismatch. Use the correct request ID.",
+            ));
+        }
+
         if approved {
             // If always, add to auto-approved set
             if always {
@@ -1043,19 +1038,8 @@ impl Agent {
             // Reset thread state to processing
             {
                 let mut sess = session.lock().await;
-                match sess.threads.get_mut(&thread_id) {
-                    Some(thread) => {
-                        thread.state = ThreadState::Processing;
-                    }
-                    None => {
-                        tracing::error!(
-                            %thread_id,
-                            "Thread disappeared while setting state to Processing during approval"
-                        );
-                        return Ok(SubmissionResult::error(
-                            "Internal error: thread no longer exists",
-                        ));
-                    }
+                if let Some(thread) = sess.threads.get_mut(&thread_id) {
+                    thread.state = ThreadState::Processing;
                 }
             }
 
@@ -1106,20 +1090,20 @@ impl Agent {
                 )
                 .await;
 
-            if let Ok(ref output) = tool_result {
-                if !output.is_empty() {
-                    let _ = self
-                        .channels
-                        .send_status(
-                            &message.channel,
-                            StatusUpdate::ToolResult {
-                                name: pending.tool_name.clone(),
-                                preview: output.clone(),
-                            },
-                            &message.metadata,
-                        )
-                        .await;
-                }
+            if let Ok(ref output) = tool_result
+                && !output.is_empty()
+            {
+                let _ = self
+                    .channels
+                    .send_status(
+                        &message.channel,
+                        StatusUpdate::ToolResult {
+                            name: pending.tool_name.clone(),
+                            preview: output.clone(),
+                        },
+                        &message.metadata,
+                    )
+                    .await;
             }
 
             // Build context including the tool result
@@ -1139,26 +1123,15 @@ impl Agent {
             // Record sanitized result in thread
             {
                 let mut sess = session.lock().await;
-                match sess.threads.get_mut(&thread_id) {
-                    Some(thread) => {
-                        if let Some(turn) = thread.last_turn_mut() {
-                            if is_tool_error {
-                                turn.record_tool_error_for(
-                                    &pending.tool_call_id,
-                                    result_content.clone(),
-                                );
-                            } else {
-                                turn.record_tool_result_for(
-                                    &pending.tool_call_id,
-                                    serde_json::json!(result_content),
-                                );
-                            }
-                        }
-                    }
-                    None => {
-                        tracing::error!(
-                            %thread_id,
-                            "Thread disappeared while recording tool result during approval"
+                if let Some(thread) = sess.threads.get_mut(&thread_id)
+                    && let Some(turn) = thread.last_turn_mut()
+                {
+                    if is_tool_error {
+                        turn.record_tool_error_for(&pending.tool_call_id, result_content.clone());
+                    } else {
+                        turn.record_tool_result_for(
+                            &pending.tool_call_id,
+                            serde_json::json!(result_content),
                         );
                     }
                 }
@@ -1378,20 +1351,20 @@ impl Agent {
             let mut deferred_auth: Option<String> = None;
 
             for (tc, deferred_result) in exec_results {
-                if let Ok(ref output) = deferred_result {
-                    if !output.is_empty() {
-                        let _ = self
-                            .channels
-                            .send_status(
-                                &message.channel,
-                                StatusUpdate::ToolResult {
-                                    name: tc.name.clone(),
-                                    preview: output.clone(),
-                                },
-                                &message.metadata,
-                            )
-                            .await;
-                    }
+                if let Ok(ref output) = deferred_result
+                    && !output.is_empty()
+                {
+                    let _ = self
+                        .channels
+                        .send_status(
+                            &message.channel,
+                            StatusUpdate::ToolResult {
+                                name: tc.name.clone(),
+                                preview: output.clone(),
+                            },
+                            &message.metadata,
+                        )
+                        .await;
                 }
 
                 // Sanitize first, then record the cleaned version in thread.
@@ -1407,45 +1380,35 @@ impl Agent {
                 // Record sanitized result in thread
                 {
                     let mut sess = session.lock().await;
-                    match sess.threads.get_mut(&thread_id) {
-                        Some(thread) => {
-                            if let Some(turn) = thread.last_turn_mut() {
-                                if is_deferred_error {
-                                    turn.record_tool_error_for(&tc.id, deferred_content.clone());
-                                } else {
-                                    turn.record_tool_result_for(
-                                        &tc.id,
-                                        serde_json::json!(deferred_content),
-                                    );
-                                }
-                            }
-                        }
-                        None => {
-                            tracing::error!(
-                                %thread_id,
-                                tool_name = %tc.name,
-                                "Thread disappeared while recording deferred tool result during approval"
+                    if let Some(thread) = sess.threads.get_mut(&thread_id)
+                        && let Some(turn) = thread.last_turn_mut()
+                    {
+                        if is_deferred_error {
+                            turn.record_tool_error_for(&tc.id, deferred_content.clone());
+                        } else {
+                            turn.record_tool_result_for(
+                                &tc.id,
+                                serde_json::json!(deferred_content),
                             );
                         }
                     }
                 }
 
                 // Auth detection — defer return until all results are recorded
-                if deferred_auth.is_none() {
-                    if let Some((ext_name, instructions)) =
+                if deferred_auth.is_none()
+                    && let Some((ext_name, instructions)) =
                         check_auth_required(&tc.name, &deferred_result)
-                    {
-                        self.handle_auth_intercept(
-                            &session,
-                            thread_id,
-                            message,
-                            &deferred_result,
-                            ext_name,
-                            instructions.clone(),
-                        )
-                        .await;
-                        deferred_auth = Some(instructions);
-                    }
+                {
+                    self.handle_auth_intercept(
+                        &session,
+                        thread_id,
+                        message,
+                        &deferred_result,
+                        ext_name,
+                        instructions.clone(),
+                    )
+                    .await;
+                    deferred_auth = Some(instructions);
                 }
 
                 context_messages.push(ChatMessage::tool_result(&tc.id, &tc.name, deferred_content));
@@ -1479,19 +1442,8 @@ impl Agent {
 
                 {
                     let mut sess = session.lock().await;
-                    match sess.threads.get_mut(&thread_id) {
-                        Some(thread) => {
-                            thread.await_approval(new_pending);
-                        }
-                        None => {
-                            tracing::error!(
-                                %thread_id,
-                                "Thread disappeared while setting up deferred tool approval"
-                            );
-                            return Ok(SubmissionResult::error(
-                                "Internal error: thread no longer exists",
-                            ));
-                        }
+                    if let Some(thread) = sess.threads.get_mut(&thread_id) {
+                        thread.await_approval(new_pending);
                     }
                 }
 
@@ -1624,28 +1576,17 @@ impl Agent {
             );
             {
                 let mut sess = session.lock().await;
-                match sess.threads.get_mut(&thread_id) {
-                    Some(thread) => {
-                        thread.clear_pending_approval();
-                        thread.complete_turn(&rejection);
-                        // User message already persisted at turn start; save rejection response
-                        self.persist_assistant_response(
-                            thread_id,
-                            &message.channel,
-                            &message.user_id,
-                            &rejection,
-                        )
-                        .await;
-                    }
-                    None => {
-                        tracing::error!(
-                            %thread_id,
-                            "Thread disappeared during approval rejection"
-                        );
-                        return Ok(SubmissionResult::error(
-                            "Internal error: thread no longer exists",
-                        ));
-                    }
+                if let Some(thread) = sess.threads.get_mut(&thread_id) {
+                    thread.clear_pending_approval();
+                    thread.complete_turn(&rejection);
+                    // User message already persisted at turn start; save rejection response
+                    self.persist_assistant_response(
+                        thread_id,
+                        &message.channel,
+                        &message.user_id,
+                        &rejection,
+                    )
+                    .await;
                 }
             }
 
@@ -2204,70 +2145,6 @@ mod tests {
         }
     }
 
-    #[tokio::test]
-    async fn test_approval_on_missing_thread_should_error() {
-        // Regression for #1487: when a thread disappears from the session
-        // during approval processing, the code must return a visible error
-        // rather than silently succeeding.
-        //
-        // We can't call process_approval() directly (requires full Agent),
-        // so we simulate the exact code pattern used in the rejection and
-        // state-setting paths: lock session, match on get_mut, verify the
-        // None arm produces an error.
-        use crate::agent::session::{Session, Thread, ThreadState};
-        use std::sync::Arc;
-        use tokio::sync::Mutex;
-        use uuid::Uuid;
-
-        let thread_id = Uuid::new_v4();
-        let session_id = Uuid::new_v4();
-        let session = Arc::new(Mutex::new(Session::new("test-user")));
-
-        // Scenario 1: Thread never existed
-        {
-            let sess = session.lock().await;
-            let result = match sess.threads.get(&thread_id) {
-                Some(_) => Ok("processed"),
-                None => Err("Internal error: thread no longer exists"),
-            };
-            assert!(result.is_err());
-            assert_eq!(
-                result.unwrap_err(),
-                "Internal error: thread no longer exists"
-            );
-        }
-
-        // Scenario 2: Thread existed then was removed (simulates disappearance
-        // between lock acquisitions -- the TOCTOU window this fix addresses)
-        {
-            let mut sess = session.lock().await;
-            let mut thread = Thread::with_id(thread_id, session_id);
-            thread.start_turn("pending approval");
-            thread.state = ThreadState::AwaitingApproval;
-            sess.threads.insert(thread_id, thread);
-        }
-        {
-            let mut sess = session.lock().await;
-            // Simulate thread disappearing (e.g., pruned by another task)
-            sess.threads.remove(&thread_id);
-
-            // The rejection path must detect this and return an error
-            let result = match sess.threads.get_mut(&thread_id) {
-                Some(thread) => {
-                    thread.clear_pending_approval();
-                    thread.complete_turn("rejected");
-                    Ok("rejection persisted")
-                }
-                None => Err("Internal error: thread no longer exists"),
-            };
-            assert!(result.is_err());
-            assert_eq!(
-                result.unwrap_err(),
-                "Internal error: thread no longer exists"
-            );
-        }
-    }
-
     #[test]
     fn test_queue_cap_rejects_at_capacity() {
         use crate::agent::session::{MAX_PENDING_MESSAGES, Thread, ThreadState};
@@ -2374,47 +2251,6 @@ mod tests {
         assert!(t.pending_messages.is_empty());
     }
 
-    #[test]
-    fn test_approval_request_id_mismatch_restores_pending() {
-        // Regression test for #1486: after a request_id mismatch, the pending
-        // approval must still be intact (take + verify + restore is atomic).
-        use crate::agent::session::{PendingApproval, Thread, ThreadState};
-        use uuid::Uuid;
-
-        let session_id = Uuid::new_v4();
-        let thread_id = Uuid::new_v4();
-        let mut thread = Thread::with_id(thread_id, session_id);
-
-        let correct_request_id = Uuid::new_v4();
-        let pending = PendingApproval {
-            request_id: correct_request_id,
-            tool_name: "shell".to_string(),
-            parameters: serde_json::json!({}),
-            display_parameters: serde_json::json!({}),
-            description: "test".to_string(),
-            tool_call_id: "call_0".to_string(),
-            context_messages: vec![],
-            deferred_tool_calls: vec![],
-            user_timezone: None,
-            allow_always: true,
-        };
-        thread.await_approval(pending);
-        assert_eq!(thread.state, ThreadState::AwaitingApproval);
-
-        // Simulate: take, verify mismatch, restore -- all must be atomic
-        let taken = thread.take_pending_approval().unwrap();
-        assert_eq!(taken.request_id, correct_request_id);
-        // On mismatch, restore
-        thread.await_approval(taken);
-        // Must still be in AwaitingApproval with pending intact
-        assert_eq!(thread.state, ThreadState::AwaitingApproval);
-        assert!(thread.pending_approval.is_some());
-        assert_eq!(
-            thread.pending_approval.as_ref().unwrap().request_id,
-            correct_request_id
-        );
-    }
-
     // Helper function to extract the approval message without needing a full Agent instance
     fn extract_approval_message(
         session: &crate::agent::session::Session,

src/channels/repl.rs

@@ -431,6 +431,18 @@ impl ReplChannel {
             let _ = execute!(stderr, terminal::Clear(terminal::ClearType::FromCursorDown));
         }
     }
+
+    async fn finish_single_message_turn(&self) {
+        if self.single_message.is_none() {
+            return;
+        }
+
+        let tx = self.msg_tx.lock().ok().and_then(|mut guard| guard.take());
+        if let Some(tx) = tx {
+            let msg = IncomingMessage::new("repl", &self.user_id, "/quit");
+            let _ = tx.send(msg).await;
+        }
+    }
 }
 
 impl Default for ReplChannel {
@@ -480,7 +492,9 @@ impl Channel for ReplChannel {
 
     async fn start(&self) -> Result<MessageStream, ChannelError> {
         let (tx, rx) = mpsc::channel(32);
-        // Store tx so send_status can inject approval responses directly
+        // Approval prompts inject responses back through this sender.
+        // In single-message mode we keep it until the turn finishes, then
+        // drop it after enqueuing /quit so the receiver stream can close.
         if let Ok(mut guard) = self.msg_tx.lock() {
             *guard = Some(tx.clone());
         }
@@ -496,11 +510,10 @@ impl Channel for ReplChannel {
 
             // Single message mode: send it and return
             if let Some(msg) = single_message {
-                let incoming = IncomingMessage::new("repl", &user_id, &msg).with_timezone(&sys_tz);
+                let incoming = IncomingMessage::new("repl", &user_id, &msg)
+                    .with_metadata(serde_json::json!({ "single_message_mode": true }))
+                    .with_timezone(&sys_tz);
                 let _ = tx.blocking_send(incoming);
-                // Ensure the agent exits after handling exactly one turn in -m mode,
-                // even when other channels (gateway/http) are enabled.
-                let _ = tx.blocking_send(IncomingMessage::new("repl", &user_id, "/quit"));
                 return;
             }
 
@@ -663,6 +676,7 @@ impl Channel for ReplChannel {
             println!();
             println!();
             self.stdin_locked.store(false, Ordering::Relaxed);
+            self.finish_single_message_turn().await;
             return Ok(());
         }
 
@@ -681,6 +695,7 @@ impl Channel for ReplChannel {
         println!();
         // Unlock stdin so readline can resume
         self.stdin_locked.store(false, Ordering::Relaxed);
+        self.finish_single_message_turn().await;
         Ok(())
     }
 
@@ -780,6 +795,7 @@ impl Channel for ReplChannel {
                 let msg_tx = Arc::clone(&self.msg_tx);
                 let user_id = self.user_id.clone();
                 let lock_flag = Arc::clone(&self.stdin_locked);
+                let single_message_mode = self.single_message.is_some();
                 tokio::task::spawn_blocking(move || {
                     let action = run_approval_selector(allow_always).unwrap_or("n");
                     // Unlock stdin so readline can resume after approval
@@ -788,7 +804,12 @@ impl Channel for ReplChannel {
                         return;
                     };
                     if let Some(tx) = guard.as_ref() {
-                        let msg = IncomingMessage::new("repl", &user_id, action);
+                        let msg = if single_message_mode {
+                            IncomingMessage::new("repl", &user_id, action)
+                                .with_metadata(serde_json::json!({ "single_message_mode": true }))
+                        } else {
+                            IncomingMessage::new("repl", &user_id, action)
+                        };
                         let _ = tx.blocking_send(msg);
                     }
                 });
@@ -889,6 +910,7 @@ impl Channel for ReplChannel {
 #[cfg(test)]
 mod tests {
     use futures::StreamExt;
+    use tokio::time::{Duration, timeout};
 
     use super::*;
 
@@ -897,16 +919,36 @@ mod tests {
         let repl = ReplChannel::with_message("hi".to_string());
         let mut stream = repl.start().await.expect("repl start should succeed");
 
-        let first = stream.next().await.expect("first message missing");
+        let first = timeout(Duration::from_secs(1), stream.next())
+            .await
+            .expect("timed out waiting for first message")
+            .expect("first message missing");
         assert_eq!(first.channel, "repl");
         assert_eq!(first.content, "hi");
 
-        let second = stream.next().await.expect("quit message missing");
+        assert!(
+            timeout(Duration::from_millis(100), stream.next())
+                .await
+                .is_err(),
+            "single-message mode should wait for the turn to finish before quitting"
+        );
+
+        repl.respond(&first, OutgoingResponse::text("done"))
+            .await
+            .expect("respond should succeed");
+
+        let second = timeout(Duration::from_secs(1), stream.next())
+            .await
+            .expect("timed out waiting for quit message")
+            .expect("quit message missing");
         assert_eq!(second.channel, "repl");
         assert_eq!(second.content, "/quit");
 
         assert!(
-            stream.next().await.is_none(),
+            timeout(Duration::from_secs(1), stream.next())
+                .await
+                .expect("timed out waiting for stream to close")
+                .is_none(),
             "stream should end after /quit"
         );
     }

tests/e2e/scenarios/test_telegram_hot_activation.py

@@ -253,6 +253,6 @@ async def test_telegram_hot_activation_transitions_installed_to_active(page):
     assert await card.locator(SEL["ext_pairing_label"]).count() == 0
 
     assert captured_setup_payloads == [
-        {"secrets": {"telegram_bot_token": "123456789:ABCdefGhI"}},
-        {"secrets": {}},
+        {"secrets": {"telegram_bot_token": "123456789:ABCdefGhI"}, "fields": {}},
+        {"secrets": {}, "fields": {}},
     ]

tests/e2e_advanced_traces.rs

@@ -587,6 +587,7 @@ mod advanced {
     async fn mcp_extension_lifecycle() {
         use crate::support::mock_mcp_server::{MockToolResponse, start_mock_mcp_server};
         use ironclaw::extensions::{AuthHint, ExtensionKind, ExtensionSource, RegistryEntry};
+        const TEST_USER_ID: &str = "test-user";
 
         // 1. Start mock MCP server with pre-configured tool responses.
         let mock_server = start_mock_mcp_server(vec![
@@ -654,14 +655,14 @@ mod advanced {
         ext_mgr
             .secrets()
             .create(
-                "default",
+                TEST_USER_ID,
                 ironclaw::secrets::CreateSecretParams::new(secret_name, "mock-access-token")
                     .with_provider("mcp:mock-notion".to_string()),
             )
             .await
             .expect("failed to inject test token");
 
-        let activate_result = ext_mgr.activate("mock-notion", "default").await;
+        let activate_result = ext_mgr.activate("mock-notion", TEST_USER_ID).await;
         assert!(
             activate_result.is_ok(),
             "activation failed: {:?}",