oxios_kernel/

cron.rs

//! Cron scheduler for time-based autonomous agent execution.
//!
//! Allows scheduling agents to run on cron-like schedules without user intervention.
//! Supports 5-field (Linux cron) and 6-7 field expressions via the `cron` crate.

use crate::config::CronConfig;
use crate::git_layer::GitLayer;
use crate::state_store::StateStore;
use crate::types::Priority;
use anyhow::{Result, bail};
use chrono::{DateTime, Utc};
use cron::Schedule;
use parking_lot::{Mutex, RwLock};
use serde::{Deserialize, Serialize};
use std::collections::{HashMap, HashSet};
use std::str::FromStr;
use std::sync::Arc;
use std::sync::atomic::{AtomicBool, Ordering};
use uuid::Uuid;

// ── Data types ─────────────────────────────────────────────

/// Source of a cron job.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize, Default)]
#[serde(rename_all = "lowercase")]
pub enum JobSource {
    /// Defined in config.toml.
    Config,
    /// Created via API.
    #[default]
    Api,
}

/// A cron job definition.
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct CronJob {
    /// Unique job identifier.
    pub id: Uuid,
    /// Human-readable job name.
    pub name: String,
    /// Cron expression (e.g. "0 */6 * * *").
    pub schedule: String,
    /// Goal description for the agent.
    pub goal: String,
    /// Constraints on agent behavior.
    #[serde(default)]
    pub constraints: Vec<String>,
    /// Criteria that must be met for success.
    #[serde(default)]
    pub acceptance_criteria: Vec<String>,
    /// Toolchain preset name.
    #[serde(default = "default_toolchain")]
    pub toolchain: String,
    /// Job priority.
    #[serde(default)]
    pub priority: Priority,
    /// Whether the job is active.
    #[serde(default = "default_true")]
    pub enabled: bool,
    /// Timestamp of the last execution.
    #[serde(skip_serializing_if = "Option::is_none")]
    pub last_run: Option<DateTime<Utc>>,
    /// Timestamp of the next scheduled execution.
    #[serde(skip_serializing_if = "Option::is_none")]
    pub next_run: Option<DateTime<Utc>>,
    /// Number of times this job has run.
    #[serde(default)]
    pub run_count: u64,
    /// Summary of the last execution result.
    #[serde(skip_serializing_if = "Option::is_none")]
    pub last_result: Option<String>,
    /// Whether the last execution succeeded.
    #[serde(skip_serializing_if = "Option::is_none")]
    pub last_success: Option<bool>,
    /// How the job was created.
    #[serde(default)]
    pub source: JobSource,
}

fn default_toolchain() -> String {
    "default".into()
}

fn default_true() -> bool {
    true
}

impl CronJob {
    /// Create a new cron job with a parsed schedule.
    pub fn new(name: String, schedule: String, goal: String) -> Self {
        Self {
            id: Uuid::new_v4(),
            name,
            schedule,
            goal,
            constraints: vec![],
            acceptance_criteria: vec![],
            toolchain: default_toolchain(),
            priority: Priority::default(),
            enabled: true,
            last_run: None,
            next_run: None,
            run_count: 0,
            last_result: None,
            last_success: None,
            source: JobSource::Api,
        }
    }
}

/// Result of a single cron job execution.
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct CronJobResult {
    /// ID of the executed job.
    pub job_id: Uuid,
    /// Name of the executed job.
    pub job_name: String,
    /// When execution started.
    pub started_at: DateTime<Utc>,
    /// When execution finished.
    pub finished_at: DateTime<Utc>,
    /// Whether the execution succeeded.
    pub success: bool,
    /// Human-readable result summary.
    pub summary: String,
}

/// Update fields for an existing cron job.
#[derive(Debug, Default, Deserialize)]
pub struct CronJobUpdate {
    /// New name.
    pub name: Option<String>,
    /// New cron expression.
    pub schedule: Option<String>,
    /// New goal description.
    pub goal: Option<String>,
    /// New constraints.
    pub constraints: Option<Vec<String>>,
    /// New acceptance criteria.
    pub acceptance_criteria: Option<Vec<String>>,
    /// New toolchain preset.
    pub toolchain: Option<String>,
    /// New priority.
    pub priority: Option<Priority>,
    /// Enable or disable.
    pub enabled: Option<bool>,
}

// ── CronScheduler ───────────────────────────────────────────

/// The cron scheduler for time-based autonomous agent execution.
///
/// Allows scheduling agents to run on cron-like schedules without user intervention.
/// Supports 5-field (Linux cron) and 6-7 field expressions via the `cron` crate.
pub struct CronScheduler {
    jobs: Arc<RwLock<HashMap<Uuid, CronJob>>>,
    schedules: Arc<Mutex<HashMap<Uuid, Schedule>>>,
    running_jobs: Arc<Mutex<HashSet<Uuid>>>,
    state_store: Arc<StateStore>,
    cancel: Arc<AtomicBool>,
    dirty: Arc<AtomicBool>,
    tick_interval_secs: u64,
    /// Optional git layer for version-controlled saves.
    git_layer: Option<Arc<GitLayer>>,
    /// Maximum concurrent cron job executions (default: 3).
    max_concurrent_jobs: usize,
    /// Timeout for individual cron job execution in seconds (default: 600 = 10 minutes).
    job_timeout_secs: u64,
}

impl CronScheduler {
    /// Create a new CronScheduler.
    ///
    /// # Arguments
    /// * `state_store` - State store for persisting job definitions
    /// * `tick_interval_secs` - How often to check for due jobs (in seconds)
    pub fn new(state_store: Arc<StateStore>, tick_interval_secs: u64) -> Self {
        Self {
            jobs: Arc::new(RwLock::new(HashMap::new())),
            schedules: Arc::new(Mutex::new(HashMap::new())),
            running_jobs: Arc::new(Mutex::new(HashSet::new())),
            state_store,
            cancel: Arc::new(AtomicBool::new(false)),
            dirty: Arc::new(AtomicBool::new(false)),
            tick_interval_secs,
            git_layer: None,
            max_concurrent_jobs: 3,
            job_timeout_secs: 600,
        }
    }

    /// Set the maximum concurrent cron job executions.
    ///
    /// A value of `0` is rejected — it would silently disable every cron tick
    /// (`current_running(0) >= max(0)` is always true). We clamp to `1` and
    /// warn so the caller can correct their configuration instead of losing
    /// all scheduled jobs without any signal.
    pub fn set_max_concurrent_jobs(&mut self, max: usize) {
        if max == 0 {
            tracing::warn!("set_max_concurrent_jobs(0) would disable all cron jobs; clamping to 1");
            self.max_concurrent_jobs = 1;
        } else {
            self.max_concurrent_jobs = max;
        }
    }

    /// Set the timeout for individual cron job execution in seconds.
    pub fn set_job_timeout_secs(&mut self, secs: u64) {
        self.job_timeout_secs = secs;
    }

    /// Set the git layer for version-controlled saves.
    pub fn set_git_layer(&mut self, gl: Arc<GitLayer>) {
        self.git_layer = Some(gl);
    }

    /// Normalize a cron expression: prepend seconds field if 5-field (Linux style).
    fn normalize_expr(expr: &str) -> String {
        let fields: Vec<&str> = expr.split_whitespace().collect();
        match fields.len() {
            5 => format!("0 {expr}"),
            _ => expr.to_string(),
        }
    }

    /// Parse a cron expression into a `Schedule`.
    fn parse_schedule(&self, expr: &str) -> Result<Schedule> {
        let normalized = Self::normalize_expr(expr);
        Schedule::from_str(&normalized)
            .map_err(|e| anyhow::anyhow!("Invalid cron expression '{expr}': {e}"))
    }

    /// Compute the next fire time after `after`.
    fn next_fire_time(&self, schedule: &Schedule, after: &DateTime<Utc>) -> Option<DateTime<Utc>> {
        schedule.after(after).next()
    }

    /// Add a job. Parses schedule, computes next_run, stores.
    pub async fn add_job(&self, job: CronJob) -> Result<Uuid> {
        let schedule = self.parse_schedule(&job.schedule)?;
        let next = self.next_fire_time(&schedule, &Utc::now());
        let id = job.id;

        self.schedules.lock().insert(id, schedule);
        self.jobs.write().insert(
            id,
            CronJob {
                next_run: next,
                ..job
            },
        );
        self.dirty.store(true, Ordering::Relaxed);
        self.persist_jobs().await;

        tracing::info!(
            name = %self.jobs.read().get(&id).map(|j| j.name.as_str()).unwrap_or("?"),
            %id,
            "Cron job added"
        );
        Ok(id)
    }

    /// Remove a job by ID.
    pub async fn remove_job(&self, id: Uuid) -> Result<()> {
        self.schedules.lock().remove(&id);
        self.jobs
            .write()
            .remove(&id)
            .ok_or_else(|| anyhow::anyhow!("Job {id} not found"))?;
        self.dirty.store(true, Ordering::Relaxed);
        self.persist_jobs().await;
        tracing::info!(%id, "Cron job removed");
        Ok(())
    }

    /// Update a job's fields (enabled, schedule, goal, etc).
    pub async fn update_job(&self, id: Uuid, update: CronJobUpdate) -> Result<()> {
        // Separate the sync mutation from the async persist to avoid
        // holding a !Send RwLockWriteGuard across an await point.
        let should_persist = {
            let mut jobs = self.jobs.write();
            let job = jobs
                .get_mut(&id)
                .ok_or_else(|| anyhow::anyhow!("Job {id} not found"))?;

            if let Some(name) = update.name {
                job.name = name;
            }
            if let Some(schedule) = &update.schedule {
                let parsed = self.parse_schedule(schedule)?;
                self.schedules.lock().insert(id, parsed);
                job.schedule = schedule.clone();
                // Recompute next_run
                let sched = self.schedules.lock().get(&id).cloned();
                if let Some(s) = sched {
                    job.next_run = self.next_fire_time(&s, &Utc::now());
                }
            }
            if let Some(goal) = update.goal {
                job.goal = goal;
            }
            if let Some(constraints) = update.constraints {
                job.constraints = constraints;
            }
            if let Some(criteria) = update.acceptance_criteria {
                job.acceptance_criteria = criteria;
            }
            if let Some(toolchain) = update.toolchain {
                job.toolchain = toolchain;
            }
            if let Some(priority) = update.priority {
                job.priority = priority;
            }
            if let Some(enabled) = update.enabled {
                job.enabled = enabled;
            }

            self.dirty.store(true, Ordering::Relaxed);
            true
        }; // RwLockWriteGuard dropped here, before any .await

        if should_persist {
            self.persist_jobs().await;
        }
        Ok(())
    }

    /// Toggle a job enabled/disabled.
    pub async fn toggle_job(&self, id: Uuid, enabled: bool) -> Result<()> {
        self.update_job(
            id,
            CronJobUpdate {
                enabled: Some(enabled),
                ..Default::default()
            },
        )
        .await
    }

    /// List all jobs.
    pub fn list_jobs(&self) -> Vec<CronJob> {
        self.jobs.read().values().cloned().collect()
    }

    /// Get a single job.
    pub fn get_job(&self, id: Uuid) -> Option<CronJob> {
        self.jobs.read().get(&id).cloned()
    }

    /// Check if a job is currently running.
    pub fn is_running(&self, id: Uuid) -> bool {
        self.running_jobs.lock().contains(&id)
    }

    /// Trigger a job immediately (manual execution, ignores schedule).
    /// Returns the job goal as a string for the caller to execute.
    /// The caller is responsible for calling `mark_job_completed` after execution.
    pub fn trigger_job(&self, id: Uuid) -> Result<CronJob> {
        let job = self
            .jobs
            .read()
            .get(&id)
            .cloned()
            .ok_or_else(|| anyhow::anyhow!("Job {id} not found"))?;

        if self.running_jobs.lock().contains(&id) {
            bail!("Job '{}' is already running", job.name);
        }

        self.running_jobs.lock().insert(id);
        Ok(job)
    }

    /// Mark a job execution as completed.
    pub async fn mark_job_completed(&self, id: Uuid, success: bool, summary: String) {
        self.running_jobs.lock().remove(&id);
        let new_next_run = {
            let mut jobs = self.jobs.write();
            if let Some(job) = jobs.get_mut(&id) {
                job.last_run = Some(Utc::now());
                job.last_result = Some(summary);
                job.last_success = Some(success);
                job.run_count += 1;
                // Recompute next_run
                let sched = self.schedules.lock().get(&id).cloned();
                sched.and_then(|s| self.next_fire_time(&s, &Utc::now()))
            } else {
                None
            }
        };
        if let Some(next_run) = new_next_run {
            let mut jobs = self.jobs.write();
            if let Some(job) = jobs.get_mut(&id) {
                job.next_run = Some(next_run);
            }
        }
        self.dirty.store(true, Ordering::Relaxed);
        self.persist_jobs().await;
    }

    /// Stop the scheduler loop.
    pub fn stop(&self) {
        self.cancel.store(true, Ordering::Relaxed);
        tracing::info!("Cron scheduler stop requested");
    }

    /// Start the main loop. Must be called on an `Arc<Self>`.
    ///
    /// # Arguments
    /// * `executor` - Async closure `(Uuid, String) -> Fut` where args are `(job_id, goal)`,
    ///   returning `(success, summary)`.
    ///
    /// # Example
    ///
    /// ```no_run
    /// use std::sync::Arc;
    /// use oxios_kernel::state_store::StateStore;
    /// use oxios_kernel::cron::CronScheduler;
    ///
    /// # async fn example() {
    /// let state_store = Arc::new(StateStore::new("/tmp/state".into()).unwrap());
    /// let scheduler = Arc::new(CronScheduler::new(state_store, 60));
    /// scheduler.clone().start(|id, goal| async move {
    ///     // execute the agent...
    ///     (true, "Done".to_string())
    /// }).await;
    /// # }
    /// ```
    pub async fn start<F, Fut>(self: Arc<Self>, executor: F)
    where
        F: Fn(Uuid, String) -> Fut + Send + Sync + 'static,
        Fut: std::future::Future<Output = (bool, String)> + Send + 'static,
    {
        let executor = Arc::new(executor);
        let mut interval =
            tokio::time::interval(std::time::Duration::from_secs(self.tick_interval_secs));

        tracing::info!(
            interval_secs = self.tick_interval_secs,
            "Cron scheduler started"
        );

        loop {
            tokio::select! {
               _ = interval.tick() => {
                   if self.cancel.load(Ordering::Relaxed) {
                       tracing::info!("Cron scheduler stopped");
                       return;
                   }
                   // `start` runs as `self: Arc<Self>`. Arc implements
                   // Deref<Target = Self>, so this resolves to the
                   // `&self` method without an explicit deref.
                   self.tick_once(&executor).await;
               }
            }
        }
    }

    /// Single tick: find due jobs and spawn execution.
    ///
    /// Enforces `max_concurrent_jobs` limit and `job_timeout_secs` per job.
    ///
    /// Public seam: callers (admin tooling, tests) can drive a single
    /// deterministic tick without the wall-clock-driven `start` loop.
    pub async fn tick_once<F, Fut>(&self, executor: &Arc<F>)
    where
        F: Fn(Uuid, String) -> Fut + Send + Sync + 'static,
        Fut: std::future::Future<Output = (bool, String)> + Send + 'static,
    {
        let now = Utc::now();

        // Check current concurrency before spawning more
        let current_running = self.running_jobs.lock().len();
        if current_running >= self.max_concurrent_jobs {
            tracing::debug!(
                running = current_running,
                max = self.max_concurrent_jobs,
                "Cron tick: max concurrent jobs reached, skipping"
            );
            return;
        }

        let due: Vec<(Uuid, String)> = {
            let jobs = self.jobs.read();
            jobs.iter()
                .filter(|(_, job)| {
                    job.enabled
                        && job.next_run.is_some_and(|nr| nr <= now)
                        && !self.running_jobs.lock().contains(&job.id)
                })
                .map(|(_, job)| (job.id, job.goal.clone()))
                .collect()
        };

        let total_due = due.len();
        for (spawned, (id, goal)) in due.into_iter().enumerate() {
            // Check concurrency limit before each spawn
            if self.running_jobs.lock().len() >= self.max_concurrent_jobs {
                tracing::info!(
                    spawned,
                    remaining = total_due - spawned,
                    "Cron tick: max concurrent jobs reached, deferring remaining"
                );
                break;
            }

            self.running_jobs.lock().insert(id);
            let exec = executor.clone();
            let me = self.clone();
            let timeout_secs = self.job_timeout_secs;
            tokio::spawn(async move {
                tracing::info!(%id, "Cron job triggered");
                let result = tokio::time::timeout(
                    std::time::Duration::from_secs(timeout_secs),
                    exec(id, goal),
                )
                .await;

                let (success, summary) = match result {
                    Ok((s, m)) => (s, m),
                    Err(_) => {
                        tracing::error!(%id, timeout_secs, "Cron job timed out");
                        (false, format!("Timed out after {timeout_secs} seconds"))
                    }
                };
                tracing::info!(%id, success, "Cron job completed");
                me.mark_job_completed(id, success, summary).await;
            });
        }
    }

    /// Persist all jobs to disk. Public seam for tests and the kernel handle
    /// to call directly (e.g. after a config import).
    pub(crate) async fn persist_jobs(&self) {
        let job_list: Vec<CronJob> = {
            let jobs = self.jobs.read();
            jobs.values().cloned().collect()
        };
        if let Err(e) = self.state_store.save_json("cron", "jobs", &job_list).await {
            tracing::error!("Failed to persist cron jobs: {}", e);
        }
        // Fire-and-forget git commit if git layer is configured
        if let Some(ref gl) = self.git_layer
            && gl.is_enabled()
        {
            let _ = gl.commit_file("cron/jobs.json", "cron: update jobs");
        }
    }

    /// Restore jobs from disk on startup.
    pub async fn restore_jobs(&self) {
        match self
            .state_store
            .load_json::<Vec<CronJob>>("cron", "jobs")
            .await
        {
            Ok(Some(job_list)) => {
                for mut job in job_list {
                    // Re-parse schedule and recompute next_run
                    match self.parse_schedule(&job.schedule) {
                        Ok(schedule) => {
                            job.next_run = self.next_fire_time(&schedule, &Utc::now());
                            self.schedules.lock().insert(job.id, schedule);
                            self.jobs.write().insert(job.id, job);
                        }
                        Err(e) => {
                            tracing::error!(job = %job.name, error = %e, "Skipping job with invalid schedule");
                        }
                    }
                }
                tracing::info!(count = self.jobs.read().len(), "Cron jobs restored");
            }
            Ok(None) => {
                tracing::info!("No saved cron jobs found");
            }
            Err(e) => {
                tracing::error!("Failed to restore cron jobs: {}", e);
            }
        }
    }

    /// Load jobs defined in config (called during startup).
    /// Config-defined jobs are only added if they don't already exist (API wins on conflict).
    pub async fn load_from_config(&self, config: &CronConfig) {
        if !config.enabled {
            tracing::info!("Cron scheduler is disabled in config");
            return;
        }

        for (name, inline) in &config.jobs {
            let schedule = inline.schedule.clone();
            let goal = inline.goal.clone();

            let job = CronJob {
                id: Uuid::new_v4(),
                name: name.clone(),
                schedule: schedule.clone(),
                goal,
                constraints: inline.constraints.clone(),
                acceptance_criteria: inline.acceptance_criteria.clone(),
                toolchain: inline.toolchain.clone(),
                priority: inline.priority,
                enabled: inline.enabled,
                last_run: None,
                next_run: None,
                run_count: 0,
                last_result: None,
                last_success: None,
                source: JobSource::Config,
            };

            // Check if a job with this name already exists (from API)
            {
                let jobs = self.jobs.read();
                if jobs.values().any(|j| j.name == *name) {
                    tracing::debug!(name = %name, "Skipping config job — already exists via API");
                    continue;
                }
            }

            if let Err(e) = self.add_job(job).await {
                tracing::error!(name = %name, error = %e, "Failed to load config job");
            } else {
                tracing::info!(name = %name, "Loaded cron job from config");
            }
        }
    }
}

impl Clone for CronScheduler {
    fn clone(&self) -> Self {
        Self {
            jobs: self.jobs.clone(),
            schedules: self.schedules.clone(),
            running_jobs: self.running_jobs.clone(),
            state_store: self.state_store.clone(),
            cancel: self.cancel.clone(),
            dirty: self.dirty.clone(),
            tick_interval_secs: self.tick_interval_secs,
            git_layer: self.git_layer.clone(),
            max_concurrent_jobs: self.max_concurrent_jobs,
            job_timeout_secs: self.job_timeout_secs,
        }
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use chrono::Timelike;

    fn test_store() -> Arc<StateStore> {
        let temp_dir = tempfile::tempdir().unwrap();
        Arc::new(StateStore::new(temp_dir.path().to_path_buf()).unwrap())
    }

    #[test]
    fn test_normalize_5field() {
        assert_eq!(CronScheduler::normalize_expr("0 9 * * *"), "0 0 9 * * *");
    }

    #[test]
    fn test_normalize_6field() {
        assert_eq!(CronScheduler::normalize_expr("0 0 9 * * *"), "0 0 9 * * *");
    }

    #[test]
    fn test_normalize_7field() {
        assert_eq!(
            CronScheduler::normalize_expr("0 0 9 * * * 2026"),
            "0 0 9 * * * 2026"
        );
    }

    #[test]
    fn test_parse_valid() {
        let cs = CronScheduler::new(test_store(), 60);
        assert!(cs.parse_schedule("0 9 * * *").is_ok());
    }

    #[test]
    fn test_parse_invalid() {
        let cs = CronScheduler::new(test_store(), 60);
        assert!(cs.parse_schedule("invalid").is_err());
    }

    #[test]
    fn test_next_fire_time_daily() {
        let cs = CronScheduler::new(test_store(), 60);
        let schedule = cs.parse_schedule("0 9 * * *").unwrap();
        let now = chrono::NaiveDate::from_ymd_opt(2026, 5, 6)
            .unwrap()
            .and_hms_opt(8, 0, 0)
            .unwrap();
        let now_utc = DateTime::<Utc>::from_naive_utc_and_offset(now, Utc);
        let next = cs.next_fire_time(&schedule, &now_utc);
        assert!(next.is_some());
        let next = next.unwrap();
        assert_eq!(next.hour(), 9);
    }

    #[test]
    fn test_next_fire_time_every_15min() {
        let cs = CronScheduler::new(test_store(), 60);
        let schedule = cs.parse_schedule("*/15 * * * *").unwrap();
        let now = chrono::NaiveDate::from_ymd_opt(2026, 5, 6)
            .unwrap()
            .and_hms_opt(10, 7, 0)
            .unwrap();
        let now_utc = DateTime::<Utc>::from_naive_utc_and_offset(now, Utc);
        let next = cs.next_fire_time(&schedule, &now_utc);
        assert!(next.is_some());
        let next = next.unwrap();
        assert_eq!(next.minute(), 15);
    }

    #[test]
    fn test_add_job_computes_next_run() {
        let job = CronJob::new("test".into(), "0 9 * * *".into(), "Test goal".into());
        assert!(job.next_run.is_none()); // not computed yet
        assert!(job.enabled);
        assert_eq!(job.run_count, 0);
    }

    #[test]
    fn test_job_source_default() {
        let job = CronJob::new("test".into(), "0 9 * * *".into(), "goal".into());
        assert_eq!(job.source, JobSource::Api);
    }

    #[tokio::test]
    async fn test_add_job() {
        let store = test_store();
        let cs = CronScheduler::new(store, 60);
        let job = CronJob::new("test-job".into(), "0 9 * * *".into(), "Run me".into());
        let id = cs.add_job(job).await.unwrap();
        assert!(cs.get_job(id).is_some());
        assert_eq!(cs.list_jobs().len(), 1);
    }

    #[tokio::test]
    async fn test_remove_job() {
        let store = test_store();
        let cs = CronScheduler::new(store, 60);
        let job = CronJob::new("remove-me".into(), "0 10 * * *".into(), "Gone".into());
        let id = cs.add_job(job).await.unwrap();
        cs.remove_job(id).await.unwrap();
        assert!(cs.get_job(id).is_none());
    }

    #[tokio::test]
    async fn test_trigger_job() {
        let store = test_store();
        let cs = CronScheduler::new(store, 60);
        let job = CronJob::new("trigger-me".into(), "0 11 * * *".into(), "Goal text".into());
        let id = cs.add_job(job).await.unwrap();

        let triggered = cs.trigger_job(id).unwrap();
        assert_eq!(triggered.goal, "Goal text");
        assert!(cs.is_running(id));

        cs.mark_job_completed(id, true, "ok".into()).await;
        assert!(!cs.is_running(id));
    }

    #[tokio::test]
    async fn test_trigger_already_running() {
        let store = test_store();
        let cs = CronScheduler::new(store, 60);
        let job = CronJob::new("running".into(), "0 12 * * *".into(), "goal".into());
        let id = cs.add_job(job).await.unwrap();
        cs.trigger_job(id).unwrap();
        let result = cs.trigger_job(id);
        assert!(result.is_err());
    }

    #[tokio::test]
    async fn test_update_job() {
        let store = test_store();
        let cs = CronScheduler::new(store, 60);
        let job = CronJob::new("old-name".into(), "0 9 * * *".into(), "old goal".into());
        let id = cs.add_job(job).await.unwrap();

        cs.update_job(
            id,
            CronJobUpdate {
                name: Some("new-name".into()),
                goal: Some("new goal".into()),
                enabled: Some(false),
                ..Default::default()
            },
        )
        .await
        .unwrap();

        let updated = cs.get_job(id).unwrap();
        assert_eq!(updated.name, "new-name");
        assert_eq!(updated.goal, "new goal");
        assert!(!updated.enabled);
    }

    #[tokio::test]
    async fn test_toggle_job() {
        let store = test_store();
        let cs = CronScheduler::new(store, 60);
        let job = CronJob::new("toggle".into(), "0 9 * * *".into(), "goal".into());
        let id = cs.add_job(job).await.unwrap();
        assert!(cs.get_job(id).unwrap().enabled);

        cs.toggle_job(id, false).await.unwrap();
        assert!(!cs.get_job(id).unwrap().enabled);

        cs.toggle_job(id, true).await.unwrap();
        assert!(cs.get_job(id).unwrap().enabled);
    }

    #[tokio::test]
    async fn test_mark_completed_updates_next_run() {
        let store = test_store();
        let cs = CronScheduler::new(store, 60);
        let job = CronJob::new("comp".into(), "*/5 * * * *".into(), "goal".into());
        let id = cs.add_job(job).await.unwrap();

        let before = cs.get_job(id).unwrap().next_run;
        assert!(before.is_some());

        // Simulate time passing: set next_run to 5 minutes in the past
        let now = Utc::now();
        {
            let mut jobs = cs.jobs.write();
            if let Some(j) = jobs.get_mut(&id) {
                j.next_run = Some(now - chrono::Duration::minutes(5));
            }
        }

        cs.mark_job_completed(id, true, "ok".into()).await;
        let after = cs.get_job(id).unwrap().next_run;
        assert!(after.is_some());
        // After completion, next_run should be set to a future time (>= now)
        assert!(after.unwrap() >= now);
    }

    #[test]
    fn test_max_concurrent_enforced() {
        let temp_dir = tempfile::tempdir().unwrap();
        let store = Arc::new(StateStore::new(temp_dir.path().to_path_buf()).unwrap());
        let mut scheduler = CronScheduler::new(store, 60);
        scheduler.set_max_concurrent_jobs(2);
        assert_eq!(scheduler.max_concurrent_jobs, 2);
    }

    #[test]
    fn test_job_timeout_configurable() {
        let temp_dir = tempfile::tempdir().unwrap();
        let store = Arc::new(StateStore::new(temp_dir.path().to_path_buf()).unwrap());
        let mut scheduler = CronScheduler::new(store, 60);
        scheduler.set_job_timeout_secs(300);
        assert_eq!(scheduler.job_timeout_secs, 300);
    }

    // ── start-loop execution (tick_once public seam) ──────────────

    /// Build a job whose `next_run` is in the past so the next tick fires.
    async fn add_due_job(cs: &CronScheduler, name: &str, goal: &str) -> Uuid {
        let id = cs
            .add_job(CronJob::new(name.into(), "*/5 * * * *".into(), goal.into()))
            .await
            .unwrap();
        // Force next_run into the past so the next tick picks it up
        // immediately regardless of when add_job ran.
        {
            let mut jobs = cs.jobs.write();
            if let Some(j) = jobs.get_mut(&id) {
                j.next_run = Some(Utc::now() - chrono::Duration::seconds(60));
            }
        }
        id
    }

    #[tokio::test]
    async fn tick_once_runs_due_jobs_and_marks_completion() {
        let cs = Arc::new(CronScheduler::new(test_store(), 60));
        let id = add_due_job(&cs, "due", "do the thing").await;

        let called = Arc::new(Mutex::new(Vec::<(Uuid, String)>::new()));
        let called_for_tick = called.clone();
        let exec = Arc::new(move |jid: Uuid, goal: String| {
            let called = called_for_tick.clone();
            async move {
                called.lock().push((jid, goal));
                (true, "ok".into())
            }
        });

        cs.tick_once(&exec).await;

        // Allow the spawned completion task to run.
        for _ in 0..50 {
            if cs.get_job(id).unwrap().run_count >= 1 {
                break;
            }
            tokio::task::yield_now().await;
        }

        let log = called.lock().clone();
        assert_eq!(log, vec![(id, "do the thing".into())]);
        let job = cs.get_job(id).unwrap();
        assert_eq!(job.run_count, 1, "completion must increment run_count");
        assert_eq!(job.last_success, Some(true));
        assert_eq!(job.last_result.as_deref(), Some("ok"));
        assert!(job.last_run.is_some(), "completion must set last_run");
        // next_run must be advanced to a future time.
        assert!(job.next_run.unwrap() > Utc::now() - chrono::Duration::seconds(1));
        assert!(!cs.is_running(id), "completion must clear running flag");
    }

    #[tokio::test]
    async fn tick_once_skips_disabled_jobs() {
        let cs = Arc::new(CronScheduler::new(test_store(), 60));
        let id = add_due_job(&cs, "off", "ignored").await;
        cs.toggle_job(id, false).await.unwrap();

        let called = Arc::new(Mutex::new(0u32));
        let called_for_tick = called.clone();
        let exec = Arc::new(move |_jid: Uuid, _goal: String| {
            let called = called_for_tick.clone();
            async move {
                *called.lock() += 1;
                (true, String::new())
            }
        });

        cs.tick_once(&exec).await;
        tokio::task::yield_now().await;

        assert_eq!(*called.lock(), 0);
        assert_eq!(cs.get_job(id).unwrap().run_count, 0);
    }

    #[tokio::test]
    async fn tick_once_prevents_overlap_with_already_running_job() {
        // If a previous tick spawned a job whose spawned task is still
        // in flight (still in `running_jobs`), the next tick must NOT
        // re-trigger it — otherwise a slow agent would run multiple
        // copies of the same job.
        let cs = Arc::new(CronScheduler::new(test_store(), 60));
        let id = add_due_job(&cs, "long", "slow").await;
        // Manually mark the job as running, simulating an in-flight tick
        // whose spawn has not yet finished.
        cs.running_jobs.lock().insert(id);

        let called = Arc::new(Mutex::new(0u32));
        let called_for_tick = called.clone();
        let exec = Arc::new(move |_jid: Uuid, _goal: String| {
            let called = called_for_tick.clone();
            async move {
                *called.lock() += 1;
                (true, String::new())
            }
        });

        cs.tick_once(&exec).await;
        tokio::task::yield_now().await;

        assert_eq!(
            *called.lock(),
            0,
            "overlap-prevention must skip a job already in running_jobs"
        );
        // Clean up so the scheduler's Drop doesn't observe a stale entry.
        cs.running_jobs.lock().remove(&id);
    }

    #[tokio::test]
    async fn tick_once_respects_max_concurrent_jobs() {
        let mut cs = CronScheduler::new(test_store(), 60);
        cs.set_max_concurrent_jobs(2);
        let cs = Arc::new(cs);

        // Three due jobs, capacity=2 → one is deferred on this tick.
        add_due_job(&cs, "a", "a").await;
        add_due_job(&cs, "b", "b").await;
        add_due_job(&cs, "c", "c").await;

        let called = Arc::new(Mutex::new(0u32));
        let called_for_tick = called.clone();
        let exec = Arc::new(move |_jid: Uuid, _goal: String| {
            let called = called_for_tick.clone();
            async move {
                *called.lock() += 1;
                (true, String::new())
            }
        });

        cs.tick_once(&exec).await;
        // Wait for all spawned ticks to finish. Each increments `called`
        // synchronously before the executor returns, so this is bounded.
        for _ in 0..200 {
            if *called.lock() >= 2 {
                break;
            }
            tokio::task::yield_now().await;
        }
        assert_eq!(
            *called.lock(),
            2,
            "tick_once must enforce max_concurrent_jobs"
        );
    }
}