Merge pull request #1379 from nearai/staging-promote/6831bb4d-23264725970

chore: promote staging to staging-promote/f2cd1d37-23262791325 (2026-03-18 20:09 UTC)

.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"

src/agent/agent_loop.rs

@@ -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 } => {

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;
@@ -88,6 +88,12 @@ impl RoutineEngine {
         }
     }
 
+    /// Expose the running count for integration tests.
+    #[doc(hidden)]
+    pub fn running_count_for_test(&self) -> &Arc<AtomicUsize> {
+        &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 {
@@ -135,9 +141,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 +179,7 @@ impl RoutineEngine {
                 EventMatcher::System { .. } => continue,
             };
 
-            if routine.user_id != message.user_id {
+            if routine.user_id != user_id {
                 continue;
             }
 
@@ -181,13 +187,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 +216,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;
         }
@@ -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<dyn Database>,
+    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<dyn Database>, 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<String>) {
+        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.

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,9 +816,217 @@ 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"
         );
     }
+
+    // -----------------------------------------------------------------------
+    // 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"
+        );
+    }
 }