Skip to main content

agentos_client/
cron.rs

1//! Cron scheduling + the `CronManager`.
2//!
3//! Ported from `packages/core/src/cron/`. The `schedule` is a 5/6/7-field cron expression (croner
4//! grammar) or an ISO-8601 one-shot timestamp. `CronAction::Callback` is in-process only
5//! (non-serializable). `on_cron_event` returns NO unsubscribe in TS; the Rust equivalent is a
6//! [`tokio::sync::broadcast::Receiver`] whose drop is the unsubscribe.
7//!
8//! Timing is owned by the [`ScheduleDriver`] (mirroring TS `CronManager.schedule` delegating to
9//! `this.driver.schedule({...})`). The default [`crate::config::TimerScheduleDriver`] parses the
10//! schedule, arms the timer, reschedules cron after each fire, and tears down on dispose. The manager
11//! itself only registers job state and runs `execute_job` when the driver fires the callback.
12//!
13//! Cron fields are interpreted in the host LOCAL timezone, matching croner's default behavior.
14
15use std::collections::BTreeMap;
16use std::sync::atomic::{AtomicBool, Ordering};
17use std::sync::Arc;
18
19use chrono::{
20    DateTime, Datelike, Duration as ChronoDuration, Local, SecondsFormat, Timelike, Utc, Weekday,
21};
22use scc::HashMap as SccHashMap;
23use serde::{Deserialize, Serialize};
24use tokio::sync::broadcast;
25
26use crate::agent_os::AgentOs;
27use crate::config::{ScheduleDriver, ScheduleEntry, ScheduleHandle};
28use crate::error::ClientError;
29use crate::session::{McpServerConfig, OpenSessionInput, PermissionPolicy, PromptInput};
30
31// ---------------------------------------------------------------------------
32// Supporting types
33// ---------------------------------------------------------------------------
34
35/// Overlap policy for a cron job.
36#[derive(Debug, Clone, Copy, PartialEq, Eq, Default, Serialize, Deserialize)]
37#[serde(rename_all = "lowercase")]
38pub enum CronOverlap {
39    #[default]
40    Allow,
41    Skip,
42    Queue,
43}
44
45/// A cron action. `Callback` holds an in-process closure and cannot cross the wire.
46#[derive(Clone)]
47pub enum CronAction {
48    /// Open a fresh durable session, prompt it, then delete it.
49    Session {
50        agent_type: String,
51        prompt: String,
52        options: Option<CronSessionOptions>,
53    },
54    /// Run a command via `exec`.
55    Exec { command: String, args: Vec<String> },
56    /// Invoke a host-side callback.
57    Callback {
58        #[allow(clippy::type_complexity)]
59        callback: Arc<dyn Fn() -> futures::future::BoxFuture<'static, ()> + Send + Sync>,
60    },
61}
62
63/// Durable session options accepted by a cron action. The action owns the
64/// agent name and generates a unique session ID for each run.
65#[derive(Debug, Clone, Default, PartialEq, Serialize, Deserialize)]
66#[serde(rename_all = "camelCase")]
67pub struct CronSessionOptions {
68    pub cwd: Option<String>,
69    pub additional_directories: Option<Vec<String>>,
70    pub env: Option<BTreeMap<String, String>>,
71    pub mcp_servers: Option<Vec<McpServerConfig>>,
72    pub permission_policy: Option<PermissionPolicy>,
73    pub skip_os_instructions: Option<bool>,
74    pub additional_instructions: Option<String>,
75}
76
77/// Serializable description of a scheduled action. Callback jobs deliberately
78/// expose only their kind: the host closure is execution state, not job data.
79#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
80#[serde(tag = "type", rename_all = "lowercase")]
81pub enum CronActionInfo {
82    Session {
83        #[serde(rename = "agentType")]
84        agent_type: String,
85        prompt: String,
86        #[serde(default, skip_serializing_if = "Option::is_none")]
87        options: Option<CronSessionOptions>,
88    },
89    Exec {
90        command: String,
91        #[serde(default, skip_serializing_if = "Vec::is_empty")]
92        args: Vec<String>,
93    },
94    Callback,
95}
96
97impl From<&CronAction> for CronActionInfo {
98    fn from(action: &CronAction) -> Self {
99        match action {
100            CronAction::Session {
101                agent_type,
102                prompt,
103                options,
104            } => Self::Session {
105                agent_type: agent_type.clone(),
106                prompt: prompt.clone(),
107                options: options.clone(),
108            },
109            CronAction::Exec { command, args } => Self::Exec {
110                command: command.clone(),
111                args: args.clone(),
112            },
113            CronAction::Callback { .. } => Self::Callback,
114        }
115    }
116}
117
118impl std::fmt::Debug for CronAction {
119    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
120        match self {
121            CronAction::Session {
122                agent_type, prompt, ..
123            } => f
124                .debug_struct("Session")
125                .field("agent_type", agent_type)
126                .field("prompt", prompt)
127                .finish_non_exhaustive(),
128            CronAction::Exec { command, args } => f
129                .debug_struct("Exec")
130                .field("command", command)
131                .field("args", args)
132                .finish(),
133            CronAction::Callback { .. } => f.debug_struct("Callback").finish_non_exhaustive(),
134        }
135    }
136}
137
138/// Options for `schedule_cron`.
139#[derive(Clone)]
140pub struct CronJobOptions {
141    /// Default: a fresh UUID.
142    pub id: Option<String>,
143    /// 5/6/7-field cron expression OR an ISO-8601 one-shot timestamp.
144    pub schedule: String,
145    pub action: CronAction,
146    /// Default: [`CronOverlap::Allow`].
147    pub overlap: Option<CronOverlap>,
148}
149
150/// Snapshot info for a cron job.
151#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
152#[serde(rename_all = "camelCase")]
153pub struct CronJobInfo {
154    pub id: String,
155    pub schedule: String,
156    pub action: CronActionInfo,
157    pub overlap: CronOverlap,
158    pub last_run: Option<String>,
159    pub next_run: Option<String>,
160    pub run_count: u64,
161    pub running: bool,
162}
163
164/// A cron event emitted on each run.
165#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
166#[serde(tag = "type")]
167pub enum CronEvent {
168    #[serde(rename = "cron:fire", rename_all = "camelCase")]
169    Fire { job_id: String, time: String },
170    #[serde(rename = "cron:complete", rename_all = "camelCase")]
171    Complete {
172        job_id: String,
173        time: String,
174        duration_ms: f64,
175    },
176    #[serde(rename = "cron:error", rename_all = "camelCase")]
177    Error {
178        job_id: String,
179        time: String,
180        error: String,
181    },
182}
183
184/// Handle to a scheduled cron job. Dropping or calling [`CronJobHandle::cancel`] cancels it.
185#[derive(Clone)]
186pub struct CronJobHandle {
187    pub id: String,
188    pub(crate) manager: Arc<CronManager>,
189}
190
191impl CronJobHandle {
192    /// Cancel the job (no-op if already cancelled/unknown).
193    pub fn cancel(&self) {
194        self.manager.cancel_job(&self.id);
195    }
196}
197
198// ---------------------------------------------------------------------------
199// CronManager + CronJobState
200// ---------------------------------------------------------------------------
201
202/// Internal per-job state.
203pub(crate) struct CronJobState {
204    pub schedule: String,
205    pub action: CronAction,
206    pub overlap: CronOverlap,
207    pub last_run: parking_lot::Mutex<Option<DateTime<Utc>>>,
208    pub next_run: parking_lot::Mutex<Option<DateTime<Utc>>>,
209    pub run_count: std::sync::atomic::AtomicU64,
210    pub running: AtomicBool,
211    /// Set when a `Queue`-policy fire arrives while the job is already running; drained to exactly
212    /// one deferred run when the active run completes. Mirrors TS `CronJobState.queued`.
213    pub queued: AtomicBool,
214    /// Driver-returned timer handle. Used by `cancel`/`dispose` to tear down the armed timer through
215    /// the driver, mirroring TS `this.driver.cancel(state.handle)`.
216    pub handle: ScheduleHandle,
217}
218
219/// Owns scheduled jobs, the schedule driver, and the cron event broadcast.
220pub struct CronManager {
221    pub(crate) jobs: SccHashMap<String, CronJobState>,
222    pub(crate) schedule_lock: parking_lot::Mutex<()>,
223    pub(crate) driver: Arc<dyn ScheduleDriver>,
224    pub(crate) event_tx: broadcast::Sender<CronEvent>,
225}
226
227impl CronManager {
228    /// Create a cron manager with the given schedule driver.
229    pub(crate) fn new(driver: Arc<dyn ScheduleDriver>) -> Self {
230        let (event_tx, _rx) = broadcast::channel(256);
231        Self {
232            jobs: SccHashMap::new(),
233            schedule_lock: parking_lot::Mutex::new(()),
234            driver,
235            event_tx,
236        }
237    }
238
239    /// Cancel a job by id (no-op if unknown).
240    ///
241    /// Mirrors TS `CronManager.cancel`: cancel the driver-armed timer (`this.driver.cancel(handle)`)
242    /// and remove the job from the registry.
243    pub(crate) fn cancel_job(&self, id: &str) {
244        let _guard = self.schedule_lock.lock();
245        if let Some((_, state)) = self.jobs.remove(id) {
246            self.driver.cancel(&state.handle);
247        }
248    }
249
250    /// Dispose all jobs (called during shutdown).
251    ///
252    /// Mirrors TS `CronManager.dispose`: cancel every armed timer through the driver, clear the
253    /// registry, then call `this.driver.dispose()` to tear down all driver-held timer state.
254    pub(crate) fn dispose(&self) {
255        let _guard = self.schedule_lock.lock();
256        self.jobs.scan(|_, state| {
257            self.driver.cancel(&state.handle);
258        });
259        self.jobs.clear();
260        self.driver.dispose();
261    }
262}
263
264/// Execute a single job run, honoring the overlap policy. Emits `Fire`, then `Complete` or `Error`.
265/// Re-runs once at the end if a `Queue`-policy run was deferred while busy.
266///
267/// Mirrors TS `CronManager.executeJob`. Handler/action errors never crash the manager; on error a
268/// `cron:error` event is emitted instead of a `cron:complete`. Returns an explicitly boxed `Send`
269/// future (rather than an `async fn`) so the recursive queued re-run does not form a
270/// self-referential async auto-trait inference cycle that would defeat the `Send` bound required by
271/// [`tokio::spawn`].
272fn execute_job(
273    manager: Arc<CronManager>,
274    vm: AgentOs,
275    id: String,
276) -> futures::future::BoxFuture<'static, ()> {
277    Box::pin(execute_job_inner(manager, vm, id))
278}
279
280async fn execute_job_inner(manager: Arc<CronManager>, vm: AgentOs, id: String) {
281    let manager = &manager;
282    let vm = &vm;
283    let id = id.as_str();
284    // Overlap policy: a running job either allows a concurrent run, skips this fire, or queues
285    // exactly one deferred run.
286    {
287        let mut should_return = false;
288        let mut should_queue = false;
289        manager.jobs.read(id, |_, state| {
290            if state.running.load(Ordering::SeqCst) {
291                match state.overlap {
292                    CronOverlap::Allow => {}
293                    CronOverlap::Skip => should_return = true,
294                    CronOverlap::Queue => should_queue = true,
295                }
296            }
297        });
298        if should_return {
299            return;
300        }
301        if should_queue {
302            manager.jobs.read(id, |_, state| {
303                state.queued.store(true, Ordering::SeqCst);
304            });
305            return;
306        }
307    }
308
309    // Mark running, record this run, and snapshot the action to dispatch.
310    let action = match manager.jobs.read(id, |_, state| {
311        state.running.store(true, Ordering::SeqCst);
312        *state.last_run.lock() = Some(Utc::now());
313        state.run_count.fetch_add(1, Ordering::SeqCst);
314        state.action.clone()
315    }) {
316        Some(action) => action,
317        None => return,
318    };
319
320    let _ = manager.event_tx.send(CronEvent::Fire {
321        job_id: id.to_string(),
322        time: Utc::now().to_rfc3339_opts(SecondsFormat::Millis, true),
323    });
324
325    // TS `durationMs = Date.now() - startTime`, an integer millisecond count.
326    let start = Utc::now();
327    let result = run_action(vm, &action).await;
328    let duration_ms = (Utc::now() - start).num_milliseconds() as f64;
329
330    match result {
331        Ok(()) => {
332            let _ = manager.event_tx.send(CronEvent::Complete {
333                job_id: id.to_string(),
334                time: Utc::now().to_rfc3339_opts(SecondsFormat::Millis, true),
335                duration_ms,
336            });
337        }
338        Err(error) => {
339            let _ = manager.event_tx.send(CronEvent::Error {
340                job_id: id.to_string(),
341                time: Utc::now().to_rfc3339_opts(SecondsFormat::Millis, true),
342                error: error.to_string(),
343            });
344        }
345    }
346
347    // Clear running, recompute the next run, and drain a queued run if one was deferred.
348    let mut run_queued = false;
349    manager.jobs.read(id, |_, state| {
350        state.running.store(false, Ordering::SeqCst);
351        *state.next_run.lock() = compute_next_time(&state.schedule, Utc::now());
352        if state.queued.swap(false, Ordering::SeqCst) {
353            run_queued = true;
354        }
355    });
356
357    if run_queued {
358        let manager = Arc::clone(manager);
359        let vm = vm.clone();
360        let id = id.to_string();
361        tokio::spawn(execute_job(manager, vm, id));
362    }
363}
364
365/// Dispatch a [`CronAction`]. Mirrors TS `CronManager.runAction`.
366///
367/// `Session` opens a session, prompts it, and always deletes it (even if the prompt errors, the
368/// delete still runs, matching the TS `finally`). `Exec` sends the structured `(command, args)` argv
369/// verbatim via [`AgentOs::exec_argv`] (no string flattening / re-parsing). `Callback` awaits the
370/// in-process future.
371async fn run_action(vm: &AgentOs, action: &CronAction) -> Result<(), ClientError> {
372    match action {
373        CronAction::Session {
374            agent_type,
375            prompt,
376            options,
377        } => {
378            let options = options.clone().unwrap_or_default();
379            let session_id = format!("cron-{}", uuid::Uuid::new_v4());
380            vm.open_session(OpenSessionInput {
381                session_id: Some(session_id.clone()),
382                agent: agent_type.clone(),
383                cwd: options.cwd,
384                additional_directories: options.additional_directories,
385                env: options.env,
386                mcp_servers: options.mcp_servers,
387                permission_policy: options.permission_policy,
388                skip_os_instructions: options.skip_os_instructions,
389                additional_instructions: options.additional_instructions,
390            })
391            .await
392            .map_err(|err| ClientError::Sidecar(err.to_string()))?;
393            let content = serde_json::from_value(serde_json::json!({
394                "type": "text",
395                "text": prompt,
396            }))
397            .map_err(|err| ClientError::Sidecar(err.to_string()))?;
398            let prompt_result = vm
399                .prompt(PromptInput {
400                    session_id: Some(session_id.clone()),
401                    idempotency_key: None,
402                    content: vec![content],
403                })
404                .await;
405            // Always delete this per-run session so cron does not grow the durable catalog.
406            let delete_result = vm.delete_session(Some(&session_id)).await;
407            match (prompt_result, delete_result) {
408                (Ok(_), Ok(())) => Ok(()),
409                (Err(prompt_error), Ok(())) => Err(ClientError::Sidecar(prompt_error.to_string())),
410                (Ok(_), Err(delete_error)) => Err(ClientError::Sidecar(format!(
411                    "cron prompt completed but durable session cleanup failed: {delete_error}"
412                ))),
413                (Err(prompt_error), Err(delete_error)) => {
414                    eprintln!(
415                        "ERR_AGENTOS_CRON_SESSION_CLEANUP: prompt failed with {prompt_error}; durable session cleanup also failed: {delete_error}"
416                    );
417                    Err(ClientError::Sidecar(prompt_error.to_string()))
418                }
419            }
420        }
421        CronAction::Exec { command, args } => {
422            // Send the structured argv verbatim. Flattening `command`/`args` into a single string
423            // and re-parsing it through the `exec` command-line parser would re-split argv elements
424            // on whitespace and shell-evaluate any `$()`/backtick content; `exec_argv` preserves the
425            // structured (command, args) contract element-for-element.
426            vm.exec_argv(command, args, crate::process::ExecOptions::default())
427                .await
428                .map_err(|err| ClientError::Sidecar(err.to_string()))?;
429            Ok(())
430        }
431        CronAction::Callback { callback } => {
432            callback().await;
433            Ok(())
434        }
435    }
436}
437
438// ---------------------------------------------------------------------------
439// Schedule validation
440// ---------------------------------------------------------------------------
441
442/// A parsed schedule: either a recurring cron expression or a one-shot ISO-8601 timestamp.
443///
444/// Mirrors TS `ParsedSchedule` (`parse-schedule.ts`).
445pub(crate) enum ParsedSchedule {
446    /// A one-shot absolute timestamp.
447    Date(DateTime<Utc>),
448    /// A recurring cron expression (croner grammar).
449    Cron(CronExpr),
450}
451
452impl ParsedSchedule {
453    /// `true` for a recurring cron expression. Mirrors TS `parsed.kind === "cron"`.
454    pub(crate) fn is_cron(&self) -> bool {
455        matches!(self, ParsedSchedule::Cron(_))
456    }
457}
458
459/// Resolve the next run for an already-parsed schedule strictly after `now`. Mirrors TS
460/// `resolveSchedule(...).nextRun`: a cron yields `cron.nextRun()`; a one-shot date yields the date if
461/// it is in the future, else `None`.
462pub(crate) fn resolve_next_run(
463    parsed: &ParsedSchedule,
464    now: DateTime<Utc>,
465) -> Option<DateTime<Utc>> {
466    match parsed {
467        ParsedSchedule::Cron(cron) => cron.next_after(now),
468        ParsedSchedule::Date(date) => {
469            if date.timestamp_millis() > now.timestamp_millis() {
470                Some(*date)
471            } else {
472                None
473            }
474        }
475    }
476}
477
478/// Decide whether a schedule string looks like a one-shot ISO-8601-ish timestamp rather than a cron
479/// expression. Mirrors TS `looksLikeOneShotSchedule` /
480/// `^\d{4}-\d{2}-\d{2}(?:[T ]\d{2}:\d{2}(?::\d{2}(?:\.\d{1,3})?)?(?:Z|[+-]\d{2}:\d{2})?)?$`, with the
481/// fractional-seconds group widened to accept any number of digits so a Rust-produced RFC-3339
482/// timestamp (up to 9 fractional digits) is recognized as a one-shot.
483fn looks_like_one_shot(schedule: &str) -> bool {
484    let bytes = schedule.as_bytes();
485    let mut i = 0usize;
486
487    let is_digit = |b: u8| b.is_ascii_digit();
488
489    let take_digits = |bytes: &[u8], i: &mut usize, n: usize| -> bool {
490        for _ in 0..n {
491            match bytes.get(*i) {
492                Some(&b) if is_digit(b) => *i += 1,
493                _ => return false,
494            }
495        }
496        true
497    };
498    let take_lit = |bytes: &[u8], i: &mut usize, lit: u8| -> bool {
499        match bytes.get(*i) {
500            Some(&b) if b == lit => {
501                *i += 1;
502                true
503            }
504            _ => false,
505        }
506    };
507
508    if !take_digits(bytes, &mut i, 4) {
509        return false;
510    }
511    if !take_lit(bytes, &mut i, b'-') {
512        return false;
513    }
514    if !take_digits(bytes, &mut i, 2) {
515        return false;
516    }
517    if !take_lit(bytes, &mut i, b'-') {
518        return false;
519    }
520    if !take_digits(bytes, &mut i, 2) {
521        return false;
522    }
523
524    // Optional time portion: [T ]HH:MM(:SS(.fff)?)?(Z|[+-]HH:MM)?
525    if i == bytes.len() {
526        return true;
527    }
528    match bytes.get(i) {
529        Some(b'T') | Some(b' ') => i += 1,
530        _ => return false,
531    }
532    if !take_digits(bytes, &mut i, 2) {
533        return false;
534    }
535    if !take_lit(bytes, &mut i, b':') {
536        return false;
537    }
538    if !take_digits(bytes, &mut i, 2) {
539        return false;
540    }
541
542    // Optional :SS
543    if take_lit(bytes, &mut i, b':') {
544        if !take_digits(bytes, &mut i, 2) {
545            return false;
546        }
547        // Optional fractional seconds. The TS regex caps this at `\.\d{1,3}`, but a Rust-produced
548        // one-shot from `chrono::DateTime::to_rfc3339()` emits up to 9 fractional digits, so a valid
549        // near-future RFC-3339 timestamp must not be misclassified as a cron expression. Accept any
550        // run of one or more fractional digits.
551        if take_lit(bytes, &mut i, b'.') {
552            let mut frac = 0;
553            while matches!(bytes.get(i), Some(&b) if is_digit(b)) {
554                i += 1;
555                frac += 1;
556            }
557            if frac == 0 {
558                return false;
559            }
560        }
561    }
562
563    // Optional timezone: Z | [+-]HH:MM
564    match bytes.get(i) {
565        None => return true,
566        Some(b'Z') => {
567            i += 1;
568        }
569        Some(b'+') | Some(b'-') => {
570            i += 1;
571            if !take_digits(bytes, &mut i, 2) {
572                return false;
573            }
574            if !take_lit(bytes, &mut i, b':') {
575                return false;
576            }
577            if !take_digits(bytes, &mut i, 2) {
578                return false;
579            }
580        }
581        _ => return false,
582    }
583
584    i == bytes.len()
585}
586
587/// Parse a one-shot timestamp string into a UTC instant, matching ECMAScript `Date.parse` rules for
588/// the subset accepted by [`looks_like_one_shot`]:
589/// - a date-only string (`2026-06-04`) is UTC midnight;
590/// - a date-time string WITHOUT an offset (`2026-06-04T12:30`, `2026-06-04 12:30`) is parsed as LOCAL
591///   time;
592/// - forms with `Z` or an explicit numeric offset are parsed as written.
593fn parse_one_shot(schedule: &str) -> Option<DateTime<Utc>> {
594    use chrono::TimeZone;
595
596    // Try a full RFC-3339 timestamp first (handles Z and numeric offsets).
597    if let Ok(dt) = DateTime::parse_from_rfc3339(schedule) {
598        return Some(dt.with_timezone(&Utc));
599    }
600
601    // Normalize a space separator to `T` for the naive parsers below.
602    let normalized = schedule.replacen(' ', "T", 1);
603
604    // Date + time without a timezone: ECMAScript treats this as LOCAL time.
605    for fmt in [
606        "%Y-%m-%dT%H:%M:%S%.f",
607        "%Y-%m-%dT%H:%M:%S",
608        "%Y-%m-%dT%H:%M",
609    ] {
610        if let Ok(naive) = chrono::NaiveDateTime::parse_from_str(&normalized, fmt) {
611            return match Local.from_local_datetime(&naive) {
612                chrono::LocalResult::Single(dt) => Some(dt.with_timezone(&Utc)),
613                chrono::LocalResult::Ambiguous(dt, _) => Some(dt.with_timezone(&Utc)),
614                chrono::LocalResult::None => None,
615            };
616        }
617    }
618
619    // Date only: midnight UTC (ECMAScript date-only form is UTC).
620    if let Ok(date) = chrono::NaiveDate::parse_from_str(schedule, "%Y-%m-%d") {
621        let naive = date.and_hms_opt(0, 0, 0)?;
622        return Some(DateTime::<Utc>::from_naive_utc_and_offset(naive, Utc));
623    }
624
625    None
626}
627
628/// Parse a schedule string into a [`ParsedSchedule`]. Mirrors TS `parseSchedule`.
629pub(crate) fn parse_schedule(schedule: &str) -> std::result::Result<ParsedSchedule, ClientError> {
630    let normalized = schedule.trim();
631    if looks_like_one_shot(normalized) {
632        return match parse_one_shot(normalized) {
633            Some(date) => Ok(ParsedSchedule::Date(date)),
634            None => Err(ClientError::InvalidSchedule(schedule.to_string())),
635        };
636    }
637
638    match CronExpr::parse(normalized) {
639        Ok(cron) => Ok(ParsedSchedule::Cron(cron)),
640        Err(_) => Err(ClientError::InvalidSchedule(schedule.to_string())),
641    }
642}
643
644/// Compute the next fire time for a schedule string strictly after `now`. Returns `None` for a
645/// one-shot timestamp in the past or a cron expression with no upcoming match. Mirrors TS
646/// `computeNextTime` / `resolveSchedule(...).nextRun`.
647pub(crate) fn compute_next_time(schedule: &str, now: DateTime<Utc>) -> Option<DateTime<Utc>> {
648    let parsed = parse_schedule(schedule).ok()?;
649    resolve_next_run(&parsed, now)
650}
651
652/// Validate a schedule string. Returns the parsed next run for one-shot ISO-8601 schedules.
653///
654/// Errors `InvalidSchedule` for malformed input and `PastSchedule` for one-shot timestamps already
655/// in the past. Mirrors TS `validateScheduleForRegistration`: a one-shot timestamp that resolves to
656/// no next run is rejected as `PastSchedule`; cron expressions are accepted even when their next run
657/// is currently unknown.
658pub(crate) fn validate_schedule(
659    schedule: &str,
660    now: DateTime<Utc>,
661) -> std::result::Result<Option<DateTime<Utc>>, ClientError> {
662    let parsed = parse_schedule(schedule)?;
663    match parsed {
664        ParsedSchedule::Cron(cron) => Ok(cron.next_after(now)),
665        ParsedSchedule::Date(date) => {
666            if date.timestamp_millis() > now.timestamp_millis() {
667                Ok(Some(date))
668            } else {
669                Err(ClientError::PastSchedule(schedule.to_string()))
670            }
671        }
672    }
673}
674
675// ---------------------------------------------------------------------------
676// Cron expression parser + next-run search (croner-compatible grammar)
677// ---------------------------------------------------------------------------
678
679/// A parsed cron expression interpreted in the host LOCAL timezone (matching croner's default).
680///
681/// Implemented in-crate because the workspace has no cron-parsing dependency. Accepts the croner
682/// grammar: 5-field (`min hour dom month dow`), 6-field (leading `seconds`), and 7-field (leading
683/// `seconds`, trailing `year`) expressions; named months (`JAN`-`DEC`); named weekdays (`SUN`-`SAT`);
684/// `*`, ranges (`a-b`), steps (`*/n`, `a-b/n`, `a/n`), comma lists, `?` (treated as `*` for
685/// dom/dow), `L` (last day of month for dom, last weekday-of-month for dow), `#` (nth weekday), and
686/// `W` (nearest weekday to a day-of-month). Day-of-month and day-of-week combine with OR semantics
687/// when both are restricted, matching Vixie/croner.
688pub(crate) struct CronExpr {
689    seconds: Vec<u32>,
690    minutes: Vec<u32>,
691    hours: Vec<u32>,
692    days_of_month: Vec<u32>,
693    months: Vec<u32>,
694    days_of_week: Vec<u32>,
695    years: Option<Vec<u32>>,
696    dom_restricted: bool,
697    dow_restricted: bool,
698    /// Day-of-month `L` (last day of month).
699    dom_last: bool,
700    /// Day-of-month `LW` (last weekday, Mon-Fri, on or before the last day of the month).
701    dom_last_weekday: bool,
702    /// Day-of-month `<n>W` (nearest weekday to day `n`).
703    dom_nearest_weekday: Option<u32>,
704    /// Day-of-week `<weekday>L` (last given weekday of the month).
705    dow_last: Option<u32>,
706    /// Day-of-week `<weekday>#<n>` (nth given weekday of the month).
707    dow_nth: Option<(u32, u32)>,
708}
709
710const MONTH_NAMES: [&str; 12] = [
711    "JAN", "FEB", "MAR", "APR", "MAY", "JUN", "JUL", "AUG", "SEP", "OCT", "NOV", "DEC",
712];
713const WEEKDAY_NAMES: [&str; 7] = ["SUN", "MON", "TUE", "WED", "THU", "FRI", "SAT"];
714
715impl CronExpr {
716    fn parse(expr: &str) -> std::result::Result<Self, ()> {
717        let fields: Vec<&str> = expr.split_whitespace().collect();
718
719        // Accept 5, 6, or 7 fields. 6-field adds a leading seconds field; 7-field adds a trailing
720        // year field on top of that. Mirrors croner's field-count handling.
721        let (sec, min, hour, dom, month, dow, year): (
722            &str,
723            &str,
724            &str,
725            &str,
726            &str,
727            &str,
728            Option<&str>,
729        ) = match fields.len() {
730            5 => (
731                "0", fields[0], fields[1], fields[2], fields[3], fields[4], None,
732            ),
733            6 => (
734                fields[0], fields[1], fields[2], fields[3], fields[4], fields[5], None,
735            ),
736            7 => (
737                fields[0],
738                fields[1],
739                fields[2],
740                fields[3],
741                fields[4],
742                fields[5],
743                Some(fields[6]),
744            ),
745            _ => return Err(()),
746        };
747
748        let seconds = parse_field(sec, 0, 59, FieldKind::Plain)?;
749        let minutes = parse_field(min, 0, 59, FieldKind::Plain)?;
750        let hours = parse_field(hour, 0, 23, FieldKind::Plain)?;
751
752        let mut dom_last = false;
753        let mut dom_last_weekday = false;
754        let mut dom_nearest_weekday = None;
755        let days_of_month = parse_dom_field(
756            dom,
757            &mut dom_last,
758            &mut dom_last_weekday,
759            &mut dom_nearest_weekday,
760        )?;
761
762        let months = parse_field(month, 1, 12, FieldKind::Month)?;
763
764        let mut dow_last = None;
765        let mut dow_nth = None;
766        let days_of_week = parse_dow_field(dow, &mut dow_last, &mut dow_nth)?;
767
768        let years = match year {
769            Some(y) => Some(parse_field(y, 1970, 2099, FieldKind::Plain)?),
770            None => None,
771        };
772
773        // `?` is equivalent to `*` for matching purposes, so the field is "unrestricted".
774        let dom_restricted = dom != "*" && dom != "?";
775        let dow_restricted = dow != "*" && dow != "?";
776
777        Ok(Self {
778            seconds,
779            minutes,
780            hours,
781            days_of_month,
782            months,
783            days_of_week,
784            years,
785            dom_restricted,
786            dow_restricted,
787            dom_last,
788            dom_last_weekday,
789            dom_nearest_weekday,
790            dow_last,
791            dow_nth,
792        })
793    }
794
795    /// Find the next instant strictly after `after` (truncated to whole seconds) that matches, in the
796    /// LOCAL timezone. Scans second-by-second only when a sub-minute (seconds) constraint is present;
797    /// otherwise scans minute-by-minute. Bounded so an impossible expression terminates.
798    fn next_after(&self, after: DateTime<Utc>) -> Option<DateTime<Utc>> {
799        let local_after = after.with_timezone(&Local);
800
801        // Determine the step granularity. When seconds is the default `[0]` we can step by minutes.
802        let by_seconds = self.seconds != vec![0];
803
804        let step = if by_seconds {
805            ChronoDuration::seconds(1)
806        } else {
807            ChronoDuration::minutes(1)
808        };
809
810        let mut candidate = if by_seconds {
811            local_after.with_nanosecond(0)? + ChronoDuration::seconds(1)
812        } else {
813            local_after.with_second(0)?.with_nanosecond(0)? + ChronoDuration::minutes(1)
814        };
815
816        // Bound the search: a few years of ticks so an impossible expression terminates.
817        let max_iterations: u64 = if by_seconds {
818            // ~2 years of seconds.
819            2u64 * 366 * 24 * 60 * 60
820        } else {
821            // ~6 years of minutes (years field can push matches far out).
822            6u64 * 366 * 24 * 60
823        };
824        for _ in 0..max_iterations {
825            if self.matches_local(&candidate) {
826                return Some(candidate.with_timezone(&Utc));
827            }
828            candidate += step;
829        }
830        None
831    }
832
833    fn matches_local(&self, dt: &DateTime<Local>) -> bool {
834        if !self.seconds.contains(&dt.second()) {
835            return false;
836        }
837        if !self.minutes.contains(&dt.minute()) {
838            return false;
839        }
840        if !self.hours.contains(&dt.hour()) {
841            return false;
842        }
843        if !self.months.contains(&dt.month()) {
844            return false;
845        }
846        if let Some(years) = &self.years {
847            let year = dt.year();
848            if year < 0 || !years.contains(&(year as u32)) {
849                return false;
850            }
851        }
852
853        let dom_match = self.dom_matches(dt);
854        let dow_match = self.dow_matches(dt);
855
856        // Vixie/croner OR semantics: if both DOM and DOW are restricted, a match in either suffices;
857        // if only one is restricted, only that one is consulted; if neither, both pass.
858        match (self.dom_restricted, self.dow_restricted) {
859            (true, true) => dom_match || dow_match,
860            (true, false) => dom_match,
861            (false, true) => dow_match,
862            (false, false) => true,
863        }
864    }
865
866    fn dom_matches(&self, dt: &DateTime<Local>) -> bool {
867        let dom = dt.day();
868        if self.dom_last && dom == last_day_of_month(dt.year(), dt.month()) {
869            return true;
870        }
871        if self.dom_last_weekday {
872            // Last weekday (Mon-Fri) on or before the last day of the month: the nearest-weekday
873            // resolution of the last day handles the Saturday/Sunday shift back into the month.
874            if is_nearest_weekday(dt, last_day_of_month(dt.year(), dt.month())) {
875                return true;
876            }
877        }
878        if let Some(target) = self.dom_nearest_weekday {
879            if is_nearest_weekday(dt, target) {
880                return true;
881            }
882        }
883        self.days_of_month.contains(&dom)
884    }
885
886    fn dow_matches(&self, dt: &DateTime<Local>) -> bool {
887        let dow = weekday_sun0(dt.weekday());
888
889        if let Some(target) = self.dow_last {
890            // Last occurrence of `target` weekday in this month.
891            if dow == target {
892                let next_week = *dt + ChronoDuration::days(7);
893                if next_week.month() != dt.month() {
894                    return true;
895                }
896            }
897        }
898        if let Some((target, n)) = self.dow_nth {
899            if dow == target {
900                // 1-based occurrence index of this weekday within the month.
901                let occurrence = (dt.day() - 1) / 7 + 1;
902                if occurrence == n {
903                    return true;
904                }
905            }
906        }
907        self.days_of_week.contains(&dow)
908    }
909}
910
911/// Convert chrono `Weekday` to cron's `Sun=0..Sat=6` numbering.
912fn weekday_sun0(weekday: Weekday) -> u32 {
913    weekday.num_days_from_sunday()
914}
915
916/// Last calendar day of a given month.
917fn last_day_of_month(year: i32, month: u32) -> u32 {
918    let (ny, nm) = if month == 12 {
919        (year + 1, 1)
920    } else {
921        (year, month + 1)
922    };
923    let first_next = chrono::NaiveDate::from_ymd_opt(ny, nm, 1).expect("valid first-of-month");
924    (first_next - ChronoDuration::days(1)).day()
925}
926
927/// Whether `dt` is the nearest weekday (Mon-Fri) to day-of-month `target` within the same month,
928/// per cron `W` semantics. If `target` falls on a weekend, the nearest weekday in the same month is
929/// used (Saturday shifts to Friday, Sunday shifts to Monday); a shift never crosses the month
930/// boundary.
931fn is_nearest_weekday(dt: &DateTime<Local>, target: u32) -> bool {
932    let last = last_day_of_month(dt.year(), dt.month());
933    let target = target.min(last);
934    let target_date = chrono::NaiveDate::from_ymd_opt(dt.year(), dt.month(), target);
935    let target_date = match target_date {
936        Some(d) => d,
937        None => return false,
938    };
939    let target_weekday = target_date.weekday();
940    let resolved_day = match target_weekday {
941        Weekday::Sat => {
942            if target > 1 {
943                target - 1
944            } else {
945                // Saturday on the 1st shifts forward to Monday (the 3rd).
946                target + 2
947            }
948        }
949        Weekday::Sun => {
950            if target < last {
951                target + 1
952            } else {
953                // Sunday on the last day shifts back to Friday.
954                target - 2
955            }
956        }
957        Weekday::Mon | Weekday::Tue | Weekday::Wed | Weekday::Thu | Weekday::Fri => target,
958    };
959    dt.day() == resolved_day
960}
961
962#[derive(Clone, Copy, PartialEq, Eq)]
963enum FieldKind {
964    Plain,
965    Month,
966    Weekday,
967}
968
969/// Parse a numeric/named cron field (`*`, `?`, lists, ranges, steps) into the sorted set of matching
970/// values within `[min, max]`. `?` is treated as `*`. For [`FieldKind::Month`] names `JAN`-`DEC` are
971/// accepted.
972fn parse_field(
973    field: &str,
974    min: u32,
975    max: u32,
976    kind: FieldKind,
977) -> std::result::Result<Vec<u32>, ()> {
978    if field == "?" {
979        // `?` = no specific value; treat as the full range.
980        return Ok((min..=max).collect());
981    }
982    let mut values: Vec<u32> = Vec::new();
983    for part in field.split(',') {
984        if part.is_empty() {
985            return Err(());
986        }
987        parse_field_part(part, min, max, kind, &mut values)?;
988    }
989    if values.is_empty() {
990        return Err(());
991    }
992    values.sort_unstable();
993    values.dedup();
994    Ok(values)
995}
996
997/// Parse the day-of-month field, recognizing `L` (last day), `LW` (last weekday), and `<n>W` (nearest
998/// weekday to day `n`) in addition to the standard grammar. Mirrors croner: `W` must be preceded by
999/// `L` or a single day-of-month value in `1..=31` (`W` alone, `0W`, and `32W` are rejected).
1000fn parse_dom_field(
1001    field: &str,
1002    dom_last: &mut bool,
1003    dom_last_weekday: &mut bool,
1004    dom_nearest_weekday: &mut Option<u32>,
1005) -> std::result::Result<Vec<u32>, ()> {
1006    let upper = field.to_ascii_uppercase();
1007    if upper == "L" {
1008        *dom_last = true;
1009        // No fixed numeric days; matching handled by `dom_last`.
1010        return Ok(Vec::new());
1011    }
1012    if upper == "LW" {
1013        *dom_last_weekday = true;
1014        return Ok(Vec::new());
1015    }
1016    if let Some(stripped) = upper.strip_suffix('W') {
1017        let day: u32 = stripped.parse().map_err(|_| ())?;
1018        if !(1..=31).contains(&day) {
1019            return Err(());
1020        }
1021        *dom_nearest_weekday = Some(day);
1022        return Ok(Vec::new());
1023    }
1024    parse_field(field, 1, 31, FieldKind::Plain)
1025}
1026
1027/// Parse the day-of-week field, recognizing `<weekday>L` (last weekday-of-month) and
1028/// `<weekday>#<n>` (nth weekday-of-month), named weekdays, and `7` folded onto Sunday.
1029fn parse_dow_field(
1030    field: &str,
1031    dow_last: &mut Option<u32>,
1032    dow_nth: &mut Option<(u32, u32)>,
1033) -> std::result::Result<Vec<u32>, ()> {
1034    let upper = field.to_ascii_uppercase();
1035
1036    // `<weekday>#<n>` (nth weekday of the month).
1037    if let Some((wd, nth)) = upper.split_once('#') {
1038        let weekday = parse_weekday_token(wd)?;
1039        let n: u32 = nth.parse().map_err(|_| ())?;
1040        if !(1..=5).contains(&n) {
1041            return Err(());
1042        }
1043        *dow_nth = Some((weekday, n));
1044        return Ok(Vec::new());
1045    }
1046
1047    // `<weekday>L` (last given weekday of the month).
1048    if let Some(stripped) = upper.strip_suffix('L') {
1049        let weekday = parse_weekday_token(stripped)?;
1050        *dow_last = Some(weekday);
1051        return Ok(Vec::new());
1052    }
1053
1054    if upper == "?" || upper == "*" {
1055        let mut v = parse_field(field, 0, 7, FieldKind::Plain)?;
1056        fold_sunday(&mut v);
1057        return Ok(v);
1058    }
1059
1060    let mut values = parse_field(field, 0, 7, FieldKind::Weekday)?;
1061    fold_sunday(&mut values);
1062    Ok(values)
1063}
1064
1065/// Fold `7` (Sunday) onto `0` and dedupe.
1066fn fold_sunday(values: &mut Vec<u32>) {
1067    for v in values.iter_mut() {
1068        if *v == 7 {
1069            *v = 0;
1070        }
1071    }
1072    values.sort_unstable();
1073    values.dedup();
1074}
1075
1076/// Parse a single weekday token (numeric `0`-`7` or named `SUN`-`SAT`) to `Sun=0..Sat=6`.
1077fn parse_weekday_token(token: &str) -> std::result::Result<u32, ()> {
1078    let upper = token.to_ascii_uppercase();
1079    if let Some(idx) = WEEKDAY_NAMES.iter().position(|name| *name == upper) {
1080        return Ok(idx as u32);
1081    }
1082    let v: u32 = upper.parse().map_err(|_| ())?;
1083    match v {
1084        0..=6 => Ok(v),
1085        7 => Ok(0),
1086        _ => Err(()),
1087    }
1088}
1089
1090// Re-add FieldKind::Weekday support by extending parse_field via a wrapper for weekday names.
1091impl FieldKind {
1092    fn resolve_name(self, token: &str) -> Option<u32> {
1093        let upper = token.to_ascii_uppercase();
1094        match self {
1095            FieldKind::Plain => None,
1096            FieldKind::Month => MONTH_NAMES
1097                .iter()
1098                .position(|name| *name == upper)
1099                .map(|i| (i + 1) as u32),
1100            FieldKind::Weekday => WEEKDAY_NAMES
1101                .iter()
1102                .position(|name| *name == upper)
1103                .map(|i| i as u32),
1104        }
1105    }
1106}
1107
1108fn parse_field_part(
1109    part: &str,
1110    min: u32,
1111    max: u32,
1112    kind: FieldKind,
1113    out: &mut Vec<u32>,
1114) -> std::result::Result<(), ()> {
1115    // Split off an optional step (`.../n`).
1116    let (range_spec, step) = match part.split_once('/') {
1117        Some((range_spec, step_str)) => {
1118            let step: u32 = step_str.parse().map_err(|_| ())?;
1119            if step == 0 {
1120                return Err(());
1121            }
1122            (range_spec, Some(step))
1123        }
1124        None => (part, None),
1125    };
1126
1127    // Determine the [start, end] bounds for this part.
1128    let (start, end) = if range_spec == "*" {
1129        (min, max)
1130    } else if let Some((lo, hi)) = range_spec.split_once('-') {
1131        let lo = parse_value_token(lo, kind)?;
1132        let hi = parse_value_token(hi, kind)?;
1133        (lo, hi)
1134    } else {
1135        // A bare numeric value may not carry a step. croner rejects `5/15` / `0/5`
1136        // ("stepping with numeric prefix"); only `*` or an explicit range may precede `/`.
1137        if step.is_some() {
1138            return Err(());
1139        }
1140        let v = parse_value_token(range_spec, kind)?;
1141        (v, v)
1142    };
1143
1144    if start < min || end > max || start > end {
1145        return Err(());
1146    }
1147
1148    let step = step.unwrap_or(1);
1149    let mut v = start;
1150    while v <= end {
1151        out.push(v);
1152        v += step;
1153    }
1154    Ok(())
1155}
1156
1157/// Parse a single value token: a number, or a name (month names for [`FieldKind::Month`], weekday
1158/// names for [`FieldKind::Weekday`]).
1159fn parse_value_token(token: &str, kind: FieldKind) -> std::result::Result<u32, ()> {
1160    match kind {
1161        FieldKind::Weekday => parse_weekday_token(token),
1162        FieldKind::Month => {
1163            if let Some(v) = kind.resolve_name(token) {
1164                return Ok(v);
1165            }
1166            token.parse().map_err(|_| ())
1167        }
1168        FieldKind::Plain => token.parse().map_err(|_| ()),
1169    }
1170}
1171
1172// ---------------------------------------------------------------------------
1173// Methods
1174// ---------------------------------------------------------------------------
1175
1176impl AgentOs {
1177    /// Schedule a cron job. SYNC. Validates the schedule (errors `InvalidSchedule` / `PastSchedule`).
1178    /// `id` defaults to a UUID; `overlap` defaults to allow.
1179    ///
1180    /// Mirrors TS `AgentOs.scheduleCron` / `CronManager.schedule`: validation happens up front, the
1181    /// driver is asked to arm the timer (`this.driver.schedule({ id, schedule, callback })`), and the
1182    /// job is registered. The driver owns all timing: it parses the schedule, fires the callback,
1183    /// reschedules cron after each fire, and is cancelled on [`CronJobHandle::cancel`] /
1184    /// [`CronManager::dispose`]. The returned [`CronJobHandle`] cancels the job.
1185    pub fn schedule_cron(
1186        &self,
1187        options: CronJobOptions,
1188    ) -> std::result::Result<CronJobHandle, ClientError> {
1189        let cron = self.cron();
1190        let now = Utc::now();
1191
1192        // Validate before any state mutation, matching TS `validateScheduleForRegistration`.
1193        let next_run = validate_schedule(&options.schedule, now)?;
1194
1195        let id = options
1196            .id
1197            .unwrap_or_else(|| uuid::Uuid::new_v4().to_string());
1198        let overlap = options.overlap.unwrap_or_default();
1199
1200        // Build the driver callback that runs one job execution, mirroring TS
1201        // `callback: () => this.executeJob(id)`.
1202        let manager = Arc::clone(cron);
1203        let vm = self.clone();
1204        let callback_id = id.clone();
1205        let callback: crate::config::ScheduleCallback = Arc::new(move || {
1206            let manager = Arc::clone(&manager);
1207            let vm = vm.clone();
1208            let id = callback_id.clone();
1209            Box::pin(async move {
1210                execute_job(manager, vm, id).await;
1211            })
1212        });
1213
1214        register_cron_job(
1215            cron,
1216            id,
1217            options.schedule,
1218            options.action,
1219            overlap,
1220            next_run,
1221            callback,
1222        )
1223    }
1224
1225    /// Snapshot all cron jobs. Mirrors TS `CronManager.list`.
1226    pub fn list_cron_jobs(&self) -> Vec<CronJobInfo> {
1227        let mut result = Vec::new();
1228        self.cron().jobs.scan(|id, state| {
1229            result.push(CronJobInfo {
1230                id: id.clone(),
1231                schedule: state.schedule.clone(),
1232                action: CronActionInfo::from(&state.action),
1233                overlap: state.overlap,
1234                last_run: state
1235                    .last_run
1236                    .lock()
1237                    .as_ref()
1238                    .map(|time| time.to_rfc3339_opts(SecondsFormat::Millis, true)),
1239                next_run: state
1240                    .next_run
1241                    .lock()
1242                    .as_ref()
1243                    .map(|time| time.to_rfc3339_opts(SecondsFormat::Millis, true)),
1244                run_count: state.run_count.load(Ordering::SeqCst),
1245                running: state.running.load(Ordering::SeqCst),
1246            });
1247        });
1248        result
1249    }
1250
1251    /// Cancel a cron job. No-op if unknown; never errors. Mirrors TS `CronManager.cancel`.
1252    pub fn cancel_cron_job(&self, id: &str) {
1253        self.cron().cancel_job(id);
1254    }
1255
1256    /// Subscribe to cron events. The TS API returns no unsubscribe; dropping the receiver is the
1257    /// equivalent. Each run emits `Fire` then `Complete`|`Error`. Mirrors TS `AgentOs.onCronEvent`.
1258    pub fn cron_events(&self) -> broadcast::Receiver<CronEvent> {
1259        self.cron().event_tx.subscribe()
1260    }
1261}
1262
1263fn ensure_cron_capacity(cron: &CronManager, id: &str) -> std::result::Result<(), ClientError> {
1264    if cron.jobs.contains(id) || cron.jobs.len() < crate::CRON_JOB_LIMIT {
1265        return Ok(());
1266    }
1267
1268    Err(ClientError::Sidecar(format!(
1269        "cron job limit exceeded: at most {} jobs can be scheduled per VM",
1270        crate::CRON_JOB_LIMIT
1271    )))
1272}
1273
1274fn register_cron_job(
1275    cron: &Arc<CronManager>,
1276    id: String,
1277    schedule: String,
1278    action: CronAction,
1279    overlap: CronOverlap,
1280    next_run: Option<DateTime<Utc>>,
1281    callback: crate::config::ScheduleCallback,
1282) -> std::result::Result<CronJobHandle, ClientError> {
1283    let _guard = cron.schedule_lock.lock();
1284    ensure_cron_capacity(cron, &id)?;
1285
1286    // If replacing an existing id, cancel the old driver-armed timer before scheduling the new one.
1287    // The default timer driver's handles are id-based, so cancelling after the new schedule would
1288    // cancel the replacement.
1289    if let Some((_, old)) = cron.jobs.remove(&id) {
1290        cron.driver.cancel(&old.handle);
1291    }
1292
1293    let handle = cron.driver.schedule(ScheduleEntry {
1294        id: id.clone(),
1295        schedule: schedule.clone(),
1296        callback,
1297    });
1298
1299    let state = CronJobState {
1300        schedule,
1301        action,
1302        overlap,
1303        last_run: parking_lot::Mutex::new(None),
1304        next_run: parking_lot::Mutex::new(next_run),
1305        run_count: std::sync::atomic::AtomicU64::new(0),
1306        running: AtomicBool::new(false),
1307        queued: AtomicBool::new(false),
1308        handle,
1309    };
1310
1311    let _ = cron.jobs.insert(id.clone(), state);
1312
1313    Ok(CronJobHandle {
1314        id,
1315        manager: Arc::clone(cron),
1316    })
1317}
1318
1319#[cfg(test)]
1320mod tests {
1321    use super::{
1322        ensure_cron_capacity, register_cron_job, CronAction, CronJobState, CronManager,
1323        CronOverlap, ScheduleDriver, ScheduleEntry, ScheduleHandle,
1324    };
1325    use crate::CRON_JOB_LIMIT;
1326    use std::sync::atomic::AtomicBool;
1327    use std::sync::Arc;
1328
1329    #[derive(Default)]
1330    struct RecordingScheduleDriver {
1331        calls: parking_lot::Mutex<Vec<String>>,
1332    }
1333
1334    impl ScheduleDriver for RecordingScheduleDriver {
1335        fn schedule(&self, entry: ScheduleEntry) -> ScheduleHandle {
1336            self.calls.lock().push(format!("schedule:{}", entry.id));
1337            ScheduleHandle { id: entry.id }
1338        }
1339
1340        fn cancel(&self, handle: &ScheduleHandle) {
1341            self.calls.lock().push(format!("cancel:{}", handle.id));
1342        }
1343
1344        fn dispose(&self) {}
1345    }
1346
1347    fn dummy_state(id: String) -> CronJobState {
1348        CronJobState {
1349            schedule: "0 0 * * *".to_string(),
1350            action: CronAction::Callback {
1351                callback: Arc::new(|| Box::pin(async {})),
1352            },
1353            overlap: CronOverlap::Allow,
1354            last_run: parking_lot::Mutex::new(None),
1355            next_run: parking_lot::Mutex::new(None),
1356            run_count: std::sync::atomic::AtomicU64::new(0),
1357            running: AtomicBool::new(false),
1358            queued: AtomicBool::new(false),
1359            handle: ScheduleHandle { id },
1360        }
1361    }
1362
1363    #[test]
1364    fn cron_capacity_rejects_new_jobs_at_limit_but_allows_replacements() {
1365        let manager = CronManager::new(Arc::new(RecordingScheduleDriver::default()));
1366        for index in 0..CRON_JOB_LIMIT {
1367            let id = format!("job-{index}");
1368            assert!(
1369                manager.jobs.insert(id.clone(), dummy_state(id)).is_ok(),
1370                "seed cron job"
1371            );
1372        }
1373
1374        let error = ensure_cron_capacity(&manager, "overflow").expect_err("limit should reject");
1375        assert!(
1376            error.to_string().contains("cron job limit exceeded"),
1377            "unexpected limit error: {error}"
1378        );
1379        ensure_cron_capacity(&manager, "job-0").expect("replacement should be allowed");
1380    }
1381
1382    // ── Security: AOSCLIENT-P1-cron-exec (N-007 untrusted cron CronAction::Exec) ─────────────────
1383    //
1384    // Threat: an untrusted actor schedules a `CronAction::Exec { command, args }` whose `args`
1385    // carry data values (a path with spaces) or shell metacharacters (`$( )`, backticks). The
1386    // intent of a structured `(command, args)` action is that the args are passed VERBATIM as
1387    // argv elements — never re-split on whitespace, never re-evaluated by a shell.
1388    //
1389    // The bug (now fixed): `run_action`'s `CronAction::Exec` arm flattened the pair with
1390    // `format!("{} {}", command, args.join(" "))` and handed the STRING to `AgentOs::exec`, which
1391    // re-parsed it through `resolve_exec_command`. That round-trip (a) re-split `"a b"` into two
1392    // argv elements and (b)/(c) promoted `$(id)` / backtick elements to a real `sh -c` shell
1393    // evaluation. The fix sends the structured argv verbatim via `AgentOs::exec_argv`, bypassing
1394    // `resolve_exec_command` entirely.
1395    //
1396    // This test pins the fix: it computes the argv exactly as the fixed `CronAction::Exec` arm
1397    // does (verbatim `command` + `args`), and asserts the hostile elements survive intact. As a
1398    // negative control it also shows the OLD join+`resolve_exec_command` path corrupts them, so a
1399    // regression back to the flatten behavior fails this test.
1400    #[test]
1401    fn cron_exec_action_argv_is_not_shell_re_split_or_evaluated() {
1402        // The fixed `CronAction::Exec` arm passes `command` and `args` straight to `exec_argv`,
1403        // which sends them verbatim with no `resolve_exec_command` round-trip.
1404        fn cron_exec_argv(command: &str, args: &[&str]) -> (String, Vec<String>) {
1405            (
1406                command.to_string(),
1407                args.iter().map(|a| a.to_string()).collect(),
1408            )
1409        }
1410
1411        // The pre-fix flatten+re-parse path, kept here purely as a negative control.
1412        fn buggy_join_then_resolve(command: &str, args: &[&str]) -> (String, Vec<String>) {
1413            let joined = if args.is_empty() {
1414                command.to_string()
1415            } else {
1416                format!("{} {}", command, args.join(" "))
1417            };
1418            crate::command_line::resolve_exec_command(&joined).expect("line must resolve")
1419        }
1420
1421        // (a) A single argv element that contains a space MUST stay one argv element.
1422        let (cmd, args) = cron_exec_argv("printenv", &["a b"]);
1423        assert_eq!(
1424            (cmd.as_str(), args.as_slice()),
1425            ("printenv", &["a b".to_string()][..]),
1426            "N-007: structured argv element \"a b\" must survive as a single argv element"
1427        );
1428        // Negative control: the old path corrupts it by re-splitting on whitespace.
1429        let (_, buggy_args) = buggy_join_then_resolve("printenv", &["a b"]);
1430        assert_eq!(
1431            buggy_args,
1432            vec!["a".to_string(), "b".to_string()],
1433            "N-007 negative control: the old join+resolve path re-split \"a b\" into two argv elements"
1434        );
1435
1436        // (b) A command-substitution argv element MUST stay a literal argv element, never `sh -c`.
1437        let (cmd, args) = cron_exec_argv("printenv", &["$(id)"]);
1438        assert_eq!(
1439            (cmd.as_str(), args.as_slice()),
1440            ("printenv", &["$(id)".to_string()][..]),
1441            "N-007: command-substitution argv element \"$(id)\" must NOT be promoted to `sh -c`"
1442        );
1443        // Negative control: the old path routes the whole line through `sh -c`, evaluating `$(id)`.
1444        let (buggy_cmd, _) = buggy_join_then_resolve("printenv", &["$(id)"]);
1445        assert_eq!(
1446            buggy_cmd, "sh",
1447            "N-007 negative control: the old path promoted the `$(id)` line to a `sh -c` shell"
1448        );
1449
1450        // (c) A backtick argv element: same guarantee.
1451        let (cmd, args) = cron_exec_argv("printenv", &["`id`"]);
1452        assert_eq!(
1453            (cmd.as_str(), args.as_slice()),
1454            ("printenv", &["`id`".to_string()][..]),
1455            "N-007: backtick argv element \"`id`\" must NOT be promoted to `sh -c`"
1456        );
1457        let (buggy_cmd, _) = buggy_join_then_resolve("printenv", &["`id`"]);
1458        assert_eq!(
1459            buggy_cmd, "sh",
1460            "N-007 negative control: the old path promoted the backtick line to a `sh -c` shell"
1461        );
1462    }
1463
1464    // ── Security: AOSCLIENT-P2-cron-cap (N-008 cron job-limit flooding) ──────────────────────────
1465    //
1466    // Threat: an untrusted actor floods the cron registry to exhaust host scheduling resources.
1467    // The public `AgentOs::schedule_cron` registers through `register_cron_job` ->
1468    // `ensure_cron_capacity` (cron.rs:1162), which must cap distinct jobs at `CRON_JOB_LIMIT`
1469    // while still allowing an existing id to be REPLACED at the cap. `AgentOs::schedule_cron`
1470    // itself needs a live sidecar to construct, so we drive the exact same public registration
1471    // chokepoint (`register_cron_job`) the public method funnels into, with a recording driver.
1472    #[test]
1473    fn schedule_cron_public_path_rejects_jobs_beyond_cron_job_limit() {
1474        let driver = Arc::new(RecordingScheduleDriver::default());
1475        let manager = Arc::new(CronManager::new(driver.clone()));
1476
1477        let make_callback =
1478            || -> crate::config::ScheduleCallback { Arc::new(|| Box::pin(async {})) };
1479
1480        // Fill the registry to exactly CRON_JOB_LIMIT distinct ids through the public chokepoint.
1481        for index in 0..CRON_JOB_LIMIT {
1482            register_cron_job(
1483                &manager,
1484                format!("flood-{index}"),
1485                "0 0 * * *".to_string(),
1486                CronAction::Callback {
1487                    callback: make_callback(),
1488                },
1489                CronOverlap::Allow,
1490                None,
1491                make_callback(),
1492            )
1493            .unwrap_or_else(|err| panic!("seed job {index} should register: {err}"));
1494        }
1495        assert_eq!(manager.jobs.len(), CRON_JOB_LIMIT);
1496
1497        // The CRON_JOB_LIMIT+1-th DISTINCT id must be denied. (Match instead of `.expect_err()`
1498        // because the Ok type `CronJobHandle` does not implement Debug.)
1499        let overflow = match register_cron_job(
1500            &manager,
1501            "flood-overflow".to_string(),
1502            "0 0 * * *".to_string(),
1503            CronAction::Callback {
1504                callback: make_callback(),
1505            },
1506            CronOverlap::Allow,
1507            None,
1508            make_callback(),
1509        ) {
1510            Ok(_) => {
1511                panic!("AOSCLIENT-P2-cron-cap: the job beyond CRON_JOB_LIMIT must be rejected")
1512            }
1513            Err(err) => err,
1514        };
1515        assert!(
1516            overflow.to_string().contains("cron job limit exceeded"),
1517            "AOSCLIENT-P2-cron-cap: overflow rejection must report the cron job limit, got: {overflow}"
1518        );
1519        assert_eq!(
1520            manager.jobs.len(),
1521            CRON_JOB_LIMIT,
1522            "AOSCLIENT-P2-cron-cap: a rejected overflow job must not be inserted"
1523        );
1524
1525        // Replacing an EXISTING id while at the cap must still succeed (replace, not grow).
1526        register_cron_job(
1527            &manager,
1528            "flood-0".to_string(),
1529            "0 1 * * *".to_string(),
1530            CronAction::Callback {
1531                callback: make_callback(),
1532            },
1533            CronOverlap::Allow,
1534            None,
1535            make_callback(),
1536        )
1537        .expect("AOSCLIENT-P2-cron-cap: replacing an existing id at the cap must be allowed");
1538        assert_eq!(
1539            manager.jobs.len(),
1540            CRON_JOB_LIMIT,
1541            "AOSCLIENT-P2-cron-cap: replacing an existing id must not grow the registry past the cap"
1542        );
1543    }
1544
1545    #[test]
1546    fn cron_replacement_cancels_old_timer_before_scheduling_new_timer() {
1547        let driver = Arc::new(RecordingScheduleDriver::default());
1548        let manager = Arc::new(CronManager::new(driver.clone()));
1549        let callback: crate::config::ScheduleCallback = Arc::new(|| Box::pin(async {}));
1550
1551        register_cron_job(
1552            &manager,
1553            "same-id".to_string(),
1554            "0 0 * * *".to_string(),
1555            CronAction::Callback {
1556                callback: callback.clone(),
1557            },
1558            CronOverlap::Allow,
1559            None,
1560            callback.clone(),
1561        )
1562        .expect("initial schedule");
1563        register_cron_job(
1564            &manager,
1565            "same-id".to_string(),
1566            "0 1 * * *".to_string(),
1567            CronAction::Callback { callback },
1568            CronOverlap::Allow,
1569            None,
1570            Arc::new(|| Box::pin(async {})),
1571        )
1572        .expect("replacement schedule");
1573
1574        assert_eq!(
1575            *driver.calls.lock(),
1576            vec!["schedule:same-id", "cancel:same-id", "schedule:same-id"]
1577        );
1578        assert_eq!(manager.jobs.len(), 1);
1579    }
1580}