feat: add cron runner worker

Author: rexleimo

Cargo.lock

@@ -4,6 +4,6 @@ mod commands;
 mod tests;
 
 pub(crate) use commands::{
-    AcpCommand, AgentCommand, AgentKind, ChannelCommand, Cli, Command, ConfigCommand,
+    AcpCommand, AgentCommand, AgentKind, ChannelCommand, Cli, Command, ConfigCommand, CronCommand,
     DaemonCommand, HarnessCommand, ReleaseCommand, SkillsCommand,
 };

crates/loopforge-cli/src/cli.rs

@@ -2,6 +2,7 @@ mod acp;
 mod agent;
 mod channel;
 mod config;
+mod cron;
 mod daemon;
 mod harness;
 mod release;
@@ -16,6 +17,7 @@ pub(crate) use acp::AcpCommand;
 pub(crate) use agent::{AgentCommand, AgentKind};
 pub(crate) use channel::ChannelCommand;
 pub(crate) use config::ConfigCommand;
+pub(crate) use cron::CronCommand;
 pub(crate) use daemon::DaemonCommand;
 pub(crate) use harness::HarnessCommand;
 pub(crate) use release::ReleaseCommand;
@@ -76,6 +78,11 @@ pub(crate) enum Command {
         #[command(subcommand)]
         command: ChannelCommand,
     },
+    /// Cron scheduler helpers (stored jobs + optional runner)
+    Cron {
+        #[command(subcommand)]
+        command: CronCommand,
+    },
     /// ACP event/checkpoint inspection helpers
     Acp {
         #[command(subcommand)]

crates/loopforge-cli/src/cli/commands.rs

@@ -0,0 +1,42 @@
+#[derive(Debug, clap::Subcommand)]
+pub(crate) enum CronCommand {
+    /// Run one cron scheduler tick (process due jobs once)
+    Tick {
+        /// Max due jobs to execute in this tick
+        #[arg(long, default_value_t = 20)]
+        max_due_per_tick: usize,
+        /// Max catch-up slots per job when recovering after downtime
+        #[arg(long, default_value_t = 25)]
+        max_catchup_slots_per_job: u32,
+        /// Auto-disable a job after this many consecutive failures
+        #[arg(long, default_value_t = 5)]
+        max_consecutive_errors: u32,
+        /// Minimum delay (seconds) before retrying a failing job
+        #[arg(long, default_value_t = 30)]
+        min_retry_delay_secs: u64,
+        /// Per-job execution timeout (milliseconds)
+        #[arg(long, default_value_t = 10_000)]
+        job_timeout_ms: u64,
+    },
+    /// Run a long-lived cron worker loop
+    Worker {
+        /// Seconds between scheduler ticks
+        #[arg(long, default_value_t = 2)]
+        interval_secs: u64,
+        /// Max due jobs to execute per tick
+        #[arg(long, default_value_t = 20)]
+        max_due_per_tick: usize,
+        /// Max catch-up slots per job when recovering after downtime
+        #[arg(long, default_value_t = 25)]
+        max_catchup_slots_per_job: u32,
+        /// Auto-disable a job after this many consecutive failures
+        #[arg(long, default_value_t = 5)]
+        max_consecutive_errors: u32,
+        /// Minimum delay (seconds) before retrying a failing job
+        #[arg(long, default_value_t = 30)]
+        min_retry_delay_secs: u64,
+        /// Per-job execution timeout (milliseconds)
+        #[arg(long, default_value_t = 10_000)]
+        job_timeout_ms: u64,
+    },
+}

crates/loopforge-cli/src/cli/commands/cron.rs

@@ -144,3 +144,21 @@ fn cli_parses_onboard_starter_profile() {
         "expected `loopforge onboard --starter workspace-brief` to parse, got: {parsed:?}"
     );
 }
+
+#[test]
+fn cli_parses_cron_tick_subcommand() {
+    let parsed = Cli::try_parse_from(["loopforge", "cron", "tick"]);
+    assert!(
+        parsed.is_ok(),
+        "expected `loopforge cron tick` to parse, got: {parsed:?}"
+    );
+}
+
+#[test]
+fn cli_parses_cron_worker_subcommand() {
+    let parsed = Cli::try_parse_from(["loopforge", "cron", "worker"]);
+    assert!(
+        parsed.is_ok(),
+        "expected `loopforge cron worker` to parse, got: {parsed:?}"
+    );
+}

crates/loopforge-cli/src/cli/tests.rs

@@ -2,6 +2,7 @@ mod acp;
 mod agent;
 mod channel;
 mod config;
+mod cron;
 mod daemon;
 mod doctor;
 mod harness;
@@ -31,6 +32,7 @@ pub(crate) async fn run(cli: Cli) -> anyhow::Result<()> {
         } => doctor::run(json, strict, timeout_ms).await,
         Command::Agent { command } => agent::run(command).await,
         Command::Channel { command } => channel::run(command).await,
+        Command::Cron { command } => cron::run(command).await,
         Command::Acp { command } => acp::run(command),
         Command::Config { command } => config::run(command),
         Command::Skills { command } => skills::run(command).await,

crates/loopforge-cli/src/dispatch.rs

@@ -0,0 +1,54 @@
+use crate::{cli::CronCommand, runtime_env};
+
+pub(super) async fn run(command: CronCommand) -> anyhow::Result<()> {
+    let (_paths, agent) = runtime_env::load_agent_runtime()?;
+
+    match command {
+        CronCommand::Tick {
+            max_due_per_tick,
+            max_catchup_slots_per_job,
+            max_consecutive_errors,
+            min_retry_delay_secs,
+            job_timeout_ms,
+        } => {
+            let config = rexos::agent::CronRunnerConfig {
+                tick_interval_secs: 1,
+                max_due_per_tick,
+                max_catchup_slots_per_job,
+                max_consecutive_errors,
+                min_retry_delay_secs,
+                job_timeout_ms,
+            };
+            let summary = agent.cron_runner_tick(&config).await?;
+            println!(
+                "cron: tick_at={} due={} ran={} ok={} failed={} skipped={}",
+                summary.tick_at,
+                summary.due,
+                summary.ran,
+                summary.ok,
+                summary.failed,
+                summary.skipped
+            );
+            Ok(())
+        }
+        CronCommand::Worker {
+            interval_secs,
+            max_due_per_tick,
+            max_catchup_slots_per_job,
+            max_consecutive_errors,
+            min_retry_delay_secs,
+            job_timeout_ms,
+        } => {
+            let config = rexos::agent::CronRunnerConfig {
+                tick_interval_secs: interval_secs,
+                max_due_per_tick,
+                max_catchup_slots_per_job,
+                max_consecutive_errors,
+                min_retry_delay_secs,
+                job_timeout_ms,
+            };
+            agent.cron_runner_loop(config).await?;
+            Ok(())
+        }
+    }
+}

crates/loopforge-cli/src/dispatch/cron.rs

@@ -1,4 +1,4 @@
-use rusqlite::OptionalExtension;
+use rusqlite::{OptionalExtension, Transaction, TransactionBehavior};
 
 use crate::MemoryStore;
 
@@ -20,4 +20,32 @@ impl MemoryStore {
             .optional()?;
         Ok(value)
     }
+
+    pub fn kv_update<F>(&self, key: &str, f: F) -> anyhow::Result<Option<String>>
+    where
+        F: FnOnce(Option<String>) -> anyhow::Result<Option<String>>,
+    {
+        let tx = Transaction::new_unchecked(&self.conn, TransactionBehavior::Immediate)?;
+        let current = tx
+            .query_row("SELECT value FROM kv WHERE key=?1", (key,), |row| {
+                row.get(0)
+            })
+            .optional()?;
+
+        let next = f(current)?;
+        match &next {
+            Some(value) => {
+                tx.execute(
+                    "INSERT INTO kv (key, value) VALUES (?1, ?2)\n            ON CONFLICT(key) DO UPDATE SET value=excluded.value",
+                    (key, value),
+                )?;
+            }
+            None => {
+                tx.execute("DELETE FROM kv WHERE key=?1", (key,))?;
+            }
+        }
+
+        tx.commit()?;
+        Ok(next)
+    }
 }

crates/rexos-memory/src/kv.rs

@@ -17,3 +17,6 @@ rexos-kernel = { path = "../rexos-kernel" }
 rexos-llm = { path = "../rexos-llm" }
 rexos-memory = { path = "../rexos-memory" }
 rexos-tools = { path = "../rexos-tools" }
+
+[dev-dependencies]
+tempfile.workspace = true

crates/rexos-runtime/Cargo.toml

@@ -53,6 +53,7 @@ pub struct AgentRuntime {
 }
 
 pub use outbox::{OutboxDispatcher, OutboxDrainSummary};
+pub use scheduling::runner::{CronRunnerConfig, CronRunnerTickSummary};
 
 impl AgentRuntime {
     pub fn new(memory: MemoryStore, llms: LlmRegistry, router: ModelRouter) -> Self {