Merge remote-tracking branch 'origin/staging' into feat/multi-tenant-isolation-phases-2-4

src/agent/agent_loop.rs

@@ -1091,10 +1091,11 @@ impl Agent {
             } else {
                 drop(sess);
                 self.session_manager
-                    .resolve_thread(
+                    .resolve_thread_with_parsed_uuid(
                         &message.user_id,
                         &message.channel,
                         message.conversation_scope(),
+                        approval_thread_uuid,
                     )
                     .await
             }
@@ -1175,9 +1176,9 @@ impl Agent {
             && let Submission::UserInput { ref content } = submission
             && let Some(engine) = self.routine_engine().await
         {
-            let fired = engine
-                .check_event_triggers(&message.user_id, &message.channel, content)
-                .await;
+            // 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;
             if fired > 0 {
                 tracing::debug!(
                     channel = %message.channel,

src/agent/routine_engine.rs

@@ -24,7 +24,7 @@ use crate::agent::Scheduler;
 use crate::agent::routine::{
     NotifyConfig, Routine, RoutineAction, RoutineRun, RunStatus, Trigger, next_cron_fire,
 };
-use crate::channels::OutgoingResponse;
+use crate::channels::{IncomingMessage, OutgoingResponse};
 use crate::config::RoutineConfig;
 use crate::context::{JobContext, JobState};
 use crate::db::Database;
@@ -56,6 +56,40 @@ pub enum SandboxReadiness {
     DockerUnavailable,
 }
 
+/// Check whether an event-triggered routine's user/channel filters match an
+/// incoming message.
+///
+/// Returns `true` if:
+/// - The routine has an `Event` trigger (non-Event routines always return `false`)
+/// - The routine's `user_id` matches the message's user scope
+/// - The routine's channel filter (if any) matches the message channel
+///   case-insensitively
+///
+/// This is a pure function extracted from `check_event_triggers` so the
+/// filter logic can be unit-tested without async infrastructure.
+pub(crate) fn routine_matches_message(routine: &Routine, message: &IncomingMessage) -> bool {
+    // Only Event-triggered routines can match incoming messages.
+    if !matches!(routine.trigger, Trigger::Event { .. }) {
+        return false;
+    }
+
+    // User ownership filter — only fire routines scoped to this user.
+    if routine.user_id != message.user_id {
+        return false;
+    }
+
+    // Channel filter (case-insensitive, matching emit_system_event behavior)
+    if let Trigger::Event {
+        channel: Some(ch), ..
+    } = &routine.trigger
+        && !ch.eq_ignore_ascii_case(&message.channel)
+    {
+        return false;
+    }
+
+    true
+}
+
 /// The routine execution engine.
 pub struct RoutineEngine {
     config: RoutineConfig,
@@ -167,10 +201,7 @@ impl RoutineEngine {
     }
 
     /// Check incoming message against event triggers. Returns number of routines fired.
-    ///
-    /// Accepts only the three fields needed for matching (user scope, channel,
-    /// message content) so callers never need to clone a full `IncomingMessage`.
-    pub async fn check_event_triggers(&self, user_id: &str, channel: &str, content: &str) -> usize {
+    pub async fn check_event_triggers(&self, message: &IncomingMessage, content: &str) -> usize {
         let cache = self.event_cache.read().await;
 
         // Early return if there are no message matchers at all.
@@ -208,16 +239,24 @@ impl RoutineEngine {
                 EventMatcher::System { .. } => continue,
             };
 
-            if routine.user_id != user_id {
-                continue;
-            }
-
-            // Channel filter
-            if let Trigger::Event {
-                channel: Some(ch), ..
-            } = &routine.trigger
-                && ch != channel
-            {
+            // User ownership + channel filter (extracted for testability).
+            if !routine_matches_message(routine, message) {
+                // User mismatch is expected for multi-user setups — keep at
+                // trace to avoid one log per routine per inbound message.
+                if routine.user_id != message.user_id {
+                    tracing::trace!(
+                        routine = %routine.name,
+                        routine_user = %routine.user_id,
+                        message_user = %message.user_id,
+                        "Skipped: user scope mismatch"
+                    );
+                } else {
+                    tracing::debug!(
+                        routine = %routine.name,
+                        channel = %message.channel,
+                        "Skipped: channel mismatch"
+                    );
+                }
                 continue;
             }
 
@@ -228,14 +267,14 @@ impl RoutineEngine {
 
             // Cooldown check
             if !self.check_cooldown(routine) {
-                tracing::trace!(routine = %routine.name, "Skipped: cooldown active");
+                tracing::debug!(routine = %routine.name, "Skipped: cooldown active");
                 continue;
             }
 
             // Concurrent run check (using batch-loaded counts)
             let running_count = concurrent_counts.get(&routine.id).copied().unwrap_or(0);
             if running_count >= routine.guardrails.max_concurrent as i64 {
-                tracing::trace!(routine = %routine.name, "Skipped: max concurrent reached");
+                tracing::debug!(routine = %routine.name, "Skipped: max concurrent reached");
                 continue;
             }
 
@@ -1790,6 +1829,13 @@ pub fn spawn_cron_ticker(
         engine.check_cron_triggers().await;
 
         let mut ticker = tokio::time::interval(interval);
+        ticker.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Skip);
+        // Periodic event cache refresh so web/CLI mutations are picked up
+        // without requiring tool-path code to call refresh_event_cache().
+        // Uses wall-clock elapsed time so the refresh cadence is stable
+        // regardless of the cron tick interval configuration.
+        let refresh_interval = Duration::from_secs(60);
+        let mut last_refresh = tokio::time::Instant::now();
 
         loop {
             ticker.tick().await;
@@ -1797,7 +1843,11 @@ pub fn spawn_cron_ticker(
             // never races with FullJobWatcher instances from this process.
             engine.sync_dispatched_runs().await;
             engine.check_cron_triggers().await;
-            engine.sync_dispatched_runs().await;
+
+            if last_refresh.elapsed() >= refresh_interval {
+                engine.refresh_event_cache().await;
+                last_refresh = tokio::time::Instant::now();
+            }
         }
     })
 }
@@ -1863,7 +1913,13 @@ fn strip_html_tags(s: &str) -> String {
 
 #[cfg(test)]
 mod tests {
-    use crate::agent::routine::{NotifyConfig, RunStatus};
+    use chrono::Utc;
+    use uuid::Uuid;
+
+    use crate::agent::routine::{
+        NotifyConfig, Routine, RoutineAction, RoutineGuardrails, RunStatus, Trigger,
+    };
+    use crate::channels::IncomingMessage;
     use crate::config::RoutineConfig;
 
     #[test]
@@ -2061,6 +2117,117 @@ mod tests {
         }
     }
 
+    /// Helper to build a test routine with the given user_id and trigger.
+    fn make_routine(user_id: &str, trigger: Trigger) -> Routine {
+        Routine {
+            id: Uuid::new_v4(),
+            name: "test".to_string(),
+            description: String::new(),
+            user_id: user_id.to_string(),
+            enabled: true,
+            trigger,
+            action: RoutineAction::Lightweight {
+                prompt: String::new(),
+                context_paths: vec![],
+                max_tokens: 1000,
+                use_tools: false,
+                max_tool_rounds: 0,
+            },
+            guardrails: RoutineGuardrails::default(),
+            notify: Default::default(),
+            last_run_at: None,
+            next_fire_at: None,
+            run_count: 0,
+            consecutive_failures: 0,
+            state: serde_json::Value::Null,
+            created_at: Utc::now(),
+            updated_at: Utc::now(),
+        }
+    }
+
+    /// Helper to build a test IncomingMessage.
+    fn make_message(user_id: &str, channel: &str, content: &str) -> IncomingMessage {
+        IncomingMessage {
+            id: Uuid::new_v4(),
+            channel: channel.to_string(),
+            user_id: user_id.to_string(),
+            owner_id: user_id.to_string(),
+            sender_id: user_id.to_string(),
+            user_name: None,
+            content: content.to_string(),
+            thread_id: None,
+            conversation_scope_id: None,
+            received_at: Utc::now(),
+            metadata: serde_json::Value::Null,
+            timezone: None,
+            attachments: vec![],
+            is_internal: false,
+        }
+    }
+
+    /// Regression test for issue #1051: event triggers used case-sensitive
+    /// channel comparison, so "Telegram" != "telegram" caused silent mismatch.
+    /// Tests the actual `routine_matches_message` function used in `check_event_triggers`.
+    #[test]
+    fn test_channel_filter_is_case_insensitive() {
+        let routine = make_routine(
+            "user1",
+            Trigger::Event {
+                pattern: ".*".to_string(),
+                channel: Some("Telegram".to_string()),
+            },
+        );
+        let msg = make_message("user1", "telegram", "hello");
+
+        // Case-insensitive channel match must succeed
+        assert!(super::routine_matches_message(&routine, &msg));
+
+        // Exact case must also work
+        let msg_exact = make_message("user1", "Telegram", "hello");
+        assert!(super::routine_matches_message(&routine, &msg_exact));
+
+        // Different channel must not match
+        let msg_wrong = make_message("user1", "discord", "hello");
+        assert!(!super::routine_matches_message(&routine, &msg_wrong));
+    }
+
+    /// Regression test for issue #1051: event triggers did not filter by
+    /// user_id, so routines from user A could fire on messages from user B.
+    /// Tests the actual `routine_matches_message` function used in `check_event_triggers`.
+    #[test]
+    fn test_event_trigger_requires_user_match() {
+        let routine = make_routine(
+            "alice",
+            Trigger::Event {
+                pattern: ".*".to_string(),
+                channel: None,
+            },
+        );
+
+        // Different user must not match
+        let msg_bob = make_message("bob", "telegram", "hello");
+        assert!(!super::routine_matches_message(&routine, &msg_bob));
+
+        // Same user must match
+        let msg_alice = make_message("alice", "telegram", "hello");
+        assert!(super::routine_matches_message(&routine, &msg_alice));
+    }
+
+    /// When no channel filter is set, any channel should match (given user matches).
+    #[test]
+    fn test_no_channel_filter_matches_any_channel() {
+        let routine = make_routine(
+            "user1",
+            Trigger::Event {
+                pattern: ".*".to_string(),
+                channel: None,
+            },
+        );
+
+        let msg = make_message("user1", "whatever_channel", "hello");
+        assert!(super::routine_matches_message(&routine, &msg));
+    }
+
     #[test]
     fn test_routine_tool_denylist_blocks_self_management_tools() {
         let denylisted = vec![

src/agent/session_manager.rs

@@ -102,11 +102,30 @@ impl SessionManager {
     /// Resolve an external thread ID to an internal thread.
     ///
     /// Returns the session and thread ID. Creates both if they don't exist.
+    /// Delegates to [`resolve_thread_with_parsed_uuid`](Self::resolve_thread_with_parsed_uuid)
+    /// with `parsed_uuid: None`.
     pub async fn resolve_thread(
         &self,
         user_id: &str,
         channel: &str,
         external_thread_id: Option<&str>,
+    ) -> (Arc<Mutex<Session>>, Uuid) {
+        self.resolve_thread_with_parsed_uuid(user_id, channel, external_thread_id, None)
+            .await
+    }
+
+    /// Like [`resolve_thread`](Self::resolve_thread), but accepts a pre-parsed
+    /// UUID to skip redundant parsing when the caller has already validated
+    /// the external thread ID as a UUID (e.g. the approval routing path).
+    ///
+    /// 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_with_parsed_uuid(
+        &self,
+        user_id: &str,
+        channel: &str,
+        external_thread_id: Option<&str>,
+        parsed_uuid: Option<Uuid>,
     ) -> (Arc<Mutex<Session>>, Uuid) {
         let session = self.get_or_create_session(user_id).await;
 
@@ -116,51 +135,65 @@ impl SessionManager {
             external_thread_id: external_thread_id.map(String::from),
         };
 
-        // Check if we have a mapping
-        {
+        // Use pre-parsed UUID if available, otherwise parse from string.
+        let ext_uuid = parsed_uuid
+            .or_else(|| external_thread_id.and_then(|ext_tid| Uuid::parse_str(ext_tid).ok()));
+
+        // Validate that parsed_uuid (if provided) is consistent with external_thread_id.
+        #[cfg(debug_assertions)]
+        if let (Some(parsed), Some(ext_tid)) = (&parsed_uuid, external_thread_id) {
+            debug_assert_eq!(
+                Uuid::parse_str(ext_tid).ok().as_ref(),
+                Some(parsed),
+                "parsed_uuid must be the parsed form of external_thread_id"
+            );
+        }
+
+        // Single read lock for both the key lookup and UUID adoption check
+        let adoptable_uuid = {
             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);
                 }
             }
-        }
 
-        // 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);
+            // 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.
+            // Only attempt adoption when external_thread_id is Some, preserving
+            // the invariant that None external_thread_id never triggers adoption.
+            if external_thread_id.is_some() {
+                ext_uuid.filter(|&uuid| !thread_map.values().any(|&v| v == uuid))
+            } else {
+                None
+            }
+        }; // Single read lock dropped here
 
-            if !mapped_elsewhere {
-                let sess = session.lock().await;
-                if sess.threads.contains_key(&ext_uuid) {
-                    drop(sess);
+        // 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 it was mapped elsewhere while we were unlocked, fall through
-                    // to create a new thread, preserving channel isolation.
+                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
             }
         }
 
@@ -909,6 +942,44 @@ 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};
@@ -947,4 +1018,88 @@ mod tests {
             "should have exactly 1 thread, not a duplicate"
         );
     }
+
+    #[tokio::test]
+    async fn test_resolve_thread_with_pre_parsed_uuid_adopts_thread() {
+        use crate::agent::session::Thread;
+
+        let manager = SessionManager::new();
+        let (session, _) = manager.resolve_thread("user1", "chan1", None).await;
+
+        // Manually insert a thread with a known UUID
+        let known_id = Uuid::new_v4();
+        {
+            let mut sess = session.lock().await;
+            let thread = Thread::with_id(known_id, sess.id);
+            sess.threads.insert(known_id, thread);
+        }
+
+        // Resolve with pre-parsed UUID -- should adopt it without re-parsing
+        let (_, resolved) = manager
+            .resolve_thread_with_parsed_uuid(
+                "user1",
+                "chan1",
+                Some(&known_id.to_string()),
+                Some(known_id),
+            )
+            .await;
+        assert_eq!(resolved, known_id);
+    }
+
+    #[tokio::test]
+    async fn test_resolve_thread_with_parsed_uuid_none_delegates_to_parse() {
+        use crate::agent::session::Thread;
+
+        let manager = SessionManager::new();
+        let (session, _) = manager.resolve_thread("user2", "chan2", None).await;
+
+        // Insert a thread with a known UUID
+        let known_id = Uuid::new_v4();
+        {
+            let mut sess = session.lock().await;
+            let thread = Thread::with_id(known_id, sess.id);
+            sess.threads.insert(known_id, thread);
+        }
+
+        // Resolve with parsed_uuid=None but a valid UUID string -- should
+        // fall back to parsing the string and still adopt the thread
+        let (_, resolved) = manager
+            .resolve_thread_with_parsed_uuid("user2", "chan2", Some(&known_id.to_string()), None)
+            .await;
+        assert_eq!(resolved, known_id);
+    }
+
+    #[tokio::test]
+    async fn test_resolve_thread_with_none_external_thread_id_does_not_adopt() {
+        use crate::agent::session::Thread;
+
+        let manager = SessionManager::new();
+        let (session, default_tid) = manager.resolve_thread("user3", "chan3", None).await;
+
+        // Manually insert a thread with a known UUID (simulating a thread
+        // created by chat_new_thread_handler)
+        let known_id = Uuid::new_v4();
+        {
+            let mut sess = session.lock().await;
+            let thread = Thread::with_id(known_id, sess.id);
+            sess.threads.insert(known_id, thread);
+        }
+
+        // Resolve with external_thread_id=None but parsed_uuid=Some.
+        // This should NOT adopt the UUID — the old code prevented adoption
+        // when external_thread_id was None, and we preserve that invariant.
+        let (_, resolved) = manager
+            .resolve_thread_with_parsed_uuid("user3", "chan3", None, Some(known_id))
+            .await;
+
+        // Should return the existing default thread, not the injected UUID
+        assert_eq!(
+            resolved, default_tid,
+            "should return existing default thread when external_thread_id is None"
+        );
+        assert_ne!(
+            resolved, known_id,
+            "should NOT adopt UUID when external_thread_id is None"
+        );
+    }
 }