Add cron runner + tests

Author: Snehal-Reddy

src/cron_runner.rs

@@ -1 +1,160 @@
 //! Tick loop: load jobs.json, find due jobs, execute (inbound to agent or direct sendMessage).
+
+use std::sync::Arc;
+
+use tokio::sync::mpsc;
+
+use crate::telegram::{InboundMsg, OutboundMsg};
+use crate::tools::cron::{CronStore, JobAction};
+
+fn unix_now() -> u64 {
+    std::time::SystemTime::now()
+        .duration_since(std::time::UNIX_EPOCH)
+        .map(|d| d.as_secs())
+        .unwrap_or(0)
+}
+
+/// Run one tick: find due jobs, send to channels, mark fired. Used by runner and tests.
+pub async fn tick_once(
+    store: &CronStore,
+    inbound_tx: &mpsc::Sender<InboundMsg>,
+    outbound_tx: &mpsc::Sender<OutboundMsg>,
+    now: u64,
+) {
+    let due = store.find_due(now);
+    for job in due {
+        match job.action {
+            JobAction::Agent => {
+                let msg = InboundMsg {
+                    chat_id: job.chat_id,
+                    user_id: 0,
+                    text: job.message.clone(),
+                    channel: "cron".to_string(),
+                };
+                if inbound_tx.try_send(msg).is_err() {
+                    eprintln!("cron runner: inbound channel full, dropping agent job {}", job.id);
+                }
+            }
+            JobAction::Direct => {
+                let msg = OutboundMsg {
+                    chat_id: job.chat_id,
+                    text: job.message.clone(),
+                    channel: "cron".to_string(),
+                };
+                if outbound_tx.try_send(msg).is_err() {
+                    eprintln!("cron runner: outbound channel full, dropping direct job {}", job.id);
+                }
+            }
+        }
+        store.mark_fired(&job.id, now);
+    }
+}
+
+async fn tick_loop(
+    store: Arc<CronStore>,
+    inbound_tx: mpsc::Sender<InboundMsg>,
+    outbound_tx: mpsc::Sender<OutboundMsg>,
+    tick_secs: u64,
+) {
+    let mut interval = tokio::time::interval(std::time::Duration::from_secs(tick_secs));
+    interval.tick().await;
+    loop {
+        interval.tick().await;
+        let now = unix_now();
+        tick_once(&store, &inbound_tx, &outbound_tx, now).await;
+    }
+}
+
+/// Spawns the cron runner task. Returns the join handle (caller may ignore).
+pub fn spawn_cron_runner(
+    store: Arc<CronStore>,
+    inbound_tx: mpsc::Sender<InboundMsg>,
+    outbound_tx: mpsc::Sender<OutboundMsg>,
+    tick_interval_secs: u64,
+) -> tokio::task::JoinHandle<()> {
+    tokio::spawn(async move {
+        tick_loop(store, inbound_tx, outbound_tx, tick_interval_secs).await;
+    })
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use crate::tools::cron::{CronStore, Schedule};
+
+    #[tokio::test]
+    async fn tick_fires_due_direct_job() {
+        let dir = std::env::temp_dir().join("icrab_cron_runner_direct");
+        let _ = std::fs::remove_dir_all(&dir);
+        std::fs::create_dir_all(&dir).unwrap();
+        let store = CronStore::empty(&dir);
+        store
+            .add(
+                None,
+                "Reminder".to_string(),
+                JobAction::Direct,
+                Schedule::Once { at_unix: 100 },
+                12345,
+            )
+            .unwrap();
+        let (inbound_tx, _inbound_rx) = mpsc::channel(8);
+        let (outbound_tx, mut outbound_rx) = mpsc::channel(8);
+        tick_once(&store, &inbound_tx, &outbound_tx, 500).await;
+        let msg = outbound_rx.try_recv().unwrap();
+        assert_eq!(msg.chat_id, 12345);
+        assert_eq!(msg.text, "Reminder");
+        assert_eq!(msg.channel, "cron");
+        let job = store.get("job-1").unwrap();
+        assert!(job.last_run.is_some());
+        assert!(!job.enabled);
+        let _ = std::fs::remove_dir_all(&dir);
+    }
+
+    #[tokio::test]
+    async fn tick_fires_due_agent_job() {
+        let dir = std::env::temp_dir().join("icrab_cron_runner_agent");
+        let _ = std::fs::remove_dir_all(&dir);
+        std::fs::create_dir_all(&dir).unwrap();
+        let store = CronStore::empty(&dir);
+        store
+            .add(
+                None,
+                "Agent task".to_string(),
+                JobAction::Agent,
+                Schedule::Once { at_unix: 100 },
+                999,
+            )
+            .unwrap();
+        let (inbound_tx, mut inbound_rx) = mpsc::channel(8);
+        let (outbound_tx, _outbound_rx) = mpsc::channel(8);
+        tick_once(&store, &inbound_tx, &outbound_tx, 500).await;
+        let msg = inbound_rx.try_recv().unwrap();
+        assert_eq!(msg.chat_id, 999);
+        assert_eq!(msg.text, "Agent task");
+        assert_eq!(msg.channel, "cron");
+        assert_eq!(msg.user_id, 0);
+        let _ = std::fs::remove_dir_all(&dir);
+    }
+
+    #[tokio::test]
+    async fn tick_skips_not_due() {
+        let dir = std::env::temp_dir().join("icrab_cron_runner_skip");
+        let _ = std::fs::remove_dir_all(&dir);
+        std::fs::create_dir_all(&dir).unwrap();
+        let store = CronStore::empty(&dir);
+        store
+            .add(
+                None,
+                "Later".to_string(),
+                JobAction::Direct,
+                Schedule::Once { at_unix: 99999 },
+                1,
+            )
+            .unwrap();
+        let (inbound_tx, _inbound_rx) = mpsc::channel(8);
+        let (outbound_tx, mut outbound_rx) = mpsc::channel(8);
+        tick_once(&store, &inbound_tx, &outbound_tx, 500).await;
+        assert!(outbound_rx.try_recv().is_err());
+        let _ = std::fs::remove_dir_all(&dir);
+    }
+}

src/main.rs

@@ -5,12 +5,16 @@
 use std::path::PathBuf;
 use std::sync::Arc;
 
+use tokio::sync::mpsc;
+
 use icrab::agent;
 use icrab::agent::subagent_manager::SubagentManager;
 use icrab::config;
+use icrab::cron_runner;
 use icrab::llm::HttpProvider;
 use icrab::telegram::{self, OutboundMsg};
 use icrab::tools;
+use icrab::tools::cron::{CronStore, CronTool};
 use icrab::tools::spawn::SpawnTool;
 use icrab::tools::subagent::SubagentTool;
 
@@ -57,14 +61,30 @@ async fn main() {
         SUBAGENT_MAX_ITERATIONS,
     ));
 
-    // Main registry: core + spawn tool.
+    // Main registry: core + spawn + cron (cron is main-agent-only).
     let registry = tools::build_core_registry(&cfg);
     registry.register(SpawnTool::new(Arc::clone(&manager)));
     registry.register(SubagentTool::new(Arc::clone(&manager)));
 
-    let (mut inbound_rx, outbound_tx) = telegram::spawn_telegram(&cfg);
+    let (inbound_tx, mut inbound_rx) = mpsc::channel(64);
+    let outbound_tx = telegram::spawn_telegram(&cfg, inbound_tx.clone());
     eprintln!("Telegram poller and sender started");
 
+    let cron_store = Arc::new(
+        CronStore::load(&workspace).unwrap_or_else(|e| {
+            eprintln!("cron store: {}", e);
+            CronStore::empty(&workspace)
+        }),
+    );
+    cron_runner::spawn_cron_runner(
+        Arc::clone(&cron_store),
+        inbound_tx.clone(),
+        outbound_tx.clone(),
+        60,
+    );
+    registry.register(CronTool::new(Arc::clone(&cron_store)));
+    drop(inbound_tx);
+
     while let Some(msg) = inbound_rx.recv().await {
         let tool_ctx = tools::ToolCtx {
             workspace: workspace.clone(),

src/telegram.rs

@@ -321,18 +321,21 @@ async fn send_loop(client: TelegramClient, mut outbound_rx: mpsc::Receiver<Outbo
     }
 }
 
-/// Spawns the Telegram poll task and send task; returns channels for main/agent.
+/// Spawns the Telegram poll task and send task; returns outbound sender.
 ///
-/// Main holds `inbound_rx` and `outbound_tx`. Poll loop pushes allowed user messages to inbound;
-/// main/agent sends replies via outbound_tx. Shutdown in v1: process kill; later add cancel token.
-pub fn spawn_telegram(config: &Config) -> (mpsc::Receiver<InboundMsg>, mpsc::Sender<OutboundMsg>) {
+/// Caller creates the inbound channel and passes `inbound_tx` so other producers (e.g. cron runner)
+/// can inject messages. Poll loop pushes allowed user messages to inbound; main/agent sends
+/// replies via returned outbound_tx. Shutdown in v1: process kill; later add cancel token.
+pub fn spawn_telegram(
+    config: &Config,
+    inbound_tx: mpsc::Sender<InboundMsg>,
+) -> mpsc::Sender<OutboundMsg> {
     let telegram = config.telegram.as_ref().expect("config validated");
     let bot_token = telegram.bot_token.clone().expect("config validated");
     let allowed_user_ids = telegram.allowed_user_ids.clone();
     let api_base = telegram.api_base.as_deref();
 
     let client = TelegramClient::with_base_url(&bot_token, api_base);
-    let (inbound_tx, inbound_rx) = mpsc::channel(CHANNEL_CAP);
     let (outbound_tx, outbound_rx) = mpsc::channel(CHANNEL_CAP);
 
     let poll_client = TelegramClient {
@@ -347,5 +350,5 @@ pub fn spawn_telegram(config: &Config) -> (mpsc::Receiver<InboundMsg>, mpsc::Sen
         send_loop(client, outbound_rx).await;
     });
 
-    (inbound_rx, outbound_tx)
+    outbound_tx
 }