clawmates/crates/cm-scheduler/src/lib.rs

//! Routines (spec §7.6): agent-created scheduled tasks. Due routines are
//! claimed atomically (one firing even with replicas) and each firing
//! drives a REAL run through the runtime in the routine's dedicated
//! session — gated tools inside a routine still hit the approval queue.

use cm_db::repo::{agents, routines, sessions};
use cm_runtime::Runtime;
use sqlx::PgPool;
use time::OffsetDateTime;

pub use cm_runtime::scheduling::next_occurrence;

#[derive(Debug, thiserror::Error)]
pub enum ScheduleError {
    #[error(transparent)]
    Pattern(#[from] cm_runtime::scheduling::ScheduleError),
    #[error(transparent)]
    Db(#[from] cm_db::DbError),
}

pub struct Scheduler {
    pool: PgPool,
    runtime: Runtime,
}

impl Scheduler {
    pub fn new(pool: PgPool, runtime: Runtime) -> Scheduler {
        Scheduler { pool, runtime }
    }

    /// Fires every due routine once and reschedules it. Returns how many
    /// fired. Time is a parameter so tests control the clock.
    pub async fn tick(&self, now: OffsetDateTime) -> Result<usize, ScheduleError> {
        let due = routines::claim_due(&self.pool, now).await?;
        for routine in &due {
            // Reschedule first: a firing failure must not stall the clock.
            let next = next_occurrence(&routine.schedule_cron, now).ok();
            routines::set_next_run(&self.pool, routine.id, next).await?;

            let message = routine.action["message"].as_str().unwrap_or_default();
            if message.is_empty() {
                continue;
            }
            let agent_id = cm_domain::AgentId::from(routine.agent_id);
            let Ok(agent) = agents::get(&self.pool, agent_id).await else {
                continue; // deleted agent: routine is orphaned
            };

            // Each routine runs in one dedicated, recognizable session.
            let title = format!("⏰ {}", routine.name);
            let session = match sessions::list_by_agent(&self.pool, agent_id)
                .await?
                .into_iter()
                .find(|s| s.title == title)
            {
                Some(existing) => existing,
                None => sessions::create(&self.pool, agent_id, agent.workspace_id, &title).await?,
            };
            let _ = self.runtime.send_message(session.id, message).await;
        }
        Ok(due.len())
    }

    /// The production loop: ticks on an interval with the real clock.
    pub fn spawn(self, interval: std::time::Duration) {
        tokio::spawn(async move {
            let mut tick = tokio::time::interval(interval);
            loop {
                tick.tick().await;
                let _ = self.tick(OffsetDateTime::now_utc()).await;
            }
        });
    }
}