clawmates/crates/cm-scheduler/tests/scheduler.rs

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);
}

/// The crash window this exists to close.
///
/// The scheduler advances `next_run_at` before dispatching, so a process that
/// dies between the two used to drop the occurrence with nothing anywhere
/// recording that it was owed. The claim row is what makes that recoverable:
/// a slot left `claimed` is a crash mid-fire, and the next tick retries it.
#[tokio::test]
async fn an_occurrence_claimed_but_never_settled_is_retried() {
    let pool = cm_testkit::test_pool().await;
    let agent = seeded(&pool).await;
    let now = time::OffsetDateTime::now_utc();
    let slot = now - time::Duration::minutes(1);

    let routine = cm_db::repo::routines::create(
        &pool,
        agent.id,
        "Nightly sweep",
        "* * * * *",
        json!({"message": "sweep"}),
        slot,
    )
    .await
    .unwrap();

    use cm_db::repo::routines::FireClaim;

    // First claim: nobody has this occurrence.
    assert_eq!(
        cm_db::repo::routines::claim_fire(&pool, routine.id, slot)
            .await
            .unwrap(),
        FireClaim::Fresh
    );

    // Simulate a crash: claimed, never settled. The next attempt must be told
    // it is safe to retry — no completion was ever recorded, so nothing
    // downstream saw a result.
    assert_eq!(
        cm_db::repo::routines::claim_fire(&pool, routine.id, slot)
            .await
            .unwrap(),
        FireClaim::Retry
    );

    // Once settled, the same occurrence must never fire again — this is the
    // branch that keeps a scheduled mission to one container across restarts.
    cm_db::repo::routines::complete_fire(&pool, routine.id, slot, None)
        .await
        .unwrap();
    assert_eq!(
        cm_db::repo::routines::claim_fire(&pool, routine.id, slot)
            .await
            .unwrap(),
        FireClaim::Settled
    );

    // A *different* occurrence of the same routine is independent.
    let later = slot + time::Duration::minutes(1);
    assert_eq!(
        cm_db::repo::routines::claim_fire(&pool, routine.id, later)
            .await
            .unwrap(),
        FireClaim::Fresh
    );
}

/// A failed dispatch settles the slot rather than leaving it retryable.
/// Retrying a persistently failing action every tick is how a broken routine
/// becomes a denial-of-service against the thing it talks to.
#[tokio::test]
async fn a_failed_dispatch_is_terminal_for_that_occurrence() {
    let pool = cm_testkit::test_pool().await;
    let agent = seeded(&pool).await;
    let slot = time::OffsetDateTime::now_utc() - time::Duration::minutes(1);
    let routine = cm_db::repo::routines::create(
        &pool,
        agent.id,
        "Flaky",
        "* * * * *",
        json!({"message": "x"}),
        slot,
    )
    .await
    .unwrap();

    use cm_db::repo::routines::FireClaim;
    cm_db::repo::routines::claim_fire(&pool, routine.id, slot)
        .await
        .unwrap();
    cm_db::repo::routines::complete_fire(&pool, routine.id, slot, Some("gateway timed out"))
        .await
        .unwrap();

    assert_eq!(
        cm_db::repo::routines::claim_fire(&pool, routine.id, slot)
            .await
            .unwrap(),
        FireClaim::Settled,
        "a failed occurrence must not be retried forever"
    );

    let err: Option<String> =
        sqlx::query_scalar("SELECT error FROM routine_fires WHERE routine_id = $1")
            .bind(routine.id)
            .fetch_one(&pool)
            .await
            .unwrap();
    assert_eq!(err.as_deref(), Some("gateway timed out"));
}