use std::sync::Arc; use std::time::Duration; use cm_domain::{ AccessPolicy, Agent, AgentId, AgentStatus, MessageRole, Role, User, UserId, Workspace, WorkspaceId, }; use cm_llm::ScriptedProvider; use cm_runtime::{Runtime, RuntimeConfig}; use cm_scheduler::{next_occurrence, Scheduler}; use serde_json::json; use time::macros::datetime; #[test] fn next_occurrence_follows_the_cron_pattern() { let after = datetime!(2026-06-10 08:30:00 UTC); // Daily at 09:00. assert_eq!( next_occurrence("0 9 * * *", after).unwrap(), datetime!(2026-06-10 09:00:00 UTC) ); // Already past 09:00 today → tomorrow. let late = datetime!(2026-06-10 09:30:00 UTC); assert_eq!( next_occurrence("0 9 * * *", late).unwrap(), datetime!(2026-06-11 09:00:00 UTC) ); // Every minute. assert_eq!( next_occurrence("* * * * *", after).unwrap(), datetime!(2026-06-10 08:31:00 UTC) ); // Mondays only (2026-06-10 is a Wednesday). assert_eq!( next_occurrence("0 9 * * MON", after).unwrap(), datetime!(2026-06-15 09:00:00 UTC) ); } #[test] fn invalid_cron_patterns_are_errors() { let after = datetime!(2026-06-10 08:30:00 UTC); assert!(next_occurrence("not a cron", after).is_err()); assert!(next_occurrence("99 99 * * *", after).is_err()); } async fn seeded(pool: &sqlx::PgPool) -> Agent { let workspace = Workspace { id: WorkspaceId::new(), name: "Acme".into(), plan: "team".into(), }; cm_db::repo::workspaces::insert(pool, &workspace) .await .unwrap(); let owner = User { id: UserId::new(), workspace_id: workspace.id, email: format!("{}@acme.test", UserId::new()), role: Role::Owner, display_name: "Owner".into(), created_at: time::OffsetDateTime::UNIX_EPOCH, }; cm_db::repo::users::insert(pool, &owner).await.unwrap(); let agent = Agent { id: AgentId::new(), workspace_id: workspace.id, name: "Scout".into(), job_title: "Analyst".into(), system_prompt: String::new(), avatar: String::new(), accent: String::new(), wallpaper: String::new(), managed_by: owner.id, status: AgentStatus::Online, }; cm_db::repo::agents::insert(pool, &agent, &AccessPolicy::default()) .await .unwrap(); agent } #[tokio::test] async fn due_routines_fire_real_runs_exactly_once() { let pool = cm_testkit::test_pool().await; let agent = seeded(&pool).await; let runtime = Runtime::new( pool.clone(), Arc::new(ScriptedProvider::from_toml("").unwrap()), RuntimeConfig::basic("scripted", 1024), ); let scheduler = Scheduler::new(pool.clone(), runtime); // A routine that became due a minute ago. let now = time::OffsetDateTime::now_utc(); cm_db::repo::routines::create( &pool, agent.id, "Morning digest", "* * * * *", json!({"message": "compile the digest"}), now - time::Duration::minutes(1), ) .await .unwrap(); let fired = scheduler.tick(now).await.unwrap(); assert_eq!(fired, 1); // Claimed: an immediate second tick fires nothing. assert_eq!(scheduler.tick(now).await.unwrap(), 0); // The routine's clock advanced beyond now. let routines = cm_db::repo::routines::list_by_agent(&pool, agent.id) .await .unwrap(); assert!(routines[0].next_run_at.unwrap() > now); assert!(routines[0].last_run_at.is_some()); // The firing produced a REAL run in the routine's dedicated session. let mut found = false; for _ in 0..100 { let sessions = cm_db::repo::sessions::list_by_agent(&pool, agent.id) .await .unwrap(); if let Some(session) = sessions.iter().find(|s| s.title == "⏰ Morning digest") { let history = cm_db::repo::messages::history(&pool, session.id) .await .unwrap(); if history.len() == 2 && history[0].message.role == MessageRole::User && history[1].message.content["text"] .as_str() .unwrap_or_default() .contains("compile the digest") { found = true; break; } } tokio::time::sleep(Duration::from_millis(50)).await; } assert!(found, "routine run never landed in its session"); // Re-firing later reuses the same session instead of spamming new ones. cm_db::repo::routines::set_next_run(&pool, routines[0].id, Some(now)) .await .unwrap(); scheduler.tick(now).await.unwrap(); for _ in 0..100 { let sessions = cm_db::repo::sessions::list_by_agent(&pool, agent.id) .await .unwrap(); let routine_sessions: Vec<_> = sessions .iter() .filter(|s| s.title == "⏰ Morning digest") .collect(); assert_eq!(routine_sessions.len(), 1); let history = cm_db::repo::messages::history(&pool, routine_sessions[0].id) .await .unwrap(); if history.len() == 4 { return; } tokio::time::sleep(Duration::from_millis(50)).await; } panic!("second firing never landed"); } #[tokio::test] async fn paused_routines_do_not_fire() { let pool = cm_testkit::test_pool().await; let agent = seeded(&pool).await; let runtime = Runtime::new( pool.clone(), Arc::new(ScriptedProvider::from_toml("").unwrap()), RuntimeConfig::basic("scripted", 1024), ); let scheduler = Scheduler::new(pool.clone(), runtime); let now = time::OffsetDateTime::now_utc(); let routine = cm_db::repo::routines::create( &pool, agent.id, "Paused digest", "* * * * *", json!({"message": "nope"}), now - time::Duration::minutes(1), ) .await .unwrap(); sqlx::query("UPDATE routines SET status = 'paused' WHERE id = $1") .bind(routine.id) .execute(&pool) .await .unwrap(); assert_eq!(scheduler.tick(now).await.unwrap(), 0); }