mirror of
https://github.com/outbackdingo/optimclaw.git
synced 2026-08-25 14:53:34 +00:00
Fix duplicate LLM responses for matched event routines (#1275)
* fix: consume matched event routine messages * style: run rustfmt for event routine fix * fix: preserve preprocessing for routine-triggered messages * fix: match routines against rewritten input * refactor: narrow check_event_triggers API and simplify routine_engine_slot Address Copilot review feedback: - Change check_event_triggers to accept (user_id, channel, content) instead of &IncomingMessage, eliminating the need to clone the full message (including attachments) when hooks rewrite content. - Remove routine_trigger_message and the Cow<IncomingMessage> indirection; the event-trigger check now inlines the is_internal + UserInput guard and passes the post-hook content string directly. - Make routine_engine_slot non-optional since Agent::new() always initializes it. Removes the redundant Option wrapper and simplifies accessor/setter methods. Co-Authored-By: Claude Opus 4.6 (1M context) <[email protected]> --------- Co-authored-by: Claude Opus 4.6 (1M context) <[email protected]>
This commit is contained in:
Henry Park
co-authored by
Claude Opus 4.6
Claude Opus 4.6
parent
f2cd1d37bc
commit
20202700db
+30
-34
@@ -161,9 +161,10 @@ pub struct Agent {
|
||||
pub(super) heartbeat_config: Option<HeartbeatConfig>,
|
||||
pub(super) hygiene_config: Option<crate::config::HygieneConfig>,
|
||||
pub(super) routine_config: Option<RoutineConfig>,
|
||||
/// Optional slot to expose the routine engine to the gateway for manual triggering.
|
||||
/// Shared routine-engine slot used for internal event matching and for exposing
|
||||
/// the engine to gateway/manual trigger entry points.
|
||||
pub(super) routine_engine_slot:
|
||||
Option<Arc<tokio::sync::RwLock<Option<Arc<crate::agent::routine_engine::RoutineEngine>>>>>,
|
||||
Arc<tokio::sync::RwLock<Option<Arc<crate::agent::routine_engine::RoutineEngine>>>>,
|
||||
}
|
||||
|
||||
impl Agent {
|
||||
@@ -228,16 +229,21 @@ impl Agent {
|
||||
heartbeat_config,
|
||||
hygiene_config,
|
||||
routine_config,
|
||||
routine_engine_slot: None,
|
||||
routine_engine_slot: Arc::new(tokio::sync::RwLock::new(None)),
|
||||
}
|
||||
}
|
||||
|
||||
/// Set the routine engine slot for exposing the engine to the gateway.
|
||||
/// Replace the routine-engine slot with a shared one so the gateway and
|
||||
/// agent reference the same engine.
|
||||
pub fn set_routine_engine_slot(
|
||||
&mut self,
|
||||
slot: Arc<tokio::sync::RwLock<Option<Arc<crate::agent::routine_engine::RoutineEngine>>>>,
|
||||
) {
|
||||
self.routine_engine_slot = Some(slot);
|
||||
self.routine_engine_slot = slot;
|
||||
}
|
||||
|
||||
async fn routine_engine(&self) -> Option<Arc<crate::agent::routine_engine::RoutineEngine>> {
|
||||
self.routine_engine_slot.read().await.clone()
|
||||
}
|
||||
|
||||
// Convenience accessors
|
||||
@@ -633,9 +639,7 @@ impl Agent {
|
||||
// via a local to use in the message loop below.
|
||||
|
||||
// Expose engine to gateway for manual triggering
|
||||
if let Some(ref slot) = self.routine_engine_slot {
|
||||
*slot.write().await = Some(Arc::clone(&engine));
|
||||
}
|
||||
*self.routine_engine_slot.write().await = Some(Arc::clone(&engine));
|
||||
|
||||
tracing::debug!(
|
||||
"Routines enabled: cron ticker every {}s, max {} concurrent",
|
||||
@@ -655,9 +659,6 @@ impl Agent {
|
||||
None
|
||||
};
|
||||
|
||||
// Extract engine ref for use in message loop
|
||||
let routine_engine_for_loop = routine_handle.as_ref().map(|(_, e)| Arc::clone(e));
|
||||
|
||||
// Main message loop
|
||||
tracing::debug!("Agent {} ready and listening", self.config.name);
|
||||
|
||||
@@ -693,29 +694,6 @@ impl Agent {
|
||||
// Store successfully extracted document text in workspace for indexing
|
||||
self.store_extracted_documents(&message).await;
|
||||
|
||||
// Event-triggered routines consume plain user input before it enters
|
||||
// the normal chat/tool pipeline. This avoids a duplicate turn where
|
||||
// the main agent responds and the routine also fires on the same
|
||||
// inbound message.
|
||||
if !message.is_internal
|
||||
&& matches!(
|
||||
SubmissionParser::parse(&message.content),
|
||||
Submission::UserInput { .. }
|
||||
)
|
||||
&& let Some(ref engine) = routine_engine_for_loop
|
||||
{
|
||||
let fired = engine.check_event_triggers(&message).await;
|
||||
if fired > 0 {
|
||||
tracing::debug!(
|
||||
channel = %message.channel,
|
||||
user = %message.user_id,
|
||||
fired,
|
||||
"Consumed inbound user message with matching event-triggered routine(s)"
|
||||
);
|
||||
continue;
|
||||
}
|
||||
}
|
||||
|
||||
match self.handle_message(&message).await {
|
||||
Ok(Some(response)) if !response.is_empty() => {
|
||||
// Hook: BeforeOutbound — allow hooks to modify or suppress outbound
|
||||
@@ -1032,6 +1010,24 @@ impl Agent {
|
||||
message.content.len()
|
||||
);
|
||||
|
||||
if !message.is_internal
|
||||
&& 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;
|
||||
if fired > 0 {
|
||||
tracing::debug!(
|
||||
channel = %message.channel,
|
||||
user = %message.user_id,
|
||||
fired,
|
||||
"Consumed inbound user message with matching event-triggered routine(s)"
|
||||
);
|
||||
return Ok(Some(String::new()));
|
||||
}
|
||||
}
|
||||
|
||||
// Process based on submission type
|
||||
let result = match submission {
|
||||
Submission::UserInput { content } => {
|
||||
|
||||
@@ -23,7 +23,7 @@ use crate::agent::Scheduler;
|
||||
use crate::agent::routine::{
|
||||
NotifyConfig, Routine, RoutineAction, RoutineRun, RunStatus, Trigger, next_cron_fire,
|
||||
};
|
||||
use crate::channels::{IncomingMessage, OutgoingResponse};
|
||||
use crate::channels::OutgoingResponse;
|
||||
use crate::config::RoutineConfig;
|
||||
use crate::context::JobContext;
|
||||
use crate::db::Database;
|
||||
@@ -135,9 +135,9 @@ impl RoutineEngine {
|
||||
|
||||
/// Check incoming message against event triggers. Returns number of routines fired.
|
||||
///
|
||||
/// Called synchronously from the main loop after handle_message(). The actual
|
||||
/// execution is spawned async so this returns quickly.
|
||||
pub async fn check_event_triggers(&self, message: &IncomingMessage) -> usize {
|
||||
/// 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 {
|
||||
let cache = self.event_cache.read().await;
|
||||
let mut fired = 0;
|
||||
|
||||
@@ -173,7 +173,7 @@ impl RoutineEngine {
|
||||
EventMatcher::System { .. } => continue,
|
||||
};
|
||||
|
||||
if routine.user_id != message.user_id {
|
||||
if routine.user_id != user_id {
|
||||
continue;
|
||||
}
|
||||
|
||||
@@ -181,13 +181,13 @@ impl RoutineEngine {
|
||||
if let Trigger::Event {
|
||||
channel: Some(ch), ..
|
||||
} = &routine.trigger
|
||||
&& ch != &message.channel
|
||||
&& ch != channel
|
||||
{
|
||||
continue;
|
||||
}
|
||||
|
||||
// Regex match
|
||||
if !re.is_match(&message.content) {
|
||||
if !re.is_match(content) {
|
||||
continue;
|
||||
}
|
||||
|
||||
@@ -210,7 +210,7 @@ impl RoutineEngine {
|
||||
continue;
|
||||
}
|
||||
|
||||
let detail = truncate(&message.content, 200);
|
||||
let detail = truncate(content, 200);
|
||||
self.spawn_fire(routine.clone(), "event", Some(detail));
|
||||
fired += 1;
|
||||
}
|
||||
|
||||
@@ -238,7 +238,13 @@ mod tests {
|
||||
"default",
|
||||
"deploy to production now",
|
||||
);
|
||||
let fired = engine.check_event_triggers(&matching_msg).await;
|
||||
let fired = engine
|
||||
.check_event_triggers(
|
||||
&matching_msg.user_id,
|
||||
&matching_msg.channel,
|
||||
&matching_msg.content,
|
||||
)
|
||||
.await;
|
||||
assert!(
|
||||
fired >= 1,
|
||||
"Expected >= 1 routine fired on match, got {fired}"
|
||||
@@ -255,7 +261,13 @@ mod tests {
|
||||
"default",
|
||||
"check the staging environment",
|
||||
);
|
||||
let fired_neg = engine.check_event_triggers(&non_matching_msg).await;
|
||||
let fired_neg = engine
|
||||
.check_event_triggers(
|
||||
&non_matching_msg.user_id,
|
||||
&non_matching_msg.channel,
|
||||
&non_matching_msg.content,
|
||||
)
|
||||
.await;
|
||||
assert_eq!(fired_neg, 0, "Expected 0 routines fired on non-match");
|
||||
}
|
||||
|
||||
@@ -315,7 +327,9 @@ mod tests {
|
||||
"guest-sender",
|
||||
"deploy to production now",
|
||||
);
|
||||
let guest_fired = engine.check_event_triggers(&guest_msg).await;
|
||||
let guest_fired = engine
|
||||
.check_event_triggers(&guest_msg.user_id, &guest_msg.channel, &guest_msg.content)
|
||||
.await;
|
||||
assert_eq!(
|
||||
guest_fired, 0,
|
||||
"Guest scope must not fire owner event routines"
|
||||
@@ -338,7 +352,9 @@ mod tests {
|
||||
"owner-sender",
|
||||
"deploy to production now",
|
||||
);
|
||||
let owner_fired = engine.check_event_triggers(&owner_msg).await;
|
||||
let owner_fired = engine
|
||||
.check_event_triggers(&owner_msg.user_id, &owner_msg.channel, &owner_msg.content)
|
||||
.await;
|
||||
assert!(
|
||||
owner_fired >= 1,
|
||||
"Owner scope should fire matching owner event routine"
|
||||
@@ -562,7 +578,9 @@ mod tests {
|
||||
"default",
|
||||
"test-cooldown trigger",
|
||||
);
|
||||
let fired1 = engine.check_event_triggers(&msg).await;
|
||||
let fired1 = engine
|
||||
.check_event_triggers(&msg.user_id, &msg.channel, &msg.content)
|
||||
.await;
|
||||
assert!(fired1 >= 1, "First fire should work");
|
||||
|
||||
// Give spawn time, then update last_run_at to simulate recent execution.
|
||||
@@ -577,7 +595,9 @@ mod tests {
|
||||
engine.refresh_event_cache().await;
|
||||
|
||||
// Second fire should be blocked by cooldown.
|
||||
let fired2 = engine.check_event_triggers(&msg).await;
|
||||
let fired2 = engine
|
||||
.check_event_triggers(&msg.user_id, &msg.channel, &msg.content)
|
||||
.await;
|
||||
assert_eq!(fired2, 0, "Second fire should be blocked by cooldown");
|
||||
}
|
||||
|
||||
@@ -745,7 +765,9 @@ mod tests {
|
||||
engine.refresh_event_cache().await;
|
||||
|
||||
let msg = IncomingMessage::new("test", "default", "DISABLE_ME");
|
||||
let fired_before = engine.check_event_triggers(&msg).await;
|
||||
let fired_before = engine
|
||||
.check_event_triggers(&msg.user_id, &msg.channel, &msg.content)
|
||||
.await;
|
||||
assert!(fired_before >= 1, "Expected routine to fire before disable");
|
||||
|
||||
// Simulate what routines_toggle_handler now does: update DB, then refresh.
|
||||
@@ -754,7 +776,9 @@ mod tests {
|
||||
db.update_routine(&routine).await.expect("update_routine");
|
||||
engine.refresh_event_cache().await;
|
||||
|
||||
let fired_after = engine.check_event_triggers(&msg).await;
|
||||
let fired_after = engine
|
||||
.check_event_triggers(&msg.user_id, &msg.channel, &msg.content)
|
||||
.await;
|
||||
assert_eq!(
|
||||
fired_after, 0,
|
||||
"Disabled routine must not fire after cache refresh"
|
||||
@@ -780,7 +804,10 @@ mod tests {
|
||||
|
||||
let msg = IncomingMessage::new("test", "default", "DELETE_ME");
|
||||
assert!(
|
||||
engine.check_event_triggers(&msg).await >= 1,
|
||||
engine
|
||||
.check_event_triggers(&msg.user_id, &msg.channel, &msg.content)
|
||||
.await
|
||||
>= 1,
|
||||
"Expected routine to fire before delete"
|
||||
);
|
||||
|
||||
@@ -789,7 +816,9 @@ mod tests {
|
||||
engine.refresh_event_cache().await;
|
||||
|
||||
assert_eq!(
|
||||
engine.check_event_triggers(&msg).await,
|
||||
engine
|
||||
.check_event_triggers(&msg.user_id, &msg.channel, &msg.content)
|
||||
.await,
|
||||
0,
|
||||
"Deleted routine must not fire after cache refresh"
|
||||
);
|
||||
|
||||
Reference in New Issue
Block a user