From 994a0b194fd3b59db9daa3e3b75ade71940205bd Mon Sep 17 00:00:00 2001 From: Nick Pismenkov <50764773+nickpismenkov@users.noreply.github.com> Date: Sat, 14 Mar 2026 13:06:59 -0700 Subject: [PATCH] fix: N+1 query pattern in event trigger loop (routine_engine) (#1163) * fix: N+1 query pattern in event trigger loop (routine_engine) * fix: linter --- src/agent/routine_engine.rs | 64 ++- src/db/libsql/routines.rs | 57 +- src/db/mod.rs | 4 + src/db/postgres.rs | 9 + src/history/store.rs | 39 ++ tests/batch_query_tests.rs | 509 +++++++++++++++++ .../e2e/scenarios/test_routine_event_batch.py | 534 ++++++++++++++++++ 7 files changed, 1211 insertions(+), 5 deletions(-) create mode 100644 tests/batch_query_tests.rs create mode 100644 tests/e2e/scenarios/test_routine_event_batch.py diff --git a/src/agent/routine_engine.rs b/src/agent/routine_engine.rs index 53b5c883..739b20d7 100644 --- a/src/agent/routine_engine.rs +++ b/src/agent/routine_engine.rs @@ -139,6 +139,32 @@ impl RoutineEngine { let cache = self.event_cache.read().await; let mut fired = 0; + // Collect routine IDs for batch query + let routine_ids: Vec = cache + .iter() + .filter_map(|matcher| match matcher { + EventMatcher::Message { routine, .. } => Some(routine.id), + EventMatcher::System { .. } => None, + }) + .collect(); + + if routine_ids.is_empty() { + return 0; + } + + // Single batch query instead of N queries + let concurrent_counts = match self + .store + .count_running_routine_runs_batch(&routine_ids) + .await + { + Ok(counts) => counts, + Err(e) => { + tracing::error!("Failed to batch-load concurrent counts: {}", e); + return 0; + } + }; + for matcher in cache.iter() { let (routine, re) = match matcher { EventMatcher::Message { routine, regex } => (routine, regex), @@ -164,8 +190,9 @@ impl RoutineEngine { continue; } - // Concurrent run check - if !self.check_concurrent(routine).await { + // 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"); continue; } @@ -197,6 +224,35 @@ impl RoutineEngine { let cache = self.event_cache.read().await; let mut fired = 0; + // Collect routine IDs for batch query + let routine_ids: Vec = cache + .iter() + .filter_map(|matcher| match matcher { + EventMatcher::System { routine } => Some(routine.id), + EventMatcher::Message { .. } => None, + }) + .collect(); + + if routine_ids.is_empty() { + return 0; + } + + // Single batch query instead of N queries + let concurrent_counts = match self + .store + .count_running_routine_runs_batch(&routine_ids) + .await + { + Ok(counts) => counts, + Err(e) => { + tracing::error!( + "Failed to batch-load concurrent counts for system events: {}", + e + ); + return 0; + } + }; + for matcher in cache.iter() { let routine = match matcher { EventMatcher::System { routine } => routine, @@ -248,7 +304,9 @@ impl RoutineEngine { continue; } - if !self.check_concurrent(routine).await { + // 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::debug!(routine = %routine.name, "Skipped: max concurrent reached"); continue; } diff --git a/src/db/libsql/routines.rs b/src/db/libsql/routines.rs index 3f2629ea..dd9fd6c0 100644 --- a/src/db/libsql/routines.rs +++ b/src/db/libsql/routines.rs @@ -1,13 +1,15 @@ //! Routine-related RoutineStore implementation for LibSqlBackend. +use std::collections::{HashMap, HashSet}; + use async_trait::async_trait; use chrono::{DateTime, Utc}; use libsql::params; use uuid::Uuid; use super::{ - LibSqlBackend, ROUTINE_COLUMNS, ROUTINE_RUN_COLUMNS, fmt_opt_ts, fmt_ts, get_i64, opt_text, - opt_text_owned, row_to_routine_libsql, row_to_routine_run_libsql, + LibSqlBackend, ROUTINE_COLUMNS, ROUTINE_RUN_COLUMNS, fmt_opt_ts, fmt_ts, get_i64, get_text, + opt_text, opt_text_owned, row_to_routine_libsql, row_to_routine_run_libsql, }; use crate::agent::routine::{Routine, RoutineRun, RunStatus}; use crate::db::RoutineStore; @@ -409,6 +411,57 @@ impl RoutineStore for LibSqlBackend { } } + async fn count_running_routine_runs_batch( + &self, + routine_ids: &[Uuid], + ) -> Result, DatabaseError> { + if routine_ids.is_empty() { + return Ok(HashMap::new()); + } + + let mut counts = HashMap::new(); + let conn = self.connect().await?; + + // Query all running routines and filter in memory + // This is simpler for libSQL than building dynamic parameter lists + let mut rows = conn + .query( + "SELECT routine_id, COUNT(*) as cnt FROM routine_runs + WHERE status = 'running' + GROUP BY routine_id", + params![], + ) + .await + .map_err(|e| { + DatabaseError::Query(format!("Failed to batch count running routines: {}", e)) + })?; + + let routine_id_set: HashSet = routine_ids.iter().copied().collect(); + + while let Some(row) = rows + .next() + .await + .map_err(|e| DatabaseError::Query(e.to_string()))? + { + let id_str: String = get_text(&row, 0); + let id = Uuid::parse_str(&id_str) + .map_err(|e| DatabaseError::Query(format!("Invalid routine UUID: {}", e)))?; + + // Only include if this routine ID was requested + if routine_id_set.contains(&id) { + let cnt: i64 = get_i64(&row, 1); + counts.insert(id, cnt); + } + } + + // Ensure all requested IDs are in the map (defaults to 0 for no running runs) + for id in routine_ids { + counts.entry(*id).or_insert(0); + } + + Ok(counts) + } + async fn link_routine_run_to_job( &self, run_id: Uuid, diff --git a/src/db/mod.rs b/src/db/mod.rs index 4afd1db8..a306c14b 100644 --- a/src/db/mod.rs +++ b/src/db/mod.rs @@ -387,6 +387,10 @@ pub trait RoutineStore: Send + Sync { limit: i64, ) -> Result, DatabaseError>; async fn count_running_routine_runs(&self, routine_id: Uuid) -> Result; + async fn count_running_routine_runs_batch( + &self, + routine_ids: &[Uuid], + ) -> Result, DatabaseError>; async fn link_routine_run_to_job( &self, run_id: Uuid, diff --git a/src/db/postgres.rs b/src/db/postgres.rs index 2cf6a65a..8c18e252 100644 --- a/src/db/postgres.rs +++ b/src/db/postgres.rs @@ -487,6 +487,15 @@ impl RoutineStore for PgBackend { self.store.count_running_routine_runs(routine_id).await } + async fn count_running_routine_runs_batch( + &self, + routine_ids: &[Uuid], + ) -> Result, DatabaseError> { + self.store + .count_running_routine_runs_batch(routine_ids) + .await + } + async fn link_routine_run_to_job( &self, run_id: Uuid, diff --git a/src/history/store.rs b/src/history/store.rs index 83f60d70..17fa96fd 100644 --- a/src/history/store.rs +++ b/src/history/store.rs @@ -1,5 +1,8 @@ //! PostgreSQL store for persisting agent data. +#[cfg(feature = "postgres")] +use std::collections::HashMap; + use chrono::{DateTime, Utc}; #[cfg(feature = "postgres")] use deadpool_postgres::{Config, Pool}; @@ -1294,6 +1297,42 @@ impl Store { Ok(row.get("cnt")) } + /// Batch-load concurrent run counts for multiple routines in a single query. + /// Returns a map where missing routine IDs default to 0. + #[cfg(feature = "postgres")] + pub async fn count_running_routine_runs_batch( + &self, + routine_ids: &[Uuid], + ) -> Result, DatabaseError> { + if routine_ids.is_empty() { + return Ok(HashMap::new()); + } + + let conn = self.conn().await?; + let rows = conn + .query( + "SELECT routine_id, COUNT(*) as cnt FROM routine_runs + WHERE routine_id = ANY($1) AND status = 'running' + GROUP BY routine_id", + &[&routine_ids], + ) + .await?; + + let mut counts = HashMap::new(); + for row in rows { + let id: Uuid = row.get("routine_id"); + let cnt: i64 = row.get("cnt"); + counts.insert(id, cnt); + } + + // Ensure all requested IDs are in the map (defaults to 0 for no running runs) + for id in routine_ids { + counts.entry(*id).or_insert(0); + } + + Ok(counts) + } + /// Link a routine run to a dispatched job. pub async fn link_routine_run_to_job( &self, diff --git a/tests/batch_query_tests.rs b/tests/batch_query_tests.rs new file mode 100644 index 00000000..d7365287 --- /dev/null +++ b/tests/batch_query_tests.rs @@ -0,0 +1,509 @@ +//! Tests for batch loading routine concurrent counts (N+1 query fix). +//! +//! Verifies: +//! 1. Batch query returns correct counts for multiple routines +//! 2. Concurrent limit enforcement uses batch counts correctly + +#[cfg(feature = "libsql")] +mod tests { + use std::sync::Arc; + + use chrono::Utc; + use uuid::Uuid; + + use ironclaw::agent::routine::{ + Routine, RoutineAction, RoutineGuardrails, RoutineRun, RunStatus, Trigger, + }; + use ironclaw::db::Database; + + async fn create_test_db() -> (Arc, tempfile::TempDir) { + use ironclaw::db::libsql::LibSqlBackend; + + let temp_dir = tempfile::tempdir().expect("tempdir"); + let db_path = temp_dir.path().join("test.db"); + let backend = LibSqlBackend::new_local(&db_path) + .await + .expect("LibSqlBackend"); + backend.run_migrations().await.expect("migrations"); + let db: Arc = Arc::new(backend); + (db, temp_dir) + } + + // ----------------------------------------------------------------------- + // Test 1: Batch query returns correct counts for multiple routines + // ----------------------------------------------------------------------- + + #[tokio::test] + async fn batch_query_empty_list() { + let (db, _tmp) = create_test_db().await; + let counts = db + .count_running_routine_runs_batch(&[]) + .await + .expect("batch query should not fail"); + assert!(counts.is_empty(), "Empty input should return empty map"); + } + + #[tokio::test] + async fn batch_query_single_routine() { + let (db, _tmp) = create_test_db().await; + let routine_id = Uuid::new_v4(); + + // Create routine + let routine = Routine { + id: routine_id, + name: "test-routine".to_string(), + description: "Test".to_string(), + user_id: "default".to_string(), + enabled: true, + trigger: Trigger::Cron { + schedule: "* * * * *".to_string(), + timezone: None, + }, + action: RoutineAction::Lightweight { + prompt: "test".to_string(), + context_paths: vec![], + max_tokens: 1000, + use_tools: false, + max_tool_rounds: 3, + }, + guardrails: RoutineGuardrails { + cooldown: std::time::Duration::from_secs(0), + max_concurrent: 5, + dedup_window: None, + }, + notify: Default::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"); + + // Create 3 running runs + for _ in 0..3 { + let run = RoutineRun { + id: Uuid::new_v4(), + 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(&run).await.expect("create run"); + } + + // Batch query for single routine + let counts = db + .count_running_routine_runs_batch(&[routine_id]) + .await + .expect("batch query should work"); + + assert_eq!(counts.len(), 1, "Should return 1 routine"); + assert_eq!(counts[&routine_id], 3, "Should count 3 running runs"); + } + + #[tokio::test] + async fn batch_query_multiple_routines_different_counts() { + let (db, _tmp) = create_test_db().await; + + let r1 = Uuid::new_v4(); + let r2 = Uuid::new_v4(); + let r3 = Uuid::new_v4(); + + // Create 3 routines + for routine_id in [r1, r2, r3] { + let routine = Routine { + id: routine_id, + name: format!("routine-{}", routine_id), + description: "Test".to_string(), + user_id: "default".to_string(), + enabled: true, + trigger: Trigger::Cron { + schedule: "* * * * *".to_string(), + timezone: None, + }, + action: RoutineAction::Lightweight { + prompt: "test".to_string(), + context_paths: vec![], + max_tokens: 1000, + use_tools: false, + max_tool_rounds: 3, + }, + guardrails: RoutineGuardrails { + cooldown: std::time::Duration::from_secs(0), + max_concurrent: 5, + dedup_window: None, + }, + notify: Default::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"); + } + + // r1: 2 running + for _ in 0..2 { + let run = RoutineRun { + id: Uuid::new_v4(), + routine_id: r1, + 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(&run).await.expect("create run"); + } + + // r2: 1 running + let run = RoutineRun { + id: Uuid::new_v4(), + routine_id: r2, + 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(&run).await.expect("create run"); + + // r3: 0 running (but has 1 Ok result) + let run = RoutineRun { + id: Uuid::new_v4(), + routine_id: r3, + trigger_type: "cron".to_string(), + trigger_detail: None, + started_at: Utc::now(), + completed_at: Some(Utc::now()), + status: RunStatus::Ok, + result_summary: None, + tokens_used: None, + job_id: None, + created_at: Utc::now(), + }; + db.create_routine_run(&run).await.expect("create run"); + + // Single batch query for all 3 + let counts = db + .count_running_routine_runs_batch(&[r1, r2, r3]) + .await + .expect("batch query should work"); + + assert_eq!(counts.len(), 3, "Should return 3 routines"); + assert_eq!(counts[&r1], 2, "r1 should have 2 running"); + assert_eq!(counts[&r2], 1, "r2 should have 1 running"); + assert_eq!( + counts[&r3], 0, + "r3 should have 0 running (Ok status is not running)" + ); + } + + #[tokio::test] + async fn batch_query_missing_routines_default_to_zero() { + let (db, _tmp) = create_test_db().await; + + let r1 = Uuid::new_v4(); + let r2 = Uuid::new_v4(); + let r3 = Uuid::new_v4(); // This one won't exist + + // Only create r1 + let routine = Routine { + id: r1, + name: "routine-1".to_string(), + description: "Test".to_string(), + user_id: "default".to_string(), + enabled: true, + trigger: Trigger::Cron { + schedule: "* * * * *".to_string(), + timezone: None, + }, + action: RoutineAction::Lightweight { + prompt: "test".to_string(), + context_paths: vec![], + max_tokens: 1000, + use_tools: false, + max_tool_rounds: 3, + }, + guardrails: RoutineGuardrails { + cooldown: std::time::Duration::from_secs(0), + max_concurrent: 5, + dedup_window: None, + }, + notify: Default::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"); + + // r1 has 1 running + let run = RoutineRun { + id: Uuid::new_v4(), + routine_id: r1, + 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(&run).await.expect("create run"); + + // Query for r1, r2 (doesn't exist), r3 (doesn't exist) + let counts = db + .count_running_routine_runs_batch(&[r1, r2, r3]) + .await + .expect("batch query should work"); + + assert_eq!(counts.len(), 3, "Should have all 3 routine IDs"); + assert_eq!(counts[&r1], 1, "r1 should have 1 running"); + assert_eq!(counts[&r2], 0, "r2 should default to 0"); + assert_eq!(counts[&r3], 0, "r3 should default to 0"); + } + + #[tokio::test] + async fn batch_query_only_counts_running_status() { + let (db, _tmp) = create_test_db().await; + let routine_id = Uuid::new_v4(); + + // Create routine + let routine = Routine { + id: routine_id, + name: "test-routine".to_string(), + description: "Test".to_string(), + user_id: "default".to_string(), + enabled: true, + trigger: Trigger::Cron { + schedule: "* * * * *".to_string(), + timezone: None, + }, + action: RoutineAction::Lightweight { + prompt: "test".to_string(), + context_paths: vec![], + max_tokens: 1000, + use_tools: false, + max_tool_rounds: 3, + }, + guardrails: RoutineGuardrails { + cooldown: std::time::Duration::from_secs(0), + max_concurrent: 5, + dedup_window: None, + }, + notify: Default::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"); + + // Create 5 runs with mixed statuses + let statuses = [ + RunStatus::Running, + RunStatus::Running, + RunStatus::Ok, + RunStatus::Failed, + RunStatus::Attention, + ]; + + for status in statuses.iter() { + let run = RoutineRun { + id: Uuid::new_v4(), + routine_id, + trigger_type: "cron".to_string(), + trigger_detail: None, + started_at: Utc::now(), + completed_at: Some(Utc::now()), + status: *status, + result_summary: None, + tokens_used: None, + job_id: None, + created_at: Utc::now(), + }; + db.create_routine_run(&run).await.expect("create run"); + } + + // Batch query should only count Running status + let counts = db + .count_running_routine_runs_batch(&[routine_id]) + .await + .expect("batch query should work"); + + assert_eq!( + counts[&routine_id], 2, + "Should only count 2 Running status runs" + ); + } + + // ----------------------------------------------------------------------- + // Test 2: Concurrent limit enforcement uses batch counts + // ----------------------------------------------------------------------- + + #[tokio::test] + async fn concurrent_limit_enforcement_with_batch_counts() { + let (db, _tmp) = create_test_db().await; + + let r1 = Uuid::new_v4(); + let r2 = Uuid::new_v4(); + + // Create 2 routines with max_concurrent=1 (r1) and max_concurrent=2 (r2) + for (routine_id, max_concurrent) in [(r1, 1), (r2, 2)] { + let routine = Routine { + id: routine_id, + name: format!("routine-{}", routine_id), + description: "Test".to_string(), + user_id: "default".to_string(), + enabled: true, + trigger: Trigger::Cron { + schedule: "* * * * *".to_string(), + timezone: None, + }, + action: RoutineAction::Lightweight { + prompt: "test".to_string(), + context_paths: vec![], + max_tokens: 1000, + use_tools: false, + max_tool_rounds: 3, + }, + guardrails: RoutineGuardrails { + cooldown: std::time::Duration::from_secs(0), + max_concurrent, + dedup_window: None, + }, + notify: Default::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"); + } + + // r1: create 1 running run (will hit max_concurrent=1) + let run = RoutineRun { + id: Uuid::new_v4(), + routine_id: r1, + 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(&run).await.expect("create run"); + + // r2: create 2 running runs (will hit max_concurrent=2) + for _ in 0..2 { + let run = RoutineRun { + id: Uuid::new_v4(), + routine_id: r2, + 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(&run).await.expect("create run"); + } + + // Batch query should return correct counts + let counts = db + .count_running_routine_runs_batch(&[r1, r2]) + .await + .expect("batch query should work"); + + // Verify counts match the limits + assert_eq!( + counts[&r1], 1, + "r1 should have 1 running (at max_concurrent=1)" + ); + assert_eq!( + counts[&r2], 2, + "r2 should have 2 running (at max_concurrent=2)" + ); + + // Now verify the limit enforcement logic + let r1_routine = db + .get_routine(r1) + .await + .expect("get routine") + .expect("routine exists"); + let r2_routine = db + .get_routine(r2) + .await + .expect("get routine") + .expect("routine exists"); + + let r1_at_limit = counts[&r1] >= r1_routine.guardrails.max_concurrent as i64; + let r2_at_limit = counts[&r2] >= r2_routine.guardrails.max_concurrent as i64; + + assert!(r1_at_limit, "r1 should be detected as at limit"); + assert!(r2_at_limit, "r2 should be detected as at limit"); + + // If we add one more run to r2, it should exceed limit + let run = RoutineRun { + id: Uuid::new_v4(), + routine_id: r2, + 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(&run).await.expect("create run"); + + // Re-query to get updated counts + let counts = db + .count_running_routine_runs_batch(&[r1, r2]) + .await + .expect("batch query should work"); + + let r2_exceeded_limit = counts[&r2] > r2_routine.guardrails.max_concurrent as i64; + assert!(r2_exceeded_limit, "r2 should have exceeded its limit"); + } +} diff --git a/tests/e2e/scenarios/test_routine_event_batch.py b/tests/e2e/scenarios/test_routine_event_batch.py new file mode 100644 index 00000000..d8c59e6d --- /dev/null +++ b/tests/e2e/scenarios/test_routine_event_batch.py @@ -0,0 +1,534 @@ +""" +E2E tests for event-triggered routines with batch loading. + +These tests verify that the N+1 query fix correctly: +1. Fires event-triggered routines on matching messages +2. Enforces concurrent limits via batch-loaded counts +3. Maintains performance with multiple simultaneous triggers +4. Works correctly through the full UI and agent loop + +Playwright-based UI tests + SSE verification. +""" + +import asyncio +import json +import pytest +from datetime import datetime, timedelta +from typing import List, Dict, Any + +from playwright.async_api import async_playwright, Page, Browser, BrowserContext + + +@pytest.fixture +async def browser_and_context(): + """Create a Playwright browser and context for testing.""" + async with async_playwright() as p: + browser = await p.chromium.launch(headless=True) + context = await browser.new_context() + yield browser, context + await context.close() + await browser.close() + + +class EventTriggerHelper: + """Helper methods for event trigger testing.""" + + def __init__(self, page: Page): + self.page = page + + async def navigate_to_routines(self): + """Navigate to the routines page.""" + await self.page.goto("http://localhost:8000/routines") + await self.page.wait_for_load_state("networkidle") + + async def create_event_routine( + self, + name: str, + trigger_regex: str, + channel: str = "slack", + max_concurrent: int = 1, + ) -> str: + """ + Create an event-triggered routine via UI. + Returns the routine ID. + """ + await self.navigate_to_routines() + + # Click "New Routine" button + await self.page.click('button:has-text("New Routine")') + await self.page.wait_for_selector('input[name="routine_name"]') + + # Fill routine details + await self.page.fill('input[name="routine_name"]', name) + await self.page.fill( + 'textarea[name="routine_description"]', + f"Test routine: {name}", + ) + + # Select "Event Trigger" type + await self.page.click('label:has-text("Event Trigger")') + await self.page.wait_for_selector('input[name="trigger_regex"]') + + # Fill trigger details + await self.page.fill('input[name="trigger_regex"]', trigger_regex) + await self.page.select_option('select[name="trigger_channel"]', channel) + + # Set guardrails + await self.page.fill('input[name="max_concurrent"]', str(max_concurrent)) + + # Select lightweight action + await self.page.click('label:has-text("Lightweight")') + await self.page.fill( + 'textarea[name="lightweight_prompt"]', + "Acknowledge the message and confirm trigger worked.", + ) + + # Save routine + await self.page.click('button:has-text("Save Routine")') + await self.page.wait_for_selector('text=Routine created successfully') + + # Extract routine ID from success message or URL + routine_id = await self.page.locator('data-testid=routine-id').text_content() + return routine_id.strip() if routine_id else None + + async def create_multiple_routines( + self, base_name: str, count: int, trigger_regex: str = None + ) -> List[str]: + """Create multiple event-triggered routines.""" + routine_ids = [] + for i in range(count): + name = f"{base_name}_{i}" + regex = trigger_regex or f"({i}|{base_name})" + routine_id = await self.create_event_routine(name, regex) + routine_ids.append(routine_id) + await asyncio.sleep(0.1) # Small delay between creations + return routine_ids + + async def send_chat_message(self, message: str) -> List[str]: + """ + Send a chat message and return SSE events received. + Captures all routine firing events. + """ + await self.page.goto("http://localhost:8000/chat") + await self.page.wait_for_selector('input[placeholder*="message"]', timeout=5000) + + # Collect SSE events + sse_events = [] + + async def capture_sse(response): + """Intercept SSE events.""" + if "event-stream" in response.headers.get("content-type", ""): + text = await response.text() + for line in text.split("\n"): + if line.startswith("data:"): + try: + event = json.loads(line[5:]) + sse_events.append(event) + except json.JSONDecodeError: + pass + + self.page.on("response", capture_sse) + + # Send message + await self.page.fill('input[placeholder*="message"]', message) + await self.page.press('input[placeholder*="message"]', "Enter") + + # Wait for response + await self.page.wait_for_selector('text=Message processed', timeout=10000) + await asyncio.sleep(0.5) # Allow time for SSE events + + self.page.remove_listener("response", capture_sse) + return sse_events + + async def get_routine_execution_log(self, routine_id: str) -> List[Dict]: + """Get execution log entries for a routine.""" + await self.page.goto(f"http://localhost:8000/routines/{routine_id}/executions") + await self.page.wait_for_load_state("networkidle") + + # Extract log entries from table + rows = await self.page.locator("tbody tr").all() + executions = [] + + for row in rows: + cells = await row.locator("td").all() + if len(cells) >= 3: + execution = { + "timestamp": await cells[0].text_content(), + "status": await cells[1].text_content(), + "details": await cells[2].text_content(), + } + executions.append(execution) + + return executions + + async def check_database_queries_in_logs( + self, max_queries_expected: int = 1 + ) -> int: + """Check debug logs for database query count.""" + await self.page.goto("http://localhost:8000/debug/logs?filter=database") + await self.page.wait_for_load_state("networkidle") + + # Count batch queries + log_lines = await self.page.locator("tr:has-text('batch')").all() + batch_count = len(log_lines) + + # Count individual COUNT queries (should be 0 after fix) + count_queries = await self.page.locator("tr:has-text('COUNT')").all() + count_query_count = len(count_queries) + + return batch_count, count_query_count + + +# ============================================================================= +# Tests +# ============================================================================= + + +@pytest.mark.asyncio +async def test_create_event_trigger_routine(browser_and_context): + """Test creating an event-triggered routine via UI.""" + browser, context = browser_and_context + page = await context.new_page() + helper = EventTriggerHelper(page) + + try: + routine_id = await helper.create_event_routine( + name="Test Trigger", + trigger_regex="test|demo", + channel="slack", + max_concurrent=1, + ) + + assert routine_id is not None, "Routine ID should be returned" + assert len(routine_id) > 0, "Routine ID should not be empty" + + finally: + await page.close() + + +@pytest.mark.asyncio +async def test_event_trigger_fires_on_matching_message(browser_and_context): + """Test that event-triggered routine fires when message matches.""" + browser, context = browser_and_context + page = await context.new_page() + helper = EventTriggerHelper(page) + + try: + # Create routine + routine_id = await helper.create_event_routine( + name="Alert Handler", + trigger_regex="urgent|critical|alert", + channel="slack", + ) + + # Send matching message + sse_events = await helper.send_chat_message("URGENT: Server down!") + + # Verify routine fired (look for event in SSE stream) + routine_fired = any( + event.get("type") == "routine_fired" and event.get("routine_id") == routine_id + for event in sse_events + ) + assert routine_fired, "Routine should fire on matching message" + + # Check execution log + executions = await helper.get_routine_execution_log(routine_id) + assert len(executions) > 0, "Execution should be logged" + assert "success" in executions[0]["status"].lower() + + finally: + await page.close() + + +@pytest.mark.asyncio +async def test_event_trigger_skips_non_matching_message(browser_and_context): + """Test that event-triggered routine skips when message doesn't match.""" + browser, context = browser_and_context + page = await context.new_page() + helper = EventTriggerHelper(page) + + try: + # Create routine + routine_id = await helper.create_event_routine( + name="Alert Handler", + trigger_regex="urgent|critical|alert", + channel="slack", + ) + + # Send non-matching message + sse_events = await helper.send_chat_message("Hello, how are you?") + + # Verify routine did NOT fire + routine_fired = any( + event.get("type") == "routine_fired" and event.get("routine_id") == routine_id + for event in sse_events + ) + assert not routine_fired, "Routine should not fire on non-matching message" + + finally: + await page.close() + + +@pytest.mark.asyncio +async def test_multiple_routines_fire_on_matching_message(browser_and_context): + """Test that multiple event-triggered routines fire on same message.""" + browser, context = browser_and_context + page = await context.new_page() + helper = EventTriggerHelper(page) + + try: + # Create 3 overlapping routines + routine_ids = await helper.create_multiple_routines( + base_name="Handler", count=3, trigger_regex="alert|warning|error" + ) + + # Send matching message + sse_events = await helper.send_chat_message("ERROR: Database connection failed") + + # Verify all 3 routines fired + fired_count = sum( + 1 + for event in sse_events + if event.get("type") == "routine_fired" and event.get("routine_id") in routine_ids + ) + + assert ( + fired_count >= 3 + ), f"Expected all 3 routines to fire, got {fired_count}" + + finally: + await page.close() + + +@pytest.mark.asyncio +async def test_concurrent_limit_prevents_additional_fires(browser_and_context): + """Test that concurrent limit is enforced via batch counts.""" + browser, context = browser_and_context + page = await context.new_page() + helper = EventTriggerHelper(page) + + try: + # Create routine with max_concurrent=1 + routine_id = await helper.create_event_routine( + name="Limited Handler", + trigger_regex="process|task", + max_concurrent=1, + ) + + # Trigger first message + await helper.send_chat_message("Process message 1") + await asyncio.sleep(1) + + # Check first execution logged + executions_1 = await helper.get_routine_execution_log(routine_id) + assert len(executions_1) >= 1 + + # Trigger second message while first is still running + sse_events = await helper.send_chat_message("Process message 2") + + # Second routine should be skipped (concurrent limit) + routine_skipped = any( + event.get("type") == "routine_skipped" + and event.get("reason") == "max_concurrent_reached" + and event.get("routine_id") == routine_id + for event in sse_events + ) + assert routine_skipped, "Routine should be skipped when concurrent limit reached" + + finally: + await page.close() + + +@pytest.mark.asyncio +async def test_rapid_messages_with_multiple_triggers_efficiency(browser_and_context): + """Test efficiency of batch loading with multiple rapid messages.""" + browser, context = browser_and_context + page = await context.new_page() + helper = EventTriggerHelper(page) + + try: + # Create 5 overlapping routines + routine_ids = await helper.create_multiple_routines( + base_name="Rapid", count=5, trigger_regex="test|demo|check" + ) + + # Send 10 matching messages rapidly + for i in range(10): + message = f"test message {i}" + await helper.send_chat_message(message) + await asyncio.sleep(0.1) + + # Check database logs for query efficiency + batch_count, count_query_count = await helper.check_database_queries_in_logs() + + # After fix: should have ~10 batch queries (1 per message) + # Before fix: would have ~50 individual COUNT queries (5 routines × 10 messages) + assert ( + count_query_count == 0 + ), f"Should have 0 individual COUNT queries after fix, got {count_query_count}" + assert ( + batch_count <= 15 + ), f"Should have <=15 batch queries for 10 messages, got {batch_count}" + + finally: + await page.close() + + +@pytest.mark.asyncio +async def test_channel_filter_applied_correctly(browser_and_context): + """Test that channel filter prevents non-matching messages.""" + browser, context = browser_and_context + page = await context.new_page() + helper = EventTriggerHelper(page) + + try: + # Create routine for Slack channel + slack_routine_id = await helper.create_event_routine( + name="Slack Handler", + trigger_regex="alert", + channel="slack", + ) + + # Simulate message from Telegram channel + # (Note: In real UI, would need to change channel context) + page.goto( + "http://localhost:8000/chat?channel=telegram" + ) # Switch channel + await helper.send_chat_message("alert: something urgent") + + # Routine should not fire (different channel) + executions = await helper.get_routine_execution_log(slack_routine_id) + + # Check if any recent execution (last 5 min) exists + recent = [ + e + for e in executions + if (datetime.now() - datetime.fromisoformat(e["timestamp"])).total_seconds() + < 300 + ] + assert ( + len(recent) == 0 + ), "Routine should not fire for different channel" + + finally: + await page.close() + + +@pytest.mark.asyncio +async def test_batch_query_failure_handling(browser_and_context): + """Test graceful handling of batch query failures.""" + browser, context = browser_and_context + page = await context.new_page() + helper = EventTriggerHelper(page) + + try: + # Create routine + routine_id = await helper.create_event_routine( + name="Error Handler", + trigger_regex="test", + ) + + # Simulate database error in logs (if possible with test hooks) + # For now, just verify error handling doesn't crash UI + await helper.send_chat_message("test message") + + # Check that UI remains responsive + assert await page.locator("text=Message processed").is_visible() + + finally: + await page.close() + + +@pytest.mark.asyncio +async def test_routine_execution_history_display(browser_and_context): + """Test that execution history correctly displays routine firings.""" + browser, context = browser_and_context + page = await context.new_page() + helper = EventTriggerHelper(page) + + try: + # Create routine + routine_id = await helper.create_event_routine( + name="History Test", + trigger_regex="test", + ) + + # Trigger routine 3 times + for i in range(3): + await helper.send_chat_message(f"test message {i}") + await asyncio.sleep(0.2) + + # Check execution log + executions = await helper.get_routine_execution_log(routine_id) + assert len(executions) >= 3, "Should have at least 3 executions logged" + + # Verify all are recent (within last 5 minutes) + for execution in executions[:3]: + timestamp = datetime.fromisoformat(execution["timestamp"]) + age = datetime.now() - timestamp + assert age < timedelta(minutes=5), "Execution should be recent" + + finally: + await page.close() + + +@pytest.mark.asyncio +async def test_concurrent_batch_loads_independent(browser_and_context): + """Test that concurrent messages each get independent batch queries.""" + browser, context = browser_and_context + page = await context.new_page() + helper = EventTriggerHelper(page) + + try: + # Create 5 routines matching different patterns + r1_id = await helper.create_event_routine( + name="Pattern A", trigger_regex="alpha|alpha_only" + ) + r2_id = await helper.create_event_routine( + name="Pattern B", trigger_regex="beta|beta_only" + ) + r3_id = await helper.create_event_routine( + name="Pattern AB", trigger_regex="alpha|beta|common" + ) + + # Send overlapping messages + # Message 1: matches r1, r3 + sse1 = await helper.send_chat_message("alpha common") + await asyncio.sleep(0.1) + + # Message 2: matches r2, r3 + sse2 = await helper.send_chat_message("beta common") + await asyncio.sleep(0.1) + + # Verify correct routines fired + r1_fired_msg1 = any( + e.get("routine_id") == r1_id for e in sse1 if e.get("type") == "routine_fired" + ) + r2_fired_msg2 = any( + e.get("routine_id") == r2_id for e in sse2 if e.get("type") == "routine_fired" + ) + r3_fired_both = ( + any( + e.get("routine_id") == r3_id for e in sse1 if e.get("type") == "routine_fired" + ) + and any( + e.get("routine_id") == r3_id for e in sse2 if e.get("type") == "routine_fired" + ) + ) + + assert r1_fired_msg1, "Routine 1 should fire on message 1" + assert r2_fired_msg2, "Routine 2 should fire on message 2" + assert r3_fired_both, "Routine 3 should fire on both messages" + + finally: + await page.close() + + +# ============================================================================= +# Integration with existing test patterns +# ============================================================================= + + +if __name__ == "__main__": + # Run tests with: pytest tests/e2e/scenarios/test_routine_event_batch.py -v + pytest.main([__file__, "-v", "-s"])