1use 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#[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#[derive(Clone)]
47pub enum CronAction {
48 Session {
50 agent_type: String,
51 prompt: String,
52 options: Option<CronSessionOptions>,
53 },
54 Exec { command: String, args: Vec<String> },
56 Callback {
58 #[allow(clippy::type_complexity)]
59 callback: Arc<dyn Fn() -> futures::future::BoxFuture<'static, ()> + Send + Sync>,
60 },
61}
62
63#[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#[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#[derive(Clone)]
140pub struct CronJobOptions {
141 pub id: Option<String>,
143 pub schedule: String,
145 pub action: CronAction,
146 pub overlap: Option<CronOverlap>,
148}
149
150#[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#[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#[derive(Clone)]
186pub struct CronJobHandle {
187 pub id: String,
188 pub(crate) manager: Arc<CronManager>,
189}
190
191impl CronJobHandle {
192 pub fn cancel(&self) {
194 self.manager.cancel_job(&self.id);
195 }
196}
197
198pub(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 pub queued: AtomicBool,
214 pub handle: ScheduleHandle,
217}
218
219pub 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 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 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 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
264fn 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 {
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 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 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 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
365async 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 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 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
438pub(crate) enum ParsedSchedule {
446 Date(DateTime<Utc>),
448 Cron(CronExpr),
450}
451
452impl ParsedSchedule {
453 pub(crate) fn is_cron(&self) -> bool {
455 matches!(self, ParsedSchedule::Cron(_))
456 }
457}
458
459pub(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
478fn 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 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 if take_lit(bytes, &mut i, b':') {
544 if !take_digits(bytes, &mut i, 2) {
545 return false;
546 }
547 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 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
587fn parse_one_shot(schedule: &str) -> Option<DateTime<Utc>> {
594 use chrono::TimeZone;
595
596 if let Ok(dt) = DateTime::parse_from_rfc3339(schedule) {
598 return Some(dt.with_timezone(&Utc));
599 }
600
601 let normalized = schedule.replacen(' ', "T", 1);
603
604 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 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
628pub(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
644pub(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
652pub(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
675pub(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 dom_last: bool,
700 dom_last_weekday: bool,
702 dom_nearest_weekday: Option<u32>,
704 dow_last: Option<u32>,
706 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 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 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 fn next_after(&self, after: DateTime<Utc>) -> Option<DateTime<Utc>> {
799 let local_after = after.with_timezone(&Local);
800
801 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 let max_iterations: u64 = if by_seconds {
818 2u64 * 366 * 24 * 60 * 60
820 } else {
821 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 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 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 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 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
911fn weekday_sun0(weekday: Weekday) -> u32 {
913 weekday.num_days_from_sunday()
914}
915
916fn 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
927fn 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 target + 2
947 }
948 }
949 Weekday::Sun => {
950 if target < last {
951 target + 1
952 } else {
953 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
969fn parse_field(
973 field: &str,
974 min: u32,
975 max: u32,
976 kind: FieldKind,
977) -> std::result::Result<Vec<u32>, ()> {
978 if field == "?" {
979 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
997fn 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 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
1027fn 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 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 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
1065fn 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
1076fn 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
1090impl 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 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 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 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
1157fn 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
1172impl AgentOs {
1177 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 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 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 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 pub fn cancel_cron_job(&self, id: &str) {
1253 self.cron().cancel_job(id);
1254 }
1255
1256 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 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 #[test]
1401 fn cron_exec_action_argv_is_not_shell_re_split_or_evaluated() {
1402 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 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 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 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 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 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 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 #[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 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 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 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}