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, routine_runs, routines, sessions, teams, topology_runs};
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. A
            // one-shot routine (Scheduled mode, a specific date/time) fires once
            // and never reschedules.
            let one_shot = routine
                .action
                .get("one_shot")
                .and_then(|v| v.as_bool())
                .unwrap_or(false);
            let next = if one_shot {
                None
            } else {
                next_occurrence(&routine.schedule_cron, now).ok()
            };
            routines::set_next_run(&self.pool, routine.id, next).await?;

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

            // Topology routine: fire the whole team's stored topology as one
            // durable run (the entire team loops, not just the coordinator).
            if let Some(topo) = routine.action.get("topology") {
                let team_id = topo
                    .get("team_id")
                    .and_then(|v| v.as_str())
                    .and_then(|s| s.parse::<uuid::Uuid>().ok());
                let task = topo
                    .get("task")
                    .and_then(|v| v.as_str())
                    .unwrap_or(routine.name.as_str());
                let run_id = routine_runs::start(&self.pool, routine.id).await.ok();
                let res: Result<(), String> = match team_id {
                    Some(tid) => match teams::get_team(&self.pool, tid, agent.workspace_id).await {
                        Ok(team) => topology_runs::enqueue_run(
                            &self.pool,
                            uuid::Uuid::now_v7(),
                            agent.workspace_id,
                            task,
                            &team.graph,
                        )
                        .await
                        .map_err(|e| e.to_string()),
                        Err(e) => Err(e.to_string()),
                    },
                    None => Err("routine topology action missing team_id".to_string()),
                };
                if let Some(rid) = run_id {
                    let (status, err) = match &res {
                        Ok(_) => ("ok", None),
                        Err(e) => ("error", Some(e.clone())),
                    };
                    let _ = routine_runs::finish(&self.pool, rid, status, err.as_deref()).await;
                }
                continue;
            }

            let message = routine.action["message"].as_str().unwrap_or_default();
            if message.is_empty() {
                continue;
            }

            // 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?,
            };
            // Journal the firing for the dashboard routines panel.
            let run_id = routine_runs::start(&self.pool, routine.id).await.ok();
            let res = self.runtime.send_message(session.id, message).await;
            if let Some(rid) = run_id {
                let (status, err) = match &res {
                    Ok(_) => ("ok", None),
                    Err(e) => ("error", Some(format!("{e}"))),
                };
                let _ = routine_runs::finish(&self.pool, rid, status, err.as_deref()).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;
            }
        });
    }
}