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;

/// Most occurrences one tick will dispatch.
///
/// A backlog — a clock jump, a long outage, or a cron expression that
/// accidentally resolves to "every minute" — would otherwise fan out every
/// missed occurrence at once. For a topology routine that is one container
/// each. The remainder stays due and is picked up by the following tick.
const MAX_FIRES_PER_TICK: usize = 25;

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.
    ///
    /// Each occurrence is claimed in `routine_fires` before it is dispatched,
    /// and settled after. That ordering is what makes a firing survive a
    /// restart: the clock still advances first (a failing action must not
    /// stall the schedule), but the claim row remembers that the occurrence
    /// was owed, so a crash between reschedule and dispatch is retried instead
    /// of silently skipped — and an occurrence already dispatched is never
    /// dispatched twice.
    pub async fn tick(&self, now: OffsetDateTime) -> Result<usize, ScheduleError> {
        let due = routines::claim_due(&self.pool, now).await?;
        // Cap the fan-out. A backlog (clock jump, long outage, a cron that
        // resolves to "every minute" by accident) would otherwise dispatch
        // every missed occurrence in one tick — for topology routines that is
        // one container each.
        let mut fired = 0usize;
        for routine in due.iter().take(MAX_FIRES_PER_TICK) {
            // The occurrence's own timestamp identifies the slot. `claim_due`
            // does not clear `next_run_at`, so this is still the value that
            // came due.
            let slot = routine.next_run_at.unwrap_or(now);
            match routines::claim_fire(&self.pool, routine.id, slot).await {
                Ok(routines::FireClaim::Fresh) | Ok(routines::FireClaim::Retry) => {}
                Ok(routines::FireClaim::Settled) => {
                    // Already dispatched by a previous tick or replica. Let the
                    // clock advance below, but do not run the work again.
                    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()
                    };
                    let _ = routines::set_next_run(&self.pool, routine.id, next).await;
                    continue;
                }
                Err(e) => {
                    // Could not take the slot. Leaving `next_run_at` untouched
                    // means the occurrence is still due and the next tick tries
                    // again — the safe direction.
                    eprintln!("scheduler: claiming fire for routine {}: {e}", routine.id);
                    continue;
                }
            }
            fired += 1;

            // Reschedule before dispatching: a firing failure must not stall
            // the clock. The claim above is what keeps this from losing the
            // occurrence outright. 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 agent = match agents::get(&self.pool, agent_id).await {
                Ok(a) => a,
                Err(e) => {
                    // Orphaned routine (deleted agent, or a row we cannot
                    // read). Settle the slot rather than leaving it `claimed`:
                    // an unsettled claim looks like a crash mid-fire, so every
                    // tick would re-claim the same routine forever and the
                    // table would grow one stuck row per occurrence.
                    eprintln!(
                        "scheduler: routine {} references agent {agent_id} which could not be \
                         read ({e}) — settling the occurrence as failed",
                        routine.id
                    );
                    let _ = routines::complete_fire(
                        &self.pool,
                        routine.id,
                        slot,
                        Some(&format!("agent {agent_id} unreadable: {e}")),
                    )
                    .await;
                    continue;
                }
            };

            // 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;
                }
                let topo_err = res.as_ref().err().cloned();
                let _ = routines::complete_fire(&self.pool, routine.id, slot, topo_err.as_deref())
                    .await;
                continue;
            }

            let message = routine.action["message"].as_str().unwrap_or_default();
            if message.is_empty() {
                // Neither a topology nor a message action: there is nothing to
                // dispatch. Settle it so the slot is not mistaken for a crash.
                eprintln!(
                    "scheduler: routine {} has no `topology` or `message` action — nothing to fire",
                    routine.id
                );
                let _ = routines::complete_fire(
                    &self.pool,
                    routine.id,
                    slot,
                    Some("routine action has neither `topology` nor `message`"),
                )
                .await;
                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;
            let send_err = res.as_ref().err().map(|e| format!("{e}"));
            if let Some(rid) = run_id {
                let (status, err) = match &res {
                    Ok(_) => ("ok", None),
                    Err(_) => ("error", send_err.clone()),
                };
                let _ = routine_runs::finish(&self.pool, rid, status, err.as_deref()).await;
            }
            let _ =
                routines::complete_fire(&self.pool, routine.id, slot, send_err.as_deref()).await;
        }
        if due.len() > MAX_FIRES_PER_TICK {
            eprintln!(
                "scheduler: {} routines were due; fired {MAX_FIRES_PER_TICK} this tick, \
                 {} deferred to the next one",
                due.len(),
                due.len() - MAX_FIRES_PER_TICK,
            );
        }
        Ok(fired)
    }

    /// 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;
                // A persistently failing tick used to be invisible: the result
                // was discarded, so a scheduler that stopped firing looked
                // exactly like one with nothing to do.
                if let Err(e) = self.tick(OffsetDateTime::now_utc()).await {
                    eprintln!("scheduler: tick failed: {e}");
                }
            }
        });
    }
}