mirror of
https://github.com/outbackdingo/optimclaw.git
synced 2026-08-25 14:53:34 +00:00
fix(agent): case-insensitive channel match and user_id filter for event triggers (#1211)
* fix(agent): case-insensitive channel match and user_id filter for event triggers (#1051, #1076) Event-triggered routines had two bugs preventing them from firing: 1. Channel comparison was case-sensitive (e.g., "Telegram" != "telegram"), while emit_system_event already used eq_ignore_ascii_case. Fixed to match. 2. No user_id scoping — routines from any user were evaluated against every message. Added ownership check so routines only fire for their owner's messages. Also adds periodic event cache refresh (every ~60s) in the cron ticker so web/CLI mutations are picked up without requiring the tool path. Upgrades skip-reason logging from trace to debug for debuggability. Closes #1051 Refs #1076 Co-Authored-By: Claude Opus 4.6 (1M context) <[email protected]> * fix: correct refresh_every from 6 to 4 to match 15s default interval The default cron_check_interval_secs is 15s, not 10s. With refresh_every=6, the cache would refresh every 90s instead of the intended ~60s. Fix to 4 ticks (4 * 15s = 60s). Co-Authored-By: Claude Opus 4.6 (1M context) <[email protected]> * fix(agent): address #1211 review -- extract routine_matches_message, fix refresh interval Extract user/channel filter logic from check_event_triggers into a standalone pure function routine_matches_message(). Rewrite tests to call this function directly with controlled Routine and IncomingMessage values, so they exercise the real code path and would catch a revert. Add test_no_channel_filter_matches_any_channel for the None channel case. Co-Authored-By: Claude Opus 4.6 (1M context) <[email protected]> * ci: re-trigger CI with latest changes Co-Authored-By: Claude Opus 4.6 <[email protected]> * fix: add missing IncomingMessage fields in test helper Co-Authored-By: Claude Opus 4.6 <[email protected]> * fix(agent): address review -- time-based refresh, trace-level user mismatch, scope guard (#1211) - Use tokio::time::Instant for cache refresh instead of tick counting - Downgrade user-mismatch log to trace to reduce noise - Add early return false for non-Event triggers in routine_matches_message - Fix doc comment to say 'user scope' instead of 'message sender' Co-Authored-By: Claude Opus 4.6 (1M context) <[email protected]> * style: run cargo fmt on agent_loop.rs https://claude.ai/code/session_01ABGWibdKVQ3b6pEKtxPPkM * fix(agent): resolve clippy warnings for unused binding and needless borrow Fix unused `content` variable in event trigger guard (use `content: _`) and remove redundant `&` on `message` which was already a reference. https://claude.ai/code/session_01PzBK21BbUAuZbrfLpoz4Xb * fix(test): update check_event_triggers call sites to new single-arg signature The staging merge brought e2e_routine_heartbeat tests that still used the old 3-argument check_event_triggers(user_id, channel, content) signature. Updated all 11 call sites to pass &IncomingMessage directly. [skip-regression-check] https://claude.ai/code/session_012GrkTDrtDFkpJos2hkgTcE * fix(agent): address review feedback on event trigger handling - Use post-hook content for event trigger matching so BeforeInbound hooks that rewrite input are respected - Set MissedTickBehavior::Skip on cron ticker to avoid burst catch-up after delays Co-Authored-By: Claude Opus 4.6 (1M context) <[email protected]> * style: cargo fmt https://claude.ai/code/session_01Va9wwvATNWFAx35GG7Zek7 --------- Co-authored-by: Claude Opus 4.6 (1M context) <[email protected]> Co-authored-by: firat.sertgoz <[email protected]>
This commit is contained in:
Zaki Manian
co-authored by
Claude Opus 4.6
firat.sertgoz
Claude Opus 4.6
firat.sertgoz
parent
01678be61d
commit
d3d517fd67
@@ -1139,9 +1139,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,
|
||||
|
||||
+186
-19
@@ -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;
|
||||
}
|
||||
|
||||
@@ -1781,6 +1820,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;
|
||||
@@ -1788,7 +1834,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();
|
||||
}
|
||||
}
|
||||
})
|
||||
}
|
||||
@@ -1854,7 +1904,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]
|
||||
@@ -2052,6 +2108,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![
|
||||
|
||||
@@ -561,11 +561,7 @@ mod tests {
|
||||
"deploy to production now",
|
||||
);
|
||||
let fired = engine
|
||||
.check_event_triggers(
|
||||
&matching_msg.user_id,
|
||||
&matching_msg.channel,
|
||||
&matching_msg.content,
|
||||
)
|
||||
.check_event_triggers(&matching_msg, &matching_msg.content)
|
||||
.await;
|
||||
assert!(
|
||||
fired >= 1,
|
||||
@@ -584,11 +580,7 @@ mod tests {
|
||||
"check the staging environment",
|
||||
);
|
||||
let fired_neg = engine
|
||||
.check_event_triggers(
|
||||
&non_matching_msg.user_id,
|
||||
&non_matching_msg.channel,
|
||||
&non_matching_msg.content,
|
||||
)
|
||||
.check_event_triggers(&non_matching_msg, &non_matching_msg.content)
|
||||
.await;
|
||||
assert_eq!(fired_neg, 0, "Expected 0 routines fired on non-match");
|
||||
}
|
||||
@@ -652,7 +644,7 @@ mod tests {
|
||||
"deploy to production now",
|
||||
);
|
||||
let guest_fired = engine
|
||||
.check_event_triggers(&guest_msg.user_id, &guest_msg.channel, &guest_msg.content)
|
||||
.check_event_triggers(&guest_msg, &guest_msg.content)
|
||||
.await;
|
||||
assert_eq!(
|
||||
guest_fired, 0,
|
||||
@@ -677,7 +669,7 @@ mod tests {
|
||||
"deploy to production now",
|
||||
);
|
||||
let owner_fired = engine
|
||||
.check_event_triggers(&owner_msg.user_id, &owner_msg.channel, &owner_msg.content)
|
||||
.check_event_triggers(&owner_msg, &owner_msg.content)
|
||||
.await;
|
||||
assert!(
|
||||
owner_fired >= 1,
|
||||
@@ -906,9 +898,7 @@ mod tests {
|
||||
"default",
|
||||
"test-cooldown trigger",
|
||||
);
|
||||
let fired1 = engine
|
||||
.check_event_triggers(&msg.user_id, &msg.channel, &msg.content)
|
||||
.await;
|
||||
let fired1 = engine.check_event_triggers(&msg, &msg.content).await;
|
||||
assert!(fired1 >= 1, "First fire should work");
|
||||
|
||||
// Give spawn time, then update last_run_at to simulate recent execution.
|
||||
@@ -923,9 +913,7 @@ mod tests {
|
||||
engine.refresh_event_cache().await;
|
||||
|
||||
// Second fire should be blocked by cooldown.
|
||||
let fired2 = engine
|
||||
.check_event_triggers(&msg.user_id, &msg.channel, &msg.content)
|
||||
.await;
|
||||
let fired2 = engine.check_event_triggers(&msg, &msg.content).await;
|
||||
assert_eq!(fired2, 0, "Second fire should be blocked by cooldown");
|
||||
}
|
||||
|
||||
@@ -1095,9 +1083,7 @@ mod tests {
|
||||
engine.refresh_event_cache().await;
|
||||
|
||||
let msg = IncomingMessage::new("test", "default", "DISABLE_ME");
|
||||
let fired_before = engine
|
||||
.check_event_triggers(&msg.user_id, &msg.channel, &msg.content)
|
||||
.await;
|
||||
let fired_before = engine.check_event_triggers(&msg, &msg.content).await;
|
||||
assert!(fired_before >= 1, "Expected routine to fire before disable");
|
||||
|
||||
// Simulate what routines_toggle_handler now does: update DB, then refresh.
|
||||
@@ -1106,9 +1092,7 @@ mod tests {
|
||||
db.update_routine(&routine).await.expect("update_routine");
|
||||
engine.refresh_event_cache().await;
|
||||
|
||||
let fired_after = engine
|
||||
.check_event_triggers(&msg.user_id, &msg.channel, &msg.content)
|
||||
.await;
|
||||
let fired_after = engine.check_event_triggers(&msg, &msg.content).await;
|
||||
assert_eq!(
|
||||
fired_after, 0,
|
||||
"Disabled routine must not fire after cache refresh"
|
||||
@@ -1134,10 +1118,7 @@ mod tests {
|
||||
|
||||
let msg = IncomingMessage::new("test", "default", "DELETE_ME");
|
||||
assert!(
|
||||
engine
|
||||
.check_event_triggers(&msg.user_id, &msg.channel, &msg.content)
|
||||
.await
|
||||
>= 1,
|
||||
engine.check_event_triggers(&msg, &msg.content).await >= 1,
|
||||
"Expected routine to fire before delete"
|
||||
);
|
||||
|
||||
@@ -1146,9 +1127,7 @@ mod tests {
|
||||
engine.refresh_event_cache().await;
|
||||
|
||||
assert_eq!(
|
||||
engine
|
||||
.check_event_triggers(&msg.user_id, &msg.channel, &msg.content)
|
||||
.await,
|
||||
engine.check_event_triggers(&msg, &msg.content).await,
|
||||
0,
|
||||
"Deleted routine must not fire after cache refresh"
|
||||
);
|
||||
@@ -1462,8 +1441,9 @@ mod tests {
|
||||
db.create_routine(&routine).await.expect("create_routine");
|
||||
engine.refresh_event_cache().await;
|
||||
|
||||
let trigger_msg = IncomingMessage::new("test", "default", "owner-gate");
|
||||
let fired = engine
|
||||
.check_event_triggers("default", "test", "owner-gate")
|
||||
.check_event_triggers(&trigger_msg, &trigger_msg.content)
|
||||
.await;
|
||||
assert_eq!(fired, 1, "expected one matching event routine");
|
||||
|
||||
|
||||
Reference in New Issue
Block a user