From 20202700dbef968297e24976ed45edaae10ce135 Mon Sep 17 00:00:00 2001 From: Henry Park Date: Wed, 18 Mar 2026 12:29:35 -0700 Subject: [PATCH 1/3] 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 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) --------- Co-authored-by: Claude Opus 4.6 (1M context) --- src/agent/agent_loop.rs | 64 ++++++++++++++++------------------ src/agent/routine_engine.rs | 16 ++++----- tests/e2e_routine_heartbeat.rs | 49 ++++++++++++++++++++------ 3 files changed, 77 insertions(+), 52 deletions(-) diff --git a/src/agent/agent_loop.rs b/src/agent/agent_loop.rs index 83d971ef..132ba4a1 100644 --- a/src/agent/agent_loop.rs +++ b/src/agent/agent_loop.rs @@ -161,9 +161,10 @@ pub struct Agent { pub(super) heartbeat_config: Option, pub(super) hygiene_config: Option, pub(super) routine_config: Option, - /// 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>>>, } 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>>>, ) { - self.routine_engine_slot = Some(slot); + self.routine_engine_slot = slot; + } + + async fn routine_engine(&self) -> Option> { + 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 } => { diff --git a/src/agent/routine_engine.rs b/src/agent/routine_engine.rs index bf044139..ec8ab851 100644 --- a/src/agent/routine_engine.rs +++ b/src/agent/routine_engine.rs @@ -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; } diff --git a/tests/e2e_routine_heartbeat.rs b/tests/e2e_routine_heartbeat.rs index 48fb1ef4..3388feb8 100644 --- a/tests/e2e_routine_heartbeat.rs +++ b/tests/e2e_routine_heartbeat.rs @@ -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" ); From 42ffefabe4003368e75e6470d48d40528b81d8ef Mon Sep 17 00:00:00 2001 From: Henry Park Date: Wed, 18 Mar 2026 12:29:44 -0700 Subject: [PATCH 2/3] fix: remove -x from coverage pytest to prevent suite-blocking failures (#1360) One flaky test (test_builtin_echo_tool timeout) was stopping the entire e2e coverage suite via -x, preventing 118+ remaining tests from running and generating coverage data. Tests are independent (each gets a fresh browser context via the function-scoped page fixture), so removing -x is safe. Co-authored-by: Claude Opus 4.6 (1M context) --- .github/workflows/coverage.yml | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/.github/workflows/coverage.yml b/.github/workflows/coverage.yml index e7371677..2f885b16 100644 --- a/.github/workflows/coverage.yml +++ b/.github/workflows/coverage.yml @@ -174,7 +174,7 @@ jobs: - name: Run E2E tests run: | - pytest tests/e2e/ -v -x --timeout=120 + pytest tests/e2e/ -v --timeout=120 env: RUST_LOG: ironclaw=info RUST_BACKTRACE: "1" From 6831bb4d7b2bf7bf841c07de098ec023ddb26a5c Mon Sep 17 00:00:00 2001 From: Henry Park Date: Wed, 18 Mar 2026 12:29:58 -0700 Subject: [PATCH 3/3] fix: full_job routine concurrency tracks linked job lifetime (#1372) * fix: add FullJobWatcher to track full_job lifecycle for concurrency (#1318) full_job routines previously bypassed max_concurrent and global concurrency limits because execute_full_job() returned RunStatus::Ok immediately after dispatch. This meant running_count was decremented and the routine_run row was finalized before the actual job completed. Introduce FullJobWatcher struct that polls store.get_job() every 5s until the linked job reaches a non-active state, then maps the final JobState to RunStatus. execute_full_job now creates and awaits the watcher, keeping both the DB-level running row and the in-memory running_count elevated for the full job duration. Co-Authored-By: Claude Opus 4.6 (1M context) * test: full_job concurrency regression tests (issue #1318) Add two integration tests verifying full_job routine concurrency: 1. full_job_max_concurrent_blocks_second_fire_while_first_active: Inserts a Running routine_run (simulating an in-flight full_job) and verifies fire_manual returns MaxConcurrent error for max_concurrent=1. 2. global_concurrency_counts_live_full_job_runs: Elevates running_count to simulate a live full_job holding the global slot, verifies check_cron_triggers skips due routines, then releases the slot and verifies the routine fires. Also makes running_count_for_test() unconditionally public so integration tests (separate crate) can access it. Co-Authored-By: Claude Opus 4.6 (1M context) * style: fmt and clippy fixes for full_job concurrency tests Co-Authored-By: Claude Opus 4.6 (1M context) * fix: address PR review feedback on FullJobWatcher - Add #[doc(hidden)] to running_count_for_test() to hide from public API - Derive MAX_POLLS from POLL_INTERVAL to keep constants coupled - Check job state before first sleep to finalize promptly for fast jobs - Update execute_full_job doc comment to reflect blocking behavior Co-Authored-By: Claude Opus 4.6 (1M context) --------- Co-authored-by: Claude Opus 4.6 (1M context) --- src/agent/routine_engine.rs | 106 +++++++++++++++-- tests/e2e_routine_heartbeat.rs | 206 +++++++++++++++++++++++++++++++++ 2 files changed, 305 insertions(+), 7 deletions(-) diff --git a/src/agent/routine_engine.rs b/src/agent/routine_engine.rs index ec8ab851..14360d85 100644 --- a/src/agent/routine_engine.rs +++ b/src/agent/routine_engine.rs @@ -88,6 +88,12 @@ impl RoutineEngine { } } + /// Expose the running count for integration tests. + #[doc(hidden)] + pub fn running_count_for_test(&self) -> &Arc { + &self.running_count + } + /// Refresh the in-memory event trigger cache from DB. pub async fn refresh_event_cache(&self) { match self.store.list_event_routines().await { @@ -508,6 +514,88 @@ impl RoutineEngine { } } +/// Watches a dispatched full_job until the linked scheduler job completes. +/// +/// Polls `store.get_job(job_id)` at a fixed interval until the job leaves +/// an active state (Pending/InProgress/Stuck). Maps the final `JobState` to +/// a `RunStatus` for the routine run. +struct FullJobWatcher { + store: Arc, + job_id: Uuid, + routine_name: String, +} + +impl FullJobWatcher { + /// Poll interval between DB checks. + const POLL_INTERVAL: Duration = Duration::from_secs(5); + /// Safety ceiling: 24 hours, derived from POLL_INTERVAL. + const MAX_POLLS: u32 = (24 * 60 * 60) / Self::POLL_INTERVAL.as_secs() as u32; + + fn new(store: Arc, job_id: Uuid, routine_name: String) -> Self { + Self { + store, + job_id, + routine_name, + } + } + + /// Block until the linked job finishes and return the mapped status + summary. + async fn wait_for_completion(&self) -> (RunStatus, Option) { + let mut polls = 0u32; + + let final_status = loop { + // Check job state before sleeping so we finalize promptly + // if the job is already done (e.g. fast-failing jobs). + match self.store.get_job(self.job_id).await { + Ok(Some(job_ctx)) => { + if !job_ctx.state.is_active() { + break Self::map_job_state(&job_ctx.state); + } + } + Ok(None) => { + tracing::warn!( + routine = %self.routine_name, + job_id = %self.job_id, + "full_job disappeared from DB while polling" + ); + break RunStatus::Failed; + } + Err(e) => { + tracing::error!( + routine = %self.routine_name, + job_id = %self.job_id, + "Error polling full_job state: {}", e + ); + break RunStatus::Failed; + } + } + + polls += 1; + if polls >= Self::MAX_POLLS { + tracing::error!( + routine = %self.routine_name, + job_id = %self.job_id, + "full_job timed out after 24 hours, treating as failed" + ); + break RunStatus::Failed; + } + + tokio::time::sleep(Self::POLL_INTERVAL).await; + }; + + let summary = format!("Job {} finished ({})", self.job_id, final_status); + (final_status, Some(summary)) + } + + fn map_job_state(state: &crate::context::JobState) -> RunStatus { + use crate::context::JobState; + match state { + JobState::Failed | JobState::Cancelled => RunStatus::Failed, + _ => RunStatus::Ok, // Completed / Submitted / Accepted + } + } +} + /// Shared context passed to the execution function. struct EngineContext { config: RoutineConfig, @@ -682,8 +770,10 @@ fn sanitize_routine_name(name: &str) -> String { /// /// Fire-and-forget: creates a job via `Scheduler::dispatch_job` (which handles /// creation, metadata, persistence, and scheduling), links the routine run to -/// the job, and returns immediately. The job runs independently via the -/// existing Worker/Scheduler with full tool access. +/// the job, then watches it via `FullJobWatcher` until it reaches a +/// non-active state (not Pending/InProgress/Stuck). Returns the final +/// `RunStatus` mapped from the job outcome. This keeps the routine run +/// active for the full job lifetime so concurrency guardrails apply. async fn execute_full_job( ctx: &EngineContext, routine: &Routine, @@ -738,13 +828,15 @@ async fn execute_full_job( routine = %routine.name, job_id = %job_id, max_iterations = max_iterations, - "Dispatched full job for routine" + "Dispatched full job for routine, watching for completion" ); - let summary = format!( - "Dispatched job {job_id} for full execution with tool access (max_iterations: {max_iterations})" - ); - Ok((RunStatus::Ok, Some(summary), None)) + // Watch the job until it finishes — keeps the routine run active + // so concurrency guardrails (running_count, routine_runs status) + // remain enforced for the full job lifetime. + let watcher = FullJobWatcher::new(ctx.store.clone(), job_id, routine.name.clone()); + let (status, summary) = watcher.wait_for_completion().await; + Ok((status, summary, None)) } /// Execute a lightweight routine with optional tool support. diff --git a/tests/e2e_routine_heartbeat.rs b/tests/e2e_routine_heartbeat.rs index 3388feb8..25432f3d 100644 --- a/tests/e2e_routine_heartbeat.rs +++ b/tests/e2e_routine_heartbeat.rs @@ -823,4 +823,210 @@ mod tests { "Deleted routine must not fire after cache refresh" ); } + + // ----------------------------------------------------------------------- + // Test: full_job per-routine concurrency blocks second fire (issue #1318) + // ----------------------------------------------------------------------- + + #[tokio::test] + async fn full_job_max_concurrent_blocks_second_fire_while_first_active() { + use ironclaw::agent::routine::{ + NotifyConfig, Routine, RoutineAction, RoutineGuardrails, RoutineRun, RunStatus, Trigger, + }; + use ironclaw::error::RoutineError; + + let (db, _tmp) = create_test_db().await; + let ws = create_workspace(&db); + + // Stub LLM — fire_manual will be rejected before any LLM call + let trace = LlmTrace::single_turn( + "stub", + "stub", + vec![TraceStep { + request_hint: None, + response: TraceResponse::Text { + content: "ROUTINE_OK".to_string(), + input_tokens: 10, + output_tokens: 5, + }, + expected_tool_results: vec![], + }], + ); + let llm = Arc::new(TraceLlm::from_trace(trace)); + let (notify_tx, _notify_rx) = tokio::sync::mpsc::channel(4); + let tools = Arc::new(ToolRegistry::new()); + let safety = Arc::new(SafetyLayer::new(&SafetyConfig { + max_output_length: 100_000, + injection_check_enabled: false, + })); + + let engine = Arc::new(RoutineEngine::new( + RoutineConfig::default(), + db.clone(), + llm, + ws, + notify_tx, + None, // no scheduler — rejected before dispatch + tools, + safety, + )); + + // Create a full_job routine with max_concurrent = 1 + let routine = Routine { + id: Uuid::new_v4(), + name: "concurrent-guard".to_string(), + description: "test max_concurrent for full_job".to_string(), + user_id: "default".to_string(), + enabled: true, + trigger: Trigger::Manual, + action: RoutineAction::FullJob { + title: "t".to_string(), + description: "d".to_string(), + max_iterations: 3, + tool_permissions: vec![], + }, + guardrails: RoutineGuardrails { + cooldown: Duration::from_secs(0), + max_concurrent: 1, + dedup_window: None, + }, + notify: NotifyConfig::default(), + last_run_at: None, + next_fire_at: None, + run_count: 0, + consecutive_failures: 0, + state: serde_json::json!({}), + created_at: Utc::now(), + updated_at: Utc::now(), + }; + db.create_routine(&routine).await.expect("create_routine"); + + // Simulate first full_job run still active: the fix keeps the + // routine_run in Running state while the linked job executes. + let active_run = RoutineRun { + id: Uuid::new_v4(), + routine_id: routine.id, + trigger_type: "cron".to_string(), + trigger_detail: None, + started_at: Utc::now(), + completed_at: None, + status: RunStatus::Running, + result_summary: None, + tokens_used: None, + job_id: None, + created_at: Utc::now(), + }; + db.create_routine_run(&active_run) + .await + .expect("create_routine_run"); + + // Attempt to fire the same routine again — must be rejected + let result = engine.fire_manual(routine.id, None).await; + assert!( + matches!(result, Err(RoutineError::MaxConcurrent { .. })), + "second fire while first full_job active must be rejected by max_concurrent=1, got: {:?}", + result + ); + } + + // ----------------------------------------------------------------------- + // Test: global running_count tracks live full_job runs (issue #1318) + // ----------------------------------------------------------------------- + + #[tokio::test] + async fn global_concurrency_counts_live_full_job_runs() { + use std::sync::atomic::Ordering; + + let (db, _tmp) = create_test_db().await; + let ws = create_workspace(&db); + + let trace = LlmTrace::single_turn( + "test-global-limit", + "check", + vec![TraceStep { + request_hint: None, + response: TraceResponse::Text { + content: "ROUTINE_OK".to_string(), + input_tokens: 50, + output_tokens: 5, + }, + expected_tool_results: vec![], + }], + ); + let llm = Arc::new(TraceLlm::from_trace(trace)); + let (notify_tx, _notify_rx) = tokio::sync::mpsc::channel(16); + let tools = Arc::new(ToolRegistry::new()); + let safety = Arc::new(SafetyLayer::new(&SafetyConfig { + max_output_length: 100_000, + injection_check_enabled: true, + })); + + // Configure global limit of 1 + let config = RoutineConfig { + max_concurrent_routines: 1, + ..RoutineConfig::default() + }; + + let engine = Arc::new(RoutineEngine::new( + config, + db.clone(), + llm, + ws, + notify_tx, + None, + tools, + safety, + )); + + // Insert a due cron routine + let mut routine = make_routine( + "global-limit-test", + Trigger::Cron { + schedule: "* * * * *".to_string(), + timezone: None, + }, + "Check status.", + ); + routine.next_fire_at = Some(Utc::now() - chrono::Duration::minutes(1)); + db.create_routine(&routine).await.expect("create_routine"); + + // Simulate one full_job from another routine holding the global slot. + // With the fix, running_count stays elevated for the full job duration. + engine + .running_count_for_test() + .fetch_add(1, Ordering::Relaxed); + + // check_cron_triggers should see global limit hit and skip + engine.check_cron_triggers().await; + tokio::time::sleep(Duration::from_millis(100)).await; + + let runs = db + .list_routine_runs(routine.id, 10) + .await + .expect("list_routine_runs"); + assert!( + runs.is_empty(), + "cron routine must not fire when global limit is reached by live full_job" + ); + + // Release the global slot + engine + .running_count_for_test() + .fetch_sub(1, Ordering::Relaxed); + + // Now the routine should fire + engine.check_cron_triggers().await; + tokio::time::sleep(Duration::from_millis(200)).await; + + // Because the first check skipped it, next_fire_at is unchanged — + // the second check should see it as still due and fire it. + let runs_after = db + .list_routine_runs(routine.id, 10) + .await + .expect("list_routine_runs"); + assert!( + !runs_after.is_empty(), + "cron routine should fire after global slot is released" + ); + } }