1use std::sync::atomic::{AtomicBool, Ordering};
16use std::sync::Arc;
17
18use chrono::{DateTime, Datelike, Duration as ChronoDuration, Local, Timelike, Utc, Weekday};
19use scc::HashMap as SccHashMap;
20use serde::{Deserialize, Serialize};
21use tokio::sync::broadcast;
22
23use crate::agent_os::AgentOs;
24use crate::config::{ScheduleDriver, ScheduleEntry, ScheduleHandle};
25use crate::error::ClientError;
26use crate::session::CreateSessionOptions;
27
28#[derive(Debug, Clone, Copy, PartialEq, Eq, Default, Serialize, Deserialize)]
34#[serde(rename_all = "lowercase")]
35pub enum CronOverlap {
36 #[default]
37 Allow,
38 Skip,
39 Queue,
40}
41
42#[derive(Clone)]
44pub enum CronAction {
45 Session {
47 agent_type: String,
48 prompt: String,
49 options: Option<CreateSessionOptions>,
50 },
51 Exec { command: String, args: Vec<String> },
53 Callback {
55 #[allow(clippy::type_complexity)]
56 callback: Arc<dyn Fn() -> futures::future::BoxFuture<'static, ()> + Send + Sync>,
57 },
58}
59
60impl std::fmt::Debug for CronAction {
61 fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
62 match self {
63 CronAction::Session {
64 agent_type, prompt, ..
65 } => f
66 .debug_struct("Session")
67 .field("agent_type", agent_type)
68 .field("prompt", prompt)
69 .finish_non_exhaustive(),
70 CronAction::Exec { command, args } => f
71 .debug_struct("Exec")
72 .field("command", command)
73 .field("args", args)
74 .finish(),
75 CronAction::Callback { .. } => f.debug_struct("Callback").finish_non_exhaustive(),
76 }
77 }
78}
79
80#[derive(Clone)]
82pub struct CronJobOptions {
83 pub id: Option<String>,
85 pub schedule: String,
87 pub action: CronAction,
88 pub overlap: Option<CronOverlap>,
90}
91
92#[derive(Debug, Clone)]
94pub struct CronJobInfo {
95 pub id: String,
96 pub schedule: String,
97 pub action: CronAction,
98 pub overlap: CronOverlap,
99 pub last_run: Option<DateTime<Utc>>,
100 pub next_run: Option<DateTime<Utc>>,
101 pub run_count: u64,
102 pub running: bool,
103}
104
105#[derive(Debug, Clone)]
107pub enum CronEvent {
108 Fire {
109 job_id: String,
110 time: DateTime<Utc>,
111 },
112 Complete {
113 job_id: String,
114 time: DateTime<Utc>,
115 duration_ms: f64,
116 },
117 Error {
118 job_id: String,
119 time: DateTime<Utc>,
120 error: String,
121 },
122}
123
124#[derive(Clone)]
126pub struct CronJobHandle {
127 pub id: String,
128 pub(crate) manager: Arc<CronManager>,
129}
130
131impl CronJobHandle {
132 pub fn cancel(&self) {
134 self.manager.cancel_job(&self.id);
135 }
136}
137
138pub(crate) struct CronJobState {
144 pub schedule: String,
145 pub action: CronAction,
146 pub overlap: CronOverlap,
147 pub last_run: parking_lot::Mutex<Option<DateTime<Utc>>>,
148 pub next_run: parking_lot::Mutex<Option<DateTime<Utc>>>,
149 pub run_count: std::sync::atomic::AtomicU64,
150 pub running: AtomicBool,
151 pub queued: AtomicBool,
154 pub handle: ScheduleHandle,
157}
158
159pub struct CronManager {
161 pub(crate) jobs: SccHashMap<String, CronJobState>,
162 pub(crate) schedule_lock: parking_lot::Mutex<()>,
163 pub(crate) driver: Arc<dyn ScheduleDriver>,
164 pub(crate) event_tx: broadcast::Sender<CronEvent>,
165}
166
167impl CronManager {
168 pub(crate) fn new(driver: Arc<dyn ScheduleDriver>) -> Self {
170 let (event_tx, _rx) = broadcast::channel(256);
171 Self {
172 jobs: SccHashMap::new(),
173 schedule_lock: parking_lot::Mutex::new(()),
174 driver,
175 event_tx,
176 }
177 }
178
179 pub(crate) fn cancel_job(&self, id: &str) {
184 let _guard = self.schedule_lock.lock();
185 if let Some((_, state)) = self.jobs.remove(id) {
186 self.driver.cancel(&state.handle);
187 }
188 }
189
190 pub(crate) fn dispose(&self) {
195 let _guard = self.schedule_lock.lock();
196 self.jobs.scan(|_, state| {
197 self.driver.cancel(&state.handle);
198 });
199 self.jobs.clear();
200 self.driver.dispose();
201 }
202}
203
204fn execute_job(
213 manager: Arc<CronManager>,
214 vm: AgentOs,
215 id: String,
216) -> futures::future::BoxFuture<'static, ()> {
217 Box::pin(execute_job_inner(manager, vm, id))
218}
219
220async fn execute_job_inner(manager: Arc<CronManager>, vm: AgentOs, id: String) {
221 let manager = &manager;
222 let vm = &vm;
223 let id = id.as_str();
224 {
227 let mut should_return = false;
228 let mut should_queue = false;
229 manager.jobs.read(id, |_, state| {
230 if state.running.load(Ordering::SeqCst) {
231 match state.overlap {
232 CronOverlap::Allow => {}
233 CronOverlap::Skip => should_return = true,
234 CronOverlap::Queue => should_queue = true,
235 }
236 }
237 });
238 if should_return {
239 return;
240 }
241 if should_queue {
242 manager.jobs.read(id, |_, state| {
243 state.queued.store(true, Ordering::SeqCst);
244 });
245 return;
246 }
247 }
248
249 let action = match manager.jobs.read(id, |_, state| {
251 state.running.store(true, Ordering::SeqCst);
252 *state.last_run.lock() = Some(Utc::now());
253 state.run_count.fetch_add(1, Ordering::SeqCst);
254 state.action.clone()
255 }) {
256 Some(action) => action,
257 None => return,
258 };
259
260 let _ = manager.event_tx.send(CronEvent::Fire {
261 job_id: id.to_string(),
262 time: Utc::now(),
263 });
264
265 let start = Utc::now();
267 let result = run_action(vm, &action).await;
268 let duration_ms = (Utc::now() - start).num_milliseconds() as f64;
269
270 match result {
271 Ok(()) => {
272 let _ = manager.event_tx.send(CronEvent::Complete {
273 job_id: id.to_string(),
274 time: Utc::now(),
275 duration_ms,
276 });
277 }
278 Err(error) => {
279 let _ = manager.event_tx.send(CronEvent::Error {
280 job_id: id.to_string(),
281 time: Utc::now(),
282 error: error.to_string(),
283 });
284 }
285 }
286
287 let mut run_queued = false;
289 manager.jobs.read(id, |_, state| {
290 state.running.store(false, Ordering::SeqCst);
291 *state.next_run.lock() = compute_next_time(&state.schedule, Utc::now());
292 if state.queued.swap(false, Ordering::SeqCst) {
293 run_queued = true;
294 }
295 });
296
297 if run_queued {
298 let manager = Arc::clone(manager);
299 let vm = vm.clone();
300 let id = id.to_string();
301 tokio::spawn(execute_job(manager, vm, id));
302 }
303}
304
305async fn run_action(vm: &AgentOs, action: &CronAction) -> Result<(), ClientError> {
312 match action {
313 CronAction::Session {
314 agent_type,
315 prompt,
316 options,
317 } => {
318 let session = vm
319 .create_session(agent_type, options.clone().unwrap_or_default())
320 .await
321 .map_err(|err| ClientError::Sidecar(err.to_string()))?;
322 let prompt_result = vm.prompt(&session.session_id, prompt).await;
323 let _ = vm.close_session(&session.session_id);
325 prompt_result.map_err(|err| ClientError::Sidecar(err.to_string()))?;
326 Ok(())
327 }
328 CronAction::Exec { command, args } => {
329 vm.exec_argv(command, args, crate::process::ExecOptions::default())
334 .await
335 .map_err(|err| ClientError::Sidecar(err.to_string()))?;
336 Ok(())
337 }
338 CronAction::Callback { callback } => {
339 callback().await;
340 Ok(())
341 }
342 }
343}
344
345pub(crate) enum ParsedSchedule {
353 Date(DateTime<Utc>),
355 Cron(CronExpr),
357}
358
359impl ParsedSchedule {
360 pub(crate) fn is_cron(&self) -> bool {
362 matches!(self, ParsedSchedule::Cron(_))
363 }
364}
365
366pub(crate) fn resolve_next_run(
370 parsed: &ParsedSchedule,
371 now: DateTime<Utc>,
372) -> Option<DateTime<Utc>> {
373 match parsed {
374 ParsedSchedule::Cron(cron) => cron.next_after(now),
375 ParsedSchedule::Date(date) => {
376 if date.timestamp_millis() > now.timestamp_millis() {
377 Some(*date)
378 } else {
379 None
380 }
381 }
382 }
383}
384
385fn looks_like_one_shot(schedule: &str) -> bool {
391 let bytes = schedule.as_bytes();
392 let mut i = 0usize;
393
394 let is_digit = |b: u8| b.is_ascii_digit();
395
396 let take_digits = |bytes: &[u8], i: &mut usize, n: usize| -> bool {
397 for _ in 0..n {
398 match bytes.get(*i) {
399 Some(&b) if is_digit(b) => *i += 1,
400 _ => return false,
401 }
402 }
403 true
404 };
405 let take_lit = |bytes: &[u8], i: &mut usize, lit: u8| -> bool {
406 match bytes.get(*i) {
407 Some(&b) if b == lit => {
408 *i += 1;
409 true
410 }
411 _ => false,
412 }
413 };
414
415 if !take_digits(bytes, &mut i, 4) {
416 return false;
417 }
418 if !take_lit(bytes, &mut i, b'-') {
419 return false;
420 }
421 if !take_digits(bytes, &mut i, 2) {
422 return false;
423 }
424 if !take_lit(bytes, &mut i, b'-') {
425 return false;
426 }
427 if !take_digits(bytes, &mut i, 2) {
428 return false;
429 }
430
431 if i == bytes.len() {
433 return true;
434 }
435 match bytes.get(i) {
436 Some(b'T') | Some(b' ') => i += 1,
437 _ => return false,
438 }
439 if !take_digits(bytes, &mut i, 2) {
440 return false;
441 }
442 if !take_lit(bytes, &mut i, b':') {
443 return false;
444 }
445 if !take_digits(bytes, &mut i, 2) {
446 return false;
447 }
448
449 if take_lit(bytes, &mut i, b':') {
451 if !take_digits(bytes, &mut i, 2) {
452 return false;
453 }
454 if take_lit(bytes, &mut i, b'.') {
459 let mut frac = 0;
460 while matches!(bytes.get(i), Some(&b) if is_digit(b)) {
461 i += 1;
462 frac += 1;
463 }
464 if frac == 0 {
465 return false;
466 }
467 }
468 }
469
470 match bytes.get(i) {
472 None => return true,
473 Some(b'Z') => {
474 i += 1;
475 }
476 Some(b'+') | Some(b'-') => {
477 i += 1;
478 if !take_digits(bytes, &mut i, 2) {
479 return false;
480 }
481 if !take_lit(bytes, &mut i, b':') {
482 return false;
483 }
484 if !take_digits(bytes, &mut i, 2) {
485 return false;
486 }
487 }
488 _ => return false,
489 }
490
491 i == bytes.len()
492}
493
494fn parse_one_shot(schedule: &str) -> Option<DateTime<Utc>> {
501 use chrono::TimeZone;
502
503 if let Ok(dt) = DateTime::parse_from_rfc3339(schedule) {
505 return Some(dt.with_timezone(&Utc));
506 }
507
508 let normalized = schedule.replacen(' ', "T", 1);
510
511 for fmt in [
513 "%Y-%m-%dT%H:%M:%S%.f",
514 "%Y-%m-%dT%H:%M:%S",
515 "%Y-%m-%dT%H:%M",
516 ] {
517 if let Ok(naive) = chrono::NaiveDateTime::parse_from_str(&normalized, fmt) {
518 return match Local.from_local_datetime(&naive) {
519 chrono::LocalResult::Single(dt) => Some(dt.with_timezone(&Utc)),
520 chrono::LocalResult::Ambiguous(dt, _) => Some(dt.with_timezone(&Utc)),
521 chrono::LocalResult::None => None,
522 };
523 }
524 }
525
526 if let Ok(date) = chrono::NaiveDate::parse_from_str(schedule, "%Y-%m-%d") {
528 let naive = date.and_hms_opt(0, 0, 0)?;
529 return Some(DateTime::<Utc>::from_naive_utc_and_offset(naive, Utc));
530 }
531
532 None
533}
534
535pub(crate) fn parse_schedule(schedule: &str) -> std::result::Result<ParsedSchedule, ClientError> {
537 let normalized = schedule.trim();
538 if looks_like_one_shot(normalized) {
539 return match parse_one_shot(normalized) {
540 Some(date) => Ok(ParsedSchedule::Date(date)),
541 None => Err(ClientError::InvalidSchedule(schedule.to_string())),
542 };
543 }
544
545 match CronExpr::parse(normalized) {
546 Ok(cron) => Ok(ParsedSchedule::Cron(cron)),
547 Err(_) => Err(ClientError::InvalidSchedule(schedule.to_string())),
548 }
549}
550
551pub(crate) fn compute_next_time(schedule: &str, now: DateTime<Utc>) -> Option<DateTime<Utc>> {
555 let parsed = parse_schedule(schedule).ok()?;
556 resolve_next_run(&parsed, now)
557}
558
559pub(crate) fn validate_schedule(
566 schedule: &str,
567 now: DateTime<Utc>,
568) -> std::result::Result<Option<DateTime<Utc>>, ClientError> {
569 let parsed = parse_schedule(schedule)?;
570 match parsed {
571 ParsedSchedule::Cron(cron) => Ok(cron.next_after(now)),
572 ParsedSchedule::Date(date) => {
573 if date.timestamp_millis() > now.timestamp_millis() {
574 Ok(Some(date))
575 } else {
576 Err(ClientError::PastSchedule(schedule.to_string()))
577 }
578 }
579 }
580}
581
582pub(crate) struct CronExpr {
596 seconds: Vec<u32>,
597 minutes: Vec<u32>,
598 hours: Vec<u32>,
599 days_of_month: Vec<u32>,
600 months: Vec<u32>,
601 days_of_week: Vec<u32>,
602 years: Option<Vec<u32>>,
603 dom_restricted: bool,
604 dow_restricted: bool,
605 dom_last: bool,
607 dom_last_weekday: bool,
609 dom_nearest_weekday: Option<u32>,
611 dow_last: Option<u32>,
613 dow_nth: Option<(u32, u32)>,
615}
616
617const MONTH_NAMES: [&str; 12] = [
618 "JAN", "FEB", "MAR", "APR", "MAY", "JUN", "JUL", "AUG", "SEP", "OCT", "NOV", "DEC",
619];
620const WEEKDAY_NAMES: [&str; 7] = ["SUN", "MON", "TUE", "WED", "THU", "FRI", "SAT"];
621
622impl CronExpr {
623 fn parse(expr: &str) -> std::result::Result<Self, ()> {
624 let fields: Vec<&str> = expr.split_whitespace().collect();
625
626 let (sec, min, hour, dom, month, dow, year): (
629 &str,
630 &str,
631 &str,
632 &str,
633 &str,
634 &str,
635 Option<&str>,
636 ) = match fields.len() {
637 5 => (
638 "0", fields[0], fields[1], fields[2], fields[3], fields[4], None,
639 ),
640 6 => (
641 fields[0], fields[1], fields[2], fields[3], fields[4], fields[5], None,
642 ),
643 7 => (
644 fields[0],
645 fields[1],
646 fields[2],
647 fields[3],
648 fields[4],
649 fields[5],
650 Some(fields[6]),
651 ),
652 _ => return Err(()),
653 };
654
655 let seconds = parse_field(sec, 0, 59, FieldKind::Plain)?;
656 let minutes = parse_field(min, 0, 59, FieldKind::Plain)?;
657 let hours = parse_field(hour, 0, 23, FieldKind::Plain)?;
658
659 let mut dom_last = false;
660 let mut dom_last_weekday = false;
661 let mut dom_nearest_weekday = None;
662 let days_of_month = parse_dom_field(
663 dom,
664 &mut dom_last,
665 &mut dom_last_weekday,
666 &mut dom_nearest_weekday,
667 )?;
668
669 let months = parse_field(month, 1, 12, FieldKind::Month)?;
670
671 let mut dow_last = None;
672 let mut dow_nth = None;
673 let days_of_week = parse_dow_field(dow, &mut dow_last, &mut dow_nth)?;
674
675 let years = match year {
676 Some(y) => Some(parse_field(y, 1970, 2099, FieldKind::Plain)?),
677 None => None,
678 };
679
680 let dom_restricted = dom != "*" && dom != "?";
682 let dow_restricted = dow != "*" && dow != "?";
683
684 Ok(Self {
685 seconds,
686 minutes,
687 hours,
688 days_of_month,
689 months,
690 days_of_week,
691 years,
692 dom_restricted,
693 dow_restricted,
694 dom_last,
695 dom_last_weekday,
696 dom_nearest_weekday,
697 dow_last,
698 dow_nth,
699 })
700 }
701
702 fn next_after(&self, after: DateTime<Utc>) -> Option<DateTime<Utc>> {
706 let local_after = after.with_timezone(&Local);
707
708 let by_seconds = self.seconds != vec![0];
710
711 let step = if by_seconds {
712 ChronoDuration::seconds(1)
713 } else {
714 ChronoDuration::minutes(1)
715 };
716
717 let mut candidate = if by_seconds {
718 local_after.with_nanosecond(0)? + ChronoDuration::seconds(1)
719 } else {
720 local_after.with_second(0)?.with_nanosecond(0)? + ChronoDuration::minutes(1)
721 };
722
723 let max_iterations: u64 = if by_seconds {
725 2u64 * 366 * 24 * 60 * 60
727 } else {
728 6u64 * 366 * 24 * 60
730 };
731 for _ in 0..max_iterations {
732 if self.matches_local(&candidate) {
733 return Some(candidate.with_timezone(&Utc));
734 }
735 candidate += step;
736 }
737 None
738 }
739
740 fn matches_local(&self, dt: &DateTime<Local>) -> bool {
741 if !self.seconds.contains(&dt.second()) {
742 return false;
743 }
744 if !self.minutes.contains(&dt.minute()) {
745 return false;
746 }
747 if !self.hours.contains(&dt.hour()) {
748 return false;
749 }
750 if !self.months.contains(&dt.month()) {
751 return false;
752 }
753 if let Some(years) = &self.years {
754 let year = dt.year();
755 if year < 0 || !years.contains(&(year as u32)) {
756 return false;
757 }
758 }
759
760 let dom_match = self.dom_matches(dt);
761 let dow_match = self.dow_matches(dt);
762
763 match (self.dom_restricted, self.dow_restricted) {
766 (true, true) => dom_match || dow_match,
767 (true, false) => dom_match,
768 (false, true) => dow_match,
769 (false, false) => true,
770 }
771 }
772
773 fn dom_matches(&self, dt: &DateTime<Local>) -> bool {
774 let dom = dt.day();
775 if self.dom_last && dom == last_day_of_month(dt.year(), dt.month()) {
776 return true;
777 }
778 if self.dom_last_weekday {
779 if is_nearest_weekday(dt, last_day_of_month(dt.year(), dt.month())) {
782 return true;
783 }
784 }
785 if let Some(target) = self.dom_nearest_weekday {
786 if is_nearest_weekday(dt, target) {
787 return true;
788 }
789 }
790 self.days_of_month.contains(&dom)
791 }
792
793 fn dow_matches(&self, dt: &DateTime<Local>) -> bool {
794 let dow = weekday_sun0(dt.weekday());
795
796 if let Some(target) = self.dow_last {
797 if dow == target {
799 let next_week = *dt + ChronoDuration::days(7);
800 if next_week.month() != dt.month() {
801 return true;
802 }
803 }
804 }
805 if let Some((target, n)) = self.dow_nth {
806 if dow == target {
807 let occurrence = (dt.day() - 1) / 7 + 1;
809 if occurrence == n {
810 return true;
811 }
812 }
813 }
814 self.days_of_week.contains(&dow)
815 }
816}
817
818fn weekday_sun0(weekday: Weekday) -> u32 {
820 weekday.num_days_from_sunday()
821}
822
823fn last_day_of_month(year: i32, month: u32) -> u32 {
825 let (ny, nm) = if month == 12 {
826 (year + 1, 1)
827 } else {
828 (year, month + 1)
829 };
830 let first_next = chrono::NaiveDate::from_ymd_opt(ny, nm, 1).expect("valid first-of-month");
831 (first_next - ChronoDuration::days(1)).day()
832}
833
834fn is_nearest_weekday(dt: &DateTime<Local>, target: u32) -> bool {
839 let last = last_day_of_month(dt.year(), dt.month());
840 let target = target.min(last);
841 let target_date = chrono::NaiveDate::from_ymd_opt(dt.year(), dt.month(), target);
842 let target_date = match target_date {
843 Some(d) => d,
844 None => return false,
845 };
846 let target_weekday = target_date.weekday();
847 let resolved_day = match target_weekday {
848 Weekday::Sat => {
849 if target > 1 {
850 target - 1
851 } else {
852 target + 2
854 }
855 }
856 Weekday::Sun => {
857 if target < last {
858 target + 1
859 } else {
860 target - 2
862 }
863 }
864 Weekday::Mon | Weekday::Tue | Weekday::Wed | Weekday::Thu | Weekday::Fri => target,
865 };
866 dt.day() == resolved_day
867}
868
869#[derive(Clone, Copy, PartialEq, Eq)]
870enum FieldKind {
871 Plain,
872 Month,
873 Weekday,
874}
875
876fn parse_field(
880 field: &str,
881 min: u32,
882 max: u32,
883 kind: FieldKind,
884) -> std::result::Result<Vec<u32>, ()> {
885 if field == "?" {
886 return Ok((min..=max).collect());
888 }
889 let mut values: Vec<u32> = Vec::new();
890 for part in field.split(',') {
891 if part.is_empty() {
892 return Err(());
893 }
894 parse_field_part(part, min, max, kind, &mut values)?;
895 }
896 if values.is_empty() {
897 return Err(());
898 }
899 values.sort_unstable();
900 values.dedup();
901 Ok(values)
902}
903
904fn parse_dom_field(
908 field: &str,
909 dom_last: &mut bool,
910 dom_last_weekday: &mut bool,
911 dom_nearest_weekday: &mut Option<u32>,
912) -> std::result::Result<Vec<u32>, ()> {
913 let upper = field.to_ascii_uppercase();
914 if upper == "L" {
915 *dom_last = true;
916 return Ok(Vec::new());
918 }
919 if upper == "LW" {
920 *dom_last_weekday = true;
921 return Ok(Vec::new());
922 }
923 if let Some(stripped) = upper.strip_suffix('W') {
924 let day: u32 = stripped.parse().map_err(|_| ())?;
925 if !(1..=31).contains(&day) {
926 return Err(());
927 }
928 *dom_nearest_weekday = Some(day);
929 return Ok(Vec::new());
930 }
931 parse_field(field, 1, 31, FieldKind::Plain)
932}
933
934fn parse_dow_field(
937 field: &str,
938 dow_last: &mut Option<u32>,
939 dow_nth: &mut Option<(u32, u32)>,
940) -> std::result::Result<Vec<u32>, ()> {
941 let upper = field.to_ascii_uppercase();
942
943 if let Some((wd, nth)) = upper.split_once('#') {
945 let weekday = parse_weekday_token(wd)?;
946 let n: u32 = nth.parse().map_err(|_| ())?;
947 if !(1..=5).contains(&n) {
948 return Err(());
949 }
950 *dow_nth = Some((weekday, n));
951 return Ok(Vec::new());
952 }
953
954 if let Some(stripped) = upper.strip_suffix('L') {
956 let weekday = parse_weekday_token(stripped)?;
957 *dow_last = Some(weekday);
958 return Ok(Vec::new());
959 }
960
961 if upper == "?" || upper == "*" {
962 let mut v = parse_field(field, 0, 7, FieldKind::Plain)?;
963 fold_sunday(&mut v);
964 return Ok(v);
965 }
966
967 let mut values = parse_field(field, 0, 7, FieldKind::Weekday)?;
968 fold_sunday(&mut values);
969 Ok(values)
970}
971
972fn fold_sunday(values: &mut Vec<u32>) {
974 for v in values.iter_mut() {
975 if *v == 7 {
976 *v = 0;
977 }
978 }
979 values.sort_unstable();
980 values.dedup();
981}
982
983fn parse_weekday_token(token: &str) -> std::result::Result<u32, ()> {
985 let upper = token.to_ascii_uppercase();
986 if let Some(idx) = WEEKDAY_NAMES.iter().position(|name| *name == upper) {
987 return Ok(idx as u32);
988 }
989 let v: u32 = upper.parse().map_err(|_| ())?;
990 match v {
991 0..=6 => Ok(v),
992 7 => Ok(0),
993 _ => Err(()),
994 }
995}
996
997impl FieldKind {
999 fn resolve_name(self, token: &str) -> Option<u32> {
1000 let upper = token.to_ascii_uppercase();
1001 match self {
1002 FieldKind::Plain => None,
1003 FieldKind::Month => MONTH_NAMES
1004 .iter()
1005 .position(|name| *name == upper)
1006 .map(|i| (i + 1) as u32),
1007 FieldKind::Weekday => WEEKDAY_NAMES
1008 .iter()
1009 .position(|name| *name == upper)
1010 .map(|i| i as u32),
1011 }
1012 }
1013}
1014
1015fn parse_field_part(
1016 part: &str,
1017 min: u32,
1018 max: u32,
1019 kind: FieldKind,
1020 out: &mut Vec<u32>,
1021) -> std::result::Result<(), ()> {
1022 let (range_spec, step) = match part.split_once('/') {
1024 Some((range_spec, step_str)) => {
1025 let step: u32 = step_str.parse().map_err(|_| ())?;
1026 if step == 0 {
1027 return Err(());
1028 }
1029 (range_spec, Some(step))
1030 }
1031 None => (part, None),
1032 };
1033
1034 let (start, end) = if range_spec == "*" {
1036 (min, max)
1037 } else if let Some((lo, hi)) = range_spec.split_once('-') {
1038 let lo = parse_value_token(lo, kind)?;
1039 let hi = parse_value_token(hi, kind)?;
1040 (lo, hi)
1041 } else {
1042 if step.is_some() {
1045 return Err(());
1046 }
1047 let v = parse_value_token(range_spec, kind)?;
1048 (v, v)
1049 };
1050
1051 if start < min || end > max || start > end {
1052 return Err(());
1053 }
1054
1055 let step = step.unwrap_or(1);
1056 let mut v = start;
1057 while v <= end {
1058 out.push(v);
1059 v += step;
1060 }
1061 Ok(())
1062}
1063
1064fn parse_value_token(token: &str, kind: FieldKind) -> std::result::Result<u32, ()> {
1067 match kind {
1068 FieldKind::Weekday => parse_weekday_token(token),
1069 FieldKind::Month => {
1070 if let Some(v) = kind.resolve_name(token) {
1071 return Ok(v);
1072 }
1073 token.parse().map_err(|_| ())
1074 }
1075 FieldKind::Plain => token.parse().map_err(|_| ()),
1076 }
1077}
1078
1079impl AgentOs {
1084 pub fn schedule_cron(
1093 &self,
1094 options: CronJobOptions,
1095 ) -> std::result::Result<CronJobHandle, ClientError> {
1096 let cron = self.cron();
1097 let now = Utc::now();
1098
1099 let next_run = validate_schedule(&options.schedule, now)?;
1101
1102 let id = options
1103 .id
1104 .unwrap_or_else(|| uuid::Uuid::new_v4().to_string());
1105 let overlap = options.overlap.unwrap_or_default();
1106
1107 let manager = Arc::clone(cron);
1110 let vm = self.clone();
1111 let callback_id = id.clone();
1112 let callback: crate::config::ScheduleCallback = Arc::new(move || {
1113 let manager = Arc::clone(&manager);
1114 let vm = vm.clone();
1115 let id = callback_id.clone();
1116 Box::pin(async move {
1117 execute_job(manager, vm, id).await;
1118 })
1119 });
1120
1121 register_cron_job(
1122 cron,
1123 id,
1124 options.schedule,
1125 options.action,
1126 overlap,
1127 next_run,
1128 callback,
1129 )
1130 }
1131
1132 pub fn list_cron_jobs(&self) -> Vec<CronJobInfo> {
1134 let mut result = Vec::new();
1135 self.cron().jobs.scan(|id, state| {
1136 result.push(CronJobInfo {
1137 id: id.clone(),
1138 schedule: state.schedule.clone(),
1139 action: state.action.clone(),
1140 overlap: state.overlap,
1141 last_run: *state.last_run.lock(),
1142 next_run: *state.next_run.lock(),
1143 run_count: state.run_count.load(Ordering::SeqCst),
1144 running: state.running.load(Ordering::SeqCst),
1145 });
1146 });
1147 result
1148 }
1149
1150 pub fn cancel_cron_job(&self, id: &str) {
1152 self.cron().cancel_job(id);
1153 }
1154
1155 pub fn cron_events(&self) -> broadcast::Receiver<CronEvent> {
1158 self.cron().event_tx.subscribe()
1159 }
1160}
1161
1162fn ensure_cron_capacity(cron: &CronManager, id: &str) -> std::result::Result<(), ClientError> {
1163 if cron.jobs.contains(id) || cron.jobs.len() < crate::CRON_JOB_LIMIT {
1164 return Ok(());
1165 }
1166
1167 Err(ClientError::Sidecar(format!(
1168 "cron job limit exceeded: at most {} jobs can be scheduled per VM",
1169 crate::CRON_JOB_LIMIT
1170 )))
1171}
1172
1173fn register_cron_job(
1174 cron: &Arc<CronManager>,
1175 id: String,
1176 schedule: String,
1177 action: CronAction,
1178 overlap: CronOverlap,
1179 next_run: Option<DateTime<Utc>>,
1180 callback: crate::config::ScheduleCallback,
1181) -> std::result::Result<CronJobHandle, ClientError> {
1182 let _guard = cron.schedule_lock.lock();
1183 ensure_cron_capacity(cron, &id)?;
1184
1185 if let Some((_, old)) = cron.jobs.remove(&id) {
1189 cron.driver.cancel(&old.handle);
1190 }
1191
1192 let handle = cron.driver.schedule(ScheduleEntry {
1193 id: id.clone(),
1194 schedule: schedule.clone(),
1195 callback,
1196 });
1197
1198 let state = CronJobState {
1199 schedule,
1200 action,
1201 overlap,
1202 last_run: parking_lot::Mutex::new(None),
1203 next_run: parking_lot::Mutex::new(next_run),
1204 run_count: std::sync::atomic::AtomicU64::new(0),
1205 running: AtomicBool::new(false),
1206 queued: AtomicBool::new(false),
1207 handle,
1208 };
1209
1210 let _ = cron.jobs.insert(id.clone(), state);
1211
1212 Ok(CronJobHandle {
1213 id,
1214 manager: Arc::clone(cron),
1215 })
1216}
1217
1218#[cfg(test)]
1219mod tests {
1220 use super::{
1221 ensure_cron_capacity, register_cron_job, CronAction, CronJobState, CronManager,
1222 CronOverlap, ScheduleDriver, ScheduleEntry, ScheduleHandle,
1223 };
1224 use crate::CRON_JOB_LIMIT;
1225 use std::sync::atomic::AtomicBool;
1226 use std::sync::Arc;
1227
1228 #[derive(Default)]
1229 struct RecordingScheduleDriver {
1230 calls: parking_lot::Mutex<Vec<String>>,
1231 }
1232
1233 impl ScheduleDriver for RecordingScheduleDriver {
1234 fn schedule(&self, entry: ScheduleEntry) -> ScheduleHandle {
1235 self.calls.lock().push(format!("schedule:{}", entry.id));
1236 ScheduleHandle { id: entry.id }
1237 }
1238
1239 fn cancel(&self, handle: &ScheduleHandle) {
1240 self.calls.lock().push(format!("cancel:{}", handle.id));
1241 }
1242
1243 fn dispose(&self) {}
1244 }
1245
1246 fn dummy_state(id: String) -> CronJobState {
1247 CronJobState {
1248 schedule: "0 0 * * *".to_string(),
1249 action: CronAction::Callback {
1250 callback: Arc::new(|| Box::pin(async {})),
1251 },
1252 overlap: CronOverlap::Allow,
1253 last_run: parking_lot::Mutex::new(None),
1254 next_run: parking_lot::Mutex::new(None),
1255 run_count: std::sync::atomic::AtomicU64::new(0),
1256 running: AtomicBool::new(false),
1257 queued: AtomicBool::new(false),
1258 handle: ScheduleHandle { id },
1259 }
1260 }
1261
1262 #[test]
1263 fn cron_capacity_rejects_new_jobs_at_limit_but_allows_replacements() {
1264 let manager = CronManager::new(Arc::new(RecordingScheduleDriver::default()));
1265 for index in 0..CRON_JOB_LIMIT {
1266 let id = format!("job-{index}");
1267 assert!(
1268 manager.jobs.insert(id.clone(), dummy_state(id)).is_ok(),
1269 "seed cron job"
1270 );
1271 }
1272
1273 let error = ensure_cron_capacity(&manager, "overflow").expect_err("limit should reject");
1274 assert!(
1275 error.to_string().contains("cron job limit exceeded"),
1276 "unexpected limit error: {error}"
1277 );
1278 ensure_cron_capacity(&manager, "job-0").expect("replacement should be allowed");
1279 }
1280
1281 #[test]
1300 fn cron_exec_action_argv_is_not_shell_re_split_or_evaluated() {
1301 fn cron_exec_argv(command: &str, args: &[&str]) -> (String, Vec<String>) {
1304 (
1305 command.to_string(),
1306 args.iter().map(|a| a.to_string()).collect(),
1307 )
1308 }
1309
1310 fn buggy_join_then_resolve(command: &str, args: &[&str]) -> (String, Vec<String>) {
1312 let joined = if args.is_empty() {
1313 command.to_string()
1314 } else {
1315 format!("{} {}", command, args.join(" "))
1316 };
1317 crate::command_line::resolve_exec_command(&joined).expect("line must resolve")
1318 }
1319
1320 let (cmd, args) = cron_exec_argv("printenv", &["a b"]);
1322 assert_eq!(
1323 (cmd.as_str(), args.as_slice()),
1324 ("printenv", &["a b".to_string()][..]),
1325 "N-007: structured argv element \"a b\" must survive as a single argv element"
1326 );
1327 let (_, buggy_args) = buggy_join_then_resolve("printenv", &["a b"]);
1329 assert_eq!(
1330 buggy_args,
1331 vec!["a".to_string(), "b".to_string()],
1332 "N-007 negative control: the old join+resolve path re-split \"a b\" into two argv elements"
1333 );
1334
1335 let (cmd, args) = cron_exec_argv("printenv", &["$(id)"]);
1337 assert_eq!(
1338 (cmd.as_str(), args.as_slice()),
1339 ("printenv", &["$(id)".to_string()][..]),
1340 "N-007: command-substitution argv element \"$(id)\" must NOT be promoted to `sh -c`"
1341 );
1342 let (buggy_cmd, _) = buggy_join_then_resolve("printenv", &["$(id)"]);
1344 assert_eq!(
1345 buggy_cmd, "sh",
1346 "N-007 negative control: the old path promoted the `$(id)` line to a `sh -c` shell"
1347 );
1348
1349 let (cmd, args) = cron_exec_argv("printenv", &["`id`"]);
1351 assert_eq!(
1352 (cmd.as_str(), args.as_slice()),
1353 ("printenv", &["`id`".to_string()][..]),
1354 "N-007: backtick argv element \"`id`\" must NOT be promoted to `sh -c`"
1355 );
1356 let (buggy_cmd, _) = buggy_join_then_resolve("printenv", &["`id`"]);
1357 assert_eq!(
1358 buggy_cmd, "sh",
1359 "N-007 negative control: the old path promoted the backtick line to a `sh -c` shell"
1360 );
1361 }
1362
1363 #[test]
1372 fn schedule_cron_public_path_rejects_jobs_beyond_cron_job_limit() {
1373 let driver = Arc::new(RecordingScheduleDriver::default());
1374 let manager = Arc::new(CronManager::new(driver.clone()));
1375
1376 let make_callback =
1377 || -> crate::config::ScheduleCallback { Arc::new(|| Box::pin(async {})) };
1378
1379 for index in 0..CRON_JOB_LIMIT {
1381 register_cron_job(
1382 &manager,
1383 format!("flood-{index}"),
1384 "0 0 * * *".to_string(),
1385 CronAction::Callback {
1386 callback: make_callback(),
1387 },
1388 CronOverlap::Allow,
1389 None,
1390 make_callback(),
1391 )
1392 .unwrap_or_else(|err| panic!("seed job {index} should register: {err}"));
1393 }
1394 assert_eq!(manager.jobs.len(), CRON_JOB_LIMIT);
1395
1396 let overflow = match register_cron_job(
1399 &manager,
1400 "flood-overflow".to_string(),
1401 "0 0 * * *".to_string(),
1402 CronAction::Callback {
1403 callback: make_callback(),
1404 },
1405 CronOverlap::Allow,
1406 None,
1407 make_callback(),
1408 ) {
1409 Ok(_) => {
1410 panic!("AOSCLIENT-P2-cron-cap: the job beyond CRON_JOB_LIMIT must be rejected")
1411 }
1412 Err(err) => err,
1413 };
1414 assert!(
1415 overflow.to_string().contains("cron job limit exceeded"),
1416 "AOSCLIENT-P2-cron-cap: overflow rejection must report the cron job limit, got: {overflow}"
1417 );
1418 assert_eq!(
1419 manager.jobs.len(),
1420 CRON_JOB_LIMIT,
1421 "AOSCLIENT-P2-cron-cap: a rejected overflow job must not be inserted"
1422 );
1423
1424 register_cron_job(
1426 &manager,
1427 "flood-0".to_string(),
1428 "0 1 * * *".to_string(),
1429 CronAction::Callback {
1430 callback: make_callback(),
1431 },
1432 CronOverlap::Allow,
1433 None,
1434 make_callback(),
1435 )
1436 .expect("AOSCLIENT-P2-cron-cap: replacing an existing id at the cap must be allowed");
1437 assert_eq!(
1438 manager.jobs.len(),
1439 CRON_JOB_LIMIT,
1440 "AOSCLIENT-P2-cron-cap: replacing an existing id must not grow the registry past the cap"
1441 );
1442 }
1443
1444 #[test]
1445 fn cron_replacement_cancels_old_timer_before_scheduling_new_timer() {
1446 let driver = Arc::new(RecordingScheduleDriver::default());
1447 let manager = Arc::new(CronManager::new(driver.clone()));
1448 let callback: crate::config::ScheduleCallback = Arc::new(|| Box::pin(async {}));
1449
1450 register_cron_job(
1451 &manager,
1452 "same-id".to_string(),
1453 "0 0 * * *".to_string(),
1454 CronAction::Callback {
1455 callback: callback.clone(),
1456 },
1457 CronOverlap::Allow,
1458 None,
1459 callback.clone(),
1460 )
1461 .expect("initial schedule");
1462 register_cron_job(
1463 &manager,
1464 "same-id".to_string(),
1465 "0 1 * * *".to_string(),
1466 CronAction::Callback { callback },
1467 CronOverlap::Allow,
1468 None,
1469 Arc::new(|| Box::pin(async {})),
1470 )
1471 .expect("replacement schedule");
1472
1473 assert_eq!(
1474 *driver.calls.lock(),
1475 vec!["schedule:same-id", "cancel:same-id", "schedule:same-id"]
1476 );
1477 assert_eq!(manager.jobs.len(), 1);
1478 }
1479}